gtiandClaude Sonnet 5
e935ed61d5
Missao 2 (fechamento de lacunas fiscais, read-only + UI/engine gated): - field_contract: pis_cofins_profile now also accepts the literal sentinels NOT_APPLICABLE/OPERATION_TIME, resolving to NA only from an audited FiscalContextEntry - never inferred from emission statistics. CFOP_ENTRADA under an already-decided entry_scope=MIXED_OR_DYNAMIC now resolves to NA instead of MANUAL, so it stops blocking product completeness and stops re-asking an already-answered question. Regression tests prove: no ADMIN decision => unchanged old behavior; decision present => gated bucket, no write. - context_keys: validate_entry accepts the two sentinels for pis_cofins_profile. - Contexto Fiscal UI: PIS/COFINS and CFOP_ENTRADA cards show the Bloco 2 cross-database evidence (5 installations, CRT 1 and 3: emission never reads the PIS/COFINS cadastral fields anywhere) and a link to an ADMIN-facing, anonymized dossie-for-the-accountant; entry_scope is now a dropdown. - release/tests: decoupled the Hub's GTI_FISCAL_VERSION from the Agent's own version (test_versions_are_coherent_across_hub_and_packages vs the new test_agent_version_is_internally_coherent) - a Hub-only release must not require touching, rebuilding or republishing the Agent. - New read-only evidence: docs/runs/GTI_dryrun_universe_summary.json (527 products, Agent 2's real database), docs/runs/BLOCO2_PIS_COFINS_CROSS_DB_V1.json (5 installations), docs/fiscal_completeness_matrix.md, docs/dossie_contador.md (internal, real names) + central/static/docs/dossie_contador.md (public, anonymized - no CNPJ/razao social/product names), docs/AGENT2_UPGRADE_RUNBOOK_V1.md. - CLAUDE.md: recorded the permanent fiscal-mission rules (no AI/scraping, no Firebird schema changes, deploy only via deploy_hub.ps1, never reenroll, no remote access to client Agent hosts, no client data in unauthenticated static routes). No schema/migration change. Zero Firebird writes (new scripts never open a write connection). Hub only - Agent code untouched. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
555 lines
42 KiB
Python
555 lines
42 KiB
Python
"""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(<input_key>) (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 _r(field, state, current, proposed=None, reason="", key=None, scope=None, **extra):
|
|
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}
|
|
|
|
|
|
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"]
|
|
own = [n for n, on in (("PROD_COZINHA=1", p.get("PROD_COZINHA") == 1), ("RECEITA_INGREDIENTES", code in ev.get("recipes", ())), ("C000032.TIPO=COMPOSICAO", code in ev.get("composed", ()))) 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")
|
|
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")
|
|
|
|
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: 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)
|
|
if prof == "LEAVE_NOT_APPLICABLE":
|
|
out["IPI"] = _r("IPI", NA, cur["IPI"], reason="STORE_IPI_PROFILE_LEAVE_NOT_APPLICABLE")
|
|
elif prof != "POPULATE_REFERENCE_RATE":
|
|
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")]
|
|
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":
|
|
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")
|
|
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
|