gti
d21a9d1140
git-subtree-dir: fiscal git-subtree-mainline:d95165fc8fgit-subtree-split:102ba84500
187 lines
8.2 KiB
Python
187 lines
8.2 KiB
Python
"""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")
|