diff --git a/agent/scripts/run_agent_service.py b/agent/scripts/run_agent_service.py index b7e71c8..9d0fd21 100644 --- a/agent/scripts/run_agent_service.py +++ b/agent/scripts/run_agent_service.py @@ -30,7 +30,8 @@ sys.path.insert(0, str(AGENT_ROOT / "src")) sys.path.insert(0, str(PROJECT_ROOT)) # ships central/app engine modules read-only (central.app.*) from gti_fiscal_agent.client import HubClient # noqa: E402 -from gti_fiscal_agent.firebird_reader import check_connection, read_products # noqa: E402 +from gti_fiscal_agent.firebird_reader import _connect as _open_read_only, check_connection, read_products # noqa: E402 +from gti_fiscal_agent.rtc_pis_cofins import assess_from_connection # noqa: E402 from gti_fiscal_agent.pipeline import run_pipeline # noqa: E402 from gti_fiscal_agent.runner import REFERENCE_SCHEMA_V1, product_fingerprint, to_hub_product # noqa: E402 from central.app.operational_state import classify_product_state, summarize_operational_states # noqa: E402 @@ -186,6 +187,18 @@ def open_write_connection(env, database_path, firebird_writes=True): ) +def _expanded_scope(env, legacy_states): + """RTC + PIS/COFINS assessment (read-only, SELECT only). Isolated: a failure here never fails the job or changes legacy coverage.""" + try: + connection = _open_read_only(env) + try: + return assess_from_connection(connection, legacy_states) + finally: + connection.close() + except Exception as error: # noqa: BLE001 + return {"scope_version": "expanded-v1", "unavailable": type(error).__name__} + + def execute_job(job, config, executor, state, env): """Run the pipeline for one claimed job. Never raises for job-domain failures (bad Firebird read, classify error) — returns (result_summary, error).""" @@ -225,7 +238,7 @@ def execute_job(job, config, executor, state, env): if write_connection is not None: write_connection.close() - operational = coverage = None + operational = coverage = expanded_scope = None if sync_kind: if cache: for product, decision in zip(to_process, decisions): @@ -234,11 +247,14 @@ def execute_job(job, config, executor, state, env): all_decisions = [state.cached_decision(p["CODIGO"]) or {} for p in products] else: all_decisions = decisions - operational = summarize_operational_states([classify_product_state(p, d) for p, d in zip(products, all_decisions)]) + product_states = [classify_product_state(p, d) for p, d in zip(products, all_decisions)] + operational = summarize_operational_states(product_states) coverage = build_job_coverage(env, client, products, all_decisions) + expanded_scope = _expanded_scope(env, {s["product_code"]: s["state"] for s in product_states}) result_summary = { "coverage": coverage, "operational": operational, + "expanded_scope": expanded_scope, "fiscal_mode": getattr(config, "fiscal_mode", None), "release_version": env.get("AGENT_VERSION") or "1.1.7", "products_total": len(products), diff --git a/agent/src/gti_fiscal_agent/rtc_pis_cofins.py b/agent/src/gti_fiscal_agent/rtc_pis_cofins.py new file mode 100644 index 0000000..c0b3d1c --- /dev/null +++ b/agent/src/gti_fiscal_agent/rtc_pis_cofins.py @@ -0,0 +1,277 @@ +"""READ-ONLY RTC (cClassTrib / CST IBS-CBS) and PIS/COFINS assessment of SGV products. + +Everything here is observation: capability detection, catalog integrity, deterministic validators, applicability and an +EXPANDED coverage view that sits next to (never replaces) the legacy scope coverage. Nothing writes to Firebird, nothing +fills a value, and no fiscal rule is invented: where local data cannot prove that a field applies, the answer is +UNKNOWN_APPLICABILITY, not a blocker. + +Vocabulary kept apart on purpose: SCHEMA_SUPPORT (the column exists) / DATA_PRESENT (it has a value) / +APPLICABILITY (the field applies to this product) / MANDATORYNESS (a missing value is a defect). Only the last one +may block, and only when applicability is proven from local, reliable data. +""" +from __future__ import annotations + +import re + +LOCAL_CATALOG_ONLY = True # the RTC catalog used is the one stored in the client's own SGV database; no official dataset is integrated here + +PRODUCT_TABLE = "C000025" +RTC_CCLASSTRIB_TABLE = "RTC_CCLASSTRIB" +RTC_CST_TABLE = "RTC_CST" +PIS_COFINS_COLUMNS = ("COD_ENTRADA_PIS", "COD_SAIDA_PIS", "ALIQ_PIS", "COD_ENTRADA_COFINS", "COD_SAIDA_COFINS", "ALIQ_COFINS", "PISCOFINS", "INCIDENCIA_PISCOFINS") +PRODUCT_COLUMNS = ("CODIGO", "CCLASSTRIB") + PIS_COFINS_COLUMNS + +# Official CST PIS/COFINS codes (2 digits). Used for FORMAT/membership only, never to pick a value. +OFFICIAL_PIS_COFINS_CST = frozenset([f"{n:02d}" for n in (*range(1, 10), 49, *range(50, 57), *range(60, 68), *range(70, 76), 98, 99)]) + +# ---------------------------------------------------------------- capabilities ------------------------------------------------- + + +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 + has_relationship = bool(has_product and has_class and has_cst) # column-level relationship: product -> class -> cst + caps = { + "HAS_PRODUCT_CCLASSTRIB": has_product, + "HAS_RTC_CCLASSTRIB_TABLE": has_class, + "HAS_RTC_CST_TABLE": has_cst, + "HAS_RTC_RELATIONSHIP": has_relationship, + "HAS_PHYSICAL_RTC_FK": bool(physical_fk), + "RTC_ENABLED_BY_PARAM": None if param_value is None else str(param_value).strip() in {"1", "S", "T", "TRUE"}, + "RTC_CATALOG_VALID": False, + "LOCAL_CATALOG_ONLY": LOCAL_CATALOG_ONLY, + } + if catalog is not None and has_class and has_cst: + caps["RTC_CATALOG_VALID"] = catalog_integrity(catalog)["valid"] + return caps + + +def catalog_integrity(catalog: dict) -> dict: + """catalog: {"classes": {cClassTrib: cst_or_None}, "cst": {code, ...}, "duplicates": int}. Pure.""" + 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"))} + +# ---------------------------------------------------------------- validators --------------------------------------------------- + + +def _text(value) -> str: + return "" if value is None else str(value).strip() + + +def validate_cclasstrib(value, catalog: dict | None) -> str: + """EMPTY | INVALID_FORMAT | NOT_IN_LOCAL_CATALOG | VALID_LOCAL | CST_MISSING | VALID_WITH_CST (deterministic, no inference).""" + 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_LOCAL_CATALOG" + if not catalog.get("cst"): + return "VALID_LOCAL" # code exists, but there is no CST table to verify its CST + cst = catalog["classes"][code] + return "VALID_WITH_CST" if cst is not None and str(cst) in catalog["cst"] else "CST_MISSING" + + +def validate_pis_cofins_cst(value) -> str: + code = _text(value) + if not code: + return "EMPTY" + return "VALID" if code in OFFICIAL_PIS_COFINS_CST else "INVALID_FORMAT" + + +def validate_pis_cofins_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_RATE" + return "VALID" if float(text.replace(",", ".")) <= 100 else "INVALID_RATE" + + +def validate_pis_cofins(product: dict) -> dict: + """Field-level validation for one product row. Flag consistency compares the ERP's own flag/incidence with its own codes.""" + out = {} + for tax in ("PIS", "COFINS"): + out[f"{tax}_ENTRY_CST"] = validate_pis_cofins_cst(product.get(f"COD_ENTRADA_{tax}")) + out[f"{tax}_EXIT_CST"] = validate_pis_cofins_cst(product.get(f"COD_SAIDA_{tax}")) + out[f"{tax}_RATE"] = validate_pis_cofins_rate(product.get(f"ALIQ_{tax}")) + flag = _text(product.get("PISCOFINS")).upper() + incidence = _text(product.get("INCIDENCIA_PISCOFINS")).upper() + has_codes = any(_text(product.get(c)) for c in PIS_COFINS_COLUMNS[:6]) + special = incidence not in ("", "NENHUMA") + inconsistent = (flag == "N" and (special or has_codes)) or (flag == "S" and incidence in ("", "NENHUMA")) + out["FLAG_CONSISTENCY"] = "INCONSISTENT" if inconsistent else ("EMPTY" if not flag else "CONSISTENT") + return out + +# ---------------------------------------------------------------- applicability ----------------------------------------------- + + +def rtc_applicability(caps: dict) -> str: + """APPLICABLE only when the ERP itself is switched to the reform (PARAMETROS.USAR_REFORMA_TRIBUTARIA) AND the model exists. + NOT_ENABLED_BY_ERP when the ERP says it does not use RTC; NOT_SUPPORTED when the schema cannot hold it; else UNKNOWN.""" + if not (caps.get("HAS_PRODUCT_CCLASSTRIB") and caps.get("HAS_RTC_RELATIONSHIP")): + return "NOT_SUPPORTED" + enabled = caps.get("RTC_ENABLED_BY_PARAM") + if enabled is None: + return "UNKNOWN_APPLICABILITY" + return "APPLICABLE" if enabled else "NOT_ENABLED_BY_ERP" + + +def pis_cofins_applicability(product: dict, schema_ok: bool) -> str: + """Product-level PIS/COFINS defaults are treated as applicable only when the product itself carries evidence of them + (a code, or a special incidence/flag). A product with no data and flag N / NENHUMA gives NO proof either way (documents + take their CST from elsewhere), so it is UNKNOWN_APPLICABILITY - not complete and not blocked.""" + if not schema_ok: + return "NOT_SUPPORTED" + evidence = any(_text(product.get(c)) for c in PIS_COFINS_COLUMNS[:6]) or _text(product.get("PISCOFINS")).upper() == "S" \ + or _text(product.get("INCIDENCIA_PISCOFINS")).upper() not in ("", "NENHUMA") + return "APPLICABLE" if evidence else "UNKNOWN_APPLICABILITY" + +# ---------------------------------------------------------------- blockers + expanded coverage ------------------------------ + + +def expanded_findings(product: dict, caps: dict, catalog: dict | None) -> dict: + """{"blockers": [(TYPE, reason)], "unknown": [area], "notes": [...]}: specific blockers only where applicability is proven.""" + blockers, unknown = [], [] + rtc = rtc_applicability(caps) + if rtc == "APPLICABLE": + state = validate_cclasstrib(product.get("CCLASSTRIB"), catalog) + if state == "EMPTY": + blockers.append(("CCLASSTRIB_MISSING", "EMPTY")) + elif state == "INVALID_FORMAT": + blockers.append(("CCLASSTRIB_INVALID", state)) + elif state == "NOT_IN_LOCAL_CATALOG": + blockers.append(("CCLASSTRIB_NOT_IN_CATALOG", state)) + elif state == "CST_MISSING": + blockers.append(("RTC_CST_INVALID", state)) + elif rtc == "UNKNOWN_APPLICABILITY": + unknown.append("RTC") + schema_ok = caps.get("_PIS_COFINS_SCHEMA", False) + pis_cofins = pis_cofins_applicability(product, schema_ok) + if pis_cofins == "APPLICABLE": + checks = validate_pis_cofins(product) + for tax in ("PIS", "COFINS"): + for slot, label in (("ENTRY_CST", "ENTRY"), ("EXIT_CST", "EXIT")): + result = checks[f"{tax}_{slot}"] + if result == "INVALID_FORMAT": + blockers.append((f"{tax}_CST_INVALID", f"{label}:{result}")) + if checks[f"{tax}_RATE"] == "INVALID_RATE": + blockers.append((f"{tax}_RATE_INVALID", "INVALID_RATE")) + exit_state = checks[f"{tax}_EXIT_CST"] + if exit_state == "EMPTY" and checks[f"{tax}_RATE"] != "EMPTY": + blockers.append((f"{tax}_CST_MISSING", "EXIT:EMPTY_WITH_RATE")) + if checks["FLAG_CONSISTENCY"] == "INCONSISTENT": + blockers.append(("PIS_COFINS_FLAG_INCONSISTENT", "FLAG_VS_DATA")) + elif pis_cofins == "UNKNOWN_APPLICABILITY": + unknown.append("PIS_COFINS") + return {"blockers": blockers, "unknown": unknown, "rtc_applicability": rtc, "pis_cofins_applicability": pis_cofins} + + +def expanded_scope_summary(products: list[dict], legacy_states: dict, caps: dict, catalog: dict | None) -> dict: + """Legacy coverage is reported unchanged (LEGACY_SCOPE_COVERAGE); EXPANDED_FISCAL_SCOPE_COVERAGE adds RTC + PIS/COFINS. + legacy_states: {product_code: state} from classify_product_state (COMPLETE/CORRECTED/BLOCKED_*/FAILED_TECHNICAL).""" + legacy = {"complete": 0, "safe_blocked": 0, "failed": 0} + expanded = {"complete": 0, "safe_blocked": 0, "failed": 0, "unknown_applicability": 0} + blockers: dict[str, int] = {} + blocked_rows = [] + caps = dict(caps) + for product in products: + code = str(product.get("CODIGO") or "") + state = legacy_states.get(code, "COMPLETE") + if state == "FAILED_TECHNICAL": + legacy["failed"] += 1 + expanded["failed"] += 1 + continue + if state.startswith("BLOCKED_"): + legacy["safe_blocked"] += 1 + expanded["safe_blocked"] += 1 # blocked in the legacy scope stays blocked: the expanded scope only ADDS conditions + continue + legacy["complete"] += 1 + found = expanded_findings(product, caps, catalog) + if found["blockers"]: + expanded["safe_blocked"] += 1 + for kind, reason in found["blockers"]: + blockers[kind] = blockers.get(kind, 0) + 1 + if len(blocked_rows) < 200: + blocked_rows.append({"product_code": code, "type": kind, "reason": reason}) + elif found["unknown"]: + expanded["unknown_applicability"] += 1 + else: + expanded["complete"] += 1 + total = len(products) + return {"scope_version": "expanded-v1", "local_catalog_only": LOCAL_CATALOG_ONLY, "total_products": total, + "legacy_scope": legacy, "expanded_scope": expanded, "blockers_by_type": dict(sorted(blockers.items())), + "blocker_samples": blocked_rows, "invariant_pass": sum(expanded.values()) == total and sum(legacy.values()) == total, + "capabilities": {k: v for k, v in caps.items() if not k.startswith("_")}} + +# ---------------------------------------------------------------- read-only loader ------------------------------------------- + + +def load_rtc_context(cursor) -> tuple[dict, dict | None, dict]: + """Read schema + catalog + ERP parameter through an already open READ ONLY cursor (SELECT only). + Returns (schema, catalog|None, extras). 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, "physical_fk": False} + if "USAR_REFORMA_TRIBUTARIA" in schema.get("PARAMETROS", ()): + try: + cursor.execute("SELECT FIRST 1 USAR_REFORMA_TRIBUTARIA FROM PARAMETROS") + row = cursor.fetchone() + extras["param_value"] = None if row is None or row[0] is None else row[0] + except Exception: # noqa: BLE001 - unreadable parameter => applicability stays unknown + 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 + if {"CCLASSTRIB", "CSTIBSCBS"} <= schema.get(RTC_CCLASSTRIB_TABLE, set()) and "CSTIBSCBS" in schema.get(RTC_CST_TABLE, set()): + cursor.execute("SELECT TRIM(CCLASSTRIB), TRIM(CSTIBSCBS) FROM RTC_CCLASSTRIB") + rows = cursor.fetchall() + classes = {c: (s or None) for c, s in rows} + cursor.execute("SELECT TRIM(CSTIBSCBS) FROM RTC_CST") + cst = {r[0] for r in cursor.fetchall()} + catalog = {"classes": classes, "cst": cst, "duplicates": len(rows) - len(classes)} + return schema, catalog, extras + + +def read_products_rtc_pis_cofins(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, legacy_states: dict) -> dict: + """One read-only pass: capabilities -> catalog integrity -> expanded scope. Any failure degrades to an explicit unavailable + marker; it never raises into the job and never touches the database beyond SELECT.""" + try: + 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"]) + caps["_PIS_COFINS_SCHEMA"] = all(c in schema.get(PRODUCT_TABLE, ()) for c in PIS_COFINS_COLUMNS[:6]) + products = read_products_rtc_pis_cofins(cursor, schema) + summary = expanded_scope_summary(products, legacy_states, caps, catalog) + summary["catalog_integrity"] = catalog_integrity(catalog) if catalog else None + summary["capabilities"]["PIS_COFINS_SCHEMA_VALID"] = caps["_PIS_COFINS_SCHEMA"] + summary["data_present"] = {c: sum(1 for p in products if _text(p.get(c))) for c in ("CCLASSTRIB",) + PIS_COFINS_COLUMNS if any(c in p for p in products[:1])} + return summary + except Exception as error: # noqa: BLE001 + return {"scope_version": "expanded-v1", "unavailable": type(error).__name__} diff --git a/agent/tests/test_rtc_pis_cofins.py b/agent/tests/test_rtc_pis_cofins.py new file mode 100644 index 0000000..752c042 --- /dev/null +++ b/agent/tests/test_rtc_pis_cofins.py @@ -0,0 +1,134 @@ +"""RTC (cClassTrib / CST IBS-CBS) and PIS/COFINS: capabilities, validators, applicability, legacy vs expanded coverage. Read-only.""" +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parents[1] / "src")) + +from gti_fiscal_agent import rtc_pis_cofins as r # noqa: E402 + +FULL_SCHEMA = {"C000025": {"CODIGO", "CCLASSTRIB", *r.PIS_COFINS_COLUMNS}, "RTC_CCLASSTRIB": {"CCLASSTRIB", "CSTIBSCBS"}, "RTC_CST": {"CSTIBSCBS"}} +CATALOG = {"classes": {"000001": "000", "200003": "200", "999999": None}, "cst": {"000", "200"}, "duplicates": 0} + + +def caps(schema=FULL_SCHEMA, param=0, catalog=CATALOG): + c = r.detect_rtc_capabilities(schema, param_value=param, catalog=catalog, physical_fk=True) + c["_PIS_COFINS_SCHEMA"] = all(col in schema.get("C000025", ()) for col in r.PIS_COFINS_COLUMNS[:6]) + return c + + +def test_database_without_rtc_fails_closed_and_is_not_blocked(): + c = caps({"C000025": {"CODIGO"}}, param=None, catalog=None) + assert not any(c[k] for k in ("HAS_PRODUCT_CCLASSTRIB", "HAS_RTC_CCLASSTRIB_TABLE", "HAS_RTC_CST_TABLE", "HAS_RTC_RELATIONSHIP", "RTC_CATALOG_VALID")) + assert r.rtc_applicability(c) == "NOT_SUPPORTED" + found = r.expanded_findings({"CODIGO": 1}, c, None) + assert found["blockers"] == [] and found["unknown"] == [] + + +def test_partial_rtc_column_without_catalog_tables_is_not_a_relationship(): + c = caps({"C000025": {"CODIGO", "CCLASSTRIB"}}, param=1, catalog=None) + assert c["HAS_PRODUCT_CCLASSTRIB"] and not c["HAS_RTC_RELATIONSHIP"] and not c["RTC_CATALOG_VALID"] + assert r.rtc_applicability(c) == "NOT_SUPPORTED" + + +def test_full_model_and_valid_catalog_capabilities(): + good = {"classes": {"000001": "000"}, "cst": {"000"}, "duplicates": 0} + c = caps(catalog=good) + assert c["HAS_RTC_RELATIONSHIP"] and c["RTC_CATALOG_VALID"] and c["HAS_PHYSICAL_RTC_FK"] and c["RTC_ENABLED_BY_PARAM"] is False and c["LOCAL_CATALOG_ONLY"] + assert not caps(catalog=CATALOG)["RTC_CATALOG_VALID"] # a class pointing to a missing CST makes the catalog invalid + + +def test_cclasstrib_validator_states(): + assert r.validate_cclasstrib(None, CATALOG) == "EMPTY" and r.validate_cclasstrib(" ", CATALOG) == "EMPTY" + assert r.validate_cclasstrib("12A456", CATALOG) == "INVALID_FORMAT" and r.validate_cclasstrib("12345", CATALOG) == "INVALID_FORMAT" + assert r.validate_cclasstrib("123456", CATALOG) == "NOT_IN_LOCAL_CATALOG" + assert r.validate_cclasstrib("200003", CATALOG) == "VALID_WITH_CST" + assert r.validate_cclasstrib("999999", CATALOG) == "CST_MISSING" + assert r.validate_cclasstrib("000001", {"classes": {"000001": None}, "cst": set()}) == "VALID_LOCAL" + assert r.validate_cclasstrib("000001", None) == "NOT_IN_LOCAL_CATALOG" + + +def test_pis_cofins_validators(): + assert r.validate_pis_cofins_cst("") == "EMPTY" and r.validate_pis_cofins_cst(None) == "EMPTY" + assert r.validate_pis_cofins_cst("01") == "VALID" and r.validate_pis_cofins_cst("73") == "VALID" + assert r.validate_pis_cofins_cst("2") == "INVALID_FORMAT" and r.validate_pis_cofins_cst("10") == "INVALID_FORMAT" + assert r.validate_pis_cofins_rate("0,65") == "VALID" and r.validate_pis_cofins_rate("3") == "VALID" and r.validate_pis_cofins_rate("") == "EMPTY" + assert r.validate_pis_cofins_rate("abc") == "INVALID_RATE" and r.validate_pis_cofins_rate("150") == "INVALID_RATE" + ok = r.validate_pis_cofins({"PISCOFINS": "S", "INCIDENCIA_PISCOFINS": "MONOFASICA", "COD_SAIDA_PIS": "04", "COD_SAIDA_COFINS": "04"}) + assert ok["FLAG_CONSISTENCY"] == "CONSISTENT" + bad = r.validate_pis_cofins({"PISCOFINS": "N", "INCIDENCIA_PISCOFINS": "NENHUMA", "COD_SAIDA_PIS": "01"}) + assert bad["FLAG_CONSISTENCY"] == "INCONSISTENT" + + +def test_empty_pis_cofins_is_unknown_applicability_not_a_blocker(): + found = r.expanded_findings({"CODIGO": 1, "PISCOFINS": "N", "INCIDENCIA_PISCOFINS": "NENHUMA"}, caps(), CATALOG) + assert found["blockers"] == [] and found["unknown"] == ["PIS_COFINS"] and found["rtc_applicability"] == "NOT_ENABLED_BY_ERP" + + +def test_populated_pis_cofins_is_applicable_and_valid_data_completes(): + row = {"CODIGO": 1, "PISCOFINS": "S", "INCIDENCIA_PISCOFINS": "SUBST.TRIBUTARIA", "COD_ENTRADA_PIS": "73", "COD_SAIDA_PIS": "01", "ALIQ_PIS": "0,65", + "COD_ENTRADA_COFINS": "73", "COD_SAIDA_COFINS": "01", "ALIQ_COFINS": "3"} + found = r.expanded_findings(row, caps(), CATALOG) + assert found["blockers"] == [] and found["unknown"] == [] and found["pis_cofins_applicability"] == "APPLICABLE" + + +def test_invalid_values_where_applicable_create_specific_blockers(): + row = {"CODIGO": 2, "PISCOFINS": "N", "INCIDENCIA_PISCOFINS": "NENHUMA", "COD_ENTRADA_PIS": "2", "COD_SAIDA_PIS": "2", "ALIQ_PIS": "2", + "COD_ENTRADA_COFINS": "2", "COD_SAIDA_COFINS": "2", "ALIQ_COFINS": "x"} + kinds = {k for k, _ in r.expanded_findings(row, caps(), CATALOG)["blockers"]} + assert {"PIS_CST_INVALID", "COFINS_CST_INVALID", "COFINS_RATE_INVALID", "PIS_COFINS_FLAG_INCONSISTENT"} <= kinds + + +def test_rtc_becomes_applicable_only_when_the_erp_enables_it(): + on = caps(param=1) + assert r.rtc_applicability(on) == "APPLICABLE" + assert {k for k, _ in r.expanded_findings({"CODIGO": 1}, on, CATALOG)["blockers"]} == {"CCLASSTRIB_MISSING"} + assert {k for k, _ in r.expanded_findings({"CODIGO": 1, "CCLASSTRIB": "123456"}, on, CATALOG)["blockers"]} == {"CCLASSTRIB_NOT_IN_CATALOG"} + assert {k for k, _ in r.expanded_findings({"CODIGO": 1, "CCLASSTRIB": "999999"}, on, CATALOG)["blockers"]} == {"RTC_CST_INVALID"} + assert {k for k, _ in r.expanded_findings({"CODIGO": 1, "CCLASSTRIB": "12x"}, on, CATALOG)["blockers"]} == {"CCLASSTRIB_INVALID"} + assert not [k for k, _ in r.expanded_findings({"CODIGO": 1, "CCLASSTRIB": "200003"}, on, CATALOG)["blockers"] if k.startswith(("CCLASS", "RTC"))] + assert r.rtc_applicability(caps(param=None)) == "UNKNOWN_APPLICABILITY" + + +def test_legacy_and_expanded_coverage_are_reported_side_by_side_and_close(): + products = [{"CODIGO": 1}, {"CODIGO": 2}, {"CODIGO": 3}, + {"CODIGO": 4, "PISCOFINS": "S", "INCIDENCIA_PISCOFINS": "MONOFASICA", "COD_SAIDA_PIS": "01", "COD_SAIDA_COFINS": "01", "COD_ENTRADA_PIS": "73", "COD_ENTRADA_COFINS": "73"}, + {"CODIGO": 5, "COD_SAIDA_PIS": "2", "PISCOFINS": "S"}] + states = {"1": "COMPLETE", "2": "BLOCKED_MISSING_EVIDENCE", "3": "FAILED_TECHNICAL", "4": "COMPLETE", "5": "CORRECTED"} + out = r.expanded_scope_summary(products, states, caps(), CATALOG) + assert out["legacy_scope"] == {"complete": 3, "safe_blocked": 1, "failed": 1} + # 1 unknown (no PIS/COFINS evidence), 1 complete (#4 valid), blocked = legacy-blocked #2 + #5 (invalid CST), failed #3 + assert out["expanded_scope"] == {"complete": 1, "safe_blocked": 2, "failed": 1, "unknown_applicability": 1} + assert out["invariant_pass"] and out["blockers_by_type"].get("PIS_CST_INVALID") == 1 and out["local_catalog_only"] + + +class FakeCursor: + def __init__(self, tables, rows): + self.tables, self.rows, self.result = tables, rows, [] + + def execute(self, sql): + assert sql.lstrip().upper().startswith("SELECT") # read-only by construction + if "RDB$RELATION_FIELDS" in sql: + self.result = [(t, c) for t, cols in self.tables.items() for c in cols] + elif "USAR_REFORMA_TRIBUTARIA" in sql: + self.result = [(0,)] + elif "RDB$RELATION_CONSTRAINTS" in sql: + self.result = [(1,)] + elif "FROM RTC_CCLASSTRIB" in sql: + self.result = [("000001", "000")] + elif "FROM RTC_CST" in sql: + self.result = [("000",)] + elif "FROM C000025" in sql: + self.result = self.rows + + def fetchall(self): + return self.result + + def fetchone(self): + return self.result[0] if self.result else None + + +def test_loader_reads_schema_catalog_and_param_with_select_only(): + cur = FakeCursor({"C000025": ["CODIGO", "CCLASSTRIB"], "RTC_CCLASSTRIB": ["CCLASSTRIB", "CSTIBSCBS"], "RTC_CST": ["CSTIBSCBS"], "PARAMETROS": ["USAR_REFORMA_TRIBUTARIA"]}, [(1, None)]) + schema, catalog, extras = r.load_rtc_context(cur) + assert catalog == {"classes": {"000001": "000"}, "cst": {"000"}, "duplicates": 0} and extras == {"param_value": 0, "physical_fk": True} + assert r.read_products_rtc_pis_cofins(cur, schema) == [{"CODIGO": 1, "CCLASSTRIB": None}] diff --git a/central/app/panel_presentation.py b/central/app/panel_presentation.py index 6203936..02221d2 100644 --- a/central/app/panel_presentation.py +++ b/central/app/panel_presentation.py @@ -120,8 +120,16 @@ def job_kind_label(value): return _human_label(value, {"SYNC_INCREMENTAL": "Sincronização incremental", "INITIAL_SYNC": "Sincronização inicial", "REPROCESS_ALL": "Reprocessar tudo", "REPROCESS_PRODUCTS": "Reprocessar produtos"}) +_RTC_LABELS = { + "CCLASSTRIB_MISSING": "cClassTrib ausente", "CCLASSTRIB_INVALID": "cClassTrib inválido", "CCLASSTRIB_NOT_IN_CATALOG": "cClassTrib fora do catálogo local", + "RTC_CST_INVALID": "CST IBS/CBS inválido", "PIS_CST_MISSING": "CST PIS ausente", "PIS_CST_INVALID": "CST PIS inválido", "PIS_RATE_INVALID": "Alíquota PIS inválida", + "COFINS_CST_MISSING": "CST COFINS ausente", "COFINS_CST_INVALID": "CST COFINS inválido", "COFINS_RATE_INVALID": "Alíquota COFINS inválida", + "PIS_COFINS_FLAG_INCONSISTENT": "Flag PIS/COFINS inconsistente", +} + + def blocker_type_label(value): - return _human_label(value, {"BLOCKED_MISSING_EVIDENCE": "Evidência fiscal ausente", "MISSING_EVIDENCE": "Evidência fiscal ausente", "TECHNICAL_FAILURE": "Falha técnica", "CONNECTION_ERROR": "Erro de conexão", "OTHER": "Outro"}) + return _human_label(value, {"BLOCKED_MISSING_EVIDENCE": "Evidência fiscal ausente", "MISSING_EVIDENCE": "Evidência fiscal ausente", "TECHNICAL_FAILURE": "Falha técnica", "CONNECTION_ERROR": "Erro de conexão", "OTHER": "Outro", **_RTC_LABELS}) _REASON_LABELS = { @@ -150,3 +158,14 @@ def health_reason_label(value): def priority_label(value): return _human_label(value, {"CRITICAL": "Crítica", "HIGH": "Alta", "NORMAL": "Média", "MEDIUM": "Média", "LOW": "Baixa"}) + + +def expanded_scope_view(summary): + """Panel-ready view of the Agent's `expanded_scope` job summary (legacy and expanded coverage side by side). None when absent/unavailable.""" + if not isinstance(summary, dict) or "expanded_scope" not in summary: + return None + legacy, expanded, total = summary.get("legacy_scope") or {}, summary["expanded_scope"] or {}, summary.get("total_products") or 0 + percent = lambda part: round(100 * part / total, 1) if total else None + return {"total": total, "legacy": legacy, "expanded": expanded, "legacy_percent": percent(legacy.get("complete", 0)), "expanded_percent": percent(expanded.get("complete", 0)), + "blockers": [{"type": key, "label": blocker_type_label(key), "count": count} for key, count in (summary.get("blockers_by_type") or {}).items()], + "capabilities": summary.get("capabilities") or {}, "local_catalog_only": bool(summary.get("local_catalog_only"))} diff --git a/central/app/web.py b/central/app/web.py index 33c5eb1..20c6567 100644 --- a/central/app/web.py +++ b/central/app/web.py @@ -15,7 +15,7 @@ from .config import settings from .blocker_resolution import BlockerResolutionUnavailable, _company_key, resolve_blocker_with_context from .context_keys import ALLOWED_KEYS, ENUMS from .installation_identity import reference_agent_version -from .panel_presentation import job_kind_label, job_status_label +from .panel_presentation import expanded_scope_view, job_kind_label, job_status_label from .control_plane import blocker_detail, blocker_filter_summary, dashboard_summary, installation_view, jobs_status_summary, list_blockers, list_installations, list_jobs, list_panel_events from .human_auth import create_session, current_session, require_csrf, verify_password from .panel_presentation import format_ui_datetime @@ -295,7 +295,8 @@ def router(session_factory): next_sync = None if installation.processing_enabled and last_sync is not None: next_sync = last_sync + timedelta(seconds=installation.sync_interval_seconds) - return {"enabled": bool(installation.processing_enabled), "interval_minutes": installation.sync_interval_seconds // 60, + last_summary = session.scalar(select(Job.result_summary).where(Job.agent_id == agent.id, Job.status == "COMPLETED", Job.kind.in_(("INITIAL_SYNC", "SYNC_INCREMENTAL", "REPROCESS_ALL"))).order_by(Job.id.desc()).limit(1)) + return {"expanded": expanded_scope_view(last_summary), "enabled": bool(installation.processing_enabled), "interval_minutes": installation.sync_interval_seconds // 60, "last_sync": format_ui_datetime(last_sync) if last_sync else None, "next_sync": format_ui_datetime(next_sync) if next_sync else None, "last_changed": installation.last_sync_changed, "last_job": {"kind": last_job.kind, "status": last_job.status, "id": last_job.id, "kind_label": job_kind_label(last_job.kind), "status_label": job_status_label(last_job.status)} if last_job else None} diff --git a/central/templates/agent_detail.html b/central/templates/agent_detail.html index 9a86e0d..8cf0022 100644 --- a/central/templates/agent_detail.html +++ b/central/templates/agent_detail.html @@ -9,6 +9,10 @@ Reprocessar tudo {% if ops.enabled %}
{% else %}
{% endif %}
Processamento
{{ "Ativo" if ops.enabled else "Pausado" }} · sincronização automática a cada {{ ops.interval_minutes }} min
Última sincronização
{{ ops.last_sync or "ainda não sincronizado" }}{% if ops.last_changed is not none %} · {{ ops.last_changed }} produto(s) alterado(s){% endif %}
Sincronização agendada
{{ ops.next_sync or ("aguardando o primeiro sync" if ops.enabled else "pausado") }}
Último job
{% if ops.last_job %}{{ ops.last_job.kind_label }} · {{ ops.last_job.status_label }}{% else %}nenhum{% endif %}
+{% if ops.expanded %}

Cobertura fiscal

Escopo anterior e escopo ampliado (cClassTrib e PIS/COFINS quando aplicáveis). Catálogo RTC local do ERP.

+
Escopo anterior
{{ ops.expanded.legacy.complete }} completos · {{ ops.expanded.legacy.safe_blocked }} bloqueados · {{ ops.expanded.legacy.failed }} falhas{% if ops.expanded.legacy_percent is not none %} · {{ ops.expanded.legacy_percent|string|replace(".", ",") }}%{% endif %}
+
Escopo ampliado
{{ ops.expanded.expanded.complete }} completos · {{ ops.expanded.expanded.safe_blocked }} bloqueados · {{ ops.expanded.expanded.failed }} falhas · {{ ops.expanded.expanded.unknown_applicability }} com aplicabilidade desconhecida
+
Novos bloqueios
{% for b in ops.expanded.blockers %}{{ b.label }} · {{ b.count }} {% else %}nenhum{% endfor %}
{% endif %}

Contexto da empresa

Cliente
{{ detail.client or "Não informado" }}
Nome de exibição
{{ detail.display_name or "Não informado" }}
Hostname
{{ detail.hostname or "Não informado" }}
Empresa
{{ detail.company_name or detail.company_code or "Não informado" }}
UF
{{ detail.company_uf or "Não informado" }}
CRT
{{ detail.company_crt or "Não informado" }}
Versão do agent
{{ detail.agent_version_display }}
Último heartbeat
{{ detail.last_seen_display }} · {{ detail.last_seen_age }}
Firebird
{{ detail.firebird_version or "Não informado" }} · ODS {{ detail.firebird_ods or "Não informado" }}
IDs técnicos
Agent {{ detail.technical_ids.agent_id }} · instalação {{ detail.technical_ids.installation_db_id }}

Eventos recentes

Ver eventos
{% for event in events[:8] %}
{{ event.occurred_display }}{{ event.event_type }}{{ event.message }}{% if event.job_id %} · Job {{ event.job_id }}{% endif %}{% if event.product_code %} · Produto {{ event.product_code }}{% endif %}Severidade: {{ event.severity }}
{% else %}

Nenhum evento.

{% endfor %}
diff --git a/central/tests/test_operational_view_models.py b/central/tests/test_operational_view_models.py index afedafd..98525c3 100644 --- a/central/tests/test_operational_view_models.py +++ b/central/tests/test_operational_view_models.py @@ -574,3 +574,14 @@ def test_remaining_technical_enums_have_human_labels(): assert health_reason_label("last heartbeat exceeds 300s") == "Sem heartbeat recente" assert health_reason_label("persistent technical failure") == "Falha técnica persistente" assert blocker_reason_label("SOMETHING_NEW") == "Something new" # unknown codes degrade readably, never crash + + +def test_expanded_scope_view_labels_and_absence(): + from central.app.panel_presentation import expanded_scope_view + assert expanded_scope_view(None) is None and expanded_scope_view({"coverage": {}}) is None + view = expanded_scope_view({"total_products": 10, "legacy_scope": {"complete": 8, "safe_blocked": 2, "failed": 0}, + "expanded_scope": {"complete": 1, "safe_blocked": 3, "failed": 0, "unknown_applicability": 6}, + "blockers_by_type": {"PIS_CST_INVALID": 1, "CCLASSTRIB_MISSING": 2}, "local_catalog_only": True}) + assert view["legacy_percent"] == 80.0 and view["expanded_percent"] == 10.0 and view["local_catalog_only"] + assert {b["label"] for b in view["blockers"]} == {"CST PIS inválido", "cClassTrib ausente"} + assert blocker_type_label("RTC_CST_INVALID") == "CST IBS/CBS inválido" and blocker_type_label("COFINS_RATE_INVALID") == "Alíquota COFINS inválida" diff --git a/docs/rtc_pis_cofins_model.md b/docs/rtc_pis_cofins_model.md new file mode 100644 index 0000000..d96dfa9 --- /dev/null +++ b/docs/rtc_pis_cofins_model.md @@ -0,0 +1,60 @@ +# Modelo RTC + PIS/COFINS no SGV (evidência: TESTESOFICIAIS.FDB) + +Tudo abaixo é observação read-only. Nenhum valor é preenchido, nenhum write é habilitado (`FIREBIRD_WRITES=0`). +Código: `agent/src/gti_fiscal_agent/rtc_pis_cofins.py` (capabilities, validadores, aplicabilidade, cobertura ampliada), +integrado ao job de sync do Agent (`expanded_scope` no `result_summary`) e exibido no detalhe do Agent. + +## 1. Modelo real do RTC + +`C000025.CCLASSTRIB` (VARCHAR(6)) → `RTC_CCLASSTRIB.CCLASSTRIB` (CHAR(6), PK) → `RTC_CCLASSTRIB.CSTIBSCBS` (CHAR(3)) → `RTC_CST.CSTIBSCBS` (CHAR(3), PK). + +- **Há FK física** nos dois elos: `FK_C000025_CCLASSTRIB` (RESTRICT/RESTRICT) e `FK_RTC_CCLASSTRIB_CST`. +- Catálogo local: 145 cClassTrib (6 dígitos, únicos, 0 duplicados, 0 órfãos) para 18 CST IBS/CBS (3 dígitos); nenhum registro com `FIM_VIGENCIA`; `INICIO_VIGENCIA` de 2025-05-05 a 2025-11-19. Cada CST tem ao menos um cClassTrib. +- Atributos do catálogo (não são valores do produto): `TIPO_ALIQUOTA`, `PRED_IBS`/`PRED_CBS` (redução), `IND_TRIB_REGULAR`, `IND_CREDPRES_OPER`, `IND_ESTORNO_CRED`, monofasia, `IND_NFE/NFCE/...` (documentos aplicáveis), `ANEXO`, `LINK`. `RTC_CST` descreve o CST (`IND_IBSCBS`, `IND_REDBC`, `IND_REDALIQ`, `IND_DIF`, `IND_CREDPRES_IBSZFM`...). +- **Não existem no produto**: CST IBS/CBS, alíquotas IBS/CBS, cCredPres. O CST IBS/CBS só se obtém **do catálogo, a partir do cClassTrib informado**. +- `PARAMETROS.USAR_REFORMA_TRIBUTARIA = 0`: o ERP não usa RTC neste banco. Documentos (`C000062`, `C000032`, `C000088`, `NFCE_ITENS`) não têm campos RTC. +- `LOCAL_CATALOG_ONLY=true`: comparamos apenas com o catálogo do próprio banco. Não há dataset oficial integrado; o ponto de extensão é `catalog` em `expanded_scope_summary`/`validate_cclasstrib`. + +## 2. PIS/COFINS no produto (`C000025`) + +| Campo | Tipo | Semântica observada | +|---|---|---| +| `COD_ENTRADA_PIS` / `COD_ENTRADA_COFINS` | VARCHAR(3) | CST de entrada (ex.: `73`); 2/521 preenchidos | +| `COD_SAIDA_PIS` / `COD_SAIDA_COFINS` | VARCHAR(3) | CST de saída (ex.: `01`, `04`); 3/521 | +| `ALIQ_PIS` / `ALIQ_COFINS` | VARCHAR(8) | percentual **textual** com vírgula (`0,65`, `3`); 2 não vazios | +| `PISCOFINS` | VARCHAR(1) | flag S/N (N=519, S=2) | +| `INCIDENCIA_PISCOFINS` | VARCHAR(30) | regime adicional: NENHUMA=519, MONOFÁSICA=1, SUBST.TRIBUTÁRIA=1 (coerente com o flag) | + +- CST oficial PIS/COFINS tem 2 dígitos; um produto traz `2` em todos os campos (formato inválido, aparenta dado de teste) e é o único com flag inconsistente (`N` com códigos). +- Não existe catálogo PIS/COFINS no SGV (nenhuma tabela com PIS/COFINS/CST no nome além de `RTC_CST`). Os códigos são texto livre. +- **Produto × documento.** `C000062` (itens de saída) grava `COD_*_PIS/COFINS` e alíquotas por item (7 de 693 com `20`/`41`/`12`); `NFCE_ITENS.CST_PIS_COFINS` tem `01` em 3.336 itens, inclusive de produtos com `INCIDENCIA=NENHUMA` e sem código no produto. Conclusão: o documento **não** depende do cadastro do produto para o CST. `PRODUCT_DEFAULT_MODEL` = campos esparsos em `C000025` (2–3 produtos); `DOCUMENT_ITEM_MODEL` = campos por item; `CANONICAL_SOURCE_FOR_PRODUCT_DEFAULT` = `C000025` somente onde preenchido, sem fonte canônica para os demais. + +## 3. Aplicabilidade (conservadora) + +Quatro coisas separadas: SCHEMA_SUPPORT, DATA_PRESENT, APPLICABILITY, MANDATORYNESS. Só a última bloqueia, e só com aplicabilidade provada por dado local. + +- **RTC**: `NOT_SUPPORTED` (schema sem o modelo) · `NOT_ENABLED_BY_ERP` (parâmetro do ERP = 0; não bloqueia nem conta como desconhecido) · `UNKNOWN_APPLICABILITY` (parâmetro ilegível) · `APPLICABLE` (parâmetro = 1 e modelo existe → cClassTrib passa a ser exigido/validado). +- **PIS/COFINS**: `APPLICABLE` só quando o próprio produto traz evidência (algum código/alíquota, `PISCOFINS='S'` ou incidência especial). Produto vazio com `N`/`NENHUMA` é `UNKNOWN_APPLICABILITY`: não é completo nem bloqueado. + +## 4. Validadores (determinísticos, sem preenchimento) + +- cClassTrib: `EMPTY`, `INVALID_FORMAT` (≠ 6 dígitos), `NOT_IN_LOCAL_CATALOG`, `VALID_LOCAL`, `CST_MISSING`, `VALID_WITH_CST`. +- CST PIS/COFINS: `EMPTY`, `VALID`, `INVALID_FORMAT` (fora do conjunto oficial 01–09, 49, 50–56, 60–67, 70–75, 98, 99). Alíquota: `EMPTY`, `VALID` (0–100, vírgula ou ponto), `INVALID_RATE`. Flag: `CONSISTENT`/`INCONSISTENT`/`EMPTY`. +- Blockers específicos (rótulos PT-BR no painel): `CCLASSTRIB_MISSING/INVALID/NOT_IN_CATALOG`, `RTC_CST_INVALID`, `PIS|COFINS_CST_MISSING/INVALID`, `PIS|COFINS_RATE_INVALID`, `PIS_COFINS_FLAG_INCONSISTENT`. Eles ficam em `expanded_scope.blockers_by_type`, **separados** dos blockers legados (não alteram o invariante 495+26+0=521 do escopo anterior). + +## 5. Cobertura: legado × ampliada + +`legacy_scope` (LEGACY_SCOPE_COVERAGE) é preservado sem alteração. `expanded_scope` (EXPANDED_FISCAL_SCOPE_COVERAGE) só **acrescenta** condições: bloqueado/falha legado continua assim; completo legado vira completo ampliado, bloqueado (blocker específico) ou `unknown_applicability`. Invariante: soma = total. + +Resultado no TESTESOFICIAIS (521 produtos): legado 495 completos / 26 bloqueados / 0 falhas; ampliado 2 completos / 27 bloqueados / 0 falhas / 492 com aplicabilidade desconhecida; blockers novos: `PIS_CST_INVALID` 2, `COFINS_CST_INVALID` 2, `PIS_COFINS_FLAG_INCONSISTENT` 1 (mesmo produto). Os 492 desconhecidos **não** são defeitos: faltam evidências para dizer se o cadastro do produto deve trazer CST de PIS/COFINS. + +## 6. Limitações + +- Sem catálogo oficial RTC nem regra tributária: nenhuma correspondência é inventada. +- A aplicabilidade de PIS/COFINS por regime (CRT 1 aqui) não está determinada por dado local confiável. +- Vigência do cClassTrib não é usada como bloqueio (o catálogo local não tem `FIM_VIGENCIA`). + +## 7. Campos candidatos a write futuro (NÃO autorizado) + +`CANDIDATE`, sem write habilitado e sem autorização: `C000025.CCLASSTRIB`, `COD_ENTRADA_PIS`, `COD_SAIDA_PIS`, `ALIQ_PIS`, `COD_ENTRADA_COFINS`, `COD_SAIDA_COFINS`, `ALIQ_COFINS`. +Não devem ser escritos: `RTC_CCLASSTRIB`, `RTC_CST` (catálogos), `PARAMETROS.USAR_REFORMA_TRIBUTARIA`, `PISCOFINS`/`INCIDENCIA_PISCOFINS` sem evidência de regime, campos de documentos fiscais, e qualquer CST IBS/CBS "inventado" no produto.