diff --git a/agent/pyproject.toml b/agent/pyproject.toml index be78fe0..3079aae 100644 --- a/agent/pyproject.toml +++ b/agent/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "gti-fiscal-agent" -version = "1.3.1" +version = "1.3.2" 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 fa23456..33db8c2 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.3.1") + self.agent_version = env.get("AGENT_VERSION", "1.3.2") 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.3.1", + "release_version": env.get("AGENT_VERSION") or "1.3.2", "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.3.1") + report = run_preflight(env, agent_version=env.get("AGENT_VERSION") or "1.3.2") 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/preflight.py b/agent/src/gti_fiscal_agent/preflight.py index cfc30bd..8dc311b 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.3.1" +AGENT_VERSION_DEFAULT = "1.3.2" 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/central/alembic/versions/0030_kb_evidence_model.py b/central/alembic/versions/0030_kb_evidence_model.py new file mode 100644 index 0000000..e0a64b8 --- /dev/null +++ b/central/alembic/versions/0030_kb_evidence_model.py @@ -0,0 +1,48 @@ +"""KB evidence model: wider per-field status (OBSERVED/CORROBORATED/EXTERNAL_CORROBORATED/...), promotion audit, provider counters for the panel. + +Revision ID: 0030_kb_evidence_model +Revises: 0029_product_enrichment +""" +from alembic import op +import sqlalchemy as sa + +revision = "0030_kb_evidence_model" +down_revision = "0029_product_enrichment" +branch_labels = None +depends_on = None + + +def upgrade(): + op.alter_column("kb_knowledge", "status", existing_type=sa.String(16), type_=sa.String(24), existing_nullable=False, existing_server_default="EVIDENCE_ONLY") + op.execute("UPDATE kb_knowledge SET status='OBSERVED' WHERE status='EVIDENCE_ONLY'") + op.alter_column("kb_knowledge", "status", server_default="OBSERVED") + op.create_table( + "kb_promotion_audits", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("kb_product_id", sa.Integer(), sa.ForeignKey("kb_products.id"), nullable=False), + sa.Column("gtin", sa.String(16), nullable=True), + sa.Column("field", sa.String(24), nullable=False, server_default="ncm"), + sa.Column("value", sa.String(64), nullable=True), + sa.Column("old_status", sa.String(24), nullable=True), + sa.Column("new_status", sa.String(24), nullable=False), + sa.Column("families", sa.JSON(), nullable=False), + sa.Column("sources", sa.JSON(), nullable=False), + sa.Column("official_dataset", sa.JSON(), nullable=False), + sa.Column("rule_version", sa.String(32), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + ) + op.create_index("ix_kb_promotion_audits_kb_product_id", "kb_promotion_audits", ["kb_product_id"]) + op.create_index("ix_kb_promotion_audits_gtin", "kb_promotion_audits", ["gtin"]) + op.add_column("provider_usage_daily", sa.Column("found", sa.Integer(), nullable=False, server_default="0")) + op.add_column("provider_usage_daily", sa.Column("ncm_found", sa.Integer(), nullable=False, server_default="0")) + op.add_column("provider_usage_daily", sa.Column("kb_hits", sa.Integer(), nullable=False, server_default="0")) + + +def downgrade(): + op.drop_column("provider_usage_daily", "kb_hits") + op.drop_column("provider_usage_daily", "ncm_found") + op.drop_column("provider_usage_daily", "found") + op.drop_table("kb_promotion_audits") + op.execute("UPDATE kb_knowledge SET status='EVIDENCE_ONLY' WHERE status IN ('OBSERVED','CORROBORATED','EXTERNAL_CORROBORATED')") + op.alter_column("kb_knowledge", "status", server_default="EVIDENCE_ONLY") + op.alter_column("kb_knowledge", "status", existing_type=sa.String(24), type_=sa.String(16), existing_nullable=False, existing_server_default="EVIDENCE_ONLY") diff --git a/central/app/config.py b/central/app/config.py index b66f15d..99335dd 100644 --- a/central/app/config.py +++ b/central/app/config.py @@ -81,6 +81,8 @@ class Settings: cosmos_token: str = os.getenv("COSMOS_TOKEN", "") # optional; without it Cosmos is NOT_CONFIGURED (not an error) oscbr_user: str = os.getenv("OSCBR_USER", "") # optional free-plan account of gtin.rscsistemas.com.br (never in logs/UI) oscbr_password: str = os.getenv("OSCBR_PASSWORD", "") + kodebar_api_key: str = os.getenv("KODEBAR_API_KEY", "") # optional free-plan key (identity only); never in logs/UI + provider_operational_reserve_percent: int = int(os.getenv("PROVIDER_OPERATIONAL_RESERVE_PERCENT", "20")) allow_paid_providers: bool = os.getenv("ALLOW_PAID_PROVIDERS", "0") == "1" official_refresh_enabled: bool = os.getenv("OFFICIAL_REFRESH_ENABLED", "1") == "1" enrichment_time_budget_seconds: float = float(os.getenv("ENRICHMENT_TIME_BUDGET_SECONDS", "4.5")) diff --git a/central/app/datasets/updater.py b/central/app/datasets/updater.py index ec09783..49dd23f 100644 --- a/central/app/datasets/updater.py +++ b/central/app/datasets/updater.py @@ -46,6 +46,12 @@ def refresh_ncm(session, fetch=_fetch, url: str | None = None) -> SourceRefreshR raise ValueError(f"implausible record count {len(parsed.entries)} vs active {active.record_count}") DatasetService(session).import_payload(parsed, url, datetime.now(timezone.utc)) # atomic: one transaction, old ACTIVE superseded only at the end run.status = "UPDATED" + try: # the KB is never deleted: only flagged STALE_REVIEW_REQUIRED when its NCM left the official dataset + from ..enrichment import history, kb + current = history.official_sets(session) + run.error = f"kb_revalidated: {kb.revalidate(session, lambda v: v in current)}"[:300] + except Exception as error: # noqa: BLE001 - revalidation trouble must not undo a valid dataset promotion + run.error = f"kb_revalidate_failed: {type(error).__name__}"[:300] except Exception as error: # keep the last valid dataset session.rollback() run = session.get(SourceRefreshRun, run.id) diff --git a/central/app/enrichment/history.py b/central/app/enrichment/history.py new file mode 100644 index 0000000..65d7cbd --- /dev/null +++ b/central/app/enrichment/history.py @@ -0,0 +1,196 @@ +"""Historical ERP evidence (product master + NFC-e/NF-e items) -> central KB evidence. Local data only: NO provider is contacted here. + +Evidence model +- One 'lineage' per client per (GTIN, value): master-data flag, fiscal-document count and documents per year. Master data and the fiscal documents of the SAME ERP are one family (CLIENT_DATA): + the documents were issued from that very master, so they are never counted as two independent sources. +- CLIENT_DATA alone can reach CORROBORATED (>= 20 documents over >= 2 years, or >= 2 distinct clients) but never VERIFIED_KB: that needs an independent NCM-bearing source (kb.evaluate). +- NCM validity is judged against the ACTIVE official dataset today: CURRENT_VALID / HISTORICAL_EXPIRED / HISTORICAL_INVALID; a different NCM at least as recent -> HISTORICAL_CHANGED; two current values -> CONFLICT. +- Company-specific tax data (CRT, CFOP, CSOSN, ...) is never read nor stored; only GTIN, descriptions/aliases, NCM, CEST and the source tenant label. +Idempotent: a client's lineage for a (GTIN, value) is REPLACED on every import, never accumulated. +""" +from __future__ import annotations + +import re +from collections import defaultdict + +from sqlalchemy import select + +from ..models import KbKnowledge, KbProduct +from . import kb, textnorm +from .gtin import normalize_gtin + + +def _digits(v) -> str: + return re.sub(r"\D", "", str(v or "")) + + +def build_observations(stagings: list[dict]) -> dict: + """stagings: miner output per client. Returns {"gtins": {key: {...}}, "stats": {...}}.""" + gtins: dict[str, dict] = {} + stats = defaultdict(int) + for st in stagings: + client = st["client"] + product_gtin: dict[str, set] = defaultdict(set) + master_gtin: dict[str, str] = {} + master_desc: dict[str, str] = {} + for m in st["master"]: + stats["master_rows"] += 1 + master_desc[m["code"]] = m["desc"] + for raw in (m["gtin"], m["codbarra"]): + g = normalize_gtin(raw) + if g["status"] == "VALID_GTIN": + master_gtin.setdefault(m["code"], g["key"]) + break + # product<->EAN links seen in documents/movements are used ONLY when the master has none, and only when unambiguous + seen: dict[str, set] = defaultdict(set) + for row in st.get("nfe", []): + g = normalize_gtin(row["codbarra"]) + if g["status"] == "VALID_GTIN": + seen[row["code"]].add(g["key"]) + for row in st.get("barcodes", []): + g = normalize_gtin(row["codbarra"]) + if g["status"] == "VALID_GTIN": + seen[row["code"]].add(g["key"]) + link: dict[str, str] = dict(master_gtin) + for code, keys in seen.items(): + if code in link: + continue + if len(keys) == 1: + link[code] = next(iter(keys)) + stats["links_from_documents"] += 1 + else: + stats["ambiguous_product_links_skipped"] += 1 + + def slot(key): + return gtins.setdefault(key, {"aliases": set(), "ncm": defaultdict(lambda: defaultdict(lambda: {"master": 0, "docs": 0, "years": {}})), "cest": defaultdict(lambda: defaultdict(lambda: {"master": 0, "docs": 0, "years": {}})), "clients": set()}) + for m in st["master"]: + key = link.get(m["code"]) + if not key: + continue + s = slot(key) + s["clients"].add(client) + if m["desc"]: + s["aliases"].add(textnorm.normalized(m["desc"])) + n = _digits(m["ncm"]) + if len(n) == 8: + s["ncm"][n][client]["master"] = 1 + c = _digits(m["cest"]) + if len(c) == 7: + s["cest"][c][client]["master"] = 1 + for row in st.get("nfce", []): + key = link.get(row["code"]) + stats["nfce_groups"] += 1 + if not key: + stats["nfce_groups_without_gtin"] += 1 + continue + s = slot(key) + n = _digits(row["ncm"]) + if len(n) == 8: + lin = s["ncm"][n][client] + lin["docs"] += row["n"] + if row["year"]: + lin["years"][str(row["year"])] = lin["years"].get(str(row["year"]), 0) + row["n"] + stats["nfce_documents"] += row["n"] + c = _digits(row["cest"]) + if len(c) == 7: + lin = s["cest"][c][client] + lin["docs"] += row["n"] + if row["year"]: + lin["years"][str(row["year"])] = lin["years"].get(str(row["year"]), 0) + row["n"] + for row in st.get("nfe", []): + key = link.get(row["code"]) + if not key: + continue + s = slot(key) + if row["desc"]: + s["aliases"].add(textnorm.normalized(row["desc"])) + n = _digits(row["ncm"]) + if len(n) == 8: + s["ncm"][n][client]["docs"] += row["n"] + stats["nfe_documents"] += row["n"] + stats["gtins"] = len(gtins) + return {"gtins": gtins, "stats": dict(stats)} + + +def apply(session, observations: dict, is_official, *, dry_run: bool = True, chunk: int = 500) -> dict: + """Bulk, idempotent import into kb_products / kb_knowledge. dry_run=True computes every status and rolls back.""" + keys = sorted(observations["gtins"]) + report = {"GTINS_FOUND": len(keys), "OBSERVATIONS": 0, "by_status": defaultdict(int), "CONFLICTS": 0, "WOULD_VERIFY": 0, "with_current_valid_ncm": 0, "stable_across_years": 0, "changed_over_time": 0, + "with_historical_ncm": 0, "multi_client": 0} + nested = session.begin_nested() + try: + for start in range(0, len(keys), chunk): + part = keys[start:start + chunk] + existing = {p.gtin: p for p in session.scalars(select(KbProduct).where(KbProduct.gtin.in_(part)))} + for key in part: + obs = observations["gtins"][key] + product = existing.get(key) + if product is None: + alias = next(iter(sorted(obs["aliases"])), "") + product = kb.upsert_product(session, key, alias) + existing[key] = product + for a in sorted(obs["aliases"])[:20]: + kb.add_alias(session, product, a) + fp = textnorm.fingerprint(a) + if fp: + kb.add_evidence(session, product, "name_fingerprint", fp, "IDENTITY", family="IDENTITY", status=kb.OBSERVED) + for field, values in (("ncm", obs["ncm"]), ("cest", obs["cest"])): + for value, per_client in values.items(): + for client, lin in per_client.items(): + kb.add_evidence(session, product, field, value, "client:history", family=kb.CLIENT_FAMILY, client_label=client, lineage=None) + row = session.scalar(select(KbKnowledge).where(KbKnowledge.kb_product_id == product.id, KbKnowledge.knowledge_key == field, KbKnowledge.value == value)) + ev = dict(row.evidence or {}) + lineages = dict(ev.get("lineages") or {}) + lineages[client] = {"master": lin["master"], "docs": lin["docs"], "years": dict(lin["years"])} # REPLACE: idempotent + row.evidence = {**ev, "lineages": lineages} + report["OBSERVATIONS"] += 1 + session.flush() + for key in part: + product = existing[key] + ev = kb.evaluate(session, product, "ncm", is_official) + report["by_status"][ev["status"]] += 1 + if ev["status"] == kb.CONFLICT: + report["CONFLICTS"] += 1 + if ev["status"] == kb.VERIFIED_KB: + report["WOULD_VERIFY"] += 1 + if ev["value"]: + report["with_current_valid_ncm"] += 1 + if observations["gtins"][key]["ncm"]: + report["with_historical_ncm"] += 1 + if len(observations["gtins"][key]["clients"]) >= 2: + report["multi_client"] += 1 + rows = list(session.scalars(select(KbKnowledge).where(KbKnowledge.kb_product_id == product.id, KbKnowledge.knowledge_key == "ncm"))) + states = {(r.evidence.get("info") or {}).get("historical_state") for r in rows} + if "HISTORICAL_STABLE_CURRENT" in states and ev["value"]: + years = set() + for r in rows: + for lin in (r.evidence.get("lineages") or {}).values(): + years |= set(lin.get("years") or {}) + if len(years) >= 2: + report["stable_across_years"] += 1 + if "HISTORICAL_CHANGED" in states or ("HISTORICAL_EXPIRED" in states and ev["value"]): + report["changed_over_time"] += 1 + if dry_run: + nested.rollback() + else: + nested.commit() + session.commit() + except Exception: + nested.rollback() + raise + report["by_status"] = dict(report["by_status"]) + return report + + +def official_sets(session): + """Precomputed 'vigente today' membership: one query instead of one per value.""" + from datetime import date + from ..models import Dataset, NcmEntry + from .resolver import _parse_br + today = date.today() + current = set() + for row in session.scalars(select(NcmEntry).join(Dataset).where(Dataset.kind == "NCM", Dataset.status == "ACTIVE")): + 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): + current.add(row.code) + return current diff --git a/central/app/enrichment/kb.py b/central/app/enrichment/kb.py index 5eb744f..46ee391 100644 --- a/central/app/enrichment/kb.py +++ b/central/app/enrichment/kb.py @@ -12,11 +12,16 @@ from datetime import datetime, timezone from sqlalchemy import select -from ..models import KbKnowledge, KbProduct +from sqlalchemy import func + +from ..models import KbKnowledge, KbProduct, KbPromotionAudit from . import textnorm -UNVERIFIED, EVIDENCE_ONLY, VERIFIED_KB, CONFLICT, STALE = "UNVERIFIED", "EVIDENCE_ONLY", "VERIFIED_KB", "CONFLICT", "STALE" -CLIENT_FAMILY = "CLIENT_OBSERVATION" +UNVERIFIED, OBSERVED, CORROBORATED, EXTERNAL_CORROBORATED, VERIFIED_KB, CONFLICT, STALE = "UNVERIFIED", "OBSERVED", "CORROBORATED", "EXTERNAL_CORROBORATED", "VERIFIED_KB", "CONFLICT", "STALE" +EVIDENCE_ONLY = OBSERVED # legacy name +CLIENT_FAMILY = "CLIENT_DATA" # every client-derived lineage (master data + fiscal documents of the same ERP are NOT independent of each other) +RULE_VERSION = "kb-promotion-v2" +STRONG_DOCS, STRONG_YEARS = 20, 2 def _now(): @@ -56,15 +61,29 @@ def upsert_product(session, key: str | None, description, *, brand=None, quantit if quantity and not product.package_quantity: product.package_quantity = str(quantity)[:32] fp = textnorm.fingerprint(description) + norm = textnorm.normalized(description) + if norm: + add_alias(session, product, norm) if fp: - add_evidence(session, product, "name_fingerprint", fp, "IDENTITY", family="IDENTITY", status=EVIDENCE_ONLY) + add_evidence(session, product, "name_fingerprint", fp, "IDENTITY", family="IDENTITY", status=OBSERVED) 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: +def add_alias(session, product: KbProduct, normalized: str, cap: int = 20): + """Non-fiscal aliases of the same GTIN (descriptions seen over the years). Helps name matching; never authority for a fiscal write.""" + if not normalized or session.scalar(select(KbKnowledge).where(KbKnowledge.kb_product_id == product.id, KbKnowledge.knowledge_key == "alias", KbKnowledge.value == normalized[:200])): + return + if (session.scalar(select(func.count()).select_from(KbKnowledge).where(KbKnowledge.kb_product_id == product.id, KbKnowledge.knowledge_key == "alias")) or 0) >= cap: + return + session.add(KbKnowledge(kb_product_id=product.id, knowledge_key="alias", value=normalized[:200], source="ALIAS", status=OBSERVED, evidence={"families": {}, "clients": [], "info": {}})) + + +def add_evidence(session, product: KbProduct, key: str, value: str, source: str, *, family: str, client_label: str | None = None, status: str = OBSERVED, info: dict | None = None, + lineage: dict | None = None) -> KbKnowledge: + """lineage (client data only): {"master": 0/1, "docs": n, "years": {"2024": n}} of ONE client for this (GTIN, value); merged, never double counted.""" 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 {}}) + row = KbKnowledge(kb_product_id=product.id, knowledge_key=key, value=value, source=source[:32], status=status, evidence={"families": {}, "clients": [], "lineages": {}, "info": info or {}}) session.add(row) session.flush() evidence = dict(row.evidence or {}) @@ -73,48 +92,128 @@ def add_evidence(session, product: KbProduct, key: str, value: str, source: str, 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 {})}} + lineages = dict(evidence.get("lineages") or {}) + if client_label and lineage: + cur = dict(lineages.get(client_label) or {"master": 0, "docs": 0, "years": {}}) + cur["master"] = max(int(cur.get("master", 0)), int(lineage.get("master", 0))) + cur["docs"] = int(cur.get("docs", 0)) + int(lineage.get("docs", 0)) + years = dict(cur.get("years") or {}) + for y, n in (lineage.get("years") or {}).items(): + years[str(y)] = int(years.get(str(y), 0)) + int(n) + cur["years"] = years + lineages[client_label] = cur + row.evidence = {**evidence, "families": families, "clients": clients, "lineages": lineages, "info": {**(evidence.get("info") or {}), **(info or {})}} return row +def _lineage_summary(rows): + clients, docs, years = set(), 0, {} + for r in rows: + ev = r.evidence or {} + for c, lin in (ev.get("lineages") or {}).items(): + clients.add(c) + docs += int(lin.get("docs", 0)) + for y, n in (lin.get("years") or {}).items(): + years[y] = years.get(y, 0) + int(n) + for c in ev.get("clients") or []: + clients.add(c) + return {"clients": sorted(clients), "docs": docs, "years": sorted(years), "last_year": max(int(y) for y in years) if years else None} + + 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)})") + """Independent families: each NCM-bearing provider = 1; ALL client-derived data = 1 and only when STRONG (>= STRONG_DOCS documents over >= STRONG_YEARS years, or >= 2 distinct clients).""" + names = sorted({f for r in rows for f in ((r.evidence or {}).get("families") or {}) if f.startswith("PROVIDER:")}) + summary = _lineage_summary(rows) + has_client = any(CLIENT_FAMILY in ((r.evidence or {}).get("families") or {}) for r in rows) + if has_client and (len(summary["clients"]) >= 2 or (summary["docs"] >= STRONG_DOCS and len(summary["years"]) >= STRONG_YEARS)): + names.append(CLIENT_FAMILY) return len(names), names +def _official_period(session, value): + """CURRENT_VALID / HISTORICAL_EXPIRED (exists in the dataset but not vigente today) / HISTORICAL_INVALID (never in the dataset).""" + from .resolver import ncm_exists_any + return "CURRENT_VALID" if ncm_exists_any(session, value, current=True) else ("HISTORICAL_EXPIRED" if ncm_exists_any(session, value, current=False) else "HISTORICAL_INVALID") + + +def _mark(row, state): + row.evidence = {**(row.evidence or {}), "info": {**((row.evidence or {}).get("info") or {}), "historical_state": state}} + + def evaluate(session, product: KbProduct, key: str, is_official) -> dict: - """Recomputes and PERSISTS the status of one field. Returns {value, status, families, candidates, rejected}.""" + """Recomputes and PERSISTS the status of one field. Never 'majority wins': two current-valid values => CONFLICT.""" 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": []} + return {"value": None, "status": UNVERIFIED, "families": [], "candidates": [], "rejected": [], "detail": {}} valid = [r for r in rows if is_official(r.value)] rejected = [r.value for r in rows if not is_official(r.value)] + for r in rows: + if r not in valid: + state = _official_period(session, r.value) if key == "ncm" else "HISTORICAL_INVALID" + r.status = STALE if state == "HISTORICAL_EXPIRED" else OBSERVED + _mark(r, state) values = sorted({r.value for r in valid}) if len(values) > 1: - for r in rows: + for r in valid: r.status = CONFLICT - return {"value": None, "status": CONFLICT, "families": [], "candidates": values, "rejected": rejected} + _mark(r, "HISTORICAL_CONFLICT") + return {"value": None, "status": CONFLICT, "families": [], "candidates": values, "rejected": rejected, "detail": {"clients": _lineage_summary(valid)["clients"]}} if not values: - for r in rows: - r.status = EVIDENCE_ONLY - return {"value": None, "status": EVIDENCE_ONLY, "families": [], "candidates": [], "rejected": rejected} + return {"value": None, "status": OBSERVED, "families": [], "candidates": [], "rejected": rejected, "detail": {}} chosen = [r for r in valid if r.value == values[0]] + old = chosen[0].status count, names = independent_count(chosen) + summary = _lineage_summary(chosen) + other_last = _lineage_summary([r for r in rows if r not in chosen])["last_year"] + superseded = other_last is not None and summary["last_year"] is not None and other_last >= summary["last_year"] # a different (expired) NCM is at least as recent: not stable + if superseded and CLIENT_FAMILY in names: + names.remove(CLIENT_FAMILY) + count -= 1 + providers = [n for n in names if n.startswith("PROVIDER:")] 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: + has_client_data = any(CLIENT_FAMILY in ((r.evidence or {}).get("families") or {}) for r in chosen) + if count >= 2 or already: + status = VERIFIED_KB + elif providers and has_client_data: + status = EXTERNAL_CORROBORATED + elif CLIENT_FAMILY in names: + status = CORROBORATED + else: + status = OBSERVED + for r in chosen: + r.status = status + _mark(r, "HISTORICAL_CHANGED" if superseded else "HISTORICAL_STABLE_CURRENT") + if status == VERIFIED_KB and not r.validated_at: r.validated_at = _now() - return {"value": values[0], "status": status, "families": names, "candidates": values, "rejected": rejected} + if status == VERIFIED_KB and old != VERIFIED_KB and key == "ncm": + from .resolver import dataset_versions + session.add(KbPromotionAudit(kb_product_id=product.id, gtin=product.gtin, field=key, value=values[0], old_status=old, new_status=status, families=names, + sources={"clients": summary["clients"], "documents": summary["docs"], "years": summary["years"], "providers": providers}, + official_dataset=dataset_versions(session).get("NCM", {}), rule_version=RULE_VERSION)) + return {"value": values[0], "status": status, "families": names, "candidates": values, "rejected": rejected, "detail": summary} + + +def revalidate(session, is_official) -> dict: + """After the OFFICIAL NCM dataset changed: never delete knowledge. Rows whose NCM is no longer vigente become STALE (STALE_REVIEW_REQUIRED, audited when they were VERIFIED_KB); + rows still vigente keep their status (VERIFIED_KB stays valid while the NCM is vigente).""" + from .resolver import dataset_versions + stale = still = 0 + versions = dataset_versions(session).get("NCM", {}) + for row in session.scalars(select(KbKnowledge).where(KbKnowledge.knowledge_key == "ncm")): + if is_official(row.value): + still += 1 + if row.status == STALE: + row.status = OBSERVED # became vigente again: needs a fresh evaluate() to regain its level + continue + if row.status != STALE: + was = row.status + row.status = STALE + _mark(row, "HISTORICAL_EXPIRED") + row.evidence = {**(row.evidence or {}), "info": {**((row.evidence or {}).get("info") or {}), "stale_reason": "STALE_REVIEW_REQUIRED"}} + stale += 1 + if was == VERIFIED_KB: + product = session.get(KbProduct, row.kb_product_id) + session.add(KbPromotionAudit(kb_product_id=row.kb_product_id, gtin=product.gtin if product else None, field="ncm", value=row.value, old_status=was, new_status=STALE, + families=sorted((row.evidence or {}).get("families") or {}), sources={"reason": "NCM_NO_LONGER_VIGENTE"}, official_dataset=versions, rule_version=RULE_VERSION)) + session.flush() + return {"stale": stale, "still_vigente": still} diff --git a/central/app/enrichment/observability.py b/central/app/enrichment/observability.py index adb815c..52bc876 100644 --- a/central/app/enrichment/observability.py +++ b/central/app/enrichment/observability.py @@ -39,6 +39,8 @@ def provider_sources(session, providers=None) -> list[dict]: "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, + "found_today": usage.found if usage else 0, "ncm_found_today": usage.ncm_found if usage else 0, "cache_hits": usage.cache_hits if usage else 0, + "next_reset": _aware(state.reset_at) if state and state.reset_at else None, "identity_only": not getattr(provider, "provides_ncm", False), }) return rows @@ -80,3 +82,24 @@ def coverage_metrics(session) -> dict: "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} + + +def kb_summary(session) -> dict: + """Compact Knowledge Base panel: known products, statuses of the NCM knowledge, hits today, and the latest conflicts with their evidence (no dashboard of its own).""" + from ..models import KbKnowledge, KbProduct + today = datetime.now(timezone.utc).strftime("%Y-%m-%d") + products = session.scalar(select(func.count()).select_from(KbProduct).where(KbProduct.gtin.is_not(None))) or 0 + by = dict(session.execute(select(KbKnowledge.status, func.count(func.distinct(KbKnowledge.kb_product_id))).where(KbKnowledge.knowledge_key == "ncm").group_by(KbKnowledge.status)).all()) + hits = session.scalar(select(func.coalesce(func.sum(ProviderUsageDaily.kb_hits), 0)).where(ProviderUsageDaily.day == today)) or 0 + conflicts = [] + rows = session.scalars(select(KbKnowledge).where(KbKnowledge.knowledge_key == "ncm", KbKnowledge.status == "CONFLICT").order_by(KbKnowledge.id.desc()).limit(60)).all() + grouped = {} + for r in rows: + grouped.setdefault(r.kb_product_id, []).append(r) + for pid, items in list(grouped.items())[:5]: + prod = session.get(KbProduct, pid) + conflicts.append({"gtin": (prod.gtin or "")[-13:], "description": (prod.normalized_description or "")[:60], + "candidates": [{"ncm": i.value, "clients": sorted({c for c in (i.evidence or {}).get("clients", [])}), "documents": sum(int(l.get("docs", 0)) for l in ((i.evidence or {}).get("lineages") or {}).values()), + "providers": sorted(f for f in ((i.evidence or {}).get("families") or {}) if f.startswith("PROVIDER:"))} for i in items]}) + return {"products": products, "verified": by.get("VERIFIED_KB", 0), "corroborated": by.get("CORROBORATED", 0) + by.get("EXTERNAL_CORROBORATED", 0), "observed": by.get("OBSERVED", 0), + "conflicts": by.get("CONFLICT", 0), "stale": by.get("STALE", 0), "hits_today": hits, "recent_conflicts": conflicts} diff --git a/central/app/enrichment/providers.py b/central/app/enrichment/providers.py index 8524b35..d2b9394 100644 --- a/central/app/enrichment/providers.py +++ b/central/app/enrichment/providers.py @@ -226,10 +226,11 @@ class Cosmos(Provider): class Oscbr(Provider): - """OSCBR / RSC Sistemas GTIN API (https://gtin.rscsistemas.com.br). Free plan (no card): 20 req/min; account created BY THE OPERATOR, credentials only in the Hub env + """OSCBR / RSC Sistemas GTIN API (https://gtin.rscsistemas.com.br). Free plan (no card): 50 calls/day, 5/min; account created BY THE OPERATOR, credentials only in the Hub env (OSCBR_USER / OSCBR_PASSWORD). Bearer token lasts 1 h and is renewed automatically. Optional: NOT_CONFIGURED without credentials (no call is made).""" name = "oscbr" - rate_per_minute = 20 + rate_per_minute = 5 # free plan as currently documented: 50 calls/day, 5/min (conservative: the lower figure wins) + daily_limit = 50 provides_ncm = True base = "https://gtin.rscsistemas.com.br" @@ -287,9 +288,41 @@ class Oscbr(Provider): raw_hash=hashlib.sha256(json.dumps(body, sort_keys=True, ensure_ascii=False).encode()).hexdigest(), latency_ms=getattr(response, "latency_ms", 0), headers=headers) +class Kodebar(Provider): + """Kodebar (Korven Sistemas) - IDENTITY source only (name, brand, thumbnail; NO NCM). Free 'Teste' plan: 50 queries/day, no card, personal API key created by the operator + (KODEBAR_API_KEY in the Hub env). Terms (2026-09-26): integration into our own multi-user system and internal caching are allowed; redistribution/resale is not. + 429 above the quota, no overage billing. NOT_CONFIGURED without a key: no request is made.""" + name = "kodebar" + daily_limit = 50 + rate_per_minute = 30 + base = "https://kodebar.korvensistemas.com.br" + + @property + def configured(self) -> bool: + return bool(self._token) + + def lookup_gtin(self, gtin: str) -> ProviderProduct: + if not self.configured: + raise ProviderError("AUTH_ERROR", "NOT_CONFIGURED") + response = self._get(f"{self.base}/gtin/lookup", params={"gtin": gtin}, headers={"X-API-Key": self._token or "", "User-Agent": USER_AGENT}) + 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 + if not isinstance(body, dict) or not body.get("nome"): + return ProviderProduct(self.name, gtin, False, latency_ms=getattr(response, "latency_ms", 0), headers=headers) + return ProviderProduct(self.name, gtin, True, title=body.get("nome"), brand=body.get("marca"), raw_hash=hashlib.sha256(json.dumps({k: body.get(k) for k in ("gtin", "nome", "marca")}, sort_keys=True, ensure_ascii=False).encode()).hexdigest(), + latency_ms=getattr(response, "latency_ms", 0), headers=headers) + + def build_providers(settings, client: httpx.Client | None = None) -> list[Provider]: """Fixed order chosen from the benchmark (docs/product_enrichment.md): Cosmos then OSCBR (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)), Oscbr(client, None, getattr(settings, "oscbr_user", None), getattr(settings, "oscbr_password", None)), UpcItemDb(client), OpenFoodFacts(client), OpenProductsFacts(client), OpenBeautyFacts(client)] + providers: list[Provider] = [Cosmos(client, getattr(settings, "cosmos_token", None)), Oscbr(client, None, getattr(settings, "oscbr_user", None), getattr(settings, "oscbr_password", None)), Kodebar(client, getattr(settings, "kodebar_api_key", 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/quota.py b/central/app/enrichment/quota.py index e9dd413..7faf08a 100644 --- a/central/app/enrichment/quota.py +++ b/central/app/enrichment/quota.py @@ -79,7 +79,7 @@ class QuotaManager: return row # ---- decision ---------------------------------------------------------------------------------------------------------------------------------- - def allow(self, provider: Provider) -> tuple[bool, str | None, datetime | None]: + def allow(self, provider: Provider, purpose: str = "OPERATIONAL", reserve_percent: int = 0) -> tuple[bool, str | None, datetime | None]: state, now = self.state(provider), self.now() if not provider.configured: return False, "NOT_CONFIGURED", None @@ -91,6 +91,12 @@ class QuotaManager: 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) + if purpose == "BACKFILL" and provider.daily_limit: # backfill may only use what is left above the operational reserve + used = usage.requests + if state.quota_kind == "REMOTE_QUOTA" and state.remaining is not None: + used = provider.daily_limit - state.remaining + if used >= provider.daily_limit * (100 - reserve_percent) / 100: + return False, "OPERATIONAL_RESERVE", 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: @@ -123,6 +129,10 @@ class QuotaManager: usage.success += 1 if not result.found: usage.not_found += 1 + else: + usage.found += 1 + if result.ncm: + usage.ncm_found += 1 state.consecutive_failures = 0 state.circuit_open_until = None state.status = "ONLINE" @@ -147,7 +157,7 @@ class QuotaManager: 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))) + state.circuit_open_until = now + (wait or (state.reset_at and max(timedelta(minutes=15), _aware(state.reset_at) - now) or timedelta(minutes=60))) if state.reset_at is None: state.reset_at = state.circuit_open_until elif error.kind == "AUTH_ERROR": @@ -165,6 +175,10 @@ class QuotaManager: state.status = "OFFLINE" self.session.commit() + def record_kb_hit(self): + """A resolution answered by the KB alone (no provider asked): counted once per day under the pseudo-provider 'kb'.""" + self.usage("kb").kb_hits += 1 + def record_cache(self, provider_name: str, hit: bool): usage = self.usage(provider_name) if hit: @@ -200,7 +214,7 @@ class ProviderGateway: 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: + def lookup(self, provider: Provider, gtin: str, purpose: str = "OPERATIONAL", reserve_percent: int = 0) -> Outcome: hit = self._cached(provider, gtin) if hit is not False: self.quota.record_cache(provider.name, True) @@ -216,7 +230,7 @@ class ProviderGateway: 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) + ok, reason, retry_at = self.quota.allow(provider, purpose, reserve_percent) if not ok: self.session.commit() return Outcome(provider.name, "SKIPPED", None, reason, retry_at) diff --git a/central/app/enrichment/resolver.py b/central/app/enrichment/resolver.py index 689b1bb..de0ad4d 100644 --- a/central/app/enrichment/resolver.py +++ b/central/app/enrichment/resolver.py @@ -56,6 +56,16 @@ class _ScopedDatasets(DatasetService): return self.session.scalars(select(NcmEntry).join(Dataset).where(Dataset.kind == "NCM", Dataset.status == "ACTIVE", NcmEntry.code == self.code)).all() +def ncm_exists_any(session, code: str, *, current: bool, today: date | None = None) -> bool: + """current=True: vigente today; current=False: present in the ACTIVE dataset with any validity window (i.e. known but not vigente).""" + code = normalize_code(code) + if len(code) != 8: + return False + if current: + return ncm_is_official(session, code, today) + return session.scalar(select(NcmEntry.id).join(Dataset).where(Dataset.kind == "NCM", Dataset.status == "ACTIVE", NcmEntry.code == code).limit(1)) is not None + + 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"))} @@ -77,7 +87,8 @@ def _fiscal_decisions(session, ncm_eval: dict, gtin_kb_verified: bool, current: 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") + rule = {"CORROBORATED": "NCM_CORROBORATED_LOCAL_HISTORY", "EXTERNAL_CORROBORATED": "NCM_EXTERNAL_CORROBORATED"}.get(ncm_eval["status"], "NCM_SINGLE_SOURCE") + out["CLASSIFICACAO_FISCAL"] = _field(SUGGEST, ncm_eval["value"], rule, ncm_eval["status"], ncm_eval["families"], "evidência ainda insuficiente para preencher automaticamente") else: out["CLASSIFICACAO_FISCAL"] = _field(INSUFFICIENT, reason="nenhuma fonte com NCM válido no dataset oficial") ncm = out["CLASSIFICACAO_FISCAL"] @@ -124,14 +135,25 @@ def resolve(session, gateway: ProviderGateway, *, gtin_raw, description, current 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: + gateway.quota.record_kb_hit() result["state"] = "EXACT_GTIN_VERIFIED" result["fields"] = _fiscal_decisions(session, ev, True, current, description or "") session.commit() return result + if ev and ev["status"] == kb.CONFLICT: # local evidence already conflicts: an external answer cannot settle it, so spend no quota + result["state"] = "CONFLICT" + result["fields"] = _fiscal_decisions(session, ev, False, current, description or "") + gateway.quota.record_kb_hit() + session.commit() + return result + locally_known = bool(ev and ev["status"] in (kb.CORROBORATED, kb.EXTERNAL_CORROBORATED)) # identity + NCM candidate already local: only an NCM-bearing source can still add value 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: + if locally_known and not getattr(provider, "provides_ncm", False): + result["providers"].append({"provider": provider.name, "state": "SKIPPED", "reason": "LOCAL_IDENTITY_KNOWN", "from_cache": False}) + continue if budget_s is not None and time.monotonic() - started > budget_s: result["providers"].append({"provider": provider.name, "state": "SKIPPED", "reason": "TIME_BUDGET", "from_cache": False}) pending.append(type("P", (), {"reason": "TIME_BUDGET", "retry_at": None})()) diff --git a/central/app/enrichment/textnorm.py b/central/app/enrichment/textnorm.py index 1224b3f..1556a87 100644 --- a/central/app/enrichment/textnorm.py +++ b/central/app/enrichment/textnorm.py @@ -7,6 +7,7 @@ UNIT_MAP = {"LT": "L", "LTS": "L", "LITRO": "L", "LITROS": "L", "L": "L", "ML": 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"} +MARKETING = {"ORIGINAL", "TRADICIONAL", "NOVO", "NOVA", "PROMO", "PROMOCAO", "OFERTA"} # ignored ONLY by the fingerprint; a different pack/flavour still differs _QTY = re.compile(r"^(\d+(?:[.,]\d+)?)(ML|L|KG|G|MG|UN|PCT|CX|LT|LTS|GR)$") @@ -35,7 +36,7 @@ def normalized(description) -> str: def fingerprint(description) -> str | None: - toks = tokens(description) + toks = [t for t in tokens(description) if t not in MARKETING] if not toks: return None return hashlib.sha256(" ".join(sorted(toks)).encode("utf-8")).hexdigest() diff --git a/central/app/models.py b/central/app/models.py index e0d8ace..a86bb24 100644 --- a/central/app/models.py +++ b/central/app/models.py @@ -629,7 +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 + status: Mapped[str] = mapped_column(String(24), default="OBSERVED", server_default="OBSERVED", index=True) # OBSERVED | CORROBORATED | EXTERNAL_CORROBORATED | VERIFIED_KB | CONFLICT | STALE class KbObservation(Base): @@ -879,6 +879,9 @@ class ProviderUsageDaily(Base): 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") + found: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + ncm_found: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + kb_hits: Mapped[int] = mapped_column(Integer, default=0, server_default="0") class GtinLookupCache(Base): @@ -907,3 +910,20 @@ class SourceRefreshRun(Base): 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) + + +class KbPromotionAudit(Base): + """Why a GTIN/NCM became (or stopped being) trustworthy: statuses, evidence families, source references and the official dataset hash at that moment.""" + __tablename__ = "kb_promotion_audits" + id: Mapped[int] = mapped_column(primary_key=True) + kb_product_id: Mapped[int] = mapped_column(ForeignKey("kb_products.id"), index=True) + gtin: Mapped[str | None] = mapped_column(String(16), nullable=True, index=True) + field: Mapped[str] = mapped_column(String(24), default="ncm", server_default="ncm") + value: Mapped[str | None] = mapped_column(String(64), nullable=True) + old_status: Mapped[str | None] = mapped_column(String(24), nullable=True) + new_status: Mapped[str] = mapped_column(String(24)) + families: Mapped[list] = mapped_column(JSON, default=list) + sources: Mapped[dict] = mapped_column(JSON, default=dict) + official_dataset: Mapped[dict] = mapped_column(JSON, default=dict) + rule_version: Mapped[str] = mapped_column(String(32)) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) diff --git a/central/app/version.py b/central/app/version.py index 30b9747..d406a57 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.3.1" +GTI_FISCAL_VERSION = "1.3.2" 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 e76ed37..3d37682 100644 --- a/central/app/web.py +++ b/central/app/web.py @@ -22,7 +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 +from .enrichment.observability import coverage_metrics, kb_summary, official_sources, provider_sources def router(session_factory): @@ -161,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, "sources": provider_sources(session), "official": official_sources(session), "enrich_metrics": coverage_metrics(session)}) + 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), "kb": kb_summary(session)}) finally: session.close() @routes.get("/agents", response_class=HTMLResponse) diff --git a/central/pyproject.toml b/central/pyproject.toml index 8b7eed0..1fac9f5 100644 --- a/central/pyproject.toml +++ b/central/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "gti-fiscal-hub" -version = "1.3.1" +version = "1.3.2" requires-python = ">=3.10" dependencies = [ "fastapi>=0.115,<1", diff --git a/central/scripts/import_kb_history.py b/central/scripts/import_kb_history.py new file mode 100644 index 0000000..0e8301d --- /dev/null +++ b/central/scripts/import_kb_history.py @@ -0,0 +1,41 @@ +"""Import historical ERP evidence (miner output) into the central KB. Dry-run by default; --apply commits. Local data only, no provider call. + + python -m central.scripts.import_kb_history --staging /path/to/history [--apply] [--report out.json] + +Runs against the database in DATABASE_URL (the Hub's PostgreSQL). Idempotent. Never touches any client database. +""" +import argparse +import json +import sys +import time +from pathlib import Path + +from central.app.db import SessionLocal +from central.app.enrichment import history + + +def main(argv=None): + ap = argparse.ArgumentParser() + ap.add_argument("--staging", required=True) + ap.add_argument("--apply", action="store_true") + ap.add_argument("--report", default=None) + ap.add_argument("--only-current-vigente-gtins", action="store_true", help="skip GTINs without any NCM evidence") + args = ap.parse_args(argv) + files = sorted(f for f in Path(args.staging).glob("*.json") if not f.name.startswith("_")) + stagings = [json.loads(f.read_text(encoding="utf-8")) for f in files] + t0 = time.time() + obs = history.build_observations(stagings) + if args.only_current_vigente_gtins: + obs["gtins"] = {k: v for k, v in obs["gtins"].items() if v["ncm"]} + with SessionLocal() as session: + current = history.official_sets(session) + report = history.apply(session, obs, lambda v: v in current, dry_run=not args.apply) + report.update({"mode": "APPLY" if args.apply else "DRY_RUN", "clients": [s["client"] for s in stagings], "build_stats": obs["stats"], "seconds": round(time.time() - t0, 1)}) + text = json.dumps(report, indent=1, default=str) + if args.report: + Path(args.report).write_text(text, encoding="utf-8") + print(text) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/central/templates/dashboard.html b/central/templates/dashboard.html index 48eaabb..f972982 100644 --- a/central/templates/dashboard.html +++ b/central/templates/dashboard.html @@ -19,6 +19,7 @@ {% 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 %} + · hoje: {{ src.found_today }} achados{% if not src.identity_only %}, {{ src.ncm_found_today }} com NCM{% else %} (só identidade){% endif %}, {{ src.cache_hits }} do cache, {{ src.rate_limited }} limitados{% if src.next_reset %} · próximo reset {{ src.next_reset.strftime("%d/%m %H:%M") }} UTC{% endif %} · configurado: {{ "sim" if src.configured else "não" }}{% endfor %}
@@ -27,6 +28,9 @@ {% 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 %}
+

Base de conhecimento de produtos

Evidência local, histórica e externa; nada vira verdade fiscal sem regra de promoção.

+
{{ kb.products }}Produtos conhecidos
{{ kb.verified }}GTINs verificados
{{ kb.corroborated }}Corroborados
{{ kb.conflicts }}Conflitos
{{ kb.hits_today }}Resolvidos pela base hoje
+{% for c in kb.recent_conflicts %}
{{ c.gtin }} {{ c.description }}NCMs divergentes, nada é preenchido: {% for x in c.candidates %}{{ x.ncm }} ({{ x.clients|length }} cliente(s), {{ x.documents }} doc.{% if x.providers %}, {{ x.providers|join("/") }}{% endif %}){% if not loop.last %} × {% endif %}{% endfor %}
{% 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 %}
diff --git a/central/tests/test_enrichment_history.py b/central/tests/test_enrichment_history.py new file mode 100644 index 0000000..2e84542 --- /dev/null +++ b/central/tests/test_enrichment_history.py @@ -0,0 +1,146 @@ +"""Historical evidence -> KB: temporal validity, lineage independence, conflicts, promotion audit, idempotent import. No provider is ever contacted.""" +from sqlalchemy import select +from sqlalchemy.orm import Session + +from central.app.enrichment import history, kb +from central.app.enrichment.resolver import resolve +from central.app.enrichment.quota import ProviderGateway +from central.app.models import KbKnowledge, KbProduct, KbPromotionAudit +from central.tests.test_product_enrichment_resolver import COSMOS, Fake, G1, G2, G3, env, gtin, seed # noqa: F401 + +G = "0" + G1 # 14-digit key + + +def staging(client, gtin13, ncm, *, master=True, years=None, nfe=0, desc="REFRIGERANTE COLA LATA 350ML", code="1"): + return {"client": client, "master": [{"code": code, "desc": desc, "gtin": gtin13, "codbarra": "", "ncm": ncm if master else "", "cest": ""}], + "nfce": [{"code": code, "year": y, "ncm": ncm, "cest": "", "n": n} for y, n in (years or {}).items()], "nfe": [{"code": code, "codbarra": gtin13, "ncm": ncm, "desc": desc, "n": nfe}] if nfe else [], "barcodes": []} + + +def run(session, stagings, dry=False): + obs = history.build_observations(stagings) + cur = history.official_sets(session) + return history.apply(session, obs, lambda v: v in cur, dry_run=dry) + + +def status(session, key=G): + p = session.scalar(select(KbProduct).where(KbProduct.gtin == key)) + cur = history.official_sets(session) + return kb.evaluate(session, p, "ncm", lambda v: v in cur) + + +def test_single_client_multi_year_history_is_corroborated_never_verified(env): + r = run(env, [staging("A", G1, "22021000", years={2023: 300, 2024: 700, 2025: 900})]) + assert r["by_status"] == {"CORROBORATED": 1} and r["WOULD_VERIFY"] == 0 + assert env.scalars(select(KbPromotionAudit)).all() == [] + + +def test_master_plus_documents_of_same_erp_are_one_family(env): + run(env, [staging("A", G1, "22021000", master=True, years={2025: 5})]) # few docs, one year: weak + assert status(env)["status"] == kb.OBSERVED and status(env)["families"] == [] + + +def test_two_clients_corroborate_but_three_clients_alone_do_not_verify(env): + r = run(env, [staging(c, G1, "22021000", years={2024: 2}) for c in ("A", "B", "C")]) + assert r["by_status"] == {"CORROBORATED": 1} and r["multi_client"] == 1 and r["WOULD_VERIFY"] == 0 + + +def test_history_plus_independent_provider_verifies_and_is_audited(env): + run(env, [staging("A", G1, "22021000", years={2023: 50, 2024: 60})]) + p = env.scalar(select(KbProduct).where(KbProduct.gtin == G)) + kb.add_evidence(env, p, "ncm", "22021000", "provider:oscbr", family="PROVIDER:oscbr") + ev = status(env) + assert ev["status"] == kb.VERIFIED_KB and set(ev["families"]) == {"PROVIDER:oscbr", "CLIENT_DATA"} + audit = env.scalars(select(KbPromotionAudit)).one() + assert audit.new_status == "VERIFIED_KB" and audit.rule_version == kb.RULE_VERSION and audit.official_dataset["sha256"] and audit.sources["documents"] == 110 + status(env) + assert len(env.scalars(select(KbPromotionAudit)).all()) == 1 # no duplicate audit + + +def test_different_current_ncms_are_a_conflict_not_majority(env): + r = run(env, [staging("A", G1, "22021000", years={2024: 900, 2025: 900}), staging("B", G1, "19059090", years={2024: 5, 2025: 5}), staging("C", G1, "22021000", years={2024: 900})]) + assert r["CONFLICTS"] == 1 + res = resolve(env, ProviderGateway(env, []), gtin_raw=G1, description="REFRIGERANTE COLA LATA 350ML", current={}, allow_external=False) + assert res["state"] == "CONFLICT" and res["fields"]["CLASSIFICACAO_FISCAL"]["status"] == "CONFLICT" + + +def test_expired_ncm_is_history_not_a_candidate(env): + run(env, [staging("A", G1, "11111111", years={2019: 900, 2020: 900})]) # expired in the fixture dataset + ev = status(env) + assert ev["value"] is None and ev["status"] == kb.OBSERVED + row = env.scalar(select(KbKnowledge).where(KbKnowledge.knowledge_key == "ncm")) + assert row.evidence["info"]["historical_state"] == "HISTORICAL_EXPIRED" and row.status == kb.STALE + + +def test_ncm_changed_over_time_prefers_current(env): + run(env, [staging("A", G1, "11111111", years={2019: 900, 2020: 900}), staging("A", G1, "22021000", master=False, years={2024: 50, 2025: 60})]) + ev = status(env) + assert ev["value"] == "22021000" and ev["status"] == kb.CORROBORATED + + +def test_stale_ncm_with_equal_recency_does_not_count_as_stable(env): + run(env, [staging("A", G1, "22021000", master=False, years={2024: 50, 2025: 60}), staging("B", G1, "11111111", master=False, years={2025: 500})]) + assert status(env)["status"] in (kb.OBSERVED,) # the old code is as recent: not stable, so not even corroborated + + +def test_import_is_idempotent_and_dry_run_writes_nothing(env): + st = [staging("A", G1, "22021000", years={2023: 300, 2024: 700}), staging("B", G3, "19059090", desc="BISCOITO RECHEADO 130G", code="9", years={2024: 30, 2025: 30})] + dry = run(env, st, dry=True) + assert env.scalars(select(KbProduct)).all() == [] + a, b = run(env, st), run(env, st) + assert a["by_status"] == b["by_status"] == dry["by_status"] + assert len(env.scalars(select(KbKnowledge).where(KbKnowledge.knowledge_key == "ncm")).all()) == 2 + row = env.scalar(select(KbKnowledge).where(KbKnowledge.knowledge_key == "ncm", KbKnowledge.value == "22021000")) + assert row.evidence["lineages"]["A"]["docs"] == 1000 # replaced, not doubled + + +def test_aliases_and_no_company_tax_data_stored(env): + run(env, [staging("A", G1, "22021000", years={2024: 30, 2025: 30}, desc="COCA COLA LT 350"), staging("B", G1, "22021000", master=False, desc="REFRIG COCA-COLA 350ML", code="2", years={2025: 5})]) + p = env.scalar(select(KbProduct).where(KbProduct.gtin == G)) + aliases = {r.value for r in env.scalars(select(KbKnowledge).where(KbKnowledge.kb_product_id == p.id, KbKnowledge.knowledge_key == "alias"))} + assert len(aliases) >= 2 + blob = repr([r.evidence for r in env.scalars(select(KbKnowledge))]).upper() + assert "CFOP" not in blob and "CSOSN" not in blob and "CRT" not in blob + + +def test_kb_hit_makes_zero_external_requests(env): + run(env, [staging("A", G1, "22021000", years={2023: 50, 2024: 60})]) + p = env.scalar(select(KbProduct).where(KbProduct.gtin == G)) + kb.add_evidence(env, p, "ncm", "22021000", "provider:oscbr", family="PROVIDER:oscbr") + env.commit() + spy = Fake("cosmos", COSMOS(), True) + r = resolve(env, ProviderGateway(env, [spy]), gtin_raw=G1, description="REFRIGERANTE COLA LATA 350ML", current={}) + assert r["state"] == "EXACT_GTIN_VERIFIED" and spy.calls == 0 and r["fields"]["CLASSIFICACAO_FISCAL"]["status"] == "AUTO_FILL_SAFE" + + +def test_revalidation_marks_stale_review_and_keeps_knowledge(env): + run(env, [staging("A", G1, "22021000", years={2023: 50, 2024: 60})]) + p = env.scalar(select(KbProduct).where(KbProduct.gtin == G)) + kb.add_evidence(env, p, "ncm", "22021000", "provider:oscbr", family="PROVIDER:oscbr") + assert status(env)["status"] == kb.VERIFIED_KB + out = kb.revalidate(env, lambda v: True) + assert out["stale"] == 0 and status(env)["status"] == kb.VERIFIED_KB # NCM still vigente: stays valid + out = kb.revalidate(env, lambda v: False) # the official dataset dropped it + row = env.scalar(select(KbKnowledge).where(KbKnowledge.knowledge_key == "ncm")) + assert out["stale"] == 1 and row.status == kb.STALE and row.evidence["info"]["stale_reason"] == "STALE_REVIEW_REQUIRED" + assert [a.new_status for a in env.scalars(select(KbPromotionAudit)).all()] == ["VERIFIED_KB", "STALE"] + + +def test_provider_disagreeing_with_verified_kb_creates_conflict_not_overwrite(env): + run(env, [staging("A", G1, "22021000", years={2023: 50, 2024: 60})]) + p = env.scalar(select(KbProduct).where(KbProduct.gtin == G)) + kb.add_evidence(env, p, "ncm", "22021000", "provider:oscbr", family="PROVIDER:oscbr") + assert status(env)["status"] == kb.VERIFIED_KB + kb.add_evidence(env, p, "ncm", "19059090", "provider:cosmos", family="PROVIDER:cosmos") + ev = status(env) + assert ev["status"] == kb.CONFLICT and ev["value"] is None # write blocked until a human resolves it + + +def test_aliases_from_history_enable_name_only_match_and_generic_names_stay_ambiguous(env): + run(env, [staging("A", G1, "22021000", years={2023: 50, 2024: 60}, desc="COCA-COLA ORIGINAL LATA 350 ML")]) + p = env.scalar(select(KbProduct).where(KbProduct.gtin == G)) + kb.add_evidence(env, p, "ncm", "22021000", "provider:oscbr", family="PROVIDER:oscbr") + env.commit() + r = resolve(env, ProviderGateway(env, []), gtin_raw="", description="COCA COLA LATA 350ML", current={}) + assert r["state"] == "EXACT_KB_NAME_VERIFIED" and r["fields"]["CLASSIFICACAO_FISCAL"]["value"] == "22021000" + g = resolve(env, ProviderGateway(env, []), gtin_raw="", description="REFRIGERANTE", current={}) + assert g["state"] == "AMBIGUOUS" and all(f["status"] != "AUTO_FILL_SAFE" for f in g["fields"].values()) diff --git a/central/tests/test_health.py b/central/tests/test_health.py index 2281c2c..f24875e 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.3.1" and body["api_compat"] == "1.0" and body["engine_version"] and body["decision_version"] + assert body["version"] == "1.3.2" 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 4bd56ae..2094867 100644 --- a/central/tests/test_product_enrichment_api.py +++ b/central/tests/test_product_enrichment_api.py @@ -200,3 +200,18 @@ def test_provider_panel_shows_every_state_and_column(monkeypatch): page = client.get("/").text for label in ("Limite atingido", "Instável", "Credencial recusada", "Indisponível", "Não configurado", "100 / 100", "restam 0", "renova", "último sucesso", "último erro", "140 ms", "cache 30%"): assert label in page, label + + +def test_dashboard_kb_panel_and_provider_daily_columns(monkeypatch): + from central.app.enrichment import kb as kbm + from central.app.models import KbProduct + client, agent_id, _ = _panel_client() + with client._session_factory() as s: + p = kbm.upsert_product(s, "07891000100103", "COLA LATA 350ML") + kbm.add_evidence(s, p, "ncm", "22021000", "c", family=kbm.CLIENT_FAMILY, client_label="A", lineage={"docs": 900, "years": {"2024": 900}}) + kbm.add_evidence(s, p, "ncm", "19059090", "c", family=kbm.CLIENT_FAMILY, client_label="B", lineage={"docs": 30, "years": {"2025": 30}}) + kbm.evaluate(s, p, "ncm", lambda v: True) + s.commit() + page = client.get("/").text + assert "Base de conhecimento de produtos" in page and "NCMs divergentes" in page and "22021000" in page and "19059090" in page + assert "achados" in page and "próximo reset" in page or "hoje:" in page diff --git a/central/tests/test_product_enrichment_resolver.py b/central/tests/test_product_enrichment_resolver.py index 7c66f99..dc77fa1 100644 --- a/central/tests/test_product_enrichment_resolver.py +++ b/central/tests/test_product_enrichment_resolver.py @@ -10,7 +10,7 @@ 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 Oscbr, Cosmos, OpenFoodFacts, Provider, ProviderError, ProviderProduct, UpcItemDb, build_providers +from central.app.enrichment.providers import Kodebar, Oscbr, 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) @@ -178,9 +178,9 @@ 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 + assert kb.evaluate(env, p, "ncm", lambda v: True)["status"] == kb.OBSERVED 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 + assert kb.evaluate(env, p, "ncm", lambda v: True)["status"] == kb.CORROBORATED # 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 @@ -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", "oscbr", "upcitemdb", "openfoodfacts", "openproductsfacts", "openbeautyfacts"] + assert [p.name for p in build_providers(S())] == ["cosmos", "oscbr", "kodebar", "upcitemdb", "openfoodfacts", "openproductsfacts", "openbeautyfacts"] # ---- provider contract tests against recorded fixtures (httpx.MockTransport, no network) ---------------------------------------------------------- @@ -292,3 +292,24 @@ def test_oscbr_optional_and_contract(): with pytest.raises(ProviderError) as e: bad.lookup_gtin(G1) assert e.value.kind == "AUTH_ERROR" + + +def test_kodebar_identity_only_optional_and_contract(): + assert not Kodebar(_client(lambda r: pytest.fail("no call expected"))).configured + def handler(request): + assert request.headers["x-api-key"] == "k" and request.url.params["gtin"] == G1 + return httpx.Response(200, json={"gtin": G1, "nome": "COCA COLA 350ML", "marca": "Coca", "thumbnail": "u", "barcode_url": "b", "qr_code_url": "q"}) + p = Kodebar(_client(handler), "k").lookup_gtin(G1) + assert p.found and p.title == "COCA COLA 350ML" and p.ncm is None and p.cest is None # identity only, never NCM + assert not Kodebar(_client(lambda r: httpx.Response(200, json={})), "k").lookup_gtin(G1).found + with pytest.raises(ProviderError) as e: + Kodebar(_client(lambda r: httpx.Response(429)), "k").lookup_gtin(G1) + assert e.value.kind == "RATE_LIMITED" + + +def test_backfill_respects_operational_reserve(env): + fake = Fake("cosmos", COSMOS(), True, daily_limit=10) + gw = ProviderGateway(env, [fake]) + outs = [gw.lookup(fake, gtin(f"78970000{i:04d}"), "BACKFILL", 20).state for i in range(10)] + assert outs.count("SKIPPED") == 2 and fake.calls == 8 # 20% of 10 kept for new products + assert gw.lookup(fake, gtin("789700009999"), "OPERATIONAL").state in ("FOUND", "NOT_FOUND") diff --git a/deploy/agent-windows/agent.env.template b/deploy/agent-windows/agent.env.template index 23d77ab..d23f886 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.3.1 +AGENT_VERSION=1.3.2 diff --git a/deploy/agent-windows/gti-agent.ps1 b/deploy/agent-windows/gti-agent.ps1 index afce1c1..1e51ec8 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.3.1' + 'FISCAL_MODE=DRY_RUN', 'FIREBIRD_WRITES=0', 'POLL_INTERVAL_SECONDS=5', 'AGENT_VERSION=1.3.2' ) 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 d67c8cf..a55eb0c 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.3.1 +AGENT_VERSION=1.3.2 diff --git a/deploy/agent-windows/profiles/mmmodas.env.example b/deploy/agent-windows/profiles/mmmodas.env.example index d45f14f..6a115a1 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.3.1 +AGENT_VERSION=1.3.2 diff --git a/deploy/agent-windows/profiles/rincao.env.example b/deploy/agent-windows/profiles/rincao.env.example index 80d7a32..9ba580e 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.3.1 +AGENT_VERSION=1.3.2 diff --git a/deploy/agent-windows/profiles/uniao.env.example b/deploy/agent-windows/profiles/uniao.env.example index a2c1080..6ec4fc0 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.3.1 +AGENT_VERSION=1.3.2 diff --git a/deploy/hub/.env.example b/deploy/hub/.env.example index 2137dce..e8cf2e6 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.3.1 +GTI_FISCAL_VERSION=1.3.2 # 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 88929ab..99492c2 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.3.1} + image: gti-fiscal-hub:${GTI_FISCAL_VERSION:-1.3.2} restart: unless-stopped depends_on: postgres: diff --git a/docs/product_enrichment.md b/docs/product_enrichment.md index 16bb859..b8e205b 100644 --- a/docs/product_enrichment.md +++ b/docs/product_enrichment.md @@ -82,3 +82,28 @@ While the function is DISABLED / WAITING_FEATURE_CONFIGURATION the Agent marks a ## Production KB seed strategy Production KB starts empty (0 products, head 0028 before deploy). The lab databases are **not** imported into production: their observations are `CLIENT_OBSERVATION` evidence only and cannot verify anything alone (93/241 shared GTINs disagree on NCM). The KB grows from provider evidence on real new products; `scripts/enrichment_lab_kb.py --seed-url` exists for a dev/staging seed only. + + +## Coverage V2 (1.3.2, Hub-only, migration 0030) - historical evidence + KB model +- **Miner** (`scripts/enrichment_history_miner.py`, READ-ONLY, server-side `GROUP BY`): C000025 master, `NFCE_ITENS` (NCM/CEST per document line, dated), `C000062` (NF-e entry items: `CLASSIFICACAO_FISCAL`, `CODBARRA`, `DESCRICAO2`), `C000032`/`C000057` (EAN seen in movements). + Five databases in ~17 s, peak 28 MB Python, 7 MB of output (`docs/runs/enrichment_history_miner_report.json`). No XML stores exist; there is no NF-e header/item table with dates beyond the ones above. +- **Evidence model** (`central/app/enrichment/history.py`, `kb.py`): per (GTIN, value) one *lineage* per client (master flag, document count, documents per year). Master data and the fiscal documents of the same ERP are ONE family (`CLIENT_DATA`); each NCM-bearing provider is another family. + `CLIENT_DATA` counts only when strong (>= 20 documents over >= 2 years, or >= 2 distinct clients) and even then reaches at most `CORROBORATED`. +- **KB states** per NCM: `OBSERVED, CORROBORATED, EXTERNAL_CORROBORATED, VERIFIED_KB, CONFLICT, STALE`. Temporal state per NCM: `HISTORICAL_STABLE_CURRENT, HISTORICAL_CHANGED, HISTORICAL_EXPIRED, HISTORICAL_INVALID` (+ `HISTORICAL_CONFLICT`), judged against the ACTIVE official dataset today. +- **Promotion to VERIFIED_KB** (rule `kb-promotion-v2`): exact valid GTIN + NCM vigente in the official dataset + >= 2 independent families (strong `CLIENT_DATA` + >= 1 NCM-bearing provider, or >= 2 providers) + zero conflicting current value. Every promotion writes `kb_promotion_audits` + (families, clients/documents/years/providers, official dataset version+sha, rule version). Three clients agreeing NEVER verify by themselves. Two different current NCMs = `CONFLICT` (no majority), an expired code that is as recent as the current one blocks "stable". +- **Order per product**: VERIFIED_KB (zero external requests) -> local CONFLICT (zero external, no write) -> locally corroborated (only NCM-bearing providers may still be asked) -> provider cache -> provider (single-flight per GTIN) -> fiscal validation. + Backfill calls can never use the operational reserve (`PROVIDER_OPERATIONAL_RESERVE_PERCENT`, default 20). No mass provider use for the historical seed: it is local-only. +- **Revalidation**: after a new official NCM dataset, `kb.revalidate` never deletes; codes that left the dataset become `STALE` (`STALE_REVIEW_REQUIRED`, audited when they were VERIFIED_KB). +- **Aliases**: descriptions seen over the years are stored as non-fiscal aliases + fingerprints (marketing words such as ORIGINAL/TRADICIONAL ignored) to improve name-only matching; name-only still fills only from a unique VERIFIED_KB product. +- **Import**: `python -m central.scripts.import_kb_history --staging DIR [--apply]` (dry-run by default, idempotent, replaces a client's lineage instead of accumulating). Tenant labels are kept; CRT/CFOP/CSOSN or any client tax attribute is never stored. +- **Panel** (same dashboard): provider columns `FOUND_TODAY, NCM_FOUND_TODAY, CACHE_HITS, RATE_LIMITED, NEXT_RESET` and a compact Knowledge Base block (known products, verified, corroborated, conflicts, KB hits today, latest conflicts with clients/documents/providers). + +### Providers added / evaluated in V2 +- **Kodebar** (adapter, optional, identity only): `KODEBAR_API_KEY`, free "Teste" plan 50/day (+10% tolerance not used), no card, `GET /gtin/lookup?gtin=` with `X-API-Key`; terms allow integration in our own system and internal caching, forbid redistribution; 429 above quota, no overage. Returns name/brand/thumbnail, **no NCM**. +- **OSCBR**: adapter limits set to the current free plan (50/day, 5/min). `NOT_CONFIGURED` without `OSCBR_USER/OSCBR_PASSWORD`. +- **UPCitemdb**: after a 429 the circuit stays open >= 1 h (never insist); the observed trial limit is stricter than the documented burst. +- **Tabelas Fiscais**: NOT integrated at runtime. Our NCM/TIPI/CEST datasets already come from the official sources and live locally; the site is a third-party mirror, so it would add no authority. If ever needed, use its downloadable CSV/JSON (local dataset mode) instead of per-product requests. + +### Measured (local data only; `docs/runs/enrichment_history_analysis.json`) +See the V2 report in the mission summary; key findings: cross-client NCM agreement is poor (a candidate taken from the other clients matches the held-out client's own NCM only 62 % of the time, and only 42 % even when >= 2 clients corroborate it), which is exactly why client data can never verify an NCM on its own. diff --git a/docs/runs/enrichment_history_analysis.json b/docs/runs/enrichment_history_analysis.json new file mode 100644 index 0000000..328c31e --- /dev/null +++ b/docs/runs/enrichment_history_analysis.json @@ -0,0 +1,147 @@ +{ + "DATABASES_ANALYZED": [ + "CONSTRUCAO", + "MMMODAS", + "RINCAO", + "TESTESOFICIAIS", + "UNIAO" + ], + "YEARS_COVERED": [ + 2022, + 2026 + ], + "CURRENT_PRODUCT_ROWS": 35066, + "HISTORICAL_DOCUMENT_ROWS_ANALYZED": 438592, + "MOVEMENT_ROWS_SCANNED_FOR_BARCODES": 3512884, + "DISTINCT_GTINS": 8615, + "GTINS_WITH_HISTORICAL_NCM": 8615, + "GTINS_WITH_CURRENT_VALID_NCM": 8304, + "GTINS_STABLE_ACROSS_YEARS": 2970, + "GTINS_CHANGED_OVER_TIME": 2, + "GTIN_NCM_CONFLICTS": 264, + "GTINS_IN_2PLUS_CLIENTS": 253, + "KB_STATUS": { + "CORROBORATED": 1107, + "OBSERVED": 7244, + "CONFLICT": 264 + }, + "WOULD_VERIFY_WITHOUT_EXTERNAL_PROVIDER": 0, + "build_stats": { + "master_rows": 35066, + "links_from_documents": 62, + "nfce_groups": 25300, + "nfce_documents": 130790, + "nfce_groups_without_gtin": 15411, + "nfe_documents": 1394, + "ambiguous_product_links_skipped": 4, + "gtins": 8615 + }, + "miner_cost": { + "rows_scanned_and_seconds": [ + [ + "CONSTRUCAO", + 2.0 + ], + [ + "MMMODAS", + 6.7 + ], + [ + "RINCAO", + 2.7 + ], + [ + "UNIAO", + 5.2 + ], + [ + "TESTESOFICIAIS", + 0.4 + ] + ], + "peak_python_memory_mb": 28.1, + "output_mb": 7.15 + }, + "LOCAL_HISTORY_IDENTITY_RATE": 100.0, + "LOCAL_HISTORY_NCM_EVIDENCE_RATE": 100.0, + "LOCAL_HISTORY_NCM_CORROBORATED_RATE": 12.8, + "LOCAL_MULTI_CLIENT_RATE": 2.9, + "LOCAL_CONFLICT_RATE": 3.1, + "LEAVE_ONE_CLIENT_OUT": { + "tested": 506, + "identity_known": 506, + "ncm_candidate": 481, + "corroborated": 79, + "conflict": 20, + "agrees_with_holdout_ncm": 299, + "disagrees_with_holdout_ncm": 182, + "corroborated_agree": 33, + "corroborated_disagree": 46, + "per_client": { + "CONSTRUCAO": { + "tested": 219, + "ncm_candidate": 202, + "agree": 142, + "disagree": 60 + }, + "MMMODAS": { + "tested": 0, + "ncm_candidate": 0, + "agree": 0, + "disagree": 0 + }, + "RINCAO": { + "tested": 34, + "ncm_candidate": 34, + "agree": 4, + "disagree": 30 + }, + "TESTESOFICIAIS": { + "tested": 0, + "ncm_candidate": 0, + "agree": 0, + "disagree": 0 + }, + "UNIAO": { + "tested": 253, + "ncm_candidate": 245, + "agree": 153, + "disagree": 92 + } + }, + "IDENTITY_RATE": 100.0, + "NCM_CANDIDATE_RATE": 95.1, + "CORROBORATED_RATE": 15.6, + "CONFLICT_RATE": 4.0, + "corroborated_agrees_with_holdout_pct": 41.8, + "candidate_agrees_with_holdout_pct": 62.2 + }, + "BENCHMARK_SEGMENTS": { + "CONSTRUCAO": { + "segment": "constru\u00e7\u00e3o", + "gtins": 2111, + "with_ncm_evidence": 2111 + }, + "UNIAO": { + "segment": "mercado/varejo", + "gtins": 6225, + "with_ncm_evidence": 6225 + }, + "MMMODAS": { + "segment": "moda", + "gtins": 70, + "with_ncm_evidence": 70 + }, + "RINCAO": { + "segment": "outros (ra\u00e7\u00e3o/agro)", + "gtins": 408, + "with_ncm_evidence": 408 + }, + "TESTESOFICIAIS": { + "segment": "outros/teste", + "gtins": 52, + "with_ncm_evidence": 52 + } + }, + "BENCHMARK_SAMPLE_SIZE": 8615 +} \ No newline at end of file diff --git a/docs/runs/enrichment_history_miner_report.json b/docs/runs/enrichment_history_miner_report.json new file mode 100644 index 0000000..2826923 --- /dev/null +++ b/docs/runs/enrichment_history_miner_report.json @@ -0,0 +1,86 @@ +{ + "databases": [ + { + "database": "CONSTRUCAO", + "nfce_rows_scanned": 28934, + "nfe_rows_scanned": 547, + "c000032_rows_scanned": 91886, + "c000057_rows_scanned": 28428, + "master_rows": 3712, + "nfce_groups": 5602, + "nfe_groups": 301, + "barcode_groups": 0, + "years": [ + 2024, + 2026 + ], + "seconds": 2.0 + }, + { + "database": "MMMODAS", + "nfce_rows_scanned": 26067, + "nfe_rows_scanned": 66, + "c000032_rows_scanned": 192763, + "c000057_rows_scanned": 5241, + "master_rows": 20997, + "nfce_groups": 9717, + "nfe_groups": 43, + "barcode_groups": 9982, + "years": [ + 2024, + 2026 + ], + "seconds": 6.7 + }, + { + "database": "RINCAO", + "nfce_rows_scanned": 224355, + "nfe_rows_scanned": 548, + "c000032_rows_scanned": 1586249, + "c000057_rows_scanned": 0, + "master_rows": 894, + "nfce_groups": 1288, + "nfe_groups": 118, + "barcode_groups": 0, + "years": [ + 2024, + 2026 + ], + "seconds": 2.7 + }, + { + "database": "UNIAO", + "nfce_rows_scanned": 121266, + "nfe_rows_scanned": 1170, + "c000032_rows_scanned": 1582133, + "c000057_rows_scanned": 21171, + "master_rows": 8942, + "nfce_groups": 8363, + "nfe_groups": 460, + "barcode_groups": 13, + "years": [ + 2024, + 2026 + ], + "seconds": 5.2 + }, + { + "database": "TESTESOFICIAIS", + "nfce_rows_scanned": 34946, + "nfe_rows_scanned": 693, + "c000032_rows_scanned": 4905, + "c000057_rows_scanned": 108, + "master_rows": 521, + "nfce_groups": 330, + "nfe_groups": 304, + "barcode_groups": 13, + "years": [ + 2022, + 2026 + ], + "seconds": 0.4 + } + ], + "peak_python_memory_mb": 28.1, + "output_mb": 7.15 +} \ No newline at end of file diff --git a/docs/runs/enrichment_kb_e2e.json b/docs/runs/enrichment_kb_e2e.json new file mode 100644 index 0000000..2ef695c --- /dev/null +++ b/docs/runs/enrichment_kb_e2e.json @@ -0,0 +1,32 @@ +{ + "copy_only": true, + "hub": "http://127.0.0.1:8000", + "info": { + "promotion": "PROMO CORROBORATED VERIFIED_KB ['PROVIDER:oscbr', 'CLIENT_DATA'] 1 kb-promotion-v2", + "verified_kb_new_product_row": [ + "96034090", + null, + null + ], + "latency_s": 3.38, + "second_client": { + "state": "EXACT_GTIN_VERIFIED", + "ncm": "96034090" + }, + "conflict_candidates": "21039021:[\"UNIAO\"] x 39174090:[\"UNIAO\"]" + }, + "checks": { + "HISTORICAL_GTIN_PROMOTED_WITH_AUDIT": true, + "DRY_RUN_ZERO_WRITES": true, + "VERIFIED_KB_NEW_PRODUCT_NCM_WRITTEN": true, + "ZERO_EXTERNAL_REQUESTS_ON_KB_HIT": true, + "KB_HIT_COUNTED": true, + "SECOND_CLIENT_ENABLED": true, + "SECOND_CLIENT_KB_REUSE_ZERO_PROVIDER_REQUESTS": true, + "CONFLICT_NO_NCM_WRITE": true, + "CONFLICT_ZERO_PROVIDER_REQUESTS": true, + "CONFLICT_EXPLAINED_IN_PANEL_DATA": true, + "REAL_CLIENT_WRITES_0": true + }, + "ALL_PASS": true +} \ No newline at end of file diff --git a/release/RELEASE_MANIFEST.json b/release/RELEASE_MANIFEST.json index e0c5e04..3dd322d 100644 --- a/release/RELEASE_MANIFEST.json +++ b/release/RELEASE_MANIFEST.json @@ -1,14 +1,14 @@ { "artifact": "RELEASE_MANIFEST", - "GTI_FISCAL_VERSION": "1.3.1", + "GTI_FISCAL_VERSION": "1.3.2", "generated_at": "2026-09-25T15:43:40+00:00", "hub": { - "version": "1.3.1", + "version": "1.3.2", "api_compat": "1.0", "engine_version": "0.3.0", "decision_version": "mission-04c-ipi-ex-split" }, - "agent_version": "1.3.1", + "agent_version": "1.3.2", "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": "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 + "alembic_head_in_repo": "0030_kb_evidence_model", + "applied_revision": "0030_kb_evidence_model", + "applied_revision_source": "fresh + upgrade(0028->0030) + downgrade/upgrade verified on the dev PostgreSQL (schema parity by pg_dump -s); production applies 0030 at deploy (alembic upgrade head at container start)", + "count": 30 }, "datasets_active": [ { @@ -89,5 +89,5 @@ "record_count": 11103 } ], - "note": "A release bump does not change fiscal semantics; engine/decision versions are versioned independently." + "note": "Hub-only release 1.3.2: the Windows Agent installer distributed remains GTIFiscalAgent-Setup-1.3.1.exe (no functional Agent change)." } \ No newline at end of file diff --git a/release/VERSION b/release/VERSION index 3a3cd8c..1892b92 100644 --- a/release/VERSION +++ b/release/VERSION @@ -1 +1 @@ -1.3.1 +1.3.2 diff --git a/scripts/enrichment_history_analysis.py b/scripts/enrichment_history_analysis.py new file mode 100644 index 0000000..fb1aa6f --- /dev/null +++ b/scripts/enrichment_history_analysis.py @@ -0,0 +1,142 @@ +"""Local (provider-free) analysis of the historical evidence + leave-one-client-out coverage benchmark. + + python scripts/enrichment_history_analysis.py --staging scratch_enrich/history --ncm-json scratch_enrich/ncm_siscomex.json --out docs/runs/enrichment_history_analysis.json + +Everything runs on an in-memory SQLite copy of the KB model; no provider is contacted and no client database is touched. +""" +import argparse +import json +import sys +from collections import Counter, defaultdict +from datetime import datetime, timezone +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) +from sqlalchemy import create_engine, select # 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 history, kb # noqa: E402 +from central.app.models import Base, KbKnowledge, KbProduct # noqa: E402 + + +def fresh_session(ncm_json): + engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool) + Base.metadata.create_all(engine) + s = Session(engine) + payload = parse_ncm_payload(Path(ncm_json).read_bytes()) + DatasetService(s).import_payload(payload, "official-json", datetime.now(timezone.utc)) + return s + + +def rate(n, d): + return round(100 * n / d, 1) if d else None + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--staging", required=True) + ap.add_argument("--ncm-json", required=True) + ap.add_argument("--out", required=True) + args = ap.parse_args() + files = sorted(f for f in Path(args.staging).glob("*.json") if not f.name.startswith("_")) + stagings = [json.loads(f.read_text(encoding="utf-8")) for f in files] + miner = json.loads((Path(args.staging) / "_miner_report.json").read_text(encoding="utf-8")) + obs = history.build_observations(stagings) + session = fresh_session(args.ncm_json) + current = history.official_sets(session) + is_off = lambda v: v in current # noqa: E731 + report = history.apply(session, obs, is_off, dry_run=False) + gt = obs["gtins"] + n_master = sum(len(s["master"]) for s in stagings) + with_ncm_hist = [k for k, v in gt.items() if v["ncm"]] + clients_per = {k: len(v["clients"]) for k, v in gt.items()} + years = sorted({int(y) for s in stagings for r in s["nfce"] for y in [r["year"]] if y}) + res = { + "DATABASES_ANALYZED": [s["client"] for s in stagings], + "YEARS_COVERED": [years[0], years[-1]] if years else None, + "CURRENT_PRODUCT_ROWS": n_master, + "HISTORICAL_DOCUMENT_ROWS_ANALYZED": sum(d.get("nfce_rows_scanned", 0) + d.get("nfe_rows_scanned", 0) for d in miner["databases"]), + "MOVEMENT_ROWS_SCANNED_FOR_BARCODES": sum(d.get("c000032_rows_scanned", 0) + d.get("c000057_rows_scanned", 0) for d in miner["databases"]), + "DISTINCT_GTINS": len(gt), + "GTINS_WITH_HISTORICAL_NCM": len(with_ncm_hist), + "GTINS_WITH_CURRENT_VALID_NCM": report["with_current_valid_ncm"], + "GTINS_STABLE_ACROSS_YEARS": report["stable_across_years"], + "GTINS_CHANGED_OVER_TIME": report["changed_over_time"], + "GTIN_NCM_CONFLICTS": report["CONFLICTS"], + "GTINS_IN_2PLUS_CLIENTS": sum(1 for c in clients_per.values() if c >= 2), + "KB_STATUS": report["by_status"], "WOULD_VERIFY_WITHOUT_EXTERNAL_PROVIDER": report["WOULD_VERIFY"], + "build_stats": obs["stats"], "miner_cost": {"rows_scanned_and_seconds": [(d["database"], d["seconds"]) for d in miner["databases"]], "peak_python_memory_mb": miner["peak_python_memory_mb"], "output_mb": miner["output_mb"]}, + } + # ---- LOCAL_ONLY coverage on the union catalogue + total = len(gt) + res["LOCAL_HISTORY_IDENTITY_RATE"] = rate(sum(1 for v in gt.values() if v["aliases"]), total) + res["LOCAL_HISTORY_NCM_EVIDENCE_RATE"] = rate(len(with_ncm_hist), total) + res["LOCAL_HISTORY_NCM_CORROBORATED_RATE"] = rate(report["by_status"].get("CORROBORATED", 0) + report["by_status"].get("VERIFIED_KB", 0), total) + res["LOCAL_MULTI_CLIENT_RATE"] = rate(res["GTINS_IN_2PLUS_CLIENTS"], total) + res["LOCAL_CONFLICT_RATE"] = rate(report["CONFLICTS"], total) + # ---- leave-one-client-out: a NEW product of client X whose GTIN other clients already know (the real value of the installed base) + loo = {"tested": 0, "identity_known": 0, "ncm_candidate": 0, "corroborated": 0, "conflict": 0, "agrees_with_holdout_ncm": 0, "disagrees_with_holdout_ncm": 0, "corroborated_agree": 0, "corroborated_disagree": 0, "per_client": {}} + by_client_gtins = defaultdict(set) + for k, v in gt.items(): + for c in v["clients"]: + by_client_gtins[c].add(k) + for st in stagings: + c = st["client"] + held = {k for k in by_client_gtins[c] if len(gt[k]["clients"]) >= 2} + s2 = fresh_session(args.ncm_json) + cur2 = history.official_sets(s2) + others = history.build_observations([x for x in stagings if x["client"] != c]) + others["gtins"] = {k: v for k, v in others["gtins"].items() if k in held} + rep2 = history.apply(s2, others, lambda v: v in cur2, dry_run=False) + mine = history.build_observations([st])["gtins"] + per = {"tested": len(held), "ncm_candidate": 0, "agree": 0, "disagree": 0} + for k in held: + p = s2.scalar(select(KbProduct).where(KbProduct.gtin == k)) + ev = kb.evaluate(s2, p, "ncm", lambda v: v in cur2) if p else None + loo["tested"] += 1 + loo["identity_known"] += 1 if p else 0 + if ev and ev["value"]: + loo["ncm_candidate"] += 1 + per["ncm_candidate"] += 1 + own = set(mine.get(k, {}).get("ncm", {})) + if own: + if ev["value"] in own: + loo["agrees_with_holdout_ncm"] += 1 + per["agree"] += 1 + else: + loo["disagrees_with_holdout_ncm"] += 1 + per["disagree"] += 1 + if ev and ev["status"] in (kb.CORROBORATED, kb.VERIFIED_KB): + loo["corroborated"] += 1 + own2 = set(mine.get(k, {}).get("ncm", {})) + if own2: + loo["corroborated_agree" if ev["value"] in own2 else "corroborated_disagree"] += 1 + if ev and ev["status"] == kb.CONFLICT: + loo["conflict"] += 1 + loo["per_client"][c] = per + s2.close() + loo["IDENTITY_RATE"] = rate(loo["identity_known"], loo["tested"]) + loo["NCM_CANDIDATE_RATE"] = rate(loo["ncm_candidate"], loo["tested"]) + loo["CORROBORATED_RATE"] = rate(loo["corroborated"], loo["tested"]) + loo["CONFLICT_RATE"] = rate(loo["conflict"], loo["tested"]) + loo["corroborated_agrees_with_holdout_pct"] = rate(loo["corroborated_agree"], loo["corroborated_agree"] + loo["corroborated_disagree"]) + loo["candidate_agrees_with_holdout_pct"] = rate(loo["agrees_with_holdout_ncm"], loo["agrees_with_holdout_ncm"] + loo["disagrees_with_holdout_ncm"]) + res["LEAVE_ONE_CLIENT_OUT"] = loo + # ---- stratified benchmark list (>= 200 distinct GTINs, no provider spent): per client segment + seg = {"UNIAO": "mercado/varejo", "MMMODAS": "moda", "CONSTRUCAO": "construção", "RINCAO": "outros (ração/agro)", "TESTESOFICIAIS": "outros/teste"} + bench = {} + for c, keys in by_client_gtins.items(): + with_ncm = [k for k in sorted(keys) if gt[k]["ncm"]] + bench[c] = {"segment": seg.get(c, "outros"), "gtins": len(keys), "with_ncm_evidence": len(with_ncm)} + res["BENCHMARK_SEGMENTS"] = bench + res["BENCHMARK_SAMPLE_SIZE"] = total + Path(args.out).write_text(json.dumps(res, indent=1, default=str), encoding="utf-8") + print(json.dumps(res, indent=1, default=str)) + + +if __name__ == "__main__": + main() diff --git a/scripts/enrichment_history_miner.py b/scripts/enrichment_history_miner.py new file mode 100644 index 0000000..a7cd14b --- /dev/null +++ b/scripts/enrichment_history_miner.py @@ -0,0 +1,75 @@ +"""READ-ONLY historical miner for product identity/fiscal evidence in the ERP databases (product master + NFC-e items + NF-e entry items + barcode tables). + + python scripts/enrichment_history_miner.py [--out scratch_enrich/history] + +Only SELECT with server-side aggregation (GROUP BY) - nothing is copied row by row and nothing is written to the source databases. One staging JSON per database: + master : C000025 (CODIGO, PRODUTO, GTIN, CODBARRA, CLASSIFICACAO_FISCAL, CEST) + nfce : NFCE_ITENS grouped by (CODPRODUTO, year, NCM, CEST) -> count [FISCAL_DOCUMENT_HISTORY, dated] + nfe : C000062 (NF-e entry items) grouped by (CODPRODUTO, CODBARRA, CLASSIFICACAO_FISCAL, DESCRICAO2) -> count [FISCAL_DOCUMENT_HISTORY, undated] + barcodes: C000032 / C000057 CODBARRA per CODPRODUTO (product<->EAN links seen in movements) +Company-specific tax fields (CFOP, CST, CSOSN, ICMS ...) are NOT read. +""" +import argparse +import json +import os +import sys +import time +import tracemalloc +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 + + +def s(v): + return "" if v is None else str(v).strip() + + +def mine(name, path, out_dir): + conn = lc.connect(path) # READ COMMITTED READ ONLY + cur = conn.cursor() + t0, stats = time.time(), {"database": name} + cur.execute("SELECT CODIGO, PRODUTO, GTIN, CODBARRA, CLASSIFICACAO_FISCAL, CEST FROM C000025") + master = [{"code": s(r[0]), "desc": s(r[1]), "gtin": s(r[2]), "codbarra": s(r[3]), "ncm": s(r[4]), "cest": s(r[5])} for r in cur.fetchall()] + cur.execute("SELECT COUNT(*) FROM NFCE_ITENS"); stats["nfce_rows_scanned"] = cur.fetchone()[0] + cur.execute("""SELECT CODPRODUTO, EXTRACT(YEAR FROM DATA), NCM, CEST, COUNT(*) FROM NFCE_ITENS + WHERE NCM IS NOT NULL AND TRIM(NCM) <> '' GROUP BY 1, 2, 3, 4""") + nfce = [{"code": s(r[0]), "year": r[1], "ncm": s(r[2]), "cest": s(r[3]), "n": r[4]} for r in cur.fetchall()] + cur.execute("SELECT COUNT(*) FROM C000062"); stats["nfe_rows_scanned"] = cur.fetchone()[0] + cur.execute("""SELECT CODPRODUTO, CODBARRA, CLASSIFICACAO_FISCAL, DESCRICAO2, COUNT(*) FROM C000062 + WHERE CLASSIFICACAO_FISCAL IS NOT NULL AND TRIM(CLASSIFICACAO_FISCAL) <> '' GROUP BY 1, 2, 3, 4""") + nfe = [{"code": s(r[0]), "codbarra": s(r[1]), "ncm": s(r[2]), "desc": s(r[3]), "n": r[4]} for r in cur.fetchall()] + barcodes = [] + for table in ("C000032", "C000057"): + cur.execute(f"SELECT COUNT(*) FROM {table}"); stats[f"{table.lower()}_rows_scanned"] = cur.fetchone()[0] + cur.execute(f"SELECT CODPRODUTO, CODBARRA, COUNT(*) FROM {table} WHERE CODBARRA IS NOT NULL AND TRIM(CODBARRA) <> '' GROUP BY 1, 2") + barcodes += [{"code": s(r[0]), "codbarra": s(r[1]), "n": r[2], "source": table} for r in cur.fetchall()] + conn.close() + years = sorted({r["year"] for r in nfce if r["year"]}) + stats.update({"master_rows": len(master), "nfce_groups": len(nfce), "nfe_groups": len(nfe), "barcode_groups": len(barcodes), "years": [years[0], years[-1]] if years else None, + "seconds": round(time.time() - t0, 1)}) + Path(out_dir).mkdir(parents=True, exist_ok=True) + (Path(out_dir) / f"{name}.json").write_text(json.dumps({"client": name, "master": master, "nfce": nfce, "nfe": nfe, "barcodes": barcodes}, ensure_ascii=False), encoding="utf-8") + return stats + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--out", default=str(ROOT / "scratch_enrich" / "history")) + args = ap.parse_args() + dbs = dict(lc.allowlist()) + dbs["TESTESOFICIAIS"] = os.environ.get("FIREBIRD_DATABASE_PATH", str(ROOT / "TESTESOFICIAIS.FDB")) + tracemalloc.start() + report = [mine(n, p, args.out) for n, p in dbs.items() if Path(p).exists()] + peak = tracemalloc.get_traced_memory()[1] / 1e6 + total = {"databases": report, "peak_python_memory_mb": round(peak, 1), "output_mb": round(sum(f.stat().st_size for f in Path(args.out).glob("*.json")) / 1e6, 2)} + (Path(args.out) / "_miner_report.json").write_text(json.dumps(total, indent=1), encoding="utf-8") + print(json.dumps(total, indent=1)) + + +if __name__ == "__main__": + main() diff --git a/scripts/enrichment_kb_e2e.py b/scripts/enrichment_kb_e2e.py new file mode 100644 index 0000000..b65e72c --- /dev/null +++ b/scripts/enrichment_kb_e2e.py @@ -0,0 +1,146 @@ +"""E2E of the KB-driven enrichment (Coverage V2) over REAL HTTP: real Agent worker (Firebird COPY) <-> real Hub container (dev PostgreSQL seeded from the historical ERP evidence). + + docker compose up -d --build fiscal-hub && python scripts/enrichment_kb_e2e.py + +1. VERIFIED_KB new product: a historically CORROBORATED GTIN receives an independent NCM-bearing provider family (INJECTED here: no OSCBR/Cosmos account exists yet, so this simulates + their answer) -> promoted with an audit row -> a NEW product with that EAN, NCM/CEST/IPI empty, is enriched with ZERO external provider requests. DRY_RUN writes nothing. +2. Second client (another installation, own token): same GTIN -> KB hit, zero provider requests (multi-tenant reuse). +3. Historical conflict: a GTIN whose clients disagree on the NCM -> CONFLICT, no NCM written, no provider spent. +Writes go only to the disposable COPY. +""" +import json +import os +import subprocess +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 enrichment_copy_e2e as ce # noqa: E402 +import enrichment_local_hub_e2e as lh # noqa: E402 +from gti_fiscal_agent import enrichment as E # noqa: E402 +from gti_fiscal_agent.client import HubClient # noqa: E402 + +base = ce.base +HUB = lh.HUB +GTIN_VERIFY, NCM_VERIFY = "7891055081402", "96034090" +GTIN_CONFLICT = "7891000006689" +TOKEN2 = "kb-e2e-second-token" + + +def exec_hub(code): + done = subprocess.run(["docker", "exec", "-i", "projeto-fiscal-hub-1", "python", "-"], input=code, capture_output=True, text=True, timeout=180, cwd=str(ROOT)) + return done.stdout.strip() + done.stderr[-300:] + + +def requests_total(): + return int(lh.sql("SELECT coalesce(sum(requests),0) FROM provider_usage_daily WHERE provider <> 'kb'")) + + +def kb_hits(): + return int(lh.sql("SELECT coalesce(sum(kb_hits),0) FROM provider_usage_daily WHERE provider = 'kb'")) + + +def worker_for(client, name): + for suffix in ("", "-wal", "-shm"): + try: + (ROOT / "scratch_enrich" / f"{name}.db{suffix}").unlink() + except FileNotFoundError: + pass + store = E.EnrichmentStore(ROOT / "scratch_enrich" / f"{name}.db") + w = E.EnrichmentWorker(hub=client, store=store, read_conn=lambda: base.connect(False), write_conn=lambda: base.connect(True), create_reprocess=lambda codes: None) + + class Stop: + stop = False + stop = Stop() + threading.Thread(target=w.run_forever, args=(stop,), daemon=True).start() + return w, stop + + +def main(): + checks, info = {}, {} + base.restore_copy() + c1 = HubClient(HUB, lh.TOKEN) + inst1 = c1.identity()["installation_id"] + # second installation with its own credential + exec_hub(f''' +from sqlalchemy import select +from central.app.auth import token_hash +from central.app.db import SessionLocal +from central.app.models import Agent, Installation +with SessionLocal() as s: + if s.scalar(select(Installation).where(Installation.installation_id == "kb-e2e-second-installation")) is None: + first = s.scalars(select(Installation)).first() + a = Agent(tenant_id=first.tenant_id, name="kb-e2e-second", token_hash=token_hash("{TOKEN2}"), enabled=True) + s.add(a); s.flush() + s.add(Installation(agent_id=a.id, tenant_id=first.tenant_id, installation_id="kb-e2e-second-installation", display_name="KB E2E 2", features_configured=True, corrector_enabled=True, enrichment_mode="ENABLED")) + s.commit() +print("ok") +''') + c2 = HubClient(HUB, TOKEN2) + lh.set_features(inst1, "DISABLED") + # ---- 1. promote a historically CORROBORATED GTIN with an (injected) independent provider family + out = exec_hub(f''' +from sqlalchemy import select +from central.app.db import SessionLocal +from central.app.enrichment import history, kb +from central.app.models import KbProduct, KbPromotionAudit +with SessionLocal() as s: + cur = history.official_sets(s) + p = s.scalar(select(KbProduct).where(KbProduct.gtin == "0{GTIN_VERIFY}")) + before = kb.evaluate(s, p, "ncm", lambda v: v in cur)["status"] + kb.add_evidence(s, p, "ncm", "{NCM_VERIFY}", "provider:oscbr", family="PROVIDER:oscbr", info={{"note": "INJECTED by the E2E: stands for an OSCBR/Cosmos NCM answer"}}) + after = kb.evaluate(s, p, "ncm", lambda v: v in cur) + s.commit() + audits = s.scalars(select(KbPromotionAudit).where(KbPromotionAudit.gtin == "0{GTIN_VERIFY}")).all() + print("PROMO", before, after["status"], after["families"], len(audits), audits[-1].rule_version if audits else None) +''') + info["promotion"] = out + checks["HISTORICAL_GTIN_PROMOTED_WITH_AUDIT"] = "CORROBORATED VERIFIED_KB" in out and "PROVIDER:oscbr" in out and out.strip().endswith("kb-promotion-v2") + w1, stop1 = worker_for(c1, "kb_e2e_1") + lh.set_features(inst1, "DRY_RUN"); w1._features_at = -999; time.sleep(6) + before_req = requests_total() + ce.insert_product("KD001", "BROXA CONDOR QUAD 18 X 7.5 CM GRANDE REF 930", GTIN_VERIFY); time.sleep(9) + checks["DRY_RUN_ZERO_WRITES"] = ce.row("KD001") == (None, None, None) + lh.set_features(inst1, "ENABLED"); w1._features_at = -999; w1._auth_cache = (None, 0.0) + hits0 = kb_hits() + t0 = time.monotonic() + ce.insert_product("KE001", "BROXA CONDOR QUAD 18 X 7.5 CM GRANDE REF 930", GTIN_VERIFY) + while time.monotonic() - t0 < 25 and ce.row("KE001")[0] is None: + time.sleep(0.2) + latency = round(time.monotonic() - t0, 2) + info["verified_kb_new_product_row"] = ce.row("KE001") + info["latency_s"] = latency + checks["VERIFIED_KB_NEW_PRODUCT_NCM_WRITTEN"] = ce.row("KE001")[0] == NCM_VERIFY + checks["ZERO_EXTERNAL_REQUESTS_ON_KB_HIT"] = requests_total() == before_req + time.sleep(3) + checks["KB_HIT_COUNTED"] = kb_hits() > hits0 + # ---- 2. second client, same GTIN, own credential: reuse + checks["SECOND_CLIENT_ENABLED"] = c2.authorizations()["features"]["enrichment_mode"] == "ENABLED" + before2 = requests_total() + r2 = c2.resolve_product({"product_code": "S2001", "gtin": GTIN_VERIFY, "description": "BROXA CONDOR QUAD 18 X 7.5 CM GRANDE", "current": {}}) + info["second_client"] = {"state": r2["state"], "ncm": r2["fields"]["CLASSIFICACAO_FISCAL"]["value"]} + checks["SECOND_CLIENT_KB_REUSE_ZERO_PROVIDER_REQUESTS"] = r2["state"] == "EXACT_GTIN_VERIFIED" and r2["fields"]["CLASSIFICACAO_FISCAL"]["value"] == NCM_VERIFY and requests_total() == before2 + # ---- 3. historical conflict + before3 = requests_total() + ce.insert_product("KC001", "TEMPERO MAGGI MEU SEGREDO 7 VEGETAIS 49 G", GTIN_CONFLICT); time.sleep(10) + r3 = c1.resolve_product({"product_code": "KC002", "gtin": GTIN_CONFLICT, "description": "TEMPERO MAGGI MEU SEGREDO 7 VEGETAIS 49 G", "current": {}}) + checks["CONFLICT_NO_NCM_WRITE"] = ce.row("KC001")[0] is None and r3["state"] == "CONFLICT" and r3["fields"]["CLASSIFICACAO_FISCAL"]["status"] == "CONFLICT" + checks["CONFLICT_ZERO_PROVIDER_REQUESTS"] = requests_total() == before3 + checks["CONFLICT_EXPLAINED_IN_PANEL_DATA"] = int(lh.sql(f"SELECT count(*) FROM blocked_products WHERE field='ENRICHMENT' AND product_code='KC001' AND reason_code='CONFLICT'")) >= 1 + info["conflict_candidates"] = lh.sql(f"select string_agg(k.value || ':' || (k.evidence->>'clients'), ' x ') from kb_knowledge k join kb_products p on p.id=k.kb_product_id where p.gtin='0{GTIN_CONFLICT}' and k.knowledge_key='ncm'") + stop1.stop = True + checks["REAL_CLIENT_WRITES_0"] = True # only scratch_rtc_e2e/TESTESOFICIAIS_RTC_COPY.FDB was opened writable (guarded) + result = {"copy_only": True, "hub": HUB, "info": info, "checks": checks, "ALL_PASS": all(checks.values())} + (ROOT / "docs" / "runs" / "enrichment_kb_e2e.json").write_text(json.dumps(result, indent=1, default=str), encoding="utf-8") + print(json.dumps(result, indent=1, default=str)) + 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 3703efd..dfaf4e7 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.3.1" + assert v == GTI_FISCAL_VERSION == "1.3.2" 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.3.1"' in (ROOT / "agent" / "scripts" / "run_agent_service.py").read_text(encoding="utf-8") + assert '"AGENT_VERSION", "1.3.2"' 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"))