funman300
88b4cad780
fifa17-migration-invariants.py Pre/post invariants across every domain the
migration authorization names: coins, ownership (+kind histogram, distinct
definitions, chemistry styles, loans, position overrides), squads,
managers, staff, consumables, club items, transfer state (market_listings),
packs, SBC, match history, plus integrity_check and foreign_key_check.
Table names are the REAL schema, not guessed: transfer state lives in
market_listings (29 rows in production), the FIFA17 opaque squad blob in
game_entity_ext.
fifa17-migration-rehearse.py Serves a migrated COPY with the candidate Rust
stack on isolated ports and validates the wire surface: club discardValue
is table-derived, squad projects, consumable categories populate, and the
apply probe is OFF (502 upstream-unavailable rather than a diagnostic ack).
Both are read-only against production: the rehearsal operates on a copy under
/home/alex/openfut-migration/, and nothing under openfut-promotion/state is
opened.
Evidence from the 2026-08-22 rehearsal is written up in the Vault runbook
"FIFA17 Rust Production Migration (rehearsed)".
91 lines
3.1 KiB
Python
Executable File
91 lines
3.1 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Pre/post invariants for the FIFA17 Rust production migration rehearsal.
|
|
|
|
Covers every domain the authorization names: coins, ownership, squads, managers,
|
|
staff, consumables, club items, transfer state, packs/unassigned, SBC state,
|
|
match state/history. Read-only; run against a COPY.
|
|
"""
|
|
import json
|
|
import sqlite3
|
|
import sys
|
|
|
|
db = sys.argv[1]
|
|
out = sys.argv[2] if len(sys.argv) > 2 else None
|
|
con = sqlite3.connect("file:%s?mode=ro" % db, uri=True)
|
|
tables = {r[0] for r in con.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table'")}
|
|
|
|
|
|
def one(sql, default=None):
|
|
try:
|
|
r = con.execute(sql).fetchone()
|
|
return r[0] if r else default
|
|
except sqlite3.Error:
|
|
return default
|
|
|
|
|
|
def count(t):
|
|
return one("SELECT COUNT(*) FROM %s" % t) if t in tables else None
|
|
|
|
|
|
def group(t, col):
|
|
if t not in tables:
|
|
return None
|
|
try:
|
|
return dict(con.execute("SELECT %s, COUNT(*) FROM %s GROUP BY 1" % (col, t)))
|
|
except sqlite3.Error:
|
|
return None
|
|
|
|
|
|
inv = {
|
|
"schema_version": one("SELECT MAX(version) FROM _sqlx_migrations"),
|
|
"tables": len(tables),
|
|
# economy
|
|
"coins": one("SELECT coins FROM clubs LIMIT 1"),
|
|
"clubs": count("clubs"),
|
|
"profiles": count("profiles"),
|
|
# ownership
|
|
"owned_cards": count("owned_cards"),
|
|
"owned_by_kind": group("owned_cards", "content_kind"),
|
|
"owned_distinct_cards": one("SELECT COUNT(DISTINCT card_id) FROM owned_cards"),
|
|
"owned_loans": one("SELECT COUNT(*) FROM owned_cards WHERE is_loan=1"),
|
|
"owned_chem_styles": one(
|
|
"SELECT COUNT(*) FROM owned_cards WHERE chemistry_style IS NOT NULL"),
|
|
"owned_pos_overrides": one(
|
|
"SELECT COUNT(*) FROM owned_cards WHERE position_override IS NOT NULL"),
|
|
# squads / managers
|
|
"squads": count("squads"),
|
|
"squad_managers": count("squad_managers"),
|
|
# club items
|
|
"club_active_items": count("club_active_items"),
|
|
"club_kit_assignments": count("club_kit_assignments"),
|
|
# transfer / market / piles
|
|
"market_listings": count("market_listings"),
|
|
"market_listings_by_status": group("market_listings", "status"),
|
|
"market_history": count("market_history"),
|
|
"packs": count("packs"),
|
|
# FIFA17 opaque squad extension (adapter-owned blob)
|
|
"game_entity_ext": count("game_entity_ext"),
|
|
"squad_players": count("squad_players"),
|
|
"seasons": count("seasons"),
|
|
"events": count("events"),
|
|
"notifications": count("notifications"),
|
|
# sbc
|
|
"sbc_submissions": count("sbc_submissions"),
|
|
"sbc_challenge_squads": count("sbc_challenge_squads"),
|
|
# matches / history
|
|
"matches": count("matches"),
|
|
"match_completions": count("match_completions"),
|
|
"season_history": count("season_history"),
|
|
"statistics": count("statistics"),
|
|
"consumable_applications": count("consumable_applications"),
|
|
}
|
|
# integrity
|
|
inv["integrity_check"] = one("PRAGMA integrity_check")
|
|
inv["foreign_key_violations"] = len(list(con.execute("PRAGMA foreign_key_check")))
|
|
con.close()
|
|
|
|
print(json.dumps(inv, indent=2, sort_keys=True))
|
|
if out:
|
|
open(out, "w").write(json.dumps(inv, sort_keys=True, indent=2))
|