"""Fiscal Field Contract V1: one contract + one resolver for EVERY persistent fiscal field of the SGV product screen (C000025). Pure functions. Every (product, field) ends in exactly one of: ALREADY_CORRECT | CHANGE_REQUIRED | NOT_APPLICABLE | MANUAL_INPUT_REQUIRED() (CORRECTED is set by the writer after a verified write) Nothing is invented: a value is proposed only when an official dataset, the company context, ERP evidence or an OPERATOR context entry proves it. A valid ERP declaration that no proven evidence contradicts is ALREADY_CORRECT (reason DECLARED_VALID_NOT_CONTRADICTED); the report keeps those apart from rule-proven results. Operator context (Hub `fiscal_context_entries`) is hierarchical: product > product group (CODGRUPO) > company. """ from .fiscal_decision import _ex_described # noqa: F401 (description evidence helper shared with the decision engine) CONTRACT_VERSION = "field-contract-v1" ALREADY, CHANGE, CORRECTED, NA, MANUAL = "ALREADY_CORRECT", "CHANGE_REQUIRED", "CORRECTED", "NOT_APPLICABLE", "MANUAL_INPUT_REQUIRED" WRITE_FIELDS = ["CLASSIFICACAO_FISCAL", "IPI", "EX_TIPI", "CEST", "SITUACAO_TRIBUTARIA", "CFOP", "CFOP_ENTRADA", "CST", "CST_ENTRADA", "CSOSN", "ALIQUOTA", "COD_ORIGEM_MERCADORIA", "CST_IPI", "COD_ENTRADA_PIS", "COD_SAIDA_PIS", "ALIQ_PIS", "COD_ENTRADA_COFINS", "COD_SAIDA_COFINS", "ALIQ_COFINS", "INCIDENCIA_PISCOFINS", "PISCOFINS", "IAT", "IPPT"] # ---- contract metadata (docs/FISCAL_FIELD_CONTRACT_V1.json is generated from this) ------------------------------------------------------------------ def _c(ui, sem, dtype, source, inputs, rule, valid, na, manual, validation, prov, write=True): return {"UI_LABEL": ui, "SEMANTICS": sem, "DATA_TYPE": dtype, "SOURCE_OF_TRUTH": source, "REQUIRED_INPUTS": inputs, "RESOLUTION_RULE": rule, "VALID_VALUES": valid, "NOT_APPLICABLE_RULE": na, "MANUAL_INPUT_RULE": manual, "WRITE_ALLOWED": write, "VALIDATION_RULE": validation, "PROVENANCE": prov} CONTRACT = { "CLASSIFICACAO_FISCAL": _c("NCM", "NCM canônico (8 dígitos)", "VARCHAR(20)", "dataset NCM (Siscomex) ACTIVE + legacy NCM table", ["current NCM"], "VALID -> ALREADY_CORRECT. LEGACY/INVALID/MISSING never auto-replaced from the description; an OPERATOR context `ncm` that the Hub validates as EXACT_NCM becomes CHANGE_REQUIRED and every dependent field is reprocessed with it", "8-digit code present in the ACTIVE NCM dataset", "never (every product has an NCM)", "MANUAL_INPUT_REQUIRED(ncm) for LEGACY/INVALID/MISSING or an invalid manual value", "Hub classify status == EXACT_NCM", "NCM dataset sha256 + engine version"), "IPI": _c("Alíquota IPI %", "alíquota numérica de IPI", "DOUBLE PRECISION", "dataset TIPI_IPI ACTIVE", ["valid NCM", "TIPI row (and EX choice when several rows)"], "TIPI numeric rate -> ALREADY_CORRECT if equal else CHANGE_REQUIRED. NT/non-numeric and blank rates never write (NOT_APPLICABLE, never zero)", "float >= 0", "TIPI rate is NT/non-numeric or blank", "MANUAL_INPUT_REQUIRED(ex_tipi) when several TIPI rows exist for the NCM", "numeric, from TIPI row", "TIPI dataset sha256"), "EX_TIPI": _c("EX TIPI", "código EX da TIPI", "VARCHAR(3)", "dataset TIPI_IPI + product description evidence", ["valid NCM", "TIPI EX rows"], "EX applies only when the product description carries the official EX description; otherwise the base rate applies (no EX)", "EX code of an existing TIPI EX row", "NCM has no EX row, or the description evidences none", "MANUAL_INPUT_REQUIRED(ex_tipi) when the ERP holds an EX that no TIPI row supports", "value in TIPI EX rows for the NCM", "TIPI dataset sha256"), "CEST": _c("CEST", "CEST (7 dígitos sem pontuação)", "VARCHAR(7)", "dataset CEST (CONFAZ 142/18) ACTIVE + BA_ST_MVA", ["valid NCM", "description"], "single defensible official CEST -> ALREADY/CHANGE; ERP CEST that is one of the official codes of the NCM -> ALREADY_CORRECT; several official candidates with a blank CEST -> MANUAL; NCM absent from the CEST list -> NOT_APPLICABLE", "official CEST code of the NCM", "NCM not in the CEST list, or the description matches no official CEST item", "MANUAL_INPUT_REQUIRED(cest) when several official candidates exist, or the ERP CEST is not official for the NCM", "digits equal to an official CEST of the NCM", "CEST dataset sha256"), "SITUACAO_TRIBUTARIA": _c("Seletor ao lado do CEST", "T=Tributado, I=Isento, N=Não tributado, S=Substituição Tributária", "VARCHAR(2)", "ST engine (BA Anexo 1 + CEST corroboration) + purchase evidence / operator st_withheld", ["ST status", "ST withheld"], "ST proven and withheld -> S. Otherwise a valid declaration stays", "T | I | N | S", "never", "MANUAL_INPUT_REQUIRED(st_withheld) when ST is proven but retention unknown; (icms_treatment) when the field is blank", "value in T/I/N/S", "BA_ST_MVA + CEST dataset sha256"), "CFOP": _c("CFOP (Nota Fiscal de Saída)", "CFOP padrão de saída (intra-UF)", "VARCHAR(4)", "dataset CFOP oficial + product role + ST", ["product_role", "ST status", "ST withheld"], "RESALE: non-ST 5102, ST withheld 5405; OWN_PRODUCTION: 5101 / 5401; a declared official 5xxx that is not contradicted by proven role/ST stays", "official CFOP starting with 5", "INPUT (not a sale item)", "MANUAL_INPUT_REQUIRED(product_role | st_status | st_withheld | service_cfop)", "code in CFOP dataset", "CFOP dataset sha256"), "CFOP_ENTRADA": _c("CFOP (Nota Fiscal de Entrada)", "CFOP padrão de entrada", "CHAR(4)", "dataset CFOP + product role + ST + company entry scope", ["product_role", "ST withheld", "entry_scope (company)"], "RESALE 1102/2102, ST withheld 1403/2403, OWN_PRODUCTION/INPUT 1101/2101 (ST 1401/2401); declared official 1xxx/2xxx that is not contradicted stays", "official CFOP starting with 1 or 2", "SERVICE", "MANUAL_INPUT_REQUIRED(product_role | entry_scope | st_withheld)", "code in CFOP dataset", "CFOP dataset sha256"), "CST": _c("CST de ICMS (Nota Fiscal de Saída)", "CST ICMS = origem(1)+CST(2), regime normal", "VARCHAR(10)", "MOC CST_ICMS + ORIGEM tables + icms_treatment", ["CRT 2/3", "icms_treatment", "origin"], "CRT 1 uses CSOSN. CRT 2/3: treatment TAXED 00 / EXEMPT 40 / NOT_TAXED 41 / ST_WITHHELD 60; a valid declaration that the treatment does not contradict stays", "3 digits: origin 0-8 + CST_ICMS code", "CRT = 1", "MANUAL_INPUT_REQUIRED(icms_treatment | origin) when blank and not derivable", "origin in ORIGEM table, last two digits in CST_ICMS", "MOC tables sha256"), "CST_ENTRADA": _c("CST de ICMS (Nota Fiscal de Entrada)", "CST ICMS de entrada, regime normal", "CHAR(4)", "MOC tables + icms_treatment", ["CRT 2/3", "icms_treatment", "origin"], "same derivation as CST for CRT 2/3", "3 digits origin+CST", "CRT = 1", "MANUAL_INPUT_REQUIRED(icms_treatment | origin)", "as CST", "MOC tables sha256"), "CSOSN": _c("CSOSN Classificação", "CSOSN de saída (Simples Nacional)", "VARCHAR(60)", "MOC CSOSN table + icms_treatment (operator or proven ST)", ["CRT 1", "icms_treatment"], "treatment TAXED -> 101/102/103 family (default 102), EXEMPT/NOT_TAXED -> 300/400 (default 400), ST_WITHHELD -> 500; a valid declaration outside a proven contradiction stays", "CSOSN code in MOC table", "CRT != 1", "MANUAL_INPUT_REQUIRED(icms_treatment) when blank/invalid", "code in CSOSN table", "MOC tables sha256"), "ALIQUOTA": _c("Aliq. ICMS", "alíquota ICMS % de destaque", "NUMERIC(15,2)", "regime + CSOSN/CST family + operator icms_rate", ["CSOSN or CST family", "icms_rate (company/group) for taxed regime-normal items"], "no ICMS highlighted (CSOSN 102/103/300/400/500/202/203/900 or CST 40/41/60/50/51) -> NOT_APPLICABLE; taxed items keep a declared value or take the operator icms_rate", "numeric >= 0", "no ICMS is highlighted for the item", "MANUAL_INPUT_REQUIRED(icms_rate | credit_rate)", "numeric", "operator context / company regime"), "COD_ORIGEM_MERCADORIA": _c("Código da origem da Mercadoria (0/1/2)", "origem da mercadoria (0 Nacional, 1 Estrangeira-importação direta, 2 Estrangeira-mercado interno)", "INTEGER", "operator origin only (no ERP field carries origin evidence)", ["origin"], "declared 0/1/2 stays; blank needs operator/evidence: the CST first digit is padding in the SGV (audit V2) and is NOT origin evidence; never defaulted to 0", "0 | 1 | 2", "never", "MANUAL_INPUT_REQUIRED(origin)", "value in {0,1,2}", "SGV DFM: hidden edit CodOrigemMercadoria + RgOrigemMercadoria.OnClick"), "CST_IPI": _c("CST IPI", "CST do IPI", "CHAR(3)", "company C000004.CONTRIBUINTE_IPI", ["company IPI taxpayer flag"], "company not an IPI taxpayer (CONTRIBUINTE_IPI != S) -> NOT_APPLICABLE; taxpayer -> declared value stays", "IPI CST code", "company is not an IPI taxpayer", "MANUAL_INPUT_REQUIRED(cst_ipi) for IPI taxpayers with a blank value", "non-blank code", "C000004.CONTRIBUINTE_IPI"), "COD_ENTRADA_PIS": _c("Pis Entrada", "CST PIS de entrada", "VARCHAR(3)", "MOC CST_PIS + operator pis_cofins_profile", ["pis_cofins_profile"], "profile in_cst (validated against CST_PIS) -> ALREADY/CHANGE; else a valid declared code stays", "CST PIS code", "never", "MANUAL_INPUT_REQUIRED(pis_cofins_profile)", "code in CST_PIS table", "MOC tables sha256"), "COD_SAIDA_PIS": _c("Pis Saída", "CST PIS de saída", "VARCHAR(3)", "MOC CST_PIS + operator pis_cofins_profile", ["pis_cofins_profile"], "profile out_cst -> ALREADY/CHANGE; else a valid declared code stays", "CST PIS code", "never", "MANUAL_INPUT_REQUIRED(pis_cofins_profile)", "code in CST_PIS table", "MOC tables sha256"), "ALIQ_PIS": _c("Alíquota (Pis)", "alíquota PIS %", "VARCHAR(8)", "operator pis_cofins_profile", ["pis_cofins_profile"], "profile pis_rate -> ALREADY/CHANGE; else a declared value stays (never assumed 0)", "decimal text", "never", "MANUAL_INPUT_REQUIRED(pis_cofins_profile)", "numeric text", "operator context"), "COD_ENTRADA_COFINS": _c("Cofins Entrada", "CST COFINS de entrada", "VARCHAR(3)", "MOC CST_COFINS + operator pis_cofins_profile", ["pis_cofins_profile"], "profile in_cst -> ALREADY/CHANGE; else a valid declared code stays", "CST COFINS code", "never", "MANUAL_INPUT_REQUIRED(pis_cofins_profile)", "code in CST_COFINS table", "MOC tables sha256"), "COD_SAIDA_COFINS": _c("Cofins Saída", "CST COFINS de saída", "VARCHAR(3)", "MOC CST_COFINS + operator pis_cofins_profile", ["pis_cofins_profile"], "profile out_cst -> ALREADY/CHANGE; else a valid declared code stays", "CST COFINS code", "never", "MANUAL_INPUT_REQUIRED(pis_cofins_profile)", "code in CST_COFINS table", "MOC tables sha256"), "ALIQ_COFINS": _c("Alíquota (Cofins)", "alíquota COFINS %", "VARCHAR(8)", "operator pis_cofins_profile", ["pis_cofins_profile"], "profile cofins_rate -> ALREADY/CHANGE; else a declared value stays (never assumed 0)", "decimal text", "never", "MANUAL_INPUT_REQUIRED(pis_cofins_profile)", "numeric text", "operator context"), "INCIDENCIA_PISCOFINS": _c("Incidência", "tipo de incidência PIS/COFINS", "VARCHAR(30)", "ERP declaration (NENHUMA / MONOFÁSICA / SUBST.TRIBUTÁRIA) + operator incidence", ["incidence (group) when blank"], "valid declaration stays; operator `incidence` may set it", "NENHUMA | MONOFÁSICA | SUBST.TRIBUTÁRIA", "never", "MANUAL_INPUT_REQUIRED(incidence)", "value in domain", "ERP domain observed in 4 databases"), "PISCOFINS": _c("PIS/COFINS (combo)", "flag PIS/COFINS N/S", "VARCHAR(1)", "ERP declaration + operator piscofins", ["piscofins when blank"], "valid N/S stays; operator `piscofins` may set it", "N | S", "never", "MANUAL_INPUT_REQUIRED(piscofins)", "value in N/S", "ERP domain observed in 4 databases"), "IAT": _c("IAT", "indicador de arredondamento (A) ou truncamento (T)", "VARCHAR(1)", "company rounding policy (operator) or ERP declaration", ["rounding (company) when blank"], "valid A/T stays; blank -> operator `rounding`", "A | T", "never", "MANUAL_INPUT_REQUIRED(rounding)", "value in A/T", "PAF-ECF/Convênio registro de produto layout (IAT)"), "IPPT": _c("IPPT", "indicador de produção própria (P) ou de terceiros (T)", "VARCHAR(1)", "product_role (own production evidence / purchase evidence / operator)", ["product_role"], "OWN_PRODUCTION -> P, RESALE -> T; INPUT/SERVICE not applicable", "P | T", "INPUT or SERVICE", "MANUAL_INPUT_REQUIRED(product_role)", "value in P/T", "PAF-ECF/Convênio registro de produto layout (IPPT)"), } for _f, _v in CONTRACT.items(): _v["ERP_FIELD"] = _f RTC_CAPABILITY = {"RTC_CCLASSTRIB": "ERP_SCHEMA_NOT_READY", "RTC_CCREDPRES": "ERP_SCHEMA_NOT_READY", "RTC_CBS_RATES": "ERP_SCHEMA_NOT_READY"} ORIGIN_UI = {"0", "1", "2"} CSOSN_FAMILY = {"TAXED": ("102", {"101", "102", "103"}), "EXEMPT": ("400", {"300", "400"}), "NOT_TAXED": ("400", {"300", "400"}), "ST_WITHHELD": ("500", {"500"}), "ST_CHARGED": ("202", {"201", "202", "203"})} CST_TREATMENT = {"TAXED": "00", "EXEMPT": "40", "NOT_TAXED": "41", "ST_WITHHELD": "60", "SUSPENDED": "50", "DEFERRED": "51"} NO_ICMS_CSOSN = {"102", "103", "300", "400", "500", "202", "203", "900"} NO_ICMS_CST = {"40", "41", "60", "50", "51"} RESALE_CFOP = {"5101", "5102", "5401", "5405"} ENTRY_CFOP = {"1101", "1102", "1401", "1403", "2101", "2102", "2401", "2403"} class Context: """Operator context lookup: product > group > company.""" def __init__(self, entries=()): self._by = {} for e in entries: self._by[(e["scope"].upper(), str(e.get("scope_key") or ""), e["key"])] = e["value"] def get(self, key, code=None, group=None, ncm=None): for scope, sk in (("PRODUCT", code), ("GROUP", group), ("NCM", ncm), ("COMPANY", "")): if scope != "COMPANY" and not sk: continue v = self._by.get((scope, str(sk or ""), key)) if v is not None and str(v).strip() != "": return str(v).strip(), scope return None, None def _s(v): return None if v is None else str(v).strip() def _blank(v): return v is None or str(v).strip() == "" def _digits(v): return "".join(c for c in str(v or "") if c.isdigit()) 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"], "legal_basis": basis, **extra} # C000032.TIPO=COMPOSICAO evidence must be a deterministic MAJORITY of that product's own movement # history to count as own-production proof, never a bare "it happened at least once" check (a beer # used once as a recipe component among 19 direct sales is not "own production"). Thresholds: # below MINORITY -> composed evidence is discarded entirely (too rare to mean anything); above # DOMINANT -> strong own-production evidence; in between (with enough events to be meaningful) -> # genuinely conflicting evidence, never guessed either way (Missão 4 / Bloco A.1). COMPOSED_MIN_TOTAL_EVENTS = 5 COMPOSED_MINORITY_RATIO = 0.30 COMPOSED_DOMINANT_RATIO = 0.70 def _composed_signal(ev, code): """None (no usable evidence), "DOMINANT", "MINORITY" or "MIXED", plus the ratio for the reason string.""" info = (ev.get("composed_ratio") or {}).get(code) if not info: return (None, None) total = info.get("composed", 0) + info.get("other", 0) if total < COMPOSED_MIN_TOTAL_EVENTS: return (None, None) ratio = info["composed"] / total if ratio >= COMPOSED_DOMINANT_RATIO: return ("DOMINANT", ratio) if ratio <= COMPOSED_MINORITY_RATIO: return ("MINORITY", ratio) return ("MIXED", ratio) def infer_role(p, ev, ctx, code, group, ncm=None): v, scope = ctx.get("product_role", code, group, ncm) if v in ("RESALE", "OWN_PRODUCTION", "INPUT", "SERVICE"): return v, [f"OPERATOR({scope})"] if _s(p.get("PRODUTO_SERVICO")): return "SERVICE", ["C000025.PRODUTO_SERVICO"] if (_s(p.get("TIPO")) or "").upper() == "MATERIA-PRIMA": return "INPUT", ["C000025.TIPO=MATERIA-PRIMA"] composed_signal, composed_ratio = _composed_signal(ev, code) if composed_signal == "MIXED": return "UNKNOWN", [f"C000032.TIPO=COMPOSICAO_MIXED_EVIDENCE(ratio={composed_ratio:.1%})"] composed_dominant = composed_signal == "DOMINANT" own = [n for n, on in (("PROD_COZINHA=1", p.get("PROD_COZINHA") == 1), ("RECEITA_INGREDIENTES", code in ev.get("recipes", ())), ("C000032.TIPO=COMPOSICAO", composed_dominant)) if on] docs = bool(ev.get("purchases", {}).get(code)) or ev.get("stock_entry", {}).get(code, False) stock = ev.get("stock_in", {}).get(code, False) if own and docs: return "UNKNOWN", own + ["ALSO_PURCHASE_DOCUMENTS_CONFLICT"] strong_own = [n for n in own if n != "PROD_COZINHA=1"] if strong_own: return "OWN_PRODUCTION", own if own: # audit V2: the kitchen-routing flag alone is not proof of own production return "UNKNOWN", own + ["PROD_COZINHA_ALONE_NOT_PRODUCTION_PROOF"] if (_s(p.get("TIPO")) or "").upper() == "VENDA" and (docs or stock): return "RESALE", ["C000025.TIPO=VENDA", "PURCHASE_DOCUMENT" if docs else "STOCK_ENTRIES(C000100.ENTRADAS>0)"] return "UNKNOWN", ["NO_OBJECTIVE_ROLE_EVIDENCE"] def _isnum(v): try: float(v) return True except (TypeError, ValueError): return False _SPED_CACHE = {} def sped_pis_cofins_match(ncm): """Active SPED EFD-Contribuicoes table 4.3.10/12/13/14/15 entry (explicit NCM prefix) covering the NCM -> list of (table, saida_cst, code, start). Empty when none.""" if "map" not in _SPED_CACHE: import json as _j from pathlib import Path as _P m = {} f = _P(__file__).resolve().parents[2] / "docs" / "sources" / "SPED_PISCOFINS_TABLES_V3.json" if f.exists(): for tab, t in _j.loads(f.read_text(encoding="utf-8"))["tables"].items(): for e in t.get("entries", []): if e["end"] is None: for pref in e["ncm_prefixes"]: m.setdefault(pref, set()).add((tab, t["saida_cst"][-1], e["code"], e["start"])) _SPED_CACHE["map"] = m hits = set() for n in (4, 6, 8): hits |= _SPED_CACHE["map"].get(ncm[:n], set()) if len(ncm) >= n else set() return sorted(hits) def resolve_product(p, d, company, refs, ev, ctx, manual_ncm=None): """p: product row (all contract columns + CODIGO, PRODUTO, CODGRUPO, TIPO, PROD_COZINHA, PRODUTO_SERVICO). d: Hub decision for the EFFECTIVE NCM (the manual NCM when one was supplied and validated). manual_ncm: (value, is_valid) or None. Returns {field: result}.""" from .fiscal_decision import decide_product from .icms_context import st_status, withheld_status code, group = _s(p["CODIGO"]), _s(p.get("CODGRUPO")) NCM = _digits(p.get("CLASSIFICACAO_FISCAL")) crt, uf = company.get("crt"), company.get("uf") out = {} cur = {f: _s(p.get(f)) for f in WRITE_FIELDS} # ---------------- NCM ---------------- ncm_ok = d.get("status") == "EXACT_NCM" if manual_ncm is not None: val, valid = manual_ncm if not valid: out["CLASSIFICACAO_FISCAL"] = _r("CLASSIFICACAO_FISCAL", MANUAL, cur["CLASSIFICACAO_FISCAL"], reason="MANUAL_NCM_NOT_VALID_IN_NCM_DATASET", key="ncm", scope="PRODUCT") elif _digits(val) == _digits(cur["CLASSIFICACAO_FISCAL"]) and d.get("status") == "EXACT_NCM": out["CLASSIFICACAO_FISCAL"] = _r("CLASSIFICACAO_FISCAL", ALREADY, cur["CLASSIFICACAO_FISCAL"], reason="OPERATOR_NCM_EQUALS_CURRENT") 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", legal_basis={"dataset": "NCM", "version": refs.get("ncm_version") or "Siscomex", "line": cur["CLASSIFICACAO_FISCAL"], "act": "Resolução Gecex / MDIC"}) else: 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") eff_ncm = out["CLASSIFICACAO_FISCAL"]["proposed"] or cur["CLASSIFICACAO_FISCAL"] prod = {**p, "CLASSIFICACAO_FISCAL": eff_ncm} # ---------------- context facts ---------------- role, role_ev = infer_role(p, ev, ctx, code, group, NCM) st_ctx, st_scope = ctx.get("st_status", code, group) ba = (d.get("classification") or {}).get("ba_st_mva") or {} if st_ctx in ("SUBJECT", "NOT_SUBJECT"): st = st_ctx elif ncm_ok and uf == "BA": s, _ = st_status(d.get("status"), d.get("ba_st_mva_status"), d.get("cest_status"), cur["CEST"], ba.get("cest"), None) st = {"ST_CONFIRMED": "SUBJECT", "ST_NOT_APPLICABLE": "NOT_SUBJECT"}.get(s, "UNKNOWN") else: st = "UNKNOWN" wh_ctx, _ = ctx.get("st_withheld", code, group) wh_doc, _ = withheld_status(ev.get("purchases", {}).get(code, [])) withheld = wh_ctx if wh_ctx in ("Y", "N") else ("Y" if wh_doc == "ST_ALREADY_WITHHELD_CONFIRMED" else None) # purchase items flagged N are weak evidence: kept as unknown treat, _ = ctx.get("icms_treatment", code, group) if not treat and st == "SUBJECT" and withheld == "Y": treat = "ST_WITHHELD" origin_ctx, _ = ctx.get("origin", code, group) # ---------------- decision-engine backed fields ---------------- if ncm_ok: rec = decide_product({**prod, "CODIGO": code, "IPI": p.get("IPI"), "EX_TIPI": p.get("EX_TIPI"), "CEST": p.get("CEST")}, d, {"uf": uf or "", "crt": crt or ""}, refs) ipi_field = "IPI" if not ncm_ok: out["IPI"], out["EX_TIPI"], out["CEST"] = blocked("IPI"), blocked("EX_TIPI"), blocked("CEST") else: # IPI ex_ctx, _ = ctx.get("ex_tipi", code, group, NCM) cands = (d.get("classification") or {}).get("ipi_candidates") or [] chosen = None if ex_ctx: chosen = next((c for c in cands if (c.get("ex") is None if ex_ctx.upper() == "NONE" else str(c.get("ex")) == ex_ctx)), None) if chosen and chosen.get("rate") not in (None, ""): try: rate = float(chosen["rate"]) except ValueError: rate = None if rate is not None: same = cur["IPI"] is not None and abs(float(cur["IPI"]) - rate) < 1e-9 out["IPI"] = _r("IPI", ALREADY if same else CHANGE, cur["IPI"], None if same else rate, "OPERATOR_EX_SELECTED_TIPI_ROW") if "IPI" not in out: r = rec["IPI"] if r["status"] == "RULE_RESOLVED": out["IPI"] = _r("IPI", ALREADY if r["reason_code"].endswith("SAME_VALUE") else CHANGE, cur["IPI"], None if r["reason_code"].endswith("SAME_VALUE") else r["proposed_value"], "TIPI_NUMERIC_RATE") elif r["status"] == "RULE_CANDIDATE": out["IPI"] = _r("IPI", MANUAL, cur["IPI"], reason="SEVERAL_TIPI_ROWS_FOR_NCM", key="ex_tipi", scope="PRODUCT", 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 / contribuinte_ipi gate is_taxpayer = company.get("contribuinte_ipi") == "S" prof, _ = ctx.get("store_ipi_profile", code, group) if not is_taxpayer: if prof == "POPULATE_REFERENCE_RATE": pass elif prof == "LEAVE_NOT_APPLICABLE": out["IPI"] = _r("IPI", NA, cur["IPI"], reason="STORE_IPI_PROFILE_LEAVE_NOT_APPLICABLE", legal_basis={"dataset": "TIPI_IPI", "version": "N/A", "line": "N/A", "act": "RIPI / CF art. 153, IV (não contribuinte)"}) elif prof is None and out["IPI"]["state"] == CHANGE: out["IPI"] = _r("IPI", MANUAL, cur["IPI"], reason="STORE_IPI_PROFILE_NOT_DECIDED", key="store_ipi_profile", scope="COMPANY") else: 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") # EX_TIPI ex_rows = [c for c in cands if c.get("ex")] if ex_ctx: if ex_ctx.upper() == "NONE": out["EX_TIPI"] = _r("EX_TIPI", NA, cur["EX_TIPI"], reason="OPERATOR_SAYS_NO_EX") elif any(str(c["ex"]) == ex_ctx for c in ex_rows): out["EX_TIPI"] = _r("EX_TIPI", ALREADY if cur["EX_TIPI"] == ex_ctx else CHANGE, cur["EX_TIPI"], None if cur["EX_TIPI"] == ex_ctx else ex_ctx, "OPERATOR_EX_VALID_IN_TIPI") else: out["EX_TIPI"] = _r("EX_TIPI", MANUAL, cur["EX_TIPI"], reason="OPERATOR_EX_NOT_IN_TIPI_ROWS", key="ex_tipi", scope="PRODUCT") else: r = rec["EX_TIPI"] if r["status"] == "RULE_RESOLVED": out["EX_TIPI"] = _r("EX_TIPI", ALREADY if r["reason_code"].endswith("SAME_VALUE") else CHANGE, cur["EX_TIPI"], None if r["reason_code"].endswith("SAME_VALUE") else r["proposed_value"], "TIPI_EX_DESCRIPTION_EVIDENCE") elif cur["EX_TIPI"] and not any(str(c["ex"]) == cur["EX_TIPI"] for c in ex_rows): out["EX_TIPI"] = _r("EX_TIPI", MANUAL, cur["EX_TIPI"], reason="ERP_EX_NOT_SUPPORTED_BY_TIPI", key="ex_tipi", scope="PRODUCT") elif cur["EX_TIPI"]: out["EX_TIPI"] = _r("EX_TIPI", ALREADY, cur["EX_TIPI"], reason="DECLARED_VALID_NOT_CONTRADICTED") else: out["EX_TIPI"] = _r("EX_TIPI", NA, cur["EX_TIPI"], reason="NO_EX_ROW_FOR_NCM" if not ex_rows else "DESCRIPTION_EVIDENCES_NO_EX_BASE_RATE_APPLIES") # CEST cest_ctx, _ = ctx.get("cest", code, group, NCM) cls = d.get("classification") or {} official = {_digits(c["code"]): c for c in (cls.get("cest_candidates") or [])} cs = d.get("cest_status") cur_cest = _digits(cur["CEST"]) if cest_ctx: if cest_ctx.upper() == "NONE": out["CEST"] = _r("CEST", NA, cur["CEST"], reason="OPERATOR_SAYS_NO_CEST") elif _digits(cest_ctx) in official: out["CEST"] = _r("CEST", ALREADY if cur_cest == _digits(cest_ctx) else CHANGE, cur["CEST"], None if cur_cest == _digits(cest_ctx) else _digits(cest_ctx), "OPERATOR_CEST_OFFICIAL_FOR_NCM") else: out["CEST"] = _r("CEST", MANUAL, cur["CEST"], reason="OPERATOR_CEST_NOT_OFFICIAL_FOR_NCM", key="cest", scope="PRODUCT") elif cs == "CEST_CONFIRMED": 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": out["CEST"] = _r("CEST", NA, cur["CEST"], reason="NCM_NOT_IN_CEST_LIST") if not cur_cest else _r("CEST", MANUAL, cur["CEST"], reason="ERP_CEST_BUT_NCM_NOT_IN_CEST_LIST", key="cest", scope="PRODUCT") elif cs == "CEST_CANDIDATES" and not cur_cest: 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()]) elif cs == "CEST_DESCRIPTION_MISMATCH" and not cur_cest: out["CEST"] = _r("CEST", NA, cur["CEST"], reason="DESCRIPTION_MATCHES_NO_OFFICIAL_CEST_ITEM") elif cur_cest: out["CEST"] = _r("CEST", MANUAL, cur["CEST"], reason="ERP_CEST_NOT_OFFICIAL_FOR_NCM", key="cest", scope="PRODUCT") else: out["CEST"] = _r("CEST", NA, cur["CEST"], reason="NO_CEST_APPLICABLE") # ---------------- SITUACAO_TRIBUTARIA ---------------- sit = cur["SITUACAO_TRIBUTARIA"] valid_sit = sit in ("T", "I", "N", "S") if not ncm_ok: out["SITUACAO_TRIBUTARIA"] = blocked("SITUACAO_TRIBUTARIA") elif st == "SUBJECT" and withheld == "Y": out["SITUACAO_TRIBUTARIA"] = _r("SITUACAO_TRIBUTARIA", ALREADY if sit == "S" else CHANGE, sit, None if sit == "S" else "S", "ST_PROVEN_AND_WITHHELD") elif st == "SUBJECT" and withheld is None: out["SITUACAO_TRIBUTARIA"] = _r("SITUACAO_TRIBUTARIA", ALREADY, sit, reason="DECLARED_S_CONSISTENT_WITH_PROVEN_ST") if sit == "S" else _r("SITUACAO_TRIBUTARIA", MANUAL, sit, reason="ST_PROVEN_RETENTION_UNKNOWN", key="st_withheld", scope="GROUP") elif valid_sit and not (st == "NOT_SUBJECT" and sit == "S"): out["SITUACAO_TRIBUTARIA"] = _r("SITUACAO_TRIBUTARIA", ALREADY, sit, reason="DECLARED_VALID_NOT_CONTRADICTED") elif valid_sit: out["SITUACAO_TRIBUTARIA"] = _r("SITUACAO_TRIBUTARIA", MANUAL, sit, reason="DECLARED_S_BUT_NOT_IN_ANY_ST_LIST", key="icms_treatment", scope="GROUP") else: out["SITUACAO_TRIBUTARIA"] = _r("SITUACAO_TRIBUTARIA", MANUAL, sit, reason="BLANK_OR_INVALID", key="icms_treatment", scope="GROUP") # ---------------- CFOP saída ---------------- cfop, cfop_valid = cur["CFOP"], refs["cfop_valid"] if role == "INPUT": out["CFOP"] = _r("CFOP", NA, cfop, reason="INPUT_IS_NOT_A_SALE_ITEM") elif role == "SERVICE": out["CFOP"] = _r("CFOP", MANUAL, cfop, reason="SERVICE_CFOP_NEEDS_OPERATOR", key="service_cfop", scope="GROUP") elif not ncm_ok: out["CFOP"] = blocked("CFOP") else: expected = None if role in ("RESALE", "OWN_PRODUCTION"): base = "5102" if role == "RESALE" else "5101" if st == "NOT_SUBJECT": expected = base elif st == "SUBJECT" and withheld == "Y": expected = "5405" if role == "RESALE" else "5401" valid_cur = cfop in cfop_valid and cfop.startswith("5") erp_st_declared = cur["CSOSN"] in ("500", "201", "202", "203") or (cur["CST"] or "")[-2:] in ("60", "10", "70", "30") if valid_cur and expected and expected in ("5102", "5101") and cfop in ("5405", "5401") and erp_st_declared: out["CFOP"] = _r("CFOP", MANUAL, cfop, reason="OFFICIAL_ST_TABLE_CONFLICTS_WITH_ERP_ST_DECLARATION", key="st_status", scope="GROUP") elif valid_cur and expected and cfop != expected and cfop in RESALE_CFOP: out["CFOP"] = _r("CFOP", CHANGE, cfop, expected, "PROVEN_ROLE_AND_ST_CONTRADICT_DECLARED_CFOP") elif valid_cur and expected == cfop: out["CFOP"] = _r("CFOP", ALREADY, cfop, reason="RULE_PROVEN_EQUALS_DECLARED") elif valid_cur: out["CFOP"] = _r("CFOP", ALREADY, cfop, reason="DECLARED_VALID_NOT_CONTRADICTED") elif expected: out["CFOP"] = _r("CFOP", CHANGE, cfop, expected, "DERIVED_FROM_ROLE_AND_ST" + ("_REPLACING_INVALID" if cfop else "_FILLING_BLANK")) else: key = "product_role" if role == "UNKNOWN" else ("st_status" if st == "UNKNOWN" else "st_withheld") out["CFOP"] = _r("CFOP", MANUAL, cfop, reason="CFOP_NEEDS_" + key.upper(), key=key, scope="GROUP") # ---------------- CFOP entrada ---------------- cfe = cur["CFOP_ENTRADA"] scope_ctx, _ = ctx.get("entry_scope", code, group) if role == "SERVICE": out["CFOP_ENTRADA"] = _r("CFOP_ENTRADA", NA, cfe, reason="SERVICE_HAS_NO_GOODS_ENTRY") elif not ncm_ok: out["CFOP_ENTRADA"] = blocked("CFOP_ENTRADA") else: exp_e = None if scope_ctx == "MIXED_OR_DYNAMIC": # Scope answered, but no single static entry CFOP is fiscally correct for every # operation: the ERP default / each operation decides 1xxx vs 2xxx. Never invent one. valid_e = cfe in cfop_valid and cfe[:1] in ("1", "2") if valid_e: out["CFOP_ENTRADA"] = _r("CFOP_ENTRADA", ALREADY, cfe, reason="DECLARED_VALID_OPERATION_DEPENDENT_MIXED_SCOPE") else: # Audited ADMIN decision (entry_scope=MIXED_OR_DYNAMIC) already says no static CFOP_ENTRADA is # ever correct here: NA (never MANUAL) so this doesn't block product completeness and never # raises a spurious "answer entry_scope" question — the scope question was already answered. out["CFOP_ENTRADA"] = _r("CFOP_ENTRADA", NA, cfe, reason="CFOP_ENTRADA_OPERATION_DEPENDENT_MIXED_SCOPE") else: if scope_ctx in ("INTRA", "INTER") and role in ("RESALE", "OWN_PRODUCTION", "INPUT") and st != "UNKNOWN" and not (st == "SUBJECT" and withheld is None): first = "1" if scope_ctx == "INTRA" else "2" stw = st == "SUBJECT" and withheld == "Y" exp_e = first + ("403" if stw else "102") if role == "RESALE" else first + ("401" if stw else "101") valid_e = cfe in cfop_valid and cfe[:1] in ("1", "2") if valid_e and exp_e and cfe != exp_e and cfe in ENTRY_CFOP: out["CFOP_ENTRADA"] = _r("CFOP_ENTRADA", CHANGE, cfe, exp_e, "PROVEN_ROLE_ST_AND_SCOPE_CONTRADICT_DECLARED_CFOP_ENTRADA") elif valid_e: out["CFOP_ENTRADA"] = _r("CFOP_ENTRADA", ALREADY, cfe, reason="DECLARED_VALID_NOT_CONTRADICTED") elif exp_e: out["CFOP_ENTRADA"] = _r("CFOP_ENTRADA", CHANGE, cfe, exp_e, "DERIVED_FROM_ROLE_ST_AND_ENTRY_SCOPE") else: key = "product_role" if role == "UNKNOWN" else ("entry_scope" if scope_ctx not in ("INTRA", "INTER") else ("st_status" if st == "UNKNOWN" else "st_withheld")) out["CFOP_ENTRADA"] = _r("CFOP_ENTRADA", MANUAL, cfe, reason="CFOP_ENTRADA_NEEDS_" + key.upper(), key=key, scope="COMPANY" if key == "entry_scope" else "GROUP") # ---------------- origin (needed by CST) ---------------- cod_o = cur["COD_ORIGEM_MERCADORIA"] cst_digit = (cur["CST"] or "")[:1] if cur["CST"] and len(cur["CST"]) == 3 and (cur["CST"][:1] in "012345678") else None if origin_ctx in ORIGIN_UI: out["COD_ORIGEM_MERCADORIA"] = _r("COD_ORIGEM_MERCADORIA", ALREADY if cod_o == origin_ctx else CHANGE, cod_o, None if cod_o == origin_ctx else int(origin_ctx), "OPERATOR_ORIGIN") elif origin_ctx: out["COD_ORIGEM_MERCADORIA"] = _r("COD_ORIGEM_MERCADORIA", MANUAL, cod_o, reason="OPERATOR_ORIGIN_NOT_REPRESENTABLE_ON_SGV_SCREEN_0_1_2", key="origin", scope="GROUP") elif cod_o in ORIGIN_UI: out["COD_ORIGEM_MERCADORIA"] = _r("COD_ORIGEM_MERCADORIA", ALREADY, cod_o, reason="DECLARED_VALID_NOT_CONTRADICTED") else: # V2 audit: the CST first digit is padding in the SGV (declared-origin 1/2 products also carry CST digit 0), so it is NOT evidence of origin out["COD_ORIGEM_MERCADORIA"] = _r("COD_ORIGEM_MERCADORIA", MANUAL, cod_o, reason="ORIGIN_NO_EVIDENCE", key="origin", scope="GROUP") origin_digit = out["COD_ORIGEM_MERCADORIA"]["proposed"] if out["COD_ORIGEM_MERCADORIA"]["proposed"] is not None else (cod_o if cod_o in ORIGIN_UI else None) origin_digit = None if origin_digit is None else str(origin_digit) # ---------------- CST / CST_ENTRADA (regime normal) ---------------- for field in ("CST", "CST_ENTRADA"): v = cur[field] if crt == "1": out[field] = _r(field, NA, v, reason="CRT1_SIMPLES_EMISSION_USES_CSOSN") continue if crt not in ("2", "3"): out[field] = _r(field, MANUAL, v, reason="CRT_UNKNOWN", key="crt", scope="COMPANY") continue valid_c = bool(v) and len(v) == 3 and v[0] in "012345678" and v[1:] in refs["cst_icms"] exp = (origin_digit + CST_TREATMENT[treat]) if (treat in CST_TREATMENT and origin_digit) else None if valid_c and exp and v != exp and v[1:] != exp[1:]: out[field] = _r(field, CHANGE, v, exp, "TREATMENT_CONTRADICTS_DECLARED_CST") elif valid_c: out[field] = _r(field, ALREADY, v, reason="DECLARED_VALID_NOT_CONTRADICTED") elif exp: out[field] = _r(field, CHANGE, v, exp, "DERIVED_FROM_TREATMENT_AND_ORIGIN") else: out[field] = _r(field, MANUAL, v, reason="CST_NEEDS_ICMS_TREATMENT" if origin_digit else "CST_NEEDS_ORIGIN", key="icms_treatment" if origin_digit else "origin", scope="GROUP") # ---------------- CSOSN ---------------- v = cur["CSOSN"] if crt != "1": out["CSOSN"] = _r("CSOSN", NA, v, reason="CSOSN_IS_THE_SIMPLES_NACIONAL_CODE_CRT_IS_" + str(crt)) elif not ncm_ok: out["CSOSN"] = blocked("CSOSN") else: fam = CSOSN_FAMILY.get(treat) if v in refs["csosn"] and (not fam or v in fam[1]): out["CSOSN"] = _r("CSOSN", ALREADY, v, reason="RULE_PROVEN_EQUALS_DECLARED" if fam else "DECLARED_VALID_NOT_CONTRADICTED") elif v in refs["csosn"] and fam: out["CSOSN"] = _r("CSOSN", CHANGE, v, fam[0], "TREATMENT_CONTRADICTS_DECLARED_CSOSN") elif fam: out["CSOSN"] = _r("CSOSN", CHANGE, v, fam[0], "DERIVED_FROM_TREATMENT" + ("_REPLACING_INVALID" if v else "_FILLING_BLANK")) else: out["CSOSN"] = _r("CSOSN", MANUAL, v, reason="CSOSN_NEEDS_ICMS_TREATMENT", key="icms_treatment", scope="GROUP") # ---------------- ALIQUOTA ---------------- v = cur["ALIQUOTA"] csosn_now = out["CSOSN"]["proposed"] or cur["CSOSN"] cst_now = out["CST"]["proposed"] or cur["CST"] rate_ctx, _ = ctx.get("icms_rate", code, group) if crt == "1" and (csosn_now in NO_ICMS_CSOSN): out["ALIQUOTA"] = _r("ALIQUOTA", NA, v, reason="SIMPLES_CSOSN_HIGHLIGHTS_NO_ICMS_RATE") elif crt in ("2", "3") and cst_now and cst_now[1:] in NO_ICMS_CST: out["ALIQUOTA"] = _r("ALIQUOTA", NA, v, reason="CST_HIGHLIGHTS_NO_ICMS_RATE") elif rate_ctx: try: r = float(rate_ctx.replace(",", ".")) same = v is not None and abs(float(v) - r) < 1e-9 out["ALIQUOTA"] = _r("ALIQUOTA", ALREADY if same else CHANGE, v, None if same else r, "OPERATOR_ICMS_RATE") except ValueError: out["ALIQUOTA"] = _r("ALIQUOTA", MANUAL, v, reason="OPERATOR_ICMS_RATE_NOT_NUMERIC", key="icms_rate", scope="COMPANY") elif v not in (None, ""): out["ALIQUOTA"] = _r("ALIQUOTA", ALREADY, v, reason="DECLARED_VALID_NOT_CONTRADICTED") else: out["ALIQUOTA"] = _r("ALIQUOTA", MANUAL, v, reason="TAXED_ITEM_WITHOUT_ICMS_RATE", key="credit_rate" if crt == "1" else "icms_rate", scope="COMPANY") # ---------------- CST IPI ---------------- v = cur["CST_IPI"] if company.get("contribuinte_ipi") != "S": out["CST_IPI"] = _r("CST_IPI", NA, v, reason="COMPANY_IS_NOT_AN_IPI_TAXPAYER_C000004.CONTRIBUINTE_IPI") elif v: out["CST_IPI"] = _r("CST_IPI", ALREADY, v, reason="DECLARED_VALID_NOT_CONTRADICTED") else: out["CST_IPI"] = _r("CST_IPI", MANUAL, v, reason="IPI_TAXPAYER_WITHOUT_CST_IPI", key="cst_ipi", scope="GROUP") # ---------------- PIS / COFINS ---------------- prof_raw, prof_scope = ctx.get("pis_cofins_profile", code, group) PIS_COFINS_FIELDS = ("COD_ENTRADA_PIS", "pis_in_cst", "cst_pis"), ("COD_SAIDA_PIS", "pis_out_cst", "cst_pis"), ("ALIQ_PIS", "pis_rate", None), \ ("COD_ENTRADA_COFINS", "cofins_in_cst", "cst_cofins"), ("COD_SAIDA_COFINS", "cofins_out_cst", "cst_cofins"), ("ALIQ_COFINS", "cofins_rate", None) if prof_raw in ("NOT_APPLICABLE", "OPERATION_TIME"): # Audited ADMIN decision only (Missão 2 / Bloco 2.4): an operator explicitly recorded, in FiscalContextEntry, # that this company/product's PIS/COFINS is never taken from the cadastral fields (real evidence: SGV emission # never reads them) or varies per operation. Never inferred automatically from emission statistics - # absent this exact recorded value, the field stays MANUAL (see the final `else` below), unchanged. for field, _pkey, _dom in PIS_COFINS_FIELDS: out[field] = _r(field, NA, cur[field], reason="ADMIN_DECISION_PIS_COFINS_" + prof_raw) prof = None else: import json as _json try: prof = _json.loads(prof_raw) if prof_raw else {} except ValueError: prof = {} for field, pkey, dom in ([] if prof is None else PIS_COFINS_FIELDS): v = cur[field] want = _s(prof.get(pkey)) if prof else None if want is not None and dom and want not in refs[dom]: out[field] = _r(field, MANUAL, v, reason="PROFILE_VALUE_NOT_IN_OFFICIAL_CST_TABLE", key="pis_cofins_profile", scope="COMPANY") elif want is not None: out[field] = _r(field, ALREADY if v == want else CHANGE, v, None if v == want else want, "OPERATOR_PIS_COFINS_PROFILE") elif v and (dom is None or v in refs[dom]): out[field] = _r(field, ALREADY, v, reason="DECLARED_VALID_NOT_CONTRADICTED") elif v: out[field] = _r(field, MANUAL, v, reason="DECLARED_CST_NOT_IN_OFFICIAL_TABLE", key="pis_cofins_profile", scope="COMPANY") elif crt == "1" and company.get("pis_cofins_unused_by_emission"): hits = sped_pis_cofins_match(NCM) if ncm_ok else [] csts = {h[1] for h in hits} if field in ("COD_SAIDA_PIS", "COD_SAIDA_COFINS") and len(csts) == 1 and role == "RESALE": want = next(iter(csts)) out[field] = _r(field, CHANGE, v, want, "SPED_TABLE_%s_ENTRY_%s_NCM_%s" % (hits[0][0], hits[0][2], NCM)) elif field in ("COD_SAIDA_PIS", "COD_SAIDA_COFINS") and len(csts) > 1: out[field] = _r(field, MANUAL, v, reason="SPED_TABLES_CONFLICT_FOR_NCM", key="pis_cofins_profile", scope="PRODUCT") elif field in ("ALIQ_PIS", "ALIQ_COFINS") and len(csts) == 1 and role == "RESALE": out[field] = _r(field, NA, v, reason="RATE_NOT_INFORMED_FOR_CST_" + next(iter(csts))) else: out[field] = _r(field, NA, v, reason="SGV_EMISSION_IGNORES_PIS_COFINS_PRODUCT_FIELDS_SIMPLES_NACIONAL") else: out[field] = _r(field, MANUAL, v, reason="NO_OFFICIAL_RULE_SOURCE_IN_CORE_DATASETS", key="pis_cofins_profile", scope="COMPANY") # ---------------- incidence / PISCOFINS / IAT / IPPT ---------------- v = cur["INCIDENCIA_PISCOFINS"] inc_ctx, _ = ctx.get("incidence", code, group) if inc_ctx == "NENHUMA": out["INCIDENCIA_PISCOFINS"] = _r("INCIDENCIA_PISCOFINS", ALREADY if v == "NENHUMA" else CHANGE, v, None if v == "NENHUMA" else "NENHUMA", "OPERATOR_INCIDENCE") elif v: out["INCIDENCIA_PISCOFINS"] = _r("INCIDENCIA_PISCOFINS", ALREADY, v, reason="DECLARED_VALID_NOT_CONTRADICTED") else: out["INCIDENCIA_PISCOFINS"] = _r("INCIDENCIA_PISCOFINS", MANUAL, v, reason="BLANK_INCIDENCE", key="incidence", scope="GROUP") v = cur["PISCOFINS"] pc_ctx, _ = ctx.get("piscofins", code, group) if pc_ctx in ("N", "S"): out["PISCOFINS"] = _r("PISCOFINS", ALREADY if v == pc_ctx else CHANGE, v, None if v == pc_ctx else pc_ctx, "OPERATOR_PISCOFINS") elif v in ("N", "S"): out["PISCOFINS"] = _r("PISCOFINS", ALREADY, v, reason="DECLARED_VALID_NOT_CONTRADICTED") else: out["PISCOFINS"] = _r("PISCOFINS", MANUAL, v, reason="BLANK_OR_INVALID", key="piscofins", scope="GROUP") v = cur["IAT"] rd_ctx, _ = ctx.get("rounding", code, group) if rd_ctx in ("A", "T"): out["IAT"] = _r("IAT", ALREADY if v == rd_ctx else CHANGE, v, None if v == rd_ctx else rd_ctx, "OPERATOR_ROUNDING_POLICY") elif v in ("A", "T"): out["IAT"] = _r("IAT", ALREADY, v, reason="DECLARED_VALID_NOT_CONTRADICTED") else: out["IAT"] = _r("IAT", MANUAL, v, reason="BLANK_OR_INVALID", key="rounding", scope="COMPANY") v = cur["IPPT"] if role in ("INPUT", "SERVICE"): out["IPPT"] = _r("IPPT", NA, v, reason="INPUT_OR_SERVICE_HAS_NO_PRODUCTION_INDICATOR") elif role == "UNKNOWN": out["IPPT"] = _r("IPPT", ALREADY, v, reason="DECLARED_VALID_NOT_CONTRADICTED") if v in ("P", "T") else _r("IPPT", MANUAL, v, reason="IPPT_NEEDS_PRODUCT_ROLE", key="product_role", scope="GROUP") else: exp = "P" if role == "OWN_PRODUCTION" else "T" out["IPPT"] = _r("IPPT", ALREADY if v == exp else CHANGE, v, None if v == exp else exp, "DERIVED_FROM_PRODUCT_ROLE:" + ",".join(role_ev)) for r in out.values(): r.setdefault("role", role) out["_facts"] = {"role": role, "role_evidence": role_ev, "st": st, "st_withheld": withheld, "icms_treatment": treat, "origin_digit": origin_digit} return out