103 lines
6.4 KiB
Python
103 lines
6.4 KiB
Python
"""Turns the deterministic cClassTrib resolution into the ONE root-cause RTC finding per product (pure).
|
|
|
|
`evaluate_product` (unified_fiscal_coverage) says WHETHER a product violates the RTC rule; this module says WHY and whether the gap is closable:
|
|
the single RTC blocker of a product is refined to the cause that matters, never several blockers for the same root cause.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from .cclasstrib_resolver import HUMAN_REASONS, Resolution
|
|
|
|
RTC_CATALOG_REASONS = ("LOCAL_OFFICIAL_", "CLASS_NOT_IN_LOCAL_CATALOG", "CLASS_NOT_IN_OFFICIAL_TABLE", "CLASS_NOT_YET_IN_FORCE", "CLASS_NOT_USABLE_IN_NFE_NFCE", "LOCAL_RTC_CATALOG_INVALID")
|
|
EVIDENCE_REASONS = ("NO_VALID_NCM", "COMPANY_REGIME_UNKNOWN")
|
|
RTC_TYPES = ("CCLASSTRIB_MISSING", "CCLASSTRIB_MISSING_UNRESOLVED", "CCLASSTRIB_AMBIGUOUS", "CCLASSTRIB_CONFLICT", "CCLASSTRIB_OFFICIAL_MISMATCH", "RTC_LOCAL_CATALOG_INCOMPATIBLE",
|
|
"RTC_RULE_EVIDENCE_MISSING", "CCLASSTRIB_INVALID_FORMAT", "CCLASSTRIB_NOT_IN_CATALOG", "RTC_CST_MISSING", "RTC_CATALOG_INVALID")
|
|
BAD_LOCAL_STATES = ("MISMATCH", "EXPIRED", "MISSING_OFFICIAL")
|
|
|
|
|
|
def _text(value) -> str:
|
|
return "" if value is None else str(value).strip()
|
|
|
|
|
|
def slim(resolution: Resolution) -> dict:
|
|
"""Small, audit-grade view kept in the job result (full evidence stays in the resolver output)."""
|
|
return {"status": resolution.status, "rule_id": resolution.rule_id, "rule_class": resolution.rule_class, "reason": resolution.reason, "resolved_cclasstrib": resolution.resolved_cclasstrib,
|
|
"resolved_cst_ibscbs": resolution.resolved_cst_ibscbs, "confidence_type": resolution.confidence_type, "auto_apply_candidate": resolution.auto_apply_candidate,
|
|
"suggested_cclasstrib": resolution.suggested_cclasstrib}
|
|
|
|
|
|
def refine_rtc_finding(found: dict, product: dict, resolution: Resolution | None, compat_state: dict | None) -> dict:
|
|
"""Mutates and returns `found` (an `evaluate_product` result). Only acts when RTC is REQUIRED for the product."""
|
|
if found.get("applicability", {}).get("RTC") != "REQUIRED" or resolution is None:
|
|
return found
|
|
value = _text(product.get("CCLASSTRIB"))
|
|
rtc_blockers = [b for b in found["blockers"] if b["category"] == "RTC"]
|
|
kinds = {b["type"] for b in rtc_blockers}
|
|
found["rtc_resolution"] = slim(resolution)
|
|
if "RTC_CATALOG_INVALID" in kinds:
|
|
return found # nothing can be proposed or compared against an invalid catalog
|
|
others = [b for b in found["blockers"] if b["category"] != "RTC"]
|
|
if "CCLASSTRIB_MISSING" in kinds:
|
|
if resolution.auto_apply_candidate:
|
|
new = ("CCLASSTRIB_MISSING", f"READY_TO_APPLY:{resolution.resolved_cclasstrib}:{resolution.rule_id}")
|
|
elif resolution.status == "SUGGESTION_ONLY":
|
|
new = ("CCLASSTRIB_MISSING_UNRESOLVED", f"{resolution.rule_id}:SUGGESTION_ONLY:{resolution.resolved_cclasstrib}")
|
|
elif resolution.status == "AMBIGUOUS":
|
|
new = ("CCLASSTRIB_AMBIGUOUS", f"{resolution.rule_id}:{resolution.reason}")
|
|
elif resolution.reason.startswith(RTC_CATALOG_REASONS):
|
|
new = ("RTC_LOCAL_CATALOG_INCOMPATIBLE", resolution.reason)
|
|
elif resolution.reason in EVIDENCE_REASONS:
|
|
new = ("RTC_RULE_EVIDENCE_MISSING", resolution.reason)
|
|
else:
|
|
new = ("CCLASSTRIB_MISSING_UNRESOLVED", f"{resolution.rule_id}:{resolution.reason}")
|
|
blocker = {"type": new[0], "category": "RTC", "reason": new[1]}
|
|
if resolution.reason in HUMAN_REASONS:
|
|
blocker["human"] = HUMAN_REASONS[resolution.reason]
|
|
found["blockers"] = others + [blocker]
|
|
return found
|
|
if not rtc_blockers and value: # a valid existing class: compare, never overwrite
|
|
state = (compat_state or {}).get(value)
|
|
if state in BAD_LOCAL_STATES:
|
|
found["blockers"] = others + [{"type": "CCLASSTRIB_OFFICIAL_MISMATCH", "category": "RTC", "reason": f"LOCAL_VS_OFFICIAL:{state}"}]
|
|
elif resolution.auto_apply_candidate and resolution.resolved_cclasstrib != value:
|
|
found["blockers"] = others + [{"type": "CCLASSTRIB_CONFLICT", "category": "RTC", "reason": f"HAS:{value};RULE:{resolution.rule_id}:{resolution.resolved_cclasstrib}"}]
|
|
else:
|
|
found["rtc_resolution"]["no_write_required"] = resolution.auto_apply_candidate and resolution.resolved_cclasstrib == value
|
|
return found
|
|
|
|
|
|
def summarize_rtc(findings: dict, compat: dict | None = None) -> dict:
|
|
"""Counts for the panel and the audit: valid, auto-resolvable (safe), suggested (diagnostic only), blocked and conflicting, catalog-incompatible."""
|
|
out = {"total": 0, "valid_existing": 0, "ready_to_apply": 0, "suggestion_only": 0, "ambiguous": 0, "insufficient": 0, "conflicts": 0, "official_mismatch": 0, "catalog_incompatible": 0,
|
|
"not_applicable": 0, "other_blocked": 0, "by_rule": {}, "by_reason": {}}
|
|
for found in findings.values():
|
|
out["total"] += 1
|
|
if found.get("applicability", {}).get("RTC") != "REQUIRED":
|
|
out["not_applicable"] += 1
|
|
continue
|
|
types = {b["type"] for b in found["blockers"] if b["category"] == "RTC"}
|
|
res = found.get("rtc_resolution") or {}
|
|
if res:
|
|
out["by_rule"][res["rule_id"]] = out["by_rule"].get(res["rule_id"], 0) + 1
|
|
out["by_reason"][res["reason"]] = out["by_reason"].get(res["reason"], 0) + 1
|
|
if not types:
|
|
out["valid_existing"] += 1
|
|
elif "CCLASSTRIB_CONFLICT" in types:
|
|
out["conflicts"] += 1
|
|
elif "CCLASSTRIB_OFFICIAL_MISMATCH" in types:
|
|
out["official_mismatch"] += 1
|
|
elif "RTC_LOCAL_CATALOG_INCOMPATIBLE" in types or "RTC_CATALOG_INVALID" in types:
|
|
out["catalog_incompatible"] += 1
|
|
elif res.get("auto_apply_candidate") and "CCLASSTRIB_MISSING" in types:
|
|
out["ready_to_apply"] += 1
|
|
elif res.get("status") == "SUGGESTION_ONLY":
|
|
out["suggestion_only"] += 1
|
|
elif "CCLASSTRIB_AMBIGUOUS" in types:
|
|
out["ambiguous"] += 1
|
|
elif types & {"CCLASSTRIB_MISSING_UNRESOLVED", "RTC_RULE_EVIDENCE_MISSING", "CCLASSTRIB_MISSING"}:
|
|
out["insufficient"] += 1
|
|
else:
|
|
out["other_blocked"] += 1
|
|
if compat:
|
|
out["catalog_compatibility"] = {"compatibility": compat["compatibility"], "counts": compat["counts"], "expired_present_locally": compat.get("expired_present_locally", 0)}
|
|
return out
|