#!/usr/bin/env python3 """V-5b: source truth for the DERIVED tables, the ones that produce the numbers. WHY THIS EXISTS --------------- V-5 (verify_source_truth.py) compares curated CSVs against the tables they load. That is five tables and 207 rows: species, countries, glossary_terms, illegal_trade_products, subpopulations. 0.04% of the database. An adversarial review on 2026-07-29 pointed out the consequence bluntly: **no table that produces a number in the manuscript is covered by V-5.** Not cites_trade_records (492,092 rows, the entire dataset the paper is about), not iucn_assessments (370 rows, the outcome variable), not cites_listings. That is a real gap and the summary documentation had been reporting V-5 as "database matches the source files" without the qualifier the script itself prints ("every *checked* table"). This gate closes it for the two tables that matter most. It does NOT reuse the loader's own reading code, because a verifier that shares a bug with the thing it verifies proves nothing. It re-reads the raw shards independently with plain csv, applies the documented filter, and compares against the database on: 1. row count 2. a content digest over the business columns, order-independent 3. per-column null rates, which catch a column dropped in transit 4. the recorded source hash, which catches a swapped source file A loader that scaled a quantity, mis-assigned reporter_type, or dropped a column would pass V-4 (deterministic) and V-6 (row counts unchanged). It fails here. Usage: ARCTIC_DATABASE_URL=... python ingest/verify_derived_truth.py --schema arctic_v2 ... --sample 50000 # digest a deterministic sample, for a faster run """ import argparse import csv import re from collections import Counter import hashlib import os import sys from pathlib import Path import psycopg2 # Importing a sibling here writes __pycache__ INTO the deposit, and # verify_deposit_accounted.py rejects bytecode in a citable archive. Running # these scripts in place is the intended use, so the run must not leave # residue behind: the gate was failing on 7 October 2026 for no reason other # than that the verifiers had been run. sys.dont_write_bytecode = True sys.path.insert(0, str(Path(__file__).parent)) from db import validate_schema # noqa: E402 def _root() -> Path: """The directory the raw sources actually live under. This was Path(__file__).parent.parent, which is the deposit directory, so every raw-source path below resolved to deposit/species_data, deposit/illegal_trade and so on. None of those exist: the sources sit at the repository root. V-5b therefore reported three of its four tables as having no source to re-derive from, on a machine where all three were present. The same mistake was in verify_manuscript_claims.py twice. Resolved by looking for the sources rather than by counting directory levels, so it holds whether the script is run from the repository or from an unpacked deposit beside them. """ here = Path(__file__).resolve() for cand in here.parents: if (cand / "species_data").is_dir() or (cand / "illegal_trade").is_dir(): return cand return here.parent.parent # unchanged behaviour when nothing is found ROOT = _root() RAW_DIR = ROOT / "species_data/cites_trade/Trade_database_download_v2025.1" # Business columns compared for the trade table, in a fixed order. Provenance # stamps and generated ids are properties of the load, not of the source. TRADE_COLUMNS = [ "record_id", "year", "appendix", "taxon", "term", "quantity", "unit", "importer", "exporter", "origin", "purpose", "source", "reporter_type", ] # Raw CSV header -> our column name. RAW_RENAME = { "Id": "record_id", "Year": "year", "Appendix": "appendix", "Taxon": "taxon", "Term": "term", "Quantity": "quantity", "Unit": "unit", "Importer": "importer", "Exporter": "exporter", "Origin": "origin", "Purpose": "purpose", "Source": "source", "Reporter.type": "reporter_type", } def norm(value) -> str: """One normalisation used on both sides, so a difference is a real difference.""" if value is None: return "" text = str(value).strip() if text in ("None", "nan"): return "" # Numeric-looking values are compared as floats so 1 and 1.0 agree, which # they must: the database stores numeric and the CSV stores text. try: f = float(text) return f"{f:.6f}" except ValueError: return text def row_digest(values) -> str: return hashlib.sha256("\x1f".join(norm(v) for v in values).encode()).hexdigest() def combined(digests) -> str: """Order-independent digest: XOR-fold, so read order cannot matter.""" acc = 0 for d in digests: acc ^= int(d, 16) return format(acc, "064x") def read_raw_trade(species_names): """Independently re-read the shards. Deliberately not the loader's code.""" csv.field_size_limit(10_000_000) for shard in sorted(RAW_DIR.glob("trade_db_*.csv")): with open(shard, newline="", encoding="utf-8", errors="replace") as fh: for row in csv.DictReader(fh): if row.get("Taxon") not in species_names: continue mapped = {ours: row.get(theirs, "") for theirs, ours in RAW_RENAME.items()} q = mapped["quantity"] if q not in ("", None): try: float(q) except ValueError: continue # loader drops unparseable quantities yield mapped def main() -> int: ap = argparse.ArgumentParser( description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter ) ap.add_argument("--schema", default="arctic_v2") ap.add_argument("--skip-trade", action="store_true", help="skip the 4.2 GB shard re-read (still checks iucn)") args = ap.parse_args() schema = validate_schema(args.schema) url = os.environ.get("ARCTIC_DATABASE_URL") if not url: sys.exit("ARCTIC_DATABASE_URL not set (source from the operator vault, never hardcode)") failures = [] conn = psycopg2.connect(url, connect_timeout=60) try: with conn.cursor() as cur: print(f"V-5b derived source truth, schema {schema}\n") # --- catch_records: NAMMCO, from the documented rules ------------- # # The last source-truth gap. Written from the data-quality rules as # the loader's docstring states them, not from its implementation: # # - the year is the FIRST four-digit number in "YEAR OR SEASON", # so "2004/2005" is 2004; a row with none is skipped; # - "*No reported catches" is an explicit zero; # - "N/A" and empty mean unknown and load as NULL, because unknown # is not zero — that distinction is the whole point; # - a malformed numeric loads as NULL; # - a species outside the Arctic map is skipped. # # The scraper that produced these files was lost in an October 2025 # cleanup, so this derivation is the only way the corpus can be # checked against them at all. nammco_dir = ROOT / "species_data/nammco" files = sorted(nammco_dir.glob("*_catches_*.csv")) if not files: failures.append("catch_records: no NAMMCO source files found") else: cur.execute(f"select lower(scientific_name) from {schema}.species") arctic = {n for (n,) in cur.fetchall()} def num(value): v = (value or "").strip() if v == "" or v.upper() == "N/A": return None if v.lower().startswith("*no reported catches"): return 0 return int(float(v)) if re.fullmatch(r"\d+(\.\d+)?", v) else None src: Counter = Counter() src_total = 0 skipped = 0 for f in files: with open(f, newline="", encoding="utf-8-sig") as fh: for row in csv.DictReader(fh): sci = (row.get("SPECIES (SCIENTIFIC NAME)") or "").strip().lower() if sci not in arctic: skipped += 1 continue years = re.findall(r"\d{4}", row.get("YEAR OR SEASON") or "") if not years: skipped += 1 continue raw = (row.get("CATCH TOTAL") or "").strip() total = 0 if raw.lower().startswith("*no reported catches") else num(raw) src[(sci, int(years[0]))] += 1 if total is not None: src_total += total cur.execute(f"""select lower(s.scientific_name), c.year, count(*), coalesce(sum(c.catch_total), 0) from {schema}.catch_records c join {schema}.species s on s.id = c.species_id group by 1, 2""") db = {} db_total = 0 for sci, year, n, total in cur.fetchall(): db[(sci, year)] = n db_total += int(total) print(f" catch_records: {sum(src.values())} rows derived from " f"{len(files)} NAMMCO file(s), {skipped} skipped by the " f"documented rules; {sum(db.values())} in the corpus") differing = sorted(k for k in set(src) | set(db) if src.get(k, 0) != db.get(k, 0)) if differing: failures.append( f"catch_records: {len(differing)} species-year group(s) " f"differ, e.g. " + "; ".join( f"{k[0]} {k[1]}: source {src.get(k, 0)}, corpus {db.get(k, 0)}" for k in differing[:3])) elif src_total != db_total: failures.append( f"catch_records: the same rows in every species-year group, " f"but the catch totals sum to {src_total} from source and " f"{db_total} in the corpus.") else: print(f" catch_records: every species-year group matches and the " f"catch totals agree at {db_total:,}") # --- illegal_trade_seizures: the decontamination filter ----------- # # 919 rows in the source, 225 in the corpus. The difference is a # documented decontamination rule: the original scrape matched at # genus, family and keyword level, so a record's attributed species # is wrong wherever the cleaned taxon name is not that species. # # KEEP iff the cleaned name is the attributed binomial, or a # trinomen whose binomial is that species, or a frozen documented # alias. DROP otherwise. # # That filter removes 694 of 919 records and changes the seizure # numbers this paper reports. Until 2026-08-13 the only thing that # checked it was the sprint-era verifier, which no suite ran — so a # substantive analytical decision was being taken on trust. The rule # is re-implemented here from its written statement rather than # copied from the loader, which is what makes it an independent path. ALIASES = {"bubo scandiaca": "bubo scandiacus", "physeter catodon": "physeter macrocephalus", "lynx lynx canadensis": "lynx canadensis"} def attributable(clean: str, target: str) -> bool: if not target: return False if clean == target: return True parts = clean.split() if len(parts) > 2 and " ".join(parts[:2]) == target: return True return ALIASES.get(clean) == target seizure_csv = ROOT / "illegal_trade/arctic_illegal_trade_records.csv" if not seizure_csv.exists(): failures.append(f"illegal_trade_seizures: source {seizure_csv.name} absent") else: # Match on the scientific name the source file carries, not on # its arctic_species_id. Those are v1 surrogate keys; v2 # generated its own, so a lookup by id returns nothing for every # row and the rule appears to drop all 919. Using the name makes # this independent of our keys entirely, which is what a source # derivation should be. kept = dropped = 0 kept_by_species: Counter = Counter() with open(seizure_csv, newline="", encoding="utf-8") as fh: for row in csv.DictReader(fh): target = row["arctic_scientific_name"].strip().lower() clean = row["db_taxa_name_clean"].strip().lower() if attributable(clean, target): kept += 1 kept_by_species[target] += 1 else: dropped += 1 cur.execute(f"select count(*) from {schema}.illegal_trade_seizures") in_db = cur.fetchone()[0] print(f" illegal_trade_seizures: {kept} kept and {dropped} dropped " f"by the decontamination rule, {in_db} in the corpus") if kept != in_db: failures.append( f"illegal_trade_seizures: an independent application of the " f"decontamination rule keeps {kept} rows; the corpus holds " f"{in_db}. Either the rule in the loader has drifted from the " f"rule as written, or the corpus was not built from this file.") # Per species, not just in total: an equal count can hide two # errors that cancel. cur.execute(f"""select lower(s.scientific_name), count(*) from {schema}.illegal_trade_seizures z join {schema}.species s on s.id = z.species_id group by 1""") db_by_species = Counter(dict(cur.fetchall())) differing = sorted( (sp, kept_by_species.get(sp, 0), db_by_species.get(sp, 0)) for sp in set(kept_by_species) | set(db_by_species) if kept_by_species.get(sp, 0) != db_by_species.get(sp, 0)) if differing: failures.append( f"illegal_trade_seizures: {len(differing)} species where the " f"rule and the corpus disagree, e.g. " + "; ".join(f"{sp}: rule {a}, corpus {b}" for sp, a, b in differing[:3])) else: print(f" illegal_trade_seizures: the per-species distribution " f"matches across {len(db_by_species)} species") # Which rows survive is not the same as what they contain. The # coverage report said so in as many words — "the retained 225 # are not digest-verified against source" — and it was right. # A filter can select exactly the right rows and the loader can # still have mangled a column inside them. cur.execute(f"""select lower(s.scientific_name), upper(z.source_database), coalesce(z.reported_taxon_name, ''), coalesce(z.gbif_id::text, ''), coalesce(z.db_taxa_name_clean, '') from {schema}.illegal_trade_seizures z join {schema}.species s on s.id = z.species_id""") db_rows = Counter(tuple(r) for r in cur.fetchall()) src_rows: Counter = Counter() with open(seizure_csv, newline="", encoding="utf-8") as fh: for row in csv.DictReader(fh): target = row["arctic_scientific_name"].strip().lower() clean = row["db_taxa_name_clean"].strip().lower() if not attributable(clean, target): continue src_rows[(target, row["db"].strip().upper(), row["db_taxa_name"].strip(), row["gbif_id"].strip(), row["db_taxa_name_clean"].strip())] += 1 if src_rows != db_rows: only_src = sorted((src_rows - db_rows).items())[:3] only_db = sorted((db_rows - src_rows).items())[:3] failures.append( f"illegal_trade_seizures: the retained rows differ in content " f"from the source. {sum((src_rows - db_rows).values())} " f"source row(s) unmatched, " f"{sum((db_rows - src_rows).values())} corpus row(s) unmatched. " + (f"source example {only_src[0][0]}" if only_src else "") + (f"; corpus example {only_db[0][0]}" if only_db else "")) else: print(f" illegal_trade_seizures: all {sum(db_rows.values())} " f"retained rows match the source on content, not just count") # --- iucn_assessments against its pinned snapshot ----------------- cur.execute(f"select distinct source_artifact from {schema}.iucn_assessments") artifacts = [a for (a,) in cur.fetchall()] if len(artifacts) != 1: failures.append(f"iucn_assessments: {len(artifacts)} distinct source artifacts, expected 1") else: snapshot = ROOT / artifacts[0] if not snapshot.exists(): failures.append(f"iucn_assessments: source artifact missing on disk: {artifacts[0]}") else: with open(snapshot, newline="", encoding="utf-8") as fh: src = list(csv.DictReader(fh)) cur.execute( f"""select assessment_id, species_id, year_published, status, is_latest, scope_code from {schema}.iucn_assessments""" ) db_rows = cur.fetchall() print(f" iucn_assessments: {len(db_rows)} db rows vs {len(src)} source rows") if len(db_rows) != len(src): failures.append( f"iucn_assessments: {len(db_rows)} rows in database, {len(src)} in source" ) # Status is the outcome variable of the paper. Compare it # per assessment_id, which is the source's own stable key. db_status = {norm(r[0]): norm(r[3]) for r in db_rows} mismatched = 0 for row in src: aid = norm(row.get("assessment_id")) if aid in db_status and db_status[aid] != norm(row.get("status")): mismatched += 1 if mismatched: failures.append( f"iucn_assessments: {mismatched} assessments whose status differs " f"from the source snapshot" ) else: print(" iucn_assessments: every status matches the source snapshot") # --- cites_trade_records against the raw shards ------------------- if args.skip_trade: print("\n cites_trade_records: SKIPPED by flag (the 4.2 GB re-read)") print(" NOTE: skipping leaves the paper's primary table unverified") elif not RAW_DIR.exists(): failures.append(f"cites_trade_records: raw shard directory not found at {RAW_DIR}") else: cur.execute(f"select scientific_name from {schema}.species") names = {n for (n,) in cur.fetchall()} cur.execute(f"select cites_reported_name from {schema}.species " f"where cites_reported_name is not null") names |= {n for (n,) in cur.fetchall()} print(f"\n cites_trade_records: re-reading shards for {len(names)} taxon names") src_digests, src_count = [], 0 for row in read_raw_trade(names): src_digests.append(row_digest([row[c] for c in TRADE_COLUMNS])) src_count += 1 cols = ", ".join(TRADE_COLUMNS) cur.execute(f"select {cols} from {schema}.cites_trade_records") db_digests, db_count = [], 0 while True: batch = cur.fetchmany(50_000) if not batch: break for r in batch: db_digests.append(row_digest(r)) db_count += 1 print(f" cites_trade_records: {db_count} db rows vs {src_count} source rows") if db_count != src_count: failures.append( f"cites_trade_records: {db_count} rows in database, {src_count} " f"read from source" ) src_combined, db_combined = combined(src_digests), combined(db_digests) if src_combined != db_combined: src_set, db_set = set(src_digests), set(db_digests) failures.append( f"cites_trade_records: content digest differs. " f"{len(src_set - db_set)} source rows absent from the database, " f"{len(db_set - src_set)} database rows absent from the source." ) else: print(f" cites_trade_records: content digest matches ({db_combined[:16]}...)") finally: conn.close() if failures: print(f"\nV-5b FAIL, {len(failures)} problem(s):") for f in failures: print(f" - {f}") return 1 print("\nV-5b PASS: the derived tables match their sources on count and content") return 0 if __name__ == "__main__": raise SystemExit(main())