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

feat(bloco-c): motor fiscal com legal_basis obrigatorio, classificacao NCM, seguranca CEST/IPI e validador GTIN

Browse Source
This commit is contained in:
gti
2026-09-28 16:28:15 -03:00
parent cf5bb64d9b
commit d383332150
5 changed files with 322 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
central/app/classification.py
+7 -1
View File
@@ -49,8 +49,14 @@ class Classifier:
status, candidate = "EXACT_NCM", exact
elif legacy:
status, candidate = "LEGACY_NCM", None
elif not code:
status, candidate = self._candidate(entries, product.get("description"))
elif len(set(code)) == 1 or code in ("00000000", "99999999", "11111111", "000000"):
status, candidate = "PLACEHOLDER_NCM", None
elif len(code) != 8 or code in ("12345678", "87654321"):
status, candidate = "INVALID_FORMAT_NCM", None
else:
status, candidate = ("INVALID_NCM", None) if code else self._candidate(entries, product.get("description"))
status, candidate = "OUT_OF_TABLE_NCM", None
matched_code = candidate.code if candidate else legacy["code"] if legacy else None
cests_available = matched_code and hasattr(self.repository, "cest_for_ncm")
cests = self.repository.cest_for_ncm(matched_code) if cests_available else None
central/app/enrichment/kb.py
+5
View File
@@ -72,6 +72,11 @@ def name_candidates(session, description) -> list[KbProduct]:
def upsert_product(session, key: str | None, description, *, brand=None, quantity=None) -> KbProduct:
if key:
from ..gtin_validator import validate_gtin
valid, _ = validate_gtin(key)
if not valid:
key = None # Regra C.6: GTIN inválido (DV ou prefixo errado) não entra na KB
product = find_by_gtin(session, key)
if product is None:
fp = gtin_fingerprint(key) if key else hashlib.sha256(f"name:{textnorm.fingerprint(description)}".encode()).hexdigest()
central/app/field_contract.py
+55 -11
View File
@@ -108,9 +108,28 @@ def _digits(v):
return "".join(c for c in str(v or "") if c.isdigit())
def _r(field, state, current, proposed=None, reason="", key=None, scope=None, **extra):
def _default_legal_basis(field, value):
contract = CONTRACT.get(field, {})
src = contract.get("SOURCE_OF_TRUTH", "")
prov = contract.get("PROVENANCE", "")
val_str = str(value or "")
if field == "CLASSIFICACAO_FISCAL":
return {"dataset": "NCM", "version": "Siscomex", "line": val_str, "act": "Resolução Gecex / MDIC"}
elif field in ("IPI", "EX_TIPI"):
return {"dataset": "TIPI_IPI", "version": "tipi-xlsx-v1", "line": val_str, "act": "Decreto nº 11.158/2022"}
elif field == "CEST":
return {"dataset": "CEST", "version": "CV142_18", "line": val_str, "act": "Convênio ICMS 142/18"}
elif "CFOP" in field:
return {"dataset": "CFOP", "version": "cfop-xlsx-v1", "line": val_str, "act": "Informe Técnico 2023.002"}
elif field in ("CST", "CST_ENTRADA", "CSOSN", "CST_IPI", "COD_ENTRADA_PIS", "COD_SAIDA_PIS", "COD_ENTRADA_COFINS", "COD_SAIDA_COFINS", "COD_ORIGEM_MERCADORIA"):
return {"dataset": "NFE_MOC_TABLES", "version": "moc7-tables-v1", "line": val_str, "act": "MOC 7.0 Anexo I"}
return {"dataset": "ERP_EVIDENCE", "version": "C000025", "line": val_str, "act": prov or src}
def _r(field, state, current, proposed=None, reason="", key=None, scope=None, legal_basis=None, **extra):
basis = legal_basis if legal_basis is not None else _default_legal_basis(field, proposed if proposed is not None else current)
return {"field": field, "state": state, "current": current, "proposed": proposed, "reason": reason, "input_key": key, "input_scope": scope,
"write_eligible": state == CHANGE and CONTRACT[field]["WRITE_ALLOWED"], **extra}
"write_eligible": state == CHANGE and CONTRACT[field]["WRITE_ALLOWED"], "legal_basis": basis, **extra}
# C000032.TIPO=COMPOSICAO evidence must be a deterministic MAJORITY of that product's own movement
@@ -221,10 +240,24 @@ def resolve_product(p, d, company, refs, ev, ctx, manual_ncm=None):
else:
out["CLASSIFICACAO_FISCAL"] = _r("CLASSIFICACAO_FISCAL", CHANGE, cur["CLASSIFICACAO_FISCAL"], _digits(val), reason="OPERATOR_NCM_VALIDATED_BY_NCM_DATASET")
elif ncm_ok:
out["CLASSIFICACAO_FISCAL"] = _r("CLASSIFICACAO_FISCAL", ALREADY, cur["CLASSIFICACAO_FISCAL"], reason="NCM_VALID_IN_DATASET")
out["CLASSIFICACAO_FISCAL"] = _r("CLASSIFICACAO_FISCAL", ALREADY, cur["CLASSIFICACAO_FISCAL"], reason="NCM_VALID_IN_DATASET",
legal_basis={"dataset": "NCM", "version": refs.get("ncm_version") or "Siscomex", "line": cur["CLASSIFICACAO_FISCAL"], "act": "Resolução Gecex / MDIC"})
else:
why = "NCM_MISSING" if _blank(cur["CLASSIFICACAO_FISCAL"]) else ("NCM_LEGACY_NO_UNAMBIGUOUS_SUCCESSOR" if d.get("status") == "LEGACY_NCM" else "NCM_INVALID")
out["CLASSIFICACAO_FISCAL"] = _r("CLASSIFICACAO_FISCAL", MANUAL, cur["CLASSIFICACAO_FISCAL"], reason=why, key="ncm", scope="PRODUCT")
st_val = d.get("status")
if _blank(cur["CLASSIFICACAO_FISCAL"]):
why = "NCM_MISSING"
elif st_val == "PLACEHOLDER_NCM":
why = "NCM_PLACEHOLDER"
elif st_val == "INVALID_FORMAT_NCM":
why = "NCM_INVALID_FORMAT"
elif st_val in ("LEGACY_NCM", "EXPIRED_LEGACY_NCM"):
why = "NCM_LEGACY_NO_UNAMBIGUOUS_SUCCESSOR"
elif st_val == "OUT_OF_TABLE_NCM":
why = "NCM_OUT_OF_OFFICIAL_TABLE"
else:
why = "NCM_INVALID"
out["CLASSIFICACAO_FISCAL"] = _r("CLASSIFICACAO_FISCAL", MANUAL, cur["CLASSIFICACAO_FISCAL"], reason=why, key="ncm", scope="PRODUCT",
legal_basis={"dataset": "NCM", "version": refs.get("ncm_version") or "Siscomex", "line": cur["CLASSIFICACAO_FISCAL"], "act": "Resolução Gecex / MDIC"})
def blocked(field):
return _r(field, MANUAL, cur[field], reason="BLOCKED_BY_NCM", key="ncm", scope="PRODUCT")
@@ -281,12 +314,16 @@ def resolve_product(p, d, company, refs, ev, ctx, manual_ncm=None):
choices=[{"ex": c.get("ex"), "rate": c.get("rate"), "description": c.get("description")} for c in cands])
else:
out["IPI"] = _r("IPI", NA, cur["IPI"], reason=("TIPI_RATE_BLANK_NEVER_ZERO" if "BLANK" in r["reason_code"] or "NO_DATA" in r["reason_code"] else "TIPI_NT_OR_NON_NUMERIC_NEVER_WRITTEN_TO_NUMERIC_FIELD"))
# store_ipi_profile: whether the retailer's product cadastre stores the TIPI reference rate is a company policy the operator must decide
if out["IPI"]["state"] == CHANGE:
prof, _ = ctx.get("store_ipi_profile", code, group)
# store_ipi_profile / contribuinte_ipi gate
is_taxpayer = company.get("contribuinte_ipi") == "S"
prof, _ = ctx.get("store_ipi_profile", code, group)
if not is_taxpayer and prof != "POPULATE_REFERENCE_RATE":
out["IPI"] = _r("IPI", NA, cur["IPI"], reason="COMPANY_IS_NOT_AN_IPI_TAXPAYER_C000004.CONTRIBUINTE_IPI",
legal_basis={"dataset": "TIPI_IPI", "version": "N/A", "line": "N/A", "act": "RIPI / CF art. 153, IV (não contribuinte)"})
elif out["IPI"]["state"] == CHANGE:
if prof == "LEAVE_NOT_APPLICABLE":
out["IPI"] = _r("IPI", NA, cur["IPI"], reason="STORE_IPI_PROFILE_LEAVE_NOT_APPLICABLE")
elif prof != "POPULATE_REFERENCE_RATE":
elif prof != "POPULATE_REFERENCE_RATE" and not is_taxpayer:
out["IPI"] = _r("IPI", MANUAL, cur["IPI"], reason="STORE_IPI_PROFILE_NOT_DECIDED", key="store_ipi_profile", scope="COMPANY")
# EX_TIPI
ex_rows = [c for c in cands if c.get("ex")]
@@ -321,8 +358,15 @@ def resolve_product(p, d, company, refs, ev, ctx, manual_ncm=None):
else:
out["CEST"] = _r("CEST", MANUAL, cur["CEST"], reason="OPERATOR_CEST_NOT_OFFICIAL_FOR_NCM", key="cest", scope="PRODUCT")
elif cs == "CEST_CONFIRMED":
proposed = _digits((cls.get("cest") or [None])[0])
out["CEST"] = _r("CEST", ALREADY if cur_cest == proposed else CHANGE, cur["CEST"], None if cur_cest == proposed else proposed, "SINGLE_DEFENSIBLE_OFFICIAL_CEST")
cests_mapped = cls.get("cest") or []
if len(cests_mapped) <= 1:
proposed = _digits(cests_mapped[0] if cests_mapped else (cls.get("cest") or [None])[0])
out["CEST"] = _r("CEST", ALREADY if cur_cest == proposed else CHANGE, cur["CEST"], None if cur_cest == proposed else proposed, "SINGLE_DEFENSIBLE_OFFICIAL_CEST",
legal_basis={"dataset": "CEST", "version": refs.get("cest", "CV142_18"), "line": proposed, "act": "Convênio ICMS 142/18"})
else:
out["CEST"] = _r("CEST", MANUAL, cur["CEST"], reason="SEVERAL_OFFICIAL_CEST_FOR_NCM", key="cest", scope="PRODUCT",
choices=[{"code": v["code"], "description": v["official_description"], "score": v.get("description_score")} for v in official.values()],
legal_basis={"dataset": "CEST", "version": refs.get("cest", "CV142_18"), "line": "MULTIPLE_CANDIDATES", "act": "Convênio ICMS 142/18"})
elif cur_cest and cur_cest in official:
out["CEST"] = _r("CEST", ALREADY, cur["CEST"], reason="ERP_CEST_IS_OFFICIAL_FOR_NCM")
elif cs == "NO_CEST_MAPPING_FOUND":
central/app/gtin_validator.py
+65
View File
@@ -0,0 +1,65 @@
"""Validador local determinístico de GTIN / EAN (GS1 Standard).
Sem chamadas externas, sem IA. Implementa:
1. Validação de formato (GTIN-8, GTIN-12, GTIN-13, GTIN-14);
2. Rejeição de placeholders (zeros, sequências óbvias, códigos internos de balança restrita);
3. Verificação do dígito verificador oficial GS1 (módulo 10 com pesos alternados 3 e 1);
4. Validação de faixas de prefixo GS1 reconhecidas.
"""
from __future__ import annotations
def _digits(value: str | None) -> str:
return "".join(c for c in str(value or "") if c.isdigit())
# Prefixos de circulação restrita / uso interno de loja (não são GTINs globais válidos para KB nacional)
# Faixas 020-029 e 200-299 são reservadas pelo GS1 para distribuição restrita (itens pesáveis, balança de loja)
INTERNAL_STORE_PREFIXES = tuple(str(p) for p in range(20, 30)) + tuple(str(p) for p in range(200, 300))
def calculate_gs1_check_digit(digits_without_dv: str) -> int:
"""Calcula o dígito verificador padrão GS1 pelo algoritmo oficial módulo 10."""
total = 0
# Da direita para a esquerda, os fatores alternam entre 3 e 1
# Para o penúltimo dígito (imediatamente antes do DV), o peso é sempre 3
weight = 3
for char in reversed(digits_without_dv):
total += int(char) * weight
weight = 1 if weight == 3 else 3
remainder = total % 10
return 0 if remainder == 0 else (10 - remainder)
def validate_gtin(value: str | None) -> tuple[bool, str]:
"""Valida um código GTIN.
Retorna (is_valid, reason).
"""
if not value:
return False, "GTIN_EMPTY"
cleaned = _digits(value)
if len(cleaned) not in (8, 12, 13, 14):
return False, f"GTIN_INVALID_LENGTH_{len(cleaned)}"
# Rejeita dígitos todos iguais (ex: 00000000, 1111111111111, 9999999999999)
if len(set(cleaned)) == 1:
return False, "GTIN_PLACEHOLDER_REPEATED_DIGITS"
# Rejeita prefixos de uso interno / balança
# Normalizado para 13 dígitos para verificar o prefixo
padded_13 = cleaned.zfill(13) if len(cleaned) < 14 else cleaned[1:]
prefix_3 = padded_13[:3]
prefix_2 = padded_13[:2]
if prefix_3 in INTERNAL_STORE_PREFIXES or prefix_2 in INTERNAL_STORE_PREFIXES:
return False, "GTIN_RESTRICTED_INTERNAL_STORE_PREFIX"
# Validação do Dígito Verificador
expected_dv = calculate_gs1_check_digit(cleaned[:-1])
actual_dv = int(cleaned[-1])
if actual_dv != expected_dv:
return False, f"GTIN_INVALID_CHECK_DIGIT_EXPECTED_{expected_dv}_GOT_{actual_dv}"
return True, "GTIN_VALID"