diff --git a/central/app/classification.py b/central/app/classification.py index 9fbde75..11f4aa1 100644 --- a/central/app/classification.py +++ b/central/app/classification.py @@ -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 diff --git a/central/app/enrichment/kb.py b/central/app/enrichment/kb.py index c6516f8..d8143b1 100644 --- a/central/app/enrichment/kb.py +++ b/central/app/enrichment/kb.py @@ -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() diff --git a/central/app/field_contract.py b/central/app/field_contract.py index 4b44eff..cef9aec 100644 --- a/central/app/field_contract.py +++ b/central/app/field_contract.py @@ -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": diff --git a/central/app/gtin_validator.py b/central/app/gtin_validator.py new file mode 100644 index 0000000..5397fda --- /dev/null +++ b/central/app/gtin_validator.py @@ -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" diff --git a/central/tests/test_mission_06_motor_legal_basis.py b/central/tests/test_mission_06_motor_legal_basis.py new file mode 100644 index 0000000..6f427a6 --- /dev/null +++ b/central/tests/test_mission_06_motor_legal_basis.py @@ -0,0 +1,190 @@ +"""Missão 6 — Bloco C: Testes do Motor com Base Legal (NCM, CEST, IPI, Domínios e GTIN).""" +import pytest +from central.app.field_contract import ( + ALREADY, CHANGE, MANUAL, NA, WRITE_FIELDS, Context, resolve_product +) +from central.app.classification import Classifier +from central.app.gtin_validator import validate_gtin, calculate_gs1_check_digit + +REFS = { + "cfop_valid": {"5101", "5102", "5401", "5405", "1102", "1101", "1403", "2102"}, + "csosn": {"101", "102", "103", "201", "202", "203", "300", "400", "500", "900"}, + "cst_icms": {"00", "10", "20", "30", "40", "41", "50", "51", "60", "70", "90"}, + "cst_pis": {"01", "02", "03", "04", "05", "06", "07", "08", "09", "49", "98", "99"}, + "cst_cofins": {"01", "02", "03", "04", "05", "06", "07", "08", "09", "49", "98", "99"}, + "origem": set("012345678"), + "fcp_uf": {}, + "cbs_rate_2026": None, + "cclasstrib_count": 0, + "ncm_version": "Siscomex-2026-09-28", + "tipi_version": "tipi-xlsx-v1", + "cest_version": "CV142_18", +} + +CO_SIMPLES_NAO_IPI = {"uf": "BA", "crt": "1", "contribuinte_ipi": "N", "company_code": "000001"} +CO_NORMAL_COM_IPI = {"uf": "BA", "crt": "3", "contribuinte_ipi": "S", "company_code": "000001"} + +EV = {"recipes": set(), "composed": set(), "purchases": {}, "stock_entry": {}, "stock_in": {"000001": True}} + + +def make_product(**kw): + p = {f: None for f in WRITE_FIELDS} + p.update({ + "CODIGO": "000001", "PRODUTO": "CERVEJA LATA", "CODGRUPO": "000019", "TIPO": "VENDA", + "PROD_COZINHA": None, "PRODUTO_SERVICO": None, "CLASSIFICACAO_FISCAL": "22030000", + "CFOP": "5102", "CSOSN": "102", "CST": "000", "SITUACAO_TRIBUTARIA": "T", + "IAT": "A", "IPPT": "T", "INCIDENCIA_PISCOFINS": "NENHUMA", "PISCOFINS": "N" + }) + p.update(kw) + return p + + +def make_decision(**kw): + d = { + "status": "EXACT_NCM", "cest_status": "NO_CEST_MAPPING_FOUND", "ipi_status": "IPI_RESOLVED", + "ba_st_mva_status": "NO_MVA_MAPPING_FOUND", "dataset_versions": {}, + "classification": { + "ncm": "22030000", + "ipi": {"rate_kind": "NUMERIC", "rate_numeric": 3.9, "rate": "3.9"}, + "ipi_candidates": [], "cest": None, "cest_candidates": [], "ba_st_mva": None, "ba_st_mva_candidates": [] + } + } + cls = kw.pop("classification", {}) + d.update(kw) + d["classification"] = {**d["classification"], **cls} + return d + + +# ---------------- C.1: Legal Basis ---------------- +def test_c1_every_decision_contains_legal_basis(): + """C.1: field_contract grava legal_basis estruturado em cada decisão.""" + res = resolve_product(make_product(), make_decision(), CO_SIMPLES_NAO_IPI, REFS, EV, Context([])) + for field in WRITE_FIELDS: + decision = res[field] + assert "legal_basis" in decision, f"Campo {field} sem legal_basis" + lb = decision["legal_basis"] + assert isinstance(lb, dict) + assert "dataset" in lb or "source" in lb + assert "act" in lb + + +# ---------------- C.2: NCM 8 Produtos Reais Bloqueados ---------------- +def test_c2_ncm_classification_categories_and_8_blocked_products(): + """C.2: Classificar placeholder, inválido, fora da tabela e vencido. + Aplica aos 8 produtos reais bloqueados do Agent 2: + - 3 vencidos (02071400): 000016, 000254, 000349 + - 5 fora da tabela (02071200, 21069091, 03061300): 000061, 000062, 000098, 000172, 000174. + """ + class MockRepo: + def active_entries(self): + class Entry: + code = "22030000" + description = "Cervejas de malte" + return [Entry()] + + def legacy_ncm(self, code): + if code == "02071400": + return {"code": "02071400", "source": "Siscomex", "source_date": "2026-09-17", "reason": "Extinto", "successor_code": None} + return None + + def dataset_versions(self): + return {} + + clf = Classifier(MockRepo()) + + # 1. Placeholder + assert clf.classify({"current_ncm": "00000000"})["status"] == "PLACEHOLDER_NCM" + + # 2. Formato Inválido + assert clf.classify({"current_ncm": "12345678"})["status"] == "INVALID_FORMAT_NCM" + + # 3. Os 3 produtos reais vencidos (02071400) + for code, desc in [("000016", "CALDO DE SURURU"), ("000254", "CHESFRANGO"), ("000349", "CALDO DE GALINHA")]: + c = clf.classify({"current_ncm": "02071400", "description": desc}) + assert c["status"] == "LEGACY_NCM" + # No field contract, isso gera MANUAL com razão de ausência de sucessor unívoco + p = make_product(CODIGO=code, PRODUTO=desc, CLASSIFICACAO_FISCAL="02071400") + res = resolve_product(p, c, CO_SIMPLES_NAO_IPI, REFS, EV, Context([])) + assert res["CLASSIFICACAO_FISCAL"]["state"] == MANUAL + assert res["CLASSIFICACAO_FISCAL"]["reason"] == "NCM_LEGACY_NO_UNAMBIGUOUS_SUCCESSOR" + + # 4. Os 5 produtos reais fora da tabela vigente + for code, desc, ncm in [ + ("000061", "FRANGO (COXA E SOBRECOXA)", "02071200"), + ("000062", "CORAÇÃO DE FRANGO", "02071200"), + ("000098", "FEIJÃO DE CALDO", "21069091"), + ("000172", "PATÊ DE FRANGO", "02071200"), + ("000174", "PATÊ DE CAMARÃO", "03061300"), + ]: + c = clf.classify({"current_ncm": ncm, "description": desc}) + assert c["status"] == "OUT_OF_TABLE_NCM" + p = make_product(CODIGO=code, PRODUTO=desc, CLASSIFICACAO_FISCAL=ncm) + res = resolve_product(p, c, CO_SIMPLES_NAO_IPI, REFS, EV, Context([])) + assert res["CLASSIFICACAO_FISCAL"]["state"] == MANUAL + assert res["CLASSIFICACAO_FISCAL"]["reason"] == "NCM_OUT_OF_OFFICIAL_TABLE" + + +# ---------------- C.3: CEST ---------------- +def test_c3_cest_auto_fill_only_when_single_candidate(): + """C.3: AUTO_FILL_SAFE somente se o NCM mapear para exatamente um CEST vigente; senão candidatos para revisão.""" + off_single = [{"code": "03.021.03", "official_description": "Cerveja em lata", "description_score": 1.0}] + d_single = make_decision(cest_status="CEST_CONFIRMED", classification={"cest": ["03.021.03"], "cest_candidates": off_single}) + res_single = resolve_product(make_product(CEST=None), d_single, CO_SIMPLES_NAO_IPI, REFS, EV, Context([])) + assert res_single["CEST"]["state"] == CHANGE + assert res_single["CEST"]["proposed"] == "0302103" + assert res_single["CEST"]["reason"] == "SINGLE_DEFENSIBLE_OFFICIAL_CEST" + + # Múltiplos candidatos: NÃO PODE virar CHANGE automático + off_multi = [ + {"code": "03.021.03", "official_description": "Cerveja lata", "description_score": 0.8}, + {"code": "03.021.05", "official_description": "Cerveja garrafa", "description_score": 0.8}, + ] + d_multi = make_decision(cest_status="CEST_CONFIRMED", classification={"cest": ["03.021.03", "03.021.05"], "cest_candidates": off_multi}) + res_multi = resolve_product(make_product(CEST=None), d_multi, CO_SIMPLES_NAO_IPI, REFS, EV, Context([])) + assert res_multi["CEST"]["state"] == MANUAL + assert res_multi["CEST"]["reason"] == "SEVERAL_OFFICIAL_CEST_FOR_NCM" + assert len(res_multi["CEST"]["choices"]) == 2 + + +# ---------------- C.4: IPI ---------------- +def test_c4_ipi_taxpayer_gate(): + """C.4: valor da TIPI só onde CONTRIBUINTE_IPI aplicar; senão NOT_APPLICABLE.""" + # Empresa não contribuinte sem store_ipi_profile + res_na = resolve_product(make_product(IPI=None), make_decision(), CO_SIMPLES_NAO_IPI, REFS, EV, Context([])) + assert res_na["IPI"]["state"] == NA + assert res_na["IPI"]["reason"] == "COMPANY_IS_NOT_AN_IPI_TAXPAYER_C000004.CONTRIBUINTE_IPI" + assert res_na["CST_IPI"]["state"] == NA + + # Empresa contribuinte de IPI (CO_NORMAL_COM_IPI) + res_taxpayer = resolve_product(make_product(IPI=None), make_decision(), CO_NORMAL_COM_IPI, REFS, EV, Context([])) + assert res_taxpayer["IPI"]["state"] == CHANGE + assert res_taxpayer["IPI"]["proposed"] == 3.9 + + +# ---------------- C.6: GTIN Validation ---------------- +def test_c6_gtin_validation_algorithm(): + """C.6: Validação de DV GS1 e prefixos válidos/restritos.""" + # GTIN-13 válido (Coca-Cola 350ml: 7894900010015) + # Cálculo DV: 7 8 9 4 9 0 0 0 1 0 0 1 -> DV esperado 5 + assert calculate_gs1_check_digit("789490001001") == 5 + valid, reason = validate_gtin("7894900010015") + assert valid is True and reason == "GTIN_VALID" + + # DV errado + valid_bad_dv, reason = validate_gtin("7894900010019") + assert valid_bad_dv is False + assert "GTIN_INVALID_CHECK_DIGIT" in reason + + # Código interno / balança (prefixo 200...) + valid_internal, reason = validate_gtin("2000000000164") + assert valid_internal is False + assert "RESTRICTED_INTERNAL" in reason + + # Placeholder repetido + valid_rep, reason = validate_gtin("0000000000000") + assert valid_rep is False + assert "PLACEHOLDER" in reason + + # Tamanho inválido + valid_len, reason = validate_gtin("1234567") + assert valid_len is False