#!/usr/bin/env python3
"""Verify and query the WebTruffle US federal awards release dated 2026-08-25.

The recipe downloads a tagged, SHA-256-pinned manifest and the award, supplier,
and award-supplier CSV products. It verifies bytes, hashes, declared grains,
exact schema headers, row counts, keys, and referential integrity before DuckDB
queries materialized VARCHAR tables. Monetary text is validated with Python
Decimal and queried only through DuckDB DECIMAL casts.

The release is a three-day changed-record window. Its cumulative award balances
are not spending during the window, and supplier totals are an alternative view
of award balances rather than another amount to add.

Requires Python 3.10+ and duckdb==1.5.5.

SPDX-License-Identifier: MIT
"""

from __future__ import annotations

import argparse
import csv
import hashlib
import json
import os
import platform
import re
import sys
import tempfile
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
from pathlib import Path
from typing import Any
from urllib.request import Request, urlopen

import duckdb

REPOSITORY = "webtruffle/us-federal-contract-awards"
RELEASE_TAG = "2026-08-25"
RELEASE_URL = f"https://github.com/{REPOSITORY}/releases/tag/{RELEASE_TAG}"
ASSET_BASE_URL = (
    f"https://github.com/{REPOSITORY}/releases/download/{RELEASE_TAG}"
)
MANIFEST_SHA256 = "fcfe470fc3ef678319f88471849f8f4d448ceaafa406bd40584b876216443c7e"
MANIFEST_BYTES = 11_153
EXPECTED_SCHEMA_VERSION = "1.0"
WINDOW_START = "2026-08-23"
WINDOW_END = "2026-08-25"
USER_AGENT = (
    "WebTruffle-US-federal-awards-Python-recipe/1.0 "
    "(+https://www.webtruffle.com/)"
)

AWARDS_FILE = "us-federal-contract-awards.csv"
SUPPLIERS_FILE = "suppliers.csv"
RELATIONSHIPS_FILE = "award-suppliers.csv"
SCHEMA_FILE = "schema.json"
INPUT_FILES = (SCHEMA_FILE, AWARDS_FILE, SUPPLIERS_FILE, RELATIONSHIPS_FILE)

PINNED_CSV_ASSETS = {
    AWARDS_FILE: {
        "bytes": 20_403_764,
        "sha256": "bcf2f1131dcd7de8df89534dae3d5f4ef8c700418bd7b60d23a81aeb5e6a69fa",
    },
    SUPPLIERS_FILE: {
        "bytes": 548_038,
        "sha256": "f9bd32daf4d8a909c70319c7d650f57bf137728580e9fc2c95b30dfc46dc695a",
    },
    RELATIONSHIPS_FILE: {
        "bytes": 4_026_066,
        "sha256": "205cb9a1ed68512ff8633c17f77cd8b2b4a240feb27b06febaad89c60d21ee3f",
    },
}

EXPECTED_PRODUCT_COUNTS = {
    "awards": 17_945,
    "suppliers": 3_145,
    "award-suppliers": 17_945,
}
EXPECTED_PRODUCT_GRAINS = {
    "awards": "one prime contract award summary per generated award identifier",
    "suppliers": "one supplier identity within this edition",
    "award-suppliers": "one prime award to its current summary recipient relationship",
}
EXPECTED_FILE_GRAINS = {
    AWARDS_FILE: "one award holder",
    SUPPLIERS_FILE: "one supplier identity",
    RELATIONSHIPS_FILE: "one award-supplier relation",
}
EXPECTED_HEADER_COUNTS = {
    AWARDS_FILE: 66,
    SUPPLIERS_FILE: 13,
    RELATIONSHIPS_FILE: 10,
}
SCHEMA_DEFINITIONS = {
    AWARDS_FILE: "award",
    SUPPLIERS_FILE: "supplier",
    RELATIONSHIPS_FILE: "award-supplier",
}
TABLES = {
    AWARDS_FILE: "awards",
    SUPPLIERS_FILE: "suppliers",
    RELATIONSHIPS_FILE: "award_suppliers",
}
MONEY_FIELDS = {
    AWARDS_FILE: (
        "total_obligated_amount_usd",
        "total_outlay_amount_usd",
        "current_total_value_of_award_usd",
        "potential_total_value_of_award_usd",
    ),
    SUPPLIERS_FILE: ("total_obligated_amount_usd",),
}
DECIMAL_TEXT = re.compile(r"^-?(?:0|[1-9]\d*)(?:\.\d+)?$")


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description=(
            "Download, verify, and query the WebTruffle 2026-08-25 US federal "
            "contract-awards release."
        )
    )
    parser.add_argument(
        "--data-dir",
        type=Path,
        default=Path(f"us-federal-contract-awards-{RELEASE_TAG}"),
        help="Directory for verified inputs and changed-window query outputs.",
    )
    return parser.parse_args()


def sha256_and_size(path: Path) -> tuple[str, int]:
    digest = hashlib.sha256()
    size = 0
    with path.open("rb") as source:
        for chunk in iter(lambda: source.read(1024 * 1024), b""):
            digest.update(chunk)
            size += len(chunk)
    return digest.hexdigest(), size


def assert_file(
    path: Path,
    expected_sha256: str,
    expected_bytes: int,
) -> tuple[str, int]:
    actual_sha256, actual_bytes = sha256_and_size(path)
    if actual_bytes != expected_bytes or actual_sha256 != expected_sha256:
        raise RuntimeError(
            f"Verification failed for {path}: expected {expected_bytes} bytes / "
            f"{expected_sha256}, got {actual_bytes} bytes / {actual_sha256}. "
            "Move or delete the unexpected file before retrying."
        )
    return actual_sha256, actual_bytes


def download_missing(
    url: str,
    destination: Path,
    expected_sha256: str,
    expected_bytes: int,
) -> None:
    if destination.exists():
        assert_file(destination, expected_sha256, expected_bytes)
        print(f"verified existing  {destination.name}")
        return

    request = Request(url, headers={"User-Agent": USER_AGENT, "Accept": "*/*"})
    descriptor, temporary_name = tempfile.mkstemp(
        dir=destination.parent,
        prefix=f".{destination.name}.",
        suffix=".part",
    )
    partial = Path(temporary_name)
    try:
        downloaded_bytes = 0
        digest = hashlib.sha256()
        with (
            urlopen(request, timeout=120) as response,
            os.fdopen(descriptor, "wb") as target,
        ):
            while chunk := response.read(1024 * 1024):
                downloaded_bytes += len(chunk)
                if downloaded_bytes > expected_bytes:
                    raise RuntimeError(
                        f"Download exceeded the declared {expected_bytes} bytes for "
                        f"{destination.name}."
                    )
                digest.update(chunk)
                target.write(chunk)
        if downloaded_bytes != expected_bytes or digest.hexdigest() != expected_sha256:
            raise RuntimeError(
                f"Verification failed for downloaded {destination.name}: expected "
                f"{expected_bytes} bytes / {expected_sha256}, got {downloaded_bytes} "
                f"bytes / {digest.hexdigest()}."
            )
        os.replace(partial, destination)
    finally:
        try:
            os.close(descriptor)
        except OSError:
            pass
        partial.unlink(missing_ok=True)
    print(f"downloaded + verified {destination.name}")


def load_manifest(data_dir: Path) -> dict[str, Any]:
    path = data_dir / "manifest.json"
    download_missing(
        f"{ASSET_BASE_URL}/manifest.json",
        path,
        MANIFEST_SHA256,
        MANIFEST_BYTES,
    )
    manifest = json.loads(path.read_text(encoding="utf-8"))
    if manifest.get("schema_version") != EXPECTED_SCHEMA_VERSION:
        raise RuntimeError(
            f"Expected schema {EXPECTED_SCHEMA_VERSION}, "
            f"got {manifest.get('schema_version')!r}."
        )
    if manifest.get("target_date") != RELEASE_TAG:
        raise RuntimeError(
            f"Expected target date {RELEASE_TAG}, got {manifest.get('target_date')!r}."
        )
    if manifest.get("coverage_start_date") != WINDOW_START:
        raise RuntimeError(
            f"Expected coverage start {WINDOW_START}, "
            f"got {manifest.get('coverage_start_date')!r}."
        )
    if manifest.get("coverage_end_date") != WINDOW_END:
        raise RuntimeError(
            f"Expected coverage end {WINDOW_END}, "
            f"got {manifest.get('coverage_end_date')!r}."
        )

    for product, expected_count in EXPECTED_PRODUCT_COUNTS.items():
        declaration = manifest.get("products", {}).get(product, {})
        if int(declaration.get("record_count", -1)) != expected_count:
            raise RuntimeError(
                f"Pinned {product} count changed: {declaration.get('record_count')!r}."
            )
        if declaration.get("grain") != EXPECTED_PRODUCT_GRAINS[product]:
            raise RuntimeError(
                f"Pinned {product} grain changed: {declaration.get('grain')!r}."
            )

    for file_name, pinned in PINNED_CSV_ASSETS.items():
        declaration = manifest.get("files", {}).get(file_name, {})
        observed = {
            "bytes": int(declaration.get("bytes", -1)),
            "sha256": declaration.get("sha256"),
        }
        if observed != pinned:
            raise RuntimeError(
                f"Pinned asset declaration changed for {file_name}: {observed}."
            )
        if int(declaration.get("record_count", -1)) != (
            EXPECTED_PRODUCT_COUNTS[
                "award-suppliers"
                if file_name == RELATIONSHIPS_FILE
                else "suppliers"
                if file_name == SUPPLIERS_FILE
                else "awards"
            ]
        ):
            raise RuntimeError(f"Pinned file count changed for {file_name}.")
        if declaration.get("grain") != EXPECTED_FILE_GRAINS[file_name]:
            raise RuntimeError(
                f"Pinned file grain changed for {file_name}: "
                f"{declaration.get('grain')!r}."
            )
    return manifest


def download_inputs(data_dir: Path, manifest: dict[str, Any]) -> None:
    declared_files = manifest["files"]
    for file_name in INPUT_FILES:
        declaration = declared_files[file_name]
        download_missing(
            f"{ASSET_BASE_URL}/{file_name}",
            data_dir / file_name,
            declaration["sha256"],
            int(declaration["bytes"]),
        )


def read_csv_header(path: Path) -> list[str]:
    with path.open("r", encoding="utf-8-sig", newline="") as source:
        return next(csv.reader(source))


def validate_schema_and_headers(
    data_dir: Path,
    manifest: dict[str, Any],
) -> dict[str, Any]:
    schema = json.loads((data_dir / SCHEMA_FILE).read_text(encoding="utf-8"))
    definitions = schema.get("$defs", {})
    product_refs = schema.get("x-webtruffle-products", {})
    expected_refs = {
        "awards": {"$ref": "#/$defs/award"},
        "suppliers": {"$ref": "#/$defs/supplier"},
        "award-suppliers": {"$ref": "#/$defs/award-supplier"},
    }
    if product_refs != expected_refs:
        raise RuntimeError(f"Unexpected schema product references: {product_refs}")

    evidence: dict[str, Any] = {}
    for file_name, definition_name in SCHEMA_DEFINITIONS.items():
        definition = definitions.get(definition_name, {})
        expected_headers = list(definition.get("properties", {}))
        headers = read_csv_header(data_dir / file_name)
        if len(headers) != len(set(headers)):
            raise RuntimeError(f"{file_name} contains duplicate column names.")
        if headers != expected_headers:
            missing = sorted(set(expected_headers) - set(headers))
            extra = sorted(set(headers) - set(expected_headers))
            raise RuntimeError(
                f"Exact header validation failed for {file_name}: "
                f"missing={missing}, extra={extra}, order_matches={headers == expected_headers}."
            )
        if len(headers) != EXPECTED_HEADER_COUNTS[file_name]:
            raise RuntimeError(
                f"Expected {EXPECTED_HEADER_COUNTS[file_name]} headers in {file_name}, "
                f"got {len(headers)}."
            )
        required = set(definition.get("required", []))
        if not required <= set(headers):
            raise RuntimeError(
                f"Required schema fields are absent from {file_name}: "
                f"{sorted(required - set(headers))}"
            )
        evidence[file_name] = {
            "schema_definition": definition_name,
            "header_count": len(headers),
            "exact_schema_order": True,
            "manifest_file_grain": manifest["files"][file_name]["grain"],
        }

    if manifest.get("record_fields") != read_csv_header(data_dir / AWARDS_FILE):
        raise RuntimeError("Manifest record_fields do not exactly match the awards CSV.")
    return evidence


def validate_decimal_text(data_dir: Path) -> dict[str, Any]:
    evidence: dict[str, Any] = {}
    for file_name, fields in MONEY_FIELDS.items():
        field_counts = {
            field: {"present": 0, "missing": 0, "maximum_scale": 0}
            for field in fields
        }
        with (data_dir / file_name).open(
            "r", encoding="utf-8-sig", newline=""
        ) as source:
            for row_number, row in enumerate(csv.DictReader(source), start=2):
                for field in fields:
                    text = row[field]
                    if text == "":
                        field_counts[field]["missing"] += 1
                        continue
                    if not DECIMAL_TEXT.fullmatch(text):
                        raise RuntimeError(
                            f"Non-canonical decimal text in {file_name}:{row_number} "
                            f"field {field}: {text!r}."
                        )
                    try:
                        value = Decimal(text)
                    except InvalidOperation as error:
                        raise RuntimeError(
                            f"Invalid decimal in {file_name}:{row_number} field {field}."
                        ) from error
                    if not value.is_finite():
                        raise RuntimeError(
                            f"Non-finite decimal in {file_name}:{row_number} field {field}."
                        )
                    scale = max(0, -value.as_tuple().exponent)
                    if scale > 2:
                        raise RuntimeError(
                            f"Money value exceeds two decimal places in "
                            f"{file_name}:{row_number} field {field}: {text!r}."
                        )
                    field_counts[field]["present"] += 1
                    field_counts[field]["maximum_scale"] = max(
                        field_counts[field]["maximum_scale"], scale
                    )
        evidence[file_name] = field_counts
    return evidence


def register_csv_tables(
    connection: duckdb.DuckDBPyConnection,
    data_dir: Path,
) -> dict[str, Any]:
    evidence: dict[str, Any] = {}
    for file_name, table_name in TABLES.items():
        source_view = f"{table_name}_source"
        connection.read_csv(
            str(data_dir / file_name),
            header=True,
            all_varchar=True,
        ).create_view(source_view)
        connection.execute(
            f"CREATE TEMP TABLE {table_name} AS SELECT * FROM {source_view}"
        )
        connection.execute(f"DROP VIEW {source_view}")
        table_info = connection.execute(
            f"PRAGMA table_info('{table_name}')"
        ).fetchall()
        types = {row[2] for row in table_info}
        if types != {"VARCHAR"}:
            raise RuntimeError(
                f"{table_name} contains inferred non-VARCHAR columns: {sorted(types)}"
            )
        evidence[table_name] = {
            "columns": len(table_info),
            "source_type": "VARCHAR",
        }
    return evidence


def reassert_input_files(
    data_dir: Path,
    manifest: dict[str, Any],
) -> dict[str, dict[str, Any]]:
    evidence: dict[str, dict[str, Any]] = {}
    digest, size = assert_file(
        data_dir / "manifest.json",
        MANIFEST_SHA256,
        MANIFEST_BYTES,
    )
    evidence["manifest.json"] = {"bytes": size, "sha256": digest}
    for file_name in INPUT_FILES:
        declaration = manifest["files"][file_name]
        digest, size = assert_file(
            data_dir / file_name,
            declaration["sha256"],
            int(declaration["bytes"]),
        )
        evidence[file_name] = {"bytes": size, "sha256": digest}
    return evidence


def assert_product_counts(
    connection: duckdb.DuckDBPyConnection,
    manifest: dict[str, Any],
) -> dict[str, int]:
    observed = {
        "awards": connection.execute("SELECT count(*) FROM awards").fetchone()[0],
        "suppliers": connection.execute(
            "SELECT count(*) FROM suppliers"
        ).fetchone()[0],
        "award-suppliers": connection.execute(
            "SELECT count(*) FROM award_suppliers"
        ).fetchone()[0],
    }
    manifest_counts = {
        product: int(manifest["products"][product]["record_count"])
        for product in EXPECTED_PRODUCT_COUNTS
    }
    if observed != EXPECTED_PRODUCT_COUNTS or observed != manifest_counts:
        raise RuntimeError(
            "Product row-count mismatch: "
            f"pinned={EXPECTED_PRODUCT_COUNTS}, manifest={manifest_counts}, "
            f"observed={observed}."
        )
    return observed


def fetch_named_row(
    connection: duckdb.DuckDBPyConnection,
    sql: str,
) -> dict[str, Any]:
    cursor = connection.execute(sql)
    row = cursor.fetchone()
    if row is None:
        raise RuntimeError("Expected one validation row, got none.")
    return dict(zip((column[0] for column in cursor.description), row))


def assert_keys_and_window(
    connection: duckdb.DuckDBPyConnection,
) -> dict[str, Any]:
    checks = fetch_named_row(
        connection,
        f"""
        SELECT
          (SELECT count(*) FROM awards
            WHERE nullif(trim(award_id), '') IS NULL) AS missing_award_keys,
          (SELECT count(*) - count(DISTINCT award_id) FROM awards)
            AS duplicate_award_keys,
          (SELECT count(*) FROM suppliers
            WHERE nullif(trim(supplier_id), '') IS NULL) AS missing_supplier_keys,
          (SELECT count(*) - count(DISTINCT supplier_id) FROM suppliers)
            AS duplicate_supplier_keys,
          (SELECT count(*) FROM award_suppliers
            WHERE nullif(trim(award_id), '') IS NULL
               OR nullif(trim(supplier_id), '') IS NULL)
            AS missing_relationship_keys,
          (SELECT count(*) - count(DISTINCT award_id) FROM award_suppliers)
            AS duplicate_relationship_award_keys,
          (SELECT count(*) FROM award_suppliers relationship
            LEFT JOIN awards award USING (award_id)
            WHERE award.award_id IS NULL) AS orphan_relationship_award_keys,
          (SELECT count(*) FROM award_suppliers relationship
            LEFT JOIN suppliers supplier USING (supplier_id)
            WHERE supplier.supplier_id IS NULL) AS orphan_relationship_supplier_keys,
          (SELECT count(*) FROM awards award
            LEFT JOIN award_suppliers relationship USING (award_id)
            WHERE relationship.award_id IS NULL) AS awards_without_relationships,
          (SELECT count(*) FROM suppliers supplier
            LEFT JOIN award_suppliers relationship USING (supplier_id)
            WHERE relationship.supplier_id IS NULL) AS suppliers_without_relationships,
          (SELECT count(*) FROM awards award
            JOIN award_suppliers relationship USING (award_id)
            WHERE award.supplier_id IS DISTINCT FROM relationship.supplier_id)
            AS award_relationship_supplier_mismatches,
          (SELECT count(*) FROM awards
            WHERE try_cast(base_action_date AS DATE) IS NULL)
            AS invalid_or_missing_base_action_dates,
          (SELECT count(*) FROM awards
            WHERE try_cast(base_action_date AS DATE) < DATE '{WINDOW_START}')
            AS base_action_dates_before_window,
          (SELECT count(*) FROM awards
            WHERE try_cast(base_action_date AS DATE)
              BETWEEN DATE '{WINDOW_START}' AND DATE '{WINDOW_END}')
            AS base_action_dates_inside_window,
          (SELECT count(*) FROM awards
            WHERE try_cast(base_action_date AS DATE) > DATE '{WINDOW_END}')
            AS base_action_dates_after_window,
          (SELECT count(*) FROM awards
            WHERE nullif(total_outlay_amount_usd, '') IS NOT NULL)
            AS outlay_present_rows,
          (SELECT count(*) FROM awards WHERE change_type = 'new')
            AS first_release_new_rows,
          (SELECT count(DISTINCT edition_date) FROM awards)
            AS distinct_award_edition_dates,
          (SELECT count(*) FROM awards WHERE edition_date <> '{RELEASE_TAG}')
            AS wrong_award_edition_dates,
          (SELECT count(*) FROM awards WHERE source <> 'usaspending')
            AS wrong_award_sources,
          (SELECT count(*) FROM suppliers WHERE source <> 'usaspending')
            AS wrong_supplier_sources,
          (SELECT count(*) FROM award_suppliers WHERE source <> 'usaspending')
            AS wrong_relationship_sources
        """
    )
    expected = {
        "missing_award_keys": 0,
        "duplicate_award_keys": 0,
        "missing_supplier_keys": 0,
        "duplicate_supplier_keys": 0,
        "missing_relationship_keys": 0,
        "duplicate_relationship_award_keys": 0,
        "orphan_relationship_award_keys": 0,
        "orphan_relationship_supplier_keys": 0,
        "awards_without_relationships": 0,
        "suppliers_without_relationships": 0,
        "award_relationship_supplier_mismatches": 0,
        "invalid_or_missing_base_action_dates": 0,
        "base_action_dates_before_window": 16_825,
        "base_action_dates_inside_window": 1_120,
        "base_action_dates_after_window": 0,
        "outlay_present_rows": 1_580,
        "first_release_new_rows": 17_945,
        "distinct_award_edition_dates": 1,
        "wrong_award_edition_dates": 0,
        "wrong_award_sources": 0,
        "wrong_supplier_sources": 0,
        "wrong_relationship_sources": 0,
    }
    if checks != expected:
        raise RuntimeError(
            f"Pinned key/window checkpoints failed: expected {expected}, got {checks}."
        )
    return checks


def validate_duckdb_decimal_semantics(
    connection: duckdb.DuckDBPyConnection,
) -> dict[str, Any]:
    cast_evidence: dict[str, Any] = {}
    table_fields = {
        "awards": MONEY_FIELDS[AWARDS_FILE],
        "suppliers": MONEY_FIELDS[SUPPLIERS_FILE],
    }
    for table_name, fields in table_fields.items():
        cast_evidence[table_name] = {}
        for field in fields:
            invalid, present = connection.execute(
                f"""
                SELECT
                  count_if(
                    nullif({field}, '') IS NOT NULL
                    AND try_cast({field} AS DECIMAL(38, 2)) IS NULL
                  ),
                  count_if(nullif({field}, '') IS NOT NULL)
                FROM {table_name}
                """
            ).fetchone()
            if invalid:
                raise RuntimeError(
                    f"DuckDB DECIMAL validation failed for {table_name}.{field}: "
                    f"{invalid} values could not be represented exactly."
                )
            cast_evidence[table_name][field] = {
                "present": present,
                "duckdb_type": "DECIMAL(38, 2)",
                "invalid": invalid,
            }

    award_total, supplier_total, award_amount_rows, supplier_amount_rows = (
        connection.execute(
            """
            SELECT
              (SELECT sum(try_cast(total_obligated_amount_usd AS DECIMAL(38, 2)))
               FROM awards),
              (SELECT sum(try_cast(total_obligated_amount_usd AS DECIMAL(38, 2)))
               FROM suppliers),
              (SELECT count(*) FROM awards
               WHERE nullif(total_obligated_amount_usd, '') IS NOT NULL),
              (SELECT sum(try_cast(obligation_amount_award_count AS BIGINT))
               FROM suppliers)
            """
        ).fetchone()
    )
    if not isinstance(award_total, Decimal) or not isinstance(supplier_total, Decimal):
        raise TypeError("DuckDB did not return obligation totals as Decimal values.")
    if award_total != supplier_total:
        raise RuntimeError(
            "Supplier obligation totals do not reconcile to the alternative award view."
        )
    if award_amount_rows != supplier_amount_rows:
        raise RuntimeError(
            "Supplier obligation denominators do not reconcile to award rows."
        )
    return {
        "casts": cast_evidence,
        "obligation_totals_reconcile_as_alternative_views": True,
        "obligation_amount_award_rows": award_amount_rows,
        "supplier_obligation_denominator_rows": supplier_amount_rows,
        "amount_rule": (
            "Award and supplier totals are compared for reconciliation only. "
            "They are never added together."
        ),
    }


CHANGED_WINDOW_CHECKPOINTS_SQL = f"""
SELECT
  '{RELEASE_TAG}' AS release_tag,
  '{WINDOW_START}' AS changed_window_start,
  '{WINDOW_END}' AS changed_window_end,
  (SELECT count(*) FROM awards) AS changed_award_records,
  (SELECT count(*) FROM suppliers) AS suppliers_in_changed_window,
  (SELECT count(*) FROM award_suppliers) AS award_supplier_relationships,
  (SELECT count(*) FROM awards
    WHERE try_cast(base_action_date AS DATE) < DATE '{WINDOW_START}')
    AS awards_with_base_date_before_changed_window,
  (SELECT count(*) FROM awards
    WHERE try_cast(base_action_date AS DATE)
      BETWEEN DATE '{WINDOW_START}' AND DATE '{WINDOW_END}')
    AS awards_with_base_date_inside_changed_window,
  (SELECT count(*) FROM awards
    WHERE nullif(total_outlay_amount_usd, '') IS NOT NULL)
    AS awards_with_outlay_present,
  (SELECT count(*) FROM awards WHERE change_type = 'new')
    AS rows_new_to_first_pipeline_release
""".strip()


CHANGED_WINDOW_AGENCY_COUNTS_SQL = f"""
SELECT
  '{RELEASE_TAG}' AS release_tag,
  '{WINDOW_START} through {WINDOW_END} last_modified_date' AS changed_window_basis,
  coalesce(nullif(awarding_agency_name, ''), '(missing)') AS awarding_agency_name,
  count(*) AS changed_award_records,
  count(DISTINCT supplier_id) AS supplier_ids_present_in_changed_window,
  count_if(nullif(total_outlay_amount_usd, '') IS NOT NULL)
    AS changed_award_records_with_outlay_present
FROM awards
GROUP BY coalesce(nullif(awarding_agency_name, ''), '(missing)')
ORDER BY changed_award_records DESC, awarding_agency_name
LIMIT 25
""".strip()


CHANGED_WINDOW_FIELD_COVERAGE_SQL = f"""
SELECT
  '{RELEASE_TAG}' AS release_tag,
  '{WINDOW_START} through {WINDOW_END} last_modified_date' AS changed_window_basis,
  field_name,
  present_records,
  total_records,
  round(present_records * 100.0 / total_records, 2) AS presence_percent
FROM (
  SELECT 'total_obligated_amount_usd' AS field_name,
    count_if(nullif(total_obligated_amount_usd, '') IS NOT NULL) AS present_records,
    count(*) AS total_records FROM awards
  UNION ALL
  SELECT 'total_outlay_amount_usd',
    count_if(nullif(total_outlay_amount_usd, '') IS NOT NULL), count(*) FROM awards
  UNION ALL
  SELECT 'current_total_value_of_award_usd',
    count_if(nullif(current_total_value_of_award_usd, '') IS NOT NULL), count(*) FROM awards
  UNION ALL
  SELECT 'potential_total_value_of_award_usd',
    count_if(nullif(potential_total_value_of_award_usd, '') IS NOT NULL), count(*) FROM awards
  UNION ALL
  SELECT 'number_of_offers_received',
    count_if(nullif(number_of_offers_received, '') IS NOT NULL), count(*) FROM awards
  UNION ALL
  SELECT 'set_aside_code',
    count_if(nullif(set_aside_code, '') IS NOT NULL), count(*) FROM awards
  UNION ALL
  SELECT 'naics_code',
    count_if(nullif(naics_code, '') IS NOT NULL), count(*) FROM awards
  UNION ALL
  SELECT 'product_or_service_code',
    count_if(nullif(product_or_service_code, '') IS NOT NULL), count(*) FROM awards
) AS coverage
ORDER BY field_name
""".strip()


CHANGED_WINDOW_SUPPLIER_REVIEW_SQL = f"""
SELECT
  '{RELEASE_TAG}' AS release_tag,
  '{WINDOW_START} through {WINDOW_END} last_modified_date' AS changed_window_basis,
  supplier_id,
  supplier_identifier_scheme,
  recipient_name,
  try_cast(award_count AS BIGINT) AS changed_award_records,
  try_cast(obligation_amount_award_count AS BIGINT)
    AS changed_award_records_with_obligation_present,
  first_base_action_date,
  latest_last_modified_at
FROM suppliers
ORDER BY changed_award_records DESC, supplier_id
LIMIT 50
""".strip()


CHANGED_WINDOW_JOIN_VALIDATION_SQL = """
SELECT
  (SELECT count(*) FROM awards) AS award_rows,
  (SELECT count(DISTINCT award_id) FROM awards) AS distinct_award_keys,
  (SELECT count(*) FROM suppliers) AS supplier_rows,
  (SELECT count(DISTINCT supplier_id) FROM suppliers) AS distinct_supplier_keys,
  (SELECT count(*) FROM award_suppliers) AS relationship_rows,
  (SELECT count(*) FROM award_suppliers relationship
    LEFT JOIN awards award USING (award_id)
    WHERE award.award_id IS NULL) AS orphan_award_keys,
  (SELECT count(*) FROM award_suppliers relationship
    LEFT JOIN suppliers supplier USING (supplier_id)
    WHERE supplier.supplier_id IS NULL) AS orphan_supplier_keys,
  (SELECT sum(try_cast(total_obligated_amount_usd AS DECIMAL(38, 2)))
     FROM awards)
    =
  (SELECT sum(try_cast(total_obligated_amount_usd AS DECIMAL(38, 2)))
     FROM suppliers)
    AS alternative_view_obligation_totals_reconcile,
  'Do not add award and supplier totals together' AS aggregation_rule
""".strip()


def write_query(
    connection: duckdb.DuckDBPyConnection,
    sql: str,
    destination: Path,
) -> dict[str, Any]:
    rows = connection.execute(f"SELECT count(*) FROM ({sql}) AS result").fetchone()[0]
    connection.sql(sql).write_csv(
        str(destination),
        header=True,
        overwrite=True,
        use_tmp_file=True,
    )
    digest, size = sha256_and_size(destination)
    print(f"wrote {destination.name:<48} {rows:>7,} rows")
    return {"rows": rows, "bytes": size, "sha256": digest}


def main() -> None:
    args = parse_args()
    data_dir = args.data_dir.expanduser().resolve()
    data_dir.mkdir(parents=True, exist_ok=True)

    manifest = load_manifest(data_dir)
    download_inputs(data_dir, manifest)
    schema_evidence = validate_schema_and_headers(data_dir, manifest)
    python_decimal_evidence = validate_decimal_text(data_dir)

    connection = duckdb.connect()
    varchar_evidence = register_csv_tables(connection, data_dir)
    product_counts = assert_product_counts(connection, manifest)
    key_window_checks = assert_keys_and_window(connection)
    decimal_evidence = validate_duckdb_decimal_semantics(connection)

    input_evidence = reassert_input_files(data_dir, manifest)
    with tempfile.TemporaryDirectory(
        dir=data_dir,
        prefix=".changed-window-results.",
    ) as temporary_directory:
        staging_dir = Path(temporary_directory)
        outputs = {
            "changed-window-checkpoints.csv": write_query(
                connection,
                CHANGED_WINDOW_CHECKPOINTS_SQL,
                staging_dir / "changed-window-checkpoints.csv",
            ),
            "changed-window-agency-record-counts.csv": write_query(
                connection,
                CHANGED_WINDOW_AGENCY_COUNTS_SQL,
                staging_dir / "changed-window-agency-record-counts.csv",
            ),
            "changed-window-field-coverage.csv": write_query(
                connection,
                CHANGED_WINDOW_FIELD_COVERAGE_SQL,
                staging_dir / "changed-window-field-coverage.csv",
            ),
            "changed-window-supplier-review.csv": write_query(
                connection,
                CHANGED_WINDOW_SUPPLIER_REVIEW_SQL,
                staging_dir / "changed-window-supplier-review.csv",
            ),
            "changed-window-join-validation.csv": write_query(
                connection,
                CHANGED_WINDOW_JOIN_VALIDATION_SQL,
                staging_dir / "changed-window-join-validation.csv",
            ),
        }
        expected_output_rows = {
            "changed-window-checkpoints.csv": 1,
            "changed-window-agency-record-counts.csv": 25,
            "changed-window-field-coverage.csv": 8,
            "changed-window-supplier-review.csv": 50,
            "changed-window-join-validation.csv": 1,
        }
        observed_output_rows = {
            file_name: evidence["rows"] for file_name, evidence in outputs.items()
        }
        if observed_output_rows != expected_output_rows:
            raise RuntimeError(
                "Pinned-release output checkpoints failed: expected "
                f"{expected_output_rows}, got {observed_output_rows}."
            )

        recipe_path = Path(__file__).resolve()
        recipe_sha256, recipe_bytes = sha256_and_size(recipe_path)
        provenance = {
            "dataset_id": "us-federal-contract-awards",
            "recipe": "us-federal-contract-awards-python",
            "recipe_version": "1.0",
            "recipe_file": {
                "name": recipe_path.name,
                "bytes": recipe_bytes,
                "sha256": recipe_sha256,
            },
            "release_tag": RELEASE_TAG,
            "release_url": RELEASE_URL,
            "schema_version": manifest["schema_version"],
            "target_date": manifest["target_date"],
            "release_generated_at": manifest["generated_at"],
            "changed_window": {
                "date_field": "last_modified_date",
                "start": WINDOW_START,
                "end": WINDOW_END,
                "inclusive": True,
            },
            "queried_at_utc": datetime.now(timezone.utc).isoformat(),
            "duckdb_version": duckdb.__version__,
            "python_runtime": {
                "minimum_version": "3.10",
                "version": sys.version,
                "implementation": platform.python_implementation(),
                "platform": platform.platform(),
            },
            "product_counts": product_counts,
            "product_grains": EXPECTED_PRODUCT_GRAINS,
            "schema_and_headers": schema_evidence,
            "materialized_source_types": varchar_evidence,
            "python_decimal_validation": python_decimal_evidence,
            "duckdb_decimal_validation": decimal_evidence,
            "key_and_window_checks": key_window_checks,
            "inputs": input_evidence,
            "source_licenses": manifest["source_licenses"],
            "outputs": outputs,
            "queries": {
                "changed-window-checkpoints.csv": CHANGED_WINDOW_CHECKPOINTS_SQL,
                "changed-window-agency-record-counts.csv": (
                    CHANGED_WINDOW_AGENCY_COUNTS_SQL
                ),
                "changed-window-field-coverage.csv": (
                    CHANGED_WINDOW_FIELD_COVERAGE_SQL
                ),
                "changed-window-supplier-review.csv": (
                    CHANGED_WINDOW_SUPPLIER_REVIEW_SQL
                ),
                "changed-window-join-validation.csv": (
                    CHANGED_WINDOW_JOIN_VALIDATION_SQL
                ),
            },
            "interpretation": {
                "changed_window": (
                    "Rows were newly reported or updated in the edition's three-day "
                    "source last-modified window; they are not contracts awarded in "
                    "that period or a complete active-contract inventory."
                ),
                "first_release_new": (
                    "The pinned first release labels every row new to this pipeline. "
                    "That does not mean every base award was newly signed."
                ),
                "amounts": (
                    "Award obligations are cumulative balances, not changed-window "
                    "flow or cash payments. Supplier obligation totals are the same "
                    "award balances grouped through the supplier view. Never add the "
                    "award and supplier totals together."
                ),
                "relationships": (
                    "Award-supplier rows describe prime-recipient relationships, not "
                    "subawards or subcontractors."
                ),
                "authority": (
                    "Use each award's source_url and the current USAspending record "
                    "for decision-critical facts."
                ),
            },
        }
        staged_provenance_path = (
            staging_dir / "changed-window-query-provenance.json"
        )
        staged_provenance_path.write_text(
            json.dumps(provenance, indent=2, sort_keys=True) + "\n",
            encoding="utf-8",
        )

        final_provenance_path = data_dir / staged_provenance_path.name
        final_provenance_path.unlink(missing_ok=True)
        for file_name in outputs:
            os.replace(staging_dir / file_name, data_dir / file_name)
        os.replace(staged_provenance_path, final_provenance_path)
        print(f"wrote {final_provenance_path.name}")

    connection.close()
    print(
        "verified checkpoint: 17,945 awards; 3,145 suppliers; 17,945 "
        "relationships; 16,825 base dates before the changed window; "
        "1,580 outlay-present rows; 0 orphan keys"
    )


if __name__ == "__main__":
    main()