"""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" _SEARCH_URL = "https://world.openfoodfacts.org/api/v2/search" # HTTP statuses worth retrying: rate limiting and transient server errors. _RETRY_STATUS = frozenset({429, 500, 502, 503, 504}) # Fields requested from the search API so a returned product can be transformed # without an extra per-barcode round trip. _SEARCH_FIELDS = ( "code,product_name,product_name_en,product_name_zh,brands,quantity," "categories,categories_tags,countries,ingredients_text,allergens_tags," "additives_tags,nutriments,nutriscore_grade,serving_size," "image_front_url,image_url,last_modified_t" ) 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, max_retries: int = 4, backoff_base: float = 2.0, ) -> None: self._client = client or httpx.Client(headers={"User-Agent": USER_AGENT}, timeout=30.0) self._min_interval = min_interval self._max_retries = max_retries self._backoff_base = backoff_base 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 _get(self, url: str, params: dict | None = None) -> httpx.Response: """GET with throttling and retry/backoff on transient errors. Retries on connection/timeout errors and on retryable HTTP statuses (429 and 5xx, which OFF returns intermittently when overloaded), using exponential backoff that honours a ``Retry-After`` header when present. """ last_exc: Exception | None = None for attempt in range(self._max_retries + 1): self._throttle() try: resp = self._client.get(url, params=params) except httpx.TransportError as exc: last_exc = exc else: if resp.status_code < 400 or resp.status_code not in _RETRY_STATUS: resp.raise_for_status() return resp last_exc = httpx.HTTPStatusError( f"retryable status {resp.status_code}", request=resp.request, response=resp ) if attempt < self._max_retries: retry_after = self._retry_after(last_exc) time.sleep(retry_after if retry_after is not None else self._backoff_base**attempt) assert last_exc is not None raise last_exc @staticmethod def _retry_after(exc: Exception | None) -> float | None: resp = getattr(exc, "response", None) if resp is None: return None value = resp.headers.get("Retry-After") if not value: return None try: return float(value) except ValueError: return None def fetch_barcode(self, barcode: str) -> dict | None: """Fetch a single product by barcode; return the raw `product` dict.""" resp = self._get(_API_URL.format(barcode=barcode)) 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 fetch_modified_since( self, since_t: int, *, page_size: int = 100, max_pages: int = 10, ) -> Iterator[dict]: """Yield products modified after ``since_t`` (unix ``last_modified_t``). Uses the OFF search API sorted by ``last_modified_t`` (most recent first) and paginates until it reaches products at or before the watermark, an empty/short page, or ``max_pages``. This is the incremental ingestion path: callers persist the highest ``last_modified_t`` they processed as the next watermark. """ for page in range(1, max_pages + 1): resp = self._get( _SEARCH_URL, params={ "fields": _SEARCH_FIELDS, "sort_by": "last_modified_t", "page": page, "page_size": page_size, }, ) products = resp.json().get("products") or [] if not products: return reached_old = False for prod in products: if int(prod.get("last_modified_t") or 0) <= since_t: reached_old = True break yield prod if reached_old or len(products) < page_size: return def fetch_by_country( self, country: str, *, page_size: int = 100, max_pages: int = 10, sort_by: str = "unique_scans_n", ) -> Iterator[dict]: """Yield products sold in ``country`` (an OFF ``countries_tags_en`` slug). Used to seed a market-specific catalogue (e.g. ``china``). Results are sorted by ``sort_by`` (default ``unique_scans_n`` so the most-scanned, best-known products come first) and de-duplicated across pages, since OFF's popularity ordering is not stable between page requests. """ seen: set[str] = set() for page in range(1, max_pages + 1): resp = self._get( _SEARCH_URL, params={ "fields": _SEARCH_FIELDS, "countries_tags_en": country, "sort_by": sort_by, "page": page, "page_size": page_size, }, ) products = resp.json().get("products") or [] if not products: return new_on_page = 0 for prod in products: code = str(prod.get("code") or "") if code and code in seen: continue if code: seen.add(code) new_on_page += 1 yield prod if len(products) < page_size or new_on_page == 0: return def is_cn_gs1(code: str | None) -> bool: """Return True for a GS1 China company prefix (barcodes starting 690-699). These identify products registered with GS1 China, i.e. genuinely domestic items, as opposed to imported goods merely tagged as sold in China. """ if not code: return False code = code.strip() return len(code) >= 3 and code[:2] == "69" and code[2].isdigit() 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)