scripts/build_odour_taxonomy_census_v8.py

#!/usr/bin/env python3
"""Build the v8 26-class odour taxonomy and coverage census.

This script intentionally stops at data/census artifacts. It does not assemble a
training dataset or launch training.
"""
from __future__ import annotations

from collections import Counter, defaultdict
import json
from pathlib import Path
import re
import sqlite3
from typing import Any


ROOT = Path(__file__).resolve().parents[1]
DATA = ROOT / "data"
ARTIFACTS = ROOT / "artifacts"

TAXONOMY_PATH = DATA / "odour_taxonomy_26.json"
ANCHOR_REPORT_PATH = ARTIFACTS / "odour_taxonomy_anchor_report_v8.json"
CROSSWALK_PATH = ARTIFACTS / "odour_character_crosswalk_v8.json"
CROSSWALK_V81_PATH = ARTIFACTS / "odour_character_crosswalk_v8_1.json"
CROSSWALK_CHANGELOG_PATH = ARTIFACTS / "odour_character_crosswalk_v8_1_changelog.json"
LABELS_PATH = DATA / "odour_character_labels_v8.jsonl"
CENSUS_PATH = ARTIFACTS / "odour_taxonomy_coverage_census_v8.json"
SUMMARY_PATH = ARTIFACTS / "odour_taxonomy_coverage_census_v8.md"
SOURCE_COVERAGE_PATH = ARTIFACTS / "odour_character_source_coverage_v8_1.json"

FORMULA_DATASETS = [
    DATA / "empirical_dataset_v8.jsonl",
]
ANCHOR_RESOLUTION_DATASETS = [
    DATA / "tgsc_training_formulas_v8.jsonl",
    DATA / "wisemoor_training_records_v9.jsonl",
    DATA / "fraterworks_free_formulas.jsonl",
]
SUBSTANTIVITY_PATH = DATA / "substantivity_targets_poucher_enriched_v8_phase1.jsonl"
ARCTANDER_PATH = DATA / "literature_flat" / "arctander_monographs.jsonl"

SOLVENT_CAS = {"64-17-5", "67-63-0", "25265-71-8", "84-66-2", "110-98-5", "57-55-6"}


TAXONOMY_ROWS: list[dict[str, Any]] = [
    {"code": "A", "name": "ALI-FAT-IC", "description": "Fatty, waxy, soapy, clean", "reference_materials": ["Aliphatic aldehydes", "alcohols"]},
    {"code": "B", "name": "Berg-ICEBERG", "description": "Cooling, borneol, mint, camphor", "reference_materials": ["Menthol", "camphor", "eucalyptol"]},
    {"code": "C", "name": "CITRUS", "description": "Sour, sharp, citrus peel", "reference_materials": ["Citral", "orange", "lemon", "lime"]},
    {"code": "D", "name": "DAIRY", "description": "Milky, cream, butter, cheese", "reference_materials": ["Diacetyl", "butyrate", "lactone", "valerate"]},
    {"code": "E", "name": "EDIBLE", "description": "Vegetable, nut, fish, meat", "reference_materials": ["Thiazoles", "pyrazines", "sulphides"]},
    {"code": "F", "name": "FRUIT", "description": "Sour, sweet fruits, strawberry", "reference_materials": ["Allyl caproate", "Verdox"]},
    {"code": "G", "name": "GREEN", "description": "Cut-grass, leaves", "reference_materials": ["cis-3-Hexenol", "Triplal"]},
    {"code": "H", "name": "HERB (Cool)", "description": "Cool herbaceous notes", "reference_materials": ["Lavender", "sage"]},
    {"code": "I", "name": "IRIS", "description": "Orris, violet", "reference_materials": ["Ionones", "methyl ionone"]},
    {"code": "J", "name": "JASMIN", "description": "Fruity, oily, narcotic, jasmin", "reference_materials": ["HCA", "benzyl acetate"]},
    {"code": "K", "name": "KONIFER", "description": "Pine, pine needle", "reference_materials": ["Terpineol", "bornyl acetate"]},
    {"code": "L", "name": "LIGHT Chemical Floral", "description": "Fresh light floral chemical", "reference_materials": ["Linalool", "Vertenex", "DMBC"]},
    {"code": "M", "name": "MUGUET", "description": "Lily of the valley, green, fresh", "reference_materials": ["Hydroxy", "Lilial", "Lyral"]},
    {"code": "N", "name": "NARCOTIC", "description": "Heavy sweet florals, absolutes", "reference_materials": ["Narcissus", "ylang ylang", "tuberose"]},
    {"code": "O", "name": "ORCHID", "description": "Aromatic, deep floral", "reference_materials": ["Salicylates", "benzoates"]},
    {"code": "P", "name": "PHENOL", "description": "Phenol, medicinal, honey", "reference_materials": ["p-Cresol", "ethyl phenyl acetate"]},
    {"code": "Q", "name": "Queen of the ORIENT", "description": "Resin, balsam", "reference_materials": ["Benzoin", "tolu", "terpenes"]},
    {"code": "R", "name": "ROSE", "description": "Rose otto, absolute, geranium", "reference_materials": ["Citronellol", "PEA", "rhodinol"]},
    {"code": "S", "name": "SPICE (Hot)", "description": "Hot culinary, spice", "reference_materials": ["Clove", "cinnamon", "thyme"]},
    {"code": "T", "name": "TAR SMOKE", "description": "Smoke, tar, burnt", "reference_materials": ["Cade", "birch tar"]},
    {"code": "U", "name": "Urine Faecal ANIMAL", "description": "Animal, faecal, leather", "reference_materials": ["Civet", "castoreum", "ambergris"]},
    {"code": "V", "name": "VANILLA", "description": "Sweet edible, vanilla", "reference_materials": ["Vanillin", "coumarin", "heliotropin"]},
    {"code": "W", "name": "WOOD", "description": "Wood, oily", "reference_materials": ["Cedar", "santal", "vetivert", "patchouli"]},
    {"code": "X", "name": "X-rated MUSK", "description": "Sexy, musk, sensual, sweet", "reference_materials": ["Musk ketone", "Galaxolide"]},
    {"code": "Y", "name": "EARTHY MOSSY", "description": "Yeast, fungal, moss, marine", "reference_materials": ["Oakmoss", "Calone"]},
    {"code": "Z", "name": "ZOLVENTS", "description": "Odourless solvents, solubilisers", "reference_materials": ["DEP", "DPG", "IPP", "ethanol", "PG"]},
]


# Only these non-specific anchors are treated as family/natural-complex anchors.
FAMILY_ANCHORS = {
    "aliphatic aldehydes", "alcohols", "orange", "lemon", "lime", "butyrate",
    "lactone", "valerate", "thiazoles", "pyrazines", "sulphides", "lavender",
    "sage", "ionones", "narcissus", "ylang ylang", "tuberose", "salicylates",
    "benzoates", "benzoin", "tolu", "terpenes", "clove", "cinnamon", "thyme",
    "cade", "birch tar", "civet", "castoreum", "ambergris", "cedar", "santal",
    "vetivert", "oakmoss",
}


# Explicit single-material aliases. These avoid substring matching and document
# abbreviations present in the source table.
ANCHOR_ALIASES = {
    "hca": "alpha hexyl cinnamic aldehyde",
    "pea": "phenylethyl alcohol",
    "dmbc": "dimethyl benzyl carbinyl acetate",
    "hydroxy": "hydroxycitronellal",
    "ipp": "isopropyl palmitate",
    "pg": "propylene glycol",
    "diacetyl": "diacetyl",
    "eucalyptol": "eucalyptol",
    "terpineol": "terpineol",
    "methyl ionone": "methylionone",
    "allyl caproate": "allyl hexanoate",
    "patchouli": "patchouli alcohol",
    "ethyl phenyl acetate": "ethyl phenylacetate",
}


# Exact CAS fallbacks for source-table single anchors absent from the local exact
# alias index. They are only used after exact normalized lookup fails.
EXACT_SINGLE_ANCHOR_FALLBACKS = {
    "diacetyl": {"cas": "431-03-8", "name": "Diacetyl", "smiles": "CC(=O)C(C)=O"},
    "hca": {"cas": "101-86-0", "name": "alpha-Hexylcinnamaldehyde", "smiles": "CCCCCCC(=CC1=CC=CC=C1)C=O"},
    "pea": {"cas": "60-12-8", "name": "Phenylethyl alcohol", "smiles": "C1=CC=C(C=C1)CCO"},
    "hydroxy": {"cas": "107-75-5", "name": "Hydroxycitronellal", "smiles": "CC(CO)CCC=C(C)C=O"},
    "ipp": {"cas": "142-91-6", "name": "Isopropyl palmitate", "smiles": "CCCCCCCCCCCCCCCC(=O)OC(C)C"},
    "pg": {"cas": "57-55-6", "name": "Propylene glycol", "smiles": "CC(O)CO"},
}


DESCRIPTOR_CROSSWALK = {
    "A": ["fatty", "waxy", "soapy", "clean", "aldehydic"],
    "B": ["cooling", "borneol", "mint", "minty", "menthol", "camphor", "camphoraceous"],
    "C": ["citrus", "orange", "lemon", "lime", "bergamot", "citrus peel"],
    "D": ["milky", "cream", "creamy", "butter", "buttery", "cheese", "cheesy", "dairy"],
    "E": ["vegetable", "nut", "nutty", "fish", "fishy", "meat", "meaty", "sulphide", "sulfide", "pyrazine", "thiazole"],
    "F": ["fruit", "fruity", "strawberry", "apple", "peach", "pear", "berry", "apricot", "pineapple"],
    "G": ["green", "cut grass", "grassy", "leaf", "leafy", "foliage"],
    "H": ["herb", "herbal", "herbaceous", "lavender", "sage"],
    "I": ["iris", "orris", "violet", "ionone"],
    "J": ["jasmin", "jasmine"],
    "K": ["pine", "pine needle", "conifer", "terpineol"],
    "L": ["fresh floral", "light floral", "linalool"],
    "M": ["muguet", "lily of the valley"],
    "N": ["heavy floral", "ylang", "tuberose", "narcissus", "narcotic"],
    "O": ["orchid", "deep floral"],
    "P": ["phenol", "phenolic", "medicinal", "honey"],
    "Q": ["resin", "resinous", "balsam", "balsamic", "benzoin", "tolu"],
    "R": ["rose", "rosy", "geranium", "citronellol", "rhodinol"],
    "S": ["spice", "spicy", "clove", "cinnamon", "thyme", "hot spice"],
    "T": ["smoke", "smoky", "tar", "burnt", "burned", "cade", "birch tar"],
    "U": ["animal", "animalic", "faecal", "fecal", "leather", "leathery", "civet", "castoreum", "ambergris", "urine"],
    "V": ["vanilla", "vanillic", "coumarin", "coumarinic", "heliotropin", "tonka"],
    "W": ["wood", "woody", "cedar", "sandalwood", "santal", "vetiver", "vetivert", "patchouli"],
    "X": ["musk", "musky", "sensual", "galaxolide"],
    "Y": ["earthy", "mossy", "moss", "fungal", "marine", "oakmoss", "calone", "seaweed"],
}

AMBIGUOUS_UNMAPPED_DESCRIPTORS = {
    "aromatic": ["H", "O", "Q"],
    "floral": ["J", "L", "M", "N", "O", "R"],
    "fresh": ["A", "C", "G", "L", "M"],
    "oily": ["J", "W"],
    "powdery": ["I", "V", "X"],
    "sweet": ["F", "N", "V", "X"],
}

CROSSWALK_CHANGELOG = [
    {
        "descriptor": "narcotic",
        "old": "J",
        "new": "N",
        "reason": "The taxonomy defines N as NARCOTIC; mapping the descriptor to JASMIN made a specific N-class descriptor collapse into class J.",
    },
    {
        "descriptor": "powdery",
        "old": "I",
        "new": None,
        "reason": "Powdery is multi-class-capable across iris, heliotrope/vanilla, and musk contexts; it is masked unless accompanied by a more specific descriptor.",
    },
    {
        "descriptor": "aromatic",
        "old": "O",
        "new": None,
        "reason": "Aromatic is multi-class-capable in source prose and is too broad to force into ORCHID without a deep-floral/orchid qualifier.",
    },
]


def norm(text: str) -> str:
    return re.sub(r"[^a-z0-9]+", " ", str(text).lower()).strip()


def compact_name(text: str) -> str:
    """Conservative exact-match alias cleanup for trade names and dilutions."""
    text = str(text)
    text = re.sub(r"[®™]", "", text)
    text = re.sub(r"[\"“”]", "", text)
    text = re.sub(r"\([^)]*\)", " ", text)
    text = re.sub(r"\b\d+(?:\.\d+)?\s*%.*$", " ", text, flags=re.IGNORECASE)
    text = re.sub(
        r"\b(signature|supra|crist|crystal|pure|extra|absolute|resinoid|oil|eo)\b",
        " ",
        text,
        flags=re.IGNORECASE,
    )
    return norm(text)


def alias_keys(text: str) -> set[str]:
    keys = {norm(text), compact_name(text)}
    return {key for key in keys if key}


def word_pattern(term: str) -> re.Pattern[str]:
    parts = [re.escape(p) for p in norm(term).split()]
    return re.compile(r"(?<![a-z0-9])" + r"\s+".join(parts) + r"(?![a-z0-9])")


def inchi_key(smiles: str | None) -> str | None:
    if not smiles or smiles.startswith("NATURAL:"):
        return None
    try:
        from rdkit import Chem
        from rdkit.Chem import inchi

        mol = Chem.MolFromSmiles(smiles.removeprefix("SMILES:"))
        return inchi.MolToInchiKey(mol) if mol is not None else None
    except Exception:
        return None


def formula_components(record: dict[str, Any]) -> list[dict[str, Any]]:
    return record.get("metadata", {}).get("initial_components") or record.get("formula", [])


def load_jsonl(path: Path) -> list[dict[str, Any]]:
    with path.open() as f:
        return [json.loads(line) for line in f if line.strip()]


def load_formula_records() -> list[dict[str, Any]]:
    records: list[dict[str, Any]] = []
    for path in FORMULA_DATASETS:
        records.extend(load_jsonl(path))
    return records


def build_alias_index(records: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
    aliases: dict[str, dict[str, Any]] = {}

    db = ROOT / "src" / "pino" / "registry.db"
    con = sqlite3.connect(db)
    con.row_factory = sqlite3.Row
    for row in con.execute("select cas,name,smiles from aroma_chemicals"):
        rec = dict(row)
        if rec.get("name"):
            for key in alias_keys(rec["name"]):
                aliases.setdefault(key, rec)
        if rec.get("cas"):
            aliases.setdefault(norm(rec["cas"]), rec)
    con.close()

    resolution_records = list(records)
    for path in ANCHOR_RESOLUTION_DATASETS:
        if path.exists():
            resolution_records.extend(load_jsonl(path))

    for record in resolution_records:
        for comp in formula_components(record):
            if comp.get("name") and comp.get("cas"):
                for key in alias_keys(comp["name"]):
                    aliases.setdefault(key, {
                        "cas": comp.get("cas"),
                        "name": comp.get("name"),
                        "smiles": comp.get("smiles") or "",
                    })
            if comp.get("cas"):
                aliases.setdefault(norm(comp["cas"]), {
                    "cas": comp.get("cas"),
                    "name": comp.get("name") or "",
                    "smiles": comp.get("smiles") or "",
                })
    return aliases


def resolve_anchor(term: str, alias_index: dict[str, dict[str, Any]]) -> dict[str, Any]:
    key = norm(term)
    if key in FAMILY_ANCHORS:
        return {
            "term": term,
            "status": "class-anchor-family",
            "provenance": "taxonomy-reference",
            "reason": "family_or_natural_complex_not_single_molecule",
        }

    lookup_key = norm(ANCHOR_ALIASES.get(key, key))
    rec = alias_index.get(lookup_key)
    method = "exact-normalized-local-alias"
    if rec is None and key in EXACT_SINGLE_ANCHOR_FALLBACKS:
        rec = EXACT_SINGLE_ANCHOR_FALLBACKS[key]
        method = "exact-single-anchor-fallback"

    if rec is None:
        return {
            "term": term,
            "status": "unresolved-single-anchor",
            "provenance": "taxonomy-reference",
            "lookup_key": lookup_key,
        }

    smiles = rec.get("smiles") or ""
    ikey = inchi_key(smiles)
    return {
        "term": term,
        "status": "resolved-single-molecule" if ikey else "resolved-cas-only",
        "provenance": "taxonomy-reference",
        "resolution_method": method,
        "canonical": {
            "cas": rec.get("cas"),
            "name": rec.get("name"),
            "smiles": smiles or None,
            "inchikey": ikey,
        },
    }


def build_taxonomy(alias_index: dict[str, dict[str, Any]]) -> tuple[dict[str, Any], dict[str, list[dict[str, Any]]]]:
    by_class: dict[str, list[dict[str, Any]]] = {}
    classes = []
    for row in TAXONOMY_ROWS:
        anchors = [resolve_anchor(term, alias_index) for term in row["reference_materials"]]
        by_class[row["code"]] = anchors
        classes.append({
            **row,
            "is_carrier": row["code"] == "Z",
            "reference_material_anchors": anchors,
        })
    taxonomy = {
        "metadata": {
            "pimt_version": "v8",
            "title": "Systematic Odour Classification Groups For Perfumery",
            "publisher": "Professional Perfumer's World / Perfumer's Bulletin",
            "source_location": "sample table p.35",
            "provenance": "structured-from-authoritative-source",
            "notes": [
                "Reference terms are transcribed from the supplied taxonomy work order.",
                "Family or natural-complex anchors are retained as class anchors but excluded from single-molecule label counts.",
            ],
        },
        "classes": classes,
    }
    return taxonomy, by_class


def material_universe(records: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
    universe: dict[str, dict[str, Any]] = {}
    for record in records:
        for comp in formula_components(record):
            cas = str(comp.get("cas") or "").strip()
            if not cas:
                continue
            entry = universe.setdefault(cas, {
                "cas": cas,
                "names": set(),
                "smiles": comp.get("smiles") or "",
                "formula_count": 0,
            })
            if comp.get("name"):
                entry["names"].add(str(comp["name"]))
            if comp.get("smiles") and not entry.get("smiles"):
                entry["smiles"] = comp.get("smiles")
            entry["formula_count"] += 1
    for entry in universe.values():
        entry["names"] = sorted(entry["names"])
    return universe


def load_substantivity_targets() -> dict[str, dict[str, Any]]:
    targets: dict[str, dict[str, Any]] = {}
    if SUBSTANTIVITY_PATH.exists():
        for row in load_jsonl(SUBSTANTIVITY_PATH):
            targets[str(row["cas"])] = row
    return targets


def anchor_labels(by_class: dict[str, list[dict[str, Any]]]) -> dict[str, dict[str, Any]]:
    labels: dict[str, dict[str, Any]] = {}
    for code, anchors in by_class.items():
        for anchor in anchors:
            if anchor["status"] not in {"resolved-single-molecule", "resolved-cas-only"}:
                continue
            cas = anchor["canonical"].get("cas")
            if not cas:
                continue
            labels[cas] = {
                "cas": cas,
                "class_code": code,
                "provenance": "taxonomy-reference",
                "source_term": anchor["term"],
                "source": "Systematic Odour Classification Groups For Perfumery sample table p.35",
            }
    return labels


def registry_records() -> dict[str, dict[str, Any]]:
    rows: dict[str, dict[str, Any]] = {}
    db = ROOT / "src" / "pino" / "registry.db"
    con = sqlite3.connect(db)
    con.row_factory = sqlite3.Row
    for row in con.execute("select cas,name,smiles from aroma_chemicals"):
        rec = dict(row)
        if rec.get("cas"):
            rows[str(rec["cas"])] = rec
    con.close()
    return rows


def enrich_universe(
    universe: dict[str, dict[str, Any]],
    subst: dict[str, dict[str, Any]],
    labels: dict[str, dict[str, Any]],
) -> dict[str, dict[str, Any]]:
    enriched = {
        cas: {**entry, "names": set(entry.get("names", []))}
        for cas, entry in universe.items()
    }
    registry = registry_records()
    for cas in set(subst) | set(labels) | set(registry):
        if cas not in set(subst) | set(labels) | set(universe):
            continue
        entry = enriched.setdefault(cas, {"cas": cas, "names": set(), "smiles": "", "formula_count": 0})
        if cas in subst and subst[cas].get("name"):
            entry["names"].add(str(subst[cas]["name"]))
        if cas in labels and labels[cas].get("source_term"):
            entry["names"].add(str(labels[cas]["source_term"]))
        if cas in registry:
            if registry[cas].get("name"):
                entry["names"].add(str(registry[cas]["name"]))
            if registry[cas].get("smiles") and not entry.get("smiles"):
                entry["smiles"] = registry[cas].get("smiles") or ""
    for entry in enriched.values():
        entry["names"] = sorted(entry["names"])
    return enriched


def load_arctander_records() -> list[dict[str, Any]]:
    out: list[dict[str, Any]] = []
    if not ARCTANDER_PATH.exists():
        return out
    for line in ARCTANDER_PATH.read_text().splitlines():
        if not line.strip():
            continue
        outer = json.loads(line)
        rec = json.loads(outer["record"]) if isinstance(outer.get("record"), str) else outer
        out.append(rec)
    return out


def arctander_aliases(rec: dict[str, Any]) -> set[str]:
    aliases = set()
    name = str(rec.get("name") or "")
    aliases.update(alias_keys(name))
    raw_lines = [line.strip() for line in str(rec.get("raw_text") or "").splitlines()]
    title_lines: list[str] = []
    for line in raw_lines[1:8]:
        if not line:
            continue
        alpha = re.sub(r"[^A-Za-z]+", "", line)
        if not alpha:
            break
        uppercase_ratio = sum(1 for ch in alpha if ch.isupper()) / max(1, len(alpha))
        if "." in line or uppercase_ratio < 0.65:
            break
        title_lines.append(line)
    if len(title_lines) > 1:
        aliases.update(alias_keys(" ".join(title_lines)))
    # Synonyms are noisy OCR, so only accept short delimited aliases. This
    # remains exact matching after normalization and catches period/newline
    # synonym fragments such as "Anisic alcohol." without substring matching.
    synonyms = str(rec.get("synonyms") or "")
    for part in re.split(r"[,;/.\n]", synonyms):
        part = part.strip()
        if 3 <= len(part) <= 60 and not re.search(r"\d", part):
            aliases.update(alias_keys(part))
    return {alias for alias in aliases if alias}


def build_arctander_index(records: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
    candidates: dict[str, list[dict[str, Any]]] = defaultdict(list)
    for rec in records:
        for alias in arctander_aliases(rec):
            candidates[alias].append(rec)
    return {
        alias: matches[0]
        for alias, matches in candidates.items()
        if len({m.get("monograph_number") for m in matches}) == 1
    }


def material_aliases(entry: dict[str, Any]) -> set[str]:
    aliases = set()
    for name in entry.get("names", []):
        aliases.update(alias_keys(name))
    return aliases


def source_odor_text(rec: dict[str, Any]) -> str:
    raw = str(rec.get("raw_text") or "")
    raw = re.sub(r"-\s*\n\s*", "", raw)
    raw = re.sub(r"\s*\n\s*", " ", raw)
    raw = re.sub(r"\s+", " ", raw).strip()
    sentences = [
        s.strip()
        for s in re.split(r"(?<=[.!?])\s+", raw)
        if s.strip()
    ]
    odor_sentences = [
        s
        for s in sentences
        if re.search(r"\b(?:odou?r|smell|aroma)\b", s, flags=re.IGNORECASE)
    ]
    text = " ".join(odor_sentences)
    desc = str(rec.get("odor_description") or "").strip()
    if desc and norm(desc) not in {"intermittent", "not available", "none"}:
        text = f"{desc}. {text}".strip()
    text = re.sub(r"\b[Tt]aste\b.*?(?=\.|$)", " ", text)
    text = re.sub(r"\bIt\s*blends\b[^.]*\bodou?r type\b", " ", text, flags=re.IGNORECASE)
    text = re.sub(r"\bIt\s*may find\b.*", " ", text, flags=re.IGNORECASE)
    text = re.sub(r"\bMuguet bases\b", "bases", text, flags=re.IGNORECASE)
    text = re.sub(r"\b([A-Za-z]+)\s+leaf oil\b", r"\1 oil", text, flags=re.IGNORECASE)
    text = re.sub(r"\s+", " ", text).strip()
    return text


def crosswalk_patterns() -> dict[str, list[tuple[str, re.Pattern[str]]]]:
    return {
        code: [(term, word_pattern(term)) for term in terms]
        for code, terms in DESCRIPTOR_CROSSWALK.items()
    }


def apply_crosswalk(text: str) -> tuple[list[str], dict[str, list[str]], list[str]]:
    text_norm = norm(text)
    hits_by_class: dict[str, set[str]] = defaultdict(set)
    for code, code_patterns in crosswalk_patterns().items():
        for term, pat in code_patterns:
            if pat.search(text_norm):
                hits_by_class[code].add(term)
    ambiguous_hits = [
        term for term in sorted(AMBIGUOUS_UNMAPPED_DESCRIPTORS)
        if word_pattern(term).search(text_norm)
    ]
    return (
        sorted(hits_by_class),
        {code: sorted(terms) for code, terms in hits_by_class.items()},
        ambiguous_hits,
    )


def arctander_labels(
    universe: dict[str, dict[str, Any]],
    existing: dict[str, dict[str, Any]],
) -> tuple[dict[str, dict[str, Any]], Counter[str], dict[str, Any]]:
    records = load_arctander_records()
    index = build_arctander_index(records)
    labels: dict[str, dict[str, Any]] = {}
    provenance_counts: Counter[str] = Counter()
    available_unlabelled = []
    masked_with_source = []

    for cas, entry in sorted(universe.items()):
        aliases = material_aliases(entry)
        matched_alias = next((alias for alias in sorted(aliases) if alias in index), None)
        if not matched_alias:
            continue
        rec = index[matched_alias]
        if cas not in existing:
            available_unlabelled.append(cas)
        if cas in existing:
            continue

        odor_text = source_odor_text(rec)
        classes, hits_by_class, ambiguous_hits = apply_crosswalk(odor_text)
        if not classes:
            masked_with_source.append(cas)
            provenance_counts["arctander_source_available_masked_no_clean_crosswalk_hit"] += 1
            continue

        labels[cas] = {
            "cas": cas,
            "class_code": classes[0],
            "class_codes": classes,
            "provenance": "parsed-from-source",
            "source": "Arctander, Perfume and Flavor Chemicals, local monograph extract",
            "source_material_name": rec.get("name"),
            "matched_material_alias": matched_alias,
            "matched_source_descriptors_by_class": hits_by_class,
            "matched_source_descriptors": sorted({term for terms in hits_by_class.values() for term in terms}),
            "ambiguous_unmapped_descriptors_seen": ambiguous_hits,
            "source_text_excerpt": odor_text[:500],
        }
        provenance_counts["parsed_from_arctander"] += 1
        if len(classes) > 1:
            provenance_counts["parsed_from_arctander_multihot"] += 1

    coverage = {
        "arctander_monographs_total": len(records),
        "arctander_exact_aliases_indexed": len(index),
        "unlabelled_materials_with_arctander_text": len(set(available_unlabelled)),
        "unlabelled_materials_with_arctander_text_masked_no_clean_crosswalk_hit": len(set(masked_with_source)),
        "matched_unlabelled_cas_sample": sorted(set(available_unlabelled))[:50],
    }
    return labels, provenance_counts, coverage


def write_labels(labels: dict[str, dict[str, Any]]) -> None:
    with LABELS_PATH.open("w") as f:
        for cas in sorted(labels):
            f.write(json.dumps(labels[cas], sort_keys=True) + "\n")


def label_codes(label: dict[str, Any]) -> set[str]:
    codes = label.get("class_codes")
    if isinstance(codes, list):
        return {str(code) for code in codes}
    code = label.get("class_code")
    return {str(code)} if code else set()


def is_z_label(label: dict[str, Any]) -> bool:
    codes = label_codes(label)
    return bool(codes) and codes <= {"Z"}


def formula_character_sets(records: list[dict[str, Any]], labels: dict[str, dict[str, Any]]) -> dict[str, list[set[str]]]:
    by_genre: dict[str, list[set[str]]] = defaultdict(list)
    for record in records:
        if record.get("is_control"):
            continue
        genre = record.get("genre") or record.get("metadata", {}).get("generation_strategy") or "unknown"
        chars = set()
        for comp in formula_components(record):
            label = labels.get(str(comp.get("cas")))
            if not label:
                continue
            chars.update(code for code in label_codes(label) if code != "Z")
        if chars:
            by_genre[genre].append(chars)
    return by_genre


def discrimination_probe(records: list[dict[str, Any]], labels: dict[str, dict[str, Any]]) -> dict[str, Any]:
    by_genre = formula_character_sets(records, labels)
    genre_profiles: dict[str, set[str]] = {}
    threshold = 0.05
    for genre, sets in sorted(by_genre.items()):
        if len(sets) < 10:
            continue
        counts = Counter(code for s in sets for code in s)
        genre_profiles[genre] = {code for code, count in counts.items() if count / len(sets) >= threshold}

    pairwise = []
    genres = sorted(genre_profiles)
    for i, left in enumerate(genres):
        for right in genres[i + 1:]:
            a, b = genre_profiles[left], genre_profiles[right]
            pairwise.append({
                "left": left,
                "right": right,
                "jaccard": len(a & b) / max(1, len(a | b)),
                "intersection": sorted(a & b),
                "union": sorted(a | b),
            })
    mean_jaccard = sum(p["jaccard"] for p in pairwise) / len(pairwise) if pairwise else None
    return {
        "method": "per-genre formula-level set Jaccard over 26-class character labels; class active if present in >=5% of labelled formulas for genre",
        "formula_labelled_by_genre": {k: len(v) for k, v in by_genre.items()},
        "genre_profiles": {k: sorted(v) for k, v in genre_profiles.items()},
        "pairwise": pairwise,
        "mean_pairwise_jaccard": mean_jaccard,
    }


def main() -> None:
    ARTIFACTS.mkdir(exist_ok=True)
    records = load_formula_records()
    universe = material_universe(records)
    alias_index = build_alias_index(records)

    taxonomy, by_class = build_taxonomy(alias_index)
    TAXONOMY_PATH.write_text(json.dumps(taxonomy, indent=2, sort_keys=True) + "\n")

    anchor_counts = {
        code: Counter(anchor["status"] for anchor in anchors)
        for code, anchors in by_class.items()
    }
    ANCHOR_REPORT_PATH.write_text(json.dumps({
        "pimt_version": "v8",
        "taxonomy": str(TAXONOMY_PATH.relative_to(ROOT)),
        "per_class_anchor_counts": {k: dict(v) for k, v in anchor_counts.items()},
        "totals": dict(sum(anchor_counts.values(), Counter())),
    }, indent=2, sort_keys=True) + "\n")

    crosswalk_artifact = {
        "pimt_version": "v8",
        "crosswalk_version": "v8.1",
        "mapping_policy": "Exact source descriptor token/phrase match to one 26-class code; ambiguous descriptor tokens remain masked. Multi-hot labels are allowed only when multiple explicit non-ambiguous descriptors appear in the source odour text.",
        "ambiguity_policy_verified": True,
        "source_scope": ["taxonomy-reference anchors", "Arctander local monograph extract"],
        "ambiguous_unmapped_descriptors": AMBIGUOUS_UNMAPPED_DESCRIPTORS,
        "descriptor_to_class": {
            term: code
            for code, terms in DESCRIPTOR_CROSSWALK.items()
            for term in terms
        },
    }
    CROSSWALK_PATH.write_text(json.dumps(crosswalk_artifact, indent=2, sort_keys=True) + "\n")
    CROSSWALK_V81_PATH.write_text(json.dumps(crosswalk_artifact, indent=2, sort_keys=True) + "\n")
    CROSSWALK_CHANGELOG_PATH.write_text(json.dumps({
        "pimt_version": "v8",
        "crosswalk_version": "v8.1",
        "changes": CROSSWALK_CHANGELOG,
        "removed_ambiguous_descriptors": sorted(AMBIGUOUS_UNMAPPED_DESCRIPTORS),
        "policy_check": {
            "ambiguous_terms_present_in_descriptor_to_class": sorted(
                set(AMBIGUOUS_UNMAPPED_DESCRIPTORS) & set(crosswalk_artifact["descriptor_to_class"])
            ),
            "passes": not (set(AMBIGUOUS_UNMAPPED_DESCRIPTORS) & set(crosswalk_artifact["descriptor_to_class"])),
        },
    }, indent=2, sort_keys=True) + "\n")

    labels = anchor_labels(by_class)
    subst = load_substantivity_targets()
    enriched_universe = enrich_universe(universe, subst, labels)
    arctander, arctander_counts, arctander_coverage = arctander_labels(enriched_universe, labels)
    labels.update(arctander)
    write_labels(labels)

    formula_universe_cas = set(universe)
    # The work order's substantivity row refers to the full model-ready Poucher
    # target pool (~170), not only targets that happen to occur in v8 formulas.
    universe_cas = formula_universe_cas | set(subst) | set(labels)
    pre_expansion_character_cas = {
        cas
        for cas, row in anchor_labels(by_class).items()
        if cas in universe_cas and any(code != "Z" for code in label_codes(row))
    }
    pre_expansion_z_cas = {
        cas
        for cas, row in anchor_labels(by_class).items()
        if cas in universe_cas and is_z_label(row)
    }
    character_cas = {cas for cas, row in labels.items() if cas in universe_cas and any(code != "Z" for code in label_codes(row))}
    z_cas = {cas for cas, row in labels.items() if cas in universe_cas and is_z_label(row)}
    subst_cas = set(subst) & universe_cas

    class_counts: Counter[str] = Counter()
    for cas in character_cas:
        class_counts.update(code for code in label_codes(labels[cas]) if code != "Z")
    all_codes = [row["code"] for row in TAXONOMY_ROWS]
    per_class_counts = {code: int(class_counts.get(code, 0)) for code in all_codes}
    sparse_classes = {
        code: count
        for code, count in per_class_counts.items()
        if code != "Z" and count < 5
    }
    classes_ge_5 = {code: count for code, count in per_class_counts.items() if code != "Z" and count >= 5}
    classes_ge_10 = {code: count for code, count in per_class_counts.items() if code != "Z" and count >= 10}
    probe = discrimination_probe(records, labels)
    poucher_coverage = {
        "source": "Poucher Vol II local odour-classification artifacts",
        "usable_for_26_class_character_labels": False,
        "reachable_unlabelled_materials": 0,
        "reason": "Local Poucher artifacts encode top/middle/base evaporation or profile tiers, not material-level 26-class odour character descriptors.",
    }
    source_coverage = {
        "pimt_version": "v8",
        "crosswalk_version": "v8.1",
        "unlabelled_materials_before_arctander_expansion_excluding_z": len(universe_cas - pre_expansion_character_cas - pre_expansion_z_cas),
        "unlabelled_materials_after_arctander_expansion_excluding_z": len(universe_cas - character_cas - z_cas),
        "arctander": arctander_coverage,
        "poucher": poucher_coverage,
        "supplier_tds": {
            "local_corpus_found": False,
            "reachable_unlabelled_materials": 0,
            "reason": "No local supplier TDS corpus was found in this repository pass.",
        },
    }
    SOURCE_COVERAGE_PATH.write_text(json.dumps(source_coverage, indent=2, sort_keys=True) + "\n")

    census = {
        "pimt_version": "v8",
        "crosswalk_version": "v8.1",
        "dataset_scope": [str(p.relative_to(ROOT)) for p in FORMULA_DATASETS],
        "material_universe": {
            "definition": "unique CAS from empirical_dataset_v8 formula components plus model-ready Poucher substantivity targets plus resolved taxonomy anchors; solvent/carrier class reported separately",
            "unique_materials": len(universe_cas),
            "formula_visible_unique_materials": len(formula_universe_cas),
            "model_ready_poucher_substantivity_targets": len(subst),
        },
        "coverage_table": {
            "materials_with_authoritative_character_label_taxonomy_reference_plus_parsed_from_source_excluding_z": len(character_cas),
            "materials_with_substantivity_measured_poucher_model_ready": len(subst_cas),
            "materials_with_both_axes_excluding_z": len(character_cas & subst_cas),
            "materials_with_neither_axis_excluding_z": len(universe_cas - character_cas - subst_cas - z_cas),
            "class_z_carriers_character_excluded": len(z_cas),
            "per_class_character_counts": per_class_counts,
        },
        "effective_n": {
            "character_head_materials_excluding_z": len(character_cas),
            "substantivity_head_materials": len(subst_cas),
            "both_axes_materials_excluding_z": len(character_cas & subst_cas),
        },
        "label_provenance_counts_in_universe": dict(Counter(labels[cas]["provenance"] for cas in labels if cas in universe_cas)),
        "masked_unlabelled_materials_excluding_z": sorted(universe_cas - character_cas - z_cas),
        "class_z_carriers": sorted(z_cas),
        "per_class_sparsity_lt5_examples": sparse_classes,
        "per_class_counts_ge5": classes_ge_5,
        "per_class_counts_ge10": classes_ge_10,
        "source_coverage_artifact": str(SOURCE_COVERAGE_PATH.relative_to(ROOT)),
        "arctander_parse_counts": dict(arctander_counts),
        "discrimination_precheck": probe,
        "stop": "STOP_AFTER_CENSUS_NO_DATASET_ASSEMBLY_NO_TRAINING_NO_HF_UPLOAD",
        "go_no_go_frame": {
            "green": "character coverage + per-class counts sufficient -> proceed to Phase 2.5 gate then Phase 3 training",
            "amber": "character axis too sparse, substantivity axis strong -> publish/ship substantivity forward result first and grow character coverage separately",
            "red": "discrimination probe high -> do not train",
        },
    }
    CENSUS_PATH.write_text(json.dumps(census, indent=2, sort_keys=True) + "\n")

    lines = [
        "# PIMT v8 odour taxonomy coverage census",
        "",
        "| Bucket | Count |",
        "|:--|--:|",
        f"| Materials with authoritative character label (excluding Z) | {len(character_cas)} |",
        f"| Materials with substantivity (measured Poucher, model-ready, in universe) | {len(subst_cas)} |",
        f"| Materials with both axes (excluding Z) | {len(character_cas & subst_cas)} |",
        f"| Materials with neither axis (excluding Z) | {len(universe_cas - character_cas - subst_cas - z_cas)} |",
        f"| Class Z carriers, character-excluded | {len(z_cas)} |",
        "",
        "## Per-class character counts",
        "",
        "| Code | Count |",
        "|:--:|--:|",
    ]
    lines.extend(f"| {code} | {per_class_counts[code]} |" for code in all_codes)
    lines.extend([
        "",
        f"Non-Z classes >=5 labels: {len(classes_ge_5)}/25",
        f"Non-Z classes >=10 labels: {len(classes_ge_10)}/25",
        f"Both-axis material overlap: {len(character_cas & subst_cas)}",
        "",
        "## Discrimination pre-check",
        "",
        f"Mean pairwise Jaccard: {probe['mean_pairwise_jaccard'] if probe['mean_pairwise_jaccard'] is not None else 'n/a'}",
        "",
        "STOP: no empirical_dataset_v10.jsonl assembled; no training started; no HF upload.",
    ])
    SUMMARY_PATH.write_text("\n".join(lines) + "\n")

    print(json.dumps({
        "taxonomy": str(TAXONOMY_PATH.relative_to(ROOT)),
        "labels": str(LABELS_PATH.relative_to(ROOT)),
        "census": str(CENSUS_PATH.relative_to(ROOT)),
        "character_n": len(character_cas),
        "substantivity_n": len(subst_cas),
        "both_n": len(character_cas & subst_cas),
        "classes_ge5": len(classes_ge_5),
        "classes_ge10": len(classes_ge_10),
        "mean_discrimination_jaccard": probe["mean_pairwise_jaccard"],
    }, indent=2, sort_keys=True))


if __name__ == "__main__":
    main()