#!/usr/bin/env python3 """Audit existing Poucher substantivity rows for identity-resolution risks.""" from __future__ import annotations import json from collections import Counter from pathlib import Path from typing import Any DATA = Path("data") ARTIFACTS = Path("artifacts") MEASURED = DATA / "poucher_substantivity.jsonl" CANDIDATES = DATA / "poucher_substantivity_candidates.jsonl" OUT_JSON = ARTIFACTS / "poucher_substantivity_identity_audit.json" OUT_MD = ARTIFACTS / "poucher_substantivity_identity_audit.md" def load_jsonl(path: Path) -> list[dict[str, Any]]: return [json.loads(line) for line in path.read_text().splitlines() if line.strip()] HARD_IDENTITY_ERRORS = { "106-25-2": { "status": "quarantine_or_split", "reason": "Neroli rows were captured by the shorter 'nerol' prefix and collapsed into nerol.", "bad_source_names": ["Neroli, Italian", "Neroli bigarade"], }, "110-41-8": { "status": "quarantine_or_split", "reason": "A bare OCR/source name 'aldehyde' was collapsed into methyl nonyl acetaldehyde.", "bad_source_names": ["aldehyde"], }, "111-12-6": { "status": "split", "reason": "Methyl octine carbonate resolves separately from methyl heptine carbonate.", "bad_source_names": ["Methyl octine carbonate"], "suggested_cas": {"Methyl octine carbonate": "111-80-8"}, }, "115-95-7": { "status": "split", "reason": "Linalyl salicylate was assigned the linalyl acetate CAS.", "bad_source_names": ["Linalyl salicylate"], "suggested_cas": {"Linalyl salicylate": "7149-28-2"}, }, "8006-90-4": { "status": "quarantine_or_split", "reason": "Pepper was captured by the longer peppermint key.", "bad_source_names": ["Pepper"], }, "8007-01-0": { "status": "split", "reason": "Rosemary, French was captured by the shorter rose key.", "bad_source_names": ["Rosemary, French"], "suggested_cas": {"Rosemary, French": "8000-25-7"}, }, "8023-70-5": { "status": "quarantine_or_split", "reason": "Ginger was collapsed with gingergrass through prefix matching.", "bad_source_names": ["Ginger"], }, "8023-85-4": { "status": "quarantine_or_split", "reason": "Cassie absolute, Farnesiana was collapsed into the orris CAS bucket.", "bad_source_names": ["Cassie absolute, Farnesiana"], }, "8006-87-9": { "status": "split", "reason": "Santalyl phenylacetate was captured by the shorter santal key.", "bad_source_names": ["Santalyl phenylacetate"], "suggested_cas": {"Santalyl phenylacetate": "1323-75-7"}, }, } BROAD_NATURAL_COLLAPSES = { "8000-46-2": "Geranium origins are collapsed to one broad natural CAS.", "8000-48-4": "Eucalyptus and Eucalyptus citriodora are collapsed to one broad natural CAS.", "8007-46-3": "Thyme red and thyme white are collapsed to one broad natural CAS.", "8014-17-3": "Petitgrain origins are collapsed to one broad natural CAS.", "8015-64-3": "Angelica seed and root are collapsed to one broad natural CAS.", "8015-91-6": "Cinnamon leaf and bark are collapsed to one broad natural CAS.", "8021-15-0": "Opoponax oil and resin are collapsed to one broad natural CAS.", "8023-82-3": "Myrrh oil and resin are collapsed to one broad natural CAS.", "8023-91-4": "Galbanum oil and resin are collapsed to one broad natural CAS.", } def source_names(row: dict[str, Any]) -> list[str]: return [source["name"] for source in row.get("source_rows", [])] def main() -> None: measured = load_jsonl(MEASURED) candidates = load_jsonl(CANDIDATES) if CANDIDATES.exists() else [] by_cas = {row["cas"]: row for row in measured} candidate_conflicts = [ row for row in candidates if row.get("dedupe_against_existing_measured", {}).get("tag") == "conflict" ] issues = [] for cas, spec in HARD_IDENTITY_ERRORS.items(): row = by_cas.get(cas) if not row: continue present_bad_names = sorted(set(spec["bad_source_names"]) & set(source_names(row))) if not present_bad_names: continue issues.append({ "severity": "hard_identity_error", "cas": cas, "measured_name": row["name"], "measured_coefficient": row["poucher_coefficient"], "all_poucher_coefficients": row.get("all_poucher_coefficients", []), "source_names": source_names(row), "bad_source_names": present_bad_names, "status": spec["status"], "reason": spec["reason"], "suggested_cas": spec.get("suggested_cas", {}), }) broad = [] for cas, reason in BROAD_NATURAL_COLLAPSES.items(): row = by_cas.get(cas) if not row or len(set(source_names(row))) < 2: continue broad.append({ "severity": "broad_natural_collapse", "cas": cas, "measured_name": row["name"], "measured_coefficient": row["poucher_coefficient"], "all_poucher_coefficients": row.get("all_poucher_coefficients", []), "source_names": source_names(row), "status": "human_review_before_public_label", "reason": reason, }) recommendation = ( "Do not grow or publish labels until hard_identity_error rows are split " "or quarantined. Broad natural collapses can remain only with an explicit " "natural-product label policy." ) if not issues: recommendation = ( "No hard identity-error rows remain. Broad natural collapses can remain " "only with an explicit natural-product label policy." ) summary = { "measured_rows_audited": len(measured), "candidate_rows_compared": len(candidates), "candidate_conflicts": len(candidate_conflicts), "hard_identity_error_rows": len(issues), "broad_natural_collapse_rows": len(broad), "severity_counts": dict(Counter(item["severity"] for item in issues + broad)), "recommendation": recommendation, "hard_identity_errors": issues, "broad_natural_collapses": broad, } ARTIFACTS.mkdir(exist_ok=True) OUT_JSON.write_text(json.dumps(summary, indent=2, sort_keys=True) + "\n") lines = [ "# Poucher substantivity identity audit", "", f"- Measured rows audited: {len(measured)}", f"- Candidate rows compared: {len(candidates)}", f"- Candidate conflicts surfaced: {len(candidate_conflicts)}", f"- Hard identity-error rows: {len(issues)}", f"- Broad natural-collapse rows: {len(broad)}", "", "## Hard identity errors", "", ] for item in issues: fixes = item.get("suggested_cas") or {} fix_text = "; suggested split " + ", ".join(f"{name} -> {cas}" for name, cas in fixes.items()) if fixes else "; quarantine unresolved source name(s)" lines.append( f"- {item['cas']} {item['measured_name']} coeff {item['measured_coefficient']} " f"from {item['source_names']}: {item['reason']}{fix_text}" ) lines.extend(["", "## Broad natural collapses", ""]) for item in broad: lines.append( f"- {item['cas']} {item['measured_name']} coeff {item['measured_coefficient']} " f"from {item['source_names']}: {item['reason']}" ) lines.extend(["", "## Recommendation", "", summary["recommendation"], ""]) OUT_MD.write_text("\n".join(lines)) print(f"wrote {OUT_JSON}") print(f"wrote {OUT_MD}") if __name__ == "__main__": main()