"""Deterministic, auditable cClassTrib resolver (pure: no Firebird, no network, no probabilities). resolve_cclasstrib(product, company, ctx, local_catalog) -> Resolution Nothing here guesses. A class is proposed only when an EXPLICIT rule, backed by the official texts in `rtc_reference_data` (cClassTrib table 2026-06-22 and LC 214/2025 compiled through LC 227/2026), proves it. Everything else is AMBIGUOUS or INSUFFICIENT_EVIDENCE. Operation assumption (documented, product-level default): an ordinary domestic sale of the good. Operation-dependent classes (buyer is a public body, exports, ZFM/ALC, credit presumed, ...) cannot be carried by a product default; when the annex offers such alternatives the answer is AMBIGUOUS. Rule classes (what a rule is ALLOWED to do) AUTO_APPLY_SAFE positive, explicit evidence: may be written to the ERP (only when every runtime gate also holds) VALIDATION_ONLY used to validate/explain, never proposes a value SUGGESTION_ONLY diagnostic suggestion shown to a human; NEVER authorizes an UPDATE BLOCKED_BY_EVIDENCE the evidence is missing/ambiguous: safe-block with a human reason Rules RTC-R0 RTC disabled in the ERP VALIDATION_ONLY -> NOT_APPLICABLE RTC-G1 local RTC catalog failed integrity BLOCKED_BY_EVIDENCE -> INSUFFICIENT_EVIDENCE RTC-R1 regime without an official rule (Simples family, unknown) BLOCKED_BY_EVIDENCE -> INSUFFICIENT_EVIDENCE RTC-R2 no valid 8-digit NCM (decision engine did not confirm) BLOCKED_BY_EVIDENCE -> INSUFFICIENT_EVIDENCE RTC-A1 annex without a unique product-default class / conditional text / revoked annex -> AMBIGUOUS RTC-A2 NCM listed in more than one annex -> AMBIGUOUS RTC-A3 chapter whose regime depends on the nature/operation of the good -> AMBIGUOUS RTC-R3 NCM listed in EXACTLY ONE annex with a unique default class, condition-free item, regime normal (CRT 3) AUTO_APPLY_SAFE -> RESOLVED_WITH_EXPLICIT_RULE RTC-R4 NCM in NO annex and in no nature-based chapter: full taxation is the residual reading of LC 214 art. 4 SUGGESTION_ONLY -> SUGGESTION_ONLY (000001). Absence of a special regime is NOT positive evidence: never written. RTC-G2 proposed class missing/different in the local SGV catalog, out of force or unusable in NF-e/NFC-e BLOCKED_BY_EVIDENCE -> INSUFFICIENT_EVIDENCE (RTC_LOCAL_CATALOG_INCOMPATIBLE) A proposed class must additionally be MATCH_SAFE against the official table (local SGV catalog confirms every attribute that matters), in force today, usable in NF-e/NFC-e (the SGV's own lookup filter) and carry a CST that exists; otherwise the answer degrades with the exact reason. """ from __future__ import annotations import hashlib import json import re from dataclasses import dataclass, field from datetime import date from . import rtc_reference_data as ref SIMPLES_FAMILY = frozenset({"1", "2", "4"}) RESOLVED_STATUSES = ("RESOLVED_DETERMINISTIC", "RESOLVED_WITH_EXPLICIT_RULE") STATUSES = RESOLVED_STATUSES + ("SUGGESTION_ONLY", "AMBIGUOUS", "INSUFFICIENT_EVIDENCE", "NOT_APPLICABLE") RULE_VERSION = "rtc-rules-v2" AUTO_APPLY_SAFE, VALIDATION_ONLY, SUGGESTION_ONLY, BLOCKED_BY_EVIDENCE = "AUTO_APPLY_SAFE", "VALIDATION_ONLY", "SUGGESTION_ONLY", "BLOCKED_BY_EVIDENCE" RULE_CLASSES = {"RTC-R0": VALIDATION_ONLY, "RTC-G1": BLOCKED_BY_EVIDENCE, "RTC-R1": BLOCKED_BY_EVIDENCE, "RTC-R2": BLOCKED_BY_EVIDENCE, "RTC-A1": BLOCKED_BY_EVIDENCE, "RTC-A2": BLOCKED_BY_EVIDENCE, "RTC-A3": BLOCKED_BY_EVIDENCE, "RTC-R3": AUTO_APPLY_SAFE, "RTC-R4": SUGGESTION_ONLY, "RTC-G2": BLOCKED_BY_EVIDENCE} HUMAN_REASONS = {"SIMPLES_ORIENTATION_PENDING_OFFICIAL_NT": "Enquadramento RTC depende de regra ainda não determinável com segurança.", "COMPANY_REGIME_UNKNOWN": "Regime tributário da empresa não determinado."} # Goods annexes whose product-default class is unique in the official table (verified by `verify_annex_class_map`). ANNEX_DEFAULT_CLASS = {"I": "200003", "VII": "200034", "VIII": "200035", "XV": "200014"} # IX is out: 515001 (deferral with the same inputs) is a sibling class # Chapters where a reduced/special regime is defined by the NATURE of the good (not by a code list): in natura agriculture (art. 137), medicines (arts. 133/146), # fuels and biofuels (art. 172 ff), medical/accessibility devices and vehicles (arts. 131-132, 143-145, 149), books/periodicals (constitutional immunity), # gold and other special-regime goods kept out of the residual rule on purpose (conservative). SENSITIVE_CHAPTERS = frozenset([f"{n:02d}" for n in range(1, 15)] + ["27", "30", "49", "71", "87", "88", "89", "90"]) # 71 gold (financial-asset regime), 88/89 aircraft and vessels SPECIAL_PREFIXES = ("9619", "2207", "3826", "8703") # art. 147 (menstrual health), biofuel inputs, art. 149 (PcD cars) @dataclass class Resolution: status: str rule_id: str reason: str resolved_cclasstrib: str | None = None resolved_cst_ibscbs: str | None = None confidence_type: str = "NONE" inputs_used: dict = field(default_factory=dict) evidence: list = field(default_factory=list) source_references: list = field(default_factory=list) @property def rule_class(self) -> str: return RULE_CLASSES.get(self.rule_id, BLOCKED_BY_EVIDENCE) @property def auto_apply_candidate(self) -> bool: """Only an AUTO_APPLY_SAFE rule with a proven class may ever become a write. Suggestions never do.""" return self.rule_class == AUTO_APPLY_SAFE and self.status in RESOLVED_STATUSES and bool(self.resolved_cclasstrib) @property def suggested_cclasstrib(self) -> str | None: return self.resolved_cclasstrib if self.status == "SUGGESTION_ONLY" else None def as_dict(self) -> dict: return {"status": self.status, "rule_id": self.rule_id, "rule_class": self.rule_class, "rule_version": RULE_VERSION, "reason": self.reason, "resolved_cclasstrib": self.resolved_cclasstrib, "resolved_cst_ibscbs": self.resolved_cst_ibscbs, "confidence_type": self.confidence_type, "inputs_used": self.inputs_used, "evidence": self.evidence, "source_references": self.source_references, "auto_apply_candidate": self.auto_apply_candidate} def _digits(value) -> str: return re.sub(r"\D", "", "" if value is None else str(value)) def _official_by_code() -> dict: return {row["code"]: row for row in ref.OFFICIAL_CCLASSTRIB} def _num(value): try: return float(str(value).replace(",", ".")) except (TypeError, ValueError): return None # ------------------------------------------------------------------------------------------------- catalog comparison def compare_catalogs(local_rows: dict, today: date | None = None) -> dict: """local_rows: {code: {cst, tipo_aliquota, p_red_ibs, p_red_cbs, ind_nfe, ind_nfce, start, end}} read from the ERP. Never writes anywhere. Per code (the official table is the reference): MATCH_SAFE every attribute the local table carries that matters to a product default agrees (CST, reductions, NF-e/NFC-e usability) MATCH_WITH_NON_MATERIAL_DIFFERENCE only a scheduled local end date (class still in force today) differs MISMATCH CST, reductions or NF-e/NFC-e usability differ EXPIRED the official class ended MISSING_LOCAL official class absent from the SGV catalog (local outdated) MISSING_OFFICIAL local class the official table does not know""" today = today or date.today() official = _official_by_code() result, details = {}, {} for code in sorted(set(local_rows) | set(official)): loc, off = local_rows.get(code), official.get(code) if off is None: result[code] = "MISSING_OFFICIAL" elif off.get("end") and date.fromisoformat(off["end"]) < today: result[code] = "EXPIRED" details[code] = ["OFFICIAL_END_PASSED"] + (["PRESENT_LOCALLY"] if loc is not None else []) elif loc is None: result[code] = "MISSING_LOCAL" else: problems, soft, info = [], [], [] if str(loc.get("cst") or "").strip() != off["cst"]: problems.append("CST") if "p_red_ibs" in loc and "p_red_cbs" in loc and (_num(loc.get("p_red_ibs")) != _num(off["p_red_ibs"]) or _num(loc.get("p_red_cbs")) != _num(off["p_red_cbs"])): problems.append("REDUCTION") # only attributes the local table actually carries are compared if "ind_nfe" in loc and "ind_nfce" in loc and (bool(loc.get("ind_nfe")) != off["ind_nfe"] or bool(loc.get("ind_nfce")) != off["ind_nfce"]): problems.append("DOCUMENT_FLAGS") start, end = str(loc.get("start") or "")[:10], str(loc.get("end") or "")[:10] if start and date.fromisoformat(start) > today: problems.append("LOCAL_NOT_IN_FORCE_YET") elif start and off.get("start") and start != off["start"]: info.append("VIGENCY_START_DIFFERS_BOTH_IN_FORCE") # informational: the class is in force today on both sides, so the emission effect is identical if end and date.fromisoformat(end) < today: problems.append("LOCAL_CLASS_ENDED") elif end: soft.append("LOCAL_END_SCHEDULED") # the class still works today; the local table schedules an end the official one does not result[code] = "MISMATCH" if problems else ("MATCH_WITH_NON_MATERIAL_DIFFERENCE" if soft else "MATCH_SAFE") if problems or soft or info: details[code] = problems + soft + info keys = ("MATCH_SAFE", "MATCH_WITH_NON_MATERIAL_DIFFERENCE", "MISMATCH", "EXPIRED", "MISSING_LOCAL", "MISSING_OFFICIAL") counts = {k: sum(1 for v in result.values() if v == k) for k in keys} material = counts["MISMATCH"] + counts["MISSING_OFFICIAL"] # a local class the official table contradicts or does not know expired_present = sum(1 for d in details.values() if "PRESENT_LOCALLY" in d) compat = "FULL" if counts["MISSING_LOCAL"] == 0 and expired_present == 0 and material == 0 else ("PARTIAL_LOCAL_OUTDATED" if material == 0 else "MATERIAL_MISMATCH") return {"per_code": result, "details": details, "counts": counts, "expired_present_locally": expired_present, "compatibility": compat, "official_catalog": {"sha256": ref.OFFICIAL_CATALOG["sha256"], "publication": ref.OFFICIAL_CATALOG["publication"], "file": ref.OFFICIAL_CATALOG["file"]}} def local_catalog_hash(local_rows: dict) -> str: canon = json.dumps({c: {k: (None if v is None else str(v)) for k, v in sorted(r.items())} for c, r in sorted(local_rows.items())}, sort_keys=True) return hashlib.sha256(canon.encode()).hexdigest() def verify_annex_class_map() -> dict: """Self-check of ANNEX_DEFAULT_CLASS against the official table: the class must exist, be a goods class in force, and be the ONLY official class for that annex usable in NF-e/NFC-e as the ordinary-sale default. Returns {annex: [problems]} (empty lists = verified).""" official = _official_by_code() numbers = {"I": "1", "VII": "7", "VIII": "8", "XV": "15"} out = {} for annex, code in ANNEX_DEFAULT_CLASS.items(): row, problems = official.get(code), [] if row is None: problems.append("CLASS_NOT_IN_OFFICIAL_TABLE") else: if str(row.get("anexo")) != numbers[annex]: problems.append("ANNEX_FIELD_DIFFERS") if not (row["ind_nfe"] or row["ind_nfce"]): problems.append("NOT_USABLE_IN_NFE_NFCE") siblings = [r["code"] for r in ref.OFFICIAL_CCLASSTRIB if str(r.get("anexo")) == numbers[annex] and r["code"] != code and (r["ind_nfe"] or r["ind_nfce"])] if siblings: problems.append("SIBLING_CLASSES:" + ",".join(siblings)) out[annex] = problems return out # ------------------------------------------------------------------------------------------------- annex lookup def annex_matches(ncm: str) -> list[dict]: """Every annex item (any annex, revoked included) whose code list covers the NCM, with the item text and whether it is a clean (condition-free) item.""" hits = [] for roman, annex in ref.LC214_ANNEXES.items(): for item in annex["items"]: for prefix in item["prefixes"]: if ncm.startswith(prefix): hits.append({"annex": roman, "item": item["item"], "prefix": prefix, "clean": item["clean"], "revoked": annex["revoked"], "text": item["description"][:160]}) break return hits def _references(extra=None) -> list: refs = [{"document": ref.LC214_SOURCE["document"], "url": ref.LC214_SOURCE["url"], "sha256": ref.LC214_SOURCE["sha256"]}, {"document": "cClassTrib (Portal Nacional NF-e)", "publication": ref.OFFICIAL_CATALOG["publication"], "sha256": ref.OFFICIAL_CATALOG["sha256"]}] return refs + (extra or []) def _validate_class(code: str, local_rows: dict, compat: dict, today: date): """None when the class may be proposed, else the exact reason it may not.""" official = _official_by_code().get(code) if official is None: return "CLASS_NOT_IN_OFFICIAL_TABLE" loc = local_rows.get(code) if loc is None: return "CLASS_NOT_IN_LOCAL_CATALOG" state = compat["per_code"].get(code) if state != "MATCH_SAFE": return f"LOCAL_OFFICIAL_{state}" if official["start"] and date.fromisoformat(official["start"]) > today: return "CLASS_NOT_YET_IN_FORCE" if not (official["ind_nfe"] or official["ind_nfce"]): return "CLASS_NOT_USABLE_IN_NFE_NFCE" return None # ------------------------------------------------------------------------------------------------- resolver def resolve_cclasstrib(product: dict, company: dict | None, ctx: dict, local_rows: dict | None, *, compat: dict | None = None, today: date | None = None) -> Resolution: """product: {"CODIGO","CLASSIFICACAO_FISCAL","ncm_valid"(bool)}; company: {"crt"}; ctx: {"rtc_enabled": bool, "local_catalog_valid": bool}.""" today = today or date.today() ncm = _digits(product.get("CLASSIFICACAO_FISCAL")) crt = str((company or {}).get("crt") or "").strip() used = {"ncm": ncm or None, "crt": crt or None, "rtc_enabled": bool(ctx.get("rtc_enabled"))} if not ctx.get("rtc_enabled"): return Resolution("NOT_APPLICABLE", "RTC-R0", "RTC_DISABLED_IN_ERP", inputs_used=used) if not ctx.get("local_catalog_valid"): return Resolution("INSUFFICIENT_EVIDENCE", "RTC-G1", "LOCAL_RTC_CATALOG_INVALID", inputs_used=used) if crt in SIMPLES_FAMILY: return Resolution("INSUFFICIENT_EVIDENCE", "RTC-R1", "SIMPLES_ORIENTATION_PENDING_OFFICIAL_NT", inputs_used=used, source_references=_references([{"document": "NT 2025.002-RTC v1.51 (Portal NF-e, 04/08/2026)", "note": "IBS/CBS orientations for CRT 1/2/4 pending"}, {"document": "Ato Conjunto RFB/CGIBS n 4/2026 (DOU 31/07/2026)", "note": "art. 1 par. 1: DF-e obligation for Simples optantes starts 2027-01-01"}])) if crt != "3": return Resolution("INSUFFICIENT_EVIDENCE", "RTC-R1", "COMPANY_REGIME_UNKNOWN", inputs_used=used) if len(ncm) != 8 or product.get("ncm_valid") is not True: return Resolution("INSUFFICIENT_EVIDENCE", "RTC-R2", "NO_VALID_NCM", inputs_used=used) local_rows = local_rows or {} compat = compat or compare_catalogs(local_rows, today) hits = annex_matches(ncm) annexes = sorted({h["annex"] for h in hits}) if hits: if any(h["revoked"] for h in hits) and len(annexes) == 1: return Resolution("AMBIGUOUS", "RTC-A1", "LISTED_ONLY_IN_REVOKED_ANNEX", inputs_used=used, evidence=hits, source_references=_references()) live = [h for h in hits if not h["revoked"]] live_annexes = sorted({h["annex"] for h in live}) if len(live_annexes) > 1 or (len(live_annexes) == 1 and len(annexes) > 1): return Resolution("AMBIGUOUS", "RTC-A2", "LISTED_IN_MORE_THAN_ONE_ANNEX", inputs_used=used, evidence=hits, source_references=_references()) annex = live_annexes[0] if live_annexes else annexes[0] clean = [h for h in live if h["clean"]] cls = ANNEX_DEFAULT_CLASS.get(annex) if not cls or not clean or len(clean) != len(live): return Resolution("AMBIGUOUS", "RTC-A1", f"ANNEX_{annex}_NEEDS_OPERATION_OR_CONDITION", inputs_used=used, evidence=hits, source_references=_references()) problem = _validate_class(cls, local_rows, compat, today) if problem: return Resolution("INSUFFICIENT_EVIDENCE", "RTC-G2", problem, inputs_used=used, evidence=hits, source_references=_references()) official = _official_by_code()[cls] return Resolution("RESOLVED_WITH_EXPLICIT_RULE", "RTC-R3", f"ANNEX_{annex}_CLEAN_ITEM_UNIQUE_DEFAULT_CLASS", cls, official["cst"], "ANNEX_TEXT_MATCH", used, hits, _references()) if ncm[:2] in SENSITIVE_CHAPTERS or ncm.startswith(SPECIAL_PREFIXES): return Resolution("AMBIGUOUS", "RTC-A3", "SPECIAL_REGIME_DEPENDS_ON_NATURE_OR_OPERATION_OF_THE_GOOD", inputs_used=used, evidence=[{"chapter": ncm[:2]}], source_references=_references()) problem = _validate_class("000001", local_rows, compat, today) if problem: return Resolution("INSUFFICIENT_EVIDENCE", "RTC-G2", problem, inputs_used=used, source_references=_references()) return Resolution("SUGGESTION_ONLY", "RTC-R4", "NCM_IN_NO_ANNEX_AND_NO_NATURE_BASED_CHAPTER", "000001", "000", "EXCLUSION_OF_ALL_SPECIAL_REGIMES", used, [{"annex_hits": 0, "chapter": ncm[:2], "note": "absence of a special regime is not positive evidence: suggestion only, never written"}], _references())