GTI_GERENCIADOR_FISCAL/fiscal/scripts/audit_firebird_database.py

"""READ-ONLY audit of any SGV Firebird database (restaurante / mercado / oficina ...).

    python audit_firebird_database.py <FDB> [--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-only isolation and the
script executes SELECT statements only. It never enables writes for the audited database.
"""
import argparse
import json
import os
import re
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "agent" / "src")); sys.path.insert(0, str(ROOT / "src"))

FISCAL_FIELDS = ["CLASSIFICACAO_FISCAL", "NCM", "CFOP", "CFOP_ENTRADA", "CST", "CST_ENTRADA", "ALIQUOTA", "CEST", "CSOSN", "IPI", "EX_TIPI", "CST_IPI",
                 "COD_ENTRADA_PIS", "COD_SAIDA_PIS", "ALIQ_PIS", "COD_ENTRADA_COFINS", "COD_SAIDA_COFINS", "ALIQ_COFINS", "ORIGEM", "COD_ORIGEM_MERCADORIA",
                 "IAT", "INCIDENCIA_PISCOFINS", "IPPT", "GRUPO_ICMS", "SITUACAO_TRIBUTARIA", "MARGEM_AGREGADA"]
# Firebird RDB$FIELD_TYPE -> family
_FAMILY = {7: "int", 8: "int", 16: "int", 10: "float", 11: "float", 27: "float", 12: "date", 13: "time", 35: "timestamp", 14: "text", 37: "text", 261: "blob"}
# expected type family per field (from docs/ERP_FISCAL_FIELD_MAP_V1.json); text fields may also be char
EXPECTED_FAMILY = {"CLASSIFICACAO_FISCAL": "text", "NCM": "text", "CFOP": "text", "CFOP_ENTRADA": ("text", "int"), "CST": "text", "CST_ENTRADA": ("text", "int"), "ALIQUOTA": ("float", "int"),
                   "CEST": "text", "CSOSN": "text", "IPI": ("float", "int"), "EX_TIPI": "text", "CST_IPI": ("text", "int"), "COD_ENTRADA_PIS": "text", "COD_SAIDA_PIS": "text",
                   "ALIQ_PIS": "text", "COD_ENTRADA_COFINS": "text", "COD_SAIDA_COFINS": "text", "ALIQ_COFINS": "text", "ORIGEM": "text", "COD_ORIGEM_MERCADORIA": "int",
                   "IAT": "text", "INCIDENCIA_PISCOFINS": "text", "IPPT": "text", "GRUPO_ICMS": "text", "SITUACAO_TRIBUTARIA": "text", "MARGEM_AGREGADA": ("float", "int")}


def type_family(code):
    return _FAMILY.get(code, f"other:{code}")


def ncm_bucket(value):
    v = "" if value is None else str(value).strip()
    if not v:
        return "blank"
    return "eight_digits" if re.fullmatch(r"\d{8}", v) else "other_format"


def compat(columns):
    """columns: {FIELD: type_code}. Returns per-field compatibility against the expected families."""
    out = {}
    for f, exp in EXPECTED_FAMILY.items():
        if f not in columns:
            out[f] = "MISSING"
            continue
        fam = type_family(columns[f]); exp = (exp,) if isinstance(exp, str) else exp
        out[f] = "OK" if fam in exp else f"TYPE_DIFFERS({fam} not in {'/'.join(exp)})"
    return out


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("#"):
                k, _, v = line.partition("="); os.environ.setdefault(k.strip(), v.strip())


def audit(path, env):
    import fdb
    from gti_discovery.fingerprint import build_fingerprint
    from gti_fiscal_agent.firebird_reader import parse_company_context
    conn = fdb.connect(database=path, user=env["FB_USER"], password=env["FB_PASSWORD"], fb_library_name=env.get("FIREBIRD_CLIENT_LIBRARY"),
                       isolation_level=fdb.ISOLATION_LEVEL_READ_COMMITED_RO)
    try:
        cur = conn.cursor()
        res = {"database": Path(path).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
                       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""")
        meta = {}
        for t, c, ty, ln, sc in cur.fetchall():
            meta.setdefault(t, []).append({"name": c, "type": ty, "length": ln, "scale": sc})
        res["schema_fingerprint"] = build_fingerprint({"tables": [{"name": t, "columns": cols} for t, cols in sorted(meta.items())]})
        res["tables"] = len(meta)
        # company context
        if "C000004" in meta:
            try:
                cur.execute("SELECT UF, CRT FROM C000004"); res["company"] = {"status": "OK", **parse_company_context(cur.fetchall())}
            except ValueError as e:
                res["company"] = {"status": "UNAVAILABLE", "error": str(e)}
        else:
            res["company"] = {"status": "C000004_ABSENT"}
        # products
        res["C000025_present"] = "C000025" in meta
        if res["C000025_present"]:
            cols = {c["name"]: c["type"] for c in meta["C000025"]}
            res["field_mapping_compatibility"] = compat(cols)
            res["field_mapping_compatibility_summary"] = {k: sum(1 for v in res["field_mapping_compatibility"].values() if (v == k or v.startswith(k))) for k in ("OK", "MISSING", "TYPE_DIFFERS")}
            cur.execute("SELECT COUNT(*) FROM C000025"); res["product_count"] = cur.fetchone()[0]
            ncm_col = "CLASSIFICACAO_FISCAL" if "CLASSIFICACAO_FISCAL" in cols else None
            if ncm_col:
                cur.execute(f"SELECT {ncm_col}, COUNT(*) FROM C000025 GROUP BY {ncm_col}")
                b = {"eight_digits": 0, "blank": 0, "other_format": 0}
                for v, n in cur.fetchall():
                    b[ncm_bucket(v)] += n
                res["ncm_coverage"] = {**b, "share_eight_digits": round(b["eight_digits"] / res["product_count"], 4) if res["product_count"] else None}
            pop = {}
            for f in FISCAL_FIELDS:
                if f in cols:
                    cur.execute(f"SELECT COUNT(*) FROM C000025 WHERE {f} IS NOT NULL AND TRIM(CAST({f} AS VARCHAR(60))) <> ''")
                    pop[f] = cur.fetchone()[0]
            res["fiscal_field_population"] = pop
        return res
    finally:
        conn.close()


def main(argv=None):
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("fdb"); ap.add_argument("--json")
    a = ap.parse_args(argv)
    load_env()
    if not os.environ.get("FB_USER") or not os.environ.get("FB_PASSWORD"):
        raise SystemExit("FB_USER/FB_PASSWORD must be set in the environment (never on the command line)")
    if not Path(a.fdb).is_file():
        raise SystemExit(f"database file not found: {a.fdb}")
    res = audit(a.fdb, os.environ)
    text = json.dumps(res, ensure_ascii=False, indent=1, default=str)
    if a.json:
        Path(a.json).write_text(text, encoding="utf-8")
    print(text)


if __name__ == "__main__":
    main()