funman300
f0c6dcf238
Promotion prep for the contract-apply cutover, which unlike the quick-sell promotion moves BOTH binaries and applies a schema migration. fifa17-promotion-backup.py uses SQLite's online backup API, not cp. Production runs WAL with a routinely uncheckpointed WAL (515 KB at capture time); copying the main file alone is not atomic against a live writer and carries no guarantee the WAL holds no newer committed state. Emits a checksummed backup, a metadata record and a RESTORE-*.sh that removes the stale -wal/-shm BEFORE restoring -- omit that and SQLite replays the old journal over the file you just put back, resurrecting the state you were abandoning. fifa17-promotion-snapshot.py is read-only (mode=ro) and counts EVERY table rather than a hand-picked list, so a delta cannot hide in a table nobody thought to name. It also fingerprints the ownership rows, catching a row silently rewritten when counts alone would match. fifa17-contract-apply-validate.py gains --host/--db so one tool serves staging, the migration rehearsal and the production acceptance run. Defaults stay staging: there is deliberately no production default, so a bare invocation cannot touch production.
95 lines
4.1 KiB
Python
Executable File
95 lines
4.1 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""READ-ONLY state capture of an OpenFUT Core SQLite DB, for promotion gating.
|
|
|
|
Opened `mode=ro` and never written to, so it is safe against live production.
|
|
Emits a deterministic JSON document: run it before a promotion and again after,
|
|
then `diff` the two. Every table is counted, so a delta cannot hide in a table
|
|
nobody thought to list.
|
|
|
|
Usage:
|
|
fifa17-promotion-snapshot.py <db-path> [out.json]
|
|
"""
|
|
import hashlib
|
|
import json
|
|
import sqlite3
|
|
import sys
|
|
|
|
|
|
def snapshot(db_path):
|
|
con = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
|
|
con.row_factory = sqlite3.Row
|
|
one = lambda s: con.execute(s).fetchone()[0]
|
|
|
|
snap = {
|
|
"db_path": db_path,
|
|
"journal_mode": one("PRAGMA journal_mode"),
|
|
"page_size": one("PRAGMA page_size"),
|
|
"page_count": one("PRAGMA page_count"),
|
|
"sqlite_version": sqlite3.sqlite_version,
|
|
}
|
|
|
|
# Schema version + the full applied-migration ledger. A promotion that
|
|
# claims "0028 applied" must show it here, with success=1.
|
|
snap["migration_max"] = one("SELECT MAX(version) FROM _sqlx_migrations")
|
|
snap["migration_count"] = one("SELECT COUNT(*) FROM _sqlx_migrations")
|
|
snap["migrations_failed"] = one(
|
|
"SELECT COUNT(*) FROM _sqlx_migrations WHERE success <> 1")
|
|
|
|
# Every table, counted. Deliberately not a hand-picked list.
|
|
tables = [r[0] for r in con.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table' "
|
|
"AND name NOT LIKE 'sqlite_%' ORDER BY name")]
|
|
snap["table_counts"] = {t: one(f'SELECT COUNT(*) FROM "{t}"') for t in tables}
|
|
|
|
# The economically load-bearing figures, called out so a diff is readable.
|
|
snap["coins"] = dict(con.execute("SELECT id, coins FROM clubs ORDER BY id").fetchall())
|
|
snap["owned_total"] = one("SELECT COUNT(*) FROM owned_cards")
|
|
snap["owned_by_kind"] = dict(
|
|
con.execute("SELECT content_kind, COUNT(*) FROM owned_cards "
|
|
"GROUP BY 1 ORDER BY 1").fetchall())
|
|
snap["squad_players"] = one("SELECT COUNT(*) FROM squad_players")
|
|
snap["market_listings"] = one("SELECT COUNT(*) FROM market_listings")
|
|
snap["market_listings_sold"] = one("SELECT COUNT(*) FROM market_listings WHERE sold = 1")
|
|
snap["market_listings_npc"] = one("SELECT COUNT(*) FROM market_listings WHERE is_npc = 1")
|
|
snap["game_entity_ext"] = one("SELECT COUNT(*) FROM game_entity_ext")
|
|
snap["consumable_applications"] = one("SELECT COUNT(*) FROM consumable_applications")
|
|
snap["packs_total"] = one("SELECT COUNT(*) FROM packs")
|
|
snap["packs_unopened"] = one("SELECT COUNT(*) FROM packs WHERE opened = 0")
|
|
snap["match_completions"] = one("SELECT COUNT(*) FROM match_completions")
|
|
|
|
# Present only after 0028. Absent => pre-0028 DB, which is itself the signal.
|
|
cols = [r[1] for r in con.execute("PRAGMA table_info(owned_cards)")]
|
|
snap["owned_cards_columns"] = cols
|
|
snap["has_0028"] = "contract_matches" in cols
|
|
if snap["has_0028"]:
|
|
snap["contract_matches_set"] = one(
|
|
"SELECT COUNT(*) FROM owned_cards WHERE contract_matches IS NOT NULL")
|
|
snap["contract_matches_sum"] = one(
|
|
"SELECT COALESCE(SUM(contract_matches), 0) FROM owned_cards")
|
|
|
|
# A content fingerprint over ownership: catches a row silently rewritten
|
|
# even when every count stays identical.
|
|
h = hashlib.sha256()
|
|
for r in con.execute(
|
|
"SELECT id, club_id, card_id, is_loan, loan_matches_remaining, "
|
|
"chemistry_style, position_override, training_bonus, content_kind, "
|
|
"quantity FROM owned_cards ORDER BY id"):
|
|
h.update(("|".join("" if v is None else str(v) for v in r)).encode())
|
|
snap["owned_cards_fingerprint_pre0028_columns"] = h.hexdigest()
|
|
|
|
snap["integrity_check"] = one("PRAGMA integrity_check")
|
|
snap["foreign_key_check"] = len(con.execute("PRAGMA foreign_key_check").fetchall())
|
|
con.close()
|
|
return snap
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) < 2:
|
|
print(__doc__)
|
|
raise SystemExit(2)
|
|
snap = snapshot(sys.argv[1])
|
|
text = json.dumps(snap, indent=2, sort_keys=True)
|
|
print(text)
|
|
if len(sys.argv) > 2:
|
|
open(sys.argv[2], "w").write(text + "\n")
|