GTI_GERENCIADOR_FISCAL/fiscal/scripts/audit_company_context.py

"""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()