gti
d21a9d1140
git-subtree-dir: fiscal git-subtree-mainline:d95165fc8fgit-subtree-split:102ba84500
114 lines
4.0 KiB
Python
114 lines
4.0 KiB
Python
"""Run the MVP's first real, guarded Firebird write (test DB only) and verify it.
|
|
|
|
Loads a single write_eligible IPI proposal from a DRY_RUN_PLAN artifact,
|
|
re-reads the live row to check for drift, performs the guarded UPDATE via
|
|
firebird_write_executor, verifies persistence, and writes an audit log with
|
|
no secrets. Only ever targets TESTESOFICIAIS.FDB (enforced by the executor).
|
|
"""
|
|
import argparse
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def load_executor():
|
|
spec = importlib.util.spec_from_file_location("firebird_write_executor", ROOT / "scripts" / "firebird_write_executor.py")
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def load_dotenv(path: Path) -> None:
|
|
for line in path.read_text(encoding="utf-8").splitlines():
|
|
line = line.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
key, _, value = line.partition("=")
|
|
os.environ[key.strip()] = value.strip()
|
|
|
|
|
|
def pick_proposal(plan: dict, product_code: str | None) -> dict:
|
|
eligible = [p for p in plan["field_proposals"] if p["field"] == "IPI" and p["write_eligible"]]
|
|
if not eligible:
|
|
raise SystemExit("no write_eligible IPI proposal available")
|
|
if product_code:
|
|
chosen = next((p for p in eligible if p["product_code"] == product_code), None)
|
|
if chosen is None:
|
|
raise SystemExit(f"product {product_code} is not write_eligible")
|
|
return chosen
|
|
return eligible[0]
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--plan", default=str(ROOT / "docs" / "DRY_RUN_PLAN_V3.json"))
|
|
parser.add_argument("--product-code", default=None)
|
|
parser.add_argument("--audit-log", default=str(ROOT / "docs" / "runs" / "first_write_audit.json"))
|
|
args = parser.parse_args()
|
|
|
|
load_dotenv(ROOT / ".env")
|
|
executor = load_executor()
|
|
|
|
plan = json.loads(Path(args.plan).read_text(encoding="utf-8"))
|
|
proposal = pick_proposal(plan, args.product_code)
|
|
|
|
import fdb
|
|
database_path = os.environ["FIREBIRD_DATABASE_PATH"]
|
|
executor.assert_authorized_write_path(database_path)
|
|
|
|
connection = fdb.connect(
|
|
database=database_path,
|
|
user=os.environ["FB_USER"],
|
|
password=os.environ["FB_PASSWORD"],
|
|
fb_library_name=os.environ.get("FIREBIRD_CLIENT_LIBRARY"),
|
|
)
|
|
try:
|
|
cursor = connection.cursor()
|
|
cursor.execute(
|
|
"SELECT CODIGO, CLASSIFICACAO_FISCAL, IPI FROM C000025 WHERE CODIGO = ?",
|
|
(proposal["product_code"],),
|
|
)
|
|
row = cursor.fetchone()
|
|
if row is None:
|
|
raise SystemExit(f"product {proposal['product_code']} not found")
|
|
codigo, ncm, live_ipi = row
|
|
if live_ipi != proposal["current_value"]:
|
|
raise SystemExit(
|
|
f"drift before write: plan expected {proposal['current_value']!r}, "
|
|
f"live is {live_ipi!r} — refusing to write, choose another eligible product"
|
|
)
|
|
cursor.close()
|
|
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
record = executor.execute_guarded_update(
|
|
connection,
|
|
database_path=database_path,
|
|
proposal=proposal,
|
|
now=now,
|
|
)
|
|
audit = {
|
|
**record.as_dict(),
|
|
"ncm": ncm,
|
|
"database": str(executor.AUTHORIZED_WRITE_DATABASE_PATH),
|
|
"table": "C000025",
|
|
}
|
|
audit_path = Path(args.audit_log)
|
|
audit_path.parent.mkdir(parents=True, exist_ok=True)
|
|
existing = json.loads(audit_path.read_text(encoding="utf-8")) if audit_path.exists() else []
|
|
existing.append(audit)
|
|
audit_path.write_text(json.dumps(existing, indent=2, ensure_ascii=False), encoding="utf-8")
|
|
|
|
print(json.dumps(audit, indent=2, ensure_ascii=False))
|
|
finally:
|
|
connection.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|