XFEEDToolBox

"""Read-only diagnosis, with original database/WAL evidence preserved first."""
import argparse
import datetime
import json
import pathlib
import shutil
import sqlite3

parser = argparse.ArgumentParser()
parser.add_argument("source", type=pathlib.Path)
parser.add_argument("output", type=pathlib.Path)
args = parser.parse_args()
args.output.mkdir(parents=True, exist_ok=False)
report = {"utc": datetime.datetime.now(datetime.timezone.utc).isoformat(), "databases": []}
for name in ("game-events.db", "markets.db"):
    source = args.source / name
    item = {"path": str(source), "files": [], "checks": {}}
    report["databases"].append(item)
    for suffix in ("", "-wal", "-shm"):
        src = pathlib.Path(str(source) + suffix)
        if src.exists():
            target = args.output / (name + suffix)
            shutil.copy2(src, target)
            item["files"].append({"name": target.name, "bytes": target.stat().st_size})
    print(json.dumps({"backed_up": name, "files": item["files"]}), flush=True)
    # Hold one SQLite read snapshot while checking, rather than opening the
    # forensic raw copy with a possibly incomplete live WAL copy.
    try:
        with sqlite3.connect(source.as_uri() + "?mode=ro", uri=True, timeout=10) as db:
            db.execute("PRAGMA query_only=ON")
            db.execute("BEGIN")
            for sql in ("PRAGMA quick_check(20)", "PRAGMA journal_mode", "PRAGMA page_size", "PRAGMA page_count", "PRAGMA freelist_count"):
                try:
                    item["checks"][sql] = db.execute(sql).fetchall()
                except sqlite3.DatabaseError as exc:
                    item["checks"][sql] = {"error": str(exc), "code": getattr(exc, "sqlite_errorcode", None)}
            try:
                tables = db.execute("SELECT type,name,tbl_name,rootpage,sql FROM sqlite_schema ORDER BY name").fetchall()
                item["schema"] = tables
                item["table_counts"] = {}
                for kind, table, _, _, _ in tables:
                    if kind != "table" or table.startswith("sqlite_"):
                        continue
                    try:
                        count = db.execute('SELECT COUNT(*) FROM "' + table.replace('"', '""') + '"').fetchone()[0]
                        item["table_counts"][table] = count
                    except sqlite3.DatabaseError as exc:
                        item["table_counts"][table] = {"error": str(exc), "code": getattr(exc, "sqlite_errorcode", None)}
            except sqlite3.DatabaseError as exc:
                item["schema_error"] = str(exc)
    except sqlite3.DatabaseError as exc:
        item["open_error"] = str(exc)
    (args.output / "diagnosis.json").write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
    print(json.dumps({"name": name, "checks": item["checks"], "table_counts": item.get("table_counts"), "error": item.get("open_error")}, ensure_ascii=False), flush=True)