| |
| """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() |
|
|
