"""Dry-run ERP write planner: turns classifier decisions into concrete field proposals without ever issuing a database write. Every proposal carries write_eligible; any future executor must refuse proposals where it is False.""" IPI_FIELD = "IPI" # C000025.IPI NUMERIC(15,2) is the real IPI rate field; C000025.ALIQ_IPI # is a VARCHAR(1) that is unused (0/521 rows) and cannot hold a TIPI rate or "NT" marker. EX_TIPI_FIELD = "EX_TIPI" CEST_FIELD = "CEST" NCM_FIELD = "CLASSIFICACAO_FISCAL" _CEST_UNRESOLVED_REASONS = { "CEST_CANDIDATES": "MULTIPLE_CEST_CANDIDATES_NO_SINGLE_EVIDENCE", "CEST_DESCRIPTION_MISMATCH": "CEST_DESCRIPTION_SCORE_BELOW_THRESHOLD", "NO_CEST_MAPPING_FOUND": "NO_CEST_MAPPING_FOUND_FOR_NCM", } def _provenance(decision): return { "engine_version": decision.get("engine_version"), "decision_version": decision.get("decision_version"), "request_fingerprint": decision.get("request_fingerprint"), } def _same(current, proposed): return str(current).strip() == str(proposed).strip() def build_field_proposals(product_code, decision, current): """current: dict with optional 'ipi' and 'cest' current ERP values. A missing/None key means the current ERP value is unknown, not that it is blank.""" dataset_versions = decision.get("dataset_versions", {}) classification = decision.get("classification", {}) if decision.get("status") == "LEGACY_NCM": return [{ "product_code": product_code, "field": NCM_FIELD, "current_value": classification.get("ncm"), "proposed_value": None, "status": "LEGACY_NCM_REVIEW", "reason_code": "LEGACY_NCM_NO_AUTOMATIC_SUCCESSOR", "source_dataset": "NCM", "source_version": dataset_versions.get("ncm"), "provenance": _provenance(decision), "write_eligible": False, }] proposals = [] ipi_status = decision.get("ipi_status") ipi = classification.get("ipi") if ipi_status == "IPI_RESOLVED" and ipi: current_ipi = current.get("ipi") current_ipi_read = "ipi" in current rate_kind = ipi.get("rate_kind") if rate_kind == "NUMERIC": proposed_rate = ipi.get("rate_numeric") if not current_ipi_read: status, write_eligible, reason = "CURRENT_UNKNOWN", False, "ERP_CURRENT_VALUE_NOT_READ" elif _same(current_ipi, proposed_rate): status, write_eligible, reason = "SAME_VALUE", False, "TIPI_MATCHES_CURRENT_ERP" else: status, write_eligible, reason = "CHANGE_REQUIRED", True, "TIPI_RATE_DIVERGES_FROM_ERP" elif rate_kind == "NON_NUMERIC": # e.g. TIPI text marker "NT" (não tributado) — never write text into a numeric field. proposed_rate = None rate_classification = ipi.get("rate_classification", "OTHER") status, write_eligible, reason = "NON_NUMERIC_IPI_STATUS", False, f"TIPI_RATE_{rate_classification}" else: # BLANK — source spreadsheet had no rate at all; never invent zero. proposed_rate = None rate_classification = ipi.get("rate_classification", "SOURCE_BLANK") status, write_eligible, reason = rate_classification, False, f"TIPI_RATE_{rate_classification}" proposals.append({ "product_code": product_code, "field": IPI_FIELD, "current_value": current_ipi, "proposed_value": proposed_rate, "status": status, "reason_code": reason, "source_dataset": "TIPI_IPI", "source_version": dataset_versions.get("tipi_ipi"), "provenance": _provenance(decision), "write_eligible": write_eligible, }) # C000025 has no EX_TIPI row under IPI_RESOLVED by construction: the matched # TIPI entry is the base (non-EX) row, so there is nothing concrete to propose. # EX TIPI only ever comes from EX-specific TIPI rows for this NCM (ipi_candidates). # No EX row for the NCM => no proposal at all (never propose a blank/None EX). ex_candidates = [c for c in classification.get("ipi_candidates", []) if c.get("ex")] if ex_candidates: current_ex = current.get("ex_tipi") current_ex_read = "ex_tipi" in current if len(ex_candidates) == 1: proposed_ex = ex_candidates[0]["ex"] if not current_ex_read: ex_status, ex_write_eligible, ex_reason = "CURRENT_UNKNOWN", False, "ERP_CURRENT_VALUE_NOT_READ" elif _same(current_ex, proposed_ex): ex_status, ex_write_eligible, ex_reason = "SAME_VALUE", False, "TIPI_EX_MATCHES_CURRENT_ERP" else: ex_status, ex_write_eligible, ex_reason = "CHANGE_REQUIRED", True, "TIPI_EX_DIVERGES_FROM_ERP" else: proposed_ex = None ex_status, ex_write_eligible, ex_reason = "CANDIDATE", False, "MULTIPLE_EX_TIPI_CANDIDATES_NO_SINGLE_EVIDENCE" proposals.append({ "product_code": product_code, "field": EX_TIPI_FIELD, "current_value": current_ex, "proposed_value": proposed_ex, "status": ex_status, "reason_code": ex_reason, "source_dataset": "TIPI_IPI", "source_version": dataset_versions.get("tipi_ipi"), "provenance": _provenance(decision), "write_eligible": ex_write_eligible, "candidate_count": len(ex_candidates), }) cest_status = decision.get("cest_status") if cest_status is not None: current_cest = current.get("cest") current_cest_read = "cest" in current if cest_status == "CEST_CONFIRMED": proposed_cest = (classification.get("cest") or [None])[0] if not current_cest_read: status, write_eligible, reason = "CURRENT_UNKNOWN", False, "ERP_CURRENT_VALUE_NOT_READ" elif _same(current_cest, proposed_cest): status, write_eligible, reason = "SAME_VALUE", False, "CEST_MATCHES_CURRENT_ERP" else: status, write_eligible, reason = "CHANGE_REQUIRED", True, "CEST_DIVERGES_FROM_ERP" else: proposed_cest = None write_eligible = False status = "CANDIDATE" if cest_status == "CEST_CANDIDATES" else "UNRESOLVED" reason = _CEST_UNRESOLVED_REASONS.get(cest_status, "CEST_NOT_RESOLVED") proposals.append({ "product_code": product_code, "field": CEST_FIELD, "current_value": current_cest, "proposed_value": proposed_cest, "status": status, "reason_code": reason, "source_dataset": "CEST", "source_version": dataset_versions.get("cest"), "provenance": _provenance(decision), "write_eligible": write_eligible, }) return proposals def mva_evidence(product_code, decision): """BA ST/MVA has no ERP field target: keep it as fiscal evidence only, never a write proposal.""" classification = decision.get("classification", {}) return { "product_code": product_code, "ncm": classification.get("ncm"), "status": decision.get("ba_st_mva_status"), "resolved": classification.get("ba_st_mva"), "candidates": classification.get("ba_st_mva_candidates", []), "source_version": decision.get("dataset_versions", {}).get("ba_st_mva"), } def split_by_eligibility(proposals): return { "eligible": [p for p in proposals if p.get("write_eligible")], "blocked": [p for p in proposals if not p.get("write_eligible")], } def execute(proposals): """Stub executor. Refuses any proposal that is not write_eligible and never issues a write itself — implementing the actual ERP UPDATE is out of scope.""" for proposal in proposals: if not proposal.get("write_eligible"): raise PermissionError( f"refused: {proposal['product_code']}/{proposal['field']} is not write_eligible " f"(status={proposal['status']})" ) raise NotImplementedError("no ERP write executor is implemented; this planner is dry-run only")