File size: 4,351 Bytes
eaf6535 | 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 | #!/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())
|