gtiandClaude Sonnet 5
9f08998800
- resolver (rules RTC-R0..R4/A1..A3/G1..G2) over the official cClassTrib table 2026-06-22 and LC 214/2025 annexes; local vs official catalog comparison - ONE root-cause RTC blocker per product (missing/unresolved/ambiguous/conflict/official mismatch/catalog outdated/evidence missing) - auto-apply: C000025.CCLASSTRIB only, three switches (FIREBIRD_WRITES + RTC_AUTO_APPLY_ENABLED + Hub authorization), per-product re-read and re-decision, never overwrites, rowcount 1, audit table rtc_write_audits (migration 0027) - copy-only write E2E (280/280 applied, only CCLASSTRIB changed, 2nd run 0 writes, stale decision 0 writes); real databases untouched - docs: cclasstrib_resolution, rtc_auto_apply, unified_fiscal_completeness Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
234 lines
14 KiB
Python
234 lines
14 KiB
Python
"""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.
|
|
|
|
Rules
|
|
RTC-R0 RTC disabled in the ERP -> NOT_APPLICABLE
|
|
RTC-R1 Simples family (CRT 1/2/4) -> INSUFFICIENT_EVIDENCE (NT 2025.002 v1.51: orientation for these taxpayers pending)
|
|
RTC-R2 no valid 8-digit NCM (decision engine did not confirm) -> INSUFFICIENT_EVIDENCE
|
|
RTC-A1 NCM listed in a REVOKED annex or an annex without a unique product-default class, or under a conditional item text -> AMBIGUOUS
|
|
RTC-A2 NCM listed in more than one annex -> AMBIGUOUS
|
|
RTC-A3 NCM in a chapter whose special regime depends on the nature/operation of the good, not on a code list -> AMBIGUOUS
|
|
RTC-R3 NCM listed in exactly one annex whose product-default class is unique, item text has no condition -> RESOLVED_WITH_EXPLICIT_RULE
|
|
RTC-R4 NCM listed in NO annex and in no nature-based chapter -> RESOLVED_WITH_EXPLICIT_RULE 000001 (integral taxation, LC 214 art. 4 general rule)
|
|
A proposed class must additionally be present in the local ERP catalog, MATCH the official table, be in force today, be usable in NF-e/NFC-e (the SGV's own
|
|
lookup filter) and carry a CST that exists; otherwise the answer degrades to INSUFFICIENT_EVIDENCE 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 + ("AMBIGUOUS", "INSUFFICIENT_EVIDENCE", "NOT_APPLICABLE")
|
|
|
|
# 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 auto_apply_candidate(self) -> bool:
|
|
return self.status in RESOLVED_STATUSES and bool(self.resolved_cclasstrib)
|
|
|
|
def as_dict(self) -> dict:
|
|
return {"status": self.status, "rule_id": self.rule_id, "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: MATCH | LOCAL_ONLY | OFFICIAL_ONLY | MISMATCH | EXPIRED (official table is the reference)."""
|
|
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] = "LOCAL_ONLY"
|
|
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] = "OFFICIAL_ONLY"
|
|
else:
|
|
problems = []
|
|
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")
|
|
result[code] = "MISMATCH" if problems else "MATCH"
|
|
if problems:
|
|
details[code] = problems
|
|
counts = {k: sum(1 for v in result.values() if v == k) for k in ("MATCH", "LOCAL_ONLY", "OFFICIAL_ONLY", "MISMATCH", "EXPIRED")}
|
|
material = counts["MISMATCH"] + counts["LOCAL_ONLY"] # a local class the official table contradicts or does not know
|
|
expired_present = sum(1 for c, d in details.items() if "PRESENT_LOCALLY" in d)
|
|
compat = "FULL" if counts["OFFICIAL_ONLY"] == 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":
|
|
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; taxation from 2027 (LC 214 art. 348)"}]))
|
|
if not crt:
|
|
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("RESOLVED_WITH_EXPLICIT_RULE", "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]}], _references())
|