scripts/extract_poucher_substantivity.py

#!/usr/bin/env python3
"""Extract Poucher's measured duration-of-evaporation coefficients.

The source table is in Poucher Vol II, Chapter 4, "Duration of evaporation
table" (parsed source pages 70-74). The numeric coefficient is the target:
it is measured by olfaction on smelling strips, not computed from molecular
features.
"""
from __future__ import annotations

import json
import re
from collections import defaultdict
from statistics import median
from pathlib import Path

from extract_poucher_tiers import match_name_to_cas

DATA = Path("data")
PAGES = DATA / "literature_flat" / "literature_pages.jsonl"
OUT = DATA / "poucher_substantivity.jsonl"
PUBCHEM_MAP = DATA / "poucher_pubchem_name_to_cas.json"


# Page 70 and page 74 are two-column layouts where the existing OCR text flattens
# columns out of visual order. These groups were verified against the page images.
PAGE_70_GROUPS: dict[int, list[str]] = {
    1: [
        "Acetophenone", "Almonds", "Amyl acetate", "Benzaldehyde",
        "Benzyl acetate", "Ethyl acetate", "Ethyl acetoacetate",
        "Iso-butyl acetate", "Methyl benzoate", "Niaouli", "Octyl acetate",
        "Phenylethyl acetate", "Phenylethyl formate",
        "Phenylethyl propionate", "Phenylethyl salicylate",
    ],
    2: [
        "Benzyl formate", "Bois de rose", "Ethyl benzoate",
        "Limes distilled", "Linalol", "Mandarin", "Methyl salicylate",
    ],
    3: [
        "Benzyl cinnamate", "Coriander", "Para cresyl methyl ether",
        "Para cresyl acetate", "Para cresyl iso-butyrate",
        "Cuminic aldehyde", "Cyclohexanyl butyrate", "Decyl formate",
        "Dimethyl benzyl carbinol", "Dimethyl benzyl acetate",
        "Ethyl decine carbonate", "Ethyl salicylate",
    ],
}

PAGE_74_GROUPS: dict[int, list[str]] = {
    54: ["Cedryl acetate"],
    55: ["Nerolidol"],
    60: ["Benzyl phenylacetate", "Citral", "Rhodinyl formate"],
    62: ["Amyl phenylacetate"],
    65: ["Cinnamic alcohol nat."],
    70: ["Linalyl salicylate", "Jasmin decolore"],
    73: ["Cassie absolute, Farnesiana"],
    77: ["Methyl naphthyl ketone"],
    79: ["Civette absolute"],
    80: ["Hydroxy citronellal"],
    85: ["Phenylacetaldehyde dimethyl acetal"],
    87: ["Octyl aldehyde"],
    88: ["Ethyl methyl phenyl glycidate"],
    89: ["Cyclamen aldehyde"],
    90: [
        "Galbanum resin", "Opoponax resin", "Orris oleo resin",
        "Rhodinyl acetate", "Santal W.A.", "Tarragon",
        "Jasmin chassis incolore",
    ],
    91: ["Phenylethyl phenylacetate", "Undecalactone"],
    94: ["Angelica root", "Birch bud"],
    96: ["Arnica flowers"],
    100: [
        "Acet eugenol", "Ambergris extract, 3 per cent",
        "Amyl cinnamic aldehyde", "Amyloxy iso-eugenol", "Benzoin",
        "Benzophenone", "Birch tar", "Castoreum absolute",
        "Cinnamic alcohol synthetic", "Costus", "Coumarin", "Cypress",
        "Decyl aldehyde", "Ethyl vanillin", "Gamma nonyl lactone",
        "Guaiyl esters", "Immortelle absolute", "Iso-eugenol",
        "Iso-eugenol phenylacetate", "Labdanum",
        "Linalyl phenylacetate", "Methyl nonyl acetaldehyde",
        "Musks artificial", "Oakmoss", "Olibanum oil and resin",
        "Patchouli", "Pepper", "Peru balsam", "Phenylacetic acid",
        "Phenylacetic aldehyde", "Pimento", "Rhodinyl phenylacetate",
        "Santalwood E.1.", "Storax resin", "Santalyl phenylacetate",
        "Tolu balsam", "Tonka resinoid",
        "Trichlor phenyl methyl carbinyl acetate", "Undecylic aldehyde",
        "Vanillin", "Vetivert",
    ],
}

SKIP_LINES = {
    "PERFUMES, COSMETICS AND SOAPS",
    "ODOUR CLASSIFICATION AND FIXATION",
    "Duration of evaporation table",
    "Top notes",
    "Top notes-cont.",
    "Top notes-cant",
    "Middle notes",
    "Middle notes-cont.",
    "Bases",
}


def load_poucher_pages() -> dict[int, str]:
    pages: dict[int, str] = {}
    with open(PAGES) as f:
        for line in f:
            rec = json.loads(json.loads(line)["record"])
            if "Poucher" in rec.get("source", ""):
                pages[int(rec["page"])] = rec.get("text", "")
    return pages


def tier_for_coefficient(coefficient: int) -> str:
    if coefficient <= 14:
        return "top"
    if coefficient <= 60:
        return "mid"
    return "base"


def conflict_resolution_note(rows: list[dict]) -> str:
    names = {r["name"].lower() for r in rows}
    if len(names) == 1:
        return "same_material_repeat"
    return "same_cas_distinct_source_names_median_coefficient"


def clean_name(name: str) -> str:
    name = re.sub(r"\s+", " ", name).strip()
    name = name.strip(".,;:")
    return name


def load_pubchem_map() -> dict[str, dict]:
    if not PUBCHEM_MAP.exists():
        return {}
    with open(PUBCHEM_MAP) as f:
        return json.load(f).get("mappings", {})


def match_with_provenance(name: str, pubchem_map: dict[str, dict]) -> tuple[str | None, str, dict]:
    cas = match_name_to_cas(name)
    if cas:
        return cas, "poucher_name_map", {}
    mapped = pubchem_map.get(name)
    if mapped:
        return mapped["cas"], "pubchem_name_synonym", {
            "pubchem_cid": mapped.get("pubchem_cid"),
            "pubchem_name": mapped.get("pubchem_name"),
        }
    return None, "unmatched", {}


def parse_numbered_lines(text: str, page: int, initial_coefficient: int) -> tuple[list[dict], int]:
    entries: list[dict] = []
    current = initial_coefficient

    for raw in text.splitlines():
        line = clean_name(raw)
        if not line or line in SKIP_LINES or line.lower() in {s.lower() for s in SKIP_LINES}:
            continue
        if re.fullmatch(r"\d{1,3}", line):
            continue
        if re.fullmatch(r"\d{1,3}\s+PERFUMES, COSMETICS AND SOAPS", line):
            continue

        # OCR occasionally reads 8. as B.
        number_match = re.match(r"^(B|\d{1,3})\.\s*(.*)$", line)
        if number_match:
            raw_num = number_match.group(1)
            current = 8 if raw_num == "B" else int(raw_num)
            name = clean_name(number_match.group(2))
            if name:
                entries.append(entry(current, name, page))
            continue

        if current is not None and looks_like_material(line):
            entries.append(entry(current, line, page))

    return entries, current


def looks_like_material(line: str) -> bool:
    if len(line) < 2 or len(line) > 80:
        return False
    if line[0].isdigit():
        return False
    low = line.lower()
    if any(fragment in low for fragment in ["chapter", "page ", "classification", "table"]):
        return False
    return True


def entry(coefficient: int, name: str, page: int) -> dict:
    return {
        "name": clean_name(name),
        "poucher_coefficient": coefficient,
        "tier_band": tier_for_coefficient(coefficient),
        "page": page,
    }


def grouped_entries(groups: dict[int, list[str]], page: int) -> list[dict]:
    return [entry(coeff, name, page) for coeff, names in groups.items() for name in names]


def extract_entries() -> list[dict]:
    pages = load_poucher_pages()
    entries: list[dict] = []
    entries.extend(grouped_entries(PAGE_70_GROUPS, 70))

    current = 3
    for page in [71, 72, 73]:
        page_entries, current = parse_numbered_lines(pages[page], page, current)
        entries.extend(page_entries)

    entries.extend(grouped_entries(PAGE_74_GROUPS, 74))

    seen = set()
    unique = []
    for e in entries:
        key = (e["poucher_coefficient"], e["name"].lower())
        if key not in seen:
            seen.add(key)
            unique.append(e)
    return unique


def main() -> None:
    entries = extract_entries()
    pubchem_map = load_pubchem_map()

    by_cas: dict[str, dict] = {}
    unmatched: list[dict] = []
    conflicts: dict[str, list[dict]] = defaultdict(list)

    for e in entries:
        cas, match_source, match_extra = match_with_provenance(e["name"], pubchem_map)
        if not cas:
            unmatched.append(e)
            continue

        row = {
            "cas": cas,
            "name": e["name"],
            "poucher_coefficient": e["poucher_coefficient"],
            "tier_band": e["tier_band"],
            "source": "Poucher Vol II duration of evaporation table",
            "source_page": e["page"],
            "target_type": "measured_olfactive_duration_coefficient",
            "cas_match_source": match_source,
            **match_extra,
        }
        conflicts[cas].append(row)

    for cas, rows in conflicts.items():
        coeffs = sorted({r["poucher_coefficient"] for r in rows})
        source_rows = sorted(
            [
                {
                    "name": r["name"],
                    "poucher_coefficient": r["poucher_coefficient"],
                    "tier_band": r["tier_band"],
                    "source_page": r["source_page"],
                    "cas_match_source": r["cas_match_source"],
                }
                for r in rows
            ],
            key=lambda r: (r["poucher_coefficient"], r["name"]),
        )
        median_coefficient = int(round(median(r["poucher_coefficient"] for r in rows)))
        chosen = sorted(
            rows,
            key=lambda r: (
                abs(r["poucher_coefficient"] - median_coefficient),
                r["poucher_coefficient"],
                r["name"],
            ),
        )[0]
        chosen["poucher_coefficient"] = median_coefficient
        chosen["tier_band"] = tier_for_coefficient(median_coefficient)
        chosen["all_poucher_names"] = sorted({r["name"] for r in rows})
        chosen["all_poucher_coefficients"] = coeffs
        chosen["coefficient_conflict"] = len(coeffs) > 1
        chosen["conflict_resolution"] = (
            conflict_resolution_note(rows) if len(coeffs) > 1 else "none"
        )
        chosen["source_rows"] = source_rows
        by_cas[cas] = chosen

    OUT.write_text(
        "\n".join(json.dumps(row, sort_keys=True) for _, row in sorted(by_cas.items())) + "\n"
    )

    print(f"Extracted table entries: {len(entries)}")
    print(f"Matched CAS rows: {len(by_cas)}")
    print(f"Unmatched names: {len(unmatched)}")
    print(f"CAS with coefficient conflicts: {sum(1 for r in by_cas.values() if r['coefficient_conflict'])}")
    print("Tier-band distribution:")
    dist = defaultdict(int)
    for row in by_cas.values():
        dist[row["tier_band"]] += 1
    for tier in ["top", "mid", "base"]:
        print(f"  {tier}: {dist[tier]}")
    if unmatched:
        print("Unmatched sample:")
        for row in unmatched[:25]:
            print(f"  p{row['page']} c{row['poucher_coefficient']}: {row['name']}")
    print(f"Saved to {OUT}")


if __name__ == "__main__":
    main()