234 lines
12 KiB
Python
234 lines
12 KiB
Python
"""Free product-identity providers behind the Hub. Agents never talk to these; the Hub does, through the quota manager.
|
|
|
|
Documentation snapshots consulted 2026-09-25 (see docs/product_enrichment.md):
|
|
- UPCitemdb Explorer/trial: https://api.upcitemdb.com/prod/trial/lookup?upc=... - no key, 100 combined requests/day, burst 15 lookup / 5 search per 30 s,
|
|
headers X-RateLimit-Limit / -Remaining / -Reset. Product IDENTITY only (title/brand/category); no fiscal data.
|
|
- Open Food Facts: https://world.openfoodfacts.org/api/v2/product/{barcode} - 15 read req/min/IP, 10 search req/min/IP, identifiable User-Agent, ODbL data. Foods only; no fiscal data.
|
|
- Cosmos / Bluesoft: https://api.cosmos.bluesoft.com.br/gtins/{gtin} - X-Cosmos-Token (free token, 25 queries/day), returns NCM/CEST/brand/description. Optional: without a token it is NOT_CONFIGURED.
|
|
A provider result is EVIDENCE. Only the fiscal engine + official datasets decide what is written.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
|
|
import httpx
|
|
|
|
USER_AGENT = "GTIFiscalHub/1.0 (+https://gtifiscal.gtigrupo.com.br; product identity lookup)"
|
|
|
|
|
|
class ProviderError(Exception):
|
|
"""kind: RATE_LIMITED | AUTH_ERROR | TIMEOUT | SERVER_ERROR | NETWORK | BAD_RESPONSE"""
|
|
|
|
def __init__(self, kind: str, detail: str = "", headers: dict | None = None, retry_after: int | None = None):
|
|
super().__init__(f"{kind}: {detail}")
|
|
self.kind, self.detail, self.headers, self.retry_after = kind, detail, headers or {}, retry_after
|
|
|
|
|
|
@dataclass
|
|
class ProviderProduct:
|
|
provider: str
|
|
gtin: str
|
|
found: bool
|
|
title: str | None = None
|
|
brand: str | None = None
|
|
category: str | None = None
|
|
quantity: str | None = None
|
|
ncm: str | None = None
|
|
cest: str | None = None
|
|
raw_hash: str = ""
|
|
latency_ms: int = 0
|
|
headers: dict = field(default_factory=dict)
|
|
|
|
def as_data(self) -> dict:
|
|
return {k: v for k, v in {"title": self.title, "brand": self.brand, "category": self.category, "quantity": self.quantity, "ncm": self.ncm, "cest": self.cest}.items() if v}
|
|
|
|
|
|
def _digits(value) -> str | None:
|
|
text = "".join(ch for ch in str(value or "") if ch.isdigit())
|
|
return text or None
|
|
|
|
|
|
class Provider:
|
|
name = "base"
|
|
paid = False
|
|
supports_search = False
|
|
daily_limit: int | None = None
|
|
rate_per_minute: int | None = None
|
|
burst: tuple[int, int] | None = None # (requests, seconds)
|
|
provides_ncm = False
|
|
|
|
def __init__(self, client: httpx.Client | None = None, token: str | None = None):
|
|
self.client = client or httpx.Client(timeout=4.0, headers={"User-Agent": USER_AGENT})
|
|
self._token = token
|
|
|
|
@property
|
|
def configured(self) -> bool:
|
|
return True
|
|
|
|
def _get(self, url: str, **kwargs) -> httpx.Response:
|
|
started = time.monotonic()
|
|
try:
|
|
response = self.client.get(url, **kwargs)
|
|
except httpx.TimeoutException as error:
|
|
raise ProviderError("TIMEOUT", type(error).__name__) from error
|
|
except httpx.HTTPError as error:
|
|
raise ProviderError("NETWORK", type(error).__name__) from error
|
|
response.latency_ms = int((time.monotonic() - started) * 1000) # type: ignore[attr-defined]
|
|
headers = {k.lower(): v for k, v in response.headers.items()}
|
|
if response.status_code == 429:
|
|
retry = headers.get("retry-after")
|
|
raise ProviderError("RATE_LIMITED", "HTTP 429", headers, int(retry) if retry and retry.isdigit() else None)
|
|
if response.status_code in (401, 403):
|
|
raise ProviderError("AUTH_ERROR", f"HTTP {response.status_code}", headers)
|
|
if response.status_code >= 500:
|
|
raise ProviderError("SERVER_ERROR", f"HTTP {response.status_code}", headers)
|
|
return response
|
|
|
|
def lookup_gtin(self, gtin: str) -> ProviderProduct: # pragma: no cover - interface
|
|
raise NotImplementedError
|
|
|
|
def search_description(self, text: str) -> list[ProviderProduct]:
|
|
return []
|
|
|
|
def health(self) -> dict:
|
|
return {"configured": self.configured}
|
|
|
|
|
|
class UpcItemDb(Provider):
|
|
name = "upcitemdb"
|
|
supports_search = True
|
|
daily_limit = 100
|
|
burst = (15, 30)
|
|
|
|
def lookup_gtin(self, gtin: str) -> ProviderProduct:
|
|
response = self._get("https://api.upcitemdb.com/prod/trial/lookup", params={"upc": gtin})
|
|
if response.status_code == 404:
|
|
return ProviderProduct(self.name, gtin, False, latency_ms=getattr(response, "latency_ms", 0), headers=dict(response.headers))
|
|
if response.status_code != 200:
|
|
raise ProviderError("BAD_RESPONSE", f"HTTP {response.status_code}")
|
|
try:
|
|
body = response.json()
|
|
except ValueError as error:
|
|
raise ProviderError("BAD_RESPONSE", "not JSON") from error
|
|
items = body.get("items") or []
|
|
headers = {k.lower(): v for k, v in response.headers.items()}
|
|
if not items:
|
|
return ProviderProduct(self.name, gtin, False, latency_ms=getattr(response, "latency_ms", 0), headers=headers)
|
|
item = items[0]
|
|
raw = hashlib.sha256(json.dumps(item, sort_keys=True, ensure_ascii=False).encode()).hexdigest()
|
|
return ProviderProduct(self.name, gtin, True, title=item.get("title"), brand=item.get("brand"), category=item.get("category"),
|
|
quantity=item.get("size") or item.get("weight") or None, raw_hash=raw, latency_ms=getattr(response, "latency_ms", 0), headers=headers)
|
|
|
|
def search_description(self, text: str) -> list[ProviderProduct]:
|
|
response = self._get("https://api.upcitemdb.com/prod/trial/search", params={"s": text, "match_mode": "1", "type": "product"})
|
|
if response.status_code != 200:
|
|
return []
|
|
out = []
|
|
for item in (response.json().get("items") or [])[:5]:
|
|
code = _digits(item.get("ean") or item.get("upc")) or ""
|
|
out.append(ProviderProduct(self.name, code, True, title=item.get("title"), brand=item.get("brand"), category=item.get("category"),
|
|
raw_hash=hashlib.sha256(json.dumps(item, sort_keys=True, ensure_ascii=False).encode()).hexdigest()))
|
|
return out
|
|
|
|
|
|
class OpenFoodFacts(Provider):
|
|
name = "openfoodfacts"
|
|
supports_search = True
|
|
rate_per_minute = 15
|
|
host = "world.openfoodfacts.org"
|
|
|
|
def lookup_gtin(self, gtin: str) -> ProviderProduct:
|
|
response = self._get(f"https://{self.host}/api/v2/product/{gtin}", params={"fields": "code,product_name,brands,quantity,categories,generic_name"})
|
|
headers = {k.lower(): v for k, v in response.headers.items()}
|
|
if response.status_code == 404:
|
|
return ProviderProduct(self.name, gtin, False, latency_ms=getattr(response, "latency_ms", 0), headers=headers)
|
|
if response.status_code != 200:
|
|
raise ProviderError("BAD_RESPONSE", f"HTTP {response.status_code}")
|
|
try:
|
|
body = response.json()
|
|
except ValueError as error:
|
|
raise ProviderError("BAD_RESPONSE", "not JSON") from error
|
|
product = body.get("product") or {}
|
|
if body.get("status") != 1 or not product:
|
|
return ProviderProduct(self.name, gtin, False, latency_ms=getattr(response, "latency_ms", 0), headers=headers)
|
|
title = product.get("product_name") or product.get("generic_name")
|
|
return ProviderProduct(self.name, gtin, bool(title), title=title, brand=(product.get("brands") or "").split(",")[0].strip() or None, category=(product.get("categories") or "").split(",")[0].strip() or None,
|
|
quantity=product.get("quantity"), raw_hash=hashlib.sha256(json.dumps(product, sort_keys=True, ensure_ascii=False).encode()).hexdigest(),
|
|
latency_ms=getattr(response, "latency_ms", 0), headers=headers)
|
|
|
|
def search_description(self, text: str) -> list[ProviderProduct]:
|
|
response = self._get("https://world.openfoodfacts.org/cgi/search.pl", params={"search_terms": text, "json": "1", "page_size": "5", "fields": "code,product_name,brands,quantity"})
|
|
if response.status_code != 200:
|
|
return []
|
|
return [ProviderProduct(self.name, _digits(p.get("code")) or "", True, title=p.get("product_name"), brand=(p.get("brands") or "").split(",")[0].strip() or None, quantity=p.get("quantity"))
|
|
for p in (response.json().get("products") or [])[:5] if p.get("product_name")]
|
|
|
|
|
|
class OpenProductsFacts(OpenFoodFacts):
|
|
"""Same open (ODbL) platform and API as Open Food Facts, for non-food goods. Identity only (no NCM)."""
|
|
name = "openproductsfacts"
|
|
host = "world.openproductsfacts.org"
|
|
supports_search = False
|
|
|
|
|
|
class OpenBeautyFacts(OpenFoodFacts):
|
|
name = "openbeautyfacts"
|
|
host = "world.openbeautyfacts.org"
|
|
supports_search = False
|
|
|
|
|
|
class Cosmos(Provider):
|
|
name = "cosmos"
|
|
supports_search = True
|
|
daily_limit = 25
|
|
provides_ncm = True
|
|
|
|
@property
|
|
def configured(self) -> bool:
|
|
return bool(self._token)
|
|
|
|
def _headers(self) -> dict:
|
|
return {"X-Cosmos-Token": self._token or "", "User-Agent": USER_AGENT, "Content-Type": "application/json"}
|
|
|
|
def lookup_gtin(self, gtin: str) -> ProviderProduct:
|
|
if not self.configured:
|
|
raise ProviderError("AUTH_ERROR", "NOT_CONFIGURED")
|
|
response = self._get(f"https://api.cosmos.bluesoft.com.br/gtins/{gtin}", headers=self._headers())
|
|
headers = {k.lower(): v for k, v in response.headers.items()}
|
|
if response.status_code == 404:
|
|
return ProviderProduct(self.name, gtin, False, latency_ms=getattr(response, "latency_ms", 0), headers=headers)
|
|
if response.status_code != 200:
|
|
raise ProviderError("BAD_RESPONSE", f"HTTP {response.status_code}")
|
|
try:
|
|
body = response.json()
|
|
except ValueError as error:
|
|
raise ProviderError("BAD_RESPONSE", "not JSON") from error
|
|
ncm = _digits((body.get("ncm") or {}).get("code"))
|
|
cest = _digits((body.get("cest") or {}).get("code"))
|
|
brand = (body.get("brand") or {}).get("name") if isinstance(body.get("brand"), dict) else body.get("brand")
|
|
return ProviderProduct(self.name, gtin, True, title=body.get("description"), brand=brand, category=(body.get("gpc") or {}).get("description"), quantity=body.get("net_weight") and str(body.get("net_weight")),
|
|
ncm=ncm, cest=cest, raw_hash=hashlib.sha256(json.dumps(body, sort_keys=True, ensure_ascii=False).encode()).hexdigest(), latency_ms=getattr(response, "latency_ms", 0), headers=headers)
|
|
|
|
def search_description(self, text: str) -> list[ProviderProduct]:
|
|
if not self.configured:
|
|
return []
|
|
response = self._get("https://api.cosmos.bluesoft.com.br/products", params={"query": text}, headers=self._headers())
|
|
if response.status_code != 200:
|
|
return []
|
|
out = []
|
|
for item in (response.json().get("products") or [])[:5]:
|
|
out.append(ProviderProduct(self.name, _digits(item.get("gtin")) or "", True, title=item.get("description"), brand=(item.get("brand") or {}).get("name") if isinstance(item.get("brand"), dict) else None,
|
|
ncm=_digits((item.get("ncm") or {}).get("code")) if isinstance(item.get("ncm"), dict) else None))
|
|
return out
|
|
|
|
|
|
def build_providers(settings, client: httpx.Client | None = None) -> list[Provider]:
|
|
"""Fixed order chosen from the benchmark (docs/product_enrichment.md): Cosmos (fiscal-capable) first when configured, then UPCitemdb, then the Open *Facts family (food, general products, beauty; identity only). Paid providers do not exist here."""
|
|
providers: list[Provider] = [Cosmos(client, getattr(settings, "cosmos_token", None)), UpcItemDb(client), OpenFoodFacts(client), OpenProductsFacts(client), OpenBeautyFacts(client)]
|
|
if any(p.paid for p in providers) and not getattr(settings, "allow_paid_providers", False):
|
|
raise RuntimeError("paid providers are not allowed (ALLOW_PAID_PROVIDERS=false)")
|
|
return providers
|