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