OpenFUT/scripts/core-snapshot.py

#!/usr/bin/env python3
"""Authoritative Core state snapshot for FIFA17 FUT loop verification.

Queries openfut-core's authoritative read API directly (NOT the FIFA UI, NOT the
host projection, NOT Python) and emits a machine-readable JSON snapshot with the
integrity fields needed to detect, across any economy operation or a full stack
restart:

  * coin drift            (balance vs club.coins cross-check + absolute value)
  * duplicate ownership   (repeated owned_card_id)
  * missing ownership     (squad references an owned_card_id not in the collection)
  * resurrection          (owned-set hash changes when it must not)
  * entitlement dup/count (unopened pack entitlements)
  * stale squad mappings  (squad player ids absent from ownership)

This is verification/oracle tooling, not production behavior. It reads only.

Usage:
  core-snapshot.py [--core URL] [--game fifa17] [--squad-ns fifa17.squad]
                   [--out FILE] [--label NAME]
Env fallbacks: OPENFUT_CORE_URL, OPENFUT_GAME, OPENFUT_SQUAD_NS.
Exit code 0 always for a successful read; integrity problems are reported IN the
snapshot (`integrity.ok` / `integrity.problems`) so callers can diff/assert.
"""
import argparse
import hashlib
import json
import os
import sys
import urllib.error
import urllib.request
from datetime import datetime, timezone


def fetch(core, game, path):
    req = urllib.request.Request(core + path, headers={"X-OpenFUT-Game": game})
    try:
        with urllib.request.urlopen(req, timeout=15) as f:
            return f.status, f.read()
    except urllib.error.HTTPError as e:
        return e.code, e.read()
    except Exception as e:  # noqa: BLE001 - surface transport errors in the snapshot
        return None, str(e).encode()


def fetch_json(core, game, path):
    st, body = fetch(core, game, path)
    if st != 200:
        raise SystemExit(f"Core read {path} failed: status={st} body={body[:200]!r}")
    return json.loads(body)


def sha(items):
    h = hashlib.sha256()
    for it in items:
        h.update(str(it).encode())
        h.update(b"\x00")
    return h.hexdigest()


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--core", default=os.environ.get("OPENFUT_CORE_URL", "http://127.0.0.1:18101"))
    ap.add_argument("--game", default=os.environ.get("OPENFUT_GAME", "fifa17"))
    ap.add_argument("--squad-ns", default=os.environ.get("OPENFUT_SQUAD_NS", "fifa17.squad"))
    ap.add_argument("--out")
    ap.add_argument("--label", default="")
    args = ap.parse_args()

    balance = fetch_json(args.core, args.game, "/economy/balance")
    entitlements = fetch_json(args.core, args.game, "/economy/entitlements")
    profile = fetch_json(args.core, args.game, "/profile")
    club = fetch_json(args.core, args.game, "/club")
    collection = fetch_json(args.core, args.game, "/collection")["collection"]
    squad = fetch_json(args.core, args.game, f"/squad/ext?namespace={args.squad_ns}")

    coins_balance = balance.get("balance")
    coins_club = club.get("coins")

    owned_ids = sorted(x["owned_card_id"] for x in collection)
    card_ids = sorted(x["card"]["id"] for x in collection)
    dup_owned = sorted({i for i in owned_ids if owned_ids.count(i) > 1}) if len(owned_ids) != len(set(owned_ids)) else []
    loans = [x["owned_card_id"] for x in collection if x.get("is_loan")]

    # Squad player ownership references (from the opaque extension payload).
    squad_player_ids = []
    ext = squad.get("extension") or {}
    payload_raw = ext.get("payload")
    kit_numbers = {}
    if payload_raw:
        try:
            payload = json.loads(payload_raw)
            kit_numbers = payload.get("kit_numbers", {}) or {}
            squad_player_ids = sorted(kit_numbers.keys())
        except (json.JSONDecodeError, TypeError):
            pass

    owned_set = set(owned_ids)
    squad_missing = sorted(pid for pid in squad_player_ids if pid not in owned_set)

    problems = []
    if coins_balance != coins_club:
        problems.append(f"coin cross-check mismatch: balance={coins_balance} club={coins_club}")
    if dup_owned:
        problems.append(f"duplicate owned_card_id: {dup_owned[:10]} (+{max(0,len(dup_owned)-10)} more)")
    if squad_missing:
        problems.append(f"squad references non-owned ids: {squad_missing}")

    snap = {
        "label": args.label,
        "captured_at": datetime.now(timezone.utc).isoformat(),
        "core": args.core,
        "game": args.game,
        "coins": {"balance": coins_balance, "club": coins_club},
        "profile": {"username": profile.get("username"), "level": profile.get("level"), "xp": profile.get("xp")},
        "entitlements": {"count": len(entitlements), "ids": entitlements},
        "collection": {
            "owned": len(collection),
            "distinct_owned_card_id": len(set(owned_ids)),
            "distinct_card_id": len(set(card_ids)),
            "loans": len(loans),
            "duplicate_owned_card_id": dup_owned,
            "owned_set_sha256": sha(owned_ids),
            "card_multiset_sha256": sha(card_ids),
        },
        "squad": {
            "namespace": args.squad_ns,
            "verdict": squad.get("verdict"),
            "player_count": len(squad_player_ids),
            "player_ids": squad_player_ids,
            "missing_from_ownership": squad_missing,
            "ext_schema_version": ext.get("schema_version"),
        },
        "integrity": {"ok": not problems, "problems": problems},
    }

    text = json.dumps(snap, indent=2, sort_keys=True)
    if args.out:
        with open(args.out, "w") as f:
            f.write(text + "\n")
    print(text)
    return 0


if __name__ == "__main__":
    sys.exit(main())