File size: 7,788 Bytes
83db774 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 | #!/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()
|