diff --git a/agent/pyproject.toml b/agent/pyproject.toml index 829cc92..a529120 100644 --- a/agent/pyproject.toml +++ b/agent/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "gti-fiscal-agent" -version = "1.3.4" +version = "1.3.5" 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 03c215a..c09b4c0 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.4") + self.agent_version = env.get("AGENT_VERSION", "1.3.5") 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.4", + "release_version": env.get("AGENT_VERSION") or "1.3.5", "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.4") + report = run_preflight(env, agent_version=env.get("AGENT_VERSION") or "1.3.5") 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/client.py b/agent/src/gti_fiscal_agent/client.py index 2cd95c2..95e5803 100644 --- a/agent/src/gti_fiscal_agent/client.py +++ b/agent/src/gti_fiscal_agent/client.py @@ -45,6 +45,10 @@ class HubClient: """Asks the Hub to resolve a NEW product (the Agent never contacts external providers). Bounded: the Hub answers within its own time budget.""" response = httpx.post(f"{self.url}/v1/products/resolve", json=payload, headers=self._headers(), timeout=20) response.raise_for_status(); return response.json() + def pending_enrichment(self): + """Codes the Hub flagged for a new evaluation (panel button, Reprocessar/Sincronizar, KB decision).""" + response = httpx.get(f"{self.url}/v1/products/enrichment/pending", headers=self._headers(), timeout=15) + response.raise_for_status(); return response.json().get("codes") or [] def report_enrichment(self, payload): response = _resilient(lambda: httpx.post(f"{self.url}/v1/products/enrichment/applied", json=payload, headers=self._headers(), timeout=30)) response.raise_for_status(); return response.json() diff --git a/agent/src/gti_fiscal_agent/enrichment.py b/agent/src/gti_fiscal_agent/enrichment.py index a3034f0..397dabc 100644 --- a/agent/src/gti_fiscal_agent/enrichment.py +++ b/agent/src/gti_fiscal_agent/enrichment.py @@ -29,6 +29,7 @@ READ_FIELDS = ("CODIGO", "PRODUTO", "GTIN", "CODBARRA") + ALLOWED_FIELDS # POLL_SECONDS = 2.0 SAFETY_DIFF_SECONDS = 60.0 OVERRIDE_CHECK_SECONDS = 20.0 +PENDING_POLL_SECONDS = 10.0 # cheap GET: products the Hub flagged for a new evaluation WATCH_WINDOW_SECONDS = 120.0 # a fresh product is re-read for a while: the operator may add the EAN seconds after the first save RETRY_BACKOFF = (30, 60, 120, 300, 600, 1800) @@ -170,7 +171,7 @@ def read_rows(conn, codes) -> dict[str, dict]: return out -def write_field(conn, code: str, field: str, value) -> dict: +def write_field(conn, code: str, field: str, value, observed=None, guard_observed: bool = False) -> dict: """PRODUCT_ENRICHMENT_WRITE_V1: one field, one short transaction, only if the field is still empty. Returns {outcome, reason?}.""" if field not in ALLOWED_FIELDS: raise ValueError(f"field {field!r} is outside the PRODUCT_ENRICHMENT_WRITE_V1 allowlist") @@ -188,6 +189,9 @@ def write_field(conn, code: str, field: str, value) -> dict: conn.rollback() return {"outcome": "ABORTED", "reason": "PRODUCT_ROW_NOT_UNIQUE" if rows else "PRODUCT_NOT_FOUND"} raw = rows[0][0] + if guard_observed and _text(raw) != _text(observed): # STALE_CONFLICT: the value changed between the Hub decision and now (e.g. a person typed something) -> zero write + conn.rollback() + return {"outcome": "ABORTED", "reason": "STALE_CONFLICT"} if not writable_now(field, raw): conn.rollback() return {"outcome": "ABORTED", "reason": "VALUE_ALREADY_SET_NEVER_OVERWRITTEN"} @@ -235,6 +239,7 @@ class EnrichmentWorker: self._last_override = 0.0 self._auth_cache = (None, 0.0) self._count_ms = 0.0 + self._last_pending = 0.0 self._wconn = None self.metrics = {"detected": 0, "resolved": 0, "fields_written": 0, "detect_latency_ms": []} @@ -372,7 +377,7 @@ class EnrichmentWorker: if mode == "DRY_RUN": reports.append({**base, "outcome": "DRY_RUN", "reason": "DRY_RUN_NO_WRITE"}) continue - outcome = self._write_safely(code, field, value) + outcome = self._write_safely(code, field, value, observed=current.get(field)) reports.append({**base, "outcome": outcome["outcome"], "reason": outcome.get("reason")}) if outcome["outcome"] == "APPLIED": written[field] = value @@ -391,14 +396,14 @@ class EnrichmentWorker: except Exception: # noqa: BLE001 logger.warning("reprocess request for %s failed", code) - def _write_safely(self, code, field, value) -> dict: + def _write_safely(self, code, field, value, observed=None) -> dict: """Per field: fresh central authorization (fail closed) + guarded typed UPDATE. write_field re-reads the field INSIDE its own transaction and refuses when anything is there.""" if not self._fresh_enabled(): return {"outcome": "ABORTED", "reason": "AUTHORIZATION_CHANGED"} try: if self._wconn is None: self._wconn = self._open(self.write_conn) - return write_field(self._wconn, code, field, value) + return write_field(self._wconn, code, field, value, observed=observed, guard_observed=True) except Exception as error: # noqa: BLE001 - one field failing never breaks the others or the SGV self._close_write() return {"outcome": "FAILED", "reason": type(error).__name__} @@ -411,6 +416,26 @@ class EnrichmentWorker: except Exception: # noqa: BLE001 pass + def reevaluate_flagged(self) -> list[str]: + """'Reavaliar' in the panel, Reprocessar/Sincronizar and KB decisions: re-run the normal resolve for tracked, not-yet-enriched products (KB/cache first, the Hub decides about providers).""" + try: + codes = self.hub.pending_enrichment() + except Exception: # noqa: BLE001 - Hub offline: try again next cycle + return [] + done = [] + for code in codes[:25]: + item = self.store.get(code) + if item is None: + self.store.enqueue(code, self.clock()) + item = self.store.get(code) + try: + self.process(item) + except Exception as error: # noqa: BLE001 + self._close_write() + logger.warning("re-evaluation of %s failed: %s", code, type(error).__name__) + done.append(code) + return done + # ---- human override watch -------------------------------------------------------------------------------------------------------------------------- def check_overrides(self): tracked = self.store.tracked_written() @@ -468,6 +493,9 @@ class EnrichmentWorker: if row and f"{pick_gtin(row)}|{_text(row['PRODUTO'])}" != item.get("snapshot"): self.process(item) done.append(item["code"]) + if self.clock() - self._last_pending >= PENDING_POLL_SECONDS: + self._last_pending = self.clock() + done += self.reevaluate_flagged() if self.clock() - self._last_override >= OVERRIDE_CHECK_SECONDS: self._last_override = self.clock() self.check_overrides() diff --git a/agent/src/gti_fiscal_agent/preflight.py b/agent/src/gti_fiscal_agent/preflight.py index e5e6eaf..d3edba0 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.4" +AGENT_VERSION_DEFAULT = "1.3.5" REQUIRED_ENV = ("GTI_HUB_URL", "AGENT_TOKEN", "FIREBIRD_DATABASE_PATH", "FB_USER", "FB_PASSWORD") REQUIRED_CAPABILITIES = ("supports_ncm", "supports_ipi", "supports_cest") FISCAL_MODES = ("DRY_RUN", "AUTO_APPLY") diff --git a/agent/tests/test_enrichment.py b/agent/tests/test_enrichment.py index d4054a5..f437346 100644 --- a/agent/tests/test_enrichment.py +++ b/agent/tests/test_enrichment.py @@ -49,6 +49,10 @@ class Hub: fields[f] = {"status": "ALREADY_SET", "value": None} return {"state": self.state, "fields": fields, "dataset_versions": {"NCM": {"sha256": "abc"}}} + def pending_enrichment(self): + codes, self.flagged = list(getattr(self, "flagged", [])), [] + return codes + def report_enrichment(self, payload): self.reports.append(payload) for item in payload["fields"]: @@ -295,3 +299,84 @@ def test_write_field_guards_the_placeholder_swap(): assert E.write_field(Conn(db), "c", "CLASSIFICACAO_FISCAL", "19059090")["outcome"] == "APPLIED" assert E.write_field(Conn(db), "d", "CLASSIFICACAO_FISCAL", "19059090")["outcome"] == "APPLIED" assert [r[0] for r in db.execute("SELECT CLASSIFICACAO_FISCAL FROM C000025 ORDER BY CODIGO")] == ["19059090", "22021000", "19059090", "19059090"] + + +# ---- restart / race / multiple products / stale guard / reevaluation / provenance -------------------------------------------------------------- +def make_worker(db, hub, store, now): + return E.EnrichmentWorker(hub=hub, store=store, read_conn=lambda: Conn(db), write_conn=lambda: Conn(db), create_reprocess=lambda codes: None, clock=lambda: now[0], sleep=lambda s: None) + + +def test_product_created_while_agent_is_down_is_found_after_restart(env): + db, hub, store, worker, now, _ = env + worker.step() + add(db, "D1") # created while the Agent is stopped / restarting + worker2 = make_worker(db, hub, store, now) # new process, same persisted state + now[0] += 5 + assert worker2.step() == ["D1"] and row(db, "D1")[0] == "22021000" + + +def test_many_products_between_polls_and_a_deleted_one(env): + db, hub, store, worker, now, _ = env + worker.step() + for i in range(5): + add(db, f"M{i}") + db.execute("DELETE FROM C000025 WHERE CODIGO='M2'") + db.commit() + add(db, "M9") + now[0] += 5 + done = worker.step() + assert sorted(done) == ["M0", "M1", "M3", "M4", "M9"] and all(row(db, c)[0] == "22021000" for c in done) + + +def test_reevaluation_flagged_by_the_hub_retries_an_unresolved_product(env): + db, hub, store, worker, now, _ = env + worker.step() + hub.decisions = {} # first attempt: nothing known yet + add(db, "U1") + now[0] += 5 + worker.step() + assert row(db, "U1")[0] is None and store.get("U1")["state"] == "DONE" + hub.decisions = dict(DECISIONS) # new evidence (KB decision) + 'Reavaliar'/'Reprocessar' + hub.flagged = ["U1"] + now[0] += 30 + assert "U1" in worker.step() and row(db, "U1")[0] == "22021000" + now[0] += 30 + before = db.total_changes + worker.step() + assert db.total_changes == before + + +def test_stale_write_is_refused_when_a_person_changed_the_value_after_the_decision(env): + db = sqlite3.connect(":memory:") + db.execute("CREATE TABLE C000025(CODIGO TEXT, CLASSIFICACAO_FISCAL TEXT)") + db.execute("INSERT INTO C000025 VALUES('a','19059090')") + db.execute("INSERT INTO C000025 VALUES('b','00000000')") + db.commit() + assert E.write_field(Conn(db), "a", "CLASSIFICACAO_FISCAL", "22021000", observed="00000000", guard_observed=True)["reason"] == "STALE_CONFLICT" + assert E.write_field(Conn(db), "b", "CLASSIFICACAO_FISCAL", "22021000", observed="00000000", guard_observed=True)["outcome"] == "APPLIED" + + +def test_placeholder_after_a_human_override_is_not_filled_again(env): + db, hub, store, worker, now, _ = env + worker.step() + add_with_ncm(db, "HP1", "00000000") # born with the placeholder: eligible + now[0] += 5 + worker.step() + assert row(db, "HP1")[0] == "22021000" + db.execute("UPDATE C000025 SET CLASSIFICACAO_FISCAL='00000000' WHERE CODIGO='HP1'") # a person puts the SAME value back: a human decision now + db.commit() + now[0] += 30 + worker.step() + assert store.is_overridden("HP1", "CLASSIFICACAO_FISCAL") + hub.flagged = ["HP1"] + now[0] += 30 + worker.step() + assert row(db, "HP1")[0] == "00000000" # provenance matters: same value, different meaning + + +def test_automatic_mode_needs_no_sync_button_or_restart(env): + db, hub, store, worker, now, _ = env + worker.step() + add(db, "A1") + now[0] += 3 + assert worker.step() == ["A1"] and row(db, "A1")[0] == "22021000" # save -> next poll -> written diff --git a/central/alembic/versions/0031_enrichment_reevaluate.py b/central/alembic/versions/0031_enrichment_reevaluate.py new file mode 100644 index 0000000..a2a0f8b --- /dev/null +++ b/central/alembic/versions/0031_enrichment_reevaluate.py @@ -0,0 +1,28 @@ +"""Enrichment re-evaluation + visible per-product status (no silent failures). + +Revision ID: 0031_enrichment_reevaluate +Revises: 0030_kb_evidence_model +""" +from alembic import op +import sqlalchemy as sa + +revision = "0031_enrichment_reevaluate" +down_revision = "0030_kb_evidence_model" +branch_labels = None +depends_on = None + + +def upgrade(): + op.add_column("product_enrichments", sa.Column("reevaluate", sa.Boolean(), nullable=False, server_default=sa.text("false"))) + op.add_column("product_enrichments", sa.Column("last_attempt_at", sa.DateTime(timezone=True), nullable=True)) + op.add_column("product_enrichments", sa.Column("ui_status", sa.String(32), nullable=True)) + op.add_column("product_enrichments", sa.Column("reason_code", sa.String(48), nullable=True)) + op.create_index("ix_product_enrichments_reevaluate", "product_enrichments", ["reevaluate"]) + op.create_index("ix_product_enrichments_ui_status", "product_enrichments", ["ui_status"]) + + +def downgrade(): + op.drop_index("ix_product_enrichments_ui_status", table_name="product_enrichments") + op.drop_index("ix_product_enrichments_reevaluate", table_name="product_enrichments") + for name in ("reason_code", "ui_status", "last_attempt_at", "reevaluate"): + op.drop_column("product_enrichments", name) diff --git a/central/app/enrichment/api.py b/central/app/enrichment/api.py index ccdbc51..c7e4b05 100644 --- a/central/app/enrichment/api.py +++ b/central/app/enrichment/api.py @@ -10,6 +10,7 @@ from sqlalchemy import select from ..blocked_products import upsert_blocker from ..config import settings from ..models import Installation, ProductEnrichment, ProductFieldAudit +from . import status as enrichment_status, textnorm from .placeholders import is_placeholder from .providers import build_providers from .quota import ProviderGateway @@ -80,6 +81,8 @@ def router(principal, providers_factory=None): decision["technical_placeholder"] = str(current.get(name)) row.gtin, row.description, row.latency_ms = result.get("gtin"), description, latency row.attempts = (row.attempts or 0) + 1 + row.ui_status, row.reason_code = enrichment_status.classify(result, has_gtin=bool(result.get("gtin")), specific_name=textnorm.is_specific(description)) + row.last_attempt_at, row.reevaluate = datetime.now(timezone.utc), False row.status = "PENDING" if pending else "RESOLVED" row.decisions = {"state": result["state"], "fields": {k: {kk: vv for kk, vv in v.items() if kk != "evidence"} for k, v in result["fields"].items()}} row.next_attempt_at = _aware(result["retry_at"]) if pending else None @@ -124,6 +127,10 @@ def router(principal, providers_factory=None): elif outcome == "HUMAN_OVERRIDDEN": overridden[item["field"]] = True row.fields_applied, row.human_overridden = applied_fields, overridden + ui, why = enrichment_status.after_write([i for i in (payload.get("fields") or []) if isinstance(i, dict)], installation.enrichment_mode) + if ui: + row.ui_status, row.reason_code = ui, why + row.reevaluate = False row.status = "APPLIED" if applied_fields and not overridden else ("PARTIAL" if applied_fields else row.status if row.status != "PENDING" else "RESOLVED") row.updated_at = datetime.now(timezone.utc) session.commit() @@ -131,6 +138,19 @@ def router(principal, providers_factory=None): finally: session.close() + @routes.get("/enrichment/pending") + def pending_reevaluation(identity=Depends(principal)): + """Tracked products flagged for a new evaluation (panel button, REPROCESS/SYNC, KB decision). The Agent polls this cheaply and re-runs resolve for each code.""" + session, agent = identity + try: + installation = session.scalar(select(Installation).where(Installation.agent_id == agent.id)) + if installation is None or not installation.features_configured or installation.enrichment_mode == "DISABLED": + return {"codes": []} + codes = list(session.scalars(select(ProductEnrichment.product_code).where(ProductEnrichment.installation_id == installation.id, ProductEnrichment.reevaluate.is_(True)).order_by(ProductEnrichment.id).limit(25))) + return {"codes": codes} + finally: + session.close() + @routes.get("/enrichment/authorization") def authorization(identity=Depends(principal)): """Fresh central authorization read by the Agent right before EVERY field write.""" diff --git a/central/app/enrichment/kb.py b/central/app/enrichment/kb.py index 46ee391..c6516f8 100644 --- a/central/app/enrichment/kb.py +++ b/central/app/enrichment/kb.py @@ -36,13 +36,39 @@ def find_by_gtin(session, key: str | None) -> KbProduct | None: return session.scalar(select(KbProduct).where(KbProduct.gtin == key)) if key else None +NEUTRAL_EXTRAS = frozenset({"REFRIGERANTE", "REFRIG", "REFR", "BEBIDA", "PRODUTO", "PROD"}) # category nouns only: ZERO / DIET / a flavour / another size are NEVER neutral + + +def _tokset(text) -> frozenset: + return frozenset(t for t in textnorm.tokens(text) if t not in textnorm.MARKETING) + + +def alias_match(query: frozenset, alias: frozenset) -> bool: + """Deterministic, conservative: same token set, or one set contains the other and the difference is only neutral category nouns (so "COCA COLA LATA 350 ML" = "REFRIGERANTE COCA COLA LATA 350 ML", but not "... ZERO ...").""" + if not query or not alias: + return False + if query == alias: + return True + if query < alias: + return (alias - query) <= NEUTRAL_EXTRAS + if alias < query: + return (query - alias) <= NEUTRAL_EXTRAS + return False + + def name_candidates(session, description) -> list[KbProduct]: - """Exact-fingerprint matches (0, 1 or several).""" + """Every KB product with an alias that matches the description under `alias_match` (exact fingerprint included). 0, 1 or several - the caller decides (unique VERIFIED only).""" fp = textnorm.fingerprint(description) if not fp: return [] - ids = list(session.scalars(select(KbKnowledge.kb_product_id).where(KbKnowledge.knowledge_key == "name_fingerprint", KbKnowledge.value == fp).distinct())) - return [p for p in (session.get(KbProduct, i) for i in ids) if p] + ids = set(session.scalars(select(KbKnowledge.kb_product_id).where(KbKnowledge.knowledge_key == "name_fingerprint", KbKnowledge.value == fp).distinct())) + query = _tokset(description) + anchor = max((t for t in query if not t.replace(".", "").isdigit() and len(t) > 2), key=len, default=None) # most selective informative token, then verify in Python + if anchor: + for row in session.execute(select(KbKnowledge.kb_product_id, KbKnowledge.value).where(KbKnowledge.knowledge_key == "alias", KbKnowledge.value.contains(anchor))): + if alias_match(query, _tokset(row[1])): + ids.add(row[0]) + return [p for p in (session.get(KbProduct, i) for i in sorted(ids)) if p] def upsert_product(session, key: str | None, description, *, brand=None, quantity=None) -> KbProduct: @@ -140,6 +166,28 @@ def _mark(row, state): row.evidence = {**(row.evidence or {}), "info": {**((row.evidence or {}).get("info") or {}), "historical_state": state}} +OPERATOR_FAMILY = "OPERATOR_ATTESTATION" + + +def attest(session, gtin_key: str, ncm: str, operator: str, is_official, note: str = "") -> dict: + """A human fiscal decision by an ADMIN (audited by the caller). Only a value that is ALREADY a candidate in the KB (some local/provider evidence) and vigente in the official dataset can be attested; + the dissenting candidates become STALE (REJECTED_BY_OPERATOR) and the product is promoted with the promotion audit naming the operator.""" + product = find_by_gtin(session, gtin_key) + if product is None: + raise ValueError("GTIN_UNKNOWN_IN_KB") + row = session.scalar(select(KbKnowledge).where(KbKnowledge.kb_product_id == product.id, KbKnowledge.knowledge_key == "ncm", KbKnowledge.value == ncm)) + if row is None: + raise ValueError("NCM_IS_NOT_A_CANDIDATE_OF_THIS_GTIN") + if not is_official(ncm): + raise ValueError("NCM_NOT_VIGENTE_IN_OFFICIAL_DATASET") + add_evidence(session, product, "ncm", ncm, f"operator:{operator}"[:32], family=OPERATOR_FAMILY, info={"attested_by": operator, "attested_at": _now().isoformat(), "note": note[:200]}) + for other in session.scalars(select(KbKnowledge).where(KbKnowledge.kb_product_id == product.id, KbKnowledge.knowledge_key == "ncm", KbKnowledge.value != ncm)): + if OPERATOR_FAMILY not in ((other.evidence or {}).get("families") or {}): + other.status = STALE + _mark(other, "REJECTED_BY_OPERATOR") + return evaluate(session, product, "ncm", is_official) + + def evaluate(session, product: KbProduct, key: str, is_official) -> dict: """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))) @@ -152,6 +200,14 @@ def evaluate(session, product: KbProduct, key: str, is_official) -> dict: state = _official_period(session, r.value) if key == "ncm" else "HISTORICAL_INVALID" r.status = STALE if state == "HISTORICAL_EXPIRED" else OBSERVED _mark(r, state) + attested = {r.value for r in valid if OPERATOR_FAMILY in ((r.evidence or {}).get("families") or {})} + if len(attested) == 1: # a human decision resolves the candidates: the others are recorded as rejected, not "outvoted" + winner = next(iter(attested)) + for r in valid: + if r.value != winner: + r.status = STALE + _mark(r, "REJECTED_BY_OPERATOR") + valid = [r for r in valid if r.value == winner] values = sorted({r.value for r in valid}) if len(values) > 1: for r in valid: @@ -169,6 +225,9 @@ def evaluate(session, product: KbProduct, key: str, is_official) -> dict: if superseded and CLIENT_FAMILY in names: names.remove(CLIENT_FAMILY) count -= 1 + if any(OPERATOR_FAMILY in ((r.evidence or {}).get("families") or {}) for r in chosen): + names = sorted(set(names) | {OPERATOR_FAMILY} | ({CLIENT_FAMILY} if any(CLIENT_FAMILY in ((r.evidence or {}).get("families") or {}) for r in chosen) else set())) + count = len(names) providers = [n for n in names if n.startswith("PROVIDER:")] already = any(r.status == VERIFIED_KB and r.validated_at for r in chosen) has_client_data = any(CLIENT_FAMILY in ((r.evidence or {}).get("families") or {}) for r in chosen) @@ -188,7 +247,8 @@ def evaluate(session, product: KbProduct, key: str, is_official) -> dict: 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}, + sources={"clients": summary["clients"], "documents": summary["docs"], "years": summary["years"], "providers": providers, + "operator": next((((r.evidence or {}).get("info") or {}).get("attested_by") for r in chosen if OPERATOR_FAMILY in ((r.evidence or {}).get("families") or {})), None)}, 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} diff --git a/central/app/enrichment/observability.py b/central/app/enrichment/observability.py index 52bc876..0feedcd 100644 --- a/central/app/enrichment/observability.py +++ b/central/app/enrichment/observability.py @@ -84,7 +84,39 @@ def coverage_metrics(session) -> dict: "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: +def _candidate_cards(session, products) -> list[dict]: + from ..models import Dataset, KbKnowledge, NcmEntry + out = [] + for prod in products: + rows = list(session.scalars(select(KbKnowledge).where(KbKnowledge.kb_product_id == prod.id, KbKnowledge.knowledge_key == "ncm"))) + cands = [] + for r in rows: + ev = r.evidence or {} + text = session.scalar(select(NcmEntry.description).join(Dataset).where(Dataset.kind == "NCM", Dataset.status == "ACTIVE", NcmEntry.code == r.value).limit(1)) + cands.append({"ncm": r.value, "official_text": text, "status": r.status, "clients": sorted(ev.get("clients", [])), "documents": sum(int(l.get("docs", 0)) for l in (ev.get("lineages") or {}).values()), + "providers": sorted(f for f in (ev.get("families") or {}) if f.startswith("PROVIDER:")), "attested": (ev.get("info") or {}).get("attested_by")}) + status = "VERIFIED_KB" if any(c["status"] == "VERIFIED_KB" for c in cands) else ("CONFLICT" if any(c["status"] == "CONFLICT" for c in cands) else (cands[0]["status"] if cands else "OBSERVED")) + out.append({"gtin": (prod.gtin or "")[-13:], "gtin_key": prod.gtin, "description": (prod.normalized_description or "")[:70], "status": status, "candidates": [c for c in cands if c["status"] != "STALE" or c["attested"] is None][:4]}) + return out + + +def kb_search(session, query: str, limit: int = 6) -> list[dict]: + from ..models import KbProduct + from . import kb as kbm + q = (query or "").strip() + if not q: + return [] + digits = "".join(ch for ch in q if ch.isdigit()) + products = [] + if len(digits) >= 8 and len(digits) == len(q.replace(" ", "")): + p = kbm.find_by_gtin(session, digits.zfill(14)) + products = [p] if p else [] + else: + products = kbm.name_candidates(session, q)[:limit] + return _candidate_cards(session, products) + + +def kb_summary(session, query: str | None = None) -> 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") @@ -102,4 +134,4 @@ def kb_summary(session) -> dict: "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} + "conflicts": by.get("CONFLICT", 0), "stale": by.get("STALE", 0), "hits_today": hits, "recent_conflicts": conflicts, "search": kb_search(session, query) if query else []} diff --git a/central/app/enrichment/status.py b/central/app/enrichment/status.py new file mode 100644 index 0000000..2e871e4 --- /dev/null +++ b/central/app/enrichment/status.py @@ -0,0 +1,86 @@ +"""Human-facing status of every tracked product: nothing may silently 'do nothing'. Computed from the resolver result and refined when the Agent reports the write outcome.""" +from __future__ import annotations + +from sqlalchemy import select, update + +LABELS = { + "ENRIQUECIDO": "Enriquecido: campos seguros preenchidos.", + "PRONTO_PARA_PREENCHER": "Evidência suficiente: aguardando a escrita segura pelo Agent.", + "SIMULADO": "Simulação (DRY_RUN): há candidato, nada foi gravado.", + "AGUARDANDO_EVIDENCIA": "Aguardando evidência: nenhuma fonte confiável ainda.", + "EAN_AUSENTE": "Sem EAN e sem produto único verificado na base: informe o código de barras ou peça a decisão de um administrador.", + "EAN_INVALIDO": "Código de barras inválido: nenhuma fonte externa foi consultada.", + "NOME_AMBIGUO": "Nome genérico ou ambíguo demais para identificar o produto com segurança.", + "SEM_MATCH_DE_PRODUTO": "Nenhum produto conhecido corresponde a este nome/EAN.", + "NCM_NAO_VERIFICADO": "Produto identificado, mas o NCM ainda não foi verificado (falta segunda fonte independente ou decisão de administrador).", + "CONFLITO_DE_NCM": "As fontes divergem sobre o NCM: nada é preenchido até um administrador decidir.", + "PROVIDER_SEM_QUOTA": "Limite diário das fontes externas atingido; será repetido após o reset.", + "PROVIDER_NAO_CONFIGURADO": "Nenhuma fonte externa com NCM está configurada.", + "AGUARDANDO_RETRY": "Fontes indisponíveis agora; nova tentativa automática agendada.", + "HUMAN_OVERRIDE": "Alterado por uma pessoa: nunca mais preenchido automaticamente.", + "TECHNICAL_PLACEHOLDER": "NCM provisório (00000000) reconhecido como 'ainda não classificado'.", + "WRITE_FAILED": "A escrita segura foi recusada ou falhou (veja o motivo).", + "VERIFICACAO_POS_WRITE_FALHOU": "A escrita foi feita mas a releitura não confirmou o valor.", + "NADA_A_PREENCHER": "Todos os campos elegíveis já estão preenchidos.", + "FUNCAO_DESATIVADA": "Enriquecimento desativado para esta instalação.", +} + + +def classify(result: dict, *, has_gtin: bool, specific_name: bool) -> tuple[str, str]: + """(ui_status, reason_code) from a resolve() result.""" + fields = result.get("fields") or {} + ncm = fields.get("CLASSIFICACAO_FISCAL") or {} + state = result.get("state") + if state == "FEATURE_DISABLED": + return "FUNCAO_DESATIVADA", state + if state == "INVALID_GTIN": + return "EAN_INVALIDO", state + if state == "PENDING_PROVIDER_QUOTA": + return "PROVIDER_SEM_QUOTA", state + if state in ("PENDING_PROVIDERS_OFFLINE", "PENDING_TIME_BUDGET"): + return "AGUARDANDO_RETRY", state + if state == "CONFLICT" or ncm.get("status") == "CONFLICT": + return "CONFLITO_DE_NCM", "CONFLICT" + if ncm.get("status") == "HUMAN_OVERRIDDEN": + return "HUMAN_OVERRIDE", "HUMAN_OVERRIDDEN" + if any(f.get("status") == "AUTO_FILL_SAFE" for f in fields.values()): + return "PRONTO_PARA_PREENCHER", state or "RESOLVED" + if ncm.get("status") == "ALREADY_SET" and not any(f.get("status") in ("AUTO_FILL_SAFE", "SUGGESTION_ONLY") for f in fields.values()): + return "NADA_A_PREENCHER", "ALL_FIELDS_SET" + if not has_gtin: + if state == "AMBIGUOUS" or not specific_name: + return "NOME_AMBIGUO", "GENERIC_OR_AMBIGUOUS_NAME" + return "EAN_AUSENTE", "NAME_ONLY_NO_VERIFIED_MATCH" + if ncm.get("status") == "SUGGESTION_ONLY": + return "NCM_NAO_VERIFICADO", ncm.get("rule") or "NCM_NOT_VERIFIED" + if state == "NOT_FOUND": + return "SEM_MATCH_DE_PRODUTO", "GTIN_UNKNOWN" + return "AGUARDANDO_EVIDENCIA", state or "INSUFFICIENT_EVIDENCE" + + +def after_write(outcomes: list[dict], mode: str) -> tuple[str | None, str | None]: + """Refines the status with what the Agent reported (APPLIED / DRY_RUN / ABORTED / FAILED / HUMAN_OVERRIDDEN).""" + kinds = {o.get("outcome") for o in outcomes} + reasons = [o.get("reason") for o in outcomes if o.get("reason")] + if "HUMAN_OVERRIDDEN" in kinds and "APPLIED" not in kinds: + return "HUMAN_OVERRIDE", "HUMAN_OVERRIDDEN" + if "APPLIED" in kinds: + return "ENRIQUECIDO", None + if "FAILED" in kinds: + return ("VERIFICACAO_POS_WRITE_FALHOU" if "POST_WRITE_VERIFICATION_FAILED" in reasons else "WRITE_FAILED"), (reasons[0] if reasons else "FAILED") + if "ABORTED" in kinds: + return "WRITE_FAILED", (reasons[0] if reasons else "ABORTED") + if "DRY_RUN" in kinds: + return "SIMULADO", "DRY_RUN_NO_WRITE" + return None, None + + +def mark_for_reevaluation(session, installation_id: int, *, codes: list[str] | None = None, gtin: str | None = None) -> int: + """Flags tracked, NOT-yet-enriched products for a new evaluation (KB/cache first; providers only when really needed). Idempotent; never touches enriched / human-overridden ones.""" + from ..models import ProductEnrichment + q = update(ProductEnrichment).where(ProductEnrichment.installation_id == installation_id, ProductEnrichment.ui_status.notin_(["ENRIQUECIDO", "HUMAN_OVERRIDE"]) | ProductEnrichment.ui_status.is_(None)) + if codes: + q = q.where(ProductEnrichment.product_code.in_(codes)) + if gtin: + q = q.where(ProductEnrichment.gtin == gtin) + return session.execute(q.values(reevaluate=True)).rowcount or 0 diff --git a/central/app/jobs.py b/central/app/jobs.py index 035068b..74f7f6f 100644 --- a/central/app/jobs.py +++ b/central/app/jobs.py @@ -30,6 +30,9 @@ def create_job(session, agent, kind, payload=None): (no duplicates, never two full jobs at once).""" if kind not in _VALID_KINDS: raise ValueError(f"unsupported job kind: {kind}") + if kind in ("REPROCESS_ALL", "REPROCESS_PRODUCTS", "SYNC_INCREMENTAL"): + _flag_enrichment_reevaluation(session, agent) # "Reprocessar"/"Sincronizar" also re-evaluate the tracked, not-yet-enriched products (KB/cache first), even when an equivalent job is already queued + session.commit() if kind in SYNC_KINDS: # A full job subsumes an incremental one, not the other way round: asking for a # full reprocess while only an incremental is queued must still queue the full. @@ -43,6 +46,16 @@ def create_job(session, agent, kind, payload=None): return job +def _flag_enrichment_reevaluation(session, agent): + try: + installation = session.scalar(select(Installation).where(Installation.agent_id == agent.id)) + if installation is not None and installation.features_configured and installation.enrichment_mode != "DISABLED": + from .enrichment.status import mark_for_reevaluation + mark_for_reevaluation(session, installation.id) + except Exception: # noqa: BLE001 - never let the enrichment bookkeeping break job creation + pass + + def recover_stale_running(session, agent, now=None): now = now or datetime.now(timezone.utc) cutoff = now - STALE_RUNNING_AFTER diff --git a/central/app/models.py b/central/app/models.py index a86bb24..0184a0c 100644 --- a/central/app/models.py +++ b/central/app/models.py @@ -816,6 +816,10 @@ class ProductEnrichment(Base): attempts: Mapped[int] = mapped_column(Integer, default=0, server_default="0") next_attempt_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True) latency_ms: Mapped[int | None] = mapped_column(Integer, nullable=True) + reevaluate: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false", index=True) # set by panel / REPROCESS / SYNC / KB change; consumed by the Agent + last_attempt_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + ui_status: Mapped[str | None] = mapped_column(String(32), nullable=True, index=True) # human-facing state (never a silent product) + reason_code: Mapped[str | None] = mapped_column(String(48), nullable=True) detected_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) diff --git a/central/app/version.py b/central/app/version.py index 18c7a88..63781c1 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.4" +GTI_FISCAL_VERSION = "1.3.5" 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 3d37682..b6c29e2 100644 --- a/central/app/web.py +++ b/central/app/web.py @@ -22,6 +22,7 @@ from .panel_presentation import format_ui_datetime from .models import AdminUser, Agent, AgentEnrollment, AgentRelease, AgentSnapshot, BlockedProduct, FiscalContextEntry, Installation, Job, ProductEnrichment, ProductFieldAudit, RtcCandidate, RtcWriteAudit, Tenant from .jobs import create_job from .audit import record_audit +from .enrichment import kb as kb_mod, status as enrichment_status from .enrichment.observability import coverage_metrics, kb_summary, official_sources, provider_sources @@ -158,10 +159,10 @@ def router(session_factory): return {key: items[0] for key, items in parse_qs((await request.body()).decode()).items() if items} @routes.get("/", response_class=HTMLResponse) - def dashboard(request: Request): + def dashboard(request: Request, kb_q: str | None = None): 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), "kb": kb_summary(session)}) + try: return render(request, "dashboard.html", {"summary": dashboard_summary(session), "auth": auth_data, "kb_q": kb_q or "", "is_admin": _is_privileged(auth_data[1]), "csrf_token": request.cookies.get("gti_csrf", ""), "sources": provider_sources(session), "official": official_sources(session), "enrich_metrics": coverage_metrics(session), "kb": kb_summary(session, kb_q)}) finally: session.close() @routes.get("/agents", response_class=HTMLResponse) @@ -302,10 +303,18 @@ def router(session_factory): row = session.scalar(select(ProductEnrichment).where(ProductEnrichment.installation_id == installation.id, ProductEnrichment.product_code == enrich_code.strip()[:64])) audits = session.scalars(select(ProductFieldAudit).where(ProductFieldAudit.installation_id == installation.id, ProductFieldAudit.product_code == enrich_code.strip()[:64]).order_by(ProductFieldAudit.id.desc()).limit(20)).all() if row else [] lookup = {"found": row is not None, "row": row, "audits": audits} + tracked = [] + for r in session.scalars(select(ProductEnrichment).where(ProductEnrichment.installation_id == installation.id).order_by(ProductEnrichment.id.desc()).limit(15)): + fields = ((r.decisions or {}).get("fields") or {}) + ncm = fields.get("CLASSIFICACAO_FISCAL") or {} + tracked.append({"code": r.product_code, "description": r.description, "gtin": r.gtin, "ui_status": r.ui_status or "AGUARDANDO_EVIDENCIA", "label": enrichment_status.LABELS.get(r.ui_status or "AGUARDANDO_EVIDENCIA", ""), + "reason": r.reason_code, "detected": r.detected_at, "last_attempt": r.last_attempt_at, "next_retry": r.next_attempt_at, "kb_status": ncm.get("kb_status"), "ncm_candidate": ncm.get("value"), + "cest_candidate": (fields.get("CEST") or {}).get("value"), "ipi_candidate": (fields.get("IPI") or {}).get("value"), "applied": r.fields_applied or {}, "placeholder": ncm.get("technical_placeholder"), + "reevaluate": bool(r.reevaluate), "mode": installation.enrichment_mode}) mode = installation.enrichment_mode waiting = not installation.features_configured return {"configured": bool(installation.features_configured), "waiting": waiting, "corrector_enabled": bool(installation.corrector_enabled), "enrichment_mode": mode, - "mode_label": ENRICHMENT_MODE_LABELS.get(mode, mode), "counts": counts, "total": sum(counts.values()), "applied": applied, "lookup": lookup, + "mode_label": ENRICHMENT_MODE_LABELS.get(mode, mode), "counts": counts, "total": sum(counts.values()), "applied": applied, "lookup": lookup, "tracked": tracked, "state_label": "Aguardando configuração de funcionalidades" if waiting else ENRICHMENT_MODE_LABELS.get(mode, mode)} @routes.post("/agents/{agent_id}/features") @@ -335,6 +344,54 @@ def router(session_factory): return RedirectResponse(f"/agents/{agent_id}", status_code=303) finally: session.close() + @routes.post("/agents/{agent_id}/enrichment/reevaluate") + async def reevaluate_enrichment(request: Request, agent_id: int): + """Explicit, idempotent 'evaluate again' for one product (code=...) or every tracked product that is not enriched yet. KB/cache first; providers only when really needed.""" + session, auth_data, redirect = page_auth(request) + if redirect: return redirect + try: + values = await form_values(request); web_session, admin = auth_data + require_csrf(web_session, values.get("csrf_token")) + agent, installation = _rtc_target(session, agent_id) + if not agent or not installation: return HTMLResponse("Agent not found", status_code=404) + if not installation.features_configured or installation.enrichment_mode == "DISABLED": + return HTMLResponse("O enriquecimento está desativado nesta instalação (Funcionalidades).", status_code=409) + code = (values.get("code") or "").strip()[:64] + n = enrichment_status.mark_for_reevaluation(session, installation.id, codes=[code] if code else None) + record_audit(session, admin_user_id=admin.id, tenant_id=installation.tenant_id, action="ENRICHMENT_REEVALUATE", target_type="installation", target_id=installation.installation_id, + context={"product": code or "ALL_UNRESOLVED", "flagged": n, "operator": admin.display_name or admin.username}) + session.commit() + return RedirectResponse(f"/agents/{agent_id}#enriquecimento", status_code=303) + finally: session.close() + + @routes.post("/kb/attest") + async def kb_attest(request: Request): + """A human fiscal decision (ADMIN only): choose which candidate NCM is right for a GTIN. Only an existing, vigente candidate can be chosen; the others are recorded as rejected.""" + session, auth_data, redirect = page_auth(request) + if redirect: return redirect + try: + values = await form_values(request); web_session, admin = auth_data + require_csrf(web_session, values.get("csrf_token")) + if not _is_privileged(admin): + record_audit(session, admin_user_id=admin.id, tenant_id=None, action="KB_ATTEST_DENIED", target_type="kb_product", target_id=values.get("gtin") or "", context={"role": admin.external_role or admin.auth_source}, severity="WARNING") + session.commit() + return HTMLResponse("Ação restrita a administradores do GTI Gerenciador (perfil ADMIN).", status_code=403) + gtin, ncm = (values.get("gtin") or "").strip(), (values.get("ncm") or "").strip() + from .enrichment import history + current = history.official_sets(session) + try: + result = kb_mod.attest(session, gtin, ncm, admin.display_name or admin.username, lambda v: v in current, note=(values.get("note") or "")[:200]) + except ValueError as error: + return HTMLResponse(f"Decisão recusada: {error}", status_code=422) + record_audit(session, admin_user_id=admin.id, tenant_id=None, action="KB_NCM_ATTESTED", target_type="kb_product", target_id=gtin, + context={"ncm": ncm, "status": result["status"], "families": result["families"], "operator": admin.display_name or admin.username, "role": admin.external_role or admin.auth_source}, severity="WARNING") + for inst in session.scalars(select(Installation).where(Installation.enrichment_mode != "DISABLED", Installation.features_configured.is_(True))): + enrichment_status.mark_for_reevaluation(session, inst.id, gtin=gtin) + enrichment_status.mark_for_reevaluation(session, inst.id) # products identified by name are found through the KB aliases + session.commit() + return RedirectResponse("/#kb-resumo", status_code=303) + finally: session.close() + def _ops_view(session, installation, agent): """Processing status shown on the Agent detail (no clutter: 5 facts).""" last_job = session.scalar(select(Job).where(Job.agent_id == agent.id).order_by(Job.id.desc()).limit(1)) diff --git a/central/pyproject.toml b/central/pyproject.toml index 193eff8..24b8bc3 100644 --- a/central/pyproject.toml +++ b/central/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "gti-fiscal-hub" -version = "1.3.4" +version = "1.3.5" requires-python = ">=3.10" dependencies = [ "fastapi>=0.115,<1", diff --git a/central/templates/agent_detail.html b/central/templates/agent_detail.html index 5dc2377..411d401 100644 --- a/central/templates/agent_detail.html +++ b/central/templates/agent_detail.html @@ -9,14 +9,22 @@ Reprocessar tudo {% if ops.enabled %}
{% else %}
{% endif %}
Processamento
{{ "Ativo" if ops.enabled else "Pausado" }} · sincronização automática a cada {{ ops.interval_minutes }} min
Última sincronização
{{ ops.last_sync or "ainda não sincronizado" }}{% if ops.last_changed is not none %} · {{ ops.last_changed }} produto(s) alterado(s){% endif %}
Sincronização agendada
{{ ops.next_sync or ("aguardando o primeiro sync" if ops.enabled else "pausado") }}
{% if ops.rtc %}
Reforma Tributária
{{ ops.rtc.label }}
{% endif %}{% if ops.rtc and ops.rtc.enabled %}{% if ops.rtc.products and not ops.rtc.products.applicable %}
cClassTrib
{{ ops.rtc.products.label }}
{% elif ops.rtc.products %}
Produtos RTC
{{ ops.rtc.products.headline }}
Total {{ ops.rtc.products.total }} · com cClassTrib válido {{ ops.rtc.products.valid }} · prontos para preenchimento {{ ops.rtc.products.ready }} · ambíguos/bloqueados {{ ops.rtc.products.blocked }}
{% endif %} -
Preenchimento automático
{{ ops.rtc.apply.state }}{% if ops.rtc.apply.applied_last_job %} · {{ ops.rtc.apply.applied_last_job }} preenchido(s) no último job{% endif %} · Controle e candidatos cClassTrib
{% endif %}
Último job
{% if ops.last_job %}{{ ops.last_job.kind_label }} · {{ ops.last_job.status_label }}{% else %}nenhum{% endif %}
-

Funcionalidades

Decididas no Hub para esta instalação — o Agent não tem chave local.

+
Preenchimento cClassTrib (RTC) — não é o Enriquecimento
{{ ops.rtc.apply.state }}{% if ops.rtc.apply.applied_last_job %} · {{ ops.rtc.apply.applied_last_job }} preenchido(s) no último job{% endif %} · Controle e candidatos cClassTrib
{% endif %}
Último job
{% if ops.last_job %}{{ ops.last_job.kind_label }} · {{ ops.last_job.status_label }}{% else %}nenhum{% endif %}
+

Funcionalidades (Enriquecimento de produtos novos)

Decididas no Hub para esta instalação — o Agent não tem chave local. O Enriquecimento de produtos novos NÃO é o Preenchimento cClassTrib (RTC), que tem controle próprio em Controle e candidatos cClassTrib.

{% if features.waiting %}

Aguardando configuração de funcionalidades: esta instalação ainda não processa nada.

{% endif %}
Corretor fiscal
{{ "Ativo" if features.corrector_enabled else "Desativado" }}
Enriquecimento de produtos novos
{{ features.state_label }}
{% if is_admin %}
{% else %}

Somente administradores alteram as funcionalidades.

{% endif %}

Enriquecimento

Produtos novos cadastrados no SGV; o catálogo histórico nunca é enviado a fontes externas.

Produtos acompanhados
{{ features.total }}
Campos preenchidos
{{ features.applied }}
{% for status, n in features.counts.items() %}
{{ status }}
{{ n }}
{% endfor %}
+
{% if features.tracked %} +{% for r in features.tracked %} + + + +{% endfor %}
ProdutoEstadoMotivo / candidatosÚltima tentativa
{{ r.code }} {{ r.description }}
EAN: {{ r.gtin or "não informado" }} · detectado {{ r.detected.strftime("%d/%m %H:%M") if r.detected else "—" }}		{{ r.ui_status }}{% if r.placeholder %}
NCM provisório {{ r.placeholder }}{% endif %}		{{ r.label }}{% if r.kb_status %} · base: {{ r.kb_status }}{% endif %}{% if r.ncm_candidate %} · NCM candidato {{ r.ncm_candidate }}{% endif %}{% if r.cest_candidate %} · CEST {{ r.cest_candidate }}{% endif %}{% if r.ipi_candidate %} · IPI {{ r.ipi_candidate }}{% endif %}{% if r.applied %} · gravado: {{ r.applied.keys()|join(", ") }}{% endif %}{{ r.last_attempt.strftime("%d/%m %H:%M:%S") if r.last_attempt else "—" }}{% if r.next_retry %}
próximo retry {{ r.next_retry.strftime("%d/%m %H:%M") }}{% endif %}{% if r.reevaluate %}
reavaliação pedida{% endif %}		{% if features.enrichment_mode != "DISABLED" and r.ui_status not in ["ENRIQUECIDO","HUMAN_OVERRIDE"] %}
{% endif %}
+{% if features.enrichment_mode != "DISABLED" %}
Só reavalia (base/cache primeiro); preço editado não é produto novo.
{% endif %} +{% else %}

Nenhum produto novo acompanhado ainda. Só entram produtos cadastrados depois da ativação.

{% endif %}
{% if features.lookup %}{% if features.lookup.found %}
Produto
{{ features.lookup.row.product_code }} · {{ features.lookup.row.description }}
GTIN
{{ features.lookup.row.gtin or "sem GTIN válido" }}
Estado
{{ features.lookup.row.status }}{% if features.lookup.row.decisions.state %} · {{ features.lookup.row.decisions.state }}{% endif %}
{% for a in features.lookup.audits %}
{{ a.field }}: {{ a.old_value or "vazio" }} → {{ a.new_value or "—" }}{{ a.outcome }} · regra {{ a.rule or "—" }} · base {{ a.kb_status or "—" }}{% if a.evidence %} · {{ a.evidence|join(", ") }}{% endif %}
{% endfor %}{% else %}

Produto sem registro de enriquecimento.

{% endif %}{% endif %}

Contexto da empresa

Cliente
{{ detail.client or "Não informado" }}
Nome de exibição
{{ detail.display_name or "Não informado" }}
Hostname
{{ detail.hostname or "Não informado" }}
Empresa
{{ detail.company_name or detail.company_code or "Não informado" }}
UF
{{ detail.company_uf or "Não informado" }}
CRT
{{ detail.company_crt or "Não informado" }}
Versão do agent
{{ detail.agent_version_display }}
Último heartbeat
{{ detail.last_seen_display }} · {{ detail.last_seen_age }}
Firebird
{{ detail.firebird_version or "Não informado" }} · ODS {{ detail.firebird_ods or "Não informado" }}
IDs técnicos
Agent {{ detail.technical_ids.agent_id }} · instalação {{ detail.technical_ids.installation_db_id }}
diff --git a/central/templates/dashboard.html b/central/templates/dashboard.html index f972982..9e85ea6 100644 --- a/central/templates/dashboard.html +++ b/central/templates/dashboard.html @@ -30,6 +30,10 @@

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.search %}
{{ c.gtin }} {{ c.description }} · {{ c.status }} +{% for x in c.candidates %}NCM {{ x.ncm }} — {{ x.official_text or "fora do dataset oficial vigente" }} ({{ x.clients|length }} cliente(s), {{ x.documents }} doc.{% if x.providers %}, {{ x.providers|join("/") }}{% endif %}{% if x.attested %}, decidido por {{ x.attested }}{% endif %}){% if is_admin and x.official_text and c.status != "VERIFIED_KB" %} +
{% endif %}
{% endfor %}
{% endfor %} {% 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_reeval.py b/central/tests/test_enrichment_reeval.py new file mode 100644 index 0000000..496d7c7 --- /dev/null +++ b/central/tests/test_enrichment_reeval.py @@ -0,0 +1,141 @@ +"""Visible status, re-evaluation (panel / Reprocessar / Sincronizar / KB decision), operator attestation, deterministic name-only match, RTC != Enrichment.""" +from sqlalchemy import select +from sqlalchemy.orm import Session + +from central.app.enrichment import api as enrich_api, history, kb, status +from central.app.enrichment.quota import ProviderGateway +from central.app.enrichment.resolver import resolve +from central.app.jobs import create_job +from central.app.models import Agent, AuditEvent, Installation, KbKnowledge, KbProduct, KbPromotionAudit, ProductEnrichment +from central.tests.test_control_panel_pages import _panel_client +from central.tests.test_enrichment_history import G, run as seed_history, staging +from central.tests.test_product_enrichment_api import H, _second_agent, resolve as api_resolve, setup +from central.tests.test_product_enrichment_resolver import COSMOS, Fake, G1, env, gtin # noqa: F401 +from central.tests.test_rtc_security import _become, _post + + +def _kb_conflict(env): + seed_history(env, [staging("RINCAO", G1, "19059090", years={2024: 900, 2025: 900}, desc="COCA LATA 350 ML"), staging("UNIAO", G1, "22021000", years={2024: 20, 2025: 13}, desc="REFRIGERANTE COCA COLA LATA 350 ML")]) + + +def test_attestation_resolves_a_conflict_with_audit_and_only_among_candidates(env): + _kb_conflict(env) + cur = history.official_sets(env) + p = env.scalar(select(KbProduct).where(KbProduct.gtin == G)) + assert kb.evaluate(env, p, "ncm", lambda v: v in cur)["status"] == kb.CONFLICT + for bad, why in (("21069090", "NCM_IS_NOT_A_CANDIDATE_OF_THIS_GTIN"), ("22030000x", "NCM_IS_NOT_A_CANDIDATE_OF_THIS_GTIN")): + try: + kb.attest(env, G, bad, "Rafael", lambda v: v in cur) + assert False + except ValueError as error: + assert str(error) == why + res = kb.attest(env, G, "22021000", "Rafael", lambda v: v in cur, note="lata 350ml = refrigerante") + assert res["status"] == kb.VERIFIED_KB and res["value"] == "22021000" and set(res["families"]) == {"OPERATOR_ATTESTATION", "CLIENT_DATA"} + audit = env.scalars(select(KbPromotionAudit)).one() + assert audit.sources["operator"] == "Rafael" and audit.new_status == "VERIFIED_KB" + rejected = env.scalar(select(KbKnowledge).where(KbKnowledge.knowledge_key == "ncm", KbKnowledge.value == "19059090")) + assert rejected.status == kb.STALE and rejected.evidence["info"]["historical_state"] == "REJECTED_BY_OPERATOR" + + +def test_name_only_match_is_deterministic_and_conservative(env): + _kb_conflict(env) + cur = history.official_sets(env) + kb.attest(env, G, "22021000", "Rafael", lambda v: v in cur) + zero = gtin("789490070001") + seed_history(env, [staging("UNIAO", zero, "22021000", years={2024: 20, 2025: 13}, desc="REFRIGERANTE COCA COLA ZERO LATA 350 ML", code="9")]) + env.commit() + r = resolve(env, ProviderGateway(env, []), gtin_raw="", description="COCA-COLA ORIGINAL LATA 350 ML", current={"CLASSIFICACAO_FISCAL": "00000000"}) + assert r["state"] == "EXACT_KB_NAME_VERIFIED" and r["fields"]["CLASSIFICACAO_FISCAL"]["status"] == "AUTO_FILL_SAFE" and r["fields"]["CLASSIFICACAO_FISCAL"]["value"] == "22021000" + ambiguous = resolve(env, ProviderGateway(env, []), gtin_raw="", description="COCA COLA ZERO LATA 350 ML", current={}) + assert ambiguous["fields"]["CLASSIFICACAO_FISCAL"]["status"] != "AUTO_FILL_SAFE" # ZERO is not a neutral extra word: no cross-match + generic = resolve(env, ProviderGateway(env, []), gtin_raw="", description="REFRIGERANTE", current={}) + assert generic["state"] == "AMBIGUOUS" + assert not kb.alias_match(frozenset({"COCA", "COLA", "LATA", "350", "ML"}), frozenset({"COCA", "COLA", "ZERO", "LATA", "350", "ML"})) + assert kb.alias_match(frozenset({"COCA", "COLA", "LATA", "350", "ML"}), frozenset({"REFRIGERANTE", "COCA", "COLA", "LATA", "350", "ML"})) + + +def test_every_tracked_product_has_a_visible_status(monkeypatch): + client, engine = setup(monkeypatch) + out = api_resolve(client, "N1", "", "COCA-COLA ORIGINAL LATA 350 ML", {"CLASSIFICACAO_FISCAL": "00000000"}) + with Session(engine) as s: + row = s.scalars(select(ProductEnrichment)).one() + assert row.ui_status == "EAN_AUSENTE" and row.reason_code == "NAME_ONLY_NO_VERIFIED_MATCH" and row.last_attempt_at is not None + assert out["fields"]["CLASSIFICACAO_FISCAL"]["technical_placeholder"] == "00000000" + generic = api_resolve(client, "N2", "", "REFRIGERANTE", {}) + invalid = api_resolve(client, "N3", "7891000100104", "PRODUTO X", {}) + with Session(engine) as s: + st = {r.product_code: r.ui_status for r in s.scalars(select(ProductEnrichment))} + assert st["N2"] == "NOME_AMBIGUO" and st["N3"] == "EAN_INVALIDO" + assert all(status.LABELS.get(v) for v in st.values()) + from central.tests.test_product_enrichment_resolver import G1 as g + done = api_resolve(client, "N4", g, "REFRIGERANTE COLA LATA 350ML", {}) + with Session(engine) as s: + assert s.scalars(select(ProductEnrichment).where(ProductEnrichment.product_code == "N4")).one().ui_status == "PRONTO_PARA_PREENCHER" + client.post("/v1/products/enrichment/applied", json={"product_code": "N4", "fields": [{"field": "CLASSIFICACAO_FISCAL", "outcome": "APPLIED", "old": "00000000", "new": "22021000"}]}, headers=H) + client.post("/v1/products/enrichment/applied", json={"product_code": "N1", "fields": [{"field": "CLASSIFICACAO_FISCAL", "outcome": "ABORTED", "reason": "STALE_CONFLICT"}]}, headers=H) + with Session(engine) as s: + st = {r.product_code: (r.ui_status, r.reason_code) for r in s.scalars(select(ProductEnrichment))} + assert st["N4"][0] == "ENRIQUECIDO" and st["N1"] == ("WRITE_FAILED", "STALE_CONFLICT") + + +def test_reprocess_sync_and_panel_flag_only_unresolved_products_for_reevaluation(monkeypatch): + client, engine = setup(monkeypatch) + api_resolve(client, "U1", "", "COCA-COLA ORIGINAL LATA 350 ML", {"CLASSIFICACAO_FISCAL": "00000000"}) + api_resolve(client, "E1", G1[:0] or gtin("789100010010"), "REFRIGERANTE COLA LATA 350ML", {}) + client.post("/v1/products/enrichment/applied", json={"product_code": "E1", "fields": [{"field": "CLASSIFICACAO_FISCAL", "outcome": "APPLIED", "old": None, "new": "22021000"}]}, headers=H) + assert client.get("/v1/products/enrichment/pending", headers=H).json() == {"codes": []} + with Session(engine) as s: + agent = s.scalars(select(Agent)).first() + for kind in ("REPROCESS_ALL", "SYNC_INCREMENTAL"): + create_job(s, agent, kind, {}) + assert s.scalars(select(ProductEnrichment).where(ProductEnrichment.reevaluate.is_(True))).all()[0].product_code == "U1" + for r in s.scalars(select(ProductEnrichment)): + r.reevaluate = False + s.commit() + for j in s.scalars(select(__import__("central.app.models", fromlist=["Job"]).Job)): + j.status = "COMPLETED" + s.commit() + with Session(engine) as s: + create_job(s, s.scalars(select(Agent)).first(), "REPROCESS_PRODUCTS", {"codigos": ["U1"]}) + assert client.get("/v1/products/enrichment/pending", headers=H).json() == {"codes": ["U1"]} # the enriched product E1 is never re-evaluated + api_resolve(client, "U1", "", "COCA-COLA ORIGINAL LATA 350 ML", {"CLASSIFICACAO_FISCAL": "00000000"}) + assert client.get("/v1/products/enrichment/pending", headers=H).json() == {"codes": []} # consumed by the new evaluation + + +def test_panel_reevaluate_and_attest_routes_are_csrf_admin_and_audited(monkeypatch): + client, agent_id, _ = _panel_client() + with client._session_factory() as s: + seed_history(s, [staging("RINCAO", G1, "19059090", years={2024: 900, 2025: 900}, desc="COCA LATA 350 ML"), staging("UNIAO", G1, "22021000", years={2024: 20, 2025: 13}, desc="REFRIGERANTE COCA COLA LATA 350 ML")]) + from central.tests.test_product_enrichment_resolver import seed + seed(s) + i = s.scalars(select(Installation)).one() + i.enrichment_mode, i.features_configured = "ENABLED", True + s.commit() + _become(client, "PADRAO") + assert _post(client, "/kb/attest", gtin="0" + G1, ncm="22021000").status_code == 403 + _become(client, "ADMIN") + assert client.post("/kb/attest", data={"csrf_token": "bad", "gtin": "0" + G1, "ncm": "22021000"}, follow_redirects=False).status_code in (400, 403) + assert _post(client, "/kb/attest", gtin="0" + G1, ncm="21069090").status_code == 422 + assert _post(client, "/kb/attest", gtin="0" + G1, ncm="22021000").status_code == 303 + assert _post(client, f"/agents/{agent_id}/enrichment/reevaluate", code="").status_code == 303 + with client._session_factory() as s: + acts = [e.action for e in s.scalars(select(AuditEvent))] + assert "KB_NCM_ATTESTED" in acts and "KB_ATTEST_DENIED" in acts and "ENRICHMENT_REEVALUATE" in acts + page = client.get("/?kb_q=COCA COLA LATA 350 ML").text + assert "VERIFIED_KB" in page and "decidido por" in page + + +def test_rtc_mode_and_enrichment_mode_are_independent_switches_with_distinct_labels(): + client, agent_id, _ = _panel_client() + _become(client, "ADMIN") + assert _post(client, f"/agents/{agent_id}/features", enrichment_mode="DRY_RUN", corrector_enabled="1").status_code == 303 + with client._session_factory() as s: + i = s.scalars(select(Installation)).one() + assert i.enrichment_mode == "DRY_RUN" and i.rtc_write_mode == "DISABLED" + assert _post(client, f"/agents/{agent_id}/rtc-mode", mode="DRY_RUN").status_code == 303 + assert _post(client, f"/agents/{agent_id}/rtc-mode", mode="ENABLED", confirm="PREENCHER").status_code == 303 + with client._session_factory() as s: + i = s.scalars(select(Installation)).one() + assert i.rtc_write_mode == "ENABLED" and i.enrichment_mode == "DRY_RUN" # turning cClassTrib to Automatic never touches the enrichment + page = client.get(f"/agents/{agent_id}").text + assert "Preenchimento cClassTrib (RTC)" in page and "Enriquecimento de produtos novos" in page diff --git a/central/tests/test_health.py b/central/tests/test_health.py index 9c97b10..b77cafc 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.4" and body["api_compat"] == "1.0" and body["engine_version"] and body["decision_version"] + assert body["version"] == "1.3.5" 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/deploy/agent-windows/agent.env.template b/deploy/agent-windows/agent.env.template index d507aec..9845763 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.4 +AGENT_VERSION=1.3.5 diff --git a/deploy/agent-windows/gti-agent.ps1 b/deploy/agent-windows/gti-agent.ps1 index 9f4acd6..07b697d 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.4' + 'FISCAL_MODE=DRY_RUN', 'FIREBIRD_WRITES=0', 'POLL_INTERVAL_SECONDS=5', 'AGENT_VERSION=1.3.5' ) 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 595fab1..f65d269 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.4 +AGENT_VERSION=1.3.5 diff --git a/deploy/agent-windows/profiles/mmmodas.env.example b/deploy/agent-windows/profiles/mmmodas.env.example index ff37fbe..532d132 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.4 +AGENT_VERSION=1.3.5 diff --git a/deploy/agent-windows/profiles/rincao.env.example b/deploy/agent-windows/profiles/rincao.env.example index 8cfacbe..d38de38 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.4 +AGENT_VERSION=1.3.5 diff --git a/deploy/agent-windows/profiles/uniao.env.example b/deploy/agent-windows/profiles/uniao.env.example index 1afb70f..fd44620 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.4 +AGENT_VERSION=1.3.5 diff --git a/deploy/hub/.env.example b/deploy/hub/.env.example index 3c117b9..fe0474d 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.4 +GTI_FISCAL_VERSION=1.3.5 # 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 401587e..b56b7c1 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.4} + image: gti-fiscal-hub:${GTI_FISCAL_VERSION:-1.3.5} restart: unless-stopped depends_on: postgres: diff --git a/docs/product_enrichment.md b/docs/product_enrichment.md index 994207e..29ed7d2 100644 --- a/docs/product_enrichment.md +++ b/docs/product_enrichment.md @@ -117,3 +117,19 @@ The SGV refuses to save a product without NCM, so operators type a filler. An em Both tables are identical (parity test). The audit records `old = 00000000`. - Scope reminders: products created while the function was DISABLED / WAITING stay history (re-baseline); REPROCESS_ALL/SYNC do not push products through the enrichment (only NEW products after the baseline do); a product WITHOUT EAN is resolved by name only from a unique VERIFIED_KB match. + + +## Product-visible pipeline, re-evaluation and operator decision (1.3.5, migration 0031) +- **No silent product**: every tracked product has `ui_status` + `reason_code` (Agent detail, section *Funcionalidades*): ENRIQUECIDO, PRONTO_PARA_PREENCHER, SIMULADO, AGUARDANDO_EVIDENCIA, EAN_AUSENTE, EAN_INVALIDO, NOME_AMBIGUO, SEM_MATCH_DE_PRODUTO, + NCM_NAO_VERIFICADO, CONFLITO_DE_NCM, PROVIDER_SEM_QUOTA, AGUARDANDO_RETRY, HUMAN_OVERRIDE, TECHNICAL_PLACEHOLDER (note), WRITE_FAILED, VERIFICACAO_POS_WRITE_FALHOU, NADA_A_PREENCHER, with EAN, detection time, KB status, candidates, last attempt and next retry. +- **Re-evaluation** (`product_enrichments.reevaluate`): set by the panel (*Reavaliar* / *Reavaliar todos os pendentes*), by **Reprocessar / Sincronizar** (`create_job`, even when an equivalent job is already queued) and by a KB decision; the Agent polls + `GET /v1/products/enrichment/pending` every 10 s and re-runs the normal resolve for those tracked, not-yet-enriched products (KB/cache first; the Hub decides about providers; enriched / human-overridden products are never flagged). + Price edits are NOT a new product; *Reavaliar* is the explicit new evaluation. REPROCESS_ALL does not push the whole catalogue through the enrichment: only the tracked unresolved set. +- **Stale-write guard** (Agent): a write is refused (`STALE_CONFLICT`) when the field no longer holds the value the Hub decided on (a person typed something meanwhile). +- **Name-only** (`kb.alias_match`): the query token set equals an alias token set, or one contains the other and the difference is only neutral category nouns (REFRIGERANTE, BEBIDA...). ZERO / DIET / a flavour / another size are never neutral, so + "COCA COLA LATA 350 ML" = "REFRIGERANTE COCA COLA LATA 350 ML" but != "... ZERO ...". Writing by name still requires ONE matching KB product whose NCM is VERIFIED_KB. +- **Operator decision (ADMIN)**: `POST /kb/attest` (dashboard, *Base de conhecimento*: search by EAN or name, official NCM text next to each candidate, button *Confirmar este NCM*). Only an existing, vigente candidate can be chosen; the others become `STALE (REJECTED_BY_OPERATOR)`; + the family `OPERATOR_ATTESTATION` + the local evidence of that value = VERIFIED_KB with a promotion audit naming the operator; every affected tracked product is flagged for re-evaluation, so the write happens automatically right after the click. +- **RTC != Enrichment**: distinct labels on the Agent detail and a test that each switch never changes the other. +- **Supplier NF-e**: investigated, no independent supplier evidence exists in the five databases (entry-item tables C000022/C000075 are empty; C000061/C000062/NFCE are documents ISSUED by the client from its own master); no `SUPPLIER_NFE` family was created. +- **Free NCM sources re-checked (2026-09-26)**: Kodebar returns NCM only for medicines (CMED) and needs a key; DotCompany API is prepaid; Pesquisa GTIN has a free plan of 5 queries/MONTH (not integrated); OSCBR V3 NCM is unproven. The official GTIN->NCM source is the SEFAZ CCG web service, which needs the client's ICP-Brasil certificate (not used). diff --git a/docs/runs/enrichment_acceptance_dev.json b/docs/runs/enrichment_acceptance_dev.json new file mode 100644 index 0000000..b6eb9b2 --- /dev/null +++ b/docs/runs/enrichment_acceptance_dev.json @@ -0,0 +1,34 @@ +{ + "dev_rehearsal": true, + "info": { + "kb_before": "ROWS [('22021000', 'OBSERVED'), ('22030000', 'OBSERVED')]", + "status_before_decision": "EAN_AUSENTE|NAME_ONLY_NO_VERIFIED_MATCH", + "attest": "ATTEST VERIFIED_KB ['CLIENT_DATA', 'OPERATOR_ATTESTATION'] flagged 15", + "status_dry_run_after_decision": "SIMULADO|DRY_RUN_NO_WRITE", + "AC001_after": [ + "22021000", + null, + null, + "200004", + "COCA-COLA ORIGINAL LATA 350 ML" + ], + "audit": "CLASSIFICACAO_FISCAL:00000000>22021000:APPLIED:VERIFIED_KB", + "ean_latency_s": 2.39 + }, + "checks": { + "SILENT_FAILURE_FIXED_STATUS_VISIBLE": true, + "NCM_UNCHANGED_UNTIL_DECISION": true, + "ATTESTED_TO_VERIFIED_KB": true, + "DRY_RUN_SHOWS_CANDIDATE_ZERO_WRITES": true, + "NAME_ONLY_NCM_WRITTEN_BY_AGENT": true, + "CCLASSTRIB_PRESERVED": true, + "AUDIT_OLD_PLACEHOLDER_NEW_NCM": true, + "FINAL_STATUS_ENRIQUECIDO": true, + "SECOND_RUN_WRITES_0": true, + "ZERO_PROVIDER_REQUESTS_ON_KB_RESOLUTION": true, + "EAN_PATH_KB_HIT_WRITE": true, + "EXISTING_REAL_NCM_PRESERVED": true, + "HUMAN_OVERRIDE_PRESERVED": true + }, + "ALL_PASS": true +} \ No newline at end of file diff --git a/release/RELEASE_MANIFEST.json b/release/RELEASE_MANIFEST.json index 52107d9..63f5bb8 100644 --- a/release/RELEASE_MANIFEST.json +++ b/release/RELEASE_MANIFEST.json @@ -1,14 +1,14 @@ { "artifact": "RELEASE_MANIFEST", - "GTI_FISCAL_VERSION": "1.3.4", + "GTI_FISCAL_VERSION": "1.3.5", "generated_at": "2026-09-25T15:43:40+00:00", "hub": { - "version": "1.3.4", + "version": "1.3.5", "api_compat": "1.0", "engine_version": "0.3.0", "decision_version": "mission-04c-ipi-ex-split" }, - "agent_version": "1.3.4", + "agent_version": "1.3.5", "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": "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 + "alembic_head_in_repo": "0031_enrichment_reevaluate", + "applied_revision": "0031_enrichment_reevaluate", + "applied_revision_source": "fresh + upgrade(0030->0031) + downgrade/upgrade verified on the dev PostgreSQL (schema parity by pg_dump -s); production applies 0031 at deploy", + "count": 31 }, "datasets_active": [ { @@ -89,5 +89,5 @@ "record_count": 11103 } ], - "note": "1.3.4: Agent + Hub. Agent 1.3.4 replaces only an approved TECHNICAL_PLACEHOLDER (NCM 00000000) besides empty fields; real values are never overwritten." + "note": "1.3.5: Hub + Agent (re-evaluation polling, stale-write guard). Placeholder semantics arrived in 1.3.4." } \ No newline at end of file diff --git a/release/VERSION b/release/VERSION index d0149fe..80e78df 100644 --- a/release/VERSION +++ b/release/VERSION @@ -1 +1 @@ -1.3.4 +1.3.5 diff --git a/scripts/enrichment_acceptance_dev.py b/scripts/enrichment_acceptance_dev.py new file mode 100644 index 0000000..030c558 --- /dev/null +++ b/scripts/enrichment_acceptance_dev.py @@ -0,0 +1,142 @@ +"""Acceptance rehearsal of the production scenario on DEV (real HTTP Hub + real Agent worker + Firebird COPY): + + product WITHOUT EAN, NCM=00000000, cClassTrib typed by the operator -> visible status (no silent failure) -> an ADMIN decides the NCM of the conflicting GTIN in the KB -> + the flagged re-evaluation finds the unique verified product by NAME -> the Agent writes the NCM (typed write) -> read-back + audit -> second run writes 0. + Also: DRY_RUN shows the candidate without writing; EAN path (KB hit, zero provider requests) with latency; human override is preserved. + + docker compose up -d --build fiscal-hub && python scripts/enrichment_acceptance_dev.py +The ADMIN decision is applied through the same code as the panel route (kb.attest + mark_for_reevaluation), with operator name 'E2E-OPERATOR'. Writes go only to the disposable COPY. +""" +import json +import sys +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_kb_e2e as ke # noqa: E402 +import enrichment_local_hub_e2e as lh # noqa: E402 +from gti_fiscal_agent.client import HubClient # noqa: E402 + +base, ce = ke.base, ke.ce +GTIN = "7894900010015" + + +def insert(code, name, ncm, ean, cclasstrib): + conn = base.connect(True); cur = conn.cursor() + cur.execute("INSERT INTO C000025 (CODIGO, PRODUTO, CODBARRA, CLASSIFICACAO_FISCAL, CCLASSTRIB) VALUES (?, ?, ?, ?, ?)", (code, name, ean, ncm, cclasstrib)) + conn.commit(); conn.close() + + +def row(code): + conn = base.connect(False); cur = conn.cursor() + cur.execute("SELECT CLASSIFICACAO_FISCAL, CEST, IPI, CCLASSTRIB, PRODUTO FROM C000025 WHERE CODIGO=?", (code,)) + r = cur.fetchone(); conn.close() + return tuple(None if v is None else (v if isinstance(v, float) else str(v).strip()) for v in r) if r else None + + +def ui(code, inst): + return lh.sql(f"select ui_status||'|'||coalesce(reason_code,'') from product_enrichments where product_code='{code}' and installation_id=(select id from installations where installation_id='{inst}')") + + +def main(): + checks, info = {}, {} + base.restore_copy() + c1 = HubClient(lh.HUB, lh.TOKEN) + inst = c1.identity()["installation_id"] + lh.set_features(inst, "DISABLED") + lh.sql(f"delete from product_field_audits where installation_id=(select id from installations where installation_id='{inst}')") + lh.sql(f"delete from product_enrichments where installation_id=(select id from installations where installation_id='{inst}')") + # start from the real situation: the KB has a CONFLICT for the Coca-Cola can (RINCAO x UNIAO), nothing verified + state = ke.exec_hub(f''' +from sqlalchemy import select +from central.app.db import SessionLocal +from central.app.enrichment import kb +from central.app.models import KbKnowledge, KbProduct +with SessionLocal() as s: + p = s.scalar(select(KbProduct).where(KbProduct.gtin == "0{GTIN}")) + rows = s.scalars(select(KbKnowledge).where(KbKnowledge.kb_product_id == p.id, KbKnowledge.knowledge_key == "ncm")).all() + for r in rows: # reset any earlier rehearsal: drop attestation + ev = dict(r.evidence or {{}}); fam = dict(ev.get("families") or {{}}); fam.pop("OPERATOR_ATTESTATION", None); ev["families"] = fam; r.evidence = ev; r.status = "OBSERVED"; r.validated_at = None + s.commit() + print("ROWS", sorted((r.value, r.status) for r in rows)) +''') + info["kb_before"] = state + w, stop = ke.worker_for(c1, "acc_dev") + lh.set_features(inst, "DRY_RUN"); w._features_at = -999; time.sleep(6) + # 1) DRY_RUN: product without EAN, placeholder NCM + insert("AC001", "COCA-COLA ORIGINAL LATA 350 ML", "00000000", None, "200004") + time.sleep(12) + info["status_before_decision"] = ui("AC001", inst) + checks["SILENT_FAILURE_FIXED_STATUS_VISIBLE"] = info["status_before_decision"].startswith("EAN_AUSENTE") + checks["NCM_UNCHANGED_UNTIL_DECISION"] = row("AC001")[0] == "00000000" + # 2) ADMIN decision in the KB (same code path as the panel route) -> flagged re-evaluation, still DRY_RUN + out = ke.exec_hub(f''' +from central.app.db import SessionLocal +from central.app.enrichment import history, kb, status +from central.app.models import Installation +from sqlalchemy import select +with SessionLocal() as s: + cur = history.official_sets(s) + r = kb.attest(s, "0{GTIN}", "22021000", "E2E-OPERATOR", lambda v: v in cur, note="E2E rehearsal") + n = 0 + for i in s.scalars(select(Installation).where(Installation.enrichment_mode != "DISABLED", Installation.features_configured.is_(True))): + n += status.mark_for_reevaluation(s, i.id) + s.commit() + print("ATTEST", r["status"], r["families"], "flagged", n) +''') + info["attest"] = out + checks["ATTESTED_TO_VERIFIED_KB"] = "VERIFIED_KB" in out + t0 = time.monotonic() + while time.monotonic() - t0 < 40 and not ui("AC001", inst).startswith("SIMULADO"): + time.sleep(1) + info["status_dry_run_after_decision"] = ui("AC001", inst) + checks["DRY_RUN_SHOWS_CANDIDATE_ZERO_WRITES"] = ui("AC001", inst).startswith("SIMULADO") and row("AC001")[0] == "00000000" + # 3) AUTOMATIC: reevaluate (like the panel button) -> written by the Agent + lh.set_features(inst, "ENABLED"); w._features_at = -999; w._auth_cache = (None, 0.0); time.sleep(3) + lh.sql(f"update product_enrichments set reevaluate=true where product_code='AC001' and installation_id=(select id from installations where installation_id='{inst}')") + t0 = time.monotonic() + while time.monotonic() - t0 < 45 and row("AC001")[0] == "00000000": + time.sleep(0.5) + time.sleep(4) + r = row("AC001") + info["AC001_after"] = r + checks["NAME_ONLY_NCM_WRITTEN_BY_AGENT"] = r[0] == "22021000" + checks["CCLASSTRIB_PRESERVED"] = r[3] == "200004" + audit = lh.sql(f"select field||':'||coalesce(old_value,'')||'>'||coalesce(new_value,'')||':'||outcome||':'||coalesce(kb_status,'') from product_field_audits where product_code='AC001' and outcome='APPLIED' order by id") + info["audit"] = audit + checks["AUDIT_OLD_PLACEHOLDER_NEW_NCM"] = "CLASSIFICACAO_FISCAL:00000000>22021000:APPLIED" in audit + checks["FINAL_STATUS_ENRIQUECIDO"] = ui("AC001", inst).startswith("ENRIQUECIDO") + # 4) second run writes 0, zero provider requests + before_req, before_writes = ke.requests_total(), int(lh.sql("select count(*) from product_field_audits where outcome='APPLIED'")) + lh.sql(f"update product_enrichments set reevaluate=true where product_code='AC001' and installation_id=(select id from installations where installation_id='{inst}')") + time.sleep(25) + checks["SECOND_RUN_WRITES_0"] = int(lh.sql("select count(*) from product_field_audits where outcome='APPLIED'")) == before_writes and row("AC001")[0] == "22021000" + checks["ZERO_PROVIDER_REQUESTS_ON_KB_RESOLUTION"] = ke.requests_total() == before_req + # 5) EAN path on the now-verified GTIN: latency + zero provider requests + t0 = time.monotonic() + insert("AC002", "COCA-COLA ORIGINAL LATA 350 ML", "00000000", GTIN, "200004") + while time.monotonic() - t0 < 25 and row("AC002")[0] == "00000000": + time.sleep(0.2) + info["ean_latency_s"] = round(time.monotonic() - t0, 2) + checks["EAN_PATH_KB_HIT_WRITE"] = row("AC002")[0] == "22021000" and ke.requests_total() == before_req + # 6) real NCM preserved + human override preserved + insert("AC003", "COCA-COLA ORIGINAL LATA 350 ML", "19059090", GTIN, "200004"); time.sleep(10) + checks["EXISTING_REAL_NCM_PRESERVED"] = row("AC003")[0] == "19059090" + conn = base.connect(True); cur = conn.cursor(); cur.execute("UPDATE C000025 SET CLASSIFICACAO_FISCAL='19059090' WHERE CODIGO='AC001'"); conn.commit(); conn.close() + time.sleep(E_OVERRIDE_WAIT := 28) + lh.sql(f"update product_enrichments set reevaluate=true where product_code='AC001' and installation_id=(select id from installations where installation_id='{inst}')") + time.sleep(20) + checks["HUMAN_OVERRIDE_PRESERVED"] = row("AC001")[0] == "19059090" + stop.stop = True + result = {"dev_rehearsal": True, "info": info, "checks": checks, "ALL_PASS": all(checks.values())} + (ROOT / "docs" / "runs" / "enrichment_acceptance_dev.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 c6fcaf1..9db798a 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.4" + assert v == GTI_FISCAL_VERSION == "1.3.5" 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.4"' in (ROOT / "agent" / "scripts" / "run_agent_service.py").read_text(encoding="utf-8") + assert '"AGENT_VERSION", "1.3.5"' 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"))