weave/pipelines/jobs/ingest_congress.py

"""Ingestion pipeline for loading congress data from unitedstates/congress JSON files.

Loads legislators, bills, votes, vote records, and bill text into the data_science_dev database.
Expects the parent directory to contain congress-tracker/ and congress-legislators/ as siblings.

Usage:
    ingest-congress /path/to/parent/
    ingest-congress /path/to/parent/ --congress 118
    ingest-congress /path/to/parent/ --congress 118 --only bills
"""

from __future__ import annotations

import logging
from dataclasses import dataclass
from datetime import date
from pathlib import Path  # noqa: TC003 needed at runtime for typer CLI argument
from typing import TYPE_CHECKING, Annotated

import orjson
import typer
import yaml
from sqlalchemy import func, select
from sqlalchemy.orm import Session

from pipelines.common import configure_logger
from pipelines.pipelines.jobs.congress_vote_context import (
    build_billstatus_text_version_index,
    build_vote_action_matches,
    classify_votes,
    coerce_raw_ref,
    derive_session_number,
    finish_ingest_run,
    filter_context_supported_congress_dirs,
    ingest_bill_status_context as rebuild_bill_status_context,
    parse_date_like,
    parse_vote_source_url,
    parsed_vote_datetime,
    register_source_artifact,
    require_billstatus_artifacts,
    resolve_vote_position_meanings,
    resolve_vote_text_targets,
    start_ingest_run,
)
from pipelines.orm.common import get_postgres_engine
from pipelines.orm.data_science_dev.congress import (
    Bill,
    BillText,
    Legislator,
    LegislatorSocialMedia,
    Vote,
    VoteActionMatch,
    VoteClassification,
    VoteContextAudit,
    VoteRecord,
)
from pipelines.parallelize import parallelize_thread

if TYPE_CHECKING:
    from collections.abc import Iterator, Sequence

    from sqlalchemy.engine import Engine

logger = logging.getLogger(__name__)

BATCH_SIZE = 10_000
PARALLEL_FILE_CHUNK_SIZE = 1_000
PARALLEL_PROGRESS_TRACKER = 250

app = typer.Typer(help="Ingest unitedstates/congress data into data_science_dev.")


@app.command()
def main(
    parent_dir: Annotated[
        Path,
        typer.Argument(
            help="Parent directory containing congress-tracker/ and congress-legislators/"
        ),
    ],
    congress: Annotated[
        int | None, typer.Option(help="Only ingest a specific congress number")
    ] = None,
    only: Annotated[
        str | None,
        typer.Option(
            help=(
                "Only run a specific step: legislators, legislators-social-media, "
                "bills, bill-text, votes, bill-status-context, vote-context-match, "
                "vote-context-classify, vote-text-resolve, vote-polarity-resolve, "
                "vote-context-diagnostics"
            )
        ),
    ] = None,
) -> None:
    """Ingest congress data from unitedstates/congress JSON files."""
    configure_logger(level="INFO")

    data_dir = parent_dir / "congress-tracker/congress/data/"
    legislators_dir = parent_dir / "congress-legislators"

    if not data_dir.is_dir():
        typer.echo(f"Expected congress-tracker/ directory: {data_dir}", err=True)
        raise typer.Exit(code=1)

    if not legislators_dir.is_dir():
        typer.echo(
            f"Expected congress-legislators/ directory: {legislators_dir}", err=True
        )
        raise typer.Exit(code=1)

    engine = get_postgres_engine(name="DATA_SCIENCE_DEV")

    congress_dirs = _resolve_congress_dirs(data_dir, congress)
    if not congress_dirs:
        typer.echo("No congress directories found.", err=True)
        raise typer.Exit(code=1)

    logger.info("Found %d congress directories to process", len(congress_dirs))

    with Session(engine) as session:
        ingest_run = start_ingest_run(
            session,
            source_snapshot_label=str(parent_dir),
            repo_root=Path(__file__).resolve().parent.parent,
        )
        ingest_run_id = ingest_run.id

    steps: dict[str, tuple] = {
        "legislators": (ingest_legislators, (engine, legislators_dir)),
        "legislators-social-media": (ingest_social_media, (engine, legislators_dir)),
        "bills": (ingest_bills, (engine, congress_dirs)),
        "bill-text": (ingest_bill_text, (engine, congress_dirs, ingest_run_id)),
        "votes": (ingest_votes, (engine, congress_dirs, ingest_run_id)),
        "bill-status-context": (
            ingest_bill_status_context_step,
            (engine, congress_dirs, ingest_run_id),
        ),
        "vote-context-match": (vote_context_match_step, (engine, congress_dirs)),
        "vote-context-classify": (vote_context_classify_step, (engine, congress_dirs)),
        "vote-text-resolve": (vote_text_resolve_step, (engine, congress_dirs)),
        "vote-polarity-resolve": (vote_polarity_resolve_step, (engine, congress_dirs)),
        "vote-context-diagnostics": (
            vote_context_diagnostics_step,
            (engine, congress_dirs),
        ),
    }

    if only:
        if only not in steps:
            typer.echo(
                f"Unknown step: {only}. Choose from: {', '.join(steps)}", err=True
            )
            raise typer.Exit(code=1)
        steps = {only: steps[only]}

    try:
        for step_name, (step_func, step_args) in steps.items():
            logger.info("=== Starting step: %s ===", step_name)
            step_func(*step_args)
            logger.info("=== Finished step: %s ===", step_name)
    except Exception:
        with Session(engine) as session:
            finish_ingest_run(session, ingest_run_id, status="failed")
        raise

    with Session(engine) as session:
        finish_ingest_run(session, ingest_run_id, status="completed")
    logger.info("ingest-congress done")


def _resolve_congress_dirs(data_dir: Path, congress: int | None) -> list[Path]:
    """Find congress number directories under data_dir."""
    if congress is not None:
        target = data_dir / str(congress)
        return [target] if target.is_dir() else []
    return sorted(
        path for path in data_dir.iterdir() if path.is_dir() and path.name.isdigit()
    )


def _flush_batch(session: Session, batch: list[object], label: str) -> int:
    """Add a batch of ORM objects to the session and commit. Returns count added."""
    if not batch:
        return 0
    session.add_all(batch)
    session.commit()
    count = len(batch)
    logger.info("Committed %d %s", count, label)
    batch.clear()
    return count


@dataclass(frozen=True)
class LoadedJsonFile:
    path: Path
    data: dict | None


@dataclass(frozen=True)
class PreparedBillTextInput:
    bill_id: int
    bill_key: tuple[int, str, int]
    version_code: str
    text_content: str | None
    version_data: dict | None
    source_file: Path | None
    billstatus_version: object | None


def _chunked[T](items: Sequence[T], chunk_size: int) -> Iterator[Sequence[T]]:
    """Yield fixed-size slices from a sequence."""
    for start in range(0, len(items), chunk_size):
        yield items[start : start + chunk_size]


def _load_json_file(*, path: Path) -> LoadedJsonFile:
    """Read one JSON file off disk for later serial DB processing."""
    return LoadedJsonFile(path=path, data=_read_json(path))


def _prepare_bill_text_input(
    *,
    bill_id: int,
    bill_key: tuple[int, str, int],
    version_dir: Path,
    billstatus_version: object | None,
) -> PreparedBillTextInput:
    """Load one bill-text directory off disk for later serial DB processing."""
    source_file = version_dir / "data.json"
    if not source_file.exists():
        source_file = _find_text_source_file(version_dir)
    return PreparedBillTextInput(
        bill_id=bill_id,
        bill_key=bill_key,
        version_code=version_dir.name,
        text_content=_read_bill_text(version_dir),
        version_data=_read_json(version_dir / "data.json"),
        source_file=source_file if source_file and source_file.exists() else None,
        billstatus_version=billstatus_version,
    )


# ---------------------------------------------------------------------------
# Legislators — loaded from congress-legislators YAML files
# ---------------------------------------------------------------------------


def ingest_legislators(engine: Engine, legislators_dir: Path) -> None:
    """Load legislators from congress-legislators YAML files."""
    legislators_data = _load_legislators_yaml(legislators_dir)
    logger.info("Loaded %d legislators from YAML files", len(legislators_data))

    with Session(engine) as session:
        existing_legislators = {
            legislator.bioguide_id: legislator
            for legislator in session.scalars(select(Legislator)).all()
        }
        logger.info("Found %d existing legislators in DB", len(existing_legislators))

        total_inserted = 0
        total_updated = 0
        for entry in legislators_data:
            bioguide_id = entry.get("id", {}).get("bioguide")
            if not bioguide_id:
                continue

            fields = _parse_legislator(entry)
            if existing := existing_legislators.get(bioguide_id):
                changed = False
                for field, value in fields.items():
                    if value is not None and getattr(existing, field) != value:
                        setattr(existing, field, value)
                        changed = True
                if changed:
                    total_updated += 1
            else:
                session.add(Legislator(bioguide_id=bioguide_id, **fields))
                total_inserted += 1

        session.commit()
    logger.info(
        "Inserted %d new legislators, updated %d existing",
        total_inserted,
        total_updated,
    )


def _load_legislators_yaml(legislators_dir: Path) -> list[dict]:
    """Load and combine legislators-current.yaml and legislators-historical.yaml."""
    legislators: list[dict] = []
    for filename in ("legislators-current.yaml", "legislators-historical.yaml"):
        path = legislators_dir / filename
        if not path.exists():
            logger.warning("Legislators file not found: %s", path)
            continue
        with path.open() as file:
            data = yaml.safe_load(file)
            if isinstance(data, list):
                legislators.extend(data)
    return legislators


def _parse_legislator(entry: dict) -> dict:
    """Extract Legislator fields from a congress-legislators YAML entry."""
    ids = entry.get("id", {})
    name = entry.get("name", {})
    bio = entry.get("bio", {})
    terms = entry.get("terms", [])
    latest_term = terms[-1] if terms else {}

    fec_ids = ids.get("fec")
    fec_ids_joined = ",".join(fec_ids) if isinstance(fec_ids, list) else fec_ids

    chamber = latest_term.get("type")
    chamber_normalized = {"rep": "House", "sen": "Senate"}.get(chamber, chamber)

    return {
        "thomas_id": ids.get("thomas"),
        "lis_id": ids.get("lis"),
        "govtrack_id": ids.get("govtrack"),
        "opensecrets_id": ids.get("opensecrets"),
        "fec_ids": fec_ids_joined,
        "first_name": name.get("first"),
        "last_name": name.get("last"),
        "official_full_name": name.get("official_full"),
        "nickname": name.get("nickname"),
        "birthday": bio.get("birthday"),
        "gender": bio.get("gender"),
        "current_party": latest_term.get("party"),
        "current_state": latest_term.get("state"),
        "current_district": latest_term.get("district"),
        "current_chamber": chamber_normalized,
    }


# ---------------------------------------------------------------------------
# Social Media — loaded from legislators-social-media.yaml
# ---------------------------------------------------------------------------

SOCIAL_MEDIA_PLATFORMS = {
    "twitter": "https://twitter.com/{account}",
    "facebook": "https://facebook.com/{account}",
    "youtube": "https://youtube.com/{account}",
    "instagram": "https://instagram.com/{account}",
    "mastodon": None,
}


def ingest_social_media(engine: Engine, legislators_dir: Path) -> None:
    """Load social media accounts from legislators-social-media.yaml."""
    social_media_path = legislators_dir / "legislators-social-media.yaml"
    if not social_media_path.exists():
        logger.warning("Social media file not found: %s", social_media_path)
        return

    with social_media_path.open() as file:
        social_media_data = yaml.safe_load(file)

    if not isinstance(social_media_data, list):
        logger.warning("Unexpected format in %s", social_media_path)
        return

    logger.info(
        "Loaded %d entries from legislators-social-media.yaml", len(social_media_data)
    )

    with Session(engine) as session:
        legislator_map = _build_legislator_map(session)
        existing_accounts = {
            (account.legislator_id, account.platform)
            for account in session.scalars(select(LegislatorSocialMedia)).all()
        }
        logger.info(
            "Found %d existing social media accounts in DB", len(existing_accounts)
        )

        total_inserted = 0
        total_updated = 0
        for entry in social_media_data:
            bioguide_id = entry.get("id", {}).get("bioguide")
            if not bioguide_id:
                continue

            legislator_id = legislator_map.get(bioguide_id)
            if legislator_id is None:
                continue

            social = entry.get("social", {})
            for platform, url_template in SOCIAL_MEDIA_PLATFORMS.items():
                account_name = social.get(platform)
                if not account_name:
                    continue

                url = (
                    url_template.format(account=account_name) if url_template else None
                )

                if (legislator_id, platform) in existing_accounts:
                    total_updated += 1
                else:
                    session.add(
                        LegislatorSocialMedia(
                            legislator_id=legislator_id,
                            platform=platform,
                            account_name=str(account_name),
                            url=url,
                            source="https://github.com/unitedstates/congress-legislators",
                        )
                    )
                    existing_accounts.add((legislator_id, platform))
                    total_inserted += 1

        session.commit()
    logger.info(
        "Inserted %d new social media accounts, updated %d existing",
        total_inserted,
        total_updated,
    )


def _iter_voters(position_group: object) -> Iterator[dict]:
    """Yield voter dicts from a vote position group (handles list, single dict, or string)."""
    if isinstance(position_group, dict):
        yield position_group
    elif isinstance(position_group, list):
        for voter in position_group:
            if isinstance(voter, dict):
                yield voter


# ---------------------------------------------------------------------------
# Bills
# ---------------------------------------------------------------------------


def ingest_bills(engine: Engine, congress_dirs: list[Path]) -> None:
    """Load bill data.json files."""
    with Session(engine) as session:
        existing_bills = {
            (bill.congress, bill.bill_type, bill.number)
            for bill in session.scalars(select(Bill)).all()
        }
        logger.info("Found %d existing bills in DB", len(existing_bills))

        total_inserted = 0
        batch: list[Bill] = []
        for congress_dir in congress_dirs:
            bills_dir = congress_dir / "bills"
            if not bills_dir.is_dir():
                continue
            bill_files = sorted(bills_dir.rglob("data.json"))
            logger.info(
                "Scanning %d bill files from %s",
                len(bill_files),
                congress_dir.name,
            )
            for chunk in _chunked(bill_files, PARALLEL_FILE_CHUNK_SIZE):
                results = parallelize_thread(
                    _load_json_file,
                    [{"path": bill_file} for bill_file in chunk],
                    progress_tracker=PARALLEL_PROGRESS_TRACKER,
                )
                for loaded in results.results:
                    if loaded.data is None:
                        continue
                    bill = _parse_bill(loaded.data, existing_bills)
                    if bill is not None:
                        batch.append(bill)
                        if len(batch) >= BATCH_SIZE:
                            total_inserted += _flush_batch(session, batch, "bills")

        total_inserted += _flush_batch(session, batch, "bills")
    logger.info("Inserted %d new bills total", total_inserted)


def _parse_bill(data: dict, existing_bills: set[tuple[int, str, int]]) -> Bill | None:
    """Parse a bill data.json dict into a Bill ORM object, skipping existing."""
    raw_congress = data.get("congress")
    bill_type = data.get("bill_type")
    raw_number = data.get("number")
    if raw_congress is None or bill_type is None or raw_number is None:
        return None
    congress = int(raw_congress)
    number = int(raw_number)
    if (congress, bill_type, number) in existing_bills:
        return None

    sponsor_bioguide = None
    sponsor = data.get("sponsor")
    if sponsor:
        sponsor_bioguide = sponsor.get("bioguide_id")

    return Bill(
        congress=congress,
        bill_type=bill_type,
        number=number,
        title=data.get("short_title") or data.get("official_title"),
        title_short=data.get("short_title"),
        official_title=data.get("official_title"),
        status=data.get("status"),
        status_at=data.get("status_at"),
        sponsor_bioguide_id=sponsor_bioguide,
        subjects_top_term=data.get("subjects_top_term"),
    )


# ---------------------------------------------------------------------------
# Votes (and vote records)
# ---------------------------------------------------------------------------


def ingest_votes(
    engine: Engine,
    congress_dirs: list[Path],
    ingest_run_id: int | None,
) -> None:
    """Load raw vote data and member positions without any guessed text linkage."""
    legislator_map = _build_legislator_map_for_engine(engine)
    logger.info("Loaded %d legislators into lookup map", len(legislator_map))
    with Session(engine) as session:
        existing_votes = {
            (vote.congress, vote.chamber, vote.session_number, vote.roll_number)
            for vote in session.scalars(select(Vote)).all()
        }
        logger.info("Found %d existing votes in DB", len(existing_votes))

        total_inserted = 0
        batch: list[Vote] = []
        for congress_dir in congress_dirs:
            votes_dir = congress_dir / "votes"
            if not votes_dir.is_dir():
                continue
            vote_files = sorted(votes_dir.rglob("data.json"))
            logger.info(
                "Scanning %d vote files from %s",
                len(vote_files),
                congress_dir.name,
            )
            for chunk in _chunked(vote_files, PARALLEL_FILE_CHUNK_SIZE):
                results = parallelize_thread(
                    _load_json_file,
                    [{"path": vote_file} for vote_file in chunk],
                    progress_tracker=PARALLEL_PROGRESS_TRACKER,
                )
                for loaded in results.results:
                    if loaded.data is None:
                        continue
                    chamber = normalize_vote_chamber(loaded.data.get("chamber"))
                    if chamber is None:
                        continue
                    artifact = register_source_artifact(
                        session,
                        path=loaded.path,
                        source_kind="vote_json",
                        congress=int(loaded.data.get("congress", congress_dir.name)),
                        chamber=chamber,
                        ingest_run_id=ingest_run_id,
                        source_url=parse_vote_source_url(loaded.data),
                    )
                    vote = _parse_vote(
                        loaded.data,
                        legislator_map,
                        existing_votes,
                        raw_vote_source_artifact_id=artifact.id,
                    )
                    if vote is not None:
                        batch.append(vote)
                        if len(batch) >= BATCH_SIZE:
                            total_inserted += _flush_batch(session, batch, "votes")

        total_inserted += _flush_batch(session, batch, "votes")
    logger.info("Inserted %d new votes total", total_inserted)


def _build_legislator_map(session: Session) -> dict[str, int]:
    """Build a mapping of bioguide_id -> legislator.id."""
    return {
        legislator.bioguide_id: legislator.id
        for legislator in session.scalars(select(Legislator)).all()
    }


def _build_bill_map(session: Session) -> dict[tuple[int, str, int], int]:
    """Build a mapping of (congress, bill_type, number) -> bill.id."""
    return {
        (bill.congress, bill.bill_type, bill.number): bill.id
        for bill in session.scalars(select(Bill)).all()
    }


def _build_legislator_map_for_engine(engine: Engine) -> dict[str, int]:
    """Build the legislator lookup in a short-lived read session."""
    with Session(engine) as session:
        return _build_legislator_map(session)


def _build_bill_map_for_engine(engine: Engine) -> dict[tuple[int, str, int], int]:
    """Build the bill lookup in a short-lived read session."""
    with Session(engine) as session:
        return _build_bill_map(session)


def _parse_vote(
    data: dict,
    legislator_map: dict[str, int],
    existing_votes: set[tuple[int, str, int, int]],
    *,
    raw_vote_source_artifact_id: int | None,
) -> Vote | None:
    """Parse a vote data.json dict into a Vote ORM object with records."""
    raw_congress = data.get("congress")
    chamber = data.get("chamber")
    raw_number = data.get("number")
    vote_date = data.get("date")
    if (
        raw_congress is None
        or chamber is None
        or raw_number is None
        or vote_date is None
    ):
        return None

    raw_session = data.get("session")
    congress = int(raw_congress)
    number = int(raw_number)
    parsed_vote_date = _coerce_iso_date(vote_date)
    session_year = parsed_vote_date.year
    if raw_session is None:
        session_number = derive_session_number(congress, session_year)
    else:
        session_number = int(raw_session)

    # Normalize chamber from "h"/"s" to "House"/"Senate"
    chamber_normalized = normalize_vote_chamber(chamber)
    if chamber_normalized is None:
        return None

    if (congress, chamber_normalized, session_number, number) in existing_votes:
        return None

    raw_votes = data.get("votes", {})
    vote_counts = _count_votes(raw_votes)
    vote_records = _build_vote_records(raw_votes, legislator_map)

    return Vote(
        congress=congress,
        chamber=chamber_normalized,
        session_year=session_year,
        session_number=session_number,
        roll_number=number,
        vote_type=data.get("type"),
        question=data.get("question"),
        result=data.get("result"),
        result_text=data.get("result_text"),
        vote_date=parsed_vote_date,
        vote_datetime=parsed_vote_datetime(data),
        raw_vote_source_url=parse_vote_source_url(data),
        raw_bill_ref=coerce_raw_ref(data.get("bill")),
        raw_amendment_ref=coerce_raw_ref(data.get("amendment")),
        raw_nomination_ref=coerce_raw_ref(data.get("nomination")),
        raw_treaty_ref=coerce_raw_ref(data.get("treaty")),
        raw_vote_source_artifact_id=raw_vote_source_artifact_id,
        vote_records=vote_records,
        **vote_counts,
    )


def _count_votes(raw_votes: dict) -> dict[str, int]:
    """Count voters per position category, correctly handling dict and list formats."""
    yea_count = 0
    nay_count = 0
    not_voting_count = 0
    present_count = 0

    for position, position_group in raw_votes.items():
        voter_count = sum(1 for _ in _iter_voters(position_group))
        if position in ("Yea", "Aye"):
            yea_count += voter_count
        elif position in ("Nay", "No"):
            nay_count += voter_count
        elif position == "Not Voting":
            not_voting_count += voter_count
        elif position == "Present":
            present_count += voter_count

    return {
        "yea_count": yea_count,
        "nay_count": nay_count,
        "not_voting_count": not_voting_count,
        "present_count": present_count,
    }


def _build_vote_records(
    raw_votes: dict, legislator_map: dict[str, int]
) -> list[VoteRecord]:
    """Build VoteRecord objects from raw vote data."""
    records: list[VoteRecord] = []
    for position, position_group in raw_votes.items():
        for voter in _iter_voters(position_group):
            bioguide_id = voter.get("id")
            if not bioguide_id:
                continue
            legislator_id = legislator_map.get(bioguide_id)
            if legislator_id is None:
                continue
            records.append(
                VoteRecord(
                    legislator_id=legislator_id,
                    position=position,
                )
            )
    return records


def normalize_vote_chamber(raw_chamber: str | None) -> str | None:
    """Normalize vote JSON chamber codes."""
    if raw_chamber is None:
        return None
    value = raw_chamber.strip().lower()
    return {"h": "House", "house": "House", "s": "Senate", "senate": "Senate"}.get(
        value,
        raw_chamber,
    )


# ---------------------------------------------------------------------------
# Bill Text
# ---------------------------------------------------------------------------


def ingest_bill_text(
    engine: Engine,
    congress_dirs: list[Path],
    ingest_run_id: int | None,
) -> None:
    """Load bill text from text-versions directories with official metadata when available."""
    bill_map = _build_bill_map_for_engine(engine)
    with Session(engine) as session:
        logger.info("Loaded %d bills into lookup map", len(bill_map))
        billstatus_text_index = build_billstatus_text_version_index(congress_dirs)
        logger.info(
            "Loaded bill status text metadata for %d bills",
            len(billstatus_text_index),
        )
        existing_bill_texts = {
            (bill_text.bill_id, bill_text.version_code)
            for bill_text in session.scalars(select(BillText)).all()
        }
        logger.info(
            "Found %d existing bill text versions in DB", len(existing_bill_texts)
        )

        total_inserted = 0
        batch: list[BillText] = []
        for congress_dir in congress_dirs:
            logger.info("Scanning bill texts from %s", congress_dir.name)
            for bill_text in _iter_bill_texts(
                session,
                congress_dir,
                bill_map,
                existing_bill_texts,
                billstatus_text_index,
                ingest_run_id,
            ):
                batch.append(bill_text)
                if len(batch) >= BATCH_SIZE:
                    total_inserted += _flush_batch(session, batch, "bill texts")

        total_inserted += _flush_batch(session, batch, "bill texts")
    logger.info("Inserted %d new bill text versions total", total_inserted)


def _iter_bill_texts(
    session: Session,
    congress_dir: Path,
    bill_map: dict[tuple[int, str, int], int],
    existing_bill_texts: set[tuple[int, str]],
    billstatus_text_index: dict[tuple[int, str, int], dict[str, object]],
    ingest_run_id: int | None,
) -> Iterator[BillText]:
    """Yield BillText objects for a single congress directory, skipping existing."""
    bills_dir = congress_dir / "bills"
    if not bills_dir.is_dir():
        return

    tasks: list[dict[str, object]] = []
    for bill_dir in bills_dir.rglob("text-versions"):
        if not bill_dir.is_dir():
            continue
        bill_key = _bill_key_from_dir(bill_dir.parent, congress_dir)
        if bill_key is None:
            continue
        bill_id = bill_map.get(bill_key)
        if bill_id is None:
            continue

        for version_dir in sorted(bill_dir.iterdir()):
            if not version_dir.is_dir():
                continue
            if (bill_id, version_dir.name) in existing_bill_texts:
                continue
            tasks.append(
                {
                    "bill_id": bill_id,
                    "bill_key": bill_key,
                    "version_dir": version_dir,
                    "billstatus_version": billstatus_text_index.get(bill_key, {}).get(
                        version_dir.name.lower()
                    ),
                }
            )

    for chunk in _chunked(tasks, PARALLEL_FILE_CHUNK_SIZE):
        results = parallelize_thread(
            _prepare_bill_text_input,
            list(chunk),
            progress_tracker=PARALLEL_PROGRESS_TRACKER,
        )
        for prepared in results.results:
            source_artifact_id = None
            if prepared.source_file is not None:
                artifact = register_source_artifact(
                    session,
                    path=prepared.source_file,
                    source_kind="bill_text_artifact",
                    congress=prepared.bill_key[0],
                    chamber=None,
                    ingest_run_id=ingest_run_id,
                    source_url=None,
                )
                source_artifact_id = artifact.id
            metadata = _merge_bill_text_metadata(
                version_code=prepared.version_code,
                version_data=prepared.version_data,
                billstatus_version=prepared.billstatus_version,
            )
            yield BillText(
                bill_id=prepared.bill_id,
                version_code=prepared.version_code,
                version_name=metadata["version_name"],
                date=metadata["date"],
                text_content=prepared.text_content,
                source_datetime_raw=metadata["source_datetime_raw"],
                text_url_xml=metadata["text_url_xml"],
                text_url_pdf=metadata["text_url_pdf"],
                text_url_html=metadata["text_url_html"],
                source_artifact_id=source_artifact_id,
            )


def _bill_key_from_dir(
    bill_dir: Path, congress_dir: Path
) -> tuple[int, str, int] | None:
    """Extract (congress, bill_type, number) from directory structure."""
    congress = int(congress_dir.name)
    bill_type = bill_dir.parent.name
    name = bill_dir.name
    # Directory name is like "hr3590" — strip the type prefix to get the number
    number_str = name[len(bill_type) :]
    if not number_str.isdigit():
        return None
    return (congress, bill_type, int(number_str))


def _read_bill_text(version_dir: Path) -> str | None:
    """Read bill text from a version directory, preferring .txt over .xml."""
    for extension in ("txt", "htm", "html", "xml"):
        candidates = list(version_dir.glob(f"document.{extension}"))
        if not candidates:
            candidates = list(version_dir.glob(f"*.{extension}"))
        if candidates:
            try:
                return candidates[0].read_text(encoding="utf-8")
            except Exception:
                logger.exception("Failed to read %s", candidates[0])
    return None


def _find_text_source_file(version_dir: Path) -> Path | None:
    """Locate one representative local file for the bill text artifact manifest."""
    for extension in ("txt", "htm", "html", "xml"):
        candidates = list(version_dir.glob(f"document.{extension}"))
        if not candidates:
            candidates = list(version_dir.glob(f"*.{extension}"))
        if candidates:
            return candidates[0]
    return None


def _merge_bill_text_metadata(
    *,
    version_code: str,
    version_data: dict | None,
    billstatus_version: object | None,
) -> dict[str, object | None]:
    """Merge unitedstates/congress text-version metadata with official billstatus metadata."""
    version_name = None
    issued_on = None
    source_datetime_raw = None
    text_url_xml = None
    text_url_pdf = None
    text_url_html = None

    if version_data:
        version_name = version_data.get("version_name")
        issued_on = parse_date_like(version_data.get("issued_on"))
        urls = version_data.get("urls")
        if isinstance(urls, dict):
            text_url_xml = urls.get("xml") or urls.get("formatted_xml")
            text_url_pdf = urls.get("pdf")
            text_url_html = urls.get("html") or urls.get("formatted_text")

    if billstatus_version is not None:
        version_name = getattr(billstatus_version, "version_name", None) or version_name
        issued_on = getattr(billstatus_version, "version_date", None) or issued_on
        source_datetime_raw = (
            getattr(billstatus_version, "source_datetime_raw", None) or source_datetime_raw
        )
        text_url_xml = getattr(billstatus_version, "text_url_xml", None) or text_url_xml
        text_url_pdf = getattr(billstatus_version, "text_url_pdf", None) or text_url_pdf
        text_url_html = getattr(billstatus_version, "text_url_html", None) or text_url_html

    return {
        "version_code": version_code,
        "version_name": version_name,
        "date": issued_on,
        "source_datetime_raw": source_datetime_raw,
        "text_url_xml": text_url_xml,
        "text_url_pdf": text_url_pdf,
        "text_url_html": text_url_html,
    }


def _coerce_iso_date(value: str | date) -> date:
    """Normalize YYYY-MM-DD strings from the source data into date objects."""
    if isinstance(value, date):
        return value
    return date.fromisoformat(value[:10])


def ingest_bill_status_context_step(
    engine: Engine,
    congress_dirs: list[Path],
    ingest_run_id: int | None,
) -> None:
    """Rebuild official bill/amendment context from offline artifacts."""
    supported_congress_dirs = filter_context_supported_congress_dirs(congress_dirs)
    if not supported_congress_dirs:
        logger.info("No congress directories support offline BILLSTATUS context; skipping.")
        return
    require_billstatus_artifacts(supported_congress_dirs)
    bill_map = _build_bill_map_for_engine(engine)
    with Session(engine) as session:
        rebuild_bill_status_context(
            session,
            congress_dirs=supported_congress_dirs,
            bill_map=bill_map,
            ingest_run_id=ingest_run_id,
        )


def vote_context_match_step(engine: Engine, congress_dirs: list[Path]) -> None:
    """Persist canonical vote->action matches from recorded vote tuples."""
    supported_congress_dirs = filter_context_supported_congress_dirs(congress_dirs)
    if not supported_congress_dirs:
        logger.info("No congress directories support offline BILLSTATUS context; skipping.")
        return
    require_billstatus_artifacts(supported_congress_dirs)
    congress_numbers = [int(path.name) for path in supported_congress_dirs]
    with Session(engine) as session:
        build_vote_action_matches(session, congress_numbers=congress_numbers)


def vote_context_classify_step(engine: Engine, congress_dirs: list[Path]) -> None:
    """Classify votes and measure relationships after action matching."""
    supported_congress_dirs = filter_context_supported_congress_dirs(congress_dirs)
    if not supported_congress_dirs:
        logger.info("No congress directories support offline BILLSTATUS context; skipping.")
        return
    require_billstatus_artifacts(supported_congress_dirs)
    congress_numbers = [int(path.name) for path in supported_congress_dirs]
    bill_map = _build_bill_map_for_engine(engine)
    with Session(engine) as session:
        classify_votes(
            session,
            congress_numbers=congress_numbers,
            bill_map=bill_map,
        )


def vote_text_resolve_step(engine: Engine, congress_dirs: list[Path]) -> None:
    """Resolve official text targets for direct legislative text votes."""
    supported_congress_dirs = filter_context_supported_congress_dirs(congress_dirs)
    if not supported_congress_dirs:
        logger.info("No congress directories support offline BILLSTATUS context; skipping.")
        return
    require_billstatus_artifacts(supported_congress_dirs)
    congress_numbers = [int(path.name) for path in supported_congress_dirs]
    with Session(engine) as session:
        resolve_vote_text_targets(session, congress_numbers=congress_numbers)


def vote_polarity_resolve_step(engine: Engine, congress_dirs: list[Path]) -> None:
    """Resolve position polarity/effect metadata for all classified votes."""
    supported_congress_dirs = filter_context_supported_congress_dirs(congress_dirs)
    if not supported_congress_dirs:
        logger.info("No congress directories support offline BILLSTATUS context; skipping.")
        return
    require_billstatus_artifacts(supported_congress_dirs)
    congress_numbers = [int(path.name) for path in supported_congress_dirs]
    with Session(engine) as session:
        resolve_vote_position_meanings(session, congress_numbers=congress_numbers)


def vote_context_diagnostics_step(engine: Engine, congress_dirs: list[Path]) -> None:
    """Log aggregate vote-context coverage and unresolved audit rows."""
    supported_congress_dirs = filter_context_supported_congress_dirs(congress_dirs)
    if not supported_congress_dirs:
        logger.info("No congress directories support offline BILLSTATUS context; skipping.")
        return
    congress_numbers = [int(path.name) for path in supported_congress_dirs]
    with Session(engine) as session:
        classification_counts = session.execute(
            select(
                VoteClassification.subject_type,
                VoteClassification.vote_relationship,
                func.count(VoteClassification.vote_id),
            )
            .join(Vote, Vote.id == VoteClassification.vote_id)
            .where(Vote.congress.in_(congress_numbers))
            .group_by(
                VoteClassification.subject_type,
                VoteClassification.vote_relationship,
            )
            .order_by(
                VoteClassification.subject_type,
                VoteClassification.vote_relationship,
            )
        ).all()
        match_counts = session.execute(
            select(VoteActionMatch.match_method, func.count(VoteActionMatch.id))
            .join(Vote, Vote.id == VoteActionMatch.vote_id)
            .where(Vote.congress.in_(congress_numbers), VoteActionMatch.is_selected.is_(True))
            .group_by(VoteActionMatch.match_method)
            .order_by(VoteActionMatch.match_method)
        ).all()
        unresolved_audits = session.scalar(
            select(func.count(VoteContextAudit.id))
            .join(Vote, Vote.id == VoteContextAudit.vote_id)
            .where(Vote.congress.in_(congress_numbers), VoteContextAudit.severity.in_(("warning", "error")))
        )

    for subject_type, vote_relationship, count in classification_counts:
        logger.info(
            "vote-context subject=%s relationship=%s count=%d",
            subject_type,
            vote_relationship,
            count,
        )
    for match_method, count in match_counts:
        logger.info("vote-context selected_match method=%s count=%d", match_method, count)
    logger.info("vote-context unresolved audit rows=%d", unresolved_audits or 0)


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _read_json(path: Path) -> dict | None:
    """Read and parse a JSON file, returning None on failure."""
    try:
        return orjson.loads(path.read_bytes())
    except FileNotFoundError:
        return None
    except Exception:
        logger.exception("Failed to parse %s", path)
        return None


if __name__ == "__main__":
    app()