scripts/build_tgsc_odor_cache.py

#!/usr/bin/env python3
"""One-off builder for TGSC ingredient odor cache.

Fetches organoleptic descriptors for every unique TGSC ingredient link found
in a TGSC demo formula dataset and persists them to data/tgsc_odor_cache.json.
Subsequent ingest_tgsc.py runs use this cache and complete instantly.
"""

from __future__ import annotations

import importlib.util
import json
import sys
import time
from pathlib import Path

import requests
from bs4 import BeautifulSoup

src = Path(__file__).parent.parent / "src"
if str(src) not in sys.path:
    sys.path.insert(0, str(src))

_INGEST_PATH = Path(__file__).parent / "ingest_tgsc.py"
_ingest_spec = importlib.util.spec_from_file_location("ingest_tgsc", _INGEST_PATH)
assert _ingest_spec and _ingest_spec.loader
_ingest_module = importlib.util.module_from_spec(_ingest_spec)
sys.modules["ingest_tgsc"] = _ingest_module
_ingest_spec.loader.exec_module(_ingest_module)

_absolute_tgsc_url = _ingest_module._absolute_tgsc_url
_load_tgsc_odor_cache = _ingest_module._load_tgsc_odor_cache
_save_tgsc_odor_cache = _ingest_module._save_tgsc_odor_cache
_is_known_header = _ingest_module._is_known_header


def _fetch(link: str) -> dict:
    result = {
        "url": link,
        "odor_type": [],
        "odor_descriptions": [],
        "flavor_type": [],
        "taste_descriptions": [],
        "vapor_pressure_pa": None,
        "error": None,
    }
    try:
        resp = requests.get(link, headers={"User-Agent": "PinoDataIngest/1.0"}, timeout=20)
        resp.raise_for_status()
    except Exception as exc:
        result["error"] = str(exc)
        return result

    soup = BeautifulSoup(resp.text, "html.parser")
    page_text = soup.get_text("\n")
    lines = [l.strip() for l in page_text.splitlines() if l.strip()]

    i = 0
    while i < len(lines):
        line = lines[i]
        low = line.lower()

        if low.startswith("odor type:"):
            j = i + 1
            while j < len(lines) and not _is_known_header(lines[j]):
                result["odor_type"].append(lines[j])
                j += 1
            i = j
            continue

        if low.startswith("odor description:"):
            j = i + 1
            while j < len(lines) and not _is_known_header(lines[j]):
                result["odor_descriptions"].append(lines[j])
                j += 1
            i = j
            continue

        if low.startswith("flavor type:"):
            j = i + 1
            while j < len(lines) and not _is_known_header(lines[j]):
                result["flavor_type"].append(lines[j])
                j += 1
            i = j
            continue

        if low.startswith("taste description:"):
            j = i + 1
            while j < len(lines) and not _is_known_header(lines[j]):
                result["taste_descriptions"].append(lines[j])
                j += 1
            i = j
            continue

        i += 1

    # Vapor pressure: look for "Vapor Pressure: X Pa (Y mm Hg)"
    for line in lines:
        if "vapor pressure" in line.lower():
            m = __import__("re").search(r"([0-9.]+)\s*Pa", line)
            if m:
                result["vapor_pressure_pa"] = float(m.group(1))
            break

    return result


def main() -> int:
    data_path = Path(__file__).parent.parent / "data" / "tgsc_demo_formulas_v2.jsonl"
    if not data_path.exists():
        print(f"Dataset not found: {data_path}")
        return 1

    records = [json.loads(line) for line in data_path.open()]
    links: set[str] = set()
    for rec in records:
        for item in rec.get("formula", []):
            link = item.get("link")
            abs_url = _absolute_tgsc_url(link)
            if abs_url:
                links.add(abs_url)

    print(f"Unique TGSC ingredient links to fetch: {len(links)}")

    cache = _load_tgsc_odor_cache()
    print(f"Already cached: {len(cache)}")

    todo = [l for l in links if l not in cache]
    print(f"To fetch: {len(todo)}")

    for idx, link in enumerate(todo, 1):
        cache[link] = _fetch(link)
        if idx % 50 == 0:
            _save_tgsc_odor_cache(cache)
            print(f"Fetched {idx}/{len(todo)}; cache size: {len(cache)}")
        time.sleep(0.2)

    _save_tgsc_odor_cache(cache)
    print(f"Done. Cache size: {len(cache)}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())