feat: central RTC write modes (DISABLED/DRY_RUN/APPROVAL/ENABLED), R4 suggestion-only, preview/approval, per-record rollback, bulk blocker upsert; 1.2.0
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
gti
Claude Sonnet 5
@@ -8,17 +8,27 @@ Nothing here guesses. A class is proposed only when an EXPLICIT rule, backed by
|
||||
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 -> 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.
|
||||
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
|
||||
|
||||
@@ -32,7 +42,12 @@ 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")
|
||||
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
|
||||
@@ -55,14 +70,23 @@ class Resolution:
|
||||
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:
|
||||
return self.status in RESOLVED_STATUSES and bool(self.resolved_cclasstrib)
|
||||
"""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, "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}
|
||||
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:
|
||||
@@ -85,34 +109,50 @@ def _num(value):
|
||||
|
||||
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)."""
|
||||
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] = "LOCAL_ONLY"
|
||||
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] = "OFFICIAL_ONLY"
|
||||
result[code] = "MISSING_LOCAL"
|
||||
else:
|
||||
problems = []
|
||||
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")
|
||||
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")
|
||||
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"]}}
|
||||
|
||||
@@ -174,7 +214,7 @@ def _validate_class(code: str, local_rows: dict, compat: dict, today: date):
|
||||
if loc is None:
|
||||
return "CLASS_NOT_IN_LOCAL_CATALOG"
|
||||
state = compat["per_code"].get(code)
|
||||
if state != "MATCH":
|
||||
if state != "MATCH_SAFE":
|
||||
return f"LOCAL_OFFICIAL_{state}"
|
||||
if official["start"] and date.fromisoformat(official["start"]) > today:
|
||||
return "CLASS_NOT_YET_IN_FORCE"
|
||||
@@ -198,8 +238,9 @@ def resolve_cclasstrib(product: dict, company: dict | None, ctx: dict, local_row
|
||||
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:
|
||||
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)
|
||||
@@ -229,5 +270,5 @@ def resolve_cclasstrib(product: dict, company: dict | None, ctx: dict, local_row
|
||||
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())
|
||||
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())
|
||||
|
||||
Reference in New Issue
Block a user