lixu
766a573989
- adapters/openfoodfacts.py: OFF API 适配器(限速+User-Agent) + JSONL/.gz dump 读取 - etl/transform.py: 纯函数转换(GTIN 校验/净含量解析+归一/营养 per_100g 双能量/过敏原添加剂清洗/关键词分类映射) - etl/load.py: psycopg upsert(product/food_detail/product_image) + product_source 字段级溯源 - jobs/seed_off.py: CLI(--barcodes API / --dump 文件 / --limit) - 测试: 13 个离线 transform 单测(fixture) + 可跳过的 DB 集成测试 - 依赖: 增加 psycopg[binary] - 实跑: 从 OFF API 拉 5 个真实条码入本地 postgres 验证通过 - docs/etl-openfoodfacts.md: 流程/运行/字段映射 Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
87 lines
2.7 KiB
Python
87 lines
2.7 KiB
Python
"""Open Food Facts (OFF) source adapter.
|
|
|
|
Fetches raw product records either from the OFF read API (one product per
|
|
barcode) or from a downloaded JSONL dump file. OFF data is licensed under the
|
|
Open Database License (ODbL); product images are CC-BY-SA. We record OFF as the
|
|
source for every field we ingest.
|
|
|
|
The adapter is read-only and rate-limited to stay well within OFF's API limits
|
|
(<= ~15 req/min/IP for product reads) and to be a good citizen.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import time
|
|
from collections.abc import Iterator
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
|
|
SOURCE_NAME = "openfoodfacts"
|
|
OFF_LICENSE = "ODbL"
|
|
USER_AGENT = "OpenGoods/0.1 (+https://github.com/baicai2026-baicai/goods) public-good product API"
|
|
|
|
# Conservative client-side spacing between API calls (seconds).
|
|
_DEFAULT_MIN_INTERVAL = 4.0
|
|
_API_URL = "https://world.openfoodfacts.org/api/v2/product/{barcode}.json"
|
|
|
|
|
|
class OpenFoodFactsAdapter:
|
|
"""Read product records from the OFF API."""
|
|
|
|
source_name = SOURCE_NAME
|
|
|
|
def __init__(
|
|
self,
|
|
client: httpx.Client | None = None,
|
|
min_interval: float = _DEFAULT_MIN_INTERVAL,
|
|
) -> None:
|
|
self._client = client or httpx.Client(headers={"User-Agent": USER_AGENT}, timeout=30.0)
|
|
self._min_interval = min_interval
|
|
self._last_call = 0.0
|
|
|
|
def _throttle(self) -> None:
|
|
elapsed = time.monotonic() - self._last_call
|
|
wait = self._min_interval - elapsed
|
|
if wait > 0:
|
|
time.sleep(wait)
|
|
self._last_call = time.monotonic()
|
|
|
|
def fetch_barcode(self, barcode: str) -> dict | None:
|
|
"""Fetch a single product by barcode; return the raw `product` dict."""
|
|
self._throttle()
|
|
resp = self._client.get(_API_URL.format(barcode=barcode))
|
|
resp.raise_for_status()
|
|
payload = resp.json()
|
|
if payload.get("status") != 1:
|
|
return None
|
|
return payload["product"]
|
|
|
|
def fetch(self, barcodes: list[str]) -> Iterator[dict]:
|
|
"""Yield raw product records for the given barcodes."""
|
|
for code in barcodes:
|
|
record = self.fetch_barcode(code)
|
|
if record is not None:
|
|
yield record
|
|
|
|
|
|
def read_dump(path: str | Path) -> Iterator[dict]:
|
|
"""Yield raw product records from an OFF JSONL dump file.
|
|
|
|
Each line is one product JSON object (the format of OFF's .jsonl export).
|
|
Supports plain or .gz files.
|
|
"""
|
|
p = Path(path)
|
|
if p.suffix == ".gz":
|
|
import gzip
|
|
|
|
opener = lambda: gzip.open(p, "rt", encoding="utf-8") # noqa: E731
|
|
else:
|
|
opener = lambda: open(p, encoding="utf-8") # noqa: E731
|
|
with opener() as fh:
|
|
for line in fh:
|
|
line = line.strip()
|
|
if line:
|
|
yield json.loads(line)
|