#!/usr/bin/env python3 """Build an exact-CAS Good Scents odor subset from the Pyrfume mirror.""" from __future__ import annotations from collections import defaultdict import csv import io import json from pathlib import Path import re from urllib.request import Request, urlopen ROOT = Path(__file__).resolve().parents[1] DATA = ROOT / "data" PROFILES = DATA / "material_profiles_v11.jsonl" OUTPUT = DATA / "goodscents_odor_descriptions_pyrfume_v11.jsonl" BASE = "https://raw.githubusercontent.com/pyrfume/pyrfume-data/main/goodscents" OPL_URL = f"{BASE}/data_rw_opl.csv" ODOR_URL = f"{BASE}/data_rw_odor.csv" MOLECULES_URL = f"{BASE}/molecules.csv" CAS_RE = re.compile(r"^\d{2,7}-\d{2}-\d$") def valid_cas(value: str) -> bool: value = value.strip() if not CAS_RE.fullmatch(value): return False digits = value.replace("-", "") return sum(int(digit) * weight for weight, digit in enumerate(reversed(digits[:-1]), 1)) % 10 == int(digits[-1]) def download_csv(url: str) -> list[dict[str, str]]: request = Request(url, headers={"User-Agent": "PinoDataIngest/1.0"}) with urlopen(request, timeout=60) as response: # noqa: S310 - fixed HTTPS source text = response.read().decode("utf-8-sig") return list(csv.DictReader(io.StringIO(text))) def inchikey(smiles: str) -> str | None: from rdkit import Chem from rdkit.Chem import inchi if not smiles or smiles.startswith("NATURAL:"): return None molecule = Chem.MolFromSmiles(smiles) return inchi.MolToInchiKey(molecule) if molecule is not None else None def build_rows( opl_rows: list[dict[str, str]], odor_rows: list[dict[str, str]], molecule_rows: list[dict[str, str]], profiles: list[dict[str, object]], ) -> tuple[list[dict[str, object]], dict[str, int]]: profile_identities = {str(row.get("cas") or "") for row in profiles} profile_by_inchikey: dict[str, set[str]] = defaultdict(set) for profile in profiles: smiles = str(profile.get("smiles") or "").removeprefix("SMILES:") key = inchikey(smiles) if smiles else None if key: profile_by_inchikey[key].add(str(profile.get("cas") or profile.get("profile_id"))) unique_profile_by_inchikey = { key: next(iter(values)) for key, values in profile_by_inchikey.items() if len(values) == 1 } cid_to_inchikey = { str(row.get("CID") or "").strip(): key for row in molecule_rows if (key := inchikey(str(row.get("IsomericSMILES") or ""))) } candidates: dict[str, set[tuple[str, str, str]]] = defaultdict(set) for row in opl_rows: cas = str(row.get("CAS Number") or "").strip() tgsc_id = str(row.get("TGSC ID") or "").strip() if not tgsc_id or not valid_cas(cas): continue if cas in profile_identities: candidates[tgsc_id].add((cas, cas, "tgsc-id-to-unique-exact-cas")) key = cid_to_inchikey.get(str(row.get("CID") or "").strip()) identity = unique_profile_by_inchikey.get(key or "") if identity: candidates[tgsc_id].add( (identity, cas, "tgsc-cid-to-exact-structure-inchikey") ) structure_source_cas: dict[str, set[str]] = defaultdict(set) for values in candidates.values(): for identity, source_cas, method in values: if method == "tgsc-cid-to-exact-structure-inchikey": structure_source_cas[identity].add(source_cas) ambiguous_structure_identities = { identity for identity, values in structure_source_cas.items() if len(values) > 1 } if ambiguous_structure_identities: candidates = { tgsc_id: { value for value in values if not ( value[2] == "tgsc-cid-to-exact-structure-inchikey" and value[0] in ambiguous_structure_identities ) } for tgsc_id, values in candidates.items() } tgsc_to_identity: dict[str, tuple[str, str, str]] = {} ambiguous_tgsc_ids = 0 for tgsc_id, values in candidates.items(): exact_cas = {value for value in values if value[2] == "tgsc-id-to-unique-exact-cas"} usable = exact_cas or values if not usable: continue if len(usable) == 1: tgsc_to_identity[tgsc_id] = next(iter(usable)) else: ambiguous_tgsc_ids += 1 output: list[dict[str, object]] = [] seen: set[tuple[str, str]] = set() for row in odor_rows: tgsc_id = str(row.get("TGSC ID") or "").strip() resolved = tgsc_to_identity.get(tgsc_id) identity, source_cas, join_method = resolved or (None, None, None) description = " ".join(str(row.get("Description") or "").split()) signature = (identity or "", description.casefold()) if not identity or not description or signature in seen: continue seen.add(signature) item: dict[str, object] = { "cas": identity, "source_cas": source_cas, "text": description, "tgsc_id": tgsc_id, "join_method": join_method, "provenance": "goodscents-via-pyrfume-mirror", "source_dataset": ODOR_URL, } for source_key, output_key in ( ("Source", "original_source"), ("Source Year", "source_year"), ("Sample Supplier", "sample_supplier"), ): value = str(row.get(source_key) or "").strip() if value: item[output_key] = value output.append(item) output.sort(key=lambda row: (str(row["cas"]), str(row["text"]).casefold())) return output, { "profiles_with_rows": len({str(row["cas"]) for row in output}), "description_rows": len(output), "ambiguous_tgsc_ids_rejected": ambiguous_tgsc_ids, "exact_structure_profiles": len({ str(row["cas"]) for row in output if row["join_method"] == "tgsc-cid-to-exact-structure-inchikey" }), "ambiguous_structure_profiles_rejected": len(ambiguous_structure_identities), "identity_policy": ( "exact CAS first; otherwise exact full InChIKey with one profile and one source CAS; " "ambiguous structures rejected" ), } def main() -> int: profiles = [json.loads(line) for line in PROFILES.read_text().splitlines() if line.strip()] rows, report = build_rows( download_csv(OPL_URL), download_csv(ODOR_URL), download_csv(MOLECULES_URL), profiles ) OUTPUT.write_text("".join(json.dumps(row, sort_keys=True) + "\n" for row in rows)) print(json.dumps({"output": str(OUTPUT.relative_to(ROOT)), **report}, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())