gti
d21a9d1140
git-subtree-dir: fiscal git-subtree-mainline:d95165fc8fgit-subtree-split:102ba84500
43 lines
3.3 KiB
Python
43 lines
3.3 KiB
Python
"""Guarded rollback of one field to the BEFORE values recorded in the local-completion write audit.
|
|
usage: python scripts/rollback_field.py FIELD [--db NAME ...] Writes a rollback audit next to the original audit and verifies 100% against the BEFORE snapshot."""
|
|
import argparse, gzip, json, pickle, sys
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
from dotenv import load_dotenv
|
|
import local_completion as lc
|
|
|
|
ap = argparse.ArgumentParser(); ap.add_argument("field"); ap.add_argument("--db", nargs="*"); ap.add_argument("--run-id"); ap.add_argument("--product"); ap.add_argument("--plan", action="store_true", help="only the codes listed in rollback_plan.json"); a = ap.parse_args()
|
|
plan = json.loads((Path(__file__).resolve().parents[1] / "docs" / "runs" / "local_completion" / "rollback_plan.json").read_text()) if a.plan else None
|
|
load_dotenv(lc.ROOT / ".env")
|
|
ex = lc.load_executor(); al = lc.allowlist()
|
|
out = {}
|
|
for name in a.db or list(al):
|
|
path = al[name]; todo = []
|
|
for line in gzip.open(lc.STATE / f"{name}_write_audit.ndjson.gz", "rt", encoding="utf-8"):
|
|
r = json.loads(line)
|
|
if r.get("outcome") == "SUCCEEDED" and (a.run_id is None or r.get("run_id") == a.run_id) and (a.product is None or r.get("product_code") == a.product) and a.field in r["fields"] and (plan is None or r["product_code"] in plan.get(name, {}).get(a.field, [])):
|
|
todo.append((r["product_code"], r["fields"][a.field]["before"], r["fields"][a.field]["after"]))
|
|
conn = lc.connect(path, rw=True); ok = fail = 0; audit = []
|
|
for code, before, after in todo:
|
|
try:
|
|
res = ex.execute_guarded_product_update(conn, database_path=path, product_code=code, changes={a.field: (after, before)}, now=datetime.now(timezone.utc).isoformat(), restore_before=True)
|
|
audit.append({**res, "outcome": "ROLLED_BACK"}); ok += 1
|
|
except Exception as e: # noqa: BLE001
|
|
message = str(e)
|
|
outcome = "CONFLICT" if "expected" in message.lower() or "changed" in message.lower() else "FAILED"
|
|
audit.append({"product_code": code, "outcome": outcome, "error": message}); fail += 1
|
|
conn.close()
|
|
suffix = "_plan" if plan else ""
|
|
with gzip.open(lc.STATE / f"{name}_rollback_{a.field}{suffix}.ndjson.gz", "wt", encoding="utf-8") as f:
|
|
for r in audit: f.write(json.dumps(r, default=str) + "\n")
|
|
# independent verification against the BEFORE snapshot
|
|
snap = pickle.load(gzip.open(lc.STATE / f"{name}_before.pkl.gz", "rb")); cols = snap["columns"]; i = cols.index(a.field)
|
|
c2 = lc.connect(path); cur = c2.cursor(); cur.execute(f"select CODIGO, {a.field} from C000025"); live = {}
|
|
for k, v in cur.fetchall(): live.setdefault(k, []).append(v)
|
|
n = lambda v: None if v is None else str(v).strip() # noqa: E731
|
|
mism = sum(1 for k, r in snap["rows"].items() if (plan is None or k in plan.get(name, {}).get(a.field, [])) and k in live and any(n(v) != n(r[i]) for v in live[k]))
|
|
out[name] = {"rollback_planned": len(todo), "rolled_back": ok, "failed": fail, "column_mismatch_vs_before_snapshot": mism, "rows_snapshot": len(snap["rows"])}
|
|
print(name, out[name], flush=True)
|
|
json.dump(out, open(lc.STATE / f"rollback_{a.field}{suffix}_summary.json", "w"), indent=1)
|