source_package.py

"""Declarative source-package loading for Arch build suites."""

from __future__ import annotations

import hashlib
import os
from dataclasses import dataclass, replace
from importlib.resources import files
from io import BytesIO
from pathlib import Path
import re
from typing import Any
from urllib.parse import unquote, urlparse
from zipfile import ZipFile

import httpx
import yaml

from arch.core import (
    Aggregation,
    AggregateConstraint,
    EntityDimension,
    GeographyDimension,
    Measure,
    PeriodDimension,
    SourceProvenance,
    AggregateFact,
    build_label,
)
from arch.sources.cells import (
    SourceArtifactMetadata,
    SourceCell,
    source_cells_from_delimited_text,
    source_cells_from_html_tables_and_text,
    source_cells_from_ods,
    source_cells_from_pdf_text_numbers,
    source_cells_from_xls,
    source_cells_from_xlsx,
)
from arch.sources.rows import (
    SourceRow,
    source_cells_from_source_rows,
    source_rows_from_census_acs_s0101_age_json,
    source_rows_from_census_acs_s2201_snap_json,
    source_rows_from_census_b01001_female_age_json,
    source_rows_from_cdc_vsrr_live_births_json,
    source_rows_from_json_table,
    source_rows_from_ees_permalink_table_html,
    source_rows_from_kff_state_indicator_gdocs_html,
    source_rows_from_ons_timeseries_json,
    source_rows_from_delimited_text,
)
from arch.sources.specs import (
    SourceRecord,
    SourceRecordSetMeasure,
    SourceRecordSetRangeLabelGuard,
    SourceRecordSetRow,
    SourceRecordSetRowGuard,
    SourceRecordSetSpec,
    SourceRecordSpec,
    build_cells_by_sheet_address,
    compile_source_record_set_specs,
    resolve_source_record,
    source_regions_from_record_set_spec,
)

SOURCE_PACKAGE_RESOURCE_PACKAGE = "packages"
SOURCE_PACKAGE_ALIASES = {
    "soi-table-1-1": Path("irs_soi/table_1_1"),
    "soi-table-1-2": Path("irs_soi/table_1_2"),
    "soi-table-1-4": Path("irs_soi/table_1_4"),
    "soi-table-2-1": Path("irs_soi/table_2_1"),
    "soi-table-2-5": Path("irs_soi/table_2_5"),
    "soi-table-4-3": Path("irs_soi/table_4_3"),
    "soi-w2-statistics-2020": Path("irs_soi/w2_statistics_2020"),
    "soi-ira-traditional-contributions-2022": Path(
        "irs_soi/ira_traditional_contributions_2022"
    ),
    "soi-ira-roth-contributions-2022": Path("irs_soi/ira_roth_contributions_2022"),
}
SOURCE_ARTIFACT_CACHE_ENV = "ARCH_SOURCE_ARTIFACT_CACHE_DIR"
SOURCE_ARTIFACT_FETCH_ENV = "ARCH_SOURCE_ARTIFACT_FETCH"
DEFAULT_SOURCE_ARTIFACT_CACHE_DIR = (
    Path.home() / ".cache" / "policyengine-arch-data" / "source-artifacts"
)
SOURCE_PACKAGE_FILENAME = "source_package.yaml"
EXCEL_COLUMN_RE = re.compile(r"^[A-Z]+$")


@dataclass(frozen=True)
class SourceArtifactSpec:
    """Declarative source artifact lookup and provenance metadata."""

    source_name: str
    source_table: str
    resource_package: str
    resource_directory: str
    manifest: str
    vintage: str
    extracted_at: str
    extraction_method: str
    parser: str = "xls_used_range"
    sheet_name: str | None = None
    archive_member: str | None = None
    artifact_year: int | None = None
    delimiter: str = ","
    selected_rows: tuple[dict[str, Any], ...] = ()

    def build_source_rows(self, year: int) -> list[SourceRow]:
        """Parse a delimited artifact for a year into full source-row records."""
        content, filename, source_url, raw_r2 = self._artifact_content(year)
        artifact = self._source_artifact_metadata(
            content,
            filename,
            source_url,
            raw_r2,
            year=year,
        )
        if self.parser in {"delimited_text_full_rows", "zip_delimited_text_full_rows"}:
            delimited_content = self._delimited_content(content, filename)
            return source_rows_from_delimited_text(
                delimited_content,
                artifact,
                sheet_name=self._sheet_name(filename, year=year),
                delimiter=self.delimiter,
            )
        if self.parser == "json_table_full_rows":
            return source_rows_from_json_table(
                content,
                artifact,
                sheet_name=self._sheet_name(filename, year=year),
            )
        if self.parser == "census_acs_s0101_age_json_rows":
            return source_rows_from_census_acs_s0101_age_json(
                content,
                artifact,
                sheet_name=self._sheet_name(filename, year=year),
            )
        if self.parser == "census_acs_s2201_snap_json_rows":
            return source_rows_from_census_acs_s2201_snap_json(
                content,
                artifact,
                sheet_name=self._sheet_name(filename, year=year),
            )
        if self.parser == "census_b01001_female_age_json_rows":
            return source_rows_from_census_b01001_female_age_json(
                content,
                artifact,
                sheet_name=self._sheet_name(filename, year=year),
            )
        if self.parser == "cdc_vsrr_live_births_json_rows":
            return source_rows_from_cdc_vsrr_live_births_json(
                content,
                artifact,
                sheet_name=self._sheet_name(filename, year=year),
            )
        if self.parser == "ons_timeseries_json_years":
            return source_rows_from_ons_timeseries_json(
                content,
                artifact,
                sheet_name=self.sheet_name or "years",
            )
        if self.parser == "ees_permalink_table_html":
            return source_rows_from_ees_permalink_table_html(
                content,
                artifact,
                sheet_name=self.sheet_name or "table",
            )
        if self.parser == "kff_state_indicator_gdocs_html_rows":
            return source_rows_from_kff_state_indicator_gdocs_html(
                content,
                artifact,
                sheet_name=self.sheet_name or "indicator",
            )
        return []

    def build_source_cells(
        self,
        year: int,
        *,
        source_rows: list[SourceRow] | None = None,
    ) -> list[SourceCell]:
        """Parse the artifact for a year into source-cell records."""
        content, filename, source_url, raw_r2 = self._artifact_content(year)
        artifact = self._source_artifact_metadata(
            content,
            filename,
            source_url,
            raw_r2,
            year=year,
        )
        if self.parser == "xls_used_range":
            return source_cells_from_xls(content, artifact)
        if self.parser == "xlsx_used_range":
            return source_cells_from_xlsx(content, artifact)
        if self.parser == "zip_xlsx_used_range":
            member_content, member_name = self._archive_member_content(
                content,
                suffixes=(".xlsx",),
            )
            member_sha256 = hashlib.sha256(member_content).hexdigest()
            return source_cells_from_xlsx(
                member_content,
                replace(
                    artifact,
                    source_file=f"{filename}!{member_name}",
                    sha256=member_sha256,
                    size_bytes=len(member_content),
                    extraction_method=(
                        f"{artifact.extraction_method}; ZIP member {member_name} "
                        f"from outer SHA-256 {artifact.sha256}"
                    ),
                ),
            )
        if self.parser == "zip_xls_used_range":
            member_content, member_name = self._archive_member_content(
                content,
                suffixes=(".xls",),
            )
            member_sha256 = hashlib.sha256(member_content).hexdigest()
            return source_cells_from_xls(
                member_content,
                replace(
                    artifact,
                    source_file=f"{filename}!{member_name}",
                    sha256=member_sha256,
                    size_bytes=len(member_content),
                    extraction_method=(
                        f"{artifact.extraction_method}; ZIP member {member_name} "
                        f"from outer SHA-256 {artifact.sha256}"
                    ),
                ),
            )
        if self.parser == "ods_used_range":
            return source_cells_from_ods(content, artifact)
        if self.parser == "html_tables_and_text":
            return source_cells_from_html_tables_and_text(content, artifact)
        if self.parser == "pdf_text_numbers":
            return source_cells_from_pdf_text_numbers(content, artifact)
        if self.parser == "delimited_text_selected_rows":
            return source_cells_from_delimited_text(
                content,
                artifact,
                sheet_name=self._sheet_name(filename, year=year),
                selected_rows=tuple(
                    {
                        key: str(_render_value(value, year=year))
                        for key, value in row.items()
                    }
                    for row in self.selected_rows
                ),
                delimiter=self.delimiter,
            )
        if self.parser in {"delimited_text_full_rows", "zip_delimited_text_full_rows"}:
            delimited_content = self._delimited_content(content, filename)
            rows = (
                source_rows
                if source_rows is not None
                else source_rows_from_delimited_text(
                    delimited_content,
                    artifact,
                    sheet_name=self._sheet_name(filename, year=year),
                    delimiter=self.delimiter,
                )
            )
            return source_cells_from_source_rows(
                rows,
                selected_rows=tuple(
                    {
                        key: str(_render_value(value, year=year))
                        for key, value in row.items()
                    }
                    for row in self.selected_rows
                ),
            )
        if self.parser == "json_table_full_rows":
            rows = (
                source_rows
                if source_rows is not None
                else source_rows_from_json_table(
                    content,
                    artifact,
                    sheet_name=self._sheet_name(filename, year=year),
                )
            )
            return source_cells_from_source_rows(
                rows,
                selected_rows=tuple(
                    {
                        key: str(_render_value(value, year=year))
                        for key, value in row.items()
                    }
                    for row in self.selected_rows
                ),
            )
        if self.parser == "census_acs_s0101_age_json_rows":
            rows = (
                source_rows
                if source_rows is not None
                else source_rows_from_census_acs_s0101_age_json(
                    content,
                    artifact,
                    sheet_name=self._sheet_name(filename, year=year),
                )
            )
            return source_cells_from_source_rows(
                rows,
                selected_rows=tuple(
                    {
                        key: str(_render_value(value, year=year))
                        for key, value in row.items()
                    }
                    for row in self.selected_rows
                ),
            )
        if self.parser == "census_b01001_female_age_json_rows":
            rows = (
                source_rows
                if source_rows is not None
                else source_rows_from_census_b01001_female_age_json(
                    content,
                    artifact,
                    sheet_name=self._sheet_name(filename, year=year),
                )
            )
            return source_cells_from_source_rows(
                rows,
                selected_rows=tuple(
                    {
                        key: str(_render_value(value, year=year))
                        for key, value in row.items()
                    }
                    for row in self.selected_rows
                ),
            )
        if self.parser == "census_acs_s2201_snap_json_rows":
            rows = (
                source_rows
                if source_rows is not None
                else source_rows_from_census_acs_s2201_snap_json(
                    content,
                    artifact,
                    sheet_name=self._sheet_name(filename, year=year),
                )
            )
            return source_cells_from_source_rows(
                rows,
                selected_rows=tuple(
                    {
                        key: str(_render_value(value, year=year))
                        for key, value in row.items()
                    }
                    for row in self.selected_rows
                ),
            )
        if self.parser == "cdc_vsrr_live_births_json_rows":
            rows = (
                source_rows
                if source_rows is not None
                else source_rows_from_cdc_vsrr_live_births_json(
                    content,
                    artifact,
                    sheet_name=self._sheet_name(filename, year=year),
                )
            )
            return source_cells_from_source_rows(
                rows,
                selected_rows=tuple(
                    {
                        key: str(_render_value(value, year=year))
                        for key, value in row.items()
                    }
                    for row in self.selected_rows
                ),
            )
        if self.parser == "ons_timeseries_json_years":
            rows = (
                source_rows
                if source_rows is not None
                else source_rows_from_ons_timeseries_json(
                    content,
                    artifact,
                    sheet_name=self.sheet_name or "years",
                )
            )
            return source_cells_from_source_rows(
                rows,
                selected_rows=tuple(
                    {
                        key: str(_render_value(value, year=year))
                        for key, value in row.items()
                    }
                    for row in self.selected_rows
                ),
            )
        if self.parser == "ees_permalink_table_html":
            rows = (
                source_rows
                if source_rows is not None
                else source_rows_from_ees_permalink_table_html(
                    content,
                    artifact,
                    sheet_name=self.sheet_name or "table",
                )
            )
            return source_cells_from_source_rows(
                rows,
                selected_rows=tuple(
                    {
                        key: str(_render_value(value, year=year))
                        for key, value in row.items()
                    }
                    for row in self.selected_rows
                ),
            )
        if self.parser == "kff_state_indicator_gdocs_html_rows":
            rows = (
                source_rows
                if source_rows is not None
                else source_rows_from_kff_state_indicator_gdocs_html(
                    content,
                    artifact,
                    sheet_name=self.sheet_name or "indicator",
                )
            )
            return source_cells_from_source_rows(
                rows,
                selected_rows=tuple(
                    {
                        key: str(_render_value(value, year=year))
                        for key, value in row.items()
                    }
                    for row in self.selected_rows
                ),
            )
        raise ValueError(f"Unsupported source artifact parser: {self.parser}")

    def _source_artifact_metadata(
        self,
        content: bytes,
        filename: str,
        source_url: str,
        raw_r2: dict[str, str],
        *,
        year: int,
    ) -> SourceArtifactMetadata:
        return SourceArtifactMetadata(
            source_name=self.source_name,
            source_table=_render_string(self.source_table, year=year),
            source_file=filename,
            url=source_url,
            vintage=_render_string(self.vintage, year=year),
            sha256=hashlib.sha256(content).hexdigest(),
            size_bytes=len(content),
            extracted_at=self.extracted_at,
            extraction_method=self.extraction_method,
            raw_r2_bucket=raw_r2.get("bucket"),
            raw_r2_key=raw_r2.get("key"),
            raw_r2_uri=raw_r2.get("uri"),
        )

    def _artifact_content(
        self,
        year: int,
    ) -> tuple[bytes, str, str, dict[str, str]]:
        manifest_path = files(self.resource_package).joinpath(
            self.resource_directory,
            self.manifest,
        )
        with manifest_path.open("r", encoding="utf-8") as file:
            manifest = yaml.safe_load(file)
        spec = _year_mapping(manifest["files"], self.artifact_year or year)
        artifact_path = files(self.resource_package).joinpath(
            self.resource_directory,
            spec["filename"],
        )
        content = _read_source_artifact_content(artifact_path, spec)
        expected_sha = spec.get("sha256")
        if expected_sha:
            _validate_source_artifact_sha(
                content,
                expected_sha=str(expected_sha),
                filename=str(spec["filename"]),
            )
        storage = spec.get("storage") if isinstance(spec, dict) else None
        raw_r2 = storage.get("r2") if isinstance(storage, dict) else {}
        return content, spec["filename"], spec["source_url"], raw_r2 or {}

    def _sheet_name(self, filename: str, *, year: int) -> str:
        if self.sheet_name:
            return _render_string(self.sheet_name, year=year)
        if self.archive_member:
            return self.archive_member
        return Path(filename).stem

    def _delimited_content(self, content: bytes, filename: str) -> bytes:
        if self.parser != "zip_delimited_text_full_rows":
            return content
        member, _member_name = self._archive_member_content(
            content,
            suffixes=(".csv", ".txt", ".tsv"),
        )
        return member

    def _archive_member_content(
        self,
        content: bytes,
        *,
        suffixes: tuple[str, ...],
    ) -> tuple[bytes, str]:
        with ZipFile(BytesIO(content)) as archive:
            member = self.archive_member or _single_archive_member(
                archive,
                suffixes=suffixes,
            )
            return archive.read(member), member


@dataclass(frozen=True)
class SourcePackageIssue:
    """One source-package authoring issue."""

    code: str
    message: str
    record_set_id: str | None = None
    row_id: str | None = None
    measure_id: str | None = None

    def to_dict(self) -> dict[str, Any]:
        """Return a JSON-serializable issue."""
        return {
            key: value
            for key, value in {
                "code": self.code,
                "message": self.message,
                "record_set_id": self.record_set_id,
                "row_id": self.row_id,
                "measure_id": self.measure_id,
            }.items()
            if value is not None
        }


@dataclass(frozen=True)
class SourcePackageValidationReport:
    """Validation report for one declarative source package."""

    package_id: str | None
    package_path: str
    year: int
    counts: dict[str, int]
    errors: tuple[SourcePackageIssue, ...]
    warnings: tuple[SourcePackageIssue, ...] = ()

    @property
    def valid(self) -> bool:
        """Whether the package has no validation errors."""
        return not self.errors

    def to_dict(self) -> dict[str, Any]:
        """Return a JSON-serializable report."""
        return {
            "valid": self.valid,
            "package_id": self.package_id,
            "package_path": self.package_path,
            "year": self.year,
            "counts": self.counts,
            "errors": [issue.to_dict() for issue in self.errors],
            "warnings": [issue.to_dict() for issue in self.warnings],
        }


@dataclass(frozen=True)
class SourcePackageScaffoldReport:
    """Report for a scaffolded source package."""

    package_id: str
    source_id: str
    package_path: str
    source_package_path: str
    replaced: bool

    def to_dict(self) -> dict[str, Any]:
        """Return a JSON-serializable report."""
        return {
            "package_id": self.package_id,
            "source_id": self.source_id,
            "package_path": self.package_path,
            "source_package_path": self.source_package_path,
            "replaced": self.replaced,
        }


@dataclass(frozen=True)
class DeclarativeRecordSet:
    """One YAML record-set payload that compiles to source-record specs."""

    payload: dict[str, Any]

    def to_record_set_spec(self, year: int) -> SourceRecordSetSpec:
        """Compile this YAML payload into the core record-set spec."""
        rows = tuple(
            _row_from_mapping(row, year=year)
            for row in _required(self.payload, "rows", "record_set")
        )
        measures = tuple(
            _measure_from_mapping(measure, year=year)
            for measure in _required(self.payload, "measures", "record_set")
        )
        return SourceRecordSetSpec(
            record_set_id=_render_required_string(
                self.payload,
                "record_set_id",
                year=year,
            ),
            record_set_spec_id=_required(
                self.payload,
                "record_set_spec_id",
                "record_set",
            ),
            source_record_id_prefix=_render_required_string(
                self.payload,
                "source_record_id_prefix",
                year=year,
            ),
            sheet_name=_record_set_sheet_name_from_mapping(
                self.payload,
                year=year,
            ),
            period_type=_required(self.payload, "period_type", "record_set"),
            period=_record_set_period_from_mapping(self.payload, year=year),
            geography_id=_required(self.payload, "geography_id", "record_set"),
            geography_level=_required(
                self.payload,
                "geography_level",
                "record_set",
            ),
            geography_name=self.payload.get("geography_name"),
            geography_vintage=self.payload.get("geography_vintage"),
            entity=_required(self.payload, "entity", "record_set"),
            entity_role=self.payload.get("entity_role"),
            domain=_required(self.payload, "domain", "record_set"),
            groupby_dimension=_required(
                self.payload,
                "groupby_dimension",
                "record_set",
            ),
            rows=rows,
            measures=measures,
            shared_filters={
                key: _render_value(value, year=year)
                for key, value in self.payload.get("shared_filters", {}).items()
            },
            shared_constraints=tuple(
                _constraint_from_mapping(constraint, year=year)
                for constraint in self.payload.get("shared_constraints", ())
            ),
        )


@dataclass(frozen=True)
class SourcePackage:
    """A declarative Arch source package."""

    package_id: str
    label: str | None
    artifact: SourceArtifactSpec
    record_sets: tuple[DeclarativeRecordSet, ...]
    package_path: Path

    def build_source_rows(self, year: int) -> list[SourceRow]:
        """Build full source rows for row-oriented artifacts."""
        return self.artifact.build_source_rows(year)

    def build_source_cells(
        self,
        year: int,
        *,
        source_rows: list[SourceRow] | None = None,
    ) -> list[SourceCell]:
        """Build whole-artifact source cells for this package."""
        return self.artifact.build_source_cells(year, source_rows=source_rows)

    def build_source_record_set_specs(
        self,
        year: int,
    ) -> list[SourceRecordSetSpec]:
        """Build compact record-set specs for this package."""
        return [record_set.to_record_set_spec(year) for record_set in self.record_sets]

    def build_source_regions(self, year: int):
        """Build source-region specs implied by the package record sets."""
        regions = []
        for spec in self.build_source_record_set_specs(year):
            regions.extend(source_regions_from_record_set_spec(spec))
        return regions

    def build_source_record_specs(self, year: int) -> list[SourceRecordSpec]:
        """Build atomic source-record specs for this package."""
        specs: list[SourceRecordSpec] = []
        for record_set in self.build_source_record_set_specs(year):
            specs.extend(compile_source_record_set_specs(record_set))
        return specs

    def build_source_records(
        self,
        year: int,
        *,
        cells: list[SourceCell] | None = None,
        source_rows: list[SourceRow] | None = None,
    ) -> list[SourceRecord]:
        """Resolve package source-record specs against source cells."""
        if cells is None:
            cells = self.build_source_cells(year, source_rows=source_rows)
        cells_by_sheet_address = build_cells_by_sheet_address(cells)
        return [
            resolve_source_record(
                cells,
                spec,
                cells_by_sheet_address=cells_by_sheet_address,
            )
            for spec in self.build_source_record_specs(year)
        ]

    def build_facts(
        self,
        year: int,
        *,
        cells: list[SourceCell] | None = None,
        source_rows: list[SourceRow] | None = None,
    ) -> list[AggregateFact]:
        """Build source-lineaged Arch aggregate facts."""
        if cells is None:
            cells = self.build_source_cells(year, source_rows=source_rows)
        source = _source_provenance_from_cells(cells)
        facts = []
        for record in self.build_source_records(
            year,
            cells=cells,
            source_rows=source_rows,
        ):
            fact = _fact_from_source_record(record, source)
            facts.append(replace(fact, label=build_label(fact)))
        return facts


def load_source_package(source: str | Path) -> SourcePackage:
    """Load a declarative source package from an alias, directory, or YAML file."""
    path = resolve_source_package_path(source)
    with path.open("r", encoding="utf-8") as file:
        payload = yaml.safe_load(file)
    schema_version = _required(payload, "schema_version", str(path))
    if schema_version != "arch.source_package.v1":
        raise ValueError(f"Unsupported source package schema: {schema_version}")
    package_dir = path.parent
    return SourcePackage(
        package_id=_required(payload, "package_id", str(path)),
        label=payload.get("label"),
        artifact=_artifact_from_mapping(
            _required(payload, "artifact", str(path)),
        ),
        record_sets=tuple(
            DeclarativeRecordSet(record_set)
            for record_set in _required(payload, "record_sets", str(path))
        ),
        package_path=package_dir,
    )


def try_load_source_package(source: str | Path) -> SourcePackage | None:
    """Return a declarative package if the source reference resolves to one."""
    try:
        return load_source_package(source)
    except FileNotFoundError:
        return None


def validate_source_package(
    source: str | Path,
    *,
    year: int = 2023,
) -> SourcePackageValidationReport:
    """Validate declarative package structure before a full build-suite run."""
    package_path = _safe_package_path(source)
    errors: list[SourcePackageIssue] = []
    warnings: list[SourcePackageIssue] = []
    counts = {
        "record_set_count": 0,
        "row_count": 0,
        "measure_count": 0,
        "source_record_count": 0,
        "source_region_count": 0,
    }

    try:
        package = load_source_package(source)
    except (FileNotFoundError, KeyError, TypeError, ValueError) as exc:
        errors.append(
            SourcePackageIssue(
                code="source_package_load_failed",
                message=str(exc),
            )
        )
        return SourcePackageValidationReport(
            package_id=None,
            package_path=str(package_path),
            year=year,
            counts=counts,
            errors=tuple(errors),
        )

    try:
        package.artifact._artifact_content(year)
    except (FileNotFoundError, KeyError, OSError, ValueError) as exc:
        errors.append(
            SourcePackageIssue(
                code="source_artifact_unavailable",
                message=str(exc),
            )
        )

    try:
        record_sets = package.build_source_record_set_specs(year)
    except (KeyError, TypeError, ValueError) as exc:
        errors.append(
            SourcePackageIssue(
                code="record_set_compile_failed",
                message=str(exc),
            )
        )
        return SourcePackageValidationReport(
            package_id=package.package_id,
            package_path=str(package.package_path),
            year=year,
            counts=counts,
            errors=tuple(errors),
            warnings=tuple(warnings),
        )

    counts["record_set_count"] = len(record_sets)
    record_set_ids: dict[str, list[int]] = {}
    source_record_ids: dict[str, list[int]] = {}
    for record_set_index, record_set in enumerate(record_sets):
        record_set_ids.setdefault(record_set.record_set_id, []).append(record_set_index)
        counts["row_count"] += len(record_set.rows)
        counts["measure_count"] += len(record_set.measures)
        errors.extend(_validate_record_set_authoring(record_set))
        try:
            counts["source_region_count"] += len(
                source_regions_from_record_set_spec(record_set)
            )
        except ValueError as exc:
            errors.append(
                SourcePackageIssue(
                    code="source_region_compile_failed",
                    message=str(exc),
                    record_set_id=record_set.record_set_id,
                )
            )
        try:
            specs = compile_source_record_set_specs(record_set)
        except (KeyError, TypeError, ValueError) as exc:
            errors.append(
                SourcePackageIssue(
                    code="source_record_compile_failed",
                    message=str(exc),
                    record_set_id=record_set.record_set_id,
                )
            )
            continue
        for spec_index, spec in enumerate(specs):
            source_record_ids.setdefault(spec.source_record_id, []).append(spec_index)
            counts["source_record_count"] += 1

    for record_set_id, indices in record_set_ids.items():
        if len(indices) > 1:
            errors.append(
                SourcePackageIssue(
                    code="duplicate_record_set_id",
                    message=f"Duplicate record-set ID appears at indices {indices}.",
                    record_set_id=record_set_id,
                )
            )
    for source_record_id, indices in source_record_ids.items():
        if len(indices) > 1:
            errors.append(
                SourcePackageIssue(
                    code="duplicate_source_record_id",
                    message=(
                        "Duplicate compiled source-record ID appears at "
                        f"indices {indices}."
                    ),
                    record_set_id=source_record_id.rsplit(".", 2)[0],
                )
            )

    return SourcePackageValidationReport(
        package_id=package.package_id,
        package_path=str(package.package_path),
        year=year,
        counts=counts,
        errors=tuple(errors),
        warnings=tuple(warnings),
    )


def scaffold_source_package(
    output_dir: str | Path,
    *,
    source_id: str,
    package_id: str,
    source_table: str | None = None,
    resource_package: str = "db",
    resource_directory: str | None = None,
    manifest: str = "manifest.yaml",
    replace_existing: bool = False,
) -> SourcePackageScaffoldReport:
    """Write a starter declarative source package."""
    package_path = Path(output_dir)
    source_package_path = package_path / SOURCE_PACKAGE_FILENAME
    replaced = source_package_path.exists()
    if replaced and not replace_existing:
        raise FileExistsError(f"Source package already exists: {source_package_path}")
    package_path.mkdir(parents=True, exist_ok=True)
    source_package_path.write_text(
        _scaffold_template(
            source_id=source_id,
            package_id=package_id,
            source_table=source_table or "TODO source table title",
            resource_package=resource_package,
            resource_directory=(resource_directory or f"data/{source_id}/TODO_table"),
            manifest=manifest,
        ),
        encoding="utf-8",
    )
    return SourcePackageScaffoldReport(
        package_id=package_id,
        source_id=source_id,
        package_path=str(package_path),
        source_package_path=str(source_package_path),
        replaced=replaced,
    )


def resolve_source_package_path(source: str | Path) -> Path:
    """Resolve a package alias, directory, or YAML file to source_package.yaml."""
    source_path = Path(source)
    if source_path.exists():
        if source_path.is_dir():
            return source_path / SOURCE_PACKAGE_FILENAME
        return source_path
    alias = SOURCE_PACKAGE_ALIASES.get(str(source))
    if alias is None:
        raise FileNotFoundError(f"Source package not found: {source}")
    repo_path = _repo_root() / "packages" / alias / SOURCE_PACKAGE_FILENAME
    if repo_path.exists():
        return repo_path
    try:
        resource_path = files(SOURCE_PACKAGE_RESOURCE_PACKAGE).joinpath(
            alias,
            SOURCE_PACKAGE_FILENAME,
        )
        with resource_path.open("rb"):
            pass
        return Path(str(resource_path))
    except (FileNotFoundError, ModuleNotFoundError):
        raise FileNotFoundError(f"Source package not found: {source}") from None


def _validate_record_set_authoring(
    record_set: SourceRecordSetSpec,
) -> list[SourcePackageIssue]:
    errors: list[SourcePackageIssue] = []
    row_ids: dict[str, list[int]] = {}
    row_ordinals: dict[int, list[int]] = {}
    measure_ids: dict[str, list[int]] = {}
    measure_ordinals: dict[int, list[int]] = {}
    for index, row in enumerate(record_set.rows):
        row_ids.setdefault(row.value_id, []).append(index)
        row_ordinals.setdefault(row.ordinal, []).append(index)
        if _row_needs_constraints(row) and not row.constraints:
            errors.append(
                SourcePackageIssue(
                    code="missing_row_constraints",
                    message=(
                        "Detail row has semantic filters but no first-class "
                        "constraints."
                    ),
                    record_set_id=record_set.record_set_id,
                    row_id=row.value_id,
                )
            )
        errors.extend(_missing_filter_bound_constraints(record_set, row))

    for index, measure in enumerate(record_set.measures):
        measure_ids.setdefault(measure.measure_id, []).append(index)
        measure_ordinals.setdefault(measure.ordinal, []).append(index)
        if not EXCEL_COLUMN_RE.match(measure.column):
            errors.append(
                SourcePackageIssue(
                    code="malformed_measure_column",
                    message=(
                        "Measure column must be an Excel column name like B or AA."
                    ),
                    record_set_id=record_set.record_set_id,
                    measure_id=measure.measure_id,
                )
            )
        if measure.concept_relation == "exact" and not (
            measure.concept_evidence_url or measure.concept_evidence_notes
        ):
            errors.append(
                SourcePackageIssue(
                    code="missing_concept_evidence",
                    message="Exact concept alignments need evidence notes or a URL.",
                    record_set_id=record_set.record_set_id,
                    measure_id=measure.measure_id,
                )
            )
        if measure.source_concept and not measure.concept_relation:
            errors.append(
                SourcePackageIssue(
                    code="missing_concept_relation",
                    message="source_concept requires concept_relation.",
                    record_set_id=record_set.record_set_id,
                    measure_id=measure.measure_id,
                )
            )
        if measure.concept_relation and not measure.source_concept:
            errors.append(
                SourcePackageIssue(
                    code="missing_source_concept",
                    message="concept_relation requires source_concept.",
                    record_set_id=record_set.record_set_id,
                    measure_id=measure.measure_id,
                )
            )

    errors.extend(
        _duplicate_issues(
            row_ids,
            code="duplicate_row_id",
            message="Duplicate row ID appears within a record set.",
            record_set_id=record_set.record_set_id,
            row=True,
        )
    )
    errors.extend(
        _duplicate_ordinal_issues(
            row_ordinals,
            code="duplicate_row_ordinal",
            message="Duplicate row ordinal appears within a record set.",
            record_set_id=record_set.record_set_id,
            row=True,
        )
    )
    errors.extend(
        _duplicate_issues(
            measure_ids,
            code="duplicate_measure_id",
            message="Duplicate measure ID appears within a record set.",
            record_set_id=record_set.record_set_id,
            row=False,
        )
    )
    errors.extend(
        _duplicate_ordinal_issues(
            measure_ordinals,
            code="duplicate_measure_ordinal",
            message="Duplicate measure ordinal appears within a record set.",
            record_set_id=record_set.record_set_id,
            row=False,
        )
    )
    return errors


def _row_needs_constraints(row: SourceRecordSetRow) -> bool:
    if row.table_record_kind == "total":
        return False
    return any(
        key not in {"filing_status", "income_range"}
        for key, value in row.filters.items()
        if value not in (None, "all")
    )


def _missing_filter_bound_constraints(
    record_set: SourceRecordSetSpec,
    row: SourceRecordSetRow,
) -> list[SourcePackageIssue]:
    errors: list[SourcePackageIssue] = []
    for suffix, operator in (("_lower_usd", ">="), ("_upper_usd", "<")):
        for filter_key, value in row.filters.items():
            if not filter_key.endswith(suffix):
                continue
            if not any(
                constraint.operator == operator and constraint.value == value
                for constraint in row.constraints
            ):
                errors.append(
                    SourcePackageIssue(
                        code="missing_bound_constraint",
                        message=(
                            f"Filter {filter_key}={value!r} needs a matching "
                            f"{operator} constraint."
                        ),
                        record_set_id=record_set.record_set_id,
                        row_id=row.value_id,
                    )
                )
    return errors


def _duplicate_issues(
    indices_by_value: dict[str, list[int]],
    *,
    code: str,
    message: str,
    record_set_id: str,
    row: bool,
) -> list[SourcePackageIssue]:
    issues = []
    for value, indices in indices_by_value.items():
        if len(indices) <= 1:
            continue
        issues.append(
            SourcePackageIssue(
                code=code,
                message=f"{message} Indices: {indices}.",
                record_set_id=record_set_id,
                row_id=value if row else None,
                measure_id=None if row else value,
            )
        )
    return issues


def _duplicate_ordinal_issues(
    indices_by_value: dict[int, list[int]],
    *,
    code: str,
    message: str,
    record_set_id: str,
    row: bool,
) -> list[SourcePackageIssue]:
    issues = []
    for value, indices in indices_by_value.items():
        if len(indices) <= 1:
            continue
        issue = SourcePackageIssue(
            code=code,
            message=f"{message} Ordinal: {value}; indices: {indices}.",
            record_set_id=record_set_id,
            row_id=str(value) if row else None,
            measure_id=None if row else str(value),
        )
        issues.append(issue)
    return issues


def _artifact_from_mapping(payload: dict[str, Any]) -> SourceArtifactSpec:
    return SourceArtifactSpec(
        source_name=_required(payload, "source_name", "artifact"),
        source_table=_required(payload, "source_table", "artifact"),
        resource_package=_required(payload, "resource_package", "artifact"),
        resource_directory=_required(payload, "resource_directory", "artifact"),
        manifest=_required(payload, "manifest", "artifact"),
        vintage=_required(payload, "vintage", "artifact"),
        extracted_at=_required(payload, "extracted_at", "artifact"),
        extraction_method=_required(payload, "extraction_method", "artifact"),
        parser=payload.get("parser", "xls_used_range"),
        sheet_name=payload.get("sheet_name"),
        archive_member=payload.get("archive_member"),
        artifact_year=(
            int(payload["artifact_year"])
            if payload.get("artifact_year") is not None
            else None
        ),
        delimiter=payload.get("delimiter", ","),
        selected_rows=tuple(payload.get("selected_rows", ())),
    )


def _row_from_mapping(payload: dict[str, Any], *, year: int) -> SourceRecordSetRow:
    return SourceRecordSetRow(
        value_id=_required(payload, "value_id", "row"),
        label=_required(payload, "label", "row"),
        ordinal=int(_required(payload, "ordinal", "row")),
        row_number=int(_required(payload, "row_number", "row")),
        row_end_number=(
            int(payload["row_end_number"])
            if payload.get("row_end_number") is not None
            else None
        ),
        geography_id=payload.get("geography_id"),
        geography_level=payload.get("geography_level"),
        geography_name=payload.get("geography_name"),
        geography_vintage=payload.get("geography_vintage"),
        filters={
            key: _render_value(value, year=year)
            for key, value in payload.get("filters", {}).items()
        },
        constraints=tuple(
            _constraint_from_mapping(constraint, year=year)
            for constraint in payload.get("constraints", ())
        ),
        value_scale=_render_value(payload.get("value_scale", 1), year=year),
        source_row_id=payload.get("source_row_id"),
        table_record_kind=payload.get("table_record_kind", "detail"),
        expected_row_header=_render_value(
            payload["expected_row_header"],
            year=year,
        )
        if "expected_row_header" in payload
        else None,
        expected_row_header_column=payload.get("expected_row_header_column"),
        guard_cells=tuple(
            _row_guard_from_mapping(guard, year=year)
            for guard in payload.get("guard_cells", ())
        ),
        range_label_guards=tuple(
            _range_label_guard_from_mapping(guard, year=year)
            for guard in payload.get("range_label_guards", ())
        ),
    )


def _row_guard_from_mapping(
    payload: dict[str, Any],
    *,
    year: int,
) -> SourceRecordSetRowGuard:
    unknown_keys = set(payload) - {"column", "expected_value", "row", "label"}
    if unknown_keys:
        unknown = ", ".join(sorted(unknown_keys))
        raise ValueError(f"Unknown row_guard field(s): {unknown}")
    column = str(_required(payload, "column", "row_guard")).upper()
    if EXCEL_COLUMN_RE.fullmatch(column) is None:
        raise ValueError(f"Malformed row_guard column: {column!r}")
    return SourceRecordSetRowGuard(
        column=column,
        expected_value=_render_value(
            _required(payload, "expected_value", "row_guard"),
            year=year,
        ),
        row=_render_value(payload.get("row", "start"), year=year),
        label=payload.get("label", "row guard"),
    )


def _range_label_guard_from_mapping(
    payload: dict[str, Any],
    *,
    year: int,
) -> SourceRecordSetRangeLabelGuard:
    unknown_keys = set(payload) - {"column", "expected_values", "label"}
    if unknown_keys:
        unknown = ", ".join(sorted(unknown_keys))
        raise ValueError(f"Unknown range_label_guard field(s): {unknown}")
    column = str(_required(payload, "column", "range_label_guard")).upper()
    if EXCEL_COLUMN_RE.fullmatch(column) is None:
        raise ValueError(f"Malformed range_label_guard column: {column!r}")
    expected_values = _range_label_values_from_payload(
        _required(payload, "expected_values", "range_label_guard"),
        year=year,
    )
    _validate_range_label_values(expected_values)
    return SourceRecordSetRangeLabelGuard(
        column=column,
        expected_values=expected_values,
        label=payload.get("label", "range label sequence"),
    )


def _range_label_values_from_payload(
    payload: Any,
    *,
    year: int,
) -> tuple[Any, ...]:
    if isinstance(payload, list):
        return tuple(_render_value(value, year=year) for value in payload)
    if not isinstance(payload, dict):
        raise ValueError("range_label_guard expected_values must be a list or mapping")
    unknown_keys = set(payload) - {"integer_range", "parts"}
    if unknown_keys:
        unknown = ", ".join(sorted(unknown_keys))
        raise ValueError(
            f"Unknown range_label_guard expected_values field(s): {unknown}"
        )
    compact_forms = {"integer_range", "parts"} & set(payload)
    if len(compact_forms) > 1:
        forms = ", ".join(sorted(compact_forms))
        raise ValueError(
            "range_label_guard expected_values must use exactly one compact "
            f"form; got {forms}"
        )
    if "integer_range" in payload:
        return _integer_range_values(payload["integer_range"], year=year)
    if "parts" in payload:
        values = []
        for part in payload["parts"]:
            values.extend(_range_label_values_from_payload(part, year=year))
        return tuple(values)
    raise ValueError("range_label_guard expected_values mapping is empty")


def _validate_range_label_values(values: tuple[Any, ...]) -> None:
    if any(value is None for value in values):
        raise ValueError("range_label_guard expected_values must not contain null")


def _integer_range_values(payload: dict[str, Any], *, year: int) -> tuple[Any, ...]:
    unknown_keys = set(payload) - {"start", "end", "final_value", "extra_values"}
    if unknown_keys:
        unknown = ", ".join(sorted(unknown_keys))
        raise ValueError(f"Unknown integer_range field(s): {unknown}")
    start = int(_render_value(_required(payload, "start", "integer_range"), year=year))
    end = int(_render_value(_required(payload, "end", "integer_range"), year=year))
    if end < start:
        raise ValueError("integer_range end must be greater than or equal to start")
    values: list[Any] = list(range(start, end + 1))
    if "final_value" in payload:
        values[-1] = _render_value(payload["final_value"], year=year)
    if "extra_values" in payload:
        extra_values = payload["extra_values"]
        if not isinstance(extra_values, list):
            extra_values = [extra_values]
        values.extend(_render_value(value, year=year) for value in extra_values)
    return tuple(values)


def _measure_from_mapping(
    payload: dict[str, Any],
    *,
    year: int,
) -> SourceRecordSetMeasure:
    return SourceRecordSetMeasure(
        measure_id=_required(payload, "measure_id", "measure"),
        label=_required(payload, "label", "measure"),
        ordinal=int(_required(payload, "ordinal", "measure")),
        column=_measure_column_from_mapping(payload, year=year),
        concept=_required(payload, "concept", "measure"),
        unit=_required(payload, "unit", "measure"),
        aggregation=_required(payload, "aggregation", "measure"),
        value_scale=_render_value(payload.get("value_scale", 1), year=year),
        source_column_id=(
            str(_render_value(payload["source_column_id"], year=year))
            if payload.get("source_column_id") is not None
            else None
        ),
        expected_cell_type=payload.get("expected_cell_type", "number"),
        expected_column_header_row=(
            int(payload["expected_column_header_row"])
            if payload.get("expected_column_header_row") is not None
            else None
        ),
        expected_column_header=_render_value(
            _year_mapping(payload["expected_column_header_by_year"], year)
            if "expected_column_header_by_year" in payload
            else payload["expected_column_header"],
            year=year,
        )
        if "expected_column_header" in payload
        or "expected_column_header_by_year" in payload
        else None,
        source_concept=payload.get("source_concept"),
        concept_relation=payload.get("concept_relation"),
        concept_authority=payload.get("concept_authority"),
        concept_evidence_url=payload.get("concept_evidence_url"),
        concept_evidence_notes=_optional_rendered_string(
            payload.get("concept_evidence_notes"),
            year=year,
        ),
        legal_vintage=(
            _render_string(payload["legal_vintage"], year=year)
            if payload.get("legal_vintage")
            else None
        ),
        filters={
            key: _render_value(value, year=year)
            for key, value in payload.get("filters", {}).items()
        },
        constraints=tuple(
            _constraint_from_mapping(constraint, year=year)
            for constraint in payload.get("constraints", ())
        ),
    )


def _measure_column_from_mapping(payload: dict[str, Any], *, year: int) -> str:
    if "column_by_year" in payload:
        return str(_year_mapping(payload["column_by_year"], year))
    return str(_required(payload, "column", "measure"))


def _record_set_period_from_mapping(
    payload: dict[str, Any],
    *,
    year: int,
) -> int | str:
    if "period_by_year" in payload:
        return _period_value(_year_mapping(payload["period_by_year"], year))
    return _period_value(
        _render_value(
            _required(payload, "period", "record_set"),
            year=year,
        )
    )


def _record_set_sheet_name_from_mapping(
    payload: dict[str, Any],
    *,
    year: int,
) -> str:
    if "sheet_name_by_year" in payload:
        return str(_year_mapping(payload["sheet_name_by_year"], year))
    return _render_string(
        str(_required(payload, "sheet_name", "record_set")), year=year
    )


def _period_value(value: Any) -> int | str:
    if isinstance(value, int):
        return value
    if isinstance(value, str):
        stripped = value.strip()
        if stripped.isdigit():
            return int(stripped)
        return stripped
    raise ValueError(f"Unsupported period value: {value!r}")


def _constraint_from_mapping(
    payload: dict[str, Any],
    *,
    year: int,
) -> AggregateConstraint:
    return AggregateConstraint(
        variable=_required(payload, "variable", "constraint"),
        operator=_required(payload, "operator", "constraint"),
        value=_render_value(_required(payload, "value", "constraint"), year=year),
        unit=payload.get("unit"),
        role=payload.get("role", "filter"),
        label=payload.get("label"),
    )


def _fact_from_source_record(
    record: SourceRecord,
    source: SourceProvenance,
) -> AggregateFact:
    spec = record.spec
    return AggregateFact(
        value=record.value,
        period=PeriodDimension(type=spec.period_type, value=spec.period),
        geography=GeographyDimension(
            level=spec.geography_level,
            id=spec.geography_id,
            vintage=spec.geography_vintage,
            name=spec.geography_name,
        ),
        entity=EntityDimension(name=spec.entity, role=spec.entity_role),
        measure=Measure(
            concept=spec.concept,
            unit=spec.unit,
            source_concept=spec.source_concept,
            concept_relation=spec.concept_relation,
            concept_authority=spec.concept_authority,
            concept_evidence_url=spec.concept_evidence_url,
            concept_evidence_notes=spec.concept_evidence_notes,
            legal_vintage=spec.legal_vintage,
        ),
        aggregation=Aggregation(method=spec.aggregation),
        source=source,
        filters=spec.filters,
        domain=spec.domain,
        source_record_id=record.source_record_id,
        source_cell_keys=record.source_cell_keys,
        source_row_keys=record.source_row_keys,
        constraints=spec.constraints,
        layout=spec.layout,
    )


def _source_provenance_from_cells(cells: list[SourceCell]) -> SourceProvenance:
    artifact = cells[0].artifact
    return SourceProvenance(
        source_name=artifact.source_name,
        source_table=artifact.source_table,
        source_file=artifact.source_file,
        url=artifact.url,
        vintage=artifact.vintage,
        extracted_at=artifact.extracted_at,
        extraction_method="CellSelectorSpec and SourceRecordSpec resolved from cells",
        method_notes=(
            "Each fact carries source_record_id, source_cell_keys, and "
            "source_row_keys when available."
        ),
        source_sha256=artifact.sha256,
        source_size_bytes=artifact.size_bytes,
        raw_r2_bucket=artifact.raw_r2_bucket,
        raw_r2_key=artifact.raw_r2_key,
        raw_r2_uri=artifact.raw_r2_uri,
    )


def _render_required_string(
    payload: dict[str, Any],
    key: str,
    *,
    year: int,
) -> str:
    return _render_string(_required(payload, key, "record_set"), year=year)


def _render_string(value: str, *, year: int) -> str:
    return value.format(year=year, filing_year=year + 1)


def _optional_rendered_string(value: Any, *, year: int) -> str | None:
    if value is None:
        return None
    return _render_string(str(value), year=year).strip()


def _render_value(value: Any, *, year: int) -> Any:
    if isinstance(value, str):
        rendered = _render_string(value, year=year)
        return int(rendered) if rendered.isdigit() else rendered
    if isinstance(value, list):
        return [_render_value(item, year=year) for item in value]
    if isinstance(value, dict):
        return {key: _render_value(item, year=year) for key, item in value.items()}
    return value


def _required(payload: dict[str, Any], key: str, context: str) -> Any:
    if key not in payload:
        raise ValueError(f"Missing required {context} field: {key}")
    return payload[key]


def _year_mapping(files_by_year: dict[Any, Any], year: int) -> dict[str, str]:
    if year in files_by_year:
        return files_by_year[year]
    if str(year) in files_by_year:
        return files_by_year[str(year)]
    raise ValueError(f"No source artifact for year {year}")


def _read_source_artifact_content(
    artifact_path: Any,
    spec: dict[str, Any],
) -> bytes:
    """Read a source artifact from package data, cache, or explicit fetch."""
    try:
        return artifact_path.read_bytes()
    except FileNotFoundError:
        pass

    cache_path = _source_artifact_cache_path(spec)
    if cache_path.exists():
        return cache_path.read_bytes()

    if not _truthy_env(SOURCE_ARTIFACT_FETCH_ENV):
        raise FileNotFoundError(
            f"Source artifact {spec['filename']} is not packaged and was not "
            f"found in {cache_path}. Set {SOURCE_ARTIFACT_FETCH_ENV}=1 to fetch "
            "and cache missing source artifacts."
        )

    content = _fetch_source_artifact_content(str(spec["source_url"]))
    expected_sha = spec.get("sha256")
    if expected_sha:
        _validate_source_artifact_sha(
            content,
            expected_sha=str(expected_sha),
            filename=str(spec["filename"]),
        )
    cache_path.parent.mkdir(parents=True, exist_ok=True)
    cache_path.write_bytes(content)
    return content


def _source_artifact_cache_path(spec: dict[str, Any]) -> Path:
    cache_root = Path(
        os.environ.get(SOURCE_ARTIFACT_CACHE_ENV, DEFAULT_SOURCE_ARTIFACT_CACHE_DIR)
    )
    source_identity = str(spec.get("source_url", spec["filename"])).encode()
    identifier = str(spec.get("sha256") or hashlib.sha256(source_identity).hexdigest())
    return cache_root / identifier / str(spec["filename"])


def _fetch_source_artifact_content(source_url: str) -> bytes:
    parsed = urlparse(source_url)
    if parsed.scheme == "file":
        return Path(unquote(parsed.path)).read_bytes()
    if parsed.scheme == "":
        return Path(source_url).read_bytes()
    if parsed.scheme not in {"http", "https"}:
        raise ValueError(f"Unsupported source artifact URL scheme: {parsed.scheme}")
    response = httpx.get(source_url, follow_redirects=True, timeout=60)
    response.raise_for_status()
    return response.content


def _validate_source_artifact_sha(
    content: bytes,
    *,
    expected_sha: str,
    filename: str,
) -> None:
    actual_sha = hashlib.sha256(content).hexdigest()
    if actual_sha != expected_sha:
        raise ValueError(
            f"Source artifact checksum mismatch for {filename}: "
            f"expected {expected_sha}, got {actual_sha}"
        )


def _truthy_env(name: str) -> bool:
    return os.environ.get(name, "").strip().lower() in {"1", "true", "yes", "on"}


def _single_archive_member(archive: ZipFile, *, suffixes: tuple[str, ...]) -> str:
    members = [
        name
        for name in archive.namelist()
        if not name.endswith("/") and name.lower().endswith(suffixes)
    ]
    if len(members) != 1:
        raise ValueError(
            "Archive parser needs exactly one matching member or an explicit "
            f"archive_member; found {members}."
        )
    return members[0]


def _repo_root() -> Path:
    return Path(__file__).resolve().parents[1]


def _safe_package_path(source: str | Path) -> Path:
    try:
        return resolve_source_package_path(source)
    except FileNotFoundError:
        return Path(source)


def _scaffold_template(
    *,
    source_id: str,
    package_id: str,
    source_table: str,
    resource_package: str,
    resource_directory: str,
    manifest: str,
) -> str:
    return f"""schema_version: arch.source_package.v1
package_id: {package_id}
label: TODO package label
artifact:
  source_name: {source_id}
  source_table: {source_table}
  resource_package: {resource_package}
  resource_directory: {resource_directory}
  manifest: {manifest}
  vintage: tax_year_{{year}}
  extracted_at: "TODO YYYY-MM-DD"
  extraction_method: xlrd whole-workbook used-range cell parse
record_sets:
  - record_set_id: {source_id}.ty{{year}}.TODO_table
    record_set_spec_id: {source_id}.TODO_table.v1
    source_record_id_prefix: {source_id}.ty{{year}}.TODO_table
    sheet_name: TODO
    period_type: tax_year
    period: "{{year}}"
    geography_id: TODO
    geography_level: country
    geography_name: TODO
    geography_vintage: TODO
    entity: tax_unit
    entity_role: filing_unit
    domain: TODO_domain
    groupby_dimension: TODO_concept
    rows:
      - value_id: all
        label: TODO row label
        ordinal: 0
        row_number: 1
        filters:
          TODO_filter: all
        table_record_kind: total
    measures:
      - measure_id: TODO_measure
        label: TODO measure label
        ordinal: 0
        column: B
        concept: TODO_concept
        unit: count
        aggregation: count
"""