108 lines
6.5 KiB
Python
108 lines
6.5 KiB
Python
"""ICMS Context Engine V1: ST applicability, ST withholding evidence, product role, CSOSN decision V2.
|
|
|
|
Pure functions, no I/O, no writes (every output has write_eligible=False). History is only used to
|
|
validate/flag; it is never an input to a decision.
|
|
"""
|
|
ICMS_CONTEXT_ENGINE_VERSION = "icms-context-v1"
|
|
|
|
|
|
def _digits(v):
|
|
return "".join(c for c in str(v or "") if c.isdigit())
|
|
|
|
|
|
def st_status(ncm_status, mva_status, cest_status, erp_cest, ba_item_cest, ba_ncm_sh_len):
|
|
"""ST_CONFIRMED needs a unique BA Anexo 1 item whose CEST the ERP product itself carries
|
|
(or a confirmed CEST). ST_NOT_APPLICABLE needs BOTH official lists (BA Anexo 1 and CEST) silent
|
|
for a valid NCM. A CEST *candidate* never confirms ST."""
|
|
if ncm_status != "EXACT_NCM":
|
|
return "ST_UNKNOWN", "NCM_NOT_VALID"
|
|
if mva_status == "MVA_RESOLVED":
|
|
if cest_status == "CEST_CONFIRMED" or (_digits(erp_cest) and _digits(erp_cest) == _digits(ba_item_cest)):
|
|
return "ST_CONFIRMED", "UNIQUE_BA_ANNEX1_ITEM_CORROBORATED_BY_CEST"
|
|
return "ST_CANDIDATE", "UNIQUE_BA_ANNEX1_ITEM_WITHOUT_CEST_CORROBORATION"
|
|
if mva_status == "MVA_CANDIDATES":
|
|
return "ST_CANDIDATE", "MULTIPLE_BA_ANNEX1_ITEMS"
|
|
if mva_status == "NO_MVA_MAPPING_FOUND":
|
|
if cest_status == "NO_CEST_MAPPING_FOUND":
|
|
return "ST_NOT_APPLICABLE", "NCM_ABSENT_FROM_BA_ANNEX1_AND_CEST_LIST"
|
|
return "ST_UNKNOWN", "CEST_LISTED_BUT_BA_ANNEX1_ABSENT"
|
|
return "ST_UNKNOWN", "NO_ST_EVIDENCE"
|
|
|
|
|
|
def withheld_status(purchase_rows):
|
|
"""purchase_rows: entrada item rows for the product with keys icms_retido (S/N), valor_retido, sub_valor.
|
|
CONFIRMED only with positive purchase-side evidence; explicit 'N'/0 is NOT_CONFIRMED; no rows UNKNOWN."""
|
|
if not purchase_rows:
|
|
return "ST_WITHHOLDING_UNKNOWN", "NO_PURCHASE_ITEM_FOR_PRODUCT"
|
|
for r in purchase_rows:
|
|
if str(r.get("icms_retido") or "").strip().upper() == "S" or float(r.get("valor_retido") or 0) > 0 or float(r.get("sub_valor") or 0) > 0:
|
|
return "ST_ALREADY_WITHHELD_CONFIRMED", "PURCHASE_ITEM_SHOWS_WITHHELD_ICMS_ST"
|
|
return "ST_ALREADY_WITHHELD_NOT_CONFIRMED", "PURCHASE_ITEMS_FLAG_ICMS_RETIDO_N_AND_ZERO_ST_VALUES"
|
|
|
|
|
|
def product_role(erp_tipo, prod_cozinha, has_recipe, sold_as_composition, service_flag, purchase_evidence):
|
|
ev = []
|
|
if service_flag:
|
|
return "SERVICE", ["PRODUTO_SERVICO"]
|
|
if str(erp_tipo or "").strip().upper() == "MATERIA-PRIMA":
|
|
return "INPUT", ["C000025.TIPO=MATERIA-PRIMA"]
|
|
own = []
|
|
if prod_cozinha: own.append("PROD_COZINHA=1")
|
|
if has_recipe: own.append("RECEITA_INGREDIENTES")
|
|
if sold_as_composition: own.append("C000032.TIPO=COMPOSICAO")
|
|
if own and purchase_evidence:
|
|
return "UNKNOWN", own + ["ALSO_HAS_PURCHASE_EVIDENCE_CONFLICT"]
|
|
if own:
|
|
return "OWN_PRODUCTION", own
|
|
if str(erp_tipo or "").strip().upper() == "VENDA" and purchase_evidence:
|
|
return "RESALE", ["C000025.TIPO=VENDA", "PURCHASE_DOCUMENT_EVIDENCE"]
|
|
return "UNKNOWN", ev or ["NO_OBJECTIVE_ROLE_EVIDENCE"]
|
|
|
|
|
|
def csosn_v2(current, company_default, ncm_status, st, withheld, role, csosn_domain):
|
|
"""States: CSOSN_RESOLVED / CANDIDATE / CONTEXT_REQUIRED / CONFLICT / UNRESOLVED.
|
|
RESOLVED only where the fiscal condition is explicitly proven (ST confirmed + withholding confirmed + role
|
|
known -> 500). Everything else stays conservative; never assumes 102."""
|
|
cur = None if current is None or str(current).strip() == "" else str(current).strip()
|
|
out = {"field": "CSOSN", "current_value": cur, "proposed_csosn": None, "write_eligible": False, "engine": ICMS_CONTEXT_ENGINE_VERSION}
|
|
def res(state, reason, conf, action, proposed=None, **kw):
|
|
return {**out, "status": state, "reason_code": reason, "confidence": conf, "action_class": action, "proposed_csosn": proposed, **kw}
|
|
if ncm_status != "EXACT_NCM":
|
|
return res("CSOSN_UNRESOLVED", "NCM_NOT_VALID", "NONE", "NO_RULE")
|
|
if role in ("INPUT", "SERVICE"):
|
|
return res("CSOSN_CONTEXT_REQUIRED", f"ROLE_{role}_NOT_A_MERCHANDISE_SALE_DEFAULT", "NONE", "CONTEXT_REQUIRED")
|
|
if st == "ST_CONFIRMED":
|
|
if withheld == "ST_ALREADY_WITHHELD_CONFIRMED" and role in ("RESALE",):
|
|
state = "CSOSN_RESOLVED"
|
|
return res(state, "ST_CONFIRMED_AND_WITHHOLDING_CONFIRMED_RESALE", "HIGH", "ALREADY_CORRECT" if cur == "500" else "AUTO_APPLY_SAFE", "500")
|
|
if withheld == "ST_ALREADY_WITHHELD_NOT_CONFIRMED" and cur == "500":
|
|
return res("CSOSN_CONFLICT", "CSOSN_500_BUT_PURCHASE_ITEMS_SHOW_NO_ST_WITHHELD", "MEDIUM", "REVIEW_REQUIRED")
|
|
if withheld == "ST_ALREADY_WITHHELD_CONFIRMED" and cur != "500":
|
|
return res("CSOSN_CONFLICT", "ST_WITHHELD_CONFIRMED_BUT_CSOSN_NOT_500", "MEDIUM", "REVIEW_REQUIRED", "500")
|
|
return res("CSOSN_CONTEXT_REQUIRED", "ST_CONFIRMED_WITHHOLDING_OR_ROLE_UNPROVEN", "NONE", "CONTEXT_REQUIRED",
|
|
consistency_flag=None if cur == "500" else "ST_CONFIRMED_BUT_CSOSN_NOT_500")
|
|
if st in ("ST_CANDIDATE", "ST_UNKNOWN"):
|
|
return res("CSOSN_CONTEXT_REQUIRED", f"{st}_APPLICABILITY_UNPROVEN", "NONE", "CONTEXT_REQUIRED")
|
|
# ST_NOT_APPLICABLE: taxed vs exempt/not-taxed treatment (102 vs 400/300) has no source
|
|
if role in ("RESALE", "OWN_PRODUCTION") and cur == company_default and cur in csosn_domain:
|
|
return res("CSOSN_CANDIDATE", "NON_ST_MATCHES_COMPANY_DEFAULT_ICMS_TREATMENT_UNPROVEN", "MEDIUM", "REVIEW_REQUIRED", cur)
|
|
if role == "UNKNOWN":
|
|
return res("CSOSN_CONTEXT_REQUIRED", "PRODUCT_ROLE_UNKNOWN", "NONE", "CONTEXT_REQUIRED")
|
|
return res("CSOSN_CANDIDATE" if cur else "CSOSN_CONTEXT_REQUIRED", "DIFFERS_FROM_OR_LACKS_COMPANY_DEFAULT_TREATMENT_UNPROVEN", "LOW", "REVIEW_REQUIRED" if cur else "CONTEXT_REQUIRED")
|
|
|
|
|
|
def st_conflict_class(cest_status, mva_status, ncm_sh_len, erp_cest, ba_item_cest, erp_cfop, history_emitted_current, non_st_but_erp_5405):
|
|
"""Primary label for an ST-related CFOP/CSOSN conflict, plus all evidence flags."""
|
|
flags = []
|
|
if non_st_but_erp_5405 or (_digits(erp_cest) and _digits(ba_item_cest) and _digits(erp_cest) != _digits(ba_item_cest)):
|
|
flags.append("BA_ST_RULE_CONFLICT")
|
|
if history_emitted_current:
|
|
flags.append("ERP_HISTORY_CONFLICT")
|
|
if cest_status == "CEST_CANDIDATES":
|
|
flags.append("MULTIPLE_CEST")
|
|
if cest_status == "CEST_DESCRIPTION_MISMATCH":
|
|
flags.append("DESCRIPTION_MISMATCH")
|
|
if ncm_sh_len is not None and ncm_sh_len <= 4:
|
|
flags.append("NCM_TOO_BROAD")
|
|
return (flags[0] if flags else "NO_EVIDENCE"), flags or ["NO_EVIDENCE"]
|