From e0ea8421686f34711dc282b8a04d46391825641a Mon Sep 17 00:00:00 2001 From: gti Date: Thu, 24 Sep 2026 15:25:31 -0300 Subject: [PATCH] docs: read-only fiscal/RTC/PIS-COFINS discovery of TESTESOFICIAIS.FDB Co-Authored-By: Claude Sonnet 5 --- docs/testesoficiais_fiscal_rtc_discovery.md | 60 +++++++ scripts/discover_testesoficiais_fiscal.py | 168 ++++++++++++++++++++ tests/test_discover_testesoficiais_guard.py | 35 ++++ 3 files changed, 263 insertions(+) create mode 100644 docs/testesoficiais_fiscal_rtc_discovery.md create mode 100644 scripts/discover_testesoficiais_fiscal.py create mode 100644 tests/test_discover_testesoficiais_guard.py diff --git a/docs/testesoficiais_fiscal_rtc_discovery.md b/docs/testesoficiais_fiscal_rtc_discovery.md new file mode 100644 index 0000000..d4c5826 --- /dev/null +++ b/docs/testesoficiais_fiscal_rtc_discovery.md @@ -0,0 +1,60 @@ +# TESTESOFICIAIS.FDB — discovery fiscal / RTC / PIS-COFINS + +Somente leitura, somente `TESTESOFICIAIS.FDB` (Firebird 2.5, WI-V6.3.9.27139, ODS 11.2; 325 tabelas, 5.390 colunas). +Isolamento `READ COMMITTED READ ONLY`; toda instrução passa por um guarda que só deixa `SELECT` chegar ao driver. +`WRITE_ATTEMPTS=0`, `WRITE_SUCCESSES=0`. Script: `scripts/discover_testesoficiais_fiscal.py` (sem credenciais no código; lê `FB_USER`/`FB_PASSWORD` do ambiente ou do `.env` local). + +Nada aqui infere cClassTrib pelo NCM nem CST IBS/CBS pelo CST de ICMS. O NCM do produto está em `C000025.CLASSIFICACAO_FISCAL` (a coluna `NCM` existe, mas está 100% vazia). + +## C000025 (produtos, 521 linhas) + +| Pergunta | Resposta | Coluna real | +|---|---|---| +| cClassTrib no produto | **SIM (schema), sem uso** | `CCLASSTRIB` VARCHAR(6), nulo, sem default — 0 preenchidos | +| CST IBS/CBS no produto | NÃO | — (só existe `CSTIBSCBS` nas tabelas auxiliares `RTC_*`) | +| Campos IBS no produto | NÃO | — | +| Campos CBS no produto | NÃO | — | +| Crédito presumido (cCredPres) no produto | NÃO | — (só indicadores `IND_CREDPRES_*` nos catálogos `RTC_*`) | + +### PIS / COFINS no produto (tudo em `C000025`, todas VARCHAR e anuláveis) + +| Item | Existe | Coluna | Preenchidos / 521 | +|---|---|---|---| +| CST PIS entrada | sim | `COD_ENTRADA_PIS` VARCHAR(3) | 2 | +| CST PIS saída | sim | `COD_SAIDA_PIS` VARCHAR(3) | 3 | +| Alíquota PIS | sim (texto) | `ALIQ_PIS` VARCHAR(8) | 3 (2 não vazios) | +| CST COFINS entrada | sim | `COD_ENTRADA_COFINS` VARCHAR(3) | 2 | +| CST COFINS saída | sim | `COD_SAIDA_COFINS` VARCHAR(3) | 3 | +| Alíquota COFINS | sim (texto) | `ALIQ_COFINS` VARCHAR(8) | 3 (2 não vazios) | +| Flag/incidência | sim | `PISCOFINS` VARCHAR(1) (N=519, S=2); `INCIDENCIA_PISCOFINS` VARCHAR(30) (NENHUMA=519, MONOFÁSICA=1, SUBST.TRIBUTÁRIA=1) | 521 | + +Os 3 produtos com dado: (saída `04`, monofásica), (entrada/saída `73`/`01`, alíquotas 0,65/3, subst. tributária) e um com o valor `2` em todos os campos (aparenta dado de teste). Ou seja: **suporte no schema, uso real quase nulo** neste banco. + +Outras colunas fiscais do produto (contexto, não RTC): `CST` (ICMS, 5 valores, 487× `040`), `CSOSN`, `CFOP` (5102 em 493), `CFOP_ENTRADA`/`CST_ENTRADA` (vazios), `CEST`, `ALIQUOTA`, `IPI`, `CST_IPI`, `GRUPO_ICMS`, `SITUACAO_TRIBUTARIA`, `COD_ORIGEM_MERCADORIA`, `ORIGEM`, `EX_TIPI`, `NCM` (vazia), `CLASSIFICACAO_FISCAL` (NCM, 54 distintos). + +## Onde está o RTC + +- `RTC_CCLASSTRIB` (145 linhas, catálogo): `CCLASSTRIB` CHAR(6), `CSTIBSCBS` CHAR(3), `DESCRICAO_CLASSTRIB`, `TIPO_ALIQUOTA`, `PRED_IBS`, `PRED_CBS`, `IND_TRIB_REGULAR`, `IND_CREDPRES_OPER`, `IND_ESTORNO_CRED`, indicadores de monofasia, `IND_NFE/NFCE/CTE/...` (documentos aplicáveis), `PUBLICACAO`, `INICIO_VIGENCIA` (2025-05-05 a 2025-11-19), `FIM_VIGENCIA`, `ANEXO`, `LINK`. Cada cClassTrib aponta um CST IBS/CBS (18 distintos). +- `RTC_CST` (18 linhas, catálogo): `CSTIBSCBS` CHAR(3), `DESCRICAO_CST`, `IND_IBSCBS`, `IND_REDBC`, `IND_REDALIQ`, `IND_TRANSFCRED`, `IND_DIF`, `IND_AJUSTE_COMPET`, `IND_IBSCBS_MONO`, `IND_CREDPRES_IBSZFM`, vigência. +- `PARAMETROS.USAR_REFORMA_TRIBUTARIA` = `0` (recurso desligado neste banco). +- Nenhuma procedure ou trigger de usuário referencia RTC/CLASSTRIB. +- **Nenhuma tabela de documento** (`C000062`, `C000032`, `C000088`, `NFCE_ITENS`…) tem `CCLASSTRIB`/`CSTIBSCBS`/IBS/CBS. Só `C000025.CCLASSTRIB` e as duas tabelas de catálogo têm essas colunas. +- Integridade: 0 produtos com `CCLASSTRIB` preenchido, portanto 0 referências ao catálogo. + +## Semântica (por tabela) + +| Classe | Campos | +|---|---| +| PRODUCT_DEFAULT | `C000025.CCLASSTRIB`, `COD_ENTRADA_PIS`, `COD_SAIDA_PIS`, `ALIQ_PIS`, `COD_ENTRADA_COFINS`, `COD_SAIDA_COFINS`, `ALIQ_COFINS`, `PISCOFINS`, `INCIDENCIA_PISCOFINS`; `C000026` (composição de custo/preço por produto, 3 linhas) com `PIS`, `COFINS`, `PIS_ENTRADA`, `COFINS_ENTRADA` e variantes `_P` | +| DOCUMENT_ITEM_FIELD | `C000062` (itens de nota de saída: `COD_ENTRADA/SAIDA_PIS/COFINS`, `PIS_ALIQUOTA/BASE/VALOR`, `COFINS_*`); `C000032` (itens de venda: `PIS`, `COFINS`); `C000088` (itens de entrada: `PIS`, `PIS_BASE`, `COFINS`, `COFINS_BASE`); `C000087`/`C000061` (cabeçalhos: `PIS`, `COFINS`, `ITEM_PIS_*`); `NFCE_ITENS` (`CST_PIS_COFINS`, `ALIQUOTA_PIS/COFINS`, `BASE_CALCULO_*`, `VALOR_*`); `MOVIMENTACAO_COMBUSTIVEL` | +| COMPANY_CONFIG | `PARAMETROS.USAR_REFORMA_TRIBUTARIA`; `C000004.PIS`/`COFINS` (vazios) | +| AUXILIARY_CONFIG | `RTC_CCLASSTRIB`, `RTC_CST` (catálogos oficiais); `C000084A/B` (`ALIQUOTA_PIS/COFINS` + `VALIDADE_*`, vazias); `CODIGO_PISCONFINS` nas tabelas `SINTEGRA_*`, `C000080`, `C000086` (vazias) | +| UNKNOWN | `C000025.FOOD_ATRIBUICAO_VALORES`; `C000026.CONTRIBUICAO_SOCIAL*` | + +Uso real em documentos (só para separar SCHEMA_SUPPORT de ACTUAL_USAGE): `NFCE_ITENS.CST_PIS_COFINS` tem valor em 4.668 de 34.946 itens (`01` em 3.336, branco em 1.332); `C000062.COD_SAIDA_PIS` tem `20` em 7 de 693 e `PIS_ALIQUOTA`/`COFINS_ALIQUOTA` em 693 (2 valores distintos). + +## Conclusão + +- `TESTESOFICIAIS_RTC_STATUS`: **suporte parcial ao RTC**. O produto tem `CCLASSTRIB` (vazio) e o banco traz os catálogos `RTC_CCLASSTRIB`/`RTC_CST`; não há CST IBS/CBS, IBS, CBS nem cCredPres no produto, nem nenhum campo RTC em tabelas de documento. Recurso desligado (`USAR_REFORMA_TRIBUTARIA=0`). +- `TESTESOFICIAIS_PIS_COFINS_STATUS`: **schema completo no produto** (CST entrada/saída e alíquota para PIS e COFINS), **uso mínimo** (2–3 de 521 produtos). +- Próximo passo recomendado: tratar `C000025.CCLASSTRIB` como campo de produto a ser lido (read-only) e validado contra `RTC_CCLASSTRIB` (existência e vigência), derivando o CST IBS/CBS **do catálogo, a partir do cClassTrib informado**, nunca de NCM ou CST de ICMS; e, para PIS/COFINS, tratar o produto como preenchimento esparso (a maioria dos 521 produtos depende de regra ou de evidência externa, não do cadastro). diff --git a/scripts/discover_testesoficiais_fiscal.py b/scripts/discover_testesoficiais_fiscal.py new file mode 100644 index 0000000..c9578e0 --- /dev/null +++ b/scripts/discover_testesoficiais_fiscal.py @@ -0,0 +1,168 @@ +"""READ-ONLY discovery of fiscal/RTC/PIS/COFINS fields in TESTESOFICIAIS.FDB (and only that database). + + python scripts/discover_testesoficiais_fiscal.py [--json out.json] + +Credentials come from the process environment (FB_USER / FB_PASSWORD, optional FIREBIRD_CLIENT_LIBRARY) or the +local ignored .env - never from the command line, never printed. The connection uses READ COMMITTED READ ONLY +isolation and every statement goes through a guard that only lets SELECT reach the driver. +""" +import argparse +import json +import os +import re +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +DATABASE = ROOT / "TESTESOFICIAIS.FDB" # fixed on purpose: this script audits no other database + +KEYWORDS = ("TRIB", "IBS", "CBS", "PIS", "COFINS", "CRED", "REFORMA", "RTC", "CLASS") +FISCAL_HINTS = ("TRIB", "CST", "CFOP", "NCM", "PIS", "COF", "IBS", "CBS", "CRED", "ALIQ", "ICMS", "IPI", "CEST", "CSOSN", "CLASS", "ORIG", "REFORMA", "RTC", "SIT_T") +_TYPES = {7: "SMALLINT", 8: "INTEGER", 16: "BIGINT", 10: "FLOAT", 27: "DOUBLE", 12: "DATE", 13: "TIME", 35: "TIMESTAMP", 14: "CHAR", 37: "VARCHAR", 261: "BLOB"} + + +class ReadOnlyCursor: + """Only SELECT statements are ever handed to the driver; anything else is counted and refused.""" + write_attempts = 0 + write_successes = 0 + + def __init__(self, cursor): + self._cursor = cursor + + def execute(self, sql, params=()): + if not re.match(r"^\s*select\b", sql, re.IGNORECASE): + ReadOnlyCursor.write_attempts += 1 + raise RuntimeError("non-SELECT statement refused") + return self._cursor.execute(sql, params) if params else self._cursor.execute(sql) + + def fetchall(self): + return self._cursor.fetchall() + + def fetchone(self): + return self._cursor.fetchone() + + +def load_env(): + env_file = ROOT / ".env" + if env_file.exists(): + for line in env_file.read_text(encoding="utf-8").splitlines(): + if "=" in line and not line.startswith("#"): + key, _, value = line.partition("=") + os.environ.setdefault(key.strip(), value.strip()) + + +def classify(table, column): + """Name-based first pass only. It is never taken as proof of meaning: the report lists evidence and a human/agent decides.""" + name = column.upper() + if re.search(r"CCLASSTRIB|CLASS_?TRIB|CLASSIFICACAO_TRIBUTARIA", name): + return "CONFIRMED_MATCH", "cClassTrib" + if re.search(r"CST_?(IBS|CBS)|(IBS|CBS)_?CST|CSTIBSCBS|CST_IBSCBS", name): + return "CONFIRMED_MATCH", "CST IBS/CBS" + if re.search(r"(^|_)(IBS|CBS)($|_)|IBS|CBS", name) and not re.search(r"CBSA", name): + return "CONFIRMED_MATCH", "IBS/CBS" + if re.search(r"CCREDPRES|CRED_?PRES", name): + return "CONFIRMED_MATCH", "crédito presumido" + if re.search(r"(^|_)PIS($|_)|_PIS|PIS_|COD_(ENTRADA|SAIDA)_PIS", name): + return "CONFIRMED_MATCH", "PIS" + if re.search(r"COFINS", name): + return "CONFIRMED_MATCH", "COFINS" + if re.search(r"REFORMA|(^|_)RTC($|_)", name): + return "POSSIBLE_MATCH", "reforma/RTC (nome)" + if re.search(r"TRIB", name): + return "POSSIBLE_MATCH", "tributação (nome genérico)" + if re.search(r"CRED", name): + return "POSSIBLE_MATCH", "crédito (nome genérico)" + if re.search(r"CLASS", name): + return "POSSIBLE_MATCH", "classificação (nome genérico)" + return "UNRELATED", "" + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--json", default=None) + args = parser.parse_args() + load_env() + sys.path.insert(0, str(ROOT / "agent" / "src")) + import fdb + + conn = fdb.connect(database=str(DATABASE), user=os.environ["FB_USER"], password=os.environ["FB_PASSWORD"], + fb_library_name=os.environ.get("FIREBIRD_CLIENT_LIBRARY"), isolation_level=fdb.ISOLATION_LEVEL_READ_COMMITED_RO) + cur = ReadOnlyCursor(conn.cursor()) + result = {"database": DATABASE.name, "mode": "READ_ONLY", "firebird_server_version": getattr(conn, "server_version", None), + "ods": f"{getattr(conn, 'ods_version', '?')}.{getattr(conn, 'ods_minor_version', '?')}"} + cur.execute("""SELECT TRIM(rf.RDB$RELATION_NAME), TRIM(rf.RDB$FIELD_NAME), f.RDB$FIELD_TYPE, COALESCE(f.RDB$CHARACTER_LENGTH, f.RDB$FIELD_LENGTH), + f.RDB$FIELD_SCALE, rf.RDB$NULL_FLAG, TRIM(rf.RDB$FIELD_SOURCE), rf.RDB$DEFAULT_SOURCE, f.RDB$DEFAULT_SOURCE, rf.RDB$FIELD_POSITION + FROM RDB$RELATION_FIELDS rf JOIN RDB$FIELDS f ON f.RDB$FIELD_NAME = rf.RDB$FIELD_SOURCE + JOIN RDB$RELATIONS r ON r.RDB$RELATION_NAME = rf.RDB$RELATION_NAME + WHERE COALESCE(r.RDB$SYSTEM_FLAG, 0) = 0 AND r.RDB$VIEW_BLR IS NULL""") + columns = [] + for table, column, ftype, length, scale, null_flag, source, rf_default, f_default, position in cur.fetchall(): + default = rf_default or f_default + default = str(default.read() if hasattr(default, "read") else default).replace("DEFAULT", "").strip() if default else None + columns.append({"table": table, "column": column, "type": _TYPES.get(ftype, f"type{ftype}") + (f"(scale {scale})" if scale else ""), "length": length, + "nullable": not bool(null_flag), "default": default, "domain": None if source.startswith("RDB$") else source, "position": position}) + result["tables"] = len({c["table"] for c in columns}) + result["total_columns"] = len(columns) + + c25 = [c for c in columns if c["table"] == "C000025"] + result["c000025_columns"] = [c["column"] for c in sorted(c25, key=lambda c: c["position"])] + result["c000025_fiscal_columns"] = [c for c in sorted(c25, key=lambda c: c["position"]) if any(h in c["column"].upper() for h in FISCAL_HINTS)] + + matches = [] + for column in columns: + verdict, concept = classify(column["table"], column["column"]) + if verdict != "UNRELATED": + matches.append({**column, "classification": verdict, "concept": concept}) + result["matches"] = matches + + def stats(table, column): + cur.execute(f'SELECT COUNT(*), COUNT("{column}"), COUNT(DISTINCT "{column}") FROM "{table}"') + total, non_null, distinct = cur.fetchone() + info = {"rows": total, "non_null": non_null, "distinct": distinct} + cur.execute(f'SELECT FIRST 6 "{column}", COUNT(*) FROM "{table}" WHERE "{column}" IS NOT NULL GROUP BY "{column}" ORDER BY 2 DESC') + info["top_values"] = [[str(v).strip()[:24] if v is not None else None, n] for v, n in cur.fetchall()] + return info + + for match in matches: + if match["type"].startswith(("BLOB",)): + continue + try: + match["usage"] = stats(match["table"], match["column"]) + except Exception as error: # noqa: BLE001 - reported, never fatal + match["usage"] = {"error": type(error).__name__} + for column in result["c000025_fiscal_columns"]: + column["classification"], column["concept"] = classify("C000025", column["column"]) + try: + column["usage"] = stats("C000025", column["column"]) + cur.execute(f"""SELECT COUNT(*) FROM "C000025" WHERE TRIM(COALESCE(CAST("{column["column"]}" AS VARCHAR(40)), '')) NOT IN ('', '0', '0.00', '0,00')""") + column["usage"]["meaningful_non_empty"] = cur.fetchone()[0] + except Exception as error: # noqa: BLE001 + column["usage"] = {"error": type(error).__name__} + # tables that are entirely about the reform (name based), with their full column list and a few rows' worth of values + rtc_tables = sorted({c["table"] for c in columns if re.search(r"RTC|IBS|CBS|REFORMA|CLASSTRIB", c["table"].upper())}) + result["rtc_tables"] = {} + for table in rtc_tables: + cols = [c for c in columns if c["table"] == table] + cur.execute(f'SELECT COUNT(*) FROM "{table}"') + result["rtc_tables"][table] = {"rows": cur.fetchone()[0], "columns": [(c["column"], c["type"], c["length"]) for c in sorted(cols, key=lambda c: c["position"])]} + if table == "RTC_CST": + cur.execute('SELECT FIRST 20 * FROM "RTC_CST"') + result["rtc_tables"][table]["sample"] = [[str(v).strip()[:60] for v in row] for row in cur.fetchall()] + cur.execute('SELECT USAR_REFORMA_TRIBUTARIA FROM PARAMETROS') + result["parametros_usar_reforma_tributaria"] = [str(v[0]) for v in cur.fetchall()] + cur.execute('SELECT COUNT(*) FROM C000025') + result["c000025_rows"] = cur.fetchone()[0] + # how many products carry a CCLASSTRIB that exists in the RTC_CCLASSTRIB catalog (orphan check, no inference) + cur.execute("""SELECT COUNT(*) FROM C000025 p WHERE TRIM(COALESCE(p.CCLASSTRIB, '')) <> '' AND EXISTS (SELECT 1 FROM RTC_CCLASSTRIB c WHERE c.CCLASSTRIB = p.CCLASSTRIB)""") + result["c000025_cclasstrib_in_catalog"] = cur.fetchone()[0] + result["write_attempts"] = ReadOnlyCursor.write_attempts + result["write_successes"] = ReadOnlyCursor.write_successes + conn.close() + text = json.dumps(result, ensure_ascii=False, indent=1, default=str) + if args.json: + Path(args.json).write_text(text, encoding="utf-8") + print(f"tables={result['tables']} columns={result['total_columns']} matches={len(matches)} writes_attempted={result['write_attempts']}") + + +if __name__ == "__main__": + main() diff --git a/tests/test_discover_testesoficiais_guard.py b/tests/test_discover_testesoficiais_guard.py new file mode 100644 index 0000000..eefcc32 --- /dev/null +++ b/tests/test_discover_testesoficiais_guard.py @@ -0,0 +1,35 @@ +import importlib.util +from pathlib import Path + +import pytest + +spec = importlib.util.spec_from_file_location("dsc", Path(__file__).parents[1] / "scripts" / "discover_testesoficiais_fiscal.py") +dsc = importlib.util.module_from_spec(spec) +spec.loader.exec_module(dsc) + + +class Recorder: + def __init__(self): + self.sql = [] + + def execute(self, sql, params=()): + self.sql.append(sql) + + +def test_only_select_reaches_the_driver_and_writes_are_counted(): + rec = Recorder() + cur = dsc.ReadOnlyCursor(rec) + dsc.ReadOnlyCursor.write_attempts = 0 + cur.execute(" select 1 from rdb$database") + for bad in ("UPDATE C000025 SET NCM='1'", "delete from c000025", "INSERT INTO x VALUES (1)", "ALTER TABLE x ADD y INT", "EXECUTE PROCEDURE p", "DROP TABLE x"): + with pytest.raises(RuntimeError): + cur.execute(bad) + assert rec.sql == [" select 1 from rdb$database"] and dsc.ReadOnlyCursor.write_attempts == 6 and dsc.ReadOnlyCursor.write_successes == 0 + + +def test_classification_does_not_infer_from_ncm_or_icms_cst(): + assert dsc.classify("C000025", "CCLASSTRIB")[0] == "CONFIRMED_MATCH" + assert dsc.classify("RTC_CST", "CSTIBSCBS")[1] == "CST IBS/CBS" + assert dsc.classify("C000025", "CLASSIFICACAO_FISCAL")[0] == "POSSIBLE_MATCH" # NCM holder, not cClassTrib + assert dsc.classify("C000025", "CST")[0] == "UNRELATED" # ICMS CST is not CST IBS/CBS + assert dsc.classify("C000025", "NCM")[0] == "UNRELATED"