diff --git a/central/alembic/versions/0029_product_enrichment.py b/central/alembic/versions/0029_product_enrichment.py new file mode 100644 index 0000000..d1e5103 --- /dev/null +++ b/central/alembic/versions/0029_product_enrichment.py @@ -0,0 +1,140 @@ +"""Per-installation functions (corrector / enrichment mode), KB field status, enrichment state + audit, provider state/usage/cache, official source refresh log. + +Revision ID: 0029_product_enrichment +Revises: 0028_rtc_write_modes +""" +from alembic import op +import sqlalchemy as sa + +revision = "0029_product_enrichment" +down_revision = "0028_rtc_write_modes" +branch_labels = None +depends_on = None + + +def _idx(table, *cols): + for col in cols: + op.create_index(f"ix_{table}_{col}", table, [col]) + + +def upgrade(): + # existing installations keep behaving exactly as before: corrector on, functions already configured + op.add_column("installations", sa.Column("corrector_enabled", sa.Boolean(), nullable=False, server_default=sa.text("true"))) + op.add_column("installations", sa.Column("enrichment_mode", sa.String(16), nullable=False, server_default="DISABLED")) + op.add_column("installations", sa.Column("features_configured", sa.Boolean(), nullable=False, server_default=sa.text("true"))) + op.add_column("agent_enrollments", sa.Column("features_selected", sa.Boolean(), nullable=False, server_default=sa.text("false"))) + op.add_column("agent_enrollments", sa.Column("feature_corrector", sa.Boolean(), nullable=False, server_default=sa.text("false"))) + op.add_column("agent_enrollments", sa.Column("feature_enrichment_mode", sa.String(16), nullable=False, server_default="DISABLED")) + op.add_column("kb_knowledge", sa.Column("status", sa.String(16), nullable=False, server_default="EVIDENCE_ONLY")) + op.create_index("ix_kb_knowledge_status", "kb_knowledge", ["status"]) + + op.create_table( + "product_enrichments", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("tenant_id", sa.Integer(), sa.ForeignKey("tenants.id"), nullable=False), + sa.Column("installation_id", sa.Integer(), sa.ForeignKey("installations.id"), nullable=False), + sa.Column("product_code", sa.String(64), nullable=False), + sa.Column("gtin", sa.String(16), nullable=True), + sa.Column("description", sa.String(255), nullable=True), + sa.Column("status", sa.String(32), nullable=False, server_default="PENDING"), + sa.Column("decisions", sa.JSON(), nullable=False), + sa.Column("fields_applied", sa.JSON(), nullable=False), + sa.Column("human_overridden", sa.JSON(), nullable=False), + sa.Column("attempts", sa.Integer(), nullable=False, server_default="0"), + sa.Column("next_attempt_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("latency_ms", sa.Integer(), nullable=True), + sa.Column("detected_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.UniqueConstraint("installation_id", "product_code", name="uq_product_enrichment"), + ) + _idx("product_enrichments", "tenant_id", "installation_id", "gtin", "status", "next_attempt_at") + + op.create_table( + "product_field_audits", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("tenant_id", sa.Integer(), sa.ForeignKey("tenants.id"), nullable=False), + sa.Column("installation_id", sa.Integer(), sa.ForeignKey("installations.id"), nullable=False), + sa.Column("agent_id", sa.Integer(), sa.ForeignKey("agents.id"), nullable=False), + sa.Column("product_code", sa.String(64), nullable=False), + sa.Column("gtin", sa.String(16), nullable=True), + sa.Column("description_snapshot", sa.String(255), nullable=True), + sa.Column("field", sa.String(48), nullable=False), + sa.Column("old_value", sa.String(64), nullable=True), + sa.Column("new_value", sa.String(64), nullable=True), + sa.Column("rule", sa.String(64), nullable=True), + sa.Column("kb_status", sa.String(16), nullable=True), + sa.Column("evidence", sa.JSON(), nullable=False), + sa.Column("dataset_versions", sa.JSON(), nullable=False), + sa.Column("authorization_origin", sa.String(48), nullable=True), + sa.Column("job_id", sa.Integer(), nullable=True), + sa.Column("outcome", sa.String(24), nullable=False, server_default="APPLIED"), + sa.Column("reason", sa.String(96), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + ) + _idx("product_field_audits", "tenant_id", "installation_id", "agent_id", "product_code") + + op.create_table( + "provider_states", + sa.Column("provider", sa.String(32), primary_key=True), + sa.Column("configured", sa.Boolean(), nullable=False, server_default=sa.text("false")), + sa.Column("status", sa.String(16), nullable=False, server_default="NOT_CONFIGURED"), + sa.Column("quota_kind", sa.String(16), nullable=False, server_default="DOCUMENTED_RATE"), + sa.Column("daily_limit", sa.Integer(), nullable=True), + sa.Column("remaining", sa.Integer(), nullable=True), + sa.Column("reset_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("rate_per_minute", sa.Integer(), nullable=True), + sa.Column("circuit_open_until", sa.DateTime(timezone=True), nullable=True), + sa.Column("consecutive_failures", sa.Integer(), nullable=False, server_default="0"), + sa.Column("last_request_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("last_success_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("last_error", sa.String(160), nullable=True), + sa.Column("last_error_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("last_latency_ms", sa.Integer(), nullable=True), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + ) + op.create_table( + "provider_usage_daily", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("provider", sa.String(32), nullable=False), + sa.Column("day", sa.String(10), nullable=False), + *[sa.Column(name, sa.Integer(), nullable=False, server_default="0") for name in ("requests", "success", "not_found", "errors", "rate_limited", "timeouts", "cache_hits", "cache_misses")], + sa.UniqueConstraint("provider", "day", name="uq_provider_usage_day"), + ) + _idx("provider_usage_daily", "provider", "day") + op.create_table( + "gtin_lookup_cache", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("gtin", sa.String(16), nullable=False), + sa.Column("provider", sa.String(32), nullable=False), + sa.Column("found", sa.Boolean(), nullable=False), + sa.Column("data", sa.JSON(), nullable=False), + sa.Column("fetched_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False), + sa.UniqueConstraint("gtin", "provider", name="uq_gtin_lookup_cache"), + ) + _idx("gtin_lookup_cache", "gtin", "expires_at") + op.create_table( + "source_refresh_runs", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("source", sa.String(32), nullable=False), + sa.Column("source_url", sa.Text(), nullable=False), + sa.Column("started_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("status", sa.String(16), nullable=False, server_default="RUNNING"), + sa.Column("sha256", sa.String(64), nullable=True), + sa.Column("version_label", sa.String(64), nullable=True), + sa.Column("record_count", sa.Integer(), nullable=True), + sa.Column("error", sa.String(300), nullable=True), + ) + _idx("source_refresh_runs", "source") + + +def downgrade(): + for table in ("source_refresh_runs", "gtin_lookup_cache", "provider_usage_daily", "provider_states", "product_field_audits", "product_enrichments"): + op.drop_table(table) + op.drop_index("ix_kb_knowledge_status", table_name="kb_knowledge") + op.drop_column("kb_knowledge", "status") + for name in ("feature_enrichment_mode", "feature_corrector", "features_selected"): + op.drop_column("agent_enrollments", name) + for name in ("features_configured", "enrichment_mode", "corrector_enabled"): + op.drop_column("installations", name) diff --git a/central/app/enrichment/__init__.py b/central/app/enrichment/__init__.py new file mode 100644 index 0000000..2bb8b91 --- /dev/null +++ b/central/app/enrichment/__init__.py @@ -0,0 +1,2 @@ +"""Product enrichment (new product -> GTIN/name -> fiscal profile -> safe fill). Deterministic only: no AI, no LLM, no embeddings, no paid providers. +Everything lives inside the existing Hub/FastAPI/PostgreSQL; providers are hidden behind the Hub (Agents never call them).""" diff --git a/central/app/enrichment/gtin.py b/central/app/enrichment/gtin.py new file mode 100644 index 0000000..cd169fe --- /dev/null +++ b/central/app/enrichment/gtin.py @@ -0,0 +1,31 @@ +"""Deterministic GTIN normalization and validation (GTIN-8, UPC/GTIN-12, EAN/GTIN-13, GTIN-14).""" +import re + +VALID, INVALID, MISSING, RESTRICTED = "VALID_GTIN", "INVALID_GTIN", "MISSING_GTIN", "RESTRICTED_GTIN" + + +def check_digit_ok(digits: str) -> bool: + body, check = digits[:-1], int(digits[-1]) + total = sum(int(d) * (3 if i % 2 == 0 else 1) for i, d in enumerate(reversed(body))) + return (10 - total % 10) % 10 == check + + +def normalize_gtin(raw) -> dict: + """Returns {status, gtin (digits as typed), key (zero-padded to 14: the KB identity), reason}. Never calls anything external. + RESTRICTED: EAN prefixes 02, 04 and 20-29 are store-internal / restricted circulation (not a global product identity), so no provider is asked.""" + text = "" if raw is None else str(raw).strip() + if not text or text.upper() in ("SEM GTIN", "SEM EAN", "0", "NULL"): + return {"status": MISSING, "gtin": None, "key": None, "reason": "EMPTY"} + digits = re.sub(r"\D", "", text) + if not digits or len(digits) not in (8, 12, 13, 14): + return {"status": INVALID, "gtin": digits or None, "key": None, "reason": "BAD_LENGTH"} + if len(set(digits)) == 1: + return {"status": INVALID, "gtin": digits, "key": None, "reason": "REPEATED_DIGITS"} + if not check_digit_ok(digits): + return {"status": INVALID, "gtin": digits, "key": None, "reason": "BAD_CHECK_DIGIT"} + key = digits.zfill(14) + prefix = key[1:3] if len(digits) == 14 else key[1:3] # key = 0 + EAN-13 layout for 12/13; GTIN-14 keeps its indicator digit in key[0] + ean13 = key[1:] + if ean13[:2] in ("02", "04") or "20" <= ean13[:2] <= "29": + return {"status": RESTRICTED, "gtin": digits, "key": key, "reason": "STORE_INTERNAL_PREFIX"} + return {"status": VALID, "gtin": digits, "key": key, "reason": None} diff --git a/central/app/enrichment/kb.py b/central/app/enrichment/kb.py new file mode 100644 index 0000000..5eb744f --- /dev/null +++ b/central/app/enrichment/kb.py @@ -0,0 +1,120 @@ +"""Central product Knowledge Base access (same PostgreSQL, existing kb_* tables) with per-field status. + +Trust model (docs/product_enrichment.md): +- A provider value or a client observation is EVIDENCE. A field becomes VERIFIED_KB only when the official dataset confirms it AND at least two INDEPENDENT evidence + families agree with zero conflict (distinct providers each count once; client observations count once and only with >= 2 distinct clients). +- Human edits never enter the KB (HUMAN_OVERRIDDEN is per-installation protection, not knowledge). Company-specific fiscal decisions (CSOSN, CFOP, ...) never live here. +""" +from __future__ import annotations + +import hashlib +from datetime import datetime, timezone + +from sqlalchemy import select + +from ..models import KbKnowledge, KbProduct +from . import textnorm + +UNVERIFIED, EVIDENCE_ONLY, VERIFIED_KB, CONFLICT, STALE = "UNVERIFIED", "EVIDENCE_ONLY", "VERIFIED_KB", "CONFLICT", "STALE" +CLIENT_FAMILY = "CLIENT_OBSERVATION" + + +def _now(): + return datetime.now(timezone.utc) + + +def gtin_fingerprint(key: str) -> str: + return hashlib.sha256(f"gtin:{key}".encode()).hexdigest() + + +def find_by_gtin(session, key: str | None) -> KbProduct | None: + return session.scalar(select(KbProduct).where(KbProduct.gtin == key)) if key else None + + +def name_candidates(session, description) -> list[KbProduct]: + """Exact-fingerprint matches (0, 1 or several).""" + fp = textnorm.fingerprint(description) + if not fp: + return [] + ids = list(session.scalars(select(KbKnowledge.kb_product_id).where(KbKnowledge.knowledge_key == "name_fingerprint", KbKnowledge.value == fp).distinct())) + return [p for p in (session.get(KbProduct, i) for i in ids) if p] + + +def upsert_product(session, key: str | None, description, *, brand=None, quantity=None) -> KbProduct: + product = find_by_gtin(session, key) + if product is None: + fp = gtin_fingerprint(key) if key else hashlib.sha256(f"name:{textnorm.fingerprint(description)}".encode()).hexdigest() + product = session.scalar(select(KbProduct).where(KbProduct.fingerprint == fp)) + if product is None: + product = KbProduct(fingerprint=fp, gtin=key, identity_kind="GTIN" if key else "FINGERPRINT", normalized_description=textnorm.normalized(description)) + session.add(product) + session.flush() + if description and not product.normalized_description: + product.normalized_description = textnorm.normalized(description) + if brand and not product.brand: + product.brand = brand + if quantity and not product.package_quantity: + product.package_quantity = str(quantity)[:32] + fp = textnorm.fingerprint(description) + if fp: + add_evidence(session, product, "name_fingerprint", fp, "IDENTITY", family="IDENTITY", status=EVIDENCE_ONLY) + return product + + +def add_evidence(session, product: KbProduct, key: str, value: str, source: str, *, family: str, client_label: str | None = None, status: str = EVIDENCE_ONLY, info: dict | None = None) -> KbKnowledge: + row = session.scalar(select(KbKnowledge).where(KbKnowledge.kb_product_id == product.id, KbKnowledge.knowledge_key == key, KbKnowledge.value == value)) + if row is None: + row = KbKnowledge(kb_product_id=product.id, knowledge_key=key, value=value, source=source[:32], status=status, evidence={"families": {}, "clients": [], "info": info or {}}) + session.add(row) + session.flush() + evidence = dict(row.evidence or {}) + families = dict(evidence.get("families") or {}) + families[family] = _now().isoformat() + clients = list(evidence.get("clients") or []) + if client_label and client_label not in clients: + clients.append(client_label) + row.evidence = {**evidence, "families": families, "clients": clients, "info": {**(evidence.get("info") or {}), **(info or {})}} + return row + + +def independent_count(rows) -> tuple[int, list[str]]: + families = {} + clients: set[str] = set() + for row in rows: + ev = row.evidence or {} + for fam in (ev.get("families") or {}): + if fam.startswith("PROVIDER:"): + families[fam] = True + elif fam == CLIENT_FAMILY: + clients.update(ev.get("clients") or []) + names = sorted(families) + if len(clients) >= 2: + names.append(f"{CLIENT_FAMILY}(x{len(clients)})") + return len(names), names + + +def evaluate(session, product: KbProduct, key: str, is_official) -> dict: + """Recomputes and PERSISTS the status of one field. Returns {value, status, families, candidates, rejected}.""" + rows = list(session.scalars(select(KbKnowledge).where(KbKnowledge.kb_product_id == product.id, KbKnowledge.knowledge_key == key))) + if not rows: + return {"value": None, "status": UNVERIFIED, "families": [], "candidates": [], "rejected": []} + valid = [r for r in rows if is_official(r.value)] + rejected = [r.value for r in rows if not is_official(r.value)] + values = sorted({r.value for r in valid}) + if len(values) > 1: + for r in rows: + r.status = CONFLICT + return {"value": None, "status": CONFLICT, "families": [], "candidates": values, "rejected": rejected} + if not values: + for r in rows: + r.status = EVIDENCE_ONLY + return {"value": None, "status": EVIDENCE_ONLY, "families": [], "candidates": [], "rejected": rejected} + chosen = [r for r in valid if r.value == values[0]] + count, names = independent_count(chosen) + already = any(r.status == VERIFIED_KB and r.validated_at for r in chosen) + status = VERIFIED_KB if (count >= 2 or already) else EVIDENCE_ONLY + for r in rows: + r.status = status if r.value == values[0] else EVIDENCE_ONLY + if status == VERIFIED_KB and r.value == values[0] and not r.validated_at: + r.validated_at = _now() + return {"value": values[0], "status": status, "families": names, "candidates": values, "rejected": rejected} diff --git a/central/app/enrichment/providers.py b/central/app/enrichment/providers.py new file mode 100644 index 0000000..20d73b8 --- /dev/null +++ b/central/app/enrichment/providers.py @@ -0,0 +1,219 @@ +"""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 + + def lookup_gtin(self, gtin: str) -> ProviderProduct: + response = self._get(f"https://world.openfoodfacts.org/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 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 Open Food Facts. Paid providers do not exist here.""" + providers: list[Provider] = [Cosmos(client, getattr(settings, "cosmos_token", None)), UpcItemDb(client), OpenFoodFacts(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 diff --git a/central/app/enrichment/quota.py b/central/app/enrichment/quota.py new file mode 100644 index 0000000..e9dd413 --- /dev/null +++ b/central/app/enrichment/quota.py @@ -0,0 +1,230 @@ +"""Central quota / circuit breaker / cache / single-flight for external product providers. Quota belongs to the HUB: 70 Agents never spend it uncoordinated.""" +from __future__ import annotations + +import threading +import time +from collections import defaultdict, deque +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone + +from sqlalchemy import select + +from ..models import GtinLookupCache, ProviderState, ProviderUsageDaily +from .providers import Provider, ProviderError, ProviderProduct + +POSITIVE_TTL = timedelta(days=30) +NEGATIVE_TTL = timedelta(hours=24) +CIRCUIT_BASE = timedelta(seconds=60) +CIRCUIT_MAX = timedelta(minutes=15) +FAILURES_TO_OPEN = 3 + +_windows: dict[str, deque] = defaultdict(deque) # per-process sliding windows (burst / per-minute governors) +_flights: dict[tuple, threading.Lock] = {} +_flights_guard = threading.Lock() + + +def utcnow() -> datetime: + return datetime.now(timezone.utc) + + +def _aware(value): + if value is not None and value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value + + +def next_utc_midnight(now: datetime) -> datetime: + return (now + timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0) + + +@dataclass +class Outcome: + provider: str + state: str # CACHE_HIT | FOUND | NOT_FOUND | SKIPPED | ERROR + product: ProviderProduct | None = None + reason: str | None = None # NOT_CONFIGURED | CIRCUIT_OPEN | RATE_LIMITED | RATE_WINDOW | TIMEOUT | AUTH_ERROR | NETWORK | SERVER_ERROR | ... + retry_at: datetime | None = None + from_cache: bool = False + + +class QuotaManager: + def __init__(self, session, now=utcnow): + self.session, self.now = session, now + + # ---- state ------------------------------------------------------------------------------------------------------------------------------------- + def state(self, provider: Provider) -> ProviderState: + row = self.session.get(ProviderState, provider.name) + if row is None: + row = ProviderState(provider=provider.name, configured=provider.configured, status="ONLINE" if provider.configured else "NOT_CONFIGURED", + daily_limit=provider.daily_limit, rate_per_minute=provider.rate_per_minute, + quota_kind="LOCAL_ESTIMATE" if provider.daily_limit else "DOCUMENTED_RATE") + self.session.add(row) + self.session.flush() + row.configured = provider.configured + if not provider.configured: + row.status = "NOT_CONFIGURED" + elif row.status == "NOT_CONFIGURED": + row.status = "ONLINE" + row.daily_limit = provider.daily_limit + row.rate_per_minute = provider.rate_per_minute + return row + + def usage(self, provider_name: str, day: str | None = None) -> ProviderUsageDaily: + day = day or self.now().strftime("%Y-%m-%d") + row = self.session.scalar(select(ProviderUsageDaily).where(ProviderUsageDaily.provider == provider_name, ProviderUsageDaily.day == day)) + if row is None: + row = ProviderUsageDaily(provider=provider_name, day=day) + self.session.add(row) + self.session.flush() + return row + + # ---- decision ---------------------------------------------------------------------------------------------------------------------------------- + def allow(self, provider: Provider) -> tuple[bool, str | None, datetime | None]: + state, now = self.state(provider), self.now() + if not provider.configured: + return False, "NOT_CONFIGURED", None + if state.circuit_open_until and _aware(state.circuit_open_until) > now: + return False, "CIRCUIT_OPEN", _aware(state.circuit_open_until) + usage = self.usage(provider.name) + if provider.daily_limit and state.quota_kind == "REMOTE_QUOTA" and state.remaining is not None and state.remaining <= 0 and state.reset_at and _aware(state.reset_at) > now: + return False, "RATE_LIMITED", _aware(state.reset_at) + if provider.daily_limit and usage.requests >= provider.daily_limit and state.quota_kind != "REMOTE_QUOTA": + state.status = "RATE_LIMITED" + return False, "RATE_LIMITED", state.reset_at and _aware(state.reset_at) or next_utc_midnight(now) + window = (provider.burst[1] if provider.burst else 60) + limit = (provider.burst[0] if provider.burst else provider.rate_per_minute) + if limit: + q, cutoff = _windows[provider.name], time.monotonic() - window + while q and q[0] < cutoff: + q.popleft() + if len(q) >= limit: + return False, "RATE_WINDOW", now + timedelta(seconds=max(1, int(window - (time.monotonic() - q[0])))) + return True, None, None + + # ---- accounting -------------------------------------------------------------------------------------------------------------------------------- + def _remote_quota(self, state: ProviderState, headers: dict): + remaining, limit, reset = headers.get("x-ratelimit-remaining"), headers.get("x-ratelimit-limit"), headers.get("x-ratelimit-reset") + if remaining is not None and str(remaining).isdigit(): + state.remaining = int(remaining) + state.quota_kind = "REMOTE_QUOTA" + if limit and str(limit).isdigit(): + state.daily_limit = int(limit) + if reset and str(reset).isdigit(): + state.reset_at = datetime.fromtimestamp(int(reset), tz=timezone.utc) + + def _sent(self, provider: Provider): + q = _windows[provider.name] + q.append(time.monotonic()) + + def record_ok(self, provider: Provider, result: ProviderProduct): + state, usage, now = self.state(provider), self.usage(provider.name), self.now() + self._sent(provider) + usage.requests += 1 + usage.success += 1 + if not result.found: + usage.not_found += 1 + state.consecutive_failures = 0 + state.circuit_open_until = None + state.status = "ONLINE" + state.last_request_at = state.last_success_at = now + state.last_latency_ms = result.latency_ms or None + self._remote_quota(state, {k.lower(): v for k, v in (result.headers or {}).items()}) + if provider.daily_limit and state.quota_kind != "REMOTE_QUOTA": + state.remaining = max(0, provider.daily_limit - usage.requests) + state.reset_at = next_utc_midnight(now) + self.session.commit() + + def record_error(self, provider: Provider, error: ProviderError): + state, usage, now = self.state(provider), self.usage(provider.name), self.now() + self._sent(provider) + usage.requests += 1 + state.last_request_at, state.last_error_at = now, now + state.last_error = f"{error.kind}: {error.detail}"[:160] + state.consecutive_failures += 1 + headers = {k.lower(): v for k, v in (error.headers or {}).items()} + self._remote_quota(state, headers) + if error.kind == "RATE_LIMITED": + usage.rate_limited += 1 + state.status = "RATE_LIMITED" + wait = timedelta(seconds=error.retry_after) if error.retry_after else None + state.circuit_open_until = now + (wait or (state.reset_at and max(timedelta(minutes=5), _aware(state.reset_at) - now) or timedelta(minutes=15))) + if state.reset_at is None: + state.reset_at = state.circuit_open_until + elif error.kind == "AUTH_ERROR": + usage.errors += 1 + state.status = "AUTH_ERROR" + state.circuit_open_until = now + timedelta(hours=1) + else: + if error.kind == "TIMEOUT": + usage.timeouts += 1 + usage.errors += 1 + state.status = "DEGRADED" + if state.consecutive_failures >= FAILURES_TO_OPEN: + delay = min(CIRCUIT_MAX, CIRCUIT_BASE * (2 ** (state.consecutive_failures - FAILURES_TO_OPEN))) + state.circuit_open_until = now + delay + state.status = "OFFLINE" + self.session.commit() + + def record_cache(self, provider_name: str, hit: bool): + usage = self.usage(provider_name) + if hit: + usage.cache_hits += 1 + else: + usage.cache_misses += 1 + + +class ProviderGateway: + """lookup(provider, gtin): cache -> single-flight -> quota/circuit -> HTTP -> account -> cache.""" + + def __init__(self, session, providers: list[Provider], now=utcnow): + self.session, self.providers, self.now = session, providers, now + self.quota = QuotaManager(session, now) + + def _cached(self, provider: Provider, gtin: str) -> ProviderProduct | None | bool: + row = self.session.scalar(select(GtinLookupCache).where(GtinLookupCache.gtin == gtin, GtinLookupCache.provider == provider.name)) + if row is None or _aware(row.expires_at) <= self.now(): + return False + if not row.found: + return None + d = row.data or {} + return ProviderProduct(provider.name, gtin, True, title=d.get("title"), brand=d.get("brand"), category=d.get("category"), quantity=d.get("quantity"), ncm=d.get("ncm"), cest=d.get("cest"), raw_hash=d.get("raw_hash", "")) + + def _store(self, provider: Provider, gtin: str, product: ProviderProduct): + now = self.now() + ttl = POSITIVE_TTL if product.found else NEGATIVE_TTL + row = self.session.scalar(select(GtinLookupCache).where(GtinLookupCache.gtin == gtin, GtinLookupCache.provider == provider.name)) + data = {**product.as_data(), "raw_hash": product.raw_hash} + if row is None: + self.session.add(GtinLookupCache(gtin=gtin, provider=provider.name, found=product.found, data=data, fetched_at=now, expires_at=now + ttl)) + else: + row.found, row.data, row.fetched_at, row.expires_at = product.found, data, now, now + ttl + self.session.commit() + + def lookup(self, provider: Provider, gtin: str) -> Outcome: + hit = self._cached(provider, gtin) + if hit is not False: + self.quota.record_cache(provider.name, True) + self.session.commit() + return Outcome(provider.name, "CACHE_HIT", hit or None, from_cache=True) if hit else Outcome(provider.name, "NOT_FOUND", None, from_cache=True) + with _flights_guard: + lock = _flights.setdefault((provider.name, gtin), threading.Lock()) + with lock: # 10 installations, same EAN -> 1 external call, the rest wait for the same answer + self.session.expire_all() + hit = self._cached(provider, gtin) + if hit is not False: + self.quota.record_cache(provider.name, True) + self.session.commit() + return Outcome(provider.name, "CACHE_HIT", hit or None, from_cache=True) if hit else Outcome(provider.name, "NOT_FOUND", None, from_cache=True) + self.quota.record_cache(provider.name, False) + ok, reason, retry_at = self.quota.allow(provider) + if not ok: + self.session.commit() + return Outcome(provider.name, "SKIPPED", None, reason, retry_at) + try: + product = provider.lookup_gtin(gtin) + except ProviderError as error: + self.quota.record_error(provider, error) + return Outcome(provider.name, "ERROR", None, error.kind, self.quota.state(provider).circuit_open_until) + self.quota.record_ok(provider, product) + self._store(provider, gtin, product) + return Outcome(provider.name, "FOUND" if product.found else "NOT_FOUND", product if product.found else None) diff --git a/central/app/enrichment/resolver.py b/central/app/enrichment/resolver.py new file mode 100644 index 0000000..c5bd6ce --- /dev/null +++ b/central/app/enrichment/resolver.py @@ -0,0 +1,180 @@ +"""Product resolver: GTIN/name -> KB -> providers -> official NCM validation -> existing fiscal engine, decided FIELD BY FIELD. + +Deterministic. No AI, no scores: discrete states only. +Resolution states: EXACT_GTIN_VERIFIED, EXACT_KB_NAME_VERIFIED, MULTI_SOURCE_MATCH, SINGLE_WEAK_SOURCE, AMBIGUOUS, CONFLICT, NOT_FOUND +(+ operational: INVALID_GTIN, PENDING_PROVIDER_QUOTA, PENDING_PROVIDERS_OFFLINE). +Field decisions: AUTO_FILL_SAFE | SUGGESTION_ONLY | CONFLICT | INSUFFICIENT_EVIDENCE | ALREADY_SET | NOT_APPLICABLE. +""" +from __future__ import annotations + +from datetime import date, datetime + +from sqlalchemy import select + +from ..classification import Classifier, normalize_code +from ..datasets.service import DatasetService +from ..models import Dataset, NcmEntry +from . import kb, textnorm +from .gtin import normalize_gtin +from .quota import ProviderGateway + +AUTO, SUGGEST, CONFLICT, INSUFFICIENT, ALREADY, NA = "AUTO_FILL_SAFE", "SUGGESTION_ONLY", "CONFLICT", "INSUFFICIENT_EVIDENCE", "ALREADY_SET", "NOT_APPLICABLE" +FIELDS = ("CLASSIFICACAO_FISCAL", "CEST", "IPI") + + +def _parse_br(text: str | None) -> date | None: + try: + d, m, y = str(text).split("/") + return date(int(y), int(m), int(d)) + except (ValueError, TypeError): + return None + + +def ncm_is_official(session, code: str, today: date | None = None) -> bool: + """8-digit code present in the ACTIVE official NCM dataset and vigente today (dd/mm/yyyy, 31/12/9999 = open).""" + code = normalize_code(code) + if len(code) != 8: + return False + today = today or date.today() + rows = session.scalars(select(NcmEntry).join(Dataset).where(Dataset.kind == "NCM", Dataset.status == "ACTIVE", NcmEntry.code == code)).all() + for row in rows: + start, end = _parse_br(row.valid_from), _parse_br(row.valid_to) + if (start is None or start <= today) and (end is None or end >= today): + return True + return False + + +def dataset_versions(session) -> dict: + return {d.kind: {"sha256": (d.sha256 or "")[:12], "version": d.version_label} for d in session.scalars(select(Dataset).where(Dataset.status == "ACTIVE"))} + + +def _field(status, value=None, rule=None, kb_status=None, evidence=None, reason=None): + return {"status": status, "value": value, "rule": rule, "kb_status": kb_status, "evidence": evidence or [], "reason": reason} + + +def _empty(current: dict, name: str) -> bool: + return not str(current.get(name) or "").strip() + + +def _fiscal_decisions(session, ncm_eval: dict, gtin_kb_verified: bool, current: dict, description: str, *, name_verified=False) -> dict: + out = {} + if not _empty(current, "CLASSIFICACAO_FISCAL"): + out["CLASSIFICACAO_FISCAL"] = _field(ALREADY, reason="campo já preenchido") + elif ncm_eval["status"] == kb.CONFLICT: + out["CLASSIFICACAO_FISCAL"] = _field(CONFLICT, kb_status=kb.CONFLICT, evidence=ncm_eval["candidates"], reason="fontes divergem no NCM") + elif ncm_eval["value"] and ncm_eval["status"] == kb.VERIFIED_KB: + out["CLASSIFICACAO_FISCAL"] = _field(AUTO, ncm_eval["value"], "NCM_KB_VERIFIED" if gtin_kb_verified or name_verified else "NCM_TWO_INDEPENDENT_SOURCES", kb.VERIFIED_KB, ncm_eval["families"]) + elif ncm_eval["value"]: + out["CLASSIFICACAO_FISCAL"] = _field(SUGGEST, ncm_eval["value"], "NCM_SINGLE_SOURCE", kb.EVIDENCE_ONLY, ncm_eval["families"], "apenas uma fonte independente") + else: + out["CLASSIFICACAO_FISCAL"] = _field(INSUFFICIENT, reason="nenhuma fonte com NCM válido no dataset oficial") + ncm = out["CLASSIFICACAO_FISCAL"] + ncm_for_engine = ncm["value"] if ncm["status"] == AUTO else (normalize_code(current.get("CLASSIFICACAO_FISCAL")) or None) + if not ncm_for_engine or (ncm["status"] == ALREADY and not ncm_is_official(session, ncm_for_engine)): + for name in ("CEST", "IPI"): + out[name] = _field(ALREADY if not _empty(current, name) else INSUFFICIENT, reason="depende de um NCM oficial válido") + return out + decision = Classifier(DatasetService(session)).classify({"current_ncm": ncm_for_engine, "description": description}) + cest = (decision["classification"].get("cest") or []) + if not _empty(current, "CEST"): + out["CEST"] = _field(ALREADY, reason="campo já preenchido") + elif decision["cest_status"] == "CEST_CONFIRMED" and len(cest) == 1 and ncm["status"] == AUTO: + out["CEST"] = _field(AUTO, cest[0], "CEST_OFFICIAL_UNIQUE_MAPPING", ncm["kb_status"], [f"NCM {ncm_for_engine}"]) + elif decision["cest_status"] == "NO_CEST_MAPPING_FOUND": + out["CEST"] = _field(NA, reason="NCM sem mapeamento de CEST no dataset oficial") + else: + out["CEST"] = _field(INSUFFICIENT, reason=f"CEST não determinístico ({decision['cest_status']})") + ipi = decision["classification"].get("ipi") or {} + if not _empty(current, "IPI"): + out["IPI"] = _field(ALREADY, reason="campo já preenchido") + elif decision["ipi_status"] == "IPI_RESOLVED" and ipi.get("rate_kind") == "NUMERIC" and ncm["status"] == AUTO: + out["IPI"] = _field(AUTO, str(ipi["rate_numeric"]), "IPI_TIPI_NUMERIC", ncm["kb_status"], [f"TIPI NCM {ncm_for_engine}"]) + elif decision["ipi_status"] in {"IPI_RESOLVED"}: + out["IPI"] = _field(NA, reason="TIPI sem alíquota numérica (NT/branca)") + else: + out["IPI"] = _field(INSUFFICIENT, reason=f"IPI não determinístico ({decision['ipi_status']})") + return out + + +def resolve(session, gateway: ProviderGateway, *, gtin_raw, description, current: dict | None = None, client_label: str | None = None, allow_external: bool = True) -> dict: + """Returns {state, gtin_status, gtin, fields:{...}, providers:[...], retry_at, dataset_versions}. Never raises for provider trouble.""" + current = {k: v for k, v in (current or {}).items()} + g = normalize_gtin(gtin_raw) + result = {"state": "NOT_FOUND", "gtin_status": g["status"], "gtin": g.get("key"), "fields": {}, "providers": [], "retry_at": None, "dataset_versions": dataset_versions(session)} + session.info["is_official"] = lambda v: ncm_is_official(session, v) + is_official = session.info["is_official"] + product, name_verified, key = None, False, g.get("key") + + if g["status"] == "VALID_GTIN": # GTIN has absolute priority over the description + product = kb.find_by_gtin(session, key) + ev = kb.evaluate(session, product, "ncm", is_official) if product else None + if ev and ev["status"] == kb.VERIFIED_KB: + result["state"] = "EXACT_GTIN_VERIFIED" + result["fields"] = _fiscal_decisions(session, ev, True, current, description or "") + session.commit() + return result + query = g["gtin"] if len(g["gtin"]) in (8, 14) else key[1:] # providers/caches use the canonical EAN-13 (UPC-12 is zero-padded to it) + found, pending = [], [] + if allow_external: + for provider in gateway.providers: + out = gateway.lookup(provider, query) + result["providers"].append({"provider": out.provider, "state": out.state, "reason": out.reason, "from_cache": out.from_cache}) + if out.state in ("FOUND", "CACHE_HIT") and out.product: + found.append(out.product) + product = kb.upsert_product(session, key, out.product.title or description, brand=out.product.brand, quantity=out.product.quantity) + if out.product.ncm: + kb.add_evidence(session, product, "ncm", out.product.ncm, f"provider:{out.provider}"[:32], family=f"PROVIDER:{out.provider}", info={"raw_hash": out.product.raw_hash}) + ev = kb.evaluate(session, product, "ncm", is_official) + if ev["status"] == kb.VERIFIED_KB: + break # enough evidence: spare the remaining quota + elif out.state in ("SKIPPED", "ERROR"): + pending.append(out) + if product is None and description: + product = kb.upsert_product(session, key, description) + ev = kb.evaluate(session, product, "ncm", is_official) if product else {"value": None, "status": kb.UNVERIFIED, "families": [], "candidates": [], "rejected": []} + result["fields"] = _fiscal_decisions(session, ev, False, current, description or "") + if ev["status"] == kb.CONFLICT: + result["state"] = "CONFLICT" + elif ev["status"] == kb.VERIFIED_KB: + result["state"] = "MULTI_SOURCE_MATCH" if found else "EXACT_GTIN_VERIFIED" + elif len(found) >= 2: + result["state"] = "MULTI_SOURCE_MATCH" + elif found: + result["state"] = "SINGLE_WEAK_SOURCE" + elif pending and not found: + quota = any(p.reason in ("RATE_LIMITED", "RATE_WINDOW") for p in pending) + offline = [p for p in pending if p.reason not in ("NOT_CONFIGURED",)] + if offline: + result["state"] = "PENDING_PROVIDER_QUOTA" if quota else "PENDING_PROVIDERS_OFFLINE" + result["retry_at"] = min((p.retry_at for p in offline if p.retry_at), default=None) + session.commit() + return result + + if g["status"] == "INVALID_GTIN": + result["state"] = "INVALID_GTIN" # never spends provider quota on an invalid GTIN + + # ---- name only: only a UNIQUE, specific description with a VERIFIED_KB fingerprint may fill anything + if not textnorm.is_specific(description): + if result["state"] != "INVALID_GTIN": + result["state"] = "AMBIGUOUS" + result["fields"] = {f: _field(INSUFFICIENT, reason="descrição genérica demais e sem GTIN válido") if _empty(current, f) else _field(ALREADY) for f in FIELDS} + return result + candidates = kb.name_candidates(session, description) + verified = [] + for cand in candidates: + ev = kb.evaluate(session, cand, "ncm", is_official) + if ev["status"] == kb.VERIFIED_KB: + verified.append((cand, ev)) + if candidates and len(verified) == len(candidates) and len({ev["value"] for _, ev in verified}) == 1: + _, ev = verified[0] + result["state"] = "EXACT_KB_NAME_VERIFIED" + result["fields"] = _fiscal_decisions(session, ev, True, current, description or "", name_verified=True) + elif len({ev["value"] for _, ev in verified}) > 1: + result["state"] = "CONFLICT" + result["fields"] = {f: _field(CONFLICT, reason="descrição corresponde a produtos com NCM divergentes") if _empty(current, f) else _field(ALREADY) for f in FIELDS} + else: + if result["state"] != "INVALID_GTIN": + result["state"] = "AMBIGUOUS" if len(candidates) > 1 else "NOT_FOUND" + result["fields"] = {f: _field(INSUFFICIENT, reason="sem correspondência verificada na base de conhecimento") if _empty(current, f) else _field(ALREADY) for f in FIELDS} + session.commit() + return result diff --git a/central/app/enrichment/textnorm.py b/central/app/enrichment/textnorm.py new file mode 100644 index 0000000..1224b3f --- /dev/null +++ b/central/app/enrichment/textnorm.py @@ -0,0 +1,50 @@ +"""Deterministic product-description normalization and fingerprint (no ML). Used to FIND candidates; never authorizes a fiscal value on its own.""" +import hashlib +import re +import unicodedata + +UNIT_MAP = {"LT": "L", "LTS": "L", "LITRO": "L", "LITROS": "L", "L": "L", "ML": "ML", "KG": "KG", "KGS": "KG", "QUILO": "KG", "G": "G", "GR": "G", "GRS": "G", "GRAMAS": "G", "MG": "MG", "UN": "UN", "UND": "UN", "UNID": "UN", "UNIDADE": "UN", "PCT": "PCT", "CX": "CX"} +ABBREVIATIONS = {"REFRIG": "REFRIGERANTE", "REFRI": "REFRIGERANTE", "CHOC": "CHOCOLATE", "BISC": "BISCOITO", "SAB": "SABONETE", "DET": "DETERGENTE", "SH": "SHAMPOO", "COND": "CONDICIONADOR", + "MACARR": "MACARRAO", "MAC": "MACARRAO", "AMAC": "AMACIANTE", "DESOD": "DESODORANTE", "PAP": "PAPEL", "HIG": "HIGIENICO", "ORIG": "ORIGINAL", "TRAD": "TRADICIONAL"} +STOP = {"DE", "DA", "DO", "DAS", "DOS", "E", "COM", "C", "PARA", "P", "EM", "A", "O"} +_QTY = re.compile(r"^(\d+(?:[.,]\d+)?)(ML|L|KG|G|MG|UN|PCT|CX|LT|LTS|GR)$") + + +def _ascii_upper(text) -> str: + return unicodedata.normalize("NFKD", str(text or "")).encode("ascii", "ignore").decode().upper() + + +def tokens(description) -> list[str]: + text = _ascii_upper(description) + text = re.sub(r"(\d)([A-Z])", r"\1 \2", text) # 350ML -> 350 ML + text = re.sub(r"([A-Z])(\d)", r"\1 \2", text) + text = re.sub(r"(\d)[,.](\d)", r"\1.\2", text) # decimal comma -> dot (kept inside the number) + text = re.sub(r"[^A-Z0-9.]+", " ", text) + out = [] + for tok in text.split(): + tok = tok.strip(".") + if not tok or tok in STOP: + continue + tok = UNIT_MAP.get(tok, ABBREVIATIONS.get(tok, tok)) + out.append(tok) + return out + + +def normalized(description) -> str: + return " ".join(tokens(description)) + + +def fingerprint(description) -> str | None: + toks = tokens(description) + if not toks: + return None + return hashlib.sha256(" ".join(sorted(toks)).encode("utf-8")).hexdigest() + + +def is_specific(description) -> bool: + """A description strong enough to be used as an identity key: at least 3 informative tokens AND a quantity (number + unit), or >= 4 tokens. + Generic names (PIPOCA, OLEO, REFRIGERANTE, SABONETE) are NEVER specific.""" + toks = tokens(description) + has_qty = any(re.fullmatch(r"\d+(?:\.\d+)?", t) and i + 1 < len(toks) and toks[i + 1] in UNIT_MAP.values() for i, t in enumerate(toks)) + informative = [t for t in toks if not re.fullmatch(r"\d+(?:\.\d+)?", t) and t not in UNIT_MAP.values()] + return (len(informative) >= 2 and has_qty and len(toks) >= 3) or len(informative) >= 4 diff --git a/central/app/models.py b/central/app/models.py index 40a58fc..e0d8ace 100644 --- a/central/app/models.py +++ b/central/app/models.py @@ -285,6 +285,11 @@ class Installation(Base): # RTC auto-apply is authorized centrally per installation and is OFF until an operator explicitly turns it on (the Agent has its own second switch). rtc_auto_apply_authorized: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false") # legacy (0027); rtc_write_mode is the source of truth rtc_write_mode: Mapped[str] = mapped_column(String(24), default="DISABLED", server_default="DISABLED") # DISABLED | DRY_RUN | APPROVAL_REQUIRED | ENABLED + # Per-installation FUNCTIONS (central, no local switch). Existing installations keep working exactly as before (corrector on, features configured); + # a NEW installation starts WAITING_FEATURE_CONFIGURATION (features_configured=False): it heartbeats but processes nothing until an admin chooses. + corrector_enabled: Mapped[bool] = mapped_column(Boolean, default=False, server_default="true") + enrichment_mode: Mapped[str] = mapped_column(String(16), default="DISABLED", server_default="DISABLED") # DISABLED | DRY_RUN | ENABLED + features_configured: Mapped[bool] = mapped_column(Boolean, default=False, server_default="true") created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) @@ -447,6 +452,10 @@ class AgentEnrollment(Base): hostname_seen: Mapped[str | None] = mapped_column(String(255), nullable=True) hostname_mismatch: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false") agent_id: Mapped[int | None] = mapped_column(ForeignKey("agents.id"), nullable=True, index=True) + # Functions chosen when the installation was created (copied to the Installation at enrollment). Nothing chosen => WAITING_FEATURE_CONFIGURATION. + features_selected: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false") + feature_corrector: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false") + feature_enrichment_mode: Mapped[str] = mapped_column(String(16), default="DISABLED", server_default="DISABLED") updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) @@ -620,6 +629,7 @@ class KbKnowledge(Base): dataset_versions: Mapped[dict] = mapped_column(JSON, default=dict) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) validated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + status: Mapped[str] = mapped_column(String(16), default="EVIDENCE_ONLY", server_default="EVIDENCE_ONLY", index=True) # UNVERIFIED | EVIDENCE_ONLY | VERIFIED_KB | CONFLICT | STALE class KbObservation(Base): @@ -786,3 +796,114 @@ class RtcCandidate(Base): basis: Mapped[str | None] = mapped_column(String(255), nullable=True) status: Mapped[str] = mapped_column(String(16), default="PENDING", server_default="PENDING", index=True) # PENDING | APPROVED | REJECTED | APPLIED | STALE updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) + + +class ProductEnrichment(Base): + """Per (installation, product) state of the new-product enrichment. Historical products never get a row (no mass provider use).""" + __tablename__ = "product_enrichments" + __table_args__ = (UniqueConstraint("installation_id", "product_code", name="uq_product_enrichment"),) + id: Mapped[int] = mapped_column(primary_key=True) + tenant_id: Mapped[int] = mapped_column(ForeignKey("tenants.id"), index=True) + installation_id: Mapped[int] = mapped_column(ForeignKey("installations.id"), index=True) + product_code: Mapped[str] = mapped_column(String(64)) + gtin: Mapped[str | None] = mapped_column(String(16), nullable=True, index=True) + description: Mapped[str | None] = mapped_column(String(255), nullable=True) + status: Mapped[str] = mapped_column(String(32), default="PENDING", server_default="PENDING", index=True) + # PENDING | RESOLVED | APPLIED | PARTIAL | NOT_RESOLVED | CONFLICT | PENDING_PROVIDER_QUOTA | PENDING_PROVIDERS_OFFLINE | INVALID_GTIN + decisions: Mapped[dict] = mapped_column(JSON, default=dict) + fields_applied: Mapped[dict] = mapped_column(JSON, default=dict) + human_overridden: Mapped[dict] = mapped_column(JSON, default=dict) + attempts: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + next_attempt_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True) + latency_ms: Mapped[int | None] = mapped_column(Integer, nullable=True) + detected_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) + + +class ProductFieldAudit(Base): + """Append-only audit of every field the enrichment wrote (or refused to). Never holds secrets.""" + __tablename__ = "product_field_audits" + id: Mapped[int] = mapped_column(primary_key=True) + tenant_id: Mapped[int] = mapped_column(ForeignKey("tenants.id"), index=True) + installation_id: Mapped[int] = mapped_column(ForeignKey("installations.id"), index=True) + agent_id: Mapped[int] = mapped_column(ForeignKey("agents.id"), index=True) + product_code: Mapped[str] = mapped_column(String(64), index=True) + gtin: Mapped[str | None] = mapped_column(String(16), nullable=True) + description_snapshot: Mapped[str | None] = mapped_column(String(255), nullable=True) + field: Mapped[str] = mapped_column(String(48)) + old_value: Mapped[str | None] = mapped_column(String(64), nullable=True) + new_value: Mapped[str | None] = mapped_column(String(64), nullable=True) + rule: Mapped[str | None] = mapped_column(String(64), nullable=True) + kb_status: Mapped[str | None] = mapped_column(String(16), nullable=True) + evidence: Mapped[dict] = mapped_column(JSON, default=dict) + dataset_versions: Mapped[dict] = mapped_column(JSON, default=dict) + authorization_origin: Mapped[str | None] = mapped_column(String(48), nullable=True) + job_id: Mapped[int | None] = mapped_column(Integer, nullable=True) + outcome: Mapped[str] = mapped_column(String(24), default="APPLIED", server_default="APPLIED") # APPLIED | NO_WRITE_REQUIRED | ABORTED | HUMAN_OVERRIDDEN + reason: Mapped[str | None] = mapped_column(String(96), nullable=True) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + + +class ProviderState(Base): + """Live state of each external product provider (owned by the Hub: quota belongs to the Hub, never to Agents).""" + __tablename__ = "provider_states" + provider: Mapped[str] = mapped_column(String(32), primary_key=True) + configured: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false") + status: Mapped[str] = mapped_column(String(16), default="NOT_CONFIGURED", server_default="NOT_CONFIGURED") # ONLINE | DEGRADED | RATE_LIMITED | AUTH_ERROR | OFFLINE | NOT_CONFIGURED + quota_kind: Mapped[str] = mapped_column(String(16), default="DOCUMENTED_RATE", server_default="DOCUMENTED_RATE") # REMOTE_QUOTA | LOCAL_ESTIMATE | DOCUMENTED_RATE + daily_limit: Mapped[int | None] = mapped_column(Integer, nullable=True) + remaining: Mapped[int | None] = mapped_column(Integer, nullable=True) + reset_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + rate_per_minute: Mapped[int | None] = mapped_column(Integer, nullable=True) + circuit_open_until: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + consecutive_failures: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + last_request_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + last_success_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + last_error: Mapped[str | None] = mapped_column(String(160), nullable=True) + last_error_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + last_latency_ms: Mapped[int | None] = mapped_column(Integer, nullable=True) + updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) + + +class ProviderUsageDaily(Base): + __tablename__ = "provider_usage_daily" + __table_args__ = (UniqueConstraint("provider", "day", name="uq_provider_usage_day"),) + id: Mapped[int] = mapped_column(primary_key=True) + provider: Mapped[str] = mapped_column(String(32), index=True) + day: Mapped[str] = mapped_column(String(10), index=True) + requests: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + success: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + not_found: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + errors: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + rate_limited: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + timeouts: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + cache_hits: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + cache_misses: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + + +class GtinLookupCache(Base): + """Positive and negative (NOT_FOUND) provider results per GTIN, with TTL. Single place that spares quota across all installations.""" + __tablename__ = "gtin_lookup_cache" + __table_args__ = (UniqueConstraint("gtin", "provider", name="uq_gtin_lookup_cache"),) + id: Mapped[int] = mapped_column(primary_key=True) + gtin: Mapped[str] = mapped_column(String(16), index=True) + provider: Mapped[str] = mapped_column(String(32)) + found: Mapped[bool] = mapped_column(Boolean, default=False) + data: Mapped[dict] = mapped_column(JSON, default=dict) + fetched_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), index=True) + + +class SourceRefreshRun(Base): + """Observability of official fiscal sources (NCM Siscomex, ...): one row per refresh attempt.""" + __tablename__ = "source_refresh_runs" + id: Mapped[int] = mapped_column(primary_key=True) + source: Mapped[str] = mapped_column(String(32), index=True) + source_url: Mapped[str] = mapped_column(Text) + started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + status: Mapped[str] = mapped_column(String(16), default="RUNNING", server_default="RUNNING") # RUNNING | UPDATED | UNCHANGED | FAILED + sha256: Mapped[str | None] = mapped_column(String(64), nullable=True) + version_label: Mapped[str | None] = mapped_column(String(64), nullable=True) + record_count: Mapped[int | None] = mapped_column(Integer, nullable=True) + error: Mapped[str | None] = mapped_column(String(300), nullable=True) diff --git a/central/tests/test_product_enrichment_resolver.py b/central/tests/test_product_enrichment_resolver.py new file mode 100644 index 0000000..5381588 --- /dev/null +++ b/central/tests/test_product_enrichment_resolver.py @@ -0,0 +1,267 @@ +"""Product enrichment resolver: 12 resolution scenarios, quota/cache/circuit, provider fixtures. No internet.""" +import json +from datetime import datetime, timedelta, timezone + +import httpx +import pytest +from sqlalchemy import create_engine, select +from sqlalchemy.orm import Session +from sqlalchemy.pool import StaticPool + +from central.app.enrichment import kb +from central.app.enrichment.gtin import check_digit_ok +from central.app.enrichment.providers import Cosmos, OpenFoodFacts, Provider, ProviderError, ProviderProduct, UpcItemDb, build_providers +from central.app.enrichment.quota import ProviderGateway +from central.app.enrichment.resolver import resolve +from central.app.models import (Base, CestEntry, CestNcmEntry, Dataset, GtinLookupCache, KbKnowledge, NcmEntry, ProviderUsageDaily, TipiEntry) + + +def gtin(body12: str) -> str: + total = sum(int(d) * (3 if i % 2 == 0 else 1) for i, d in enumerate(reversed(body12))) + return body12 + str((10 - total % 10) % 10) + + +G1, G2, G3, G4 = gtin("789100010010"), gtin("789100010020"), gtin("789100010030"), gtin("789100010040") + + +class Fake(Provider): + def __init__(self, name, data=None, ncm_provider=False, error=None, daily_limit=None): + super().__init__(client=httpx.Client()) + self.name, self.data, self.provides_ncm, self.error, self.calls = name, data or {}, ncm_provider, error, 0 + self.daily_limit = daily_limit + + def lookup_gtin(self, g): + self.calls += 1 + if self.error: + raise self.error + d = self.data.get(g) or self.data.get(g.lstrip('0')) + return ProviderProduct(self.name, g, True, **d) if d else ProviderProduct(self.name, g, False) + + +@pytest.fixture() +def env(): + engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool) + Base.metadata.create_all(engine) + session = Session(engine) + ncm = Dataset(kind="NCM", provider="Siscomex", sha256="a" * 64, version_label="v", act="a", source_url="u", status="ACTIVE", record_count=3) + session.add(ncm) + session.flush() + for code in ("22021000", "19059090", "21069090"): + session.add(NcmEntry(dataset_id=ncm.id, code=code, description="x", valid_from="01/04/2022", valid_to="31/12/9999")) + session.add(NcmEntry(dataset_id=ncm.id, code="11111111", description="expired", valid_from="01/01/2000", valid_to="31/12/2001")) + tipi = Dataset(kind="TIPI_IPI", provider="RFB", sha256="b" * 64, version_label="v", act="a", source_url="u", status="ACTIVE", record_count=1) + cest = Dataset(kind="CEST", provider="CONFAZ", sha256="c" * 64, version_label="v", act="a", source_url="u", status="ACTIVE", record_count=1) + session.add_all([tipi, cest]) + session.flush() + session.add(TipiEntry(dataset_id=tipi.id, ncm="22021000", ex=None, description="x", ipi_rate=0.0, ipi_rate_text="0", source_row=1)) + entry = CestEntry(dataset_id=cest.id, code="03.007.00", description="REFRIGERANTE COLA LATA 350ML refrigerantes", annex="I") + session.add(entry) + session.flush() + session.add(CestNcmEntry(cest_entry_id=entry.id, ncm_code="22021000")) + session.commit() + yield session + session.close() + + +def run(session, providers, g, desc, current=None, **kw): + return resolve(session, ProviderGateway(session, providers), gtin_raw=g, description=desc, current=current or {}, **kw) + + +COSMOS = lambda ncm="22021000": {G1: dict(title="REFRIGERANTE COLA LATA 350ML", brand="X", ncm=ncm)} +UPC = {G1: dict(title="Refrigerante Cola Lata 350ml", brand="X")} + + +def test_1_gtin_verified_kb_no_provider_call(env): + p = kb.upsert_product(env, kb.gtin_fingerprint("0" + G1), "COLA", brand=None) if False else kb.upsert_product(env, "0" + G1, "REFRIGERANTE COLA LATA 350ML") + kb.add_evidence(env, p, "ncm", "22021000", "provider:cosmos", family="PROVIDER:cosmos") + kb.add_evidence(env, p, "ncm", "22021000", "provider:other", family="PROVIDER:other") + env.commit() + fake = Fake("cosmos", COSMOS(), True) + r = run(env, [fake], G1, "REFRIGERANTE COLA LATA 350ML") + assert r["state"] == "EXACT_GTIN_VERIFIED" and fake.calls == 0 + f = r["fields"] + assert f["CLASSIFICACAO_FISCAL"]["status"] == "AUTO_FILL_SAFE" and f["CLASSIFICACAO_FISCAL"]["value"] == "22021000" + assert f["CEST"]["value"] == "03.007.00" and f["IPI"]["value"] == "0.0" + + +def test_2_two_independent_providers_agree_multi_source(env): + a, b = Fake("cosmos", COSMOS(), True), Fake("other", {G1: dict(title="REFRIGERANTE COLA LATA 350ML", ncm="22021000")}, True) + r = run(env, [a, b], G1, "REFRIGERANTE COLA LATA 350ML") + assert r["state"] == "MULTI_SOURCE_MATCH" + assert r["fields"]["CLASSIFICACAO_FISCAL"]["status"] == "AUTO_FILL_SAFE" + # promoted: a second installation is served from the KB alone + a2 = Fake("cosmos", COSMOS(), True) + r2 = run(env, [a2], G1, "REFRIGERANTE COLA LATA 350ML") + assert r2["state"] == "EXACT_GTIN_VERIFIED" and a2.calls == 0 + + +def test_3_single_source_ncm_is_suggestion_only(env): + r = run(env, [Fake("cosmos", COSMOS(), True)], G1, "REFRIGERANTE COLA LATA 350ML") + assert r["fields"]["CLASSIFICACAO_FISCAL"]["status"] == "SUGGESTION_ONLY" + assert r["fields"]["CEST"]["status"] == "INSUFFICIENT_EVIDENCE" and r["fields"]["IPI"]["status"] == "INSUFFICIENT_EVIDENCE" + + +def test_4_identity_only_sources_never_give_ncm(env): + r = run(env, [Fake("upcitemdb", UPC), Fake("openfoodfacts", UPC)], G1, "REFRIGERANTE COLA LATA 350ML") + assert r["state"] == "MULTI_SOURCE_MATCH" + assert r["fields"]["CLASSIFICACAO_FISCAL"]["status"] == "INSUFFICIENT_EVIDENCE" + + +def test_5_conflicting_ncm_zero_write(env): + a, b = Fake("cosmos", COSMOS("22021000"), True), Fake("other", COSMOS("19059090"), True) + r = run(env, [a, b], G1, "REFRIGERANTE COLA LATA 350ML") + assert r["state"] == "CONFLICT" + assert all(v["status"] != "AUTO_FILL_SAFE" for v in r["fields"].values()) + + +def test_6_ncm_not_in_official_dataset_rejected(env): + a, b = Fake("cosmos", COSMOS("99999999"), True), Fake("other", COSMOS("99999999"), True) + r = run(env, [a, b], G1, "REFRIGERANTE COLA LATA 350ML") + assert r["fields"]["CLASSIFICACAO_FISCAL"]["status"] == "INSUFFICIENT_EVIDENCE" + a, b = Fake("cosmos", COSMOS("11111111"), True), Fake("other", COSMOS("11111111"), True) # expired NCM is not vigente + assert run(env, [a, b], gtin("789100010050"), "X")["fields"]["CLASSIFICACAO_FISCAL"]["status"] == "INSUFFICIENT_EVIDENCE" + + +def test_7_invalid_gtin_never_calls_providers(env): + fake = Fake("cosmos", COSMOS(), True) + r = run(env, [fake], G1[:-1] + str((int(G1[-1]) + 1) % 10), "PIPOCA") + assert r["gtin_status"] == "INVALID_GTIN" and r["state"] == "INVALID_GTIN" and fake.calls == 0 + + +def test_8_gtin_not_found_anywhere(env): + r = run(env, [Fake("cosmos", {}, True), Fake("upcitemdb")], G2, "PRODUTO QUALQUER") + assert r["state"] == "NOT_FOUND" and r["fields"]["CLASSIFICACAO_FISCAL"]["status"] == "INSUFFICIENT_EVIDENCE" + + +def test_9_name_only_exact_verified_fingerprint(env): + p = kb.upsert_product(env, "0" + G1, "REFRIGERANTE COLA ORIGINAL LATA 350 ML") + for fam in ("PROVIDER:a", "PROVIDER:b"): + kb.add_evidence(env, p, "ncm", "22021000", "x", family=fam) + env.commit() + r = run(env, [Fake("cosmos")], "", "REFRIGERANTE COLA ORIGINAL LATA 350ML") + assert r["state"] == "EXACT_KB_NAME_VERIFIED" and r["fields"]["CLASSIFICACAO_FISCAL"]["value"] == "22021000" + + +def test_10_generic_name_never_fills(env): + p = kb.upsert_product(env, "0" + G1, "REFRIGERANTE") + for fam in ("PROVIDER:a", "PROVIDER:b"): + kb.add_evidence(env, p, "ncm", "22021000", "x", family=fam) + env.commit() + r = run(env, [], "", "REFRIGERANTE") + assert r["state"] == "AMBIGUOUS" and all(v["status"] != "AUTO_FILL_SAFE" for v in r["fields"].values()) + + +def test_11_name_matching_different_ncms_is_conflict(env): + for g, ncm in ((G1, "22021000"), (G3, "21069090")): + p = kb.upsert_product(env, "0" + g, "SUCO NATURAL LARANJA GARRAFA 1L") + for fam in ("PROVIDER:a", "PROVIDER:b"): + kb.add_evidence(env, p, "ncm", ncm, "x", family=fam) + env.commit() + r = run(env, [], "", "SUCO NATURAL LARANJA GARRAFA 1L") + assert r["state"] == "CONFLICT" and all(v["status"] == "CONFLICT" for v in r["fields"].values()) + + +def test_12_already_filled_fields_are_never_touched(env): + p = kb.upsert_product(env, "0" + G1, "REFRIGERANTE COLA LATA 350ML") + for fam in ("PROVIDER:a", "PROVIDER:b"): + kb.add_evidence(env, p, "ncm", "22021000", "x", family=fam) + env.commit() + r = run(env, [], G1, "REFRIGERANTE COLA LATA 350ML", current={"CLASSIFICACAO_FISCAL": "19059090", "CEST": "", "IPI": None}) + assert r["fields"]["CLASSIFICACAO_FISCAL"]["status"] == "ALREADY_SET" + + +def test_client_consensus_needs_two_distinct_clients(env): + p = kb.upsert_product(env, "0" + G1, "X") + kb.add_evidence(env, p, "ncm", "22021000", "client", family=kb.CLIENT_FAMILY, client_label="A") + kb.add_evidence(env, p, "ncm", "22021000", "client", family=kb.CLIENT_FAMILY, client_label="A") + assert kb.evaluate(env, p, "ncm", lambda v: True)["status"] == kb.EVIDENCE_ONLY + kb.add_evidence(env, p, "ncm", "22021000", "client", family=kb.CLIENT_FAMILY, client_label="B") + assert kb.evaluate(env, p, "ncm", lambda v: True)["status"] == kb.EVIDENCE_ONLY # consensus alone is one family, never authority + kb.add_evidence(env, p, "ncm", "22021000", "provider", family="PROVIDER:cosmos") + assert kb.evaluate(env, p, "ncm", lambda v: True)["status"] == kb.VERIFIED_KB + + +def test_cache_spares_quota_across_installations(env): + fake = Fake("upcitemdb", UPC) + for _ in range(5): + run(env, [fake], G1, "REFRIGERANTE COLA LATA 350ML") + assert fake.calls == 1 + usage = env.scalar(select(ProviderUsageDaily).where(ProviderUsageDaily.provider == "upcitemdb")) + assert usage.requests == 1 and usage.cache_hits == 4 + + +def test_negative_cache(env): + fake = Fake("upcitemdb") + for _ in range(3): + run(env, [fake], G2, "PRODUTO X") + assert fake.calls == 1 + + +def test_quota_exhausted_pending_with_retry(env): + fake = Fake("cosmos", COSMOS(), True, daily_limit=1) + run(env, [fake], G1, "REFRIGERANTE COLA LATA 350ML") + r = run(env, [fake], G3, "OUTRO PRODUTO") + assert r["state"] == "PENDING_PROVIDER_QUOTA" and r["retry_at"] is not None and fake.calls == 1 + + +def test_circuit_breaker_and_timeout_is_not_a_crash(env): + fake = Fake("upcitemdb", error=ProviderError("TIMEOUT", "x")) + states = [run(env, [fake], gtin(f"78910001{i:04d}"), "P")["state"] for i in range(6)] + assert fake.calls == 3 # circuit opened after 3 failures, no more calls + assert set(states) == {"PENDING_PROVIDERS_OFFLINE"} + + +def test_cosmos_without_token_is_not_configured_and_costs_nothing(env): + cosmos = Cosmos(httpx.Client(transport=httpx.MockTransport(lambda r: pytest.fail("no call expected"))), None) + r = run(env, [cosmos], G1, "REFRIGERANTE COLA LATA 350ML") + assert r["providers"][0]["reason"] == "NOT_CONFIGURED" and r["state"] == "NOT_FOUND" + + +def test_paid_provider_guard(): + class S: allow_paid_providers = False; cosmos_token = None + assert [p.name for p in build_providers(S())] == ["cosmos", "upcitemdb", "openfoodfacts"] + + +# ---- provider contract tests against recorded fixtures (httpx.MockTransport, no network) ---------------------------------------------------------- +def _client(handler): + return httpx.Client(transport=httpx.MockTransport(handler)) + + +def test_upcitemdb_contract(): + seen = {} + + def handler(request): + seen["url"] = str(request.url) + return httpx.Response(200, json={"code": "OK", "total": 1, "items": [{"ean": G1, "title": "Cola 350ml", "brand": "X", "category": "Beverages", "size": "350ml"}]}, + headers={"X-RateLimit-Limit": "100", "X-RateLimit-Remaining": "99", "X-RateLimit-Reset": "1900000000"}) + p = UpcItemDb(_client(handler)).lookup_gtin(G1) + assert p.found and p.title == "Cola 350ml" and p.ncm is None and "trial/lookup" in seen["url"] + assert p.headers["x-ratelimit-remaining"] == "99" + assert not UpcItemDb(_client(lambda r: httpx.Response(200, json={"code": "OK", "total": 0, "items": []}))).lookup_gtin(G1).found + + +def test_openfoodfacts_contract_and_user_agent(): + seen = {} + + def handler(request): + seen["ua"] = request.headers.get("user-agent") + return httpx.Response(200, json={"status": 1, "product": {"product_name": "Biscoito", "brands": "Marca, Outra", "quantity": "130 g"}}) + off = OpenFoodFacts(httpx.Client(transport=httpx.MockTransport(handler), headers={"User-Agent": "GTIFiscalHub/1.0 (+contact)"})) + p = off.lookup_gtin(G1) + assert p.found and p.brand == "Marca" and "GTIFiscalHub" in seen["ua"] + assert not OpenFoodFacts(_client(lambda r: httpx.Response(200, json={"status": 0, "status_verbose": "product not found"}))).lookup_gtin(G1).found + + +def test_cosmos_contract_and_errors(): + body = {"description": "COCA COLA 350ML", "brand": {"name": "Coca"}, "ncm": {"code": "2202.10.00"}, "cest": {"code": "03.007.00"}} + p = Cosmos(_client(lambda r: httpx.Response(200, json=body)), "tok").lookup_gtin(G1) + assert p.ncm == "22021000" and p.cest == "0300700" and p.brand == "Coca" + assert not Cosmos(_client(lambda r: httpx.Response(404, json={})), "tok").lookup_gtin(G1).found + for status, kind in ((429, "RATE_LIMITED"), (401, "AUTH_ERROR"), (503, "SERVER_ERROR")): + with pytest.raises(ProviderError) as e: + Cosmos(_client(lambda r, s=status: httpx.Response(s)), "tok").lookup_gtin(G1) + assert e.value.kind == kind + def boom(request): raise httpx.ReadTimeout("t") + with pytest.raises(ProviderError) as e: + Cosmos(_client(boom), "tok").lookup_gtin(G1) + assert e.value.kind == "TIMEOUT"