Add 'fiscal/' from commit '102ba84500dbd9655a0471288c3e278652593ca9'
git-subtree-dir: fiscal git-subtree-mainline:d95165fc8fgit-subtree-split:102ba84500
This commit is contained in:
gti
@@ -0,0 +1,129 @@
|
||||
"""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()
|
||||
Reference in New Issue
Block a user