GTI_GERENCIADOR_FISCAL/fiscal/central/app/icms_context.py

"""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"]