148 lines
8.6 KiB
Python
148 lines
8.6 KiB
Python
"""Single installation preflight for the GTI Fiscal Agent. READ-ONLY: never writes to Firebird, never prints secrets.
|
|
|
|
PREFLIGHT=PASS | FAIL (+ blockers, one line each)
|
|
|
|
Checks: config, FISCAL_MODE/FIREBIRD_WRITES, Hub reachable/auth/version, Firebird reachable + version + ODS, C000004/C000025,
|
|
capabilities (minimum: NCM, IPI, CEST), company context (UF/CRT, deterministic resolver), schema/type guards, SQLite state, log dir.
|
|
"""
|
|
import os
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
AGENT_VERSION_DEFAULT = "1.1.7"
|
|
REQUIRED_ENV = ("GTI_HUB_URL", "AGENT_TOKEN", "FIREBIRD_DATABASE_PATH", "FB_USER", "FB_PASSWORD")
|
|
REQUIRED_CAPABILITIES = ("supports_ncm", "supports_ipi", "supports_cest")
|
|
FISCAL_MODES = ("DRY_RUN", "AUTO_APPLY")
|
|
|
|
|
|
def _check(name, ok, detail="", warn=False):
|
|
return {"name": name, "status": "PASS" if ok else ("WARN" if warn else "FAIL"), "detail": detail}
|
|
|
|
|
|
def major_minor(version):
|
|
parts = str(version or "").replace("-", ".").replace("rc", ".").split(".")
|
|
return ".".join(parts[:2])
|
|
|
|
|
|
def default_http_get(url, token=None, timeout=10):
|
|
import httpx
|
|
headers = {"Authorization": f"Bearer {token}"} if token else {}
|
|
r = httpx.get(url, headers=headers, timeout=timeout)
|
|
return r.status_code, (r.json() if r.headers.get("content-type", "").startswith("application/json") else {})
|
|
|
|
|
|
def firebird_probe(env):
|
|
"""Read-only introspection of the configured database."""
|
|
import fdb
|
|
from .company_context import collect_evidence, config_from_env, resolve
|
|
conn = fdb.connect(database=env["FIREBIRD_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()
|
|
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)
|
|
FROM RDB$RELATION_FIELDS rf JOIN RDB$FIELDS f ON f.RDB$FIELD_NAME = rf.RDB$FIELD_SOURCE WHERE rf.RDB$RELATION_NAME IN ('C000004', 'C000025')""")
|
|
tables = {}
|
|
for t, c, ty, ln in cur.fetchall():
|
|
tables.setdefault(t, {})[c] = (ty, ln)
|
|
products = None
|
|
if "C000025" in tables:
|
|
cur.execute("SELECT COUNT(*) FROM C000025")
|
|
products = cur.fetchone()[0]
|
|
company = resolve(collect_evidence(cur), config_from_env(env))
|
|
return {"server_version": getattr(conn, "server_version", None), "ods": f"{getattr(conn, 'ods_version', '?')}.{getattr(conn, 'ods_minor_version', '?')}",
|
|
"tables": tables, "products": products, "company": company}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def run_preflight(env=None, *, http_get=default_http_get, probe=firebird_probe, agent_version=None):
|
|
from .capabilities import detect_capabilities, type_ok
|
|
from .company_context import preflight_report
|
|
env = env if env is not None else os.environ
|
|
checks, info = [], {}
|
|
agent_version = agent_version or env.get("AGENT_VERSION") or AGENT_VERSION_DEFAULT
|
|
|
|
missing = [k for k in REQUIRED_ENV if not (env.get(k) or "").strip()]
|
|
checks.append(_check("CONFIG", not missing, f"missing: {', '.join(missing)}" if missing else "required variables present"))
|
|
|
|
mode = (env.get("FISCAL_MODE") or "").strip().upper()
|
|
writes = (env.get("FIREBIRD_WRITES") or "0").strip() == "1"
|
|
checks.append(_check("FISCAL_MODE", mode in FISCAL_MODES, f"FISCAL_MODE={mode or 'unset'} (must be one of {FISCAL_MODES})"))
|
|
dry_conflict = mode == "DRY_RUN" and writes
|
|
checks.append(_check("FIREBIRD_WRITES", not dry_conflict, "FIREBIRD_WRITES=1 conflicts with FISCAL_MODE=DRY_RUN (writes stay disabled; fix the config)" if dry_conflict else f"requested={int(writes)}"))
|
|
effective = writes and mode == "AUTO_APPLY"
|
|
info["effective_firebird_writes"] = effective
|
|
if effective:
|
|
ok = bool((env.get("AUTHORIZED_WRITE_DATABASE_PATH") or "").strip() and (env.get("EXPECTED_SCHEMA_FINGERPRINT") or "").strip())
|
|
checks.append(_check("WRITE_AUTHORIZATION", ok, "AUTO_APPLY needs AUTHORIZED_WRITE_DATABASE_PATH and EXPECTED_SCHEMA_FINGERPRINT (two independent layers)"))
|
|
|
|
hub = (env.get("GTI_HUB_URL") or "").rstrip("/")
|
|
hub_ok = False
|
|
if hub:
|
|
try:
|
|
code, _ = http_get(f"{hub}/health")
|
|
hub_ok = code == 200
|
|
checks.append(_check("HUB_REACHABLE", hub_ok, f"GET /health -> {code}"))
|
|
except Exception as e: # noqa: BLE001
|
|
checks.append(_check("HUB_REACHABLE", False, type(e).__name__))
|
|
if hub_ok:
|
|
try:
|
|
code, body = http_get(f"{hub}/version")
|
|
hub_v = body.get("version")
|
|
info["hub_version"] = hub_v
|
|
checks.append(_check("HUB_VERSION_COMPATIBLE", code == 200 and major_minor(hub_v) == major_minor(agent_version), f"hub={hub_v} agent={agent_version} (major.minor must match)"))
|
|
except Exception as e: # noqa: BLE001
|
|
checks.append(_check("HUB_VERSION_COMPATIBLE", False, type(e).__name__))
|
|
try:
|
|
code, _ = http_get(f"{hub}/v1/datasets/status", token=env.get("AGENT_TOKEN"))
|
|
checks.append(_check("AUTH", code == 200, f"agent token -> HTTP {code}"))
|
|
except Exception as e: # noqa: BLE001
|
|
checks.append(_check("AUTH", False, type(e).__name__))
|
|
|
|
fb = None
|
|
if all((env.get(k) or "").strip() for k in ("FIREBIRD_DATABASE_PATH", "FB_USER", "FB_PASSWORD")):
|
|
try:
|
|
fb = probe(env)
|
|
checks.append(_check("FIREBIRD_REACHABLE", True, "read-only connection opened"))
|
|
info.update(firebird_version=fb.get("server_version"), ods=fb.get("ods"), products=fb.get("products"))
|
|
checks.append(_check("FIREBIRD_VERSION_ODS", bool(fb.get("server_version")), f"{fb.get('server_version')} / ODS {fb.get('ods')}"))
|
|
except Exception as e: # noqa: BLE001
|
|
checks.append(_check("FIREBIRD_REACHABLE", False, f"{type(e).__name__}: {str(e)[:80]}"))
|
|
if fb:
|
|
t = fb["tables"]
|
|
checks.append(_check("C000004", "C000004" in t, "company table present" if "C000004" in t else "missing"))
|
|
checks.append(_check("C000025", "C000025" in t, "product table present" if "C000025" in t else "missing"))
|
|
cols = {c: ty for c, (ty, _) in t.get("C000025", {}).items()}
|
|
caps = detect_capabilities(cols)
|
|
info["capabilities"] = caps
|
|
lacking = [c for c in REQUIRED_CAPABILITIES if not caps.get(c)]
|
|
checks.append(_check("CAPABILITIES", not lacking, f"missing required: {', '.join(lacking)}" if lacking else f"required present; optional true: {sum(1 for c, v in caps.items() if v and c not in REQUIRED_CAPABILITIES)}"))
|
|
rep = preflight_report(fb["company"])
|
|
info["company_context"] = rep
|
|
checks.append(_check("COMPANY_CONTEXT", rep["COMPANY_CONTEXT"] == "PASS", "; ".join(rep.get("reasons") or []) or f"{rep['classification']} source={rep['source_of_truth']}"))
|
|
checks.append(_check("UF", bool(rep.get("company_uf")), str(rep.get("company_uf"))))
|
|
checks.append(_check("CRT", bool(rep.get("company_crt")), str(rep.get("company_crt"))))
|
|
writable = [f for f in ("IPI", "CEST", "EX_TIPI", "CLASSIFICACAO_FISCAL") if f in t.get("C000025", {})]
|
|
bad = [f for f in writable if not type_ok(f, t["C000025"][f][0])]
|
|
lens = {f: t["C000025"][f][1] for f in ("CEST", "EX_TIPI") if f in t.get("C000025", {})}
|
|
checks.append(_check("SCHEMA_TYPE_GUARDS", not bad and len(writable) >= 3, f"incompatible: {bad}" if bad else f"types ok for {writable}; text lengths {lens} (the executor re-checks length/type before any write)"))
|
|
|
|
state_db = env.get("AGENT_STATE_DB") or str(Path.cwd() / "agent_state.sqlite")
|
|
try:
|
|
Path(state_db).parent.mkdir(parents=True, exist_ok=True)
|
|
c = sqlite3.connect(state_db)
|
|
c.execute("SELECT 1")
|
|
c.close()
|
|
checks.append(_check("SQLITE", True, "state database opens"))
|
|
except Exception as e: # noqa: BLE001
|
|
checks.append(_check("SQLITE", False, type(e).__name__))
|
|
log_path = Path(env.get("AGENT_LOG_PATH") or Path.cwd() / "logs" / "agent.log")
|
|
try:
|
|
log_path.parent.mkdir(parents=True, exist_ok=True)
|
|
checks.append(_check("LOG_DIRECTORY", os.access(log_path.parent, os.W_OK), "log directory writable"))
|
|
except Exception as e: # noqa: BLE001
|
|
checks.append(_check("LOG_DIRECTORY", False, type(e).__name__))
|
|
|
|
blockers = [f"{c['name']}: {c['detail']}" for c in checks if c["status"] == "FAIL"]
|
|
return {"PREFLIGHT": "FAIL" if blockers else "PASS", "blockers": blockers, "checks": checks, "agent_version": agent_version, "fiscal_mode": mode or None, **info}
|