gustavosantos/GTI_GERENCIADOR_FISCAL
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: product enrichment observability, lab KB study, benchmark, copy + local-Hub E2E, docs; 1.3.0

Browse Source 
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
gti
2026-09-25 19:26:13 -03:00
co-authored by Claude Sonnet 5
parent 5deebea14e
commit 4fc83a33d5
38 changed files with 1282 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
central/app/enrichment/resolver.py
+16 -3
View File
@@ -45,6 +45,17 @@ def ncm_is_official(session, code: str, today: date | None = None) -> bool:
return False
class _ScopedDatasets(DatasetService):
"""The Classifier only needs the ONE exact NCM row here; loading the whole 15k-row catalogue on every product would cost ~300 ms of the 5 s budget."""
def __init__(self, session, code: str):
super().__init__(session)
self.code = code
def active_entries(self):
return self.session.scalars(select(NcmEntry).join(Dataset).where(Dataset.kind == "NCM", Dataset.status == "ACTIVE", NcmEntry.code == self.code)).all()
def dataset_versions(session) -> dict:
return {d.kind: {"sha256": (d.sha256 or "")[:12], "version": d.version_label} for d in session.scalars(select(Dataset).where(Dataset.status == "ACTIVE"))}
@@ -75,11 +86,13 @@ def _fiscal_decisions(session, ncm_eval: dict, gtin_kb_verified: bool, current:
for name in ("CEST", "IPI"):
out[name] = _field(ALREADY if not _empty(current, name) else INSUFFICIENT, reason="depende de um NCM oficial válido")
return out
decision = Classifier(DatasetService(session)).classify({"current_ncm": ncm_for_engine, "description": description})
# CEST / IPI are deterministic functions of an OFFICIAL NCM: the NCM we are about to write (AUTO) or one already in the product (validated above against the official dataset)
ncm_solid = ncm["status"] == AUTO or (ncm["status"] == ALREADY and ncm_is_official(session, ncm_for_engine))
decision = Classifier(_ScopedDatasets(session, ncm_for_engine)).classify({"current_ncm": ncm_for_engine, "description": description})
cest = (decision["classification"].get("cest") or [])
if not _empty(current, "CEST"):
out["CEST"] = _field(ALREADY, reason="campo já preenchido")
elif decision["cest_status"] == "CEST_CONFIRMED" and len(cest) == 1 and ncm["status"] == AUTO:
elif decision["cest_status"] == "CEST_CONFIRMED" and len(cest) == 1 and ncm_solid:
out["CEST"] = _field(AUTO, normalize_code(cest[0]), "CEST_OFFICIAL_UNIQUE_MAPPING", ncm["kb_status"], [f"NCM {ncm_for_engine}"])
elif decision["cest_status"] == "NO_CEST_MAPPING_FOUND":
out["CEST"] = _field(NA, reason="NCM sem mapeamento de CEST no dataset oficial")
@@ -88,7 +101,7 @@ def _fiscal_decisions(session, ncm_eval: dict, gtin_kb_verified: bool, current:
ipi = decision["classification"].get("ipi") or {}
if not _empty(current, "IPI"):
out["IPI"] = _field(ALREADY, reason="campo já preenchido")
elif decision["ipi_status"] == "IPI_RESOLVED" and ipi.get("rate_kind") == "NUMERIC" and ncm["status"] == AUTO:
elif decision["ipi_status"] == "IPI_RESOLVED" and ipi.get("rate_kind") == "NUMERIC" and ncm_solid:
out["IPI"] = _field(AUTO, str(ipi["rate_numeric"]), "IPI_TIPI_NUMERIC", ncm["kb_status"], [f"TIPI NCM {ncm_for_engine}"])
elif decision["ipi_status"] in {"IPI_RESOLVED"}:
out["IPI"] = _field(NA, reason="TIPI sem alíquota numérica (NT/branca)")