Something went wrong. Try again.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307"""SQLite cache for generated fighters."""
import jsonimport sqlite3from dataclasses import asdictfrom datetime import datetime, timedelta, timezonefrom pathlib import Path
import config
def connect() -> sqlite3.Connection: """Open a SQLite connection with row factories and pragmas.""" Path(config.DB_PATH).parent.mkdir(parents=True, exist_ok=True) conn = sqlite3.connect(config.DB_PATH, detect_types=sqlite3.PARSE_DECLTYPES, timeout=10.0) conn.row_factory = sqlite3.Row conn.execute("PRAGMA foreign_keys = ON") conn.execute("PRAGMA journal_mode = WAL") conn.execute("PRAGMA busy_timeout = 10000") return conn
def init_schema(conn: sqlite3.Connection) -> None: """Create tables if they do not exist.""" conn.execute( """ CREATE TABLE IF NOT EXISTS fighters ( handle TEXT PRIMARY KEY, did TEXT, display_name TEXT, avatar TEXT, type TEXT, stats_json TEXT, moves_json TEXT, summary TEXT, collections_json TEXT, generated_at TIMESTAMP ) """ ) conn.execute( """ CREATE TABLE IF NOT EXISTS sessions ( session_id TEXT PRIMARY KEY, data_json TEXT, created_at TIMESTAMP, expires_at TIMESTAMP ) """ ) conn.execute( """ CREATE TABLE IF NOT EXISTS battle_results ( id INTEGER PRIMARY KEY AUTOINCREMENT, player_handle TEXT NOT NULL, opponent_handle TEXT NOT NULL, player_type TEXT NOT NULL, opponent_type TEXT NOT NULL, won BOOLEAN NOT NULL, turns INTEGER NOT NULL, final_player_hp INTEGER NOT NULL, final_opponent_hp INTEGER NOT NULL, pds_record_uri TEXT, created_at TIMESTAMP NOT NULL ) """ ) conn.commit()
def save_fighter( conn: sqlite3.Connection, handle: str, did: str, display_name: str, avatar: str, fighter_type: str, stats: dict, moves: list, summary: str, collections: list,) -> None: """Persist a fighter analysis, overwriting any existing row.""" now = datetime.now(timezone.utc) conn.execute( """ INSERT INTO fighters (handle, did, display_name, avatar, type, stats_json, moves_json, summary, collections_json, generated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(handle) DO UPDATE SET did = excluded.did, display_name = excluded.display_name, avatar = excluded.avatar, type = excluded.type, stats_json = excluded.stats_json, moves_json = excluded.moves_json, summary = excluded.summary, collections_json = excluded.collections_json, generated_at = excluded.generated_at """, ( handle, did, display_name, avatar, fighter_type, json.dumps(stats), json.dumps(moves), summary, json.dumps(collections), now, ), ) conn.commit()
def get_fighter(conn: sqlite3.Connection, handle: str) -> dict | None: """Return a cached fighter as a dict, or None if not cached.""" row = conn.execute( "SELECT * FROM fighters WHERE handle = ?", (handle,) ).fetchone() if row is None: return None return { "handle": row["handle"], "did": row["did"], "display_name": row["display_name"], "avatar": row["avatar"], "type": row["type"], "stats": json.loads(row["stats_json"]), "moves": json.loads(row["moves_json"]), "summary": row["summary"], "collections": json.loads(row["collections_json"]), "generated_at": row["generated_at"].isoformat(), }
def is_fresh(conn: sqlite3.Connection, handle: str, ttl_hours: int = 24) -> bool: """Return True if a cached fighter exists and is younger than ttl_hours.""" row = conn.execute( "SELECT generated_at FROM fighters WHERE handle = ?", (handle,) ).fetchone() if row is None: return False generated = row["generated_at"] if generated.tzinfo is None: generated = generated.replace(tzinfo=timezone.utc) cutoff = datetime.now(timezone.utc) - timedelta(hours=ttl_hours) return generated > cutoff
def row_count(conn: sqlite3.Connection) -> int: """Return the number of cached fighters.""" row = conn.execute("SELECT COUNT(*) FROM fighters").fetchone() return row[0] if row else 0
def save_session( conn: sqlite3.Connection, session_id: str, data: dict, ttl_seconds: int = 3600,) -> None: """Persist server-side session data.""" now = datetime.now(timezone.utc) expires = now + timedelta(seconds=ttl_seconds) conn.execute( """ INSERT INTO sessions (session_id, data_json, created_at, expires_at) VALUES (?, ?, ?, ?) ON CONFLICT(session_id) DO UPDATE SET data_json = excluded.data_json, expires_at = excluded.expires_at """, (session_id, json.dumps(data), now, expires), ) conn.commit()
def get_session(conn: sqlite3.Connection, session_id: str) -> dict | None: """Return session data if it exists and has not expired.""" row = conn.execute( "SELECT data_json, expires_at FROM sessions WHERE session_id = ?", (session_id,), ).fetchone() if row is None: return None expires = row["expires_at"] if expires.tzinfo is None: expires = expires.replace(tzinfo=timezone.utc) if datetime.now(timezone.utc) > expires: conn.execute("DELETE FROM sessions WHERE session_id = ?", (session_id,)) conn.commit() return None return json.loads(row["data_json"])
def delete_session(conn: sqlite3.Connection, session_id: str) -> None: """Remove a session.""" conn.execute("DELETE FROM sessions WHERE session_id = ?", (session_id,)) conn.commit()
def save_battle_result( conn: sqlite3.Connection, player_handle: str, opponent_handle: str, player_type: str, opponent_type: str, won: bool, turns: int, final_player_hp: int, final_opponent_hp: int, pds_record_uri: str | None = None,) -> int: """Persist an official battle result and return the row id.""" now = datetime.now(timezone.utc) cursor = conn.execute( """ INSERT INTO battle_results (player_handle, opponent_handle, player_type, opponent_type, won, turns, final_player_hp, final_opponent_hp, pds_record_uri, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( player_handle, opponent_handle, player_type, opponent_type, won, turns, final_player_hp, final_opponent_hp, pds_record_uri, now, ), ) conn.commit() return cursor.lastrowid
def get_leaderboard(conn: sqlite3.Connection, limit: int = 20) -> list[dict]: """Return win/loss totals ordered by wins descending.""" rows = conn.execute( """ SELECT player_handle, COUNT(*) FILTER (WHERE won = TRUE) AS wins, COUNT(*) FILTER (WHERE won = FALSE) AS losses, COUNT(*) AS total FROM battle_results GROUP BY player_handle ORDER BY wins DESC, total ASC LIMIT ? """, (limit,), ).fetchall() return [ { "handle": row["player_handle"], "wins": row["wins"], "losses": row["losses"], "total": row["total"], } for row in rows ]
def get_user_history(conn: sqlite3.Connection, handle: str, limit: int = 50) -> list[dict]: """Return recent battle history for a player.""" rows = conn.execute( """ SELECT player_handle, opponent_handle, player_type, opponent_type, won, turns, final_player_hp, final_opponent_hp, created_at FROM battle_results WHERE player_handle = ? ORDER BY created_at DESC LIMIT ? """, (handle, limit), ).fetchall() return [ { "player_handle": row["player_handle"], "opponent_handle": row["opponent_handle"], "player_type": row["player_type"], "opponent_type": row["opponent_type"], "won": bool(row["won"]), "turns": row["turns"], "final_player_hp": row["final_player_hp"], "final_opponent_hp": row["final_opponent_hp"], "created_at": row["created_at"].isoformat(), } for row in rows ]