86 lines
6.4 KiB
Python
86 lines
6.4 KiB
Python
"""READ-ONLY company-context audit + resolver retest over every .FDB in a directory (default ./Banco).
|
|
Writes docs/COMPANY_CONTEXT_RESOLUTION_V1.json (evidence) and docs/MULTI_DATABASE_COMPANY_CONTEXT_V1.json (outcomes).
|
|
The full CNPJ is never written: masked value + short SHA-256 fingerprint only.
|
|
"""
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "agent" / "src"))
|
|
import fdb # noqa: E402
|
|
from gti_fiscal_agent import company_context as cc # noqa: E402
|
|
|
|
for line in (ROOT / ".env").read_text(encoding="utf-8").splitlines():
|
|
if "=" in line and not line.startswith("#"):
|
|
k, _, v = line.partition("=")
|
|
os.environ.setdefault(k.strip(), v.strip())
|
|
|
|
|
|
def evidence_for(path):
|
|
conn = fdb.connect(database=str(path), 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)
|
|
try:
|
|
return cc.collect_evidence(conn.cursor())
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def table(ev):
|
|
rows = []
|
|
emissor_cnpjs = {cc.digits(e["cnpj"]) for e in ev["emissor"]}
|
|
docs_by_cnpj = {}
|
|
for d in ev["nfce_keys"]:
|
|
docs_by_cnpj[cc.digits(d["cnpj"])] = docs_by_cnpj.get(cc.digits(d["cnpj"]), 0) + d["n"]
|
|
for i, c in enumerate(ev["c000004"]):
|
|
cn = cc.digits(c.get("cnpj"))
|
|
em = next((e for e in ev["emissor"] if cc.digits(e["cnpj"]) == cn), None)
|
|
n_docs = docs_by_cnpj.get(cn, 0); n_fil = ev["nf_by_filial"].get(c.get("code"), 0)
|
|
active = bool(em) and n_docs > 0
|
|
rows.append({"C000004_CODE": c.get("code"), "CNPJ": cc.mask_cnpj(cn), "CNPJ_FINGERPRINT": cc.cnpj_fingerprint(cn), "UF_C000004": c.get("uf"), "CRT_C000004": c.get("crt"), "OPTANTE_SIMPLES": c.get("optante"),
|
|
"EMISSOR_ID": (ev["emissor"].index(em) + 1) if em else None, "UF_EMISSOR": em.get("uf") if em else None, "CRT_EMISSOR": em.get("crt") if em else None,
|
|
"DOCUMENT_COUNT": {"nfce_keys_with_this_cnpj": n_docs, "nf_C000061_with_this_CODFILIAL": n_fil}, "IS_ACTIVE_CANDIDATE": active,
|
|
"EVIDENCE": [x for x in (("EMISSOR.EMITENTE_CNPJ equals this CNPJ" if em else "no EMISSOR row with this CNPJ"), (f"{n_docs} emitted NFC-e keys carry this CNPJ" if n_docs else "no emitted NFC-e key carries this CNPJ"),
|
|
(f"{n_fil} NF-e rows use CODFILIAL={c.get('code')}" if n_fil else None)) if x]})
|
|
return rows
|
|
|
|
|
|
def main():
|
|
directory = Path(sys.argv[1]) if len(sys.argv) > 1 else ROOT / "Banco"
|
|
evid, outcomes = {}, {}
|
|
for p in sorted(directory.glob("*.FDB")):
|
|
ev = evidence_for(p)
|
|
rows = table(ev)
|
|
no_cfg = cc.resolve(ev, {})
|
|
crt_cfg = cc.resolve(ev, {"crt_source": "EMISSOR"})
|
|
multi = len(ev["c000004"]) > 1
|
|
neg = {}
|
|
if multi: # negative tests on real data: wrong company / unknown CNPJ must not resolve silently
|
|
other = next((c for c in ev["c000004"] if not next((e for e in ev["emissor"] if cc.digits(e["cnpj"]) == cc.digits(c["cnpj"])), None)), None)
|
|
neg["configured_non_emitting_company"] = cc.resolve(ev, {"company_code": other["code"], "crt_source": "EMISSOR"})["classification"] if other else None
|
|
neg["configured_unknown_cnpj"] = cc.resolve(ev, {"company_cnpj": "00000000000000"})["classification"]
|
|
crt_authority = {"EMISSOR_values": sorted({e.get("crt") for e in ev["emissor"] if e.get("crt")}), "C000004_selected_row": no_cfg.get("crt_candidates") or {"C000004": no_cfg.get("company_crt")}}
|
|
evid[p.stem] = {"c000004_rows": len(ev["c000004"]), "emissor_rows": len(ev["emissor"]), "companies": rows, "nfce_key_groups": [{"cUF": d["cuf"], "cnpj": cc.mask_cnpj(d["cnpj"]), "model": "65", "n": d["n"]} for d in ev["nfce_keys"]],
|
|
"nf_by_filial": ev["nf_by_filial"]}
|
|
outcomes[p.stem] = {"without_config": cc.preflight_report(no_cfg), "with_COMPANY_CRT_SOURCE=EMISSOR": cc.preflight_report(crt_cfg), "negative_tests": neg,
|
|
"crt_source_status": no_cfg.get("crt_status") or ("CRT_SOURCE_CONFLICT" if "CRT_SOURCE_CONFLICT" in " ".join(no_cfg["reasons"]) else "CRT_SOURCE_UNKNOWN"),
|
|
"uf_source_status": no_cfg.get("uf_status") or "UF_SOURCE_UNKNOWN", "resolution_class": no_cfg["classification"]}
|
|
classes = {}
|
|
for k, v in outcomes.items():
|
|
classes.setdefault(v["resolution_class"], []).append(k)
|
|
authority = {
|
|
"CRT": {"emission_authority": "EMISSOR.EMITENTE_CRT", "evidence": ["GtiVendas_NFCe.exe (NFC-e) and Envio NF-e.exe (NF-e) bind QrEmissor.EMITENTE_CRT/EMITENTE_UF/EMITENTE_CNPJ and never reference C000004.CRT or OPTANTE_SIMPLES",
|
|
"SGV_5.0.6.124.exe (ERP) binds C000004.CRT (qrFilialCRT/QrSPEDFilialCRT) and OPTANTE_SIMPLES for the filial registry/SPED and has no EMITENTE_* binding",
|
|
"emitted XML is not stored in the database (C000061.NFE_XML and XML_MES.XML hold file paths), so the CRT tag of issued documents could not be read directly"],
|
|
"status": "CRT_SOURCE_CONFLICT where the two values differ; the resolver refuses to choose without COMPANY_CRT_SOURCE"},
|
|
"UF": {"status": "UF_SOURCE_CONFIRMED", "evidence": "EMISSOR.EMITENTE_UF, C000004.UF and the cUF of 100% of emitted NFC-e keys agree in all 4 databases"},
|
|
"COMPANY": {"status": "confirmed", "evidence": "EMISSOR.EMITENTE_CNPJ equals exactly one C000004 CNPJ in every database and 100% of emitted NFC-e keys carry that CNPJ; other C000004 rows (RINCAO) have no documents"}}
|
|
(ROOT / "docs" / "COMPANY_CONTEXT_RESOLUTION_V1.json").write_text(json.dumps({"artifact": "COMPANY_CONTEXT_RESOLUTION_V1", "firebird_writes": 0, "authority": authority, "databases": evid}, ensure_ascii=False, indent=1), encoding="utf-8")
|
|
(ROOT / "docs" / "MULTI_DATABASE_COMPANY_CONTEXT_V1.json").write_text(json.dumps({"artifact": "MULTI_DATABASE_COMPANY_CONTEXT_V1", "firebird_writes": 0, "resolution_classes": classes, "databases": outcomes}, ensure_ascii=False, indent=1), encoding="utf-8")
|
|
print(json.dumps({k: (v["resolution_class"], v["with_COMPANY_CRT_SOURCE=EMISSOR"]["COMPANY_CONTEXT"], v["with_COMPANY_CRT_SOURCE=EMISSOR"]["company_crt"], v["negative_tests"]) for k, v in outcomes.items()}, ensure_ascii=False))
|
|
print(classes)
|
|
|
|
|
|
main()
|