dbd517b279
- Update dbis_core, cross-chain-pmm-lps, explorer-monorepo, metamask-integration, pr-workspace/chains - Omit embedded publish git dirs and empty placeholders from index Made-with: Cursor
79 lines
2.5 KiB
Python
Executable File
79 lines
2.5 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Optional: append IT inventory export metadata to SQLite (Phase 1 BFF persistence stub)."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sqlite3
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
|
|
def main() -> None:
|
|
if len(sys.argv) < 4:
|
|
print(
|
|
"usage: persist-it-snapshot-sqlite.py <db_path> <reports_dir> <drift_exit_code>",
|
|
file=sys.stderr,
|
|
)
|
|
sys.exit(2)
|
|
db_path = Path(sys.argv[1])
|
|
reports = Path(sys.argv[2])
|
|
try:
|
|
rc = int(sys.argv[3])
|
|
except ValueError:
|
|
rc = -1
|
|
|
|
drift_path = reports / "drift.json"
|
|
live_path = reports / "live_inventory.json"
|
|
drift = json.loads(drift_path.read_text(encoding="utf-8")) if drift_path.is_file() else {}
|
|
live = json.loads(live_path.read_text(encoding="utf-8")) if live_path.is_file() else {}
|
|
|
|
collected = drift.get("collected_at") or live.get("collected_at")
|
|
if not collected:
|
|
collected = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
|
|
dup = drift.get("duplicate_ips") or {}
|
|
dup_count = len(dup) if isinstance(dup, dict) else 0
|
|
guests = live.get("guests") if isinstance(live.get("guests"), list) else []
|
|
guest_count = len(guests)
|
|
|
|
db_path.parent.mkdir(parents=True, exist_ok=True)
|
|
con = sqlite3.connect(str(db_path))
|
|
try:
|
|
con.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS inventory_export_run (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
collected_at TEXT NOT NULL,
|
|
drift_exit_code INTEGER NOT NULL,
|
|
guest_count INTEGER NOT NULL,
|
|
duplicate_ip_bucket_count INTEGER NOT NULL,
|
|
drift_json TEXT NOT NULL,
|
|
created_at TEXT NOT NULL
|
|
)
|
|
"""
|
|
)
|
|
con.execute(
|
|
"""
|
|
INSERT INTO inventory_export_run
|
|
(collected_at, drift_exit_code, guest_count, duplicate_ip_bucket_count, drift_json, created_at)
|
|
VALUES (?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
collected,
|
|
rc,
|
|
guest_count,
|
|
dup_count,
|
|
drift_path.read_text(encoding="utf-8") if drift_path.is_file() else "{}",
|
|
datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
|
|
),
|
|
)
|
|
con.commit()
|
|
finally:
|
|
con.close()
|
|
print(f"SQLite snapshot row written: {db_path}", file=sys.stderr)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|