149 lines
9.9 KiB
Python
149 lines
9.9 KiB
Python
"""READ-ONLY ERP loader for the unified fiscal completeness (PIS/COFINS applicability + RTC when the ERP has the reform on).
|
|
|
|
The rules live in central/app/unified_fiscal_coverage.py (pure). This module only observes the ERP through an already open
|
|
READ ONLY connection (SELECT only): schema capabilities, RTC catalog, PARAMETROS.USAR_REFORMA_TRIBUTARIA and the product columns.
|
|
Nothing is written and no value is ever filled in.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from collections import Counter
|
|
from datetime import date, datetime, timezone
|
|
|
|
from central.app.cclasstrib_resolver import compare_catalogs, local_catalog_hash, resolve_cclasstrib
|
|
from central.app.rtc_findings import refine_rtc_finding, summarize_rtc
|
|
from central.app.unified_fiscal_coverage import PIS_COFINS_FIELDS, catalog_integrity, cclasstrib_applicability, evaluate_product, legal_effective_context, rtc_enabled
|
|
|
|
PRODUCT_TABLE = "C000025"
|
|
RTC_CCLASSTRIB_TABLE = "RTC_CCLASSTRIB"
|
|
RTC_CST_TABLE = "RTC_CST"
|
|
PRODUCT_COLUMNS = ("CODIGO", "CCLASSTRIB", "CLASSIFICACAO_FISCAL") + PIS_COFINS_FIELDS + ("PISCOFINS", "INCIDENCIA_PISCOFINS")
|
|
TRUTHY = {"1", "S", "T", "TRUE"}
|
|
|
|
|
|
def detect_rtc_capabilities(schema: dict, *, param_value=None, catalog: dict | None = None, physical_fk: bool | None = None) -> dict:
|
|
"""schema: {TABLE: {COLUMN, ...}}. Fail-closed: anything not proven is False. Old databases without RTC simply report False."""
|
|
product_cols = set(schema.get(PRODUCT_TABLE, ()))
|
|
class_cols = set(schema.get(RTC_CCLASSTRIB_TABLE, ()))
|
|
cst_cols = set(schema.get(RTC_CST_TABLE, ()))
|
|
has_product = "CCLASSTRIB" in product_cols
|
|
has_class = {"CCLASSTRIB", "CSTIBSCBS"} <= class_cols
|
|
has_cst = "CSTIBSCBS" in cst_cols
|
|
caps = {
|
|
"HAS_PRODUCT_CCLASSTRIB": has_product, "HAS_RTC_CCLASSTRIB_TABLE": has_class, "HAS_RTC_CST_TABLE": has_cst,
|
|
"HAS_RTC_RELATIONSHIP": bool(has_product and has_class and has_cst), "HAS_PHYSICAL_RTC_FK": bool(physical_fk),
|
|
"RTC_ENABLED_BY_PARAM": False if param_value is None else str(param_value).strip().upper() in TRUTHY,
|
|
"RTC_CATALOG_VALID": False,
|
|
}
|
|
if catalog is not None and has_class and has_cst:
|
|
caps["RTC_CATALOG_VALID"] = catalog_integrity(catalog)["valid"]
|
|
return caps
|
|
|
|
|
|
def load_rtc_context(cursor) -> tuple[dict, dict | None, dict]:
|
|
"""Schema + catalog + ERP parameter through a READ ONLY cursor. Never raises for a database that lacks RTC."""
|
|
schema: dict[str, set] = {}
|
|
cursor.execute("SELECT TRIM(rf.RDB$RELATION_NAME), TRIM(rf.RDB$FIELD_NAME) FROM RDB$RELATION_FIELDS rf WHERE rf.RDB$RELATION_NAME IN ('C000025','RTC_CCLASSTRIB','RTC_CST','PARAMETROS')")
|
|
for table, column in cursor.fetchall():
|
|
schema.setdefault(table, set()).add(column)
|
|
extras: dict = {"param_value": None, "param_rows": 0, "param_conflict": False, "physical_fk": False}
|
|
if "USAR_REFORMA_TRIBUTARIA" in schema.get("PARAMETROS", ()):
|
|
try:
|
|
# ALL rows are read: the source of truth is what THIS installation's database says NOW (never a cached or Hub-side value).
|
|
cursor.execute("SELECT FIRST 20 USAR_REFORMA_TRIBUTARIA FROM PARAMETROS")
|
|
values = [None if r[0] is None else str(r[0]).strip().upper() for r in cursor.fetchall()]
|
|
extras["param_rows"] = len(values)
|
|
if len(set(values)) > 1:
|
|
extras["param_conflict"] = True # rows disagree: never guess "on" (RTC stays out of scope, and the conflict is reported)
|
|
elif values:
|
|
extras["param_value"] = values[0]
|
|
except Exception: # noqa: BLE001 - unreadable parameter => RTC stays off (not enabled by the ERP)
|
|
extras["param_value"] = None
|
|
try:
|
|
cursor.execute("SELECT COUNT(*) FROM RDB$RELATION_CONSTRAINTS WHERE RDB$RELATION_NAME = 'C000025' AND RDB$CONSTRAINT_TYPE = 'FOREIGN KEY' AND RDB$CONSTRAINT_NAME CONTAINING 'CCLASSTRIB'")
|
|
extras["physical_fk"] = cursor.fetchone()[0] > 0
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
catalog = None
|
|
class_cols = schema.get(RTC_CCLASSTRIB_TABLE, set())
|
|
if {"CCLASSTRIB", "CSTIBSCBS"} <= class_cols and "CSTIBSCBS" in schema.get(RTC_CST_TABLE, set()):
|
|
validity = "INICIO_VIGENCIA" in class_cols and "FIM_VIGENCIA" in class_cols
|
|
extra = [c for c in ("TIPO_ALIQUOTA", "PRED_IBS", "PRED_CBS", "IND_NFE", "IND_NFCE") if c in class_cols]
|
|
cursor.execute("SELECT TRIM(CCLASSTRIB), TRIM(CSTIBSCBS)" + (", INICIO_VIGENCIA, FIM_VIGENCIA" if validity else "") + "".join(f", {c}" for c in extra) + " FROM RTC_CCLASSTRIB")
|
|
rows = cursor.fetchall()
|
|
classes = {r[0]: (r[1] or None) for r in rows}
|
|
cursor.execute("SELECT TRIM(CSTIBSCBS) FROM RTC_CST")
|
|
catalog = {"classes": classes, "cst": {r[0] for r in cursor.fetchall()}, "duplicates": len(rows) - len(classes)}
|
|
if validity:
|
|
catalog["validity"] = {r[0]: (r[2], r[3]) for r in rows}
|
|
offset = 4 if validity else 2
|
|
catalog["rows"] = {r[0]: {"cst": r[1], "start": (r[2] if validity else None), "end": (r[3] if validity else None), **{c: r[offset + i] for i, c in enumerate(extra)}} for r in rows}
|
|
for value in catalog["rows"].values():
|
|
for key in ("TIPO_ALIQUOTA", "PRED_IBS", "PRED_CBS", "IND_NFE", "IND_NFCE"):
|
|
if key in value:
|
|
value[key.lower()] = value.pop(key)
|
|
for flag in ("ind_nfe", "ind_nfce"):
|
|
if flag in value:
|
|
value[flag] = str(value[flag]).strip() in ("1", "True", "true")
|
|
return schema, catalog, extras
|
|
|
|
|
|
def read_products(cursor, schema: dict) -> list[dict]:
|
|
"""SELECT of only the columns that exist (old schemas keep working)."""
|
|
have = schema.get(PRODUCT_TABLE, set())
|
|
columns = [c for c in PRODUCT_COLUMNS if c in have]
|
|
if "CODIGO" not in columns:
|
|
return []
|
|
cursor.execute(f"SELECT {', '.join(columns)} FROM {PRODUCT_TABLE} ORDER BY CODIGO")
|
|
return [dict(zip(columns, row)) for row in cursor.fetchall()]
|
|
|
|
|
|
def assess_from_connection(conn, company: dict | None, *, today: date | None = None, rtc_override: bool | None = None, ncm_valid: dict | None = None) -> dict:
|
|
"""One read-only pass. Returns {"context": {...}, "findings": {product_code: evaluate_product(...)}}.
|
|
rtc_override forces RTC on/off for simulations WITHOUT touching the ERP parameter (tests / what-if only).
|
|
ncm_valid: {product_code: bool} from the decision engine (EXACT_NCM); a product whose NCM is not confirmed never gets a cClassTrib proposal."""
|
|
cursor = conn.cursor()
|
|
schema, catalog, extras = load_rtc_context(cursor)
|
|
caps = detect_rtc_capabilities(schema, param_value=extras["param_value"], catalog=catalog, physical_fk=extras["physical_fk"])
|
|
enabled = rtc_enabled(caps) if rtc_override is None else bool(rtc_override and caps["HAS_RTC_RELATIONSHIP"])
|
|
today = today or date.today()
|
|
cclasstrib_app = cclasstrib_applicability((company or {}).get("crt"), enabled, today)
|
|
rtc = {"enabled": enabled, "catalog": catalog, "catalog_valid": caps["RTC_CATALOG_VALID"], "cclasstrib_applicability": cclasstrib_app}
|
|
products = read_products(cursor, schema)
|
|
findings = {}
|
|
warn_by_type: Counter = Counter()
|
|
warn_products: set = set()
|
|
samples: list = []
|
|
applicability: Counter = Counter()
|
|
local_rows = (catalog or {}).get("rows") or {}
|
|
applicable = cclasstrib_app["state"] == "REQUIRED"
|
|
compat = compare_catalogs(local_rows, today) if (enabled and applicable and local_rows) else None
|
|
resolver_ctx = {"rtc_enabled": enabled, "cclasstrib_applicable": applicable, "local_catalog_valid": bool(caps["RTC_CATALOG_VALID"])}
|
|
for p in products:
|
|
found = evaluate_product(p, company, rtc, today)
|
|
if enabled and applicable:
|
|
code0 = str(p.get("CODIGO") or "")
|
|
resolution = resolve_cclasstrib({**p, "ncm_valid": bool((ncm_valid or {}).get(code0, False))}, company, resolver_ctx, local_rows, compat=compat, today=today)
|
|
refine_rtc_finding(found, p, resolution, compat["per_code"] if compat else None)
|
|
found["input_fingerprint"] = hashlib.sha256(json.dumps({k: (None if v is None else str(v).strip()) for k, v in p.items()} | {"rtc": enabled, "cclasstrib_applicability": cclasstrib_app}, sort_keys=True).encode()).hexdigest()
|
|
code = str(p.get("CODIGO") or "")
|
|
findings[code] = found
|
|
applicability[found["applicability"]["PIS_COFINS"]] += 1
|
|
for w in found.get("warnings", []):
|
|
warn_by_type[w["type"]] += 1
|
|
warn_products.add(code)
|
|
if len(samples) < 10:
|
|
samples.append({"product_code": code, **w})
|
|
integrity = catalog_integrity(catalog) if catalog else None
|
|
return {"context": {"rtc_enabled": enabled, "erp_rtc_enabled": enabled, "rtc_supported": caps["HAS_RTC_RELATIONSHIP"], "rtc_catalog_valid": caps["RTC_CATALOG_VALID"], "rtc_param": extras["param_value"],
|
|
"rtc_source": "SIMULATION" if rtc_override is not None else "PARAMETROS.USAR_REFORMA_TRIBUTARIA", "rtc_param_rows": extras["param_rows"], "rtc_param_conflict": extras["param_conflict"],
|
|
"rtc_observed_at": datetime.now(timezone.utc).isoformat(timespec="seconds"), "cclasstrib_applicability": cclasstrib_app,
|
|
"legal_effective_context": legal_effective_context((company or {}).get("crt")),
|
|
"catalog_integrity": integrity, "crt": (company or {}).get("crt"), "capabilities": caps, "products_evaluated": len(products),
|
|
"pis_cofins_applicability": dict(applicability),
|
|
"rtc_resolution": summarize_rtc(findings, compat) if enabled else None, "local_catalog_hash": local_catalog_hash(local_rows) if local_rows else None,
|
|
"official_catalog": (compat or {}).get("official_catalog"),
|
|
"non_canonical_warnings": {"products": len(warn_products), "by_type": dict(sorted(warn_by_type.items())), "samples": samples}},
|
|
"findings": findings}
|