feat: unified fiscal coverage - PIS/COFINS applicability + conditional RTC in one definition, one queue, one panel
- central/app/unified_fiscal_coverage.py: pure rules (PIS/COFINS NOT_APPLICABLE/REQUIRED/OPTIONAL/NEEDS_REVIEW, RTC only when the ERP switch is on) - agent: read-only loader (unified_fiscal.py) replaces the legacy/expanded module; findings feed the single operational state, coverage and blocker queue - hub: BLOCKED_FISCAL_DATA blockers (PIS/COFINS/RTC) share the queue; stale ones close on the next full pass; dashboard breakdown gains CEST/PIS/COFINS/RTC - panel: legacy/expanded coverage block removed; discreet 'Reforma Tributaria: Ativa/Desativada' - docs: unified_fiscal_completeness, pis_cofins_applicability, rtc_pis_cofins_model + official source manifests; read-only E2E script Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
gti
Claude Sonnet 5
@@ -0,0 +1,211 @@
|
||||
"""UNIFIED fiscal completeness: ONE definition of "complete" for a product (pure, no I/O).
|
||||
|
||||
mandatory(product, company, rtc) = base fiscal fields (decided elsewhere: NCM/CEST/CFOP/ICMS/IPI ...)
|
||||
+ PIS/COFINS when they APPLY to the product
|
||||
+ RTC (cClassTrib -> CST IBS/CBS) ONLY when the ERP has the tax reform switched on.
|
||||
|
||||
There is no "legacy scope" and no "expanded scope": every finding here becomes a normal blocker in the single queue and
|
||||
changes the single coverage number. Nothing here writes to the ERP or invents a fiscal value.
|
||||
|
||||
Sources (see docs/pis_cofins_applicability.md): SPED EFD-Contribuicoes table 4.3.3/4.3.4 (CST domain), 4.3.10-4.3.15 (special regimes),
|
||||
LC 123/2006 art. 13/18 (PIS/Cofins inside the Simples DAS), Leis 9.718/98, 10.637/2002, 10.833/2003 (cumulative / non-cumulative),
|
||||
NT 2025.002-RTC v1.51 (IBS/CBS fields), LC 214/2025 art. 348.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import unicodedata
|
||||
from datetime import date
|
||||
|
||||
# Official CST PIS / CST COFINS domain (SPED tables 4.3.3 / 4.3.4). Membership only - never used to pick a value.
|
||||
CST_DOMAIN = frozenset([f"{n:02d}" for n in (*range(1, 10), 49, *range(50, 57), *range(60, 68), *range(70, 76), 98, 99)])
|
||||
RATE_REQUIRED_CST = frozenset({"01", "02"}) # taxable by percentage: a rate is needed
|
||||
RATE_NOT_APPLICABLE_CST = frozenset({"03", "04", "05", "06", "07", "08", "09"}) # per-unit amount / monophasic / ST / zero / exempt / no incidence / suspended
|
||||
INCIDENCE_CST = {"MONOFASICA": frozenset({"02", "03", "04"}), "SUBSTITUICAO": frozenset({"05"})} # 4.3.10/4.3.11 (02,03,04) and 4.3.12 (05)
|
||||
SIMPLES_FAMILY = frozenset({"1", "2", "4"}) # CRT 1 Simples, 2 Simples excess sublimit, 4 MEI (LC 123/2006): PIS/Cofins are collected inside the DAS
|
||||
|
||||
PIS_COFINS_FIELDS = ("COD_ENTRADA_PIS", "COD_SAIDA_PIS", "ALIQ_PIS", "COD_ENTRADA_COFINS", "COD_SAIDA_COFINS", "ALIQ_COFINS")
|
||||
|
||||
FIELD_BY_BLOCKER_PREFIX = (("PIS", "PIS"), ("COFINS", "COFINS"), ("CCLASSTRIB", "RTC"), ("RTC", "RTC"))
|
||||
|
||||
|
||||
def _text(value) -> str:
|
||||
return "" if value is None else str(value).strip()
|
||||
|
||||
|
||||
def blocker_category(blocker_type: str) -> str:
|
||||
"""Dashboard category of a blocker type: PIS | COFINS | RTC | OTHER."""
|
||||
for prefix, category in FIELD_BY_BLOCKER_PREFIX:
|
||||
if blocker_type.startswith(prefix):
|
||||
return category
|
||||
return "OTHER"
|
||||
|
||||
|
||||
def incidence_kind(value) -> str:
|
||||
"""NONE | MONOFASICA | SUBSTITUICAO | UNKNOWN (accent/case-insensitive, encoding tolerant). Anything the SGV does not define is UNKNOWN, never NONE."""
|
||||
raw = _text(value)
|
||||
text = re.sub(r"[^A-Z]", "", unicodedata.normalize("NFKD", raw).encode("ascii", "ignore").decode().upper())
|
||||
if not raw or text == "NENHUMA":
|
||||
return "NONE"
|
||||
if text.startswith("MONOF"):
|
||||
return "MONOFASICA"
|
||||
if text.startswith("SUBST"):
|
||||
return "SUBSTITUICAO"
|
||||
return "UNKNOWN"
|
||||
|
||||
|
||||
def validate_cst(value) -> str:
|
||||
code = _text(value)
|
||||
if not code:
|
||||
return "EMPTY"
|
||||
return "VALID" if code in CST_DOMAIN else "INVALID"
|
||||
|
||||
|
||||
def cst_suggestion(value) -> str | None:
|
||||
"""A short numeric code that becomes an official CST once zero-padded to two digits ('4' -> '04'). Only a hint for the operator; never applied."""
|
||||
code = _text(value)
|
||||
return code.zfill(2) if code.isdigit() and len(code) == 1 and code.zfill(2) in CST_DOMAIN else None
|
||||
|
||||
|
||||
def validate_rate(value) -> str:
|
||||
text = _text(value)
|
||||
if not text:
|
||||
return "EMPTY"
|
||||
if not re.fullmatch(r"\d{1,3}([.,]\d{1,4})?", text):
|
||||
return "INVALID"
|
||||
return "VALID" if float(text.replace(",", ".")) <= 100 else "INVALID"
|
||||
|
||||
|
||||
# ------------------------------------------------------------------------------------------------- PIS / COFINS
|
||||
|
||||
|
||||
def pis_cofins_applicability(product: dict, crt) -> dict:
|
||||
"""{"state": NOT_APPLICABLE | REQUIRED | OPTIONAL | NEEDS_REVIEW, "basis": ...}.
|
||||
|
||||
NOT_APPLICABLE the product carries no product-level PIS/COFINS treatment at all (flag N/blank, incidence NENHUMA/blank, all six fields blank).
|
||||
The tax is then settled elsewhere: inside the DAS (Simples family) or from the company default (regime normal).
|
||||
REQUIRED the product itself claims a special treatment (flag S, or incidence MONOFASICA/SUBSTITUICAO): the exit CST (and a rate when the CST needs one) must be valid.
|
||||
OPTIONAL no special claim, but some CST/rate was typed: it is not mandatory, yet whatever is there must be valid.
|
||||
NEEDS_REVIEW the incidence text is not one the SGV defines (NENHUMA / MONOFASICA / SUBST.TRIBUTARIA): applicability cannot be decided.
|
||||
"""
|
||||
flag = _text(product.get("PISCOFINS")).upper()
|
||||
kind = incidence_kind(product.get("INCIDENCIA_PISCOFINS"))
|
||||
has_values = any(_text(product.get(c)) for c in PIS_COFINS_FIELDS)
|
||||
if kind == "UNKNOWN":
|
||||
return {"state": "NEEDS_REVIEW", "basis": "INCIDENCE_NOT_DEFINED_BY_SGV"}
|
||||
if flag == "S" or kind in ("MONOFASICA", "SUBSTITUICAO"):
|
||||
return {"state": "REQUIRED", "basis": "PRODUCT_DECLARES_SPECIAL_TREATMENT"}
|
||||
if has_values:
|
||||
return {"state": "OPTIONAL", "basis": "VALUES_TYPED_WITHOUT_SPECIAL_CLAIM"}
|
||||
return {"state": "NOT_APPLICABLE", "basis": "SIMPLES_DAS" if _text(crt) in SIMPLES_FAMILY else "COMPANY_DEFAULT"}
|
||||
|
||||
|
||||
def evaluate_pis_cofins(product: dict, crt) -> dict:
|
||||
"""{"applicability": {...}, "blockers": [(type, reason)]}. Entry and exit are separate; a missing OPTIONAL value never blocks."""
|
||||
app = pis_cofins_applicability(product, crt)
|
||||
blockers: list[tuple[str, str]] = []
|
||||
if app["state"] in ("NOT_APPLICABLE", "NEEDS_REVIEW"):
|
||||
return {"applicability": app, "blockers": blockers}
|
||||
kind = incidence_kind(product.get("INCIDENCIA_PISCOFINS"))
|
||||
flag = _text(product.get("PISCOFINS")).upper()
|
||||
for tax in ("PIS", "COFINS"):
|
||||
entry, exit_, rate = product.get(f"COD_ENTRADA_{tax}"), product.get(f"COD_SAIDA_{tax}"), product.get(f"ALIQ_{tax}")
|
||||
entry_state, exit_state, rate_state = validate_cst(entry), validate_cst(exit_), validate_rate(rate)
|
||||
if entry_state == "INVALID":
|
||||
blockers.append((f"{tax}_CST_INVALID", "ENTRY" + (f":UNPADDED_USE_{cst_suggestion(entry)}" if cst_suggestion(entry) else "")))
|
||||
if exit_state == "INVALID":
|
||||
blockers.append((f"{tax}_CST_INVALID", "EXIT" + (f":UNPADDED_USE_{cst_suggestion(exit_)}" if cst_suggestion(exit_) else "")))
|
||||
if app["state"] == "REQUIRED" and exit_state == "EMPTY":
|
||||
blockers.append((f"{tax}_CST_MISSING", "EXIT"))
|
||||
if rate_state == "INVALID":
|
||||
blockers.append((f"{tax}_RATE_INVALID", "RATE"))
|
||||
if exit_state == "VALID":
|
||||
code = _text(exit_)
|
||||
if code in RATE_REQUIRED_CST and rate_state == "EMPTY":
|
||||
blockers.append((f"{tax}_RATE_MISSING", f"CST_{code}_NEEDS_RATE"))
|
||||
if kind in INCIDENCE_CST and code not in INCIDENCE_CST[kind]:
|
||||
blockers.append((f"{tax}_CST_INCIDENCE_MISMATCH", f"{kind}:CST_{code}"))
|
||||
if kind == "NONE" and app["state"] == "REQUIRED" and code in RATE_NOT_APPLICABLE_CST - {"03"} and flag == "S":
|
||||
blockers.append((f"{tax}_CST_INCIDENCE_MISMATCH", f"NENHUMA:CST_{code}"))
|
||||
if flag == "N" and kind != "NONE":
|
||||
blockers.append(("PIS_COFINS_FLAG_INCONSISTENT", "FLAG_N_WITH_SPECIAL_INCIDENCE"))
|
||||
if flag == "S" and kind == "NONE" and not any(_text(product.get(c)) for c in PIS_COFINS_FIELDS):
|
||||
blockers.append(("PIS_COFINS_FLAG_INCONSISTENT", "FLAG_S_WITHOUT_INCIDENCE_OR_CODES"))
|
||||
return {"applicability": app, "blockers": _dedupe(blockers)}
|
||||
|
||||
|
||||
def _dedupe(items):
|
||||
seen, out = set(), []
|
||||
for item in items:
|
||||
if item not in seen:
|
||||
seen.add(item)
|
||||
out.append(item)
|
||||
return out
|
||||
|
||||
|
||||
# ------------------------------------------------------------------------------------------------- RTC
|
||||
|
||||
|
||||
def rtc_enabled(caps: dict) -> bool:
|
||||
"""RTC is in scope ONLY when the ERP is switched to the reform (PARAMETROS.USAR_REFORMA_TRIBUTARIA truthy) AND its schema can hold it."""
|
||||
return bool(caps.get("HAS_PRODUCT_CCLASSTRIB") and caps.get("HAS_RTC_RELATIONSHIP") and caps.get("RTC_ENABLED_BY_PARAM"))
|
||||
|
||||
|
||||
def catalog_integrity(catalog: dict) -> dict:
|
||||
"""catalog: {"classes": {cClassTrib: cst_or_None}, "cst": {code, ...}, "duplicates": int}."""
|
||||
classes, cst = catalog.get("classes", {}), set(catalog.get("cst", ()))
|
||||
bad_format = [c for c in classes if not re.fullmatch(r"\d{6}", str(c))]
|
||||
bad_cst_format = [c for c in cst if not re.fullmatch(r"\d{3}", str(c))]
|
||||
orphans = [c for c, s in classes.items() if s is None or str(s) not in cst]
|
||||
return {"classes": len(classes), "cst": len(cst), "orphan_classes": len(orphans), "bad_class_format": len(bad_format), "bad_cst_format": len(bad_cst_format),
|
||||
"duplicates": int(catalog.get("duplicates", 0)), "valid": bool(classes and cst and not orphans and not bad_format and not bad_cst_format and not catalog.get("duplicates"))}
|
||||
|
||||
|
||||
def validate_cclasstrib(value, catalog: dict | None, today: date | None = None) -> str:
|
||||
"""EMPTY | INVALID_FORMAT | NOT_IN_CATALOG | EXPIRED | CST_MISSING | VALID (CST IBS/CBS derived through the catalog, never stored on the product)."""
|
||||
code = _text(value)
|
||||
if not code:
|
||||
return "EMPTY"
|
||||
if not re.fullmatch(r"\d{6}", code):
|
||||
return "INVALID_FORMAT"
|
||||
if not catalog or code not in catalog.get("classes", {}):
|
||||
return "NOT_IN_CATALOG"
|
||||
validity = (catalog.get("validity") or {}).get(code)
|
||||
if validity and today is not None:
|
||||
start, end = validity
|
||||
if (start and start > today) or (end and end < today):
|
||||
return "EXPIRED"
|
||||
cst = catalog["classes"][code]
|
||||
return "VALID" if cst is not None and str(cst) in set(catalog.get("cst", ())) else "CST_MISSING"
|
||||
|
||||
|
||||
def evaluate_rtc(product: dict, rtc: dict | None, today: date | None = None) -> dict:
|
||||
"""rtc: {"enabled": bool, "catalog": {...}|None, "catalog_valid": bool}. Disabled => NOT_APPLICABLE and NO blocker, whatever CCLASSTRIB holds."""
|
||||
if not rtc or not rtc.get("enabled"):
|
||||
return {"applicability": {"state": "NOT_APPLICABLE", "basis": "RTC_DISABLED_IN_ERP"}, "blockers": []}
|
||||
app = {"state": "REQUIRED", "basis": "RTC_ENABLED_IN_ERP"}
|
||||
if not rtc.get("catalog_valid"):
|
||||
return {"applicability": app, "blockers": [("RTC_CATALOG_INVALID", "LOCAL_CATALOG_FAILED_INTEGRITY")]}
|
||||
state = validate_cclasstrib(product.get("CCLASSTRIB"), rtc.get("catalog"), today)
|
||||
mapping = {"EMPTY": "CCLASSTRIB_MISSING", "INVALID_FORMAT": "CCLASSTRIB_INVALID_FORMAT", "NOT_IN_CATALOG": "CCLASSTRIB_NOT_IN_CATALOG", "EXPIRED": "CCLASSTRIB_NOT_IN_CATALOG", "CST_MISSING": "RTC_CST_MISSING"}
|
||||
return {"applicability": app, "blockers": [(mapping[state], state)] if state in mapping else []}
|
||||
|
||||
|
||||
# ------------------------------------------------------------------------------------------------- unified
|
||||
|
||||
|
||||
def evaluate_product(product: dict, company: dict | None, rtc: dict | None, today: date | None = None) -> dict:
|
||||
"""The single fiscal-completeness verdict for the fields this module owns.
|
||||
Returns {"blockers": [{"type","category","reason"}], "applicability": {"PIS_COFINS": state, "RTC": state}, "mandatory_fields": [...], "needs_review": bool}."""
|
||||
crt = (company or {}).get("crt")
|
||||
pis = evaluate_pis_cofins(product, crt)
|
||||
rtc_result = evaluate_rtc(product, rtc, today)
|
||||
blockers = [{"type": kind, "category": blocker_category(kind), "reason": reason} for kind, reason in pis["blockers"] + rtc_result["blockers"]]
|
||||
mandatory = []
|
||||
if pis["applicability"]["state"] == "REQUIRED":
|
||||
mandatory += ["COD_SAIDA_PIS", "COD_SAIDA_COFINS"]
|
||||
if rtc_result["applicability"]["state"] == "REQUIRED":
|
||||
mandatory.append("CCLASSTRIB")
|
||||
return {"blockers": blockers, "applicability": {"PIS_COFINS": pis["applicability"]["state"], "RTC": rtc_result["applicability"]["state"]},
|
||||
"basis": {"PIS_COFINS": pis["applicability"]["basis"], "RTC": rtc_result["applicability"]["basis"]}, "mandatory_fields": mandatory,
|
||||
"needs_review": pis["applicability"]["state"] == "NEEDS_REVIEW"}
|
||||
Reference in New Issue
Block a user