GTI_GERENCIADOR_FISCAL/fiscal/central/app/rtc_findings.py

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