#!/usr/bin/env python3
"""Verify and query the WebTruffle government-tenders-rfps release dated 2026-08-08.

The recipe downloads tagged, SHA-256-pinned release assets before querying them. It
verifies the pinned manifest, every selected file's byte count and SHA-256,
the normalized-schema/header relationship, and the product row counts. DuckDB reads
the CSV columns as VARCHAR so source identifiers and source-formatted values survive;
queries cast only the fields they need.

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 sys
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from urllib.request import Request, urlopen

import duckdb


REPOSITORY = "webtruffle/government-tenders-rfps"
RELEASE_TAG = "2026-08-08"
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 = "cbbe78fe20b38cb94ebbf21144638207ba796f2b546eee2f4785b51565a9d2b1"
MANIFEST_BYTES = 33_479
EXPECTED_SCHEMA_VERSION = "2.0"
USER_AGENT = "WebTruffle-tender-Python-recipe/1.0 (+https://www.webtruffle.com/)"
INPUT_FILES = (
    "schema.json",
    "data-dictionary.json",
    "current-opportunities.csv",
    "material-changes.csv",
)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Download, verify, and query the WebTruffle 2026-08-08 tender release."
    )
    parser.add_argument(
        "--data-dir",
        type=Path,
        default=Path(f"government-tender-data-{RELEASE_TAG}"),
        help="Directory for verified inputs and 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}, 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}."
        )
    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]:
    # utf-8-sig removes the release CSV's UTF-8 byte-order mark from the first name.
    with path.open("r", encoding="utf-8-sig", newline="") as source:
        return next(csv.reader(source))


def validate_schema_boundary(data_dir: Path) -> dict[str, Any]:
    schema = json.loads((data_dir / "schema.json").read_text(encoding="utf-8"))
    schema_fields = set(schema["properties"])
    csv_headers = read_csv_header(data_dir / "current-opportunities.csv")
    header_fields = set(csv_headers)
    if len(csv_headers) != len(header_fields):
        raise RuntimeError("current-opportunities.csv contains duplicate column names.")

    missing = sorted(schema_fields - header_fields)
    extra = sorted(header_fields - schema_fields)
    if missing:
        raise RuntimeError(f"CSV is missing normalized schema fields: {missing}")
    if extra != ["observation_date"]:
        raise RuntimeError(
            "Unexpected schema/export boundary. Expected only observation_date outside "
            f"the normalized record schema, got: {extra}"
        )
    return {
        "schema_field_count": len(schema_fields),
        "csv_header_count": len(csv_headers),
        "missing_schema_fields": missing,
        "export_only_fields": extra,
    }


def register_csv_tables(connection: duckdb.DuckDBPyConnection, data_dir: Path) -> None:
    # Reading every source column as VARCHAR avoids accidental identifier coercion.
    # Materializing the verified files also prevents later path reads from changing
    # the query population while this recipe is running.
    connection.read_csv(
        str(data_dir / "current-opportunities.csv"),
        header=True,
        all_varchar=True,
    ).create_view("opportunities_source")
    connection.execute(
        "CREATE TEMP TABLE opportunities AS SELECT * FROM opportunities_source"
    )
    connection.execute("DROP VIEW opportunities_source")
    connection.read_csv(
        str(data_dir / "material-changes.csv"),
        header=True,
        all_varchar=True,
    ).create_view("material_changes_source")
    connection.execute(
        "CREATE TEMP TABLE material_changes AS SELECT * FROM material_changes_source"
    )
    connection.execute("DROP VIEW material_changes_source")


def reassert_input_files(
    data_dir: Path,
    manifest: dict[str, Any],
) -> dict[str, dict[str, Any]]:
    """Detect any input mutation between initial verification and final output."""
    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 = {
        "current_opportunities": connection.execute(
            "SELECT count(*) FROM opportunities"
        ).fetchone()[0],
        "material_changes": connection.execute(
            "SELECT count(*) FROM material_changes"
        ).fetchone()[0],
    }
    expected = {
        "current_opportunities": int(
            manifest["products"]["current_opportunities"]["record_count"]
        ),
        "material_changes": int(
            manifest["products"]["material_changes"]["record_count"]
        ),
    }
    if observed != expected:
        raise RuntimeError(f"Product row-count mismatch: expected {expected}, got {observed}")
    return observed


SOURCE_SUMMARY_SQL = f"""
SELECT
    '{RELEASE_TAG}' AS release_tag,
    '{RELEASE_TAG}' AS edition_target_date,
    source,
    count(*) AS records,
    count_if(record_stage = 'opportunity') AS opportunity_records,
    count_if(record_stage = 'amendment') AS amendment_records,
    count_if(deadline_bucket = 'unavailable') AS deadline_unavailable,
    count_if(deadline_bucket = '0-7 days') AS deadline_0_7_days,
    count_if(deadline_bucket = '8-14 days') AS deadline_8_14_days,
    count_if(deadline_bucket = '15-30 days') AS deadline_15_30_days,
    count_if(deadline_bucket = '31+ days') AS deadline_31_plus_days,
    min(source_license) AS source_license,
    count(DISTINCT source_license) AS distinct_source_license_values,
    count_if(source_license IS NULL OR source_license = '') AS missing_source_license
FROM opportunities
GROUP BY source
ORDER BY records DESC, source
""".strip()


DEADLINE_WATCHLIST_SQL = f"""
SELECT
    '{RELEASE_TAG}' AS release_tag,
    '{RELEASE_TAG}' AS edition_target_date,
    source,
    id,
    source_id,
    title,
    buyer_name,
    deadline_at,
    deadline_bucket,
    try_cast(days_to_deadline AS INTEGER) AS days_to_deadline,
    classification_primary_scheme,
    classification_division,
    classification_division_label,
    actionability_basis,
    source_license,
    source_url
FROM opportunities
WHERE deadline_bucket IN ('0-7 days', '8-14 days')
ORDER BY try_cast(days_to_deadline AS INTEGER), source, id
""".strip()


MATERIAL_DEADLINE_CHANGES_SQL = f"""
SELECT
    '{RELEASE_TAG}' AS release_tag,
    '{RELEASE_TAG}' AS edition_target_date,
    observation_date,
    source,
    id,
    source_id,
    title,
    buyer_name,
    previous_deadline_at,
    deadline_at,
    change_summary,
    changed_fields,
    field_changes,
    source_license,
    source_url
FROM material_changes
WHERE changed_fields LIKE '%"deadline_at"%'
ORDER BY source, id
""".strip()


CATEGORY_SUMMARY_SQL = f"""
SELECT
    '{RELEASE_TAG}' AS release_tag,
    '{RELEASE_TAG}' AS edition_target_date,
    source,
    classification_primary_scheme,
    classification_division,
    classification_division_label,
    source_license,
    count(*) AS records
FROM opportunities
GROUP BY
    source,
    classification_primary_scheme,
    classification_division,
    classification_division_label,
    source_license
ORDER BY
    classification_primary_scheme,
    source,
    records DESC,
    classification_division,
    classification_division_label,
    source_license
""".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:<30} {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_check = validate_schema_boundary(data_dir)

    connection = duckdb.connect()
    register_csv_tables(connection, data_dir)
    product_counts = assert_product_counts(connection, manifest)

    source_counts = dict(
        connection.execute(
            "SELECT source, count(*) FROM opportunities GROUP BY source"
        ).fetchall()
    )
    expected_source_counts = {"ted": 29_406, "sam": 8_685, "contracts_finder": 119}
    if source_counts != expected_source_counts:
        raise RuntimeError(
            f"Pinned-release source-count check failed: {source_counts}"
        )

    input_evidence = reassert_input_files(data_dir, manifest)
    with tempfile.TemporaryDirectory(
        dir=data_dir,
        prefix=".query-results.",
    ) as temporary_directory:
        staging_dir = Path(temporary_directory)
        outputs = {
            "source-summary.csv": write_query(
                connection, SOURCE_SUMMARY_SQL, staging_dir / "source-summary.csv"
            ),
            "deadline-watchlist.csv": write_query(
                connection,
                DEADLINE_WATCHLIST_SQL,
                staging_dir / "deadline-watchlist.csv",
            ),
            "material-deadline-changes.csv": write_query(
                connection,
                MATERIAL_DEADLINE_CHANGES_SQL,
                staging_dir / "material-deadline-changes.csv",
            ),
            "category-summary.csv": write_query(
                connection, CATEGORY_SUMMARY_SQL, staging_dir / "category-summary.csv"
            ),
        }

        expected_output_rows = {
            "source-summary.csv": 3,
            "deadline-watchlist.csv": 8_646,
            "material-deadline-changes.csv": 2,
            "category-summary.csv": 199,
        }
        observed_output_rows = {
            file_name: output["rows"] for file_name, output in outputs.items()
        }
        if observed_output_rows != expected_output_rows:
            raise RuntimeError(
                "Pinned-release query 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)
        license_rows = connection.execute(
            """
            SELECT
                product,
                source,
                source_license,
                count(*) AS records
            FROM (
                SELECT
                    'current_opportunities' AS product,
                    source,
                    nullif(source_license, '') AS source_license
                FROM opportunities
                UNION ALL
                SELECT
                    'material_changes' AS product,
                    source,
                    nullif(source_license, '') AS source_license
                FROM material_changes
            ) AS license_evidence
            GROUP BY product, source, source_license
            ORDER BY product, source, source_license
            """
        ).fetchall()

        provenance = {
            "recipe": "government-tender-data-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"],
            "queried_at_utc": datetime.now(timezone.utc).isoformat(),
            "duckdb_version": duckdb.__version__,
            "python_runtime": {
                "version": sys.version,
                "implementation": platform.python_implementation(),
                "platform": platform.platform(),
            },
            "product_counts": product_counts,
            "source_counts": source_counts,
            "schema_check": schema_check,
            "inputs": input_evidence,
            "source_license_evidence": [
                {
                    "product": product,
                    "source": source,
                    "source_license": source_license,
                    "records": records,
                    "missing": source_license is None,
                }
                for product, source, source_license, records in license_rows
            ],
            "outputs": outputs,
            "queries": {
                "source-summary.csv": SOURCE_SUMMARY_SQL,
                "deadline-watchlist.csv": DEADLINE_WATCHLIST_SQL,
                "material-deadline-changes.csv": MATERIAL_DEADLINE_CHANGES_SQL,
                "category-summary.csv": CATEGORY_SUMMARY_SQL,
            },
            "interpretation": {
                "current_opportunities": manifest["products"][
                    "current_opportunities"
                ]["grain"],
                "deadline_watchlist": (
                    "Deterministic candidates in the 0-7 or 8-14 day deadline bucket "
                    "as of the 2026-08-08 edition; not a live-open-tender assertion."
                ),
            },
        }
        staged_provenance_path = staging_dir / "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}")
    print(
        "verified checkpoint: 38,210 candidates; 8,646 in the edition's first "
        "two deadline buckets; 2 material deadline changes"
    )


if __name__ == "__main__":
    main()