novaalphastrikeomegaz663
2255243081
Add a market-focused seeding path so the catalogue can be built from domestic products rather than the English-heavy global default: - adapter.fetch_by_country(country): OFF search filtered by countries_tags_en, sorted by unique_scans_n (most-scanned first), de-duplicated across pages since OFF popularity ordering is unstable. - is_cn_gs1(code): True for GS1-China company prefixes (690-699), i.e. genuinely domestic items vs. imports merely sold in China. - seed_off --country <slug> [--domestic-only] [--page-size/--max-pages]: e.g. 'seed_off --country china --domestic-only' loads only 69x barcodes. Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
233 lines
8.2 KiB
Python
233 lines
8.2 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"
|
|
_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)
|