96 lines
6.0 KiB
Python
96 lines
6.0 KiB
Python
"""Product Fiscal Defaults Engine V1 — C000025.CFOP / CSOSN / CST default decisions.
|
|
|
|
Pure functions. Scope: intra-UF (BA) NFC-e-style sale of merchandise, company CRT=1 (Simples).
|
|
Historical documents are NEVER an input here; they are used by the runner only to validate.
|
|
|
|
CFOP rule chain (all must hold, else CANDIDATE/CONTEXT/UNRESOLVED):
|
|
ERP CFOP mapping HIGH (product default, saída) + NCM EXACT + current CFOP in official table + first digit 5
|
|
(saída, intra-UF) + ERP declares resale family (5102/5405: 'adquirida ou recebida de terceiros')
|
|
+ ST applicability known from BA Anexo 1 (MVA_RESOLVED -> subject, NO_MVA_MAPPING_FOUND -> not subject)
|
|
=> derived CFOP = 5405 if subject else 5102. Equal to current => RESOLVED (already correct).
|
|
Different => CANDIDATE only: one of the inputs is wrong and the engine cannot tell which
|
|
(ST retention by the supplier is not available), so it is never write-eligible.
|
|
The chain never yields a write: filling a blank CFOP would need the resale-vs-production declaration that
|
|
only the current CFOP carries, so blank => OPERATION_CONTEXT_REQUIRED.
|
|
|
|
CSOSN: no dataset states whether a product is taxed (102), exempt/not taxed (400/300), ST-charged (500)...
|
|
=> never RESOLVED. Company default C000004.ICMS_CSOSN is only a CANDIDATE anchor for non-ST products.
|
|
"""
|
|
PRODUCT_DEFAULTS_ENGINE_VERSION = "product-defaults-v1"
|
|
RESALE_CFOPS = {"5102", "5405"}
|
|
|
|
|
|
def _blank(v):
|
|
return v is None or str(v).strip() == ""
|
|
|
|
|
|
def _out(field, current, proposed, state, reason, action, **extra):
|
|
return {"field": field, "current_value": current, "proposed_value": proposed, "status": state, "reason_code": reason,
|
|
"action_class": action, "write_eligible": False, "engine": PRODUCT_DEFAULTS_ENGINE_VERSION, **extra}
|
|
|
|
|
|
def st_applicability(ncm_status, mva_status):
|
|
if ncm_status != "EXACT_NCM":
|
|
return "UNKNOWN_NCM_NOT_VALID"
|
|
if mva_status == "MVA_RESOLVED":
|
|
return "SUBJECT"
|
|
if mva_status == "NO_MVA_MAPPING_FOUND":
|
|
return "NOT_SUBJECT"
|
|
return "UNKNOWN_UNCONFIRMED"
|
|
|
|
|
|
def cfop_default(product, ncm_status, mva_status, cfop_valid):
|
|
cur = None if _blank(product.get("CFOP")) else str(product["CFOP"]).strip()
|
|
st = st_applicability(ncm_status, mva_status)
|
|
base = {"st_applicability": st}
|
|
if st == "UNKNOWN_NCM_NOT_VALID":
|
|
return _out("CFOP", cur, None, "CFOP_UNRESOLVED", "NCM_NOT_VALID_CANNOT_ASSESS_ST", "NO_RULE", **base)
|
|
if cur is None:
|
|
return _out("CFOP", cur, None, "CFOP_OPERATION_CONTEXT_REQUIRED", "NO_DECLARED_RESALE_OR_PRODUCTION_FAMILY", "OPERATION_CONTEXT_REQUIRED", **base)
|
|
if cur not in cfop_valid:
|
|
return _out("CFOP", cur, None, "CFOP_UNRESOLVED", "CURRENT_CFOP_NOT_IN_OFFICIAL_TABLE", "REVIEW_REQUIRED", **base)
|
|
if cur[0] != "5":
|
|
return _out("CFOP", cur, None, "CFOP_OPERATION_CONTEXT_REQUIRED", "DIRECTION_OR_UF_SCOPE_NOT_INTRA_UF_SAIDA", "OPERATION_CONTEXT_REQUIRED", **base)
|
|
if cur not in RESALE_CFOPS:
|
|
return _out("CFOP", cur, None, "CFOP_OPERATION_CONTEXT_REQUIRED", "NO_RULE_FOR_DECLARED_CFOP_FAMILY", "OPERATION_CONTEXT_REQUIRED", **base)
|
|
if st == "UNKNOWN_UNCONFIRMED":
|
|
return _out("CFOP", cur, None, "CFOP_OPERATION_CONTEXT_REQUIRED", "ST_APPLICABILITY_UNCONFIRMED_NEEDS_CEST_CONFIRMATION", "OPERATION_CONTEXT_REQUIRED", **base)
|
|
derived = "5405" if st == "SUBJECT" else "5102"
|
|
if derived == cur:
|
|
return _out("CFOP", cur, derived, "CFOP_DEFAULT_RESOLVED", "CONSISTENT_WITH_ST_STATUS_SAME_VALUE", "ALREADY_CORRECT", **base)
|
|
return _out("CFOP", cur, derived, "CFOP_DEFAULT_CANDIDATE", "ST_STATUS_CONFLICTS_WITH_ERP_CFOP_RETENTION_CONTEXT_MISSING", "REVIEW_REQUIRED", **base)
|
|
|
|
|
|
def csosn_default(product, ncm_status, mva_status, csosn_valid, company_default):
|
|
cur = None if _blank(product.get("CSOSN")) else str(product["CSOSN"]).strip()
|
|
st = st_applicability(ncm_status, mva_status)
|
|
base = {"st_applicability": st, "company_default": company_default}
|
|
if st == "UNKNOWN_NCM_NOT_VALID":
|
|
return _out("CSOSN", cur, None, "CSOSN_UNRESOLVED", "NCM_NOT_VALID_CANNOT_ASSESS_ST", "NO_RULE", **base)
|
|
if st == "UNKNOWN_UNCONFIRMED":
|
|
return _out("CSOSN", cur, None, "CSOSN_OPERATION_CONTEXT_REQUIRED", "ST_APPLICABILITY_UNCONFIRMED", "OPERATION_CONTEXT_REQUIRED", **base)
|
|
if st == "SUBJECT":
|
|
flag = "ST_SUBJECT_BUT_CSOSN_NOT_500" if cur != "500" else "ST_CSOSN_500_RETENTION_UNVERIFIED"
|
|
return _out("CSOSN", cur, None, "CSOSN_OPERATION_CONTEXT_REQUIRED", "ST_RETENTION_BY_SUPPLIER_UNKNOWN", "OPERATION_CONTEXT_REQUIRED", consistency_flag=flag, **base)
|
|
# NOT_SUBJECT: 102 / 400 / 300 / 103 depend on the product's ICMS treatment, for which no source exists
|
|
if cur is not None and cur == company_default and cur in csosn_valid:
|
|
return _out("CSOSN", cur, cur, "CSOSN_DEFAULT_CANDIDATE", "MATCHES_COMPANY_DEFAULT_PRODUCT_ICMS_TREATMENT_UNPROVEN", "REVIEW_REQUIRED", **base)
|
|
if cur is None:
|
|
return _out("CSOSN", cur, company_default, "CSOSN_DEFAULT_CANDIDATE", "BLANK_COMPANY_DEFAULT_ONLY_ANCHOR_TREATMENT_UNPROVEN", "REVIEW_REQUIRED", **base)
|
|
return _out("CSOSN", cur, None, "CSOSN_DEFAULT_CANDIDATE", "DIFFERS_FROM_COMPANY_DEFAULT_NO_TREATMENT_SOURCE", "REVIEW_REQUIRED", **base)
|
|
|
|
|
|
def cst_status(product, cst_icms_codes, origem_codes):
|
|
cur = None if _blank(product.get("CST")) else str(product["CST"]).strip()
|
|
valid = cur is not None and len(cur) == 3 and cur[1:] in cst_icms_codes and cur[0] in origem_codes
|
|
return {"field": "CST", "current_value": cur, "status": "CST_REFERENCE_ONLY" if valid else "CST_INVALID_REVIEW",
|
|
"reason_code": "CRT1_EMISSION_USES_CSOSN_CST_IS_ORIGIN_PLUS_CST" if valid else "NOT_ORIGIN_DIGIT_PLUS_VALID_CST",
|
|
"origin_digit": cur[0] if valid else None, "write_eligible": False, "engine": PRODUCT_DEFAULTS_ENGINE_VERSION}
|
|
|
|
|
|
def validate_against_history(proposed, emitted_values):
|
|
"""History validates only; returns MATCH / MISMATCH / UNVALIDATED."""
|
|
if not emitted_values or proposed is None:
|
|
return "UNVALIDATED"
|
|
return "MATCH" if set(emitted_values) == {proposed} else "MISMATCH"
|