#!/usr/bin/env python3 """Verify and analyze two WebTruffle procurement lifecycle-link releases. This recipe pins the 2026-08-31 releases of government-tenders-rfps and government-contract-awards. It verifies each manifest, schema, source-health file, and lifecycle CSV by byte count and SHA-256 before DuckDB sees any CSV data. The two releases use the same upstream asset names, so every local input has a dataset-qualified filename. Every CSV column is loaded as VARCHAR. The recipe checks the exact 15-column contract, row and key counts, deterministic link semantics, complete pinned edge composition, and the absence of orphan award edges. It exports bounded, deterministically ordered path samples plus full aggregate checks. Existing verified inputs make repeat runs network-free. Requires Python 3.10+ and duckdb==1.5.5. No pandas dependency is used. 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 collections.abc import Iterator from contextlib import contextmanager from datetime import datetime, timezone from pathlib import Path from typing import Any from urllib.request import Request, urlopen import duckdb RELEASE_TAG = "2026-08-31" EXPECTED_SCHEMA_VERSION = "2.0" EXPECTED_DUCKDB_VERSION = "1.5.5" DEFAULT_PATH_LIMIT = 50 MAX_PATH_LIMIT = 500 LOCK_DIRECTORY_NAME = ".procurement-lifecycle-links-python.lock" USER_AGENT = ( "WebTruffle-procurement-lifecycle-links-Python-recipe/1.0 " "(+https://www.webtruffle.com/)" ) LIFECYCLE_HEADERS = ( "link_id", "source", "from_entity_type", "from_entity_id", "from_stage", "to_entity_type", "to_entity_id", "to_stage", "relationship", "link_method", "confidence", "is_inferred", "evidence", "source_url", "observation_date", ) PINNED_RELEASES: dict[str, dict[str, Any]] = { "government-tenders-rfps": { "repository": "webtruffle/government-tenders-rfps", "generated_at": "2026-09-01T13:25:10Z", "record_count": 4_319, "lifecycle_rows": 162_684, "table": "tender_links", "sources": ( "contracts_finder", "find_a_tender", "sam", "ted", ), "files": { "manifest": { "remote_name": "manifest.json", "local_name": ( "government-tenders-rfps-2026-08-31-manifest.json" ), "bytes": 33_720, "sha256": ( "c50e511d5ff5fe80411d80cf0e07490389f138fa2afca5572167ad7b2ba2c55f" ), }, "schema": { "remote_name": "schema.json", "local_name": "government-tenders-rfps-2026-08-31-schema.json", "bytes": 18_717, "sha256": ( "0a31059590fcb48919458e7af9f97ebce04a1c9a41b6c91883578568634ba8b3" ), }, "source_health": { "remote_name": "source-health.json", "local_name": ( "government-tenders-rfps-2026-08-31-source-health.json" ), "bytes": 2_673, "sha256": ( "35f2266b60aa6f46f127b0b7254782768d074d354d6c35ad65d85507b6044665" ), }, "lifecycle_links": { "remote_name": "lifecycle-links.csv", "local_name": ( "government-tenders-rfps-2026-08-31-lifecycle-links.csv" ), "bytes": 73_352_297, "sha256": ( "d2600170c9e3f44b4471778c00556d5862eda9d8bb376661f861ad6e02397d62" ), }, }, }, "government-contract-awards": { "repository": "webtruffle/government-contract-awards", "generated_at": "2026-09-01T13:38:49Z", "record_count": 2_325, "lifecycle_rows": 88_817, "table": "award_links", "sources": ( "contracts_finder", "find_a_tender", "ted", ), "files": { "manifest": { "remote_name": "manifest.json", "local_name": ( "government-contract-awards-2026-08-31-manifest.json" ), "bytes": 31_102, "sha256": ( "570b8ad004459405201393cb8ba3c81386f55a625ff44175b29840ad5e438d29" ), }, "schema": { "remote_name": "schema.json", "local_name": ( "government-contract-awards-2026-08-31-schema.json" ), "bytes": 15_089, "sha256": ( "4fb97f596bb4c0ea34f8bdbeea6cfb986f4e94344f9bce739c45986c7746ff81" ), }, "source_health": { "remote_name": "source-health.json", "local_name": ( "government-contract-awards-2026-08-31-source-health.json" ), "bytes": 1_994, "sha256": ( "3c769d2cdd3f16a7c27feca8d77b19e7dbdf2eb704d27a54d6bfa263570a6704" ), }, "lifecycle_links": { "remote_name": "lifecycle-links.csv", "local_name": ( "government-contract-awards-2026-08-31-lifecycle-links.csv" ), "bytes": 41_722_673, "sha256": ( "6781b8516de652d3727ec1c30699134a2c57cd8ed38bd33f37027b9267479527" ), }, }, }, } EXPECTED_SOURCE_COUNTS = { ("government-tenders-rfps", "contracts_finder"): 5_129, ("government-tenders-rfps", "find_a_tender"): 2_768, ("government-tenders-rfps", "sam"): 38_843, ("government-tenders-rfps", "ted"): 115_944, ("government-contract-awards", "contracts_finder"): 5_976, ("government-contract-awards", "find_a_tender"): 4_437, ("government-contract-awards", "ted"): 78_404, } EXPECTED_EDGE_COMPOSITION = { ( "government-tenders-rfps", "contracts_finder", "source_release", "amendment", "normalized_notice", "amendment", "normalized_as", ): 46, ( "government-tenders-rfps", "contracts_finder", "source_release", "award", "normalized_notice", "award", "normalized_as", ): 4_802, ( "government-tenders-rfps", "contracts_finder", "source_release", "opportunity", "normalized_notice", "opportunity", "normalized_as", ): 270, ( "government-tenders-rfps", "contracts_finder", "source_release", "planning", "normalized_notice", "planning", "normalized_as", ): 11, ( "government-tenders-rfps", "find_a_tender", "normalized_notice", "contract", "contract", "contract", "declares_contract", ): 1_313, ( "government-tenders-rfps", "find_a_tender", "source_release", "cancellation", "normalized_notice", "cancellation", "normalized_as", ): 28, ( "government-tenders-rfps", "find_a_tender", "source_release", "contract", "normalized_notice", "contract", "normalized_as", ): 1_427, ( "government-tenders-rfps", "sam", "source_release", "opportunity", "normalized_notice", "opportunity", "normalized_as", ): 32_549, ( "government-tenders-rfps", "sam", "source_release", "planning", "normalized_notice", "planning", "normalized_as", ): 6_294, ( "government-tenders-rfps", "ted", "source_release", "award", "normalized_notice", "award", "normalized_as", ): 57_005, ( "government-tenders-rfps", "ted", "source_release", "opportunity", "normalized_notice", "opportunity", "normalized_as", ): 55_857, ( "government-tenders-rfps", "ted", "source_release", "other", "normalized_notice", "other", "normalized_as", ): 1_213, ( "government-tenders-rfps", "ted", "source_release", "planning", "normalized_notice", "planning", "normalized_as", ): 1_869, ( "government-contract-awards", "contracts_finder", "award", "award", "supplier", "supplier", "awarded_to", ): 3_153, ( "government-contract-awards", "contracts_finder", "source_notice", "notice", "award", "awarded", "declares_award", ): 2_823, ( "government-contract-awards", "find_a_tender", "award", "award", "contract", "contract", "results_in_contract", ): 885, ( "government-contract-awards", "find_a_tender", "award", "award", "supplier", "supplier", "awarded_to", ): 1_955, ( "government-contract-awards", "find_a_tender", "source_notice", "notice", "award", "awarded", "declares_award", ): 2, ( "government-contract-awards", "find_a_tender", "source_notice", "notice", "award", "contracted", "declares_award", ): 1_595, ( "government-contract-awards", "ted", "award", "award", "supplier", "supplier", "awarded_to", ): 52_361, ( "government-contract-awards", "ted", "source_notice", "notice", "award", "awarded", "declares_award", ): 26_043, } EXPECTED_GRAPH_METRICS = { "award_outgoing_edges_without_declaration": 0, "award_declarations_without_supplier_edge": 0, "duplicate_declared_award_keys": 0, "declared_award_nodes": 30_463, "outgoing_award_nodes": 30_463, "award_supplier_paths": 57_469, "award_contract_paths": 885, "contract_targets_shared_by_releases": 851, "award_contract_targets_outside_tender_graph": 34, "tender_contract_targets_outside_award_graph": 462, "cross_release_link_id_collisions": 0, } def path_limit(value: str) -> int: try: parsed = int(value) except ValueError as error: raise argparse.ArgumentTypeError("path limit must be an integer") from error if not 1 <= parsed <= MAX_PATH_LIMIT: raise argparse.ArgumentTypeError( f"path limit must be between 1 and {MAX_PATH_LIMIT}" ) return parsed def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description=( "Verify and analyze the pinned 2026-08-31 tender and award " "lifecycle-link releases." ) ) parser.add_argument( "--data-dir", type=Path, default=Path(f"procurement-lifecycle-links-{RELEASE_TAG}"), help="Directory for verified inputs and deterministic outputs.", ) parser.add_argument( "--path-limit", type=path_limit, default=DEFAULT_PATH_LIMIT, help=( "Maximum rows in each path sample " f"(1-{MAX_PATH_LIMIT}; default: {DEFAULT_PATH_LIMIT})." ), ) return parser.parse_args() def release_url(release: dict[str, Any]) -> str: return ( f"https://github.com/{release['repository']}/releases/tag/{RELEASE_TAG}" ) def asset_url(release: dict[str, Any], remote_name: str) -> str: return ( f"https://github.com/{release['repository']}/releases/download/" f"{RELEASE_TAG}/{remote_name}" ) 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; this recipe " "will not overwrite an unverified input." ) return actual_sha256, actual_bytes def file_identity(path: Path) -> dict[str, Any]: digest, size = sha256_and_size(path) return {"name": path.name, "bytes": size, "sha256": digest} def assert_recipe_unchanged( recipe_path: Path, entry_identity: dict[str, Any], checkpoint: str, ) -> dict[str, Any]: try: current = file_identity(recipe_path) except OSError as error: raise RuntimeError( "The executing recipe could not be re-read at " f"{checkpoint}: {error}. No staged outputs were published." ) from error if ( current["bytes"] != entry_identity["bytes"] or current["sha256"] != entry_identity["sha256"] ): raise RuntimeError( "The executing recipe changed during this run: entry was " f"{entry_identity['bytes']} bytes / {entry_identity['sha256']}, " f"but {checkpoint} found {current['bytes']} bytes / " f"{current['sha256']}. No staged outputs were published." ) return current @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 " f"metadata: {owner_detail}. If no writer is active, verify that " f"this is a stale lock, remove only {lock_dir}, and retry." ) from error owner = { "recipe": "procurement-lifecycle-links-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_DIRECTORY_NAME, "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 pinned {expected_bytes} bytes " f"for {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}: " f"expected {expected_bytes} bytes / {expected_sha256}, got " f"{downloaded_bytes} 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 validate_manifest( dataset_id: str, release: dict[str, Any], manifest_path: Path, ) -> dict[str, Any]: 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, "generated_at": release["generated_at"], "record_count": release["record_count"], } observed_identity = {key: manifest.get(key) for key in expected_identity} if observed_identity != expected_identity: raise RuntimeError( f"Pinned manifest identity changed for {dataset_id}: expected " f"{expected_identity}, got {observed_identity}." ) expected_product = { "record_count": release["lifecycle_rows"], "grain": "one exact official-identifier relationship", "files": ["lifecycle-links.csv", "lifecycle-links.jsonl"], "inference_policy": "No fuzzy, name-based, or probabilistic links.", } product = manifest.get("products", {}).get("lifecycle_links", {}) if product != expected_product: raise RuntimeError( f"Pinned lifecycle product changed for {dataset_id}: expected " f"{expected_product}, got {product}." ) declared_files = manifest.get("files", {}) for role in ("schema", "source_health", "lifecycle_links"): pin = release["files"][role] remote_name = pin["remote_name"] declaration = declared_files.get(remote_name, {}) expected_declaration = { "path": f"daily/{RELEASE_TAG}/{remote_name}", "bytes": pin["bytes"], "sha256": pin["sha256"], } if declaration != expected_declaration: raise RuntimeError( f"Pinned declaration changed for {dataset_id}/{remote_name}: " f"expected {expected_declaration}, got {declaration}." ) observed_sources = tuple(row.get("source") for row in manifest.get("sources", [])) if observed_sources != release["sources"]: raise RuntimeError( f"Pinned source list changed for {dataset_id}: expected " f"{release['sources']}, got {observed_sources}." ) if any(row.get("status") != "succeeded" for row in manifest["sources"]): raise RuntimeError(f"A pinned source run was not successful for {dataset_id}.") return manifest def acquire_and_validate_inputs(data_dir: Path) -> dict[str, dict[str, Any]]: manifests: dict[str, dict[str, Any]] = {} for dataset_id, release in PINNED_RELEASES.items(): manifest_pin = release["files"]["manifest"] manifest_path = data_dir / manifest_pin["local_name"] download_missing( asset_url(release, manifest_pin["remote_name"]), manifest_path, manifest_pin["sha256"], manifest_pin["bytes"], ) manifest = validate_manifest(dataset_id, release, manifest_path) manifests[dataset_id] = manifest for role in ("schema", "source_health", "lifecycle_links"): pin = release["files"][role] download_missing( asset_url(release, pin["remote_name"]), data_dir / pin["local_name"], pin["sha256"], pin["bytes"], ) return manifests def validate_json_inputs( data_dir: Path, manifests: dict[str, dict[str, Any]], ) -> dict[str, Any]: evidence: dict[str, Any] = {} for dataset_id, release in PINNED_RELEASES.items(): schema_path = data_dir / release["files"]["schema"]["local_name"] schema = json.loads(schema_path.read_text(encoding="utf-8")) expected_schema = { "title": dataset_id, "x-schema-version": EXPECTED_SCHEMA_VERSION, "required": ["id", "source"], } observed_schema = { "title": schema.get("title"), "x-schema-version": schema.get("x-schema-version"), "required": schema.get("required"), } if observed_schema != expected_schema: raise RuntimeError( f"Pinned schema identity changed for {dataset_id}: expected " f"{expected_schema}, got {observed_schema}." ) health_path = data_dir / release["files"]["source_health"]["local_name"] source_health = json.loads(health_path.read_text(encoding="utf-8")) if source_health != manifests[dataset_id].get("source_health"): raise RuntimeError( f"source-health.json does not equal the manifest declaration " f"for {dataset_id}." ) health_sources = tuple(row.get("source") for row in source_health) if health_sources != release["sources"]: raise RuntimeError( f"Pinned source-health source list changed for {dataset_id}." ) for row in source_health: if ( row.get("target_date") != RELEASE_TAG or row.get("latest_status") != "succeeded" or row.get("assessment") != "successful_nonzero" ): raise RuntimeError( f"Unexpected source-health checkpoint for {dataset_id}/" f"{row.get('source')}: {row}." ) evidence[dataset_id] = { "schema": observed_schema, "source_health_rows": len(source_health), "source_health_sources": list(health_sources), "source_health_equals_manifest": True, } return evidence 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_csv_headers(data_dir: Path) -> dict[str, Any]: evidence: dict[str, Any] = {} for dataset_id, release in PINNED_RELEASES.items(): path = data_dir / release["files"]["lifecycle_links"]["local_name"] observed = read_csv_header(path) if len(observed) != len(set(observed)): raise RuntimeError(f"Duplicate lifecycle column names in {path.name}.") if observed != LIFECYCLE_HEADERS: raise RuntimeError( f"Exact lifecycle header changed for {dataset_id}: expected " f"{LIFECYCLE_HEADERS}, got {observed}." ) evidence[dataset_id] = { "header_count": len(observed), "exact_order": True, "headers": list(observed), } return evidence def reassert_input_files(data_dir: Path) -> dict[str, Any]: evidence: dict[str, Any] = {} for dataset_id, release in PINNED_RELEASES.items(): dataset_evidence: dict[str, Any] = {} for role, pin in release["files"].items(): path = data_dir / pin["local_name"] digest, size = assert_file(path, pin["sha256"], pin["bytes"]) dataset_evidence[role] = { "remote_asset": pin["remote_name"], "local_file": pin["local_name"], "bytes": size, "sha256": digest, "url": asset_url(release, pin["remote_name"]), } evidence[dataset_id] = dataset_evidence return evidence def sql_path(path: Path) -> str: return str(path).replace("'", "''") def register_csv_tables( connection: duckdb.DuckDBPyConnection, data_dir: Path, ) -> dict[str, Any]: evidence: dict[str, Any] = {} for dataset_id, release in PINNED_RELEASES.items(): table = release["table"] path = data_dir / release["files"]["lifecycle_links"]["local_name"] connection.execute( f""" CREATE TABLE {table} AS SELECT * FROM read_csv( '{sql_path(path)}', header = true, all_varchar = true, auto_detect = true ) """ ) described = connection.execute(f"DESCRIBE {table}").fetchall() columns = tuple(row[0] for row in described) types = tuple(row[1] for row in described) if columns != LIFECYCLE_HEADERS: raise RuntimeError(f"DuckDB column order changed for {dataset_id}.") if set(types) != {"VARCHAR"}: raise RuntimeError( f"DuckDB did not materialize every {dataset_id} column as " f"VARCHAR: {types}." ) evidence[dataset_id] = { "table": table, "column_count": len(columns), "all_columns_varchar": True, } return evidence EDGE_COMPOSITION_SQL = """ WITH edges AS ( SELECT 'government-tenders-rfps' AS dataset_id, * FROM tender_links UNION ALL SELECT 'government-contract-awards' AS dataset_id, * FROM award_links ) SELECT dataset_id, source, from_entity_type, from_stage, to_entity_type, to_stage, relationship, count(*) AS edge_rows FROM edges GROUP BY ALL ORDER BY dataset_id, source, from_entity_type, from_stage, to_entity_type, to_stage, relationship """.strip() NODE_COVERAGE_SQL = """ WITH edges AS ( SELECT 'government-tenders-rfps' AS dataset_id, * FROM tender_links UNION ALL SELECT 'government-contract-awards' AS dataset_id, * FROM award_links ), nodes AS ( SELECT dataset_id, source, from_entity_type AS entity_type, from_entity_id AS entity_id, 1 AS outgoing_edge, 0 AS incoming_edge FROM edges UNION ALL SELECT dataset_id, source, to_entity_type AS entity_type, to_entity_id AS entity_id, 0 AS outgoing_edge, 1 AS incoming_edge FROM edges ) SELECT dataset_id, source, entity_type, count(DISTINCT entity_id) AS distinct_nodes, sum(outgoing_edge) AS outgoing_edges, sum(incoming_edge) AS incoming_edges, 'dataset_id × source × entity_type × entity_id' AS node_identity FROM nodes GROUP BY ALL ORDER BY dataset_id, source, entity_type """.strip() GRAPH_METRICS_CTE = """ metrics AS ( SELECT ( SELECT count(*) FROM award_links edge WHERE edge.relationship IN ('awarded_to', 'results_in_contract') AND NOT EXISTS ( SELECT 1 FROM award_links declaration WHERE declaration.relationship = 'declares_award' AND declaration.source = edge.source AND declaration.to_entity_id = edge.from_entity_id ) )::BIGINT AS award_outgoing_edges_without_declaration, ( SELECT count(*) FROM award_links declaration WHERE declaration.relationship = 'declares_award' AND NOT EXISTS ( SELECT 1 FROM award_links supplier_edge WHERE supplier_edge.relationship = 'awarded_to' AND supplier_edge.source = declaration.source AND supplier_edge.from_entity_id = declaration.to_entity_id ) )::BIGINT AS award_declarations_without_supplier_edge, ( SELECT count(*) FROM ( SELECT source, to_entity_id FROM award_links WHERE relationship = 'declares_award' GROUP BY source, to_entity_id HAVING count(*) <> 1 ) duplicate_declarations )::BIGINT AS duplicate_declared_award_keys, ( SELECT count(DISTINCT source || chr(31) || to_entity_id) FROM award_links WHERE relationship = 'declares_award' )::BIGINT AS declared_award_nodes, ( SELECT count(DISTINCT source || chr(31) || from_entity_id) FROM award_links WHERE relationship IN ('awarded_to', 'results_in_contract') )::BIGINT AS outgoing_award_nodes, ( SELECT count(*) FROM award_links declaration JOIN award_links supplier_edge ON supplier_edge.source = declaration.source AND supplier_edge.from_entity_id = declaration.to_entity_id AND supplier_edge.relationship = 'awarded_to' WHERE declaration.relationship = 'declares_award' )::BIGINT AS award_supplier_paths, ( SELECT count(*) FROM award_links declaration JOIN award_links contract_edge ON contract_edge.source = declaration.source AND contract_edge.from_entity_id = declaration.to_entity_id AND contract_edge.relationship = 'results_in_contract' WHERE declaration.relationship = 'declares_award' )::BIGINT AS award_contract_paths, ( SELECT count(*) FROM award_links contract_edge JOIN tender_links tender_contract ON tender_contract.source = contract_edge.source AND tender_contract.to_entity_id = contract_edge.to_entity_id AND tender_contract.relationship = 'declares_contract' WHERE contract_edge.relationship = 'results_in_contract' )::BIGINT AS contract_targets_shared_by_releases, ( SELECT count(*) FROM award_links contract_edge WHERE contract_edge.relationship = 'results_in_contract' AND NOT EXISTS ( SELECT 1 FROM tender_links tender_contract WHERE tender_contract.relationship = 'declares_contract' AND tender_contract.source = contract_edge.source AND tender_contract.to_entity_id = contract_edge.to_entity_id ) )::BIGINT AS award_contract_targets_outside_tender_graph, ( SELECT count(*) FROM tender_links tender_contract WHERE tender_contract.relationship = 'declares_contract' AND NOT EXISTS ( SELECT 1 FROM award_links contract_edge WHERE contract_edge.relationship = 'results_in_contract' AND contract_edge.source = tender_contract.source AND contract_edge.to_entity_id = tender_contract.to_entity_id ) )::BIGINT AS tender_contract_targets_outside_award_graph, ( SELECT count(*) FROM tender_links JOIN award_links USING (link_id) )::BIGINT AS cross_release_link_id_collisions ) """.strip() GRAPH_BOUNDARY_CHECKS_SQL = f""" WITH {GRAPH_METRICS_CTE}, checks AS ( SELECT 1 AS check_order, 'award_outgoing_edges_without_declaration' AS check_name, award_outgoing_edges_without_declaration AS observed_value, 0::BIGINT AS expected_value, 'Every award→supplier or award→contract edge has one source-matched declaration.' AS interpretation FROM metrics UNION ALL SELECT 2, 'award_declarations_without_supplier_edge', award_declarations_without_supplier_edge, 0, 'Every declared award has at least one deterministic supplier edge.' FROM metrics UNION ALL SELECT 3, 'duplicate_declared_award_keys', duplicate_declared_award_keys, 0, 'A source-scoped award key has exactly one declaration edge.' FROM metrics UNION ALL SELECT 4, 'declared_award_nodes', declared_award_nodes, 30463, 'Distinct source-scoped award nodes declared by notice edges.' FROM metrics UNION ALL SELECT 5, 'outgoing_award_nodes', outgoing_award_nodes, 30463, 'Distinct source-scoped award nodes with supplier or contract edges.' FROM metrics UNION ALL SELECT 6, 'award_supplier_paths', award_supplier_paths, 57469, 'Complete notice→award→supplier paths before output sampling.' FROM metrics UNION ALL SELECT 7, 'award_contract_paths', award_contract_paths, 885, 'Complete notice→award→contract paths before output sampling.' FROM metrics UNION ALL SELECT 8, 'contract_targets_shared_by_releases', contract_targets_shared_by_releases, 851, 'Exact contract target IDs present in both pinned release graphs.' FROM metrics UNION ALL SELECT 9, 'award_contract_targets_outside_tender_graph', award_contract_targets_outside_tender_graph, 34, 'Expected release-scope boundary; not an orphan inside the award graph.' FROM metrics UNION ALL SELECT 10, 'tender_contract_targets_outside_award_graph', tender_contract_targets_outside_award_graph, 462, 'Expected release-scope boundary; the two releases are not a closed world.' FROM metrics UNION ALL SELECT 11, 'cross_release_link_id_collisions', cross_release_link_id_collisions, 0, 'Dataset-qualified link collections have disjoint link IDs.' FROM metrics ) SELECT check_name, observed_value, expected_value, CASE WHEN observed_value = expected_value THEN 'PASS' ELSE 'FAIL' END AS status, interpretation FROM checks ORDER BY check_order """.strip() def build_release_receipts_sql() -> str: rows: list[str] = [] for dataset_id, release in PINNED_RELEASES.items(): for role, pin in release["files"].items(): values = ( dataset_id, RELEASE_TAG, release_url(release), release["generated_at"], role, pin["remote_name"], pin["local_name"], pin["bytes"], pin["sha256"], ) quoted = [ f"'{str(value).replace(chr(39), chr(39) * 2)}'" if not isinstance(value, int) else str(value) for value in values ] rows.append("(" + ", ".join(quoted) + ")") return ( "SELECT * FROM (VALUES\n " + ",\n ".join(rows) + "\n) AS receipt(\n" " dataset_id, release_tag, release_url, generated_at, file_role,\n" " remote_asset, local_file, verified_bytes, verified_sha256\n" ")\nORDER BY dataset_id, file_role" ) def build_award_supplier_paths_sql(limit: int) -> str: return f""" SELECT '{RELEASE_TAG}' AS release_tag, declaration.source, declaration.from_entity_id AS source_notice_id, declaration.to_entity_id AS award_id, declaration.to_stage AS award_stage, supplier_edge.to_entity_id AS supplier_id, declaration.link_id AS notice_to_award_link_id, supplier_edge.link_id AS award_to_supplier_link_id, declaration.source_url AS notice_source_url, supplier_edge.source_url AS supplier_edge_source_url, 'source_notice → award → supplier' AS path_shape FROM award_links declaration JOIN award_links supplier_edge ON supplier_edge.source = declaration.source AND supplier_edge.from_entity_id = declaration.to_entity_id AND supplier_edge.relationship = 'awarded_to' WHERE declaration.relationship = 'declares_award' ORDER BY declaration.source, declaration.to_entity_id, supplier_edge.to_entity_id, declaration.link_id, supplier_edge.link_id LIMIT {limit} """.strip() def build_award_contract_paths_sql(limit: int) -> str: return f""" SELECT '{RELEASE_TAG}' AS release_tag, declaration.source, declaration.from_entity_id AS source_notice_id, declaration.to_entity_id AS award_id, declaration.to_stage AS award_stage, contract_edge.to_entity_id AS contract_id, declaration.link_id AS notice_to_award_link_id, contract_edge.link_id AS award_to_contract_link_id, CASE WHEN tender_contract.to_entity_id IS NOT NULL THEN 'present' ELSE 'absent' END AS tender_graph_contract_target, declaration.source_url AS notice_source_url, contract_edge.source_url AS contract_edge_source_url, 'source_notice → award → contract' AS path_shape FROM award_links declaration JOIN award_links contract_edge ON contract_edge.source = declaration.source AND contract_edge.from_entity_id = declaration.to_entity_id AND contract_edge.relationship = 'results_in_contract' LEFT JOIN tender_links tender_contract ON tender_contract.source = contract_edge.source AND tender_contract.to_entity_id = contract_edge.to_entity_id AND tender_contract.relationship = 'declares_contract' WHERE declaration.relationship = 'declares_award' ORDER BY declaration.source, declaration.to_entity_id, contract_edge.to_entity_id, declaration.link_id, contract_edge.link_id LIMIT {limit} """.strip() def assert_graph_contract( connection: duckdb.DuckDBPyConnection, ) -> dict[str, Any]: row_counts: dict[str, int] = {} unique_counts: dict[str, int] = {} for dataset_id, release in PINNED_RELEASES.items(): table = release["table"] row_count, unique_count = connection.execute( f"SELECT count(*), count(DISTINCT link_id) FROM {table}" ).fetchone() if row_count != release["lifecycle_rows"]: raise RuntimeError( f"Pinned row count changed for {dataset_id}: expected " f"{release['lifecycle_rows']}, got {row_count}." ) if unique_count != row_count: raise RuntimeError( f"link_id is not unique for {dataset_id}: {row_count} rows, " f"{unique_count} unique IDs." ) semantics = connection.execute( f""" SELECT link_method, confidence, is_inferred, count(*) FROM {table} GROUP BY ALL ORDER BY ALL """ ).fetchall() expected_semantics = [ ("exact_official_identifier", "1.0", "False", row_count) ] if semantics != expected_semantics: raise RuntimeError( f"Pinned deterministic-link semantics changed for {dataset_id}: " f"expected {expected_semantics}, got {semantics}." ) invalid = connection.execute( f""" SELECT count_if(link_id IS NULL OR NOT regexp_full_match( link_id, '^[0-9a-f]{{64}}$' )), count_if(source IS NULL), count_if(from_entity_type IS NULL OR from_entity_id IS NULL), count_if(to_entity_type IS NULL OR to_entity_id IS NULL), count_if(relationship IS NULL), count_if(evidence IS NULL OR try_cast(evidence AS JSON) IS NULL), count_if(observation_date IS NULL) FROM {table} """ ).fetchone() if any(invalid): raise RuntimeError( f"Required lifecycle values failed for {dataset_id}: {invalid}." ) row_counts[dataset_id] = row_count unique_counts[dataset_id] = unique_count source_rows = connection.execute( """ WITH edges AS ( SELECT 'government-tenders-rfps' AS dataset_id, source FROM tender_links UNION ALL SELECT 'government-contract-awards' AS dataset_id, source FROM award_links ) SELECT dataset_id, source, count(*) FROM edges GROUP BY ALL """ ).fetchall() observed_sources = {(row[0], row[1]): row[2] for row in source_rows} if observed_sources != EXPECTED_SOURCE_COUNTS: raise RuntimeError( "Pinned source composition changed: expected " f"{EXPECTED_SOURCE_COUNTS}, got {observed_sources}." ) composition_rows = connection.execute(EDGE_COMPOSITION_SQL).fetchall() observed_composition = {tuple(row[:7]): row[7] for row in composition_rows} if observed_composition != EXPECTED_EDGE_COMPOSITION: missing = sorted(set(EXPECTED_EDGE_COMPOSITION) - set(observed_composition)) extra = sorted(set(observed_composition) - set(EXPECTED_EDGE_COMPOSITION)) changed = { key: { "expected": EXPECTED_EDGE_COMPOSITION[key], "observed": observed_composition.get(key), } for key in EXPECTED_EDGE_COMPOSITION.keys() & observed_composition.keys() if EXPECTED_EDGE_COMPOSITION[key] != observed_composition[key] } raise RuntimeError( "Pinned edge composition changed: " f"missing={missing}, extra={extra}, changed={changed}." ) metric_names = tuple(EXPECTED_GRAPH_METRICS) metric_row = connection.execute( f"WITH {GRAPH_METRICS_CTE} SELECT * FROM metrics" ).fetchone() observed_metrics = dict(zip(metric_names, metric_row, strict=True)) if observed_metrics != EXPECTED_GRAPH_METRICS: raise RuntimeError( "Pinned graph boundary changed: expected " f"{EXPECTED_GRAPH_METRICS}, got {observed_metrics}." ) node_group_count = connection.execute( f"SELECT count(*) FROM ({NODE_COVERAGE_SQL}) node_coverage" ).fetchone()[0] if node_group_count != 19: raise RuntimeError( f"Pinned node-coverage group count changed: expected 19, got " f"{node_group_count}." ) node_totals = dict( connection.execute( f""" SELECT dataset_id, sum(distinct_nodes)::BIGINT FROM ({NODE_COVERAGE_SQL}) node_coverage GROUP BY dataset_id ORDER BY dataset_id """ ).fetchall() ) expected_node_totals = { "government-contract-awards": 97_914, "government-tenders-rfps": 315_571, } if node_totals != expected_node_totals: raise RuntimeError( "Pinned canonical node totals changed: expected " f"{expected_node_totals}, got {node_totals}." ) return { "row_counts": row_counts, "unique_link_id_counts": unique_counts, "link_method": "exact_official_identifier", "confidence": "1.0", "is_inferred": "False", "source_composition": [ { "dataset_id": dataset_id, "source": source, "edge_rows": rows, } for (dataset_id, source), rows in sorted(observed_sources.items()) ], "edge_composition_groups": len(observed_composition), "node_coverage_groups": node_group_count, "node_identity": "dataset_id × source × entity_type × entity_id", "node_totals": node_totals, "combined_node_total": sum(node_totals.values()), "graph_metrics": observed_metrics, } 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 staged {destination.name:<62} {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: manifests = acquire_and_validate_inputs(data_dir) json_evidence = validate_json_inputs(data_dir, manifests) header_evidence = validate_csv_headers(data_dir) # This is the final byte-level gate before DuckDB reads either CSV. input_evidence = reassert_input_files(data_dir) connection = duckdb.connect() try: varchar_evidence = register_csv_tables(connection, data_dir) graph_evidence = assert_graph_contract(connection) queries = { "procurement-lifecycle-links-release-receipts.csv": ( build_release_receipts_sql() ), "procurement-lifecycle-links-edge-composition.csv": ( EDGE_COMPOSITION_SQL ), "procurement-lifecycle-links-node-coverage.csv": NODE_COVERAGE_SQL, "procurement-lifecycle-links-award-supplier-paths.csv": ( build_award_supplier_paths_sql(args.path_limit) ), "procurement-lifecycle-links-award-contract-paths.csv": ( build_award_contract_paths_sql(args.path_limit) ), "procurement-lifecycle-links-graph-boundary-checks.csv": ( GRAPH_BOUNDARY_CHECKS_SQL ), } with tempfile.TemporaryDirectory( dir=data_dir, prefix=".procurement-lifecycle-links-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 = { "procurement-lifecycle-links-release-receipts.csv": 8, "procurement-lifecycle-links-edge-composition.csv": 21, "procurement-lifecycle-links-node-coverage.csv": 19, "procurement-lifecycle-links-award-supplier-paths.csv": min( args.path_limit, EXPECTED_GRAPH_METRICS["award_supplier_paths"], ), "procurement-lifecycle-links-award-contract-paths.csv": min( args.path_limit, EXPECTED_GRAPH_METRICS["award_contract_paths"], ), "procurement-lifecycle-links-graph-boundary-checks.csv": len( EXPECTED_GRAPH_METRICS ), } observed_output_rows = { name: evidence["rows"] for 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 = assert_recipe_unchanged( recipe_path, recipe_entry_identity, "immediately_before_provenance_receipt", ) inputs_before_receipt = reassert_input_files(data_dir) if inputs_before_receipt != input_evidence: raise RuntimeError( "Pinned input identity changed after DuckDB processing. " "No staged outputs were published." ) provenance = { "provenance_schema_version": "1.0", "recipe": "procurement-lifecycle-links-python", "recipe_version": "1.0", "recipe_file": { "entry_identity": recipe_entry_identity, "identity_immediately_before_receipt": recipe_before_receipt, "unchanged_during_run": True, }, "release_tag": RELEASE_TAG, "releases": { dataset_id: { "release_url": release_url(release), "generated_at": release["generated_at"], "schema_version": manifests[dataset_id]["schema_version"], "lifecycle_rows": release["lifecycle_rows"], } for dataset_id, release in PINNED_RELEASES.items() }, "path_limit": args.path_limit, "exclusive_data_directory_lock": lock_evidence, "runtime_contract": { "python_minimum": "3.10", "python_version": platform.python_version(), "duckdb_version": duckdb.__version__, "pandas_used": False, }, "input_files": input_evidence, "json_validation": json_evidence, "csv_headers": header_evidence, "duckdb_materialization": varchar_evidence, "graph_validation": graph_evidence, "outputs": outputs, "queries": queries, "interpretation": { "identifiers": ( "Every edge uses exact_official_identifier with confidence " "1.0 and is_inferred=False. No fuzzy or name-based entity " "resolution is performed." ), "paths": ( "Path CSVs are deterministic samples bounded by path_limit; " "the graph-boundary CSV reports the complete path counts." ), "orphans": ( "All award outgoing edges resolve to a source-matched award " "declaration inside the award graph." ), "boundary": ( "The two releases are not a closed world. The 34 award-only " "and 462 tender-only contract targets are pinned release-scope " "boundaries, not orphan award edges." ), "authority": ( "Use source_url and the current official source record for " "decision-critical facts." ), }, } provenance_name = "procurement-lifecycle-links-provenance.json" staged_provenance = staging_dir / provenance_name staged_provenance.write_text( json.dumps(provenance, indent=2, sort_keys=True) + "\n", encoding="utf-8", ) # Recheck executable and inputs at the publication boundary. os.replace # then promotes each fully written same-filesystem temporary atomically. assert_recipe_unchanged( recipe_path, recipe_entry_identity, "immediately_before_output_publication", ) inputs_before_publication = reassert_input_files(data_dir) if inputs_before_publication != input_evidence: raise RuntimeError( "Pinned input identity changed at publication. No staged " "outputs were published." ) for file_name in queries: os.replace(staging_dir / file_name, data_dir / file_name) os.replace(staged_provenance, data_dir / provenance_name) print(f"wrote {provenance_name}") finally: connection.close() print( "verified checkpoints: 162,684 tender edges; 88,817 award edges; " "57,469 award-to-supplier paths; 885 award-to-contract paths; 0 orphan " "award edges" ) def main() -> None: recipe_path = Path(__file__).resolve() recipe_entry_identity = file_identity(recipe_path) if sys.version_info < (3, 10): raise RuntimeError( f"This recipe requires Python 3.10+; found {platform.python_version()}." ) 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()