#!/usr/bin/env python3 """Verify and analyze the WebTruffle EU/UK award release dated 2026-08-29. This recipe pins the tagged release manifest by byte count and SHA-256 before trusting its file declarations. It then verifies four CSV products plus the flat-record schema, reads every source column as VARCHAR in DuckDB, and checks headers, row counts, keys, relationships, contract coverage, decimal text, and the release's value-sharing rules. Use awards.csv for aggregation. contract-awards.csv and award-suppliers.csv are award x supplier relationship products: an award-level value can repeat when value_is_shared=true. Monetary outputs remain split by source, currency, and value_basis; this recipe performs no currency conversion or supplier allocation. 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 collections.abc import Iterator from contextlib import contextmanager 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 DATASET_ID = "government-contract-awards" REPOSITORY = "webtruffle/government-contract-awards" RELEASE_TAG = "2026-08-29" RELEASE_URL = f"https://github.com/{REPOSITORY}/releases/tag/{RELEASE_TAG}" ASSET_BASE_URL = ( f"https://github.com/{REPOSITORY}/releases/download/{RELEASE_TAG}" ) MANIFEST_BYTES = 30_982 MANIFEST_SHA256 = "2531eaa5333ce4ede47f8f501b970b192cfa4100ecc5178a9826585aab7caa89" EXPECTED_SCHEMA_VERSION = "2.0" EXPECTED_DUCKDB_VERSION = "1.5.5" LOCK_DIRECTORY_NAME = ".government-contract-awards-python.lock" USER_AGENT = ( "WebTruffle-government-contract-awards-Python-recipe/1.0 " "(+https://www.webtruffle.com/)" ) SCHEMA_FILE = "schema.json" SOURCE_HEALTH_FILE = "source-health.json" FLAT_FILE = "contract-awards.csv" AWARDS_FILE = "awards.csv" RELATIONSHIPS_FILE = "award-suppliers.csv" CONTRACTS_FILE = "contracts.csv" INPUT_FILES = ( SCHEMA_FILE, SOURCE_HEALTH_FILE, FLAT_FILE, AWARDS_FILE, RELATIONSHIPS_FILE, CONTRACTS_FILE, ) PINNED_ASSETS = { SCHEMA_FILE: { "bytes": 15_089, "sha256": "4fb97f596bb4c0ea34f8bdbeea6cfb986f4e94344f9bce739c45986c7746ff81", }, SOURCE_HEALTH_FILE: { "bytes": 2_009, "sha256": "fc08d97e3ccf67e71be4b0ed6fa0bc943d5bbccc5b5367fbebe6f439f53bab50", }, FLAT_FILE: { "bytes": 426_535, "sha256": "f30abec2fc978b54c252bc83be1bb4f16a564f8a4c95e5125c69f32a60cba2b4", }, AWARDS_FILE: { "bytes": 416_112, "sha256": "6a0534417b259e7ef43693c061095c1c0a962e7ed213faf0c7e9c80db7833ec0", }, RELATIONSHIPS_FILE: { "bytes": 125_685, "sha256": "cf5bc36865e2affe9b8f352f7ae70f56e4aff9be8296936fd86ebc485de9226c", }, CONTRACTS_FILE: { "bytes": 68_861, "sha256": "c32283079dff82e497cb9f63bd99fa14d51d667829d64bd25941a7b419075aee", }, } EXPECTED_PRODUCT_COUNTS = { "contract_awards_flat": 217, "awards": 198, "award_suppliers": 217, "contracts": 93, } EXPECTED_PRODUCT_GRAINS = { "contract_awards_flat": "award_supplier_relationship", "awards": "award_group_id", "award_suppliers": "award_group_id × supplier relationship", "contracts": "explicit source contract related to award_group_id", } EXPECTED_TABLE_COUNTS = { "flat_awards": 217, "awards": 198, "award_suppliers": 217, "contracts": 93, } TABLES = { FLAT_FILE: "flat_awards", AWARDS_FILE: "awards", RELATIONSHIPS_FILE: "award_suppliers", CONTRACTS_FILE: "contracts", } FLAT_HEADERS = ( "id", "source", "source_award_id", "source_supplier_id", "source_notice_id", "source_release_id", "notice_type", "title", "buyer_name", "buyer_country", "buyer_region", "award_status", "award_stage", "award_date", "contract_signed_at", "published_at", "contract_start_at", "contract_end_at", "award_value", "award_currency", "procurement_method", "category", "classification_scheme", "classification_codes", "place_country", "place_region", "supplier_count", "supplier_name", "supplier_identifier", "supplier_country", "source_url", "source_license", "first_seen_at", "last_seen_at", "content_hash", "award_group_id", "buyer_source_id", "buyer_identifier", "buyer_identifier_scheme", "buyer_locality", "buyer_country_iso2", "buyer_country_codes", "source_contract_id", "contract_records", "contract_status", "contract_value", "contract_currency", "tender_value", "tender_currency", "primary_value", "primary_currency", "value_basis", "value_is_shared", "sme_suitable", "vcse_suitable", "lot_ids", "lot_count", "lot_has_options", "bids_count", "sme_bids_count", "electronic_bids_count", "foreign_eu_bids_count", "foreign_non_eu_bids_count", "lowest_valid_bid_value", "highest_valid_bid_value", "bid_value_currency", "document_count", "supplier_identifier_scheme", "supplier_scale", "legal_basis_scheme", "legal_basis_id", "gpa_covered", "has_framework_agreement", "has_dynamic_purchasing_system", "has_electronic_auction", "classification_labels", "classification_records", "classification_division", "classification_division_label", "classification_reference", "place_country_iso2", "place_country_codes", "supplier_country_iso2", "supplier_country_codes", "contract_duration_days", "missing_critical_fields", "quality_flags", "source_coverage_profile", "enrichment_method", "enrichment_version", "change_type", "edition_date", "snapshot_as_of", "source_updated_at", "source_update_time_method", "verification_state", "verification_method", "days_since_award", "days_to_contract_start", "days_until_contract_end", "version_count", "last_changed_at", "changed_fields", "field_changes", "change_summary", "previous_award_status", "previous_award_value", "previous_contract_value", "previous_contract_end_at", "change_severity", "enrichment_as_of", ) AWARD_HEADERS = ( "edition_date", "snapshot_as_of", "award_group_id", "source", "source_award_id", "source_notice_id", "source_release_id", "notice_type", "title", "buyer_name", "buyer_source_id", "buyer_identifier", "buyer_identifier_scheme", "buyer_locality", "buyer_country", "buyer_country_iso2", "buyer_country_codes", "buyer_region", "award_status", "award_stage", "award_date", "contract_signed_at", "published_at", "source_updated_at", "source_update_time_method", "verification_state", "verification_method", "contract_start_at", "contract_end_at", "award_value", "award_currency", "contract_value", "contract_currency", "tender_value", "tender_currency", "primary_value", "primary_currency", "value_basis", "value_is_shared", "source_contract_ids", "source_contract_id", "contract_records", "contract_status", "procurement_method", "category", "classification_scheme", "classification_codes", "classification_labels", "classification_records", "classification_division", "classification_division_label", "classification_reference", "place_country", "place_country_iso2", "place_country_codes", "place_region", "sme_suitable", "vcse_suitable", "lot_ids", "lot_count", "lot_has_options", "bids_count", "sme_bids_count", "electronic_bids_count", "foreign_eu_bids_count", "foreign_non_eu_bids_count", "lowest_valid_bid_value", "highest_valid_bid_value", "bid_value_currency", "document_count", "legal_basis_scheme", "legal_basis_id", "gpa_covered", "has_framework_agreement", "has_dynamic_purchasing_system", "has_electronic_auction", "supplier_count", "relationship_count", "source_supplier_ids", "supplier_names", "supplier_records", "source_url", "source_license", "first_seen_at", "last_seen_at", "change_type", "version_count", "last_changed_at", "changed_fields", "field_changes", "change_summary", "change_severity", "days_since_award", "days_to_contract_start", "days_until_contract_end", "contract_duration_days", "missing_critical_fields", "quality_flags", "source_coverage_profile", "enrichment_method", "enrichment_version", "enrichment_as_of", ) RELATIONSHIP_HEADERS = ( "edition_date", "snapshot_as_of", "award_group_id", "relationship_id", "source", "source_notice_id", "source_award_id", "source_supplier_id", "supplier_name", "supplier_identifier", "supplier_identifier_scheme", "supplier_country", "supplier_country_iso2", "supplier_country_codes", "supplier_scale", "supplier_count", "source_url", "source_license", "first_seen_at", "last_seen_at", "change_type", "version_count", "last_changed_at", "changed_fields", "field_changes", "change_summary", "change_severity", "enrichment_version", "enrichment_as_of", ) CONTRACT_HEADERS = ( "edition_date", "snapshot_as_of", "contract_group_id", "award_group_id", "source", "source_notice_id", "source_award_id", "source_contract_id", "contract_status", "contract_signed_at", "contract_start_at", "contract_end_at", "contract_value", "contract_currency", "value_basis", "value_is_shared", "supplier_count", "buyer_name", "buyer_source_id", "title", "classification_scheme", "classification_codes", "classification_division", "classification_division_label", "place_country_iso2", "document_count", "days_to_contract_start", "days_until_contract_end", "contract_duration_days", "source_url", "source_license", "enrichment_version", "enrichment_as_of", ) EXPECTED_HEADERS = { FLAT_FILE: FLAT_HEADERS, AWARDS_FILE: AWARD_HEADERS, RELATIONSHIPS_FILE: RELATIONSHIP_HEADERS, CONTRACTS_FILE: CONTRACT_HEADERS, } MONEY_FIELDS = { FLAT_FILE: ( "award_value", "contract_value", "tender_value", "primary_value", "lowest_valid_bid_value", "highest_valid_bid_value", ), AWARDS_FILE: ( "award_value", "contract_value", "tender_value", "primary_value", "lowest_valid_bid_value", "highest_valid_bid_value", ), CONTRACTS_FILE: ("contract_value",), } DECIMAL_TEXT = re.compile(r"^-?(?:0|[1-9]\d*)(?:\.\d+)?$") def positive_review_limit(value: str) -> int: try: parsed = int(value) except ValueError as error: raise argparse.ArgumentTypeError("must be an integer") from error if not 1 <= parsed <= 100: raise argparse.ArgumentTypeError("must be between 1 and 100") return parsed def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description=( "Download, verify, and analyze the WebTruffle 2026-08-29 EU/UK " "government-contract-awards release." ) ) parser.add_argument( "--data-dir", type=Path, default=Path(f"government-contract-awards-{RELEASE_TAG}"), help="Directory for verified release inputs and deterministic outputs.", ) parser.add_argument( "--review-limit", type=positive_review_limit, default=25, help="Maximum shared-value award rows to export (1-100; default: 25).", ) 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 file_identity(path: Path, checkpoint: str) -> dict[str, Any]: digest, size = sha256_and_size(path) return { "name": path.name, "bytes": size, "sha256": digest, "checkpoint": checkpoint, } def assert_recipe_unchanged( recipe_path: Path, entry_identity: dict[str, Any], checkpoint: str, ) -> dict[str, Any]: try: current_identity = file_identity(recipe_path, checkpoint) except OSError as error: raise RuntimeError( "The executing recipe file could not be re-read at " f"{checkpoint}: {error}. No staged outputs from this run were " "published. Retry with a stable recipe file." ) from error if ( current_identity["bytes"] != entry_identity["bytes"] or current_identity["sha256"] != entry_identity["sha256"] ): raise RuntimeError( "The executing recipe file changed during this run: entry was " f"{entry_identity['bytes']} bytes / {entry_identity['sha256']}, but " f"{checkpoint} found {current_identity['bytes']} bytes / " f"{current_identity['sha256']}. No staged outputs from this run were " "published. Retry with a stable recipe file." ) return current_identity @contextmanager def exclusive_data_dir_lock( data_dir: Path, recipe_entry_identity: dict[str, Any], ) -> Iterator[dict[str, Any]]: lock_dir = data_dir / LOCK_DIRECTORY_NAME owner_path = lock_dir / "owner.json" try: lock_dir.mkdir() except FileExistsError as error: owner_detail = "metadata unavailable" try: owner_detail = owner_path.read_text(encoding="utf-8").strip() except OSError: pass raise RuntimeError( f"Exclusive data-directory lock already exists: {lock_dir}. " "Another recipe run may still be writing this directory. Lock metadata: " f"{owner_detail}. If no writer is active, verify that this is a stale " f"lock, remove only {lock_dir}, and retry." ) from error owner = { "recipe": "government-contract-awards-python", "recipe_entry_bytes": recipe_entry_identity["bytes"], "recipe_entry_sha256": recipe_entry_identity["sha256"], "pid": os.getpid(), "host": platform.node(), "started_at_utc": datetime.now(timezone.utc).isoformat(), "data_dir": str(data_dir), } cleanup_errors: list[str] = [] try: owner_path.write_text( json.dumps(owner, indent=2, sort_keys=True) + "\n", encoding="utf-8", ) print(f"acquired exclusive lock {lock_dir.name}") yield { "directory_name": lock_dir.name, "owner": owner, "scope": "entire run from input verification through output publication", } finally: try: owner_path.unlink(missing_ok=True) except OSError as error: cleanup_errors.append(f"could not remove {owner_path}: {error}") try: lock_dir.rmdir() except FileNotFoundError: pass except OSError as error: cleanup_errors.append(f"could not remove {lock_dir}: {error}") if cleanup_errors: raise RuntimeError( "Exclusive-lock cleanup failed; verify no writer is active, then " "remove only the reported lock paths. " + "; ".join(cleanup_errors) ) 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]: manifest_path = data_dir / "manifest.json" download_missing( f"{ASSET_BASE_URL}/manifest.json", manifest_path, MANIFEST_SHA256, MANIFEST_BYTES, ) manifest = json.loads(manifest_path.read_text(encoding="utf-8")) expected_identity = { "schema_version": EXPECTED_SCHEMA_VERSION, "dataset_id": DATASET_ID, "target_date": RELEASE_TAG, "record_count": EXPECTED_PRODUCT_COUNTS["contract_awards_flat"], } observed_identity = {key: manifest.get(key) for key in expected_identity} if observed_identity != expected_identity: raise RuntimeError( f"Pinned manifest identity changed: expected {expected_identity}, " f"got {observed_identity}." ) products = manifest.get("products", {}) for product, expected_count in EXPECTED_PRODUCT_COUNTS.items(): declaration = products.get(product, {}) observed = { "record_count": int(declaration.get("record_count", -1)), "grain": declaration.get("grain"), } expected = { "record_count": expected_count, "grain": EXPECTED_PRODUCT_GRAINS[product], } if observed != expected: raise RuntimeError( f"Pinned product declaration changed for {product}: " f"expected {expected}, got {observed}." ) declared_files = manifest.get("files", {}) for file_name, pinned in PINNED_ASSETS.items(): declaration = declared_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}: " f"expected {pinned}, got {observed}." ) if tuple(manifest.get("record_fields", ())) != FLAT_HEADERS: raise RuntimeError("Pinned manifest record_fields changed.") return manifest def download_inputs(data_dir: Path, manifest: dict[str, Any]) -> None: declarations = manifest["files"] for file_name in INPUT_FILES: declaration = declarations[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) -> tuple[str, ...]: with path.open("r", encoding="utf-8-sig", newline="") as source: return tuple(next(csv.reader(source))) def validate_schema_and_headers(data_dir: Path) -> dict[str, Any]: schema = json.loads((data_dir / SCHEMA_FILE).read_text(encoding="utf-8")) schema_headers = tuple(schema.get("properties", {})) if schema.get("x-schema-version") != EXPECTED_SCHEMA_VERSION: raise RuntimeError( f"Expected schema version {EXPECTED_SCHEMA_VERSION}, " f"got {schema.get('x-schema-version')!r}." ) if schema_headers != FLAT_HEADERS: raise RuntimeError("schema.json properties do not match the pinned flat header.") if schema.get("required") != ["id", "source"]: raise RuntimeError(f"Unexpected flat-schema required fields: {schema.get('required')}") evidence: dict[str, Any] = {} for file_name, expected in EXPECTED_HEADERS.items(): observed = read_csv_header(data_dir / file_name) if len(observed) != len(set(observed)): raise RuntimeError(f"{file_name} contains duplicate column names.") if observed != expected: missing = sorted(set(expected) - set(observed)) extra = sorted(set(observed) - set(expected)) raise RuntimeError( f"Exact header validation failed for {file_name}: missing={missing}, " f"extra={extra}, order_matches={observed == expected}." ) evidence[file_name] = { "header_count": len(observed), "exact_pinned_order": True, "ordered_headers": list(observed), } evidence[SCHEMA_FILE] = { "property_count": len(schema_headers), "exact_flat_header_order": True, "required_fields": schema["required"], "ordered_properties": list(schema_headers), } return evidence def validate_source_health( data_dir: Path, manifest: dict[str, Any], ) -> dict[str, Any]: source_health = json.loads( (data_dir / SOURCE_HEALTH_FILE).read_text(encoding="utf-8") ) if source_health != manifest.get("source_health"): raise RuntimeError("source-health.json does not exactly match the manifest copy.") field_counts = {len(row) for row in source_health} if field_counts != {18}: raise RuntimeError( f"Expected 18 fields in each source-health row, got {sorted(field_counts)}." ) observed = { row["source"]: { "target_date": row["target_date"], "assessment": row["assessment"], "latest_status": row["latest_status"], "latest_records_seen": row["latest_records_seen"], "expected_quiet_day": row["expected_quiet_day"], "last_nonzero_publication_date": row[ "last_nonzero_publication_date" ], } for row in source_health } if len(source_health) != len(observed): raise RuntimeError("source-health.json contains duplicate source keys.") expected = { "contracts_finder": { "target_date": RELEASE_TAG, "assessment": "successful_nonzero", "latest_status": "succeeded", "latest_records_seen": 124, "expected_quiet_day": False, "last_nonzero_publication_date": RELEASE_TAG, }, "find_a_tender": { "target_date": RELEASE_TAG, "assessment": "successful_nonzero", "latest_status": "succeeded", "latest_records_seen": 93, "expected_quiet_day": False, "last_nonzero_publication_date": RELEASE_TAG, }, "ted": { "target_date": RELEASE_TAG, "assessment": "successful_zero_expected_quiet_day", "latest_status": "succeeded", "latest_records_seen": 0, "expected_quiet_day": True, "last_nonzero_publication_date": "2026-08-28", }, } if observed != expected: raise RuntimeError( f"Pinned source-health checkpoints failed: expected {expected}, " f"got {observed}." ) return observed def validate_decimal_text(data_dir: Path) -> dict[str, Any]: evidence: dict[str, Any] = {} for file_name, fields in MONEY_FIELDS.items(): 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 == "": 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 exceeds two decimal places in {file_name}:" f"{row_number} field {field}: {text!r}." ) counts[field]["present"] += 1 counts[field]["maximum_scale"] = max( counts[field]["maximum_scale"], scale ) evidence[file_name] = 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 source columns: " f"{sorted(types)}." ) evidence[table_name] = { "columns": len(table_info), "source_type": "VARCHAR", } return evidence 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_counts_and_relationships( connection: duckdb.DuckDBPyConnection, ) -> dict[str, Any]: counts = { table: connection.execute(f"SELECT count(*) FROM {table}").fetchone()[0] for table in EXPECTED_TABLE_COUNTS } if counts != EXPECTED_TABLE_COUNTS: raise RuntimeError( f"Pinned table counts changed: expected {EXPECTED_TABLE_COUNTS}, got {counts}." ) checks = fetch_named_row( connection, f""" SELECT (SELECT count(*) FROM flat_awards WHERE nullif(trim(id), '') IS NULL) AS missing_flat_keys, (SELECT count(*) - count(DISTINCT id) FROM flat_awards) AS duplicate_flat_keys, (SELECT count(*) FROM awards WHERE nullif(trim(award_group_id), '') IS NULL) AS missing_award_keys, (SELECT count(*) - count(DISTINCT award_group_id) FROM awards) AS duplicate_award_keys, (SELECT count(*) FROM award_suppliers WHERE nullif(trim(relationship_id), '') IS NULL) AS missing_relationship_keys, (SELECT count(*) - count(DISTINCT relationship_id) FROM award_suppliers) AS duplicate_relationship_keys, (SELECT count(*) FROM contracts WHERE nullif(trim(contract_group_id), '') IS NULL) AS missing_contract_keys, (SELECT count(*) - count(DISTINCT contract_group_id) FROM contracts) AS duplicate_contract_keys, (SELECT count(*) FROM award_suppliers relationship LEFT JOIN awards award USING (award_group_id) WHERE award.award_group_id IS NULL) AS orphan_relationship_awards, (SELECT count(*) FROM contracts contract LEFT JOIN awards award USING (award_group_id) WHERE award.award_group_id IS NULL) AS orphan_contract_awards, (SELECT count(*) FROM awards award LEFT JOIN award_suppliers relationship USING (award_group_id) WHERE relationship.award_group_id IS NULL) AS awards_without_supplier_relationships, (SELECT count(*) FROM flat_awards flat FULL JOIN award_suppliers relationship ON flat.id = relationship.relationship_id WHERE flat.id IS NULL OR relationship.relationship_id IS NULL) AS flat_relationship_membership_mismatches, (SELECT count(*) FROM flat_awards flat JOIN award_suppliers relationship ON flat.id = relationship.relationship_id WHERE flat.award_group_id IS DISTINCT FROM relationship.award_group_id OR flat.source IS DISTINCT FROM relationship.source OR flat.source_notice_id IS DISTINCT FROM relationship.source_notice_id OR flat.source_award_id IS DISTINCT FROM relationship.source_award_id OR flat.source_supplier_id IS DISTINCT FROM relationship.source_supplier_id OR flat.supplier_name IS DISTINCT FROM relationship.supplier_name OR flat.supplier_scale IS DISTINCT FROM relationship.supplier_scale) AS flat_relationship_value_mismatches, (SELECT count(*) FROM flat_awards flat JOIN awards award USING (award_group_id) WHERE flat.source IS DISTINCT FROM award.source OR flat.source_notice_id IS DISTINCT FROM award.source_notice_id OR flat.source_award_id IS DISTINCT FROM award.source_award_id OR flat.primary_value IS DISTINCT FROM award.primary_value OR flat.primary_currency IS DISTINCT FROM award.primary_currency OR flat.value_basis IS DISTINCT FROM award.value_basis OR flat.value_is_shared IS DISTINCT FROM award.value_is_shared) AS flat_award_value_mismatches, (SELECT count(*) FROM award_suppliers relationship JOIN awards award USING (award_group_id) WHERE relationship.source IS DISTINCT FROM award.source OR relationship.source_notice_id IS DISTINCT FROM award.source_notice_id OR relationship.source_award_id IS DISTINCT FROM award.source_award_id) AS relationship_parent_identity_mismatches, (SELECT count(*) FROM contracts contract JOIN awards award USING (award_group_id) WHERE contract.source IS DISTINCT FROM award.source OR contract.source_notice_id IS DISTINCT FROM award.source_notice_id OR contract.source_award_id IS DISTINCT FROM award.source_award_id) AS contract_parent_identity_mismatches, (SELECT count(*) FROM ( SELECT award.award_group_id, try_cast(award.supplier_count AS BIGINT) AS declared_suppliers, try_cast(award.relationship_count AS BIGINT) AS declared_relationships, count(relationship.relationship_id) AS observed_relationships FROM awards award LEFT JOIN award_suppliers relationship USING (award_group_id) GROUP BY ALL ) grouped WHERE declared_suppliers IS NULL OR declared_relationships IS NULL OR declared_suppliers <> observed_relationships OR declared_relationships <> observed_relationships) AS supplier_relationship_count_mismatches, (SELECT count(*) FROM contracts contract JOIN awards award USING (award_group_id) WHERE contract.contract_value IS DISTINCT FROM award.contract_value OR contract.contract_currency IS DISTINCT FROM award.contract_currency OR contract.source_contract_id IS DISTINCT FROM award.source_contract_id) AS single_contract_projection_mismatches, (SELECT count(*) FROM ( SELECT award.award_group_id, CASE WHEN nullif(award.source_contract_id, '') IS NULL THEN 0 ELSE 1 END AS projected_contract_count, count(contract.contract_group_id) AS observed_contract_count FROM awards award LEFT JOIN contracts contract USING (award_group_id) GROUP BY award.award_group_id, projected_contract_count ) grouped WHERE projected_contract_count <> observed_contract_count) AS contract_membership_count_mismatches, (SELECT count(*) FROM awards WHERE edition_date <> '{RELEASE_TAG}') AS wrong_award_editions, (SELECT count(*) FROM award_suppliers WHERE edition_date <> '{RELEASE_TAG}') AS wrong_relationship_editions, (SELECT count(*) FROM contracts WHERE edition_date <> '{RELEASE_TAG}') AS wrong_contract_editions """, ) expected = {key: 0 for key in checks} if checks != expected: raise RuntimeError( f"Pinned key/relationship checks failed: expected {expected}, got {checks}." ) return {"table_counts": counts, "checks": checks} def validate_value_semantics( connection: duckdb.DuckDBPyConnection, ) -> dict[str, Any]: casts: dict[str, Any] = {} for file_name, fields in MONEY_FIELDS.items(): table = TABLES[file_name] casts[table] = {} 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} """ ).fetchone() if invalid: raise RuntimeError( f"DuckDB DECIMAL validation failed for {table}.{field}: " f"{invalid} values could not be represented exactly." ) casts[table][field] = { "present": present, "duckdb_type": "DECIMAL(38, 2)", "invalid": invalid, } checks = fetch_named_row( connection, """ SELECT (SELECT count(*) FROM awards WHERE lower(value_is_shared) = 'true') AS shared_value_awards, (SELECT count(*) FROM awards WHERE lower(value_is_shared) = 'false') AS unshared_value_awards, (SELECT count(*) FROM awards WHERE value_is_shared IS NULL OR value_is_shared = '') AS unavailable_value_share_flags, (SELECT max(try_cast(supplier_count AS BIGINT)) FROM awards) AS maximum_supplier_count, (SELECT count(*) FROM awards WHERE ( nullif(primary_value, '') IS NULL AND value_is_shared IS NOT NULL ) OR ( nullif(primary_value, '') IS NOT NULL AND (lower(value_is_shared) = 'true') IS DISTINCT FROM (try_cast(supplier_count AS BIGINT) > 1) )) AS shared_flag_rule_mismatches, (SELECT count(*) FROM awards WHERE nullif(primary_value, '') IS NOT NULL AND (nullif(primary_currency, '') IS NULL OR value_basis NOT IN ('award', 'contract', 'tender_estimate'))) AS valued_awards_missing_semantics, (SELECT count(*) FROM awards WHERE nullif(primary_value, '') IS NULL AND (nullif(primary_currency, '') IS NOT NULL OR value_basis <> 'unavailable' OR value_is_shared IS NOT NULL)) AS unavailable_award_semantic_mismatches, (SELECT count(*) FROM awards WHERE CASE WHEN nullif(award_value, '') IS NOT NULL THEN primary_value IS DISTINCT FROM award_value OR primary_currency IS DISTINCT FROM award_currency OR value_basis <> 'award' WHEN nullif(contract_value, '') IS NOT NULL THEN primary_value IS DISTINCT FROM contract_value OR primary_currency IS DISTINCT FROM contract_currency OR value_basis <> 'contract' WHEN nullif(tender_value, '') IS NOT NULL THEN primary_value IS DISTINCT FROM tender_value OR primary_currency IS DISTINCT FROM tender_currency OR value_basis <> 'tender_estimate' ELSE primary_value IS NOT NULL OR primary_currency IS NOT NULL OR value_basis <> 'unavailable' END) AS primary_value_precedence_mismatches, (SELECT count(*) FROM awards WHERE source = 'contracts_finder') AS contracts_finder_awards, (SELECT count(*) FROM awards WHERE source = 'find_a_tender') AS find_a_tender_awards, (SELECT count(*) FROM awards WHERE source = 'contracts_finder' AND award_stage = 'awarded') AS contracts_finder_awarded_stage, (SELECT count(*) FROM awards WHERE source = 'find_a_tender' AND award_stage = 'contracted') AS find_a_tender_contracted_stage, (SELECT count(*) FROM awards WHERE source = 'contracts_finder' AND primary_currency = 'GBP' AND value_basis = 'award') AS contracts_finder_gbp_award_basis, (SELECT count(*) FROM awards WHERE source = 'find_a_tender' AND primary_currency = 'GBP' AND value_basis = 'contract') AS find_a_tender_gbp_contract_basis, (SELECT count(*) FROM awards WHERE source = 'find_a_tender' AND primary_currency IS NULL AND value_basis = 'unavailable') AS find_a_tender_unavailable_basis, (SELECT count(*) FROM contracts) AS explicit_contract_records, (SELECT count(*) FROM contracts WHERE source = 'find_a_tender') AS find_a_tender_contract_records, (SELECT count(*) FROM contracts WHERE contract_status = 'active') AS active_contract_records, (SELECT count(*) FROM flat_awards WHERE source = 'contracts_finder') AS contracts_finder_flat_rows, (SELECT count(*) FROM flat_awards WHERE source = 'find_a_tender') AS find_a_tender_flat_rows, (SELECT count(*) FROM award_suppliers WHERE source = 'contracts_finder') AS contracts_finder_relationship_rows, (SELECT count(*) FROM award_suppliers WHERE source = 'find_a_tender') AS find_a_tender_relationship_rows, (SELECT count(*) FROM awards WHERE change_type = 'new') AS new_award_rows, (SELECT count(*) FROM awards WHERE change_type = 'unchanged') AS unchanged_award_rows """, ) expected_checks = { "shared_value_awards": 4, "unshared_value_awards": 193, "unavailable_value_share_flags": 1, "maximum_supplier_count": 16, "shared_flag_rule_mismatches": 0, "valued_awards_missing_semantics": 0, "unavailable_award_semantic_mismatches": 0, "primary_value_precedence_mismatches": 0, "contracts_finder_awards": 105, "find_a_tender_awards": 93, "contracts_finder_awarded_stage": 105, "find_a_tender_contracted_stage": 93, "contracts_finder_gbp_award_basis": 105, "find_a_tender_gbp_contract_basis": 92, "find_a_tender_unavailable_basis": 1, "explicit_contract_records": 93, "find_a_tender_contract_records": 93, "active_contract_records": 93, "contracts_finder_flat_rows": 124, "find_a_tender_flat_rows": 93, "contracts_finder_relationship_rows": 124, "find_a_tender_relationship_rows": 93, "new_award_rows": 4, "unchanged_award_rows": 194, } if checks != expected_checks: raise RuntimeError( f"Pinned value-semantic checks failed: expected {expected_checks}, " f"got {checks}." ) totals = connection.execute( """ SELECT (SELECT sum(try_cast(primary_value AS DECIMAL(38, 2))) FROM awards WHERE source = 'contracts_finder' AND primary_currency = 'GBP' AND value_basis = 'award'), (SELECT sum(try_cast(primary_value AS DECIMAL(38, 2))) FROM flat_awards WHERE source = 'contracts_finder' AND primary_currency = 'GBP' AND value_basis = 'award'), (SELECT sum(try_cast(primary_value AS DECIMAL(38, 2))) FROM awards WHERE source = 'find_a_tender' AND primary_currency = 'GBP' AND value_basis = 'contract'), (SELECT sum(try_cast(primary_value AS DECIMAL(38, 2))) FROM flat_awards WHERE source = 'find_a_tender' AND primary_currency = 'GBP' AND value_basis = 'contract') """ ).fetchone() expected_totals = ( Decimal("205967080.37"), Decimal("1724446419.04"), Decimal("299236615.29"), Decimal("299236615.29"), ) if totals != expected_totals: raise RuntimeError( f"Pinned value totals changed: expected {expected_totals}, got {totals}." ) return { "decimal_casts": casts, "checks": checks, "separate_value_group_totals": { "contracts_finder_gbp_award_basis_awards_csv": str(totals[0]), "contracts_finder_gbp_award_basis_flat_csv": str(totals[1]), "find_a_tender_gbp_contract_basis_awards_csv": str(totals[2]), "find_a_tender_gbp_contract_basis_flat_csv": str(totals[3]), }, } 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"] = { "url": f"{ASSET_BASE_URL}/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] = { "url": f"{ASSET_BASE_URL}/{file_name}", "bytes": size, "sha256": digest, } return evidence RELEASE_CHECK_SQL = f""" SELECT * FROM ( VALUES ( 1, '{RELEASE_TAG}', 'manifest.json', 'release control', {MANIFEST_BYTES}::BIGINT, '{MANIFEST_SHA256}', {MANIFEST_BYTES}::BIGINT, '{MANIFEST_SHA256}', NULL::BIGINT, NULL::BIGINT, NULL, NULL::BIGINT, NULL::BIGINT, 'release metadata and integrity declarations', 'verified' ), ( 2, '{RELEASE_TAG}', '{SCHEMA_FILE}', 'flat-record schema', {PINNED_ASSETS[SCHEMA_FILE]['bytes']}::BIGINT, '{PINNED_ASSETS[SCHEMA_FILE]['sha256']}', {PINNED_ASSETS[SCHEMA_FILE]['bytes']}::BIGINT, '{PINNED_ASSETS[SCHEMA_FILE]['sha256']}', NULL::BIGINT, {len(FLAT_HEADERS)}::BIGINT, NULL, NULL::BIGINT, NULL::BIGINT, 'schema for the flat compatibility record', 'verified' ), ( 3, '{RELEASE_TAG}', '{SOURCE_HEALTH_FILE}', 'source health', {PINNED_ASSETS[SOURCE_HEALTH_FILE]['bytes']}::BIGINT, '{PINNED_ASSETS[SOURCE_HEALTH_FILE]['sha256']}', {PINNED_ASSETS[SOURCE_HEALTH_FILE]['bytes']}::BIGINT, '{PINNED_ASSETS[SOURCE_HEALTH_FILE]['sha256']}', 3::BIGINT, 18::BIGINT, 'source', 0::BIGINT, 0::BIGINT, 'one collector health row per source', 'verified' ), ( 4, '{RELEASE_TAG}', '{FLAT_FILE}', 'compatibility export', {PINNED_ASSETS[FLAT_FILE]['bytes']}::BIGINT, '{PINNED_ASSETS[FLAT_FILE]['sha256']}', {PINNED_ASSETS[FLAT_FILE]['bytes']}::BIGINT, '{PINNED_ASSETS[FLAT_FILE]['sha256']}', (SELECT count(*) FROM flat_awards), {len(FLAT_HEADERS)}::BIGINT, 'id', (SELECT count(*) FROM flat_awards WHERE nullif(trim(id), '') IS NULL), (SELECT count(*) - count(DISTINCT id) FROM flat_awards), 'award_supplier_relationship', 'verified' ), ( 5, '{RELEASE_TAG}', '{AWARDS_FILE}', 'aggregation-safe awards', {PINNED_ASSETS[AWARDS_FILE]['bytes']}::BIGINT, '{PINNED_ASSETS[AWARDS_FILE]['sha256']}', {PINNED_ASSETS[AWARDS_FILE]['bytes']}::BIGINT, '{PINNED_ASSETS[AWARDS_FILE]['sha256']}', (SELECT count(*) FROM awards), {len(AWARD_HEADERS)}::BIGINT, 'award_group_id', (SELECT count(*) FROM awards WHERE nullif(trim(award_group_id), '') IS NULL), (SELECT count(*) - count(DISTINCT award_group_id) FROM awards), 'award_group_id', 'verified' ), ( 6, '{RELEASE_TAG}', '{RELATIONSHIPS_FILE}', 'supplier relationships', {PINNED_ASSETS[RELATIONSHIPS_FILE]['bytes']}::BIGINT, '{PINNED_ASSETS[RELATIONSHIPS_FILE]['sha256']}', {PINNED_ASSETS[RELATIONSHIPS_FILE]['bytes']}::BIGINT, '{PINNED_ASSETS[RELATIONSHIPS_FILE]['sha256']}', (SELECT count(*) FROM award_suppliers), {len(RELATIONSHIP_HEADERS)}::BIGINT, 'relationship_id', (SELECT count(*) FROM award_suppliers WHERE nullif(trim(relationship_id), '') IS NULL), (SELECT count(*) - count(DISTINCT relationship_id) FROM award_suppliers), 'award_group_id x supplier relationship', 'verified' ), ( 7, '{RELEASE_TAG}', '{CONTRACTS_FILE}', 'explicit contracts', {PINNED_ASSETS[CONTRACTS_FILE]['bytes']}::BIGINT, '{PINNED_ASSETS[CONTRACTS_FILE]['sha256']}', {PINNED_ASSETS[CONTRACTS_FILE]['bytes']}::BIGINT, '{PINNED_ASSETS[CONTRACTS_FILE]['sha256']}', (SELECT count(*) FROM contracts), {len(CONTRACT_HEADERS)}::BIGINT, 'contract_group_id', (SELECT count(*) FROM contracts WHERE nullif(trim(contract_group_id), '') IS NULL), (SELECT count(*) - count(DISTINCT contract_group_id) FROM contracts), 'explicit source contract related to award_group_id', 'verified' ) ) receipt( sort_order, release_tag, input_file, artifact_role, expected_bytes, expected_sha256, verified_bytes, verified_sha256, record_rows, field_count, key_field, missing_keys, duplicate_keys, declared_grain, verification_state ) ORDER BY sort_order """.strip() SOURCE_SUMMARY_SQL = f""" WITH source_health( source, source_status, latest_records_seen, expected_quiet_day, source_assessment, source_license ) AS ( VALUES ( 'contracts_finder', 'succeeded', 124::BIGINT, false, 'successful_nonzero', 'https://www.nationalarchives.gov.uk/doc/open-government-licence/version/3/' ), ( 'find_a_tender', 'succeeded', 93::BIGINT, false, 'successful_nonzero', 'https://www.nationalarchives.gov.uk/doc/open-government-licence/version/3/' ), ( 'ted', 'succeeded', 0::BIGINT, true, 'successful_zero_expected_quiet_day', 'https://ted.europa.eu/en/legal-notice' ) ), award_counts AS ( SELECT source, count(*) AS award_rows, count_if(nullif(primary_value, '') IS NOT NULL) AS valued_award_rows, count_if(nullif(primary_value, '') IS NULL) AS unavailable_value_rows, count_if(change_type = 'new') AS new_award_rows, count_if(change_type = 'unchanged') AS unchanged_award_rows, min(award_stage) AS award_stage FROM awards GROUP BY source ), value_rows AS ( SELECT DISTINCT source, coalesce(primary_currency, 'unavailable') || '/' || value_basis AS value_group FROM awards ), value_groups AS ( SELECT source, string_agg(value_group, '; ' ORDER BY value_group) AS value_groups FROM value_rows GROUP BY source ), relationship_counts AS ( SELECT source, count(*) AS supplier_relationship_rows FROM award_suppliers GROUP BY source ), contract_counts AS ( SELECT source, count(*) AS explicit_contract_rows FROM contracts GROUP BY source ) SELECT '{RELEASE_TAG}' AS release_tag, health.source, health.source_status, health.latest_records_seen, health.expected_quiet_day, health.source_assessment, coalesce(award.award_rows, 0) AS award_rows, coalesce(relationship.supplier_relationship_rows, 0) AS supplier_relationship_rows, coalesce(contract.explicit_contract_rows, 0) AS explicit_contract_rows, coalesce(award.valued_award_rows, 0) AS valued_award_rows, coalesce(award.unavailable_value_rows, 0) AS unavailable_value_rows, coalesce(award.new_award_rows, 0) AS new_award_rows, coalesce(award.unchanged_award_rows, 0) AS unchanged_award_rows, award.award_stage, value.value_groups, health.source_license, CASE WHEN health.source = 'ted' THEN 'Successful expected-quiet-day zero; not evidence of no EU activity' ELSE 'Edition rows only; not a complete market or period-spend total' END AS interpretation FROM source_health health LEFT JOIN award_counts award USING (source) LEFT JOIN value_groups value USING (source) LEFT JOIN relationship_counts relationship USING (source) LEFT JOIN contract_counts contract USING (source) ORDER BY health.source """.strip() VALUE_RECONCILIATION_SQL = f""" WITH safe AS ( SELECT source, primary_currency, value_basis, count(*) AS award_rows, count_if(nullif(primary_value, '') IS NOT NULL) AS valued_award_rows, sum(try_cast(primary_value AS DECIMAL(38, 2))) AS safe_award_total FROM awards GROUP BY source, primary_currency, value_basis ), relationship AS ( SELECT source, primary_currency, value_basis, count(*) AS relationship_rows, count_if(nullif(primary_value, '') IS NOT NULL) AS valued_relationship_rows, sum(try_cast(primary_value AS DECIMAL(38, 2))) AS naive_relationship_total FROM flat_awards GROUP BY source, primary_currency, value_basis ) SELECT '{RELEASE_TAG}' AS release_tag, coalesce(safe.source, relationship.source) AS source, coalesce(safe.primary_currency, relationship.primary_currency) AS primary_currency, coalesce(safe.value_basis, relationship.value_basis) AS value_basis, safe.award_rows, safe.valued_award_rows, safe.safe_award_total, relationship.relationship_rows, relationship.valued_relationship_rows, relationship.naive_relationship_total, relationship.naive_relationship_total - safe.safe_award_total AS repeated_value_inflation, round(relationship.naive_relationship_total / safe.safe_award_total, 4) AS naive_to_safe_multiplier, round( (relationship.naive_relationship_total - safe.safe_award_total) * 100 / safe.safe_award_total, 2 ) AS naive_above_safe_percent, 'Do not combine rows across source, currency, or value_basis; use awards.csv totals' AS aggregation_rule FROM safe FULL JOIN relationship ON safe.source IS NOT DISTINCT FROM relationship.source AND safe.primary_currency IS NOT DISTINCT FROM relationship.primary_currency AND safe.value_basis IS NOT DISTINCT FROM relationship.value_basis ORDER BY source, primary_currency NULLS LAST, value_basis """.strip() def build_multi_supplier_review_sql(review_limit: int) -> str: return f""" SELECT '{RELEASE_TAG}' AS release_tag, award_group_id, source, title, buyer_name, try_cast(supplier_count AS BIGINT) AS supplier_count, try_cast(relationship_count AS BIGINT) AS relationship_count, primary_currency, value_basis, try_cast(primary_value AS DECIMAL(38, 2)) AS award_level_primary_value, try_cast(primary_value AS DECIMAL(38, 2)) * try_cast(relationship_count AS BIGINT) AS naive_relationship_total, try_cast(primary_value AS DECIMAL(38, 2)) * (try_cast(relationship_count AS BIGINT) - 1) AS repeated_value_inflation, value_is_shared, supplier_names, source_url, 'Review supplier relationships; do not allocate the shared award value to each supplier' AS review_rule FROM awards WHERE lower(value_is_shared) = 'true' ORDER BY repeated_value_inflation DESC, award_group_id LIMIT {review_limit} """.strip() CONTRACT_COVERAGE_SQL = f""" WITH award_source AS ( SELECT source, count(*) AS source_award_rows, count_if(nullif(source_contract_id, '') IS NOT NULL) AS awards_with_explicit_contract, count_if(nullif(source_contract_id, '') IS NULL) AS awards_without_explicit_contract FROM awards GROUP BY source ), contract_source AS ( SELECT source, count(*) AS explicit_contract_rows, count_if(nullif(contract_value, '') IS NOT NULL) AS valued_contract_rows, count_if(nullif(contract_value, '') IS NULL) AS unavailable_contract_value_rows FROM contracts GROUP BY source ), contract_value_rows AS ( SELECT DISTINCT source, coalesce(contract_currency, 'unavailable') || '/' || value_basis AS contract_value_group FROM contracts ), contract_value_groups AS ( SELECT source, string_agg(contract_value_group, '; ' ORDER BY contract_value_group) AS contract_value_groups FROM contract_value_rows GROUP BY source ) SELECT '{RELEASE_TAG}' AS release_tag, award.source, award.source_award_rows, award.awards_with_explicit_contract, award.awards_without_explicit_contract, coalesce(contract.explicit_contract_rows, 0) AS explicit_contract_rows, coalesce(contract.valued_contract_rows, 0) AS valued_contract_rows, coalesce(contract.unavailable_contract_value_rows, 0) AS unavailable_contract_value_rows, value.contract_value_groups, CASE WHEN contract.explicit_contract_rows IS NULL THEN 'No explicit source contract rows in this edition; absence does not prove that no contract exists' ELSE 'Coverage counts are source-level; value groups are labels, not amounts to add' END AS coverage_rule FROM award_source award LEFT JOIN contract_source contract USING (source) LEFT JOIN contract_value_groups value USING (source) ORDER BY award.source """.strip() SUPPLIER_SCALE_SQL = f""" SELECT '{RELEASE_TAG}' AS release_tag, source, coalesce(nullif(supplier_scale, ''), 'unknown') AS supplier_scale, count(*) AS supplier_relationship_rows, count(DISTINCT award_group_id) AS distinct_awards, count_if(nullif(supplier_identifier, '') IS NOT NULL) AS relationships_with_supplier_identifier, count_if(nullif(supplier_identifier, '') IS NULL) AS relationships_without_supplier_identifier, 'Relationship grain; unknown scale remains unknown and is not large' AS interpretation FROM award_suppliers GROUP BY source, coalesce(nullif(supplier_scale, ''), 'unknown') ORDER BY source, supplier_scale """.strip() def write_query( connection: duckdb.DuckDBPyConnection, sql: str, destination: Path, ) -> dict[str, Any]: rows = connection.execute(f"SELECT count(*) FROM ({sql}) 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:<58} {rows:>7,} rows") return {"rows": rows, "bytes": size, "sha256": digest} def run_locked( args: argparse.Namespace, data_dir: Path, recipe_path: Path, recipe_entry_identity: dict[str, Any], lock_evidence: dict[str, Any], ) -> None: manifest = load_manifest(data_dir) download_inputs(data_dir, manifest) schema_evidence = validate_schema_and_headers(data_dir) source_health_evidence = validate_source_health(data_dir, manifest) decimal_text_evidence = validate_decimal_text(data_dir) connection = duckdb.connect() varchar_evidence = register_csv_tables(connection, data_dir) relationship_evidence = assert_counts_and_relationships(connection) value_evidence = validate_value_semantics(connection) input_evidence = reassert_input_files(data_dir, manifest) review_sql = build_multi_supplier_review_sql(args.review_limit) queries = { "government-contract-awards-release-check.csv": RELEASE_CHECK_SQL, "government-contract-awards-source-summary.csv": SOURCE_SUMMARY_SQL, "government-contract-awards-value-reconciliation.csv": ( VALUE_RECONCILIATION_SQL ), "government-contract-awards-multi-supplier-review.csv": review_sql, "government-contract-awards-contract-coverage.csv": CONTRACT_COVERAGE_SQL, "government-contract-awards-supplier-scale.csv": SUPPLIER_SCALE_SQL, } with tempfile.TemporaryDirectory( dir=data_dir, prefix=".government-contract-awards-results.", ) as temporary_directory: staging_dir = Path(temporary_directory) outputs = { file_name: write_query( connection, sql, staging_dir / file_name, ) for file_name, sql in queries.items() } expected_output_rows = { "government-contract-awards-release-check.csv": 7, "government-contract-awards-source-summary.csv": 3, "government-contract-awards-value-reconciliation.csv": 3, "government-contract-awards-multi-supplier-review.csv": min( args.review_limit, 4 ), "government-contract-awards-contract-coverage.csv": 2, "government-contract-awards-supplier-scale.csv": 6, } observed_output_rows = { file_name: evidence["rows"] for file_name, evidence in outputs.items() } if observed_output_rows != expected_output_rows: raise RuntimeError( "Pinned output row checkpoints failed: expected " f"{expected_output_rows}, got {observed_output_rows}." ) recipe_before_receipt_identity = assert_recipe_unchanged( recipe_path, recipe_entry_identity, "immediately_before_provenance_receipt", ) edition_change_counts = [ { "source": source, "change_type": change_type, "award_rows": rows, } for source, change_type, rows in connection.execute( """ SELECT source, change_type, count(*) FROM awards GROUP BY source, change_type ORDER BY source, change_type """ ).fetchall() ] provenance = { "dataset_id": DATASET_ID, "recipe": "government-contract-awards-python", "recipe_version": "1.0", "recipe_file": { "name": recipe_entry_identity["name"], "bytes": recipe_entry_identity["bytes"], "sha256": recipe_entry_identity["sha256"], "entry_identity": recipe_entry_identity, "identity_immediately_before_receipt": ( recipe_before_receipt_identity ), "unchanged_during_run": True, }, "release_tag": RELEASE_TAG, "release_url": RELEASE_URL, "manifest_url": f"{ASSET_BASE_URL}/manifest.json", "schema_version": manifest["schema_version"], "target_date": manifest["target_date"], "release_generated_at": manifest["generated_at"], "queried_at_utc": datetime.now(timezone.utc).isoformat(), "review_limit": args.review_limit, "exclusive_data_directory_lock": lock_evidence, "duckdb_version": duckdb.__version__, "python_runtime": { "minimum_version": "3.10", "version": sys.version, "implementation": platform.python_implementation(), "platform": platform.platform(), }, "input_files": input_evidence, "schema_and_headers": schema_evidence, "source_health": source_health_evidence, "manifest_source_runs": manifest["sources"], "manifest_source_coverage": manifest["coverage"], "edition_award_change_counts": edition_change_counts, "materialized_source_types": varchar_evidence, "decimal_text_validation": decimal_text_evidence, "relationships": relationship_evidence, "value_semantics": value_evidence, "source_licenses": manifest["source_licenses"], "outputs": outputs, "queries": queries, "interpretation": { "aggregation": ( "Use awards.csv, one row per award_group_id, for monetary " "aggregation. Never sum the flat or relationship products." ), "value_groups": ( "Keep source, primary_currency, and value_basis separate. " "No currency conversion is performed." ), "shared_values": ( "value_is_shared=true means one award-level value repeats over " "multiple supplier relationships; it is not allocated supplier value." ), "contracts": ( "contracts.csv contains explicit related source contracts only. " "A missing row does not establish that no contract exists." ), "suppliers": ( "Supplier scale is relationship-grain evidence. Missing scale is " "unknown, and names are not cross-source entity keys." ), "nulls": "Null remains unavailable; the recipe does not impute values.", "authority": ( "Use source_url and the current official source record for " "decision-critical facts." ), }, } staged_provenance = ( staging_dir / "government-contract-awards-provenance.json" ) staged_provenance.write_text( json.dumps(provenance, indent=2, sort_keys=True) + "\n", encoding="utf-8", ) final_provenance = data_dir / staged_provenance.name final_provenance.unlink(missing_ok=True) for file_name in outputs: os.replace(staging_dir / file_name, data_dir / file_name) os.replace(staged_provenance, final_provenance) print(f"wrote {final_provenance.name}") connection.close() print( "verified checkpoint: 198 award-grain rows; 217 supplier relationships; " "93 explicit contracts; 4 shared-value awards; 0 orphan keys" ) def main() -> None: recipe_path = Path(__file__).resolve() recipe_entry_identity = file_identity(recipe_path, "main_entry") args = parse_args() if duckdb.__version__ != EXPECTED_DUCKDB_VERSION: raise RuntimeError( f"This pinned recipe requires duckdb=={EXPECTED_DUCKDB_VERSION}; " f"found {duckdb.__version__}." ) data_dir = args.data_dir.expanduser().resolve() data_dir.mkdir(parents=True, exist_ok=True) with exclusive_data_dir_lock(data_dir, recipe_entry_identity) as lock_evidence: run_locked( args, data_dir, recipe_path, recipe_entry_identity, lock_evidence, ) if __name__ == "__main__": main()