diff --git a/.gitignore b/.gitignore index d712160..2bcfc01 100644 --- a/.gitignore +++ b/.gitignore @@ -47,3 +47,4 @@ scripts/_* sources_cache/legal/ scratch_rtc_e2e/ scratch_sandbox/ +scratch_enrich/ diff --git a/agent/pyproject.toml b/agent/pyproject.toml index 50c295a..89a595c 100644 --- a/agent/pyproject.toml +++ b/agent/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "gti-fiscal-agent" -version = "1.2.5" +version = "1.3.0" requires-python = ">=3.10" dependencies = ["httpx>=0.27,<1"] diff --git a/agent/scripts/run_agent_service.py b/agent/scripts/run_agent_service.py index 24eeabd..8d0ec21 100644 --- a/agent/scripts/run_agent_service.py +++ b/agent/scripts/run_agent_service.py @@ -107,7 +107,7 @@ class Config: self.poll_interval_seconds = float(env.get("POLL_INTERVAL_SECONDS", "5")) # RTC auto-apply is a SEPARATE switch (default OFF): FIREBIRD_WRITES alone never lets the Agent fill cClassTrib. self.rtc_auto_apply_max_per_job = int(env.get("RTC_AUTO_APPLY_MAX_PER_JOB", str(_rtc_apply.DEFAULT_MAX_PER_JOB))) - self.agent_version = env.get("AGENT_VERSION", "1.2.5") + self.agent_version = env.get("AGENT_VERSION", "1.3.0") self.schema_fingerprint = env.get("SCHEMA_FINGERPRINT", REFERENCE_SCHEMA_V1) self.state_db = env.get("AGENT_STATE_DB", str(AGENT_ROOT / "agent_state.sqlite")) # Pilot write authorization: both must be explicitly set by the installer for @@ -349,7 +349,7 @@ def execute_job(job, config, executor, state, env): "fiscal_context": fiscal_context, "rtc_auto_apply": rtc_apply, "fiscal_mode": getattr(config, "fiscal_mode", None), - "release_version": env.get("AGENT_VERSION") or "1.2.5", + "release_version": env.get("AGENT_VERSION") or "1.3.0", "products_total": len(products), "products_changed": len(to_process), "products_processed": summary["processed"], @@ -520,7 +520,7 @@ def run(config, env, *, max_iterations=None, stopper=None): def enforce_startup_preflight(env): """Full read-only preflight at startup (Hub, auth, Firebird, capabilities, company context, guards). Fail closed.""" from gti_fiscal_agent.preflight import run_preflight - report = run_preflight(env, agent_version=env.get("AGENT_VERSION") or "1.2.5") + report = run_preflight(env, agent_version=env.get("AGENT_VERSION") or "1.3.0") logger.info("startup preflight: %s", json.dumps({k: report[k] for k in ("PREFLIGHT", "blockers", "fiscal_mode", "effective_firebird_writes")}, ensure_ascii=False)) print(json.dumps({"PREFLIGHT": report["PREFLIGHT"], "blockers": report["blockers"]}, ensure_ascii=False)) if report["PREFLIGHT"] != "PASS": diff --git a/agent/src/gti_fiscal_agent/enrichment.py b/agent/src/gti_fiscal_agent/enrichment.py index c135c30..e635ff1 100644 --- a/agent/src/gti_fiscal_agent/enrichment.py +++ b/agent/src/gti_fiscal_agent/enrichment.py @@ -22,8 +22,8 @@ logger = logging.getLogger("gti_fiscal_agent.enrichment") CAPABILITY = "PRODUCT_ENRICHMENT_WRITE_V1" ALLOWED_FIELDS = ("CLASSIFICACAO_FISCAL", "CEST", "IPI") # explicit allowlist; anything else raises before touching SQL -READ_FIELDS = ("CODIGO", "PRODUTO", "GTIN") + ALLOWED_FIELDS -POLL_SECONDS = 3.0 +READ_FIELDS = ("CODIGO", "PRODUTO", "GTIN", "CODBARRA") + ALLOWED_FIELDS # the SGV operator types the EAN in CODBARRA; GTIN (fiscal) is mostly empty +POLL_SECONDS = 2.0 SAFETY_DIFF_SECONDS = 60.0 OVERRIDE_CHECK_SECONDS = 20.0 WATCH_WINDOW_SECONDS = 120.0 # a fresh product is re-read for a while: the operator may add the EAN seconds after the first save @@ -112,6 +112,21 @@ def _digits(value) -> str: return "".join(ch for ch in _text(value) if ch.isdigit()) +def gtin_check_ok(value) -> bool: + """Same acceptance as central.app.enrichment.gtin (8/12/13/14 digits + check digit); parity is asserted by a test. Only used to pick WHICH column to send: the Hub decides.""" + d = _digits(value) + if len(d) not in (8, 12, 13, 14) or len(set(d)) == 1: + return False + total = sum(int(x) * (3 if i % 2 == 0 else 1) for i, x in enumerate(reversed(d[:-1]))) + return (10 - total % 10) % 10 == int(d[-1]) + + +def pick_gtin(row) -> str: + """First VALID of GTIN / CODBARRA; otherwise whatever was typed (the Hub classifies it INVALID_GTIN/MISSING_GTIN without calling any provider).""" + candidates = [_text(row.get("GTIN")), _text(row.get("CODBARRA"))] + return next((c for c in candidates if gtin_check_ok(c)), next((c for c in candidates if c), "")) + + def read_codes(conn) -> set[str]: cur = conn.cursor() cur.execute("SELECT CODIGO FROM C000025") @@ -202,8 +217,20 @@ class EnrichmentWorker: self._last_diff = 0.0 self._last_override = 0.0 self._auth_cache = (None, 0.0) + self._count_ms = 0.0 + self._wconn = None self.metrics = {"detected": 0, "resolved": 0, "fields_written": 0, "detect_latency_ms": []} + def _open(self, factory, attempts=3): + """Firebird can drop a connection under load ('connection lost'): retry briefly instead of aborting the product.""" + for attempt in range(attempts): + try: + return factory() + except Exception: # noqa: BLE001 + if attempt == attempts - 1: + raise + self.sleep(0.3 * (attempt + 1)) + # ---- feature gate ------------------------------------------------------------------------------------------------------------------------------ def features(self, max_age=10.0) -> dict: if self._features is None or self.clock() - self._features_at > max_age: @@ -234,7 +261,7 @@ class EnrichmentWorker: # ---- detection --------------------------------------------------------------------------------------------------------------------------------- def detect(self) -> list[str]: - conn = self.read_conn() + conn = self._open(self.read_conn) try: if self.store.meta("baseline_done") != "1": codes = read_codes(conn) @@ -245,7 +272,9 @@ class EnrichmentWorker: self._last_count, self._last_diff = len(codes), self.clock() return [] started = self.clock() + t0 = time.monotonic() count = read_count(conn) + self._count_ms = (time.monotonic() - t0) * 1000 moved = self._last_count is None or count != self._last_count if not moved and self.clock() - self._last_diff < SAFETY_DIFF_SECONDS: return [] @@ -265,7 +294,7 @@ class EnrichmentWorker: # ---- one product --------------------------------------------------------------------------------------------------------------------------------- def process(self, item: dict): code, mode = item["code"], self.mode() - conn = self.read_conn() + conn = self._open(self.read_conn) try: rows = read_rows(conn, [code]) finally: @@ -274,14 +303,14 @@ class EnrichmentWorker: if row is None: self.store.update(code, state="GONE") return - snapshot = f"{_text(row['GTIN'])}|{_text(row['PRODUTO'])}" + snapshot = f"{pick_gtin(row)}|{_text(row['PRODUTO'])}" if mode not in ("DRY_RUN", "ENABLED"): self.store.update(code, state="SKIPPED_FEATURE", snapshot=snapshot) return attempts = int(item["attempts"] or 0) + 1 current = {f: row[f] for f in ALLOWED_FIELDS} try: - result = self.hub.resolve_product({"product_code": code, "gtin": _text(row["GTIN"]), "description": _text(row["PRODUTO"]), "current": {k: (None if v is None else str(v)) for k, v in current.items()}}) + result = self.hub.resolve_product({"product_code": code, "gtin": pick_gtin(row), "description": _text(row["PRODUTO"]), "current": {k: (None if v is None else str(v)) for k, v in current.items()}}) except Exception as error: # noqa: BLE001 - Hub offline / slow: the product stays pending, the SGV is unaffected delay = RETRY_BACKOFF[min(attempts - 1, len(RETRY_BACKOFF) - 1)] self.store.update(code, state="RETRY", attempts=attempts, next_attempt_at=self.clock() + delay, last_state=f"HUB_ERROR:{type(error).__name__}") @@ -322,6 +351,8 @@ class EnrichmentWorker: if outcome["outcome"] == "APPLIED": written[field] = value self.metrics["fields_written"] += 1 + self.store.update(code, written=written) # remembered at once: a crash right after a write still leaves it protected/tracked + self._close_write() self.store.update(code, state="DRY_RUN_DONE" if mode == "DRY_RUN" else "DONE", attempts=attempts, snapshot=snapshot, written=written, last_state=state, next_attempt_at=None) if reports: try: @@ -335,29 +366,31 @@ class EnrichmentWorker: logger.warning("reprocess request for %s failed", code) def _write_safely(self, code, field, value) -> dict: + """Per field: fresh central authorization (fail closed) + guarded typed UPDATE. write_field re-reads the field INSIDE its own transaction and refuses when anything is there.""" if not self._fresh_enabled(): return {"outcome": "ABORTED", "reason": "AUTHORIZATION_CHANGED"} - conn = self.read_conn() try: - fresh = read_rows(conn, [code]).get(code) - finally: - conn.close() - if fresh is None: - return {"outcome": "ABORTED", "reason": "PRODUCT_NOT_FOUND"} - if _text(fresh[field]): - return {"outcome": "ABORTED", "reason": "VALUE_ALREADY_SET_NEVER_OVERWRITTEN"} - wconn = self.write_conn() - try: - return write_field(wconn, code, field, value) - finally: - wconn.close() + if self._wconn is None: + self._wconn = self._open(self.write_conn) + return write_field(self._wconn, code, field, value) + except Exception as error: # noqa: BLE001 - one field failing never breaks the others or the SGV + self._close_write() + return {"outcome": "FAILED", "reason": type(error).__name__} + + def _close_write(self): + conn, self._wconn = self._wconn, None + if conn is not None: + try: + conn.close() + except Exception: # noqa: BLE001 + pass # ---- human override watch -------------------------------------------------------------------------------------------------------------------------- def check_overrides(self): tracked = self.store.tracked_written() if not tracked: return - conn = self.read_conn() + conn = self._open(self.read_conn) try: rows = read_rows(conn, [t["code"] for t in tracked]) finally: @@ -388,17 +421,23 @@ class EnrichmentWorker: self.detect() done = [] for item in self.store.due(self.clock()): - self.process(item) + try: + self.process(item) + except Exception as error: # noqa: BLE001 - keep the product retryable, never lose the batch + self._close_write() + attempts = int(item["attempts"] or 0) + 1 + self.store.update(item["code"], state="RETRY", attempts=attempts, next_attempt_at=self.clock() + RETRY_BACKOFF[min(attempts - 1, len(RETRY_BACKOFF) - 1)], last_state=f"ERROR:{type(error).__name__}") + logger.warning("enrichment of %s failed: %s", item["code"], type(error).__name__) done.append(item["code"]) for item in self.store.watching(self.clock()): # operator completes the record (EAN typed a moment later): re-resolve once per change rows = None - conn = self.read_conn() + conn = self._open(self.read_conn) try: rows = read_rows(conn, [item["code"]]) finally: conn.close() row = (rows or {}).get(item["code"]) - if row and f"{_text(row['GTIN'])}|{_text(row['PRODUTO'])}" != item.get("snapshot"): + if row and f"{pick_gtin(row)}|{_text(row['PRODUTO'])}" != item.get("snapshot"): self.process(item) done.append(item["code"]) if self.clock() - self._last_override >= OVERRIDE_CHECK_SECONDS: @@ -406,10 +445,14 @@ class EnrichmentWorker: self.check_overrides() return done + def poll_seconds(self) -> float: + """Adaptive: the SGV is never loaded by more than ~2.5% of one query-time. 3 s on small catalogs, longer when COUNT(*) is slow (huge catalogs); never above 30 s.""" + return min(30.0, max(POLL_SECONDS, 40 * self._count_ms / 1000)) + def run_forever(self, stopper): while not stopper.stop: try: self.step() except Exception: # noqa: BLE001 - the enrichment thread must never die or affect the SGV logger.exception("enrichment step failed") - self.sleep(POLL_SECONDS) + self.sleep(self.poll_seconds()) diff --git a/agent/src/gti_fiscal_agent/preflight.py b/agent/src/gti_fiscal_agent/preflight.py index f84217d..e065b42 100644 --- a/agent/src/gti_fiscal_agent/preflight.py +++ b/agent/src/gti_fiscal_agent/preflight.py @@ -9,7 +9,7 @@ import os import sqlite3 from pathlib import Path -AGENT_VERSION_DEFAULT = "1.2.5" +AGENT_VERSION_DEFAULT = "1.3.0" REQUIRED_ENV = ("GTI_HUB_URL", "AGENT_TOKEN", "FIREBIRD_DATABASE_PATH", "FB_USER", "FB_PASSWORD") REQUIRED_CAPABILITIES = ("supports_ncm", "supports_ipi", "supports_cest") FISCAL_MODES = ("DRY_RUN", "AUTO_APPLY") diff --git a/agent/tests/test_enrichment.py b/agent/tests/test_enrichment.py index cfdc8cf..1df92e3 100644 --- a/agent/tests/test_enrichment.py +++ b/agent/tests/test_enrichment.py @@ -66,7 +66,7 @@ DECISIONS = {"CLASSIFICACAO_FISCAL": {"status": "AUTO_FILL_SAFE", "value": "2202 @pytest.fixture() def env(tmp_path): db = sqlite3.connect(":memory:", check_same_thread=False) - db.execute("CREATE TABLE C000025(CODIGO TEXT PRIMARY KEY, PRODUTO TEXT, GTIN TEXT, CLASSIFICACAO_FISCAL TEXT, CEST TEXT, IPI REAL, OTHER TEXT)") + db.execute("CREATE TABLE C000025(CODIGO TEXT PRIMARY KEY, PRODUTO TEXT, GTIN TEXT, CLASSIFICACAO_FISCAL TEXT, CEST TEXT, IPI REAL, OTHER TEXT, CODBARRA TEXT)") db.executemany("INSERT INTO C000025(CODIGO,PRODUTO,GTIN,OTHER) VALUES(?,?,?,?)", [(f"H{i}", f"HIST {i}", "", "keep") for i in range(50)]) db.commit() hub = Hub(dict(ENABLED), DECISIONS) @@ -199,3 +199,13 @@ def test_hub_offline_keeps_product_pending_and_retries(env): hub.resolve_product = good now[0] += 40 assert worker.step() == ["N6"] and row(db, "N6")[0] == "22021000" + + +def test_gtin_check_parity_with_the_hub_and_codbarra_is_used(): + sys.path.insert(0, str(ROOT)) + from central.app.enrichment.gtin import normalize_gtin + for value in ("7891000100103", "7891000100104", "", "0", "12345678", "40170725", "036000291452", "11111111", "78910001001030", "abc", "2000000000008"): + assert E.gtin_check_ok(value) == (normalize_gtin(value)["status"] in ("VALID_GTIN", "RESTRICTED_GTIN")), value + assert E.pick_gtin({"GTIN": "", "CODBARRA": "7891000100103"}) == "7891000100103" + assert E.pick_gtin({"GTIN": "7891000100103", "CODBARRA": "999"}) == "7891000100103" + assert E.pick_gtin({"GTIN": "", "CODBARRA": "123"}) == "123" and E.pick_gtin({}) == "" diff --git a/central/app/enrichment/observability.py b/central/app/enrichment/observability.py new file mode 100644 index 0000000..adb815c --- /dev/null +++ b/central/app/enrichment/observability.py @@ -0,0 +1,82 @@ +"""Read-only views for the panel: product sources status, official dataset freshness and enrichment coverage metrics. Never exposes tokens.""" +from __future__ import annotations + +from datetime import datetime, timezone + +from sqlalchemy import func, select + +from ..config import settings +from ..datasets.updater import last_runs +from ..models import Dataset, GtinLookupCache, ProductEnrichment, ProviderState, ProviderUsageDaily +from .providers import build_providers + +STATUS_LABELS = {"ONLINE": "Online", "DEGRADED": "Instável", "RATE_LIMITED": "Limite atingido", "AUTH_ERROR": "Credencial recusada", "OFFLINE": "Indisponível", "NOT_CONFIGURED": "Não configurado"} +OFFICIAL_KINDS = (("NCM", "NCM (Siscomex)"), ("TIPI_IPI", "TIPI / IPI"), ("CEST", "CEST (CONFAZ)"), ("BA_ST_MVA", "ST/MVA Bahia")) + + +def _aware(dt): + return dt.replace(tzinfo=timezone.utc) if dt is not None and dt.tzinfo is None else dt + + +def provider_sources(session, providers=None) -> list[dict]: + providers = providers or build_providers(settings) + today = datetime.now(timezone.utc).strftime("%Y-%m-%d") + rows = [] + for provider in providers: + state = session.get(ProviderState, provider.name) + usage = session.scalar(select(ProviderUsageDaily).where(ProviderUsageDaily.provider == provider.name, ProviderUsageDaily.day == today)) + requests = usage.requests if usage else 0 + lookups = (usage.cache_hits + usage.cache_misses) if usage else 0 + status = ("NOT_CONFIGURED" if not provider.configured else (state.status if state else "ONLINE")) + rows.append({ + "provider": provider.name, "configured": bool(provider.configured), "status": status, "status_label": STATUS_LABELS.get(status, status), + "used": requests, "limit": provider.daily_limit or (state.daily_limit if state else None), "rate_per_minute": provider.rate_per_minute, + "remaining": state.remaining if state and state.remaining is not None else ((provider.daily_limit - requests) if provider.daily_limit else None), + "quota_kind": state.quota_kind if state else ("LOCAL_ESTIMATE" if provider.daily_limit else "DOCUMENTED_RATE"), + "reset_at": _aware(state.reset_at) if state and state.reset_at else None, + "last_success_at": _aware(state.last_success_at) if state and state.last_success_at else None, + "last_error": state.last_error if state else None, "last_error_at": _aware(state.last_error_at) if state and state.last_error_at else None, + "latency_ms": state.last_latency_ms if state else None, + "cache_hit_rate": round(100 * usage.cache_hits / lookups) if usage and lookups else None, + "errors": usage.errors if usage else 0, "rate_limited": usage.rate_limited if usage else 0, + }) + return rows + + +def official_sources(session) -> list[dict]: + out = [] + for kind, label in OFFICIAL_KINDS: + ds = session.scalar(select(Dataset).where(Dataset.kind == kind, Dataset.status == "ACTIVE").order_by(Dataset.id.desc()).limit(1)) + item = {"kind": kind, "label": label, "active": ds is not None, "records": ds.record_count if ds else None, "version": ds.version_label if ds else None, + "sha256": (ds.sha256[:12] if ds else None), "last_update": _aware(ds.downloaded_at) if ds else None, "status": "ATUALIZADA" if ds else "SEM DADOS", "error": None, "auto_refresh": kind == "NCM"} + if kind == "NCM": + recent = last_runs(session, "NCM", 1) + if recent: + last = recent[0] + item["last_check"], item["last_run_status"] = _aware(last.finished_at or last.started_at), last.status + if last.status == "FAILED": + item["status"], item["error"] = "FALHA NA ÚLTIMA ATUALIZAÇÃO (dados anteriores mantidos)", last.error + out.append(item) + return out + + +def coverage_metrics(session) -> dict: + total = session.scalar(select(func.count()).select_from(ProductEnrichment)) or 0 + by = dict(session.execute(select(ProductEnrichment.status, func.count()).group_by(ProductEnrichment.status)).all()) + with_gtin = session.scalar(select(func.count()).select_from(ProductEnrichment).where(ProductEnrichment.gtin.is_not(None))) or 0 + rows = [r.decisions or {} for r in session.scalars(select(ProductEnrichment).limit(5000))] + states = {} + for d in rows: + states[d.get("state")] = states.get(d.get("state"), 0) + 1 + + def rate(n): + return round(100 * n / total, 1) if total else None + auto_ncm = sum(1 for d in rows if ((d.get("fields") or {}).get("CLASSIFICACAO_FISCAL") or {}).get("status") == "AUTO_FILL_SAFE") + usage = session.execute(select(func.sum(ProviderUsageDaily.requests), func.sum(ProviderUsageDaily.cache_hits), func.sum(ProviderUsageDaily.cache_misses), func.sum(ProviderUsageDaily.rate_limited))).one() + requests, hits, misses, limited = (int(x or 0) for x in usage) + return {"tracked": total, "GTIN_VALID_RATE": rate(with_gtin), "KB_HIT_RATE": rate(states.get("EXACT_GTIN_VERIFIED", 0) + states.get("EXACT_KB_NAME_VERIFIED", 0)), + "EXTERNAL_LOOKUP_RATE": round(100 * requests / (requests + hits), 1) if (requests + hits) else None, + "PRODUCT_FOUND_RATE": rate(total - states.get("NOT_FOUND", 0) - states.get("INVALID_GTIN", 0) - states.get("AMBIGUOUS", 0)), + "NCM_AUTO_FILL_SAFE_RATE": rate(auto_ncm), "NAME_ONLY_RESOLUTION_RATE": rate(states.get("EXACT_KB_NAME_VERIFIED", 0)), "AMBIGUOUS_RATE": rate(states.get("AMBIGUOUS", 0)), + "CONFLICT_RATE": rate(states.get("CONFLICT", 0)), "PROVIDER_QUOTA_BLOCK_RATE": round(100 * limited / requests, 1) if requests else None, "states": states, "by_status": by, + "provider_cache_entries": session.scalar(select(func.count()).select_from(GtinLookupCache)) or 0, "provider_cache_hit_rate": round(100 * hits / (hits + misses), 1) if (hits + misses) else None} diff --git a/central/app/enrichment/observations.py b/central/app/enrichment/observations.py new file mode 100644 index 0000000..6853b63 --- /dev/null +++ b/central/app/enrichment/observations.py @@ -0,0 +1,27 @@ +"""ERP observations -> KB evidence. A client observation is EVIDENCE of family CLIENT_OBSERVATION; it never becomes VERIFIED_KB by itself +(needs the official dataset AND a second independent family, or >= 2 distinct clients + another family - see kb.evaluate).""" +from __future__ import annotations + +from ..classification import normalize_code +from . import kb +from .gtin import normalize_gtin + + +def observe(session, rows, client_label: str, is_official) -> dict: + """rows: iterable of {"GTIN","PRODUTO","NCM"}. Only VALID_GTIN rows with an OFFICIAL 8-digit NCM are recorded. Returns counters.""" + stats = {"rows": 0, "valid_gtin": 0, "with_official_ncm": 0, "recorded": 0} + for row in rows: + stats["rows"] += 1 + g = normalize_gtin(row.get("GTIN")) + if g["status"] != "VALID_GTIN": + continue + stats["valid_gtin"] += 1 + ncm = normalize_code(row.get("NCM")) + if len(ncm) != 8 or not is_official(ncm): + continue + stats["with_official_ncm"] += 1 + product = kb.upsert_product(session, g["key"], row.get("PRODUTO")) + kb.add_evidence(session, product, "ncm", ncm, "client:observation", family=kb.CLIENT_FAMILY, client_label=client_label) + stats["recorded"] += 1 + session.flush() + return stats diff --git a/central/app/enrichment/providers.py b/central/app/enrichment/providers.py index 20d73b8..9f9c3cd 100644 --- a/central/app/enrichment/providers.py +++ b/central/app/enrichment/providers.py @@ -138,9 +138,10 @@ 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://world.openfoodfacts.org/api/v2/product/{gtin}", params={"fields": "code,product_name,brands,quantity,categories,generic_name"}) + 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) @@ -166,6 +167,19 @@ class OpenFoodFacts(Provider): 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 @@ -212,8 +226,8 @@ class Cosmos(Provider): 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)] + """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 diff --git a/central/app/enrichment/resolver.py b/central/app/enrichment/resolver.py index d29153e..689b1bb 100644 --- a/central/app/enrichment/resolver.py +++ b/central/app/enrichment/resolver.py @@ -45,6 +45,17 @@ def ncm_is_official(session, code: str, today: date | None = None) -> bool: return False +class _ScopedDatasets(DatasetService): + """The Classifier only needs the ONE exact NCM row here; loading the whole 15k-row catalogue on every product would cost ~300 ms of the 5 s budget.""" + + def __init__(self, session, code: str): + super().__init__(session) + self.code = code + + def active_entries(self): + return self.session.scalars(select(NcmEntry).join(Dataset).where(Dataset.kind == "NCM", Dataset.status == "ACTIVE", NcmEntry.code == self.code)).all() + + 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"))} @@ -75,11 +86,13 @@ def _fiscal_decisions(session, ncm_eval: dict, gtin_kb_verified: bool, current: 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 / IPI are deterministic functions of an OFFICIAL NCM: the NCM we are about to write (AUTO) or one already in the product (validated above against the official dataset) + ncm_solid = ncm["status"] == AUTO or (ncm["status"] == ALREADY and ncm_is_official(session, ncm_for_engine)) + decision = Classifier(_ScopedDatasets(session, ncm_for_engine)).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: + elif decision["cest_status"] == "CEST_CONFIRMED" and len(cest) == 1 and ncm_solid: out["CEST"] = _field(AUTO, normalize_code(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") @@ -88,7 +101,7 @@ def _fiscal_decisions(session, ncm_eval: dict, gtin_kb_verified: bool, current: 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: + elif decision["ipi_status"] == "IPI_RESOLVED" and ipi.get("rate_kind") == "NUMERIC" and ncm_solid: 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)") diff --git a/central/app/version.py b/central/app/version.py index 3997907..40155f6 100644 --- a/central/app/version.py +++ b/central/app/version.py @@ -4,5 +4,5 @@ kept identical to this) and what an AgentRelease row's `version` column holds once published — never a second, independent version number. Fiscal semantics are versioned separately (ENGINE_VERSION/DECISION_VERSION in classification.py, DECISION_ENGINE_VERSION etc.) and are NOT changed by a release bump.""" -GTI_FISCAL_VERSION = "1.2.5" +GTI_FISCAL_VERSION = "1.3.0" API_COMPAT_MAJOR_MINOR = "1.0" # an Agent 1.0.x/1.1.x talks to any Hub 1.0.x diff --git a/central/app/web.py b/central/app/web.py index 59fa081..e76ed37 100644 --- a/central/app/web.py +++ b/central/app/web.py @@ -22,6 +22,7 @@ from .panel_presentation import format_ui_datetime from .models import AdminUser, Agent, AgentEnrollment, AgentRelease, AgentSnapshot, BlockedProduct, FiscalContextEntry, Installation, Job, ProductEnrichment, ProductFieldAudit, RtcCandidate, RtcWriteAudit, Tenant from .jobs import create_job from .audit import record_audit +from .enrichment.observability import coverage_metrics, official_sources, provider_sources def router(session_factory): @@ -160,7 +161,7 @@ def router(session_factory): def dashboard(request: Request): session, auth_data, redirect = page_auth(request) if redirect: return redirect - try: return render(request, "dashboard.html", {"summary": dashboard_summary(session), "auth": auth_data}) + try: return render(request, "dashboard.html", {"summary": dashboard_summary(session), "auth": auth_data, "sources": provider_sources(session), "official": official_sources(session), "enrich_metrics": coverage_metrics(session)}) finally: session.close() @routes.get("/agents", response_class=HTMLResponse) diff --git a/central/pyproject.toml b/central/pyproject.toml index 857c5a1..aa31279 100644 --- a/central/pyproject.toml +++ b/central/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "gti-fiscal-hub" -version = "1.2.5" +version = "1.3.0" requires-python = ">=3.10" dependencies = [ "fastapi>=0.115,<1", diff --git a/central/templates/dashboard.html b/central/templates/dashboard.html index 744f9f4..48eaabb 100644 --- a/central/templates/dashboard.html +++ b/central/templates/dashboard.html @@ -12,6 +12,23 @@ {% endfor %} +
+
+

Fontes de produtos

Consultas feitas somente pelo Hub, dentro do limite gratuito.

+ {% for src in sources %}
{{ src.provider }} {{ src.status_label }} + {% if src.limit %}{{ src.used }} / {{ src.limit }} consultas hoje · restam {{ src.remaining if src.remaining is not none else "—" }}{% if src.reset_at %} · renova {{ src.reset_at.strftime("%d/%m %H:%M") }} UTC{% endif %}{% elif src.rate_per_minute %}{{ src.rate_per_minute }} consultas/min (limite documentado) · {{ src.used }} hoje{% else %}{{ src.used }} consultas hoje{% endif %} + {% if src.latency_ms %} · {{ src.latency_ms }} ms{% endif %}{% if src.cache_hit_rate is not none %} · cache {{ src.cache_hit_rate }}%{% endif %} + {% if src.last_success_at %} · último sucesso {{ src.last_success_at.strftime("%d/%m %H:%M") }}{% endif %}{% if src.last_error %} · último erro: {{ src.last_error }}{% endif %} + · configurado: {{ "sim" if src.configured else "não" }}
{% endfor %} +
+
+

Fontes oficiais

Bases fiscais usadas para validar qualquer preenchimento.

+ {% for o in official %}
{{ o.label }} {{ o.status }} + {% if o.active %}{{ o.records }} registros · {{ o.version }} · {{ o.sha256 }}{% if o.last_update %} · baixada {{ o.last_update.strftime("%d/%m/%Y") }}{% endif %}{% endif %}{% if o.auto_refresh %} · atualização automática diária{% endif %}{% if o.error %} · {{ o.error }}{% endif %}
{% endfor %} +
+
+

Enriquecimento de produtos novos

Cobertura medida sobre os produtos novos acompanhados (o catálogo histórico não entra).

+{% if enrich_metrics.tracked %}
{% for key,label in [("GTIN_VALID_RATE","Com GTIN válido"),("KB_HIT_RATE","Resolvidos pela base"),("PRODUCT_FOUND_RATE","Produto identificado"),("NCM_AUTO_FILL_SAFE_RATE","NCM preenchível com segurança"),("AMBIGUOUS_RATE","Ambíguos"),("CONFLICT_RATE","Em conflito")] %}
{{ enrich_metrics[key] if enrich_metrics[key] is not none else "—" }}{% if enrich_metrics[key] is not none %}%{% endif %}{{ label }}
{% endfor %}
{% else %}

Nenhum produto novo acompanhado ainda.

{% endif %}

Cobertura dos produtos

Ver servidores
diff --git a/central/tests/test_health.py b/central/tests/test_health.py index ad8085a..e5b6d79 100644 --- a/central/tests/test_health.py +++ b/central/tests/test_health.py @@ -17,7 +17,7 @@ def test_health_is_unavailable_when_database_is_down(): def test_version_endpoint_reports_release_and_fiscal_versions(): body = TestClient(create_app(database_probe=lambda: True)).get("/version").json() - assert body["version"] == "1.2.5" and body["api_compat"] == "1.0" and body["engine_version"] and body["decision_version"] + assert body["version"] == "1.3.0" and body["api_compat"] == "1.0" and body["engine_version"] and body["decision_version"] def test_production_refuses_the_dev_token(monkeypatch): diff --git a/central/tests/test_product_enrichment_api.py b/central/tests/test_product_enrichment_api.py index 0b938ce..074c58a 100644 --- a/central/tests/test_product_enrichment_api.py +++ b/central/tests/test_product_enrichment_api.py @@ -94,3 +94,67 @@ def test_features_route_is_admin_only_csrf_and_audited(): assert "INSTALLATION_FEATURES" in [e.action for e in s.scalars(select(AuditEvent))] page = client.get(f"/agents/{agent_id}") assert "Funcionalidades" in page.text and "Enriquecimento" in page.text + + +def test_dashboard_shows_sources_without_tokens_and_official_freshness(monkeypatch): + from central.app.enrichment import observability, providers + monkeypatch.setattr(observability, "build_providers", lambda _s: providers.build_providers(type("S", (), {"cosmos_token": "SEGREDO-NAO-VAZAR", "allow_paid_providers": False})())) + client, agent_id, _ = _panel_client() + page = client.get("/").text + assert "Fontes de produtos" in page and "cosmos" in page and "configurado: sim" in page and "upcitemdb" in page and "openfoodfacts" in page + assert "SEGREDO-NAO-VAZAR" not in page and "Fontes oficiais" in page and "NCM (Siscomex)" in page + + +def _second_agent(engine, mode="ENABLED", corrector=True): + from central.app.auth import token_hash + from central.app.models import Agent, Tenant + with Session(engine) as s: + first = s.scalars(select(Installation)).first() + agent = Agent(tenant_id=first.tenant_id, name="second", token_hash=token_hash("second-token"), enabled=True) + s.add(agent) + s.flush() + s.add(Installation(agent_id=agent.id, tenant_id=first.tenant_id, installation_id="00000000-0000-4000-8000-0000000000bb", display_name="B", features_configured=True, corrector_enabled=corrector, enrichment_mode=mode)) + s.commit() + return {"Authorization": "Bearer second-token"} + + +def test_same_product_in_another_installation_costs_zero_extra_provider_requests(monkeypatch): + client, engine = setup(monkeypatch) + h2 = _second_agent(engine) + providers = enrich_api.PROVIDERS_OVERRIDE + first = resolve(client) + calls_after_first = [p.calls for p in providers] + second = client.post("/v1/products/resolve", json={"product_code": "X9", "gtin": G1, "description": "REFRIGERANTE COLA LATA 350ML", "current": {}}, headers=h2).json() + assert [p.calls for p in providers] == calls_after_first # other installation: served by KB, zero external calls + assert second["state"] == "EXACT_GTIN_VERIFIED" and second["fields"]["CLASSIFICACAO_FISCAL"]["value"] == first["fields"]["CLASSIFICACAO_FISCAL"]["value"] + with Session(engine) as s: + assert {r.installation_id for r in s.scalars(select(ProductEnrichment))} == {1, 2} + + +def test_installations_are_isolated_by_their_own_feature_flags(monkeypatch): + client, engine = setup(monkeypatch, "DISABLED") + h2 = _second_agent(engine, "ENABLED") + assert resolve(client)["state"] == "FEATURE_DISABLED" + body = client.post("/v1/products/resolve", json={"product_code": "N1", "gtin": G1, "description": "REFRIGERANTE COLA LATA 350ML", "current": {}}, headers=h2).json() + assert body["state"] != "FEATURE_DISABLED" and body["fields"]["CLASSIFICACAO_FISCAL"]["status"] == "AUTO_FILL_SAFE" + + +def test_quota_exhausted_gives_pending_state_and_retry_time_through_the_api(monkeypatch): + client, engine = setup(monkeypatch) + monkeypatch.setattr(enrich_api, "PROVIDERS_OVERRIDE", [Fake("cosmos", COSMOS(), True, daily_limit=1)]) + resolve(client, "A1", G1) + body = resolve(client, "A2", G4, "OUTRO PRODUTO") + assert body["state"] == "PENDING_PROVIDER_QUOTA" and body["retry_at"] + with Session(engine) as s: + row = s.scalars(select(ProductEnrichment).where(ProductEnrichment.product_code == "A2")).one() + assert row.status == "PENDING" and row.next_attempt_at is not None + assert s.scalars(select(BlockedProduct).where(BlockedProduct.reason_code == "PENDING_PROVIDER_QUOTA")).one().field == "ENRICHMENT" + + +def test_corrector_switch_off_stops_jobs_but_not_enrichment(monkeypatch): + client, engine = setup(monkeypatch) + with Session(engine) as s: + s.scalars(select(Installation)).one().corrector_enabled = False + s.commit() + assert client.get("/v1/jobs/next", headers=H).json() is None + assert resolve(client)["state"] != "FEATURE_DISABLED" diff --git a/central/tests/test_product_enrichment_resolver.py b/central/tests/test_product_enrichment_resolver.py index d793446..33d5ee8 100644 --- a/central/tests/test_product_enrichment_resolver.py +++ b/central/tests/test_product_enrichment_resolver.py @@ -223,7 +223,7 @@ def test_cosmos_without_token_is_not_configured_and_costs_nothing(env): 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"] + assert [p.name for p in build_providers(S())] == ["cosmos", "upcitemdb", "openfoodfacts", "openproductsfacts", "openbeautyfacts"] # ---- provider contract tests against recorded fixtures (httpx.MockTransport, no network) ---------------------------------------------------------- diff --git a/deploy/agent-windows/agent.env.template b/deploy/agent-windows/agent.env.template index c351d3f..620f416 100644 --- a/deploy/agent-windows/agent.env.template +++ b/deploy/agent-windows/agent.env.template @@ -19,4 +19,4 @@ FB_PASSWORD= FISCAL_MODE=DRY_RUN FIREBIRD_WRITES=0 POLL_INTERVAL_SECONDS=5 -AGENT_VERSION=1.2.4 +AGENT_VERSION=1.3.0 diff --git a/deploy/agent-windows/gti-agent.ps1 b/deploy/agent-windows/gti-agent.ps1 index 9ab8820..fb9546a 100644 --- a/deploy/agent-windows/gti-agent.ps1 +++ b/deploy/agent-windows/gti-agent.ps1 @@ -164,7 +164,7 @@ switch ($Command) { "GTI_HUB_URL=$hub", "AGENT_TOKEN=$token", "AGENT_ID=$agentId", "COMPANY_CODE=$code", "COMPANY_CNPJ=$cnpj", "COMPANY_CRT_SOURCE=$crtSource", "FIREBIRD_DATABASE_PATH=$db", "FB_USER=$fbUser", "FB_PASSWORD=$fbPass", - 'FISCAL_MODE=DRY_RUN', 'FIREBIRD_WRITES=0', 'POLL_INTERVAL_SECONDS=5', 'AGENT_VERSION=1.2.4' + 'FISCAL_MODE=DRY_RUN', 'FIREBIRD_WRITES=0', 'POLL_INTERVAL_SECONDS=5', 'AGENT_VERSION=1.3.0' ) Set-Content -Path $EnvFile -Value $lines -Encoding UTF8 } diff --git a/deploy/agent-windows/pilot-construcao.env.example b/deploy/agent-windows/pilot-construcao.env.example index 9854140..55f8b42 100644 --- a/deploy/agent-windows/pilot-construcao.env.example +++ b/deploy/agent-windows/pilot-construcao.env.example @@ -19,4 +19,4 @@ FB_PASSWORD= FISCAL_MODE=DRY_RUN FIREBIRD_WRITES=0 POLL_INTERVAL_SECONDS=5 -AGENT_VERSION=1.2.4 +AGENT_VERSION=1.3.0 diff --git a/deploy/agent-windows/profiles/mmmodas.env.example b/deploy/agent-windows/profiles/mmmodas.env.example index d0c5f1a..2989cd6 100644 --- a/deploy/agent-windows/profiles/mmmodas.env.example +++ b/deploy/agent-windows/profiles/mmmodas.env.example @@ -19,4 +19,4 @@ FB_PASSWORD= FISCAL_MODE=DRY_RUN FIREBIRD_WRITES=0 POLL_INTERVAL_SECONDS=5 -AGENT_VERSION=1.2.4 +AGENT_VERSION=1.3.0 diff --git a/deploy/agent-windows/profiles/rincao.env.example b/deploy/agent-windows/profiles/rincao.env.example index b8d3e38..4309ad1 100644 --- a/deploy/agent-windows/profiles/rincao.env.example +++ b/deploy/agent-windows/profiles/rincao.env.example @@ -19,4 +19,4 @@ FB_PASSWORD= FISCAL_MODE=DRY_RUN FIREBIRD_WRITES=0 POLL_INTERVAL_SECONDS=5 -AGENT_VERSION=1.2.4 +AGENT_VERSION=1.3.0 diff --git a/deploy/agent-windows/profiles/uniao.env.example b/deploy/agent-windows/profiles/uniao.env.example index bce7734..32f99fa 100644 --- a/deploy/agent-windows/profiles/uniao.env.example +++ b/deploy/agent-windows/profiles/uniao.env.example @@ -19,4 +19,4 @@ FB_PASSWORD= FISCAL_MODE=DRY_RUN FIREBIRD_WRITES=0 POLL_INTERVAL_SECONDS=5 -AGENT_VERSION=1.2.4 +AGENT_VERSION=1.3.0 diff --git a/deploy/hub/.env.example b/deploy/hub/.env.example index 4ff0a0f..9550e08 100644 --- a/deploy/hub/.env.example +++ b/deploy/hub/.env.example @@ -1,5 +1,5 @@ # GTI Fiscal Hub — production environment (copy to .env, never commit .env) -GTI_FISCAL_VERSION=1.2.4 +GTI_FISCAL_VERSION=1.3.0 # REQUIRED. URL-safe secret (used inside a connection URL): openssl rand -hex 24 POSTGRES_PASSWORD= diff --git a/deploy/hub/docker-compose.prod.yml b/deploy/hub/docker-compose.prod.yml index c361929..bf386dc 100644 --- a/deploy/hub/docker-compose.prod.yml +++ b/deploy/hub/docker-compose.prod.yml @@ -36,7 +36,7 @@ services: build: context: ../.. dockerfile: central/Dockerfile - image: gti-fiscal-hub:${GTI_FISCAL_VERSION:-1.2.4} + image: gti-fiscal-hub:${GTI_FISCAL_VERSION:-1.3.0} restart: unless-stopped depends_on: postgres: diff --git a/docs/PRODUCT_NEW_DETECTION_DISCOVERY_V1.json b/docs/PRODUCT_NEW_DETECTION_DISCOVERY_V1.json new file mode 100644 index 0000000..21d9e62 --- /dev/null +++ b/docs/PRODUCT_NEW_DETECTION_DISCOVERY_V1.json @@ -0,0 +1,119 @@ +{ + "artifact": "PRODUCT_NEW_DETECTION_DISCOVERY_V1", + "table": "C000025", + "decision": "SELECT COUNT(*) poll (adaptive interval) + CODIGO-set diff only when the count moves (+ 60 s safety diff). No trigger/table/procedure is created in the SGV database.", + "why": "DATA_CADASTRO/ULTIMA_ALTERACAO are NOT indexed (a filter would scan the table like COUNT does); CODIGO is the PK; the ERP already owns triggers C000025_INS/UPD/DEL that must not be touched.", + "databases": { + "CONSTRUCAO": { + "rows": 3712, + "count_ms_p50": 2.2, + "code_list_ms": 62.9, + "max_codigo_ms": 5.5, + "pk": "PK_C000025(CODIGO)", + "indexed_columns": [ + "CCLASSTRIB", + "CODIGO", + "PRODUTO" + ], + "sgv_triggers_present": [ + "C000025_DEL", + "C000025_INS", + "C000025_UPD", + "CODBARRA_CODPRO" + ], + "date_columns_unindexed": [ + "ULTIMA_ALTERACAO", + "DATA_CADASTRO" + ] + }, + "MMMODAS": { + "rows": 20997, + "count_ms_p50": 9.0, + "code_list_ms": 241.1, + "max_codigo_ms": 24.3, + "pk": "PK_C000025(CODIGO)", + "indexed_columns": [ + "CCLASSTRIB", + "CODIGO", + "PRODUTO" + ], + "sgv_triggers_present": [ + "C000025_DEL", + "C000025_INS", + "C000025_UPD", + "CODBARRA_CODPRO" + ], + "date_columns_unindexed": [ + "ULTIMA_ALTERACAO", + "DATA_CADASTRO" + ] + }, + "RINCAO": { + "rows": 894, + "count_ms_p50": 1.0, + "code_list_ms": 15.6, + "max_codigo_ms": 4.0, + "pk": "PK_C000025(CODIGO)", + "indexed_columns": [ + "CCLASSTRIB", + "CODIGO", + "PRODUTO" + ], + "sgv_triggers_present": [ + "C000025_DEL", + "C000025_INS", + "C000025_UPD", + "CODBARRA_CODPRO" + ], + "date_columns_unindexed": [ + "ULTIMA_ALTERACAO", + "DATA_CADASTRO" + ] + }, + "UNIAO": { + "rows": 8942, + "count_ms_p50": 4.3, + "code_list_ms": 119.2, + "max_codigo_ms": 11.4, + "pk": "PK_C000025(CODIGO)", + "indexed_columns": [ + "CCLASSTRIB", + "CODBARRA", + "CODIGO", + "PRODUTO" + ], + "sgv_triggers_present": [ + "C000025_DEL", + "C000025_INS", + "C000025_UPD", + "CODBARRA_CODPRO" + ], + "date_columns_unindexed": [ + "ULTIMA_ALTERACAO", + "DATA_CADASTRO" + ] + }, + "TESTESOFICIAIS": { + "rows": 521, + "count_ms_p50": 0.9, + "code_list_ms": 8.0, + "max_codigo_ms": 2.2, + "pk": "PK_C000025(CODIGO)", + "indexed_columns": [ + "CCLASSTRIB", + "CODIGO", + "PRODUTO" + ], + "sgv_triggers_present": [ + "C000025_DEL", + "C000025_INS", + "C000025_UPD", + "CODBARRA_CODPRO" + ], + "date_columns_unindexed": [ + "ULTIMA_ALTERACAO", + "DATA_CADASTRO" + ] + } + } +} \ No newline at end of file diff --git a/docs/PRODUCT_PROVIDER_BENCHMARK_V1.json b/docs/PRODUCT_PROVIDER_BENCHMARK_V1.json new file mode 100644 index 0000000..b41c995 --- /dev/null +++ b/docs/PRODUCT_PROVIDER_BENCHMARK_V1.json @@ -0,0 +1,96 @@ +{ + "measured_at": "2026-09-25T21:53:01Z", + "sample_source": "lab ERP databases (read-only), valid GTINs only", + "limit_per_provider": 30, + "providers": [ + { + "provider": "cosmos", + "configured": false, + "sample_size": 0, + "found": 0, + "not_found": 0, + "with_description": 0, + "with_brand": 0, + "with_ncm": 0, + "with_cest": 0, + "errors": 0, + "error_kinds": {}, + "latencies_ms": [], + "status": "NOT_CONFIGURED" + }, + { + "provider": "upcitemdb", + "configured": true, + "sample_size": 6, + "found": 1, + "not_found": 5, + "with_description": 1, + "with_brand": 1, + "with_ncm": 0, + "with_cest": 0, + "errors": 1, + "error_kinds": { + "RATE_LIMITED": 1 + }, + "error_rate": 0.143, + "latency_p50_ms": 139.5, + "latency_p95_ms": 140, + "found_rate": 0.167, + "status": "MEASURED" + }, + { + "provider": "openfoodfacts", + "configured": true, + "sample_size": 30, + "found": 4, + "not_found": 26, + "with_description": 4, + "with_brand": 3, + "with_ncm": 0, + "with_cest": 0, + "errors": 0, + "error_kinds": {}, + "error_rate": 0.0, + "latency_p50_ms": 655.0, + "latency_p95_ms": 827, + "found_rate": 0.133, + "status": "MEASURED" + }, + { + "provider": "openproductsfacts", + "configured": true, + "sample_size": 30, + "found": 0, + "not_found": 30, + "with_description": 0, + "with_brand": 0, + "with_ncm": 0, + "with_cest": 0, + "errors": 0, + "error_kinds": {}, + "error_rate": 0.0, + "latency_p50_ms": 640.0, + "latency_p95_ms": 969, + "found_rate": 0.0, + "status": "MEASURED" + }, + { + "provider": "openbeautyfacts", + "configured": true, + "sample_size": 30, + "found": 0, + "not_found": 30, + "with_description": 0, + "with_brand": 0, + "with_ncm": 0, + "with_cest": 0, + "errors": 0, + "error_kinds": {}, + "error_rate": 0.0, + "latency_p50_ms": 655.5, + "latency_p95_ms": 1311, + "found_rate": 0.0, + "status": "MEASURED" + } + ] +} \ No newline at end of file diff --git a/docs/product_enrichment.md b/docs/product_enrichment.md new file mode 100644 index 0000000..2f81fc6 --- /dev/null +++ b/docs/product_enrichment.md @@ -0,0 +1,67 @@ +# GTI Fiscal — Product Enrichment V1 (produto novo → EAN/nome → perfil fiscal → preenchimento seguro) + +Living document. Same Hub / FastAPI / PostgreSQL / Agent as the rest of GTI Fiscal: **no new server, service, panel or product**, no change in the SGV, no Firebird trigger/table/procedure. + +## Hard rules (verified by tests) +- `AI_ENABLED=false`, `LLM_ENABLED=false`, `EMBEDDINGS_ENABLED=false`. Everything here is deterministic. +- External cost R$ 0: only free providers; `ALLOW_PAID_PROVIDERS=false` (a paid provider makes `build_providers` raise). **Agents never call providers – only the Hub.** +- Per-installation functions live in the Hub (Agent detail → *Funcionalidades*): **Corretor fiscal** on/off and **Enriquecimento** `DISABLED | DRY_RUN | ENABLED`. ADMIN only, CSRF, audited (`INSTALLATION_FEATURES`), + typed confirmation `PREENCHER` for ENABLED. A **new** installation is `WAITING_FEATURE_CONFIGURATION` (`features_configured=false`): it heartbeats but `claim_next_job`/`schedule_automatic_sync` + hand it nothing and the enrichment worker does zero Firebird work, until features are chosen (at creation in *Nova instalação*, or later). Existing installations keep the previous behaviour (migration 0029). +- Scope = **new products only**. When the function is first active the Agent stores a baseline of the existing `C000025.CODIGO` set (local SQLite); historical products are never sent anywhere. + +## Flow +1. **Detection (Agent, ≤ 5 s)** – `SELECT COUNT(*) FROM C000025` every 2 s (adaptive up to 30 s if the count is slow); code-set diff only when the count moved (+ a 60 s safety diff). No trigger. Discovery on the five + lab databases (`docs/PRODUCT_NEW_DETECTION_DISCOVERY_V1.json`): PK `CODIGO` char(6), `DATA_CADASTRO/ULTIMA_ALTERACAO` **not indexed**, count 1–9 ms, full code list 8–241 ms, the ERP already owns triggers on the table. + The EAN typed by the operator is in **`CODBARRA`** (fiscal `GTIN` is mostly empty): the first VALID of `GTIN`/`CODBARRA` is sent. +2. **Resolve (Hub)** `POST /v1/products/resolve {product_code, gtin, description, current}` → per-field decisions. Hub time budget 4.5 s; beyond it (or on quota/offline) the state is `PENDING_*` and the Agent retries with backoff (30 s … 30 min); results are cached, so the retry is usually free. +3. **Write (Agent)** typed capability `PRODUCT_ENRICHMENT_WRITE_V1`, allowlist `CLASSIFICACAO_FISCAL, CEST, IPI`; fixed statement `UPDATE C000025 SET f=? WHERE CODIGO=? AND (f IS NULL OR TRIM(f)='')`, one short transaction per field, + guarded re-select inside the transaction, post-write verification. Before every field: fresh central authorization (fail closed). `COD_ORIGEM_MERCADORIA` is never written; cClassTrib stays with the existing RTC gates + (the Agent only requests `REPROCESS_PRODUCTS` for the product; nothing is decided here). +4. **Audit** `POST /v1/products/enrichment/applied` → `product_field_audits` (tenant, agent, installation, code, GTIN, description snapshot, field, old/new, rule, KB status, evidence, dataset versions, authorization origin, job, result). No secrets. + +## Identity and states +- GTIN: 8/12/13/14 digits + check digit → `VALID_GTIN | INVALID_GTIN | MISSING_GTIN` (+ `RESTRICTED_GTIN` for store-internal prefixes 02/04/20–29). **GTIN has absolute priority over the description.** An invalid GTIN never reaches a provider. +- Resolution states (discrete, no scores): `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, FEATURE_DISABLED`). +- Field decisions: `AUTO_FILL_SAFE | SUGGESTION_ONLY | CONFLICT | INSUFFICIENT_EVIDENCE | ALREADY_SET | NOT_APPLICABLE | HUMAN_OVERRIDDEN`. A field that is not resolved never blocks the others. + +## Knowledge base (same PostgreSQL: `kb_products`, `kb_knowledge` with per-field `status`) +`UNVERIFIED / EVIDENCE_ONLY / VERIFIED_KB / CONFLICT / STALE`. NCM becomes `VERIFIED_KB` only when it is **valid and vigente in the official NCM dataset**, has **≥ 2 independent evidence families** (each provider = 1 family; client observations = 1 family and only when ≥ 2 distinct clients) +and **no conflicting value**. Client consensus alone is never authority (measured on the lab databases: 241 GTINs shared by ≥ 2 clients, **93 with different NCMs**). Company-specific decisions (CSOSN, CFOP, …) never enter the KB. +**Human edits never train the KB.** `NCM` auto-fill: only `VERIFIED_KB`; a single non-official source is `SUGGESTION_ONLY`; conflicts write nothing. +CEST and IPI are deterministic functions of an official NCM through the existing engine (`Classifier`): they are filled when the NCM is being written (AUTO) **or is already present and official**, only for a unique official CEST mapping / a numeric TIPI rate. +Name-only: only an exact fingerprint (sorted normalized tokens, quantity/unit normalized) of a **specific** description (≥ 2 informative tokens + quantity, or ≥ 4) that maps to a unique `VERIFIED_KB` product may fill; generic names never do. +ERP observations (`scripts/enrichment_lab_kb.py`, read-only) enter as `CLIENT_OBSERVATION` evidence only. + +## Human override protection +After writing, the Agent watches the value. Any value different from what it wrote — **including a deleted value** — marks the field `HUMAN_OVERRIDDEN` (local SQLite + reported to the Hub). The Hub then answers `HUMAN_OVERRIDDEN` for that field, the enrichment never rewrites it, +and `enrichment.is_human_overridden(state_path, code, field)` is the hook every automatic writer must call. No write loops: second run = 0 writes. + +## Providers (Hub only; `central/app/enrichment/providers.py`, single interface `lookup_gtin / search_description / health`) +| Provider | Access | Limit used | Gives | +|---|---|---|---| +| Cosmos (Bluesoft) | `COSMOS_TOKEN` in the Hub env (optional) | 25/day | description, brand, **NCM, CEST** — `NOT_CONFIGURED` without a token (not an error) | +| UPCitemdb (trial) | no key | 100/day, burst 15 / 30 s (headers `X-RateLimit-*`) | title, brand, category | +| Open Food Facts / Open Products Facts / Open Beauty Facts | open (ODbL), identifiable `User-Agent` | 15 req/min | title, brand, quantity | +Quota is central (`ProviderState`, `ProviderUsageDaily`, `quota_kind = REMOTE_QUOTA | LOCAL_ESTIMATE | DOCUMENTED_RATE`), with positive (30 d) and negative (24 h) `GtinLookupCache`, single-flight per (provider, GTIN), circuit breaker +(3 failures → 60 s doubling up to 15 min; 429 → until reset; 401/403 → 1 h). Quota exhausted → `PENDING_PROVIDER_QUOTA` + retry after reset. +**Benchmark on real lab GTINs** (`docs/PRODUCT_PROVIDER_BENCHMARK_V1.json`): identity found rate is low for Brazilian goods (Open Food Facts 4/30, UPCitemdb rate-limited after 6 calls) and **only Cosmos returns NCM**. Consequently +NCM auto-fill depends on the KB (Cosmos + client observations) and the product delivers CEST/IPI immediately whenever the operator already typed a valid NCM. Cosmos without a token is not a blocker. + +## Official data +`NCM` (Siscomex JSON) is refreshed daily by `central/app/datasets/updater.py` (staging in memory → schema validation → plausibility ≥ 90 % of the active count → atomic promotion by `DatasetService.import_payload`; on any failure the last valid dataset stays). +Every attempt is a `source_refresh_runs` row (url, time, sha256, version, count, status, error). TIPI/CEST/CFOP/CST/CSOSN/PIS-COFINS/RTC datasets are reused as they are. The dashboard shows *Fontes oficiais* (records, version, sha, last update, status). + +## Panel +Dashboard: *Fontes de produtos* (status, used/limit/remaining/reset, last success/error, latency, cache-hit rate, `configurado: sim/não` — never a token), *Fontes oficiais*, small *Enriquecimento* coverage. +Agent detail: *Funcionalidades* (admin) and *Enriquecimento* (counters + product lookup with per-field provenance). Problems appear through the existing blockers (`field=ENRICHMENT`, PT-BR reasons). + +## Coverage metrics (`observability.coverage_metrics`) +`GTIN_VALID_RATE, KB_HIT_RATE, EXTERNAL_LOOKUP_RATE, PRODUCT_FOUND_RATE, NCM_AUTO_FILL_SAFE_RATE, NAME_ONLY_RESOLUTION_RATE, AMBIGUOUS_RATE, CONFLICT_RATE, PROVIDER_QUOTA_BLOCK_RATE` (+ cache stats). + +## Evidence +- Offline tests: `central/tests/test_product_enrichment_resolver.py` (12 resolution scenarios, quota/circuit/cache, provider contract fixtures), `test_product_enrichment_api.py` (feature gates, audit, override, blockers, multi-installation), `test_official_refresh.py`, `agent/tests/test_enrichment.py`. +- Real Firebird **copy-only** E2E `scripts/enrichment_copy_e2e.py` → `docs/runs/enrichment_copy_e2e.json`: detect→written P50 2.8 s / P95 3.9 s / max 4.1 s; only the 3 allowed columns; other tables changed only by the SGV's own triggers for the test rows; + second run 0 writes; DRY_RUN 0 writes; WAITING/DISABLED 0 Firebird activity and no baseline; human override incl. deleted value; Hub offline → pending → recovers. +- Migration 0029: fresh-vs-upgrade `pg_dump -s` identical; upgrade → downgrade → upgrade clean. +- Live smoke `scripts/enrichment_live_smoke.py` (1 lookup/provider, real internet, separate from the suite). diff --git a/docs/runs/enrichment_copy_e2e.json b/docs/runs/enrichment_copy_e2e.json new file mode 100644 index 0000000..24d9c22 --- /dev/null +++ b/docs/runs/enrichment_copy_e2e.json @@ -0,0 +1,47 @@ +{ + "copy_only": true, + "products_enabled": 12, + "detect_to_written_seconds": { + "p50": 2.835, + "p95": 3.89, + "max": 4.14, + "all": [ + 1.66, + 2.08, + 3.89, + 2.83, + 2.52, + 2.84, + 3.05, + 2.58, + 3.31, + 2.45, + 3.89, + 4.14 + ] + }, + "hub_resolve_calls": 15, + "tables_changed": [ + "C000025", + "C000058", + "C000100" + ], + "checks": { + "WAITING_NO_BASELINE_NO_CALLS": true, + "DISABLED_NO_BASELINE_NO_CALLS": true, + "BASELINE_TAKEN_ON_ACTIVATION": true, + "DRY_RUN_ZERO_WRITES": true, + "FIELDS_WRITTEN": true, + "SECOND_RUN_ZERO_WRITES": true, + "HISTORICAL_KNOWN_COUNT": true, + "OVERRIDE_DELETED_VALUE": true, + "OVERRIDE_REPORTED_TO_HUB": true, + "OVERRIDE_NEVER_REWRITTEN": true, + "HUB_OFFLINE_PENDING_NO_WRITE": true, + "HUB_BACK_RECOVERS": true, + "OTHER_TABLES_ONLY_ERP_TRIGGER_SIDE_EFFECTS_OF_TEST_ROWS": true, + "ORIGIN_NEVER_WRITTEN": true + }, + "ALL_PASS": true, + "poll_seconds": 2.0 +} \ No newline at end of file diff --git a/docs/runs/enrichment_local_hub_e2e.json b/docs/runs/enrichment_local_hub_e2e.json new file mode 100644 index 0000000..82f5882 --- /dev/null +++ b/docs/runs/enrichment_local_hub_e2e.json @@ -0,0 +1,21 @@ +{ + "copy_only": true, + "hub": "http://127.0.0.1:8000", + "detect_to_written_seconds_enabled": 4.22, + "checks": { + "REGISTERED_DEFAULT_DISABLED": true, + "DISABLED_NO_ACTIVITY": true, + "BASELINE_ON_ACTIVATION": true, + "DRY_RUN_ZERO_WRITES": true, + "DRY_RUN_AUDITED_IN_HUB": true, + "ENABLED_NCM_WRITTEN": true, + "WRITTEN_FIELDS_MATCH_HUB_AUDIT": true, + "EARLIER_DRY_RUN_PRODUCT_STILL_UNTOUCHED": true, + "UNKNOWN_GTIN_NO_WRITE": true, + "UNKNOWN_GTIN_BLOCKER_PT": true, + "INVALID_GTIN_NO_PROVIDER_CALL": true, + "DISABLED_AGAIN_ZERO_WRITES": true, + "CORRECTOR_OFF_NO_JOBS": true + }, + "ALL_PASS": true +} \ No newline at end of file diff --git a/release/RELEASE_MANIFEST.json b/release/RELEASE_MANIFEST.json index 4f9b5b6..6a34e7e 100644 --- a/release/RELEASE_MANIFEST.json +++ b/release/RELEASE_MANIFEST.json @@ -1,14 +1,14 @@ { "artifact": "RELEASE_MANIFEST", - "GTI_FISCAL_VERSION": "1.2.5", + "GTI_FISCAL_VERSION": "1.3.0", "generated_at": "2026-09-25T15:43:40+00:00", "hub": { - "version": "1.2.5", + "version": "1.3.0", "api_compat": "1.0", "engine_version": "0.3.0", "decision_version": "mission-04c-ipi-ex-split" }, - "agent_version": "1.2.5", + "agent_version": "1.3.0", "fiscal_engine": { "ENGINE_VERSION": "0.3.0", "DECISION_VERSION": "mission-04c-ipi-ex-split", @@ -17,10 +17,10 @@ "ICMS_CONTEXT_ENGINE_VERSION": "icms-context-v1" }, "migrations": { - "alembic_head_in_repo": "0028_rtc_write_modes", - "applied_revision": "0028_rtc_write_modes", - "applied_revision_source": "production Hub 10.0.10.117 (gti-fiscal-hub-fiscal-hub-1), verified 2026-09-25", - "count": 28 + "alembic_head_in_repo": "0029_product_enrichment", + "applied_revision": "0029_product_enrichment", + "applied_revision_source": "fresh + upgrade(0028->0029) + downgrade/upgrade verified on the dev PostgreSQL (schema parity by pg_dump -s); production applies 0029 at deploy (alembic upgrade head at container start)", + "count": 29 }, "datasets_active": [ { @@ -90,4 +90,4 @@ } ], "note": "A release bump does not change fiscal semantics; engine/decision versions are versioned independently." -} +} \ No newline at end of file diff --git a/release/VERSION b/release/VERSION index c813fe1..f0bb29e 100644 --- a/release/VERSION +++ b/release/VERSION @@ -1 +1 @@ -1.2.5 +1.3.0 diff --git a/scripts/enrichment_benchmark.py b/scripts/enrichment_benchmark.py new file mode 100644 index 0000000..bdeb484 --- /dev/null +++ b/scripts/enrichment_benchmark.py @@ -0,0 +1,79 @@ +"""Conservative provider benchmark on REAL GTINs taken (read-only) from the lab ERP databases. Free quotas only, paced below documented limits. + + python scripts/enrichment_benchmark.py --sample scratch_enrich/benchmark_gtins.json --out docs/PRODUCT_PROVIDER_BENCHMARK_V1.json [--limit 30] + +Each provider gets at most --limit lookups (UPCitemdb: also <= 12 per 30 s burst window; Open Food Facts: <= 12/min). Cosmos is measured only when COSMOS_TOKEN is set. +Reports sample_size / found / not_found / description / brand / NCM / CEST / latency / error_rate per provider. Secrets are never printed. +""" +import argparse +import json +import os +import statistics +import sys +import time +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) +import httpx # noqa: E402 + +from central.app.enrichment.providers import Cosmos, OpenBeautyFacts, OpenFoodFacts, OpenProductsFacts, ProviderError, UpcItemDb, USER_AGENT # noqa: E402 + +PACE = {"upcitemdb": 2.6, "openfoodfacts": 5.2, "openproductsfacts": 5.2, "openbeautyfacts": 5.2, "cosmos": 1.0} # seconds between calls: comfortably under the documented burst / per-minute limits + + +def bench(provider, items): + out = {"provider": provider.name, "configured": provider.configured, "sample_size": 0, "found": 0, "not_found": 0, "with_description": 0, "with_brand": 0, "with_ncm": 0, "with_cest": 0, "errors": 0, "error_kinds": {}, "latencies_ms": []} + if not provider.configured: + out["status"] = "NOT_CONFIGURED" + return out + for item in items: + started = time.monotonic() + try: + product = provider.lookup_gtin(item["gtin"]) + except ProviderError as error: + out["errors"] += 1 + out["error_kinds"][error.kind] = out["error_kinds"].get(error.kind, 0) + 1 + if error.kind in ("RATE_LIMITED", "AUTH_ERROR"): + break # never insist against a limit + time.sleep(PACE[provider.name]) + continue + out["sample_size"] += 1 + out["latencies_ms"].append(int((time.monotonic() - started) * 1000)) + if product.found: + out["found"] += 1 + out["with_description"] += bool(product.title) + out["with_brand"] += bool(product.brand) + out["with_ncm"] += bool(product.ncm) + out["with_cest"] += bool(product.cest) + else: + out["not_found"] += 1 + time.sleep(PACE[provider.name]) + n = out["sample_size"] + out["errors"] + lat = sorted(out.pop("latencies_ms")) + out.update({"error_rate": round(out["errors"] / n, 3) if n else None, "latency_p50_ms": statistics.median(lat) if lat else None, "latency_p95_ms": lat[int(0.95 * (len(lat) - 1))] if lat else None, + "found_rate": round(out["found"] / out["sample_size"], 3) if out["sample_size"] else None, "status": "MEASURED"}) + return out + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--sample", required=True) + ap.add_argument("--out", required=True) + ap.add_argument("--limit", type=int, default=30) + ap.add_argument("--only", default="", help="comma list of provider names; results are MERGED into an existing --out") + args = ap.parse_args() + items = json.loads(Path(args.sample).read_text(encoding="utf-8"))[: args.limit] + client = httpx.Client(timeout=8.0, headers={"User-Agent": USER_AGENT}) + providers = [Cosmos(client, os.environ.get("COSMOS_TOKEN")), UpcItemDb(client), OpenFoodFacts(client), OpenProductsFacts(client), OpenBeautyFacts(client)] + only = {n for n in args.only.split(",") if n} + out = Path(args.out) + result = json.loads(out.read_text(encoding="utf-8")) if only and out.exists() else {"measured_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "sample_source": "lab ERP databases (read-only), valid GTINs only", "limit_per_provider": args.limit, "providers": []} + measured = [bench(p, items) for p in providers if not only or p.name in only] + result["providers"] = [p for p in result["providers"] if p["provider"] not in {m["provider"] for m in measured}] + measured + Path(args.out).write_text(json.dumps(result, ensure_ascii=False, indent=1), encoding="utf-8") + print(json.dumps(result, ensure_ascii=False, indent=1)) + + +if __name__ == "__main__": + main() diff --git a/scripts/enrichment_copy_e2e.py b/scripts/enrichment_copy_e2e.py new file mode 100644 index 0000000..f0da06f --- /dev/null +++ b/scripts/enrichment_copy_e2e.py @@ -0,0 +1,237 @@ +"""COPY-ONLY E2E of the new-product enrichment on a REAL Firebird (disposable copy restored from its own .fbk). + + python scripts/enrichment_copy_e2e.py [--products 12] + +The Agent code under test is the real one (gti_fiscal_agent.enrichment: real fdb connections, real typed UPDATE, real threads/timing). The Hub side is the real resolver run in-process on +SQLite with the official NCM dataset and fake providers (no internet). Proves on a real Firebird: + detection latency P50/P95 (product INSERTed by an "operator" -> fields written), only-empty fields written, other columns + every other table untouched, historical rows untouched, + second run = 0 writes, DRY_RUN = 0 writes, DISABLED / WAITING = 0 Firebird activity, human override incl. deleted value, Hub offline -> pending -> recovers, authorization revoked -> 0 writes. +""" +import json +import os +import statistics +import sys +import threading +import time +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)); sys.path.insert(0, str(ROOT / "scripts")); sys.path.insert(0, str(ROOT / "agent" / "src")) +from dotenv import load_dotenv # noqa: E402 + +load_dotenv(ROOT / ".env") +import rtc_copy_write_e2e as base # noqa: E402 +from gti_fiscal_agent import enrichment as E # noqa: E402 +from sqlalchemy import create_engine # noqa: E402 +from sqlalchemy.orm import Session # noqa: E402 +from sqlalchemy.pool import StaticPool # noqa: E402 + +from central.app.datasets.ncm import parse_ncm_payload # noqa: E402 +from central.app.datasets.service import DatasetService # noqa: E402 +from central.app.enrichment import kb # noqa: E402 +from central.app.enrichment.providers import Provider, ProviderProduct # noqa: E402 +from central.app.enrichment.quota import ProviderGateway # noqa: E402 +from central.app.enrichment.resolver import resolve # noqa: E402 +from central.app.models import Base, CestEntry, CestNcmEntry, Dataset, TipiEntry # noqa: E402 +from datetime import datetime, timezone # noqa: E402 +import httpx # noqa: E402 + + +def gtin(body12): + 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) + + +class Fake(Provider): + def __init__(self, name, ncm): + super().__init__(client=httpx.Client()) + self.name, self.ncm, self.provides_ncm = name, ncm, True + + def lookup_gtin(self, g): + return ProviderProduct(self.name, g, True, title="PRODUTO DE TESTE ESPECIFICO 350ML", ncm=self.ncm) + + +class InProcessHub: + """The real Hub resolver, in-process. `online=False` simulates the Hub being unreachable.""" + + def __init__(self): + engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool) + Base.metadata.create_all(engine) + self.session = Session(engine) + DatasetService(self.session).import_payload(parse_ncm_payload((ROOT / "scratch_enrich" / "ncm_siscomex.json").read_bytes()), "official", datetime.now(timezone.utc)) + 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) + self.session.add_all([tipi, cest]); self.session.flush() + self.session.add(TipiEntry(dataset_id=tipi.id, ncm="22021000", ex=None, description="x", ipi_rate=0.0, ipi_rate_text="0", source_row=1)) + e = CestEntry(dataset_id=cest.id, code="03.007.00", description="PRODUTO DE TESTE ESPECIFICO 350ML", annex="I") + self.session.add(e); self.session.flush(); self.session.add(CestNcmEntry(cest_entry_id=e.id, ncm_code="22021000")); self.session.commit() + self.providers = [Fake("cosmos", "22021000"), Fake("second", "22021000")] + self.features = {"features_configured": True, "corrector_enabled": True, "enrichment_mode": "ENABLED"} + self.online, self.reports, self.overridden, self.calls = True, [], {}, 0 + self.lock = threading.Lock() + + def authorizations(self): + if not self.online: + raise ConnectionError("hub offline") + return {"features": dict(self.features)} + + def resolve_product(self, payload): + if not self.online: + raise ConnectionError("hub offline") + with self.lock: + self.calls += 1 + r = resolve(self.session, ProviderGateway(self.session, self.providers), gtin_raw=payload["gtin"], description=payload["description"], current=payload["current"]) + for f in self.overridden.get(payload["product_code"], ()): + r["fields"][f] = {"status": "HUMAN_OVERRIDDEN", "value": None} + r["retry_at"] = None + return r + + def report_enrichment(self, payload): + with self.lock: + self.reports.append(payload) + for it in payload["fields"]: + if it["outcome"] == "HUMAN_OVERRIDDEN": + self.overridden.setdefault(payload["product_code"], set()).add(it["field"]) + return {"ok": True} + + +_orig_connect = base.connect + + +def _connect_retry(rw): + """Same brief retry the Agent applies: local XNET connections can report 'connection lost' when several threads connect at the same instant.""" + for attempt in range(6): + try: + return _orig_connect(rw) + except Exception: # noqa: BLE001 + if attempt == 5: + raise + time.sleep(0.2 * (attempt + 1)) + + +base.connect = _connect_retry + + +def insert_product(code, name, ean): + conn = base.connect(True); cur = conn.cursor() + cur.execute("INSERT INTO C000025 (CODIGO, PRODUTO, CODBARRA) VALUES (?, ?, ?)", (code, name, ean)) + conn.commit(); conn.close() + + +def row(code): + conn = base.connect(False); cur = conn.cursor() + cur.execute("SELECT CLASSIFICACAO_FISCAL, CEST, IPI FROM C000025 WHERE CODIGO=?", (code,)) + r = cur.fetchone(); conn.close() + return tuple(None if v is None else (str(v).strip() if not isinstance(v, float) else v) for v in r) if r else None + + +def side_hashes(): + """C000058 (ERP change journal) and C000100 (stock) are maintained by the SGV's OWN triggers on C000025 INSERT/UPDATE. Only rows of the test products (code 'Z...') may differ.""" + import hashlib + conn = base.connect(False); cur = conn.cursor(); out = {} + for table, col in (("C000058", "CODREGISTRO"), ("C000100", "CODPRODUTO")): + cur.execute(f"SELECT * FROM {table} WHERE {col} NOT STARTING WITH 'Z'") + h = hashlib.sha256() + n = 0 + for r in cur.fetchall(): + h.update(repr(r).encode("utf-8", "replace")); n += 1 + out[table] = f"{n}:{h.hexdigest()[:16]}" + conn.close() + return out + + +def main(): + n = int(sys.argv[sys.argv.index("--products") + 1]) if "--products" in sys.argv else 12 + checks = {} + base.restore_copy() + before_tables, _ = base.baseline() + before_side = side_hashes() + hub = InProcessHub() + for suffix in ("", "-wal", "-shm"): + try: + (ROOT / "scratch_enrich" / f"enrich_e2e.db{suffix}").unlink() + except FileNotFoundError: + pass + store = E.EnrichmentStore(ROOT / "scratch_enrich" / "enrich_e2e.db") + worker = E.EnrichmentWorker(hub=hub, store=store, read_conn=lambda: base.connect(False), write_conn=lambda: base.connect(True), + create_reprocess=lambda codes: None) + + class Stop: + stop = False + stopper = Stop() + thread = threading.Thread(target=worker.run_forever, args=(stopper,), daemon=True) + + # --- WAITING / DISABLED: zero Firebird activity, baseline NOT taken + hub.features = {"features_configured": False} + thread.start(); time.sleep(4) + checks["WAITING_NO_BASELINE_NO_CALLS"] = store.meta("baseline_done") is None and hub.calls == 0 + hub.features = {"features_configured": True, "corrector_enabled": True, "enrichment_mode": "DISABLED"}; worker._features_at = -999; time.sleep(4) + checks["DISABLED_NO_BASELINE_NO_CALLS"] = store.meta("baseline_done") is None and hub.calls == 0 + + # --- DRY_RUN: baseline, detect, resolve, report, zero writes + hub.features = {"features_configured": True, "corrector_enabled": True, "enrichment_mode": "DRY_RUN"}; worker._features_at = -999; time.sleep(5) + checks["BASELINE_TAKEN_ON_ACTIVATION"] = store.meta("baseline_done") == "1" + hist = len(store.known()) + insert_product("ZD001", "PRODUTO DE TESTE ESPECIFICO 350ML", gtin("789200000001")); time.sleep(8) + checks["DRY_RUN_ZERO_WRITES"] = row("ZD001") == (None, None, None) and hub.reports and hub.reports[-1]["fields"][0]["outcome"] == "DRY_RUN" + + # --- ENABLED: N products inserted like an operator would, latency measured insert -> written + hub.features = {"features_configured": True, "corrector_enabled": True, "enrichment_mode": "ENABLED"}; worker._features_at = -999; worker._auth_cache = (None, 0.0) + latencies = [] + for i in range(n): + code, ean = f"ZE{i:03d}", gtin(f"78930000{i:04d}") + t0 = time.monotonic() + insert_product(code, "PRODUTO DE TESTE ESPECIFICO 350ML", ean) + while time.monotonic() - t0 < 20 and row(code) == (None, None, None): + time.sleep(0.1) + latencies.append(round(time.monotonic() - t0, 2)) + time.sleep(0.5) + ok_rows = [row(f"ZE{i:03d}") for i in range(n)] + checks["FIELDS_WRITTEN"] = all(r and r[0] == "22021000" and r[1] == "0300700" and r[2] == 0.0 for r in ok_rows) + time.sleep(6) + writes_before = sum(1 for rep in hub.reports for f in rep["fields"] if f["outcome"] == "APPLIED") + calls_before = hub.calls + time.sleep(8) # a "second run": nothing pending, nothing re-written + checks["SECOND_RUN_ZERO_WRITES"] = writes_before == sum(1 for rep in hub.reports for f in rep["fields"] if f["outcome"] == "APPLIED") and hub.calls - calls_before <= 0 + checks["HISTORICAL_KNOWN_COUNT"] = hist > 100 + + # --- human override incl. deleted value + conn = base.connect(True); cur = conn.cursor() + cur.execute("UPDATE C000025 SET CEST = NULL WHERE CODIGO = 'ZE000'"); cur.execute("UPDATE C000025 SET CLASSIFICACAO_FISCAL = '19059090' WHERE CODIGO = 'ZE001'"); conn.commit(); conn.close() + time.sleep(E.OVERRIDE_CHECK_SECONDS + 6) + checks["OVERRIDE_DELETED_VALUE"] = store.is_overridden("ZE000", "CEST") and store.is_overridden("ZE001", "CLASSIFICACAO_FISCAL") + checks["OVERRIDE_REPORTED_TO_HUB"] = "CEST" in hub.overridden.get("ZE000", set()) + # ... and a later description edit does NOT bring the value back + conn = base.connect(True); cur = conn.cursor(); cur.execute("UPDATE C000025 SET PRODUTO = 'PRODUTO DE TESTE ESPECIFICO 350ML ZERO' WHERE CODIGO = 'ZE000'"); conn.commit(); conn.close() + time.sleep(10) + checks["OVERRIDE_NEVER_REWRITTEN"] = row("ZE000")[1] in (None, "") and row("ZE001")[0] == "19059090" + + # --- Hub offline -> pending -> recovers ; authorization revoked -> 0 writes + hub.online = False + insert_product("ZO001", "PRODUTO DE TESTE ESPECIFICO 350ML", gtin("789400000001")); time.sleep(8) + checks["HUB_OFFLINE_PENDING_NO_WRITE"] = row("ZO001") == (None, None, None) and store.get("ZO001")["state"] in ("RETRY", "PENDING") + hub.online = True + with store.db: + store.db.execute("UPDATE enrich_products SET next_attempt_at=0 WHERE code='ZO001'") + time.sleep(8) + checks["HUB_BACK_RECOVERS"] = row("ZO001") == ("22021000", "0300700", 0.0) + stopper.stop = True + thread.join(timeout=10) + + # --- invariants on the real Firebird copy: only the 3 allowed columns of the new rows changed + after_tables, _ = base.baseline() + changed = base.diff_tables(before_tables, after_tables) + checks["OTHER_TABLES_ONLY_ERP_TRIGGER_SIDE_EFFECTS_OF_TEST_ROWS"] = set(changed) <= {"C000025", "C000058", "C000100"} and side_hashes() == before_side + conn = base.connect(False); cur = conn.cursor() + cur.execute("SELECT COUNT(*) FROM C000025 WHERE CODIGO STARTING WITH 'Z' AND COD_ORIGEM_MERCADORIA IS NOT NULL AND TRIM(COD_ORIGEM_MERCADORIA) <> ''"); origin = cur.fetchone()[0]; conn.close() + checks["ORIGIN_NEVER_WRITTEN"] = origin == 0 + result = {"copy_only": True, "products_enabled": n, "detect_to_written_seconds": {"p50": statistics.median(latencies), "p95": sorted(latencies)[int(0.95 * (len(latencies) - 1))], "max": max(latencies), "all": latencies}, + "hub_resolve_calls": hub.calls, "tables_changed": changed, "checks": checks, "ALL_PASS": all(checks.values()), "poll_seconds": worker.poll_seconds()} + Path(ROOT / "docs" / "runs").mkdir(parents=True, exist_ok=True) + (ROOT / "docs" / "runs" / "enrichment_copy_e2e.json").write_text(json.dumps(result, indent=1), encoding="utf-8") + print(json.dumps(result, indent=1)) + sys.exit(0 if result["ALL_PASS"] else 1) + + +if __name__ == "__main__": + main() diff --git a/scripts/enrichment_lab_kb.py b/scripts/enrichment_lab_kb.py new file mode 100644 index 0000000..3809b53 --- /dev/null +++ b/scripts/enrichment_lab_kb.py @@ -0,0 +1,103 @@ +"""READ-ONLY study of the five lab ERP databases for the product enrichment: GTIN quality, NCM coverage, cross-database agreement, and KB seeding. + + python scripts/enrichment_lab_kb.py [--seed-url sqlite:///kb.db] [--sample 40 --sample-out scratch/benchmark_gtins.json] + +Firebird is opened READ-ONLY (no write path exists here). The KB seed goes ONLY to the database given by --seed-url (default: an in-memory SQLite, i.e. statistics only) and stores +CLIENT_OBSERVATION evidence, never VERIFIED_KB by itself. Company-specific fiscal fields are not read at all. +""" +import argparse +import json +import os +import sys +from collections import Counter, defaultdict +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) +sys.path.insert(0, str(ROOT / "scripts")) +from dotenv import load_dotenv # noqa: E402 + +load_dotenv(ROOT / ".env") +import local_completion as lc # noqa: E402 +from sqlalchemy import create_engine, select # noqa: E402 +from sqlalchemy.orm import Session # noqa: E402 + +from central.app.enrichment import kb # noqa: E402 +from central.app.enrichment.gtin import normalize_gtin # noqa: E402 +from central.app.enrichment.observations import observe # noqa: E402 +from central.app.enrichment.resolver import ncm_is_official # noqa: E402 +from central.app.models import Base, Dataset, KbKnowledge, NcmEntry # noqa: E402 + + +def read_lab(name, path): + conn = lc.connect(path) + try: + cur = conn.cursor() + cur.execute("SELECT CODIGO, PRODUTO, GTIN, CLASSIFICACAO_FISCAL, CODBARRA FROM C000025") + # the SGV operator types the EAN in CODBARRA; GTIN (fiscal) is mostly empty. The first VALID one wins (same rule as the Agent). + return [{"CODIGO": r[0], "PRODUTO": (r[1] or "").strip(), "GTIN": next((x for x in (r[2], r[4]) if normalize_gtin(x)["status"] == "VALID_GTIN"), r[2] or r[4]), "NCM": r[3]} for r in cur.fetchall()] + finally: + conn.close() + + +def load_ncm_official(session, ncm_json: Path | None): + """Official NCM into the throw-away/seed session when it has none (parsed from the saved official JSON).""" + if session.scalar(select(Dataset).where(Dataset.kind == "NCM", Dataset.status == "ACTIVE")): + return + from central.app.datasets.ncm import parse_ncm_payload + from central.app.datasets.service import DatasetService + from datetime import datetime, timezone + payload = parse_ncm_payload(ncm_json.read_bytes()) + DatasetService(session).import_payload(payload, "official-json", datetime.now(timezone.utc)) + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--seed-url", default="sqlite://") + ap.add_argument("--ncm-json", default=None, help="official Siscomex JSON (used only if the seed DB has no NCM dataset)") + ap.add_argument("--sample", type=int, default=0) + ap.add_argument("--sample-out", default=None) + args = ap.parse_args() + engine = create_engine(args.seed_url) + Base.metadata.create_all(engine) + dbs = dict(lc.allowlist()) + dbs["TESTESOFICIAIS"] = os.environ.get("FIREBIRD_DATABASE_PATH", str(ROOT / "TESTESOFICIAIS.FDB")) + report, all_rows = {}, {} + with Session(engine) as session: + load_ncm_official(session, Path(args.ncm_json) if args.ncm_json else None) + official = lambda v: ncm_is_official(session, v) # noqa: E731 + for name, path in dbs.items(): + if not Path(path).exists(): + report[name] = {"error": "missing"} + continue + rows = read_lab(name, path) + all_rows[name] = rows + status = Counter(normalize_gtin(r["GTIN"])["status"] for r in rows) + stats = observe(session, rows, name, official) + ncm_ok = sum(1 for r in rows if len("".join(c for c in str(r["NCM"] or "") if c.isdigit())) == 8 and official("".join(c for c in str(r["NCM"]) if c.isdigit()))) + report[name] = {"products": len(rows), "gtin_status": dict(status), "valid_gtin_rate": round(100 * status["VALID_GTIN"] / max(1, len(rows)), 1), "official_ncm_rate": round(100 * ncm_ok / max(1, len(rows)), 1), **stats} + session.commit() + by_gtin = defaultdict(set) + for name, rows in all_rows.items(): + for r in rows: + g = normalize_gtin(r["GTIN"]) + if g["status"] == "VALID_GTIN": + by_gtin[g["key"]].add(name) + shared = {k: v for k, v in by_gtin.items() if len(v) >= 2} + conflicts = 0 + for kp in session.scalars(select(kb.KbProduct)): + ncms = {r.value for r in session.scalars(select(KbKnowledge).where(KbKnowledge.kb_product_id == kp.id, KbKnowledge.knowledge_key == "ncm"))} + conflicts += len(ncms) > 1 + summary = {"databases": report, "distinct_valid_gtin": len(by_gtin), "gtin_in_2plus_databases": len(shared), "kb_products": session.scalar(select(kb.KbProduct.id).limit(1)) is not None and session.query(kb.KbProduct).count(), + "kb_ncm_conflicts_between_clients": conflicts} + if args.sample and args.sample_out: + picks = sorted(by_gtin)[:: max(1, len(by_gtin) // args.sample)][: args.sample] + names = {normalize_gtin(r["GTIN"])["key"]: r["PRODUTO"] for rows in all_rows.values() for r in rows if normalize_gtin(r["GTIN"])["status"] == "VALID_GTIN"} + Path(args.sample_out).parent.mkdir(parents=True, exist_ok=True) + Path(args.sample_out).write_text(json.dumps([{"gtin": k[1:] if k.startswith("0") else k, "erp_description": names[k]} for k in picks], ensure_ascii=False, indent=1), encoding="utf-8") + summary["sample_written"] = len(picks) + print(json.dumps(summary, ensure_ascii=False, indent=1)) + + +if __name__ == "__main__": + main() diff --git a/scripts/enrichment_live_smoke.py b/scripts/enrichment_live_smoke.py new file mode 100644 index 0000000..b3fb858 --- /dev/null +++ b/scripts/enrichment_live_smoke.py @@ -0,0 +1,44 @@ +"""Conservative LIVE smoke of the free providers (separate from the offline test suite): 1 lookup per provider on a famous, stable GTIN. + + python scripts/enrichment_live_smoke.py + +Goes through the real ProviderGateway (cache + quota + circuit breaker) on an in-memory database; tokens are never printed. Exit 0 when every CONFIGURED provider answered +without an infrastructure error (NOT_FOUND is a valid answer; a rate limit is reported, not treated as a failure of our code). +""" +import json +import os +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) +from sqlalchemy import create_engine # noqa: E402 +from sqlalchemy.orm import Session # noqa: E402 +from sqlalchemy.pool import StaticPool # noqa: E402 + +from central.app.enrichment.providers import build_providers # noqa: E402 +from central.app.enrichment.quota import ProviderGateway # noqa: E402 +from central.app.models import Base # noqa: E402 + +GTIN = "7894900011517" # Coca-Cola 350 ml (BR): stable, widely catalogued + + +def main(): + engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool) + Base.metadata.create_all(engine) + settings = type("S", (), {"cosmos_token": os.environ.get("COSMOS_TOKEN", ""), "allow_paid_providers": False})() + out, ok = [], True + with Session(engine) as session: + gateway = ProviderGateway(session, build_providers(settings)) + for provider in gateway.providers: + result = gateway.lookup(provider, GTIN) + row = {"provider": provider.name, "configured": provider.configured, "state": result.state, "reason": result.reason, "title": result.product.title if result.product else None} + out.append(row) + if provider.configured and result.state == "ERROR" and result.reason not in ("RATE_LIMITED",): + ok = False + print(json.dumps({"gtin": GTIN, "results": out, "OK": ok}, ensure_ascii=False, indent=1)) + sys.exit(0 if ok else 1) + + +if __name__ == "__main__": + main() diff --git a/scripts/enrichment_local_hub_e2e.py b/scripts/enrichment_local_hub_e2e.py new file mode 100644 index 0000000..8880990 --- /dev/null +++ b/scripts/enrichment_local_hub_e2e.py @@ -0,0 +1,137 @@ +"""E2E of the enrichment across REAL HTTP: the real Agent worker (real Firebird COPY) <-> the real Hub container (dev PostgreSQL, real datasets, real providers). + + docker compose up -d --build fiscal-hub # Hub at http://localhost:8000 + python scripts/enrichment_local_hub_e2e.py + +Registers a throw-away dev Agent (AGENT_DEV_TOKEN), flips its per-installation features straight in the dev database (the panel route is covered by tests), seeds the KB of the dev Hub with a VERIFIED NCM for +one test GTIN (two independent provider families), then walks: DISABLED -> DRY_RUN -> ENABLED -> DISABLED, an unknown GTIN (real providers, internet) and an invalid GTIN (no provider call). +Writes go ONLY to scratch_rtc_e2e/TESTESOFICIAIS_RTC_COPY.FDB (guarded by rtc_copy_write_e2e.guard). The dev database receives test rows only. +""" +import json +import os +import subprocess +import sys +import threading +import time +import uuid +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)); sys.path.insert(0, str(ROOT / "scripts")); sys.path.insert(0, str(ROOT / "agent" / "src")) +from dotenv import load_dotenv # noqa: E402 + +load_dotenv(ROOT / ".env") +import enrichment_copy_e2e as ce # noqa: E402 (also patches base.connect with the brief retry) +from gti_fiscal_agent import enrichment as E # noqa: E402 +from gti_fiscal_agent.client import HubClient # noqa: E402 + +base = ce.base +HUB = "http://127.0.0.1:8000" # NOT "localhost": on Windows it resolves ::1 first and adds ~500 ms per request +TOKEN = os.environ["AGENT_DEV_TOKEN"] +CONTAINER = "projeto-postgres-1" + + +def sql(query): + done = subprocess.run(["docker", "exec", CONTAINER, "psql", "-U", "gti", "-d", "gti_fiscal", "-At", "-c", query], capture_output=True, text=True, timeout=60) + if done.returncode != 0: + raise RuntimeError(done.stderr[-300:]) + return done.stdout.strip() + + +def seed_kb(gtin14, description, ncm): + code = f''' +from central.app.db import SessionLocal +from central.app.enrichment import kb +with SessionLocal() as s: + p = kb.upsert_product(s, "{gtin14}", "{description}") + kb.add_evidence(s, p, "ncm", "{ncm}", "provider:cosmos", family="PROVIDER:cosmos") + kb.add_evidence(s, p, "ncm", "{ncm}", "provider:other", family="PROVIDER:other") + s.commit() +print("seeded") +''' + done = subprocess.run(["docker", "exec", "-i", "projeto-fiscal-hub-1", "python", "-"], input=code, capture_output=True, text=True, timeout=120, cwd=str(ROOT)) + assert "seeded" in done.stdout, done.stderr[-400:] + + +def set_features(installation_id, mode, corrector=True): + sql(f"UPDATE installations SET enrichment_mode='{mode}', corrector_enabled={'true' if corrector else 'false'}, features_configured=true WHERE installation_id='{installation_id}'") + + +def main(): + checks = {} + base.restore_copy() + client = HubClient(HUB, TOKEN) + try: + installation_id = client.identity().get("installation_id") or str(uuid.uuid4()) # the dev Agent may already own an installation + except Exception: # noqa: BLE001 + installation_id = str(uuid.uuid4()) + client.heartbeat_metadata({"installation_id": installation_id, "status": "IDLE", "agent_version": "1.3.0", "display_name": "E2E-ENRICH", "capabilities": {"PRODUCT_ENRICHMENT_WRITE_V1": True}}) + set_features(installation_id, "DISABLED") # start every run from the default + checks["REGISTERED_DEFAULT_DISABLED"] = client.authorizations()["features"]["enrichment_mode"] == "DISABLED" + gtin_ok = ce.gtin("789500000001") + seed_kb("0" + gtin_ok, "REFRIGERANTE COLA LATA 350ML", "22021000") + for suffix in ("", "-wal", "-shm"): + try: + (ROOT / "scratch_enrich" / f"enrich_hub_e2e.db{suffix}").unlink() + except FileNotFoundError: + pass + store = E.EnrichmentStore(ROOT / "scratch_enrich" / "enrich_hub_e2e.db") + worker = E.EnrichmentWorker(hub=client, store=store, read_conn=lambda: base.connect(False), write_conn=lambda: base.connect(True), create_reprocess=lambda codes: client.create_job("REPROCESS_PRODUCTS", {"codigos": list(codes), "reason": "ENRICHMENT"})) + + class Stop: + stop = False + stopper = Stop() + threading.Thread(target=worker.run_forever, args=(stopper,), daemon=True).start() + + def refresh(): + worker._features_at = -999 + worker._auth_cache = (None, 0.0) + + time.sleep(4) + checks["DISABLED_NO_ACTIVITY"] = store.meta("baseline_done") is None + set_features(installation_id, "DRY_RUN"); refresh(); time.sleep(6) + checks["BASELINE_ON_ACTIVATION"] = store.meta("baseline_done") == "1" + ce.insert_product("HD001", "REFRIGERANTE COLA LATA 350ML", gtin_ok); time.sleep(8) + checks["DRY_RUN_ZERO_WRITES"] = ce.row("HD001") == (None, None, None) + audits = int(sql(f"SELECT count(*) FROM product_field_audits a JOIN installations i ON i.id=a.installation_id WHERE i.installation_id='{installation_id}' AND a.outcome='DRY_RUN' AND a.product_code='HD001'")) + checks["DRY_RUN_AUDITED_IN_HUB"] = audits >= 1 + set_features(installation_id, "ENABLED"); refresh() + ce.insert_product("HE001", "REFRIGERANTE COLA LATA 350ML", gtin_ok) + t0 = time.monotonic() + while time.monotonic() - t0 < 25 and ce.row("HE001")[0] is None: + time.sleep(0.2) + latency = round(time.monotonic() - t0, 2) + r = ce.row("HE001") + checks["ENABLED_NCM_WRITTEN"] = r[0] == "22021000" + time.sleep(3) # the Agent reports the audit right after the write + applied = sql(f"SELECT field||':'||coalesce(new_value,'') FROM product_field_audits a JOIN installations i ON i.id=a.installation_id WHERE i.installation_id='{installation_id}' AND a.outcome='APPLIED' AND a.product_code='HE001' ORDER BY a.id") + checks["WRITTEN_FIELDS_MATCH_HUB_AUDIT"] = ("CLASSIFICACAO_FISCAL:22021000" in applied) and all(v in applied for v in ([f"CEST:{r[1]}"] if r[1] else []) + ([f"IPI:{r[2]}"] if r[2] not in (None,) else [])) + time.sleep(6) + hd = ce.row("HD001") + checks["EARLIER_DRY_RUN_PRODUCT_STILL_UNTOUCHED"] = hd == (None, None, None) + # unknown GTIN goes to the REAL providers (internet); nothing may be written, a human blocker must appear + unknown = ce.gtin("789600000009") + ce.insert_product("HE002", "PRODUTO INEXISTENTE ESPECIAL XYZ 123", unknown); time.sleep(14) + checks["UNKNOWN_GTIN_NO_WRITE"] = ce.row("HE002") == (None, None, None) + blockers = sql("SELECT reason_code FROM blocked_products WHERE field='ENRICHMENT' AND product_code='HE002'") + checks["UNKNOWN_GTIN_BLOCKER_PT"] = blockers in ("NOT_FOUND", "SINGLE_WEAK_SOURCE", "PENDING_PROVIDER_QUOTA", "PENDING_PROVIDERS_OFFLINE") or blockers.startswith("NOT_FOUND") + bad = unknown[:-1] + str((int(unknown[-1]) + 1) % 10) + before_req = sql("SELECT coalesce(sum(requests),0) FROM provider_usage_daily") + ce.insert_product("HE003", "PRODUTO GTIN INVALIDO", bad); time.sleep(8) + checks["INVALID_GTIN_NO_PROVIDER_CALL"] = sql("SELECT coalesce(sum(requests),0) FROM provider_usage_daily") == before_req and ce.row("HE003") == (None, None, None) + # feature switched off centrally: the next new product is not touched + set_features(installation_id, "DISABLED"); refresh(); time.sleep(4) + ce.insert_product("HE004", "REFRIGERANTE COLA LATA 350ML", gtin_ok); time.sleep(8) + checks["DISABLED_AGAIN_ZERO_WRITES"] = ce.row("HE004") == (None, None, None) + # corrector off + enrichment on: independent switches + set_features(installation_id, "ENABLED", corrector=False); refresh(); time.sleep(3) + checks["CORRECTOR_OFF_NO_JOBS"] = client.next_job() is None + stopper.stop = True + result = {"copy_only": True, "hub": HUB, "detect_to_written_seconds_enabled": latency, "checks": checks, "ALL_PASS": all(checks.values())} + (ROOT / "docs" / "runs" / "enrichment_local_hub_e2e.json").write_text(json.dumps(result, indent=1), encoding="utf-8") + print(json.dumps(result, indent=1)) + sys.exit(0 if result["ALL_PASS"] else 1) + + +if __name__ == "__main__": + main() diff --git a/tests/test_release_candidate.py b/tests/test_release_candidate.py index 576b083..98c293b 100644 --- a/tests/test_release_candidate.py +++ b/tests/test_release_candidate.py @@ -14,11 +14,11 @@ sys.path.insert(0, str(ROOT)) def test_versions_are_coherent_across_hub_agent_and_packages(): from central.app.version import GTI_FISCAL_VERSION v = (ROOT / "release" / "VERSION").read_text(encoding="utf-8").strip() - assert v == GTI_FISCAL_VERSION == "1.2.5" + assert v == GTI_FISCAL_VERSION == "1.3.0" pep = v.replace("-rc", "rc") for pp in ("central/pyproject.toml", "agent/pyproject.toml"): assert f'version = "{pep}"' in (ROOT / pp).read_text(encoding="utf-8") - assert '"AGENT_VERSION", "1.2.5"' in (ROOT / "agent" / "scripts" / "run_agent_service.py").read_text(encoding="utf-8") + assert '"AGENT_VERSION", "1.3.0"' in (ROOT / "agent" / "scripts" / "run_agent_service.py").read_text(encoding="utf-8") manifest = ROOT / "release" / "RELEASE_MANIFEST.json" if manifest.exists(): m = json.loads(manifest.read_text(encoding="utf-8"))