"""Separate storage and run tracking for prospective RCON historical capture.""" from __future__ import annotations import json import sqlite3 from collections.abc import Mapping from collections.abc import Iterator from contextlib import closing, contextmanager from datetime import datetime, timezone from pathlib import Path from .config import get_storage_path, use_postgres_rcon_storage from .normalizers import normalize_map_name from .rcon_client import load_rcon_targets from .sqlite_utils import connect_sqlite_readonly, connect_sqlite_writer COMPETITIVE_WINDOW_GAP_SECONDS = 1800 COMPETITIVE_MODE_PARTIAL = "partial" COMPETITIVE_MODE_APPROXIMATE = "approximate" COMPETITIVE_MODE_EXACT = "exact" def initialize_rcon_historical_storage( *, db_path: Path | None = None, ensure_storage: bool = True, ) -> Path: """Create the SQLite structures used by prospective RCON capture.""" if use_postgres_rcon_storage(explicit_sqlite_path=db_path): if ensure_storage: from .postgres_rcon_storage import initialize_postgres_rcon_storage initialize_postgres_rcon_storage() return get_storage_path() resolved_path = db_path or get_storage_path() resolved_path.parent.mkdir(parents=True, exist_ok=True) with _connect(resolved_path) as connection: connection.executescript( """ CREATE TABLE IF NOT EXISTS rcon_historical_targets ( id INTEGER PRIMARY KEY AUTOINCREMENT, target_key TEXT NOT NULL UNIQUE, external_server_id TEXT, display_name TEXT NOT NULL, host TEXT NOT NULL, port INTEGER NOT NULL, region TEXT, game_port INTEGER, query_port INTEGER, source_name TEXT NOT NULL, last_configured_at TEXT NOT NULL, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS rcon_historical_capture_runs ( id INTEGER PRIMARY KEY AUTOINCREMENT, mode TEXT NOT NULL, status TEXT NOT NULL, target_scope TEXT, started_at TEXT NOT NULL, completed_at TEXT, targets_seen INTEGER NOT NULL DEFAULT 0, samples_inserted INTEGER NOT NULL DEFAULT 0, duplicate_samples INTEGER NOT NULL DEFAULT 0, failed_targets INTEGER NOT NULL DEFAULT 0, notes TEXT, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS rcon_historical_samples ( id INTEGER PRIMARY KEY AUTOINCREMENT, target_id INTEGER NOT NULL, capture_run_id INTEGER, captured_at TEXT NOT NULL, source_kind TEXT NOT NULL, status TEXT NOT NULL, players INTEGER, max_players INTEGER, current_map TEXT, normalized_payload_json TEXT NOT NULL, raw_payload_json TEXT, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, UNIQUE(target_id, captured_at), FOREIGN KEY (target_id) REFERENCES rcon_historical_targets(id), FOREIGN KEY (capture_run_id) REFERENCES rcon_historical_capture_runs(id) ); CREATE TABLE IF NOT EXISTS rcon_historical_checkpoints ( target_id INTEGER PRIMARY KEY, last_successful_capture_at TEXT, last_sample_at TEXT, last_run_id INTEGER, last_run_status TEXT, last_error TEXT, last_error_at TEXT, updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (target_id) REFERENCES rcon_historical_targets(id), FOREIGN KEY (last_run_id) REFERENCES rcon_historical_capture_runs(id) ); CREATE INDEX IF NOT EXISTS idx_rcon_historical_samples_target_time ON rcon_historical_samples(target_id, captured_at DESC); CREATE TABLE IF NOT EXISTS rcon_historical_competitive_windows ( id INTEGER PRIMARY KEY AUTOINCREMENT, target_id INTEGER NOT NULL, session_key TEXT NOT NULL UNIQUE, source_kind TEXT NOT NULL, map_name TEXT, map_pretty_name TEXT, first_seen_at TEXT NOT NULL, last_seen_at TEXT NOT NULL, sample_count INTEGER NOT NULL DEFAULT 0, total_players INTEGER NOT NULL DEFAULT 0, peak_players INTEGER NOT NULL DEFAULT 0, last_players INTEGER, max_players INTEGER, status TEXT NOT NULL, confidence_mode TEXT NOT NULL, capabilities_json TEXT NOT NULL, latest_payload_json TEXT, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (target_id) REFERENCES rcon_historical_targets(id) ); CREATE INDEX IF NOT EXISTS idx_rcon_historical_windows_target_time ON rcon_historical_competitive_windows(target_id, last_seen_at DESC); """ ) return resolved_path def start_rcon_historical_capture_run( *, mode: str, target_scope: str, db_path: Path | None = None, ) -> int: """Create one run row for prospective RCON capture.""" if use_postgres_rcon_storage(explicit_sqlite_path=db_path): from .postgres_rcon_storage import start_capture_run return start_capture_run(mode=mode, target_scope=target_scope) resolved_path = initialize_rcon_historical_storage(db_path=db_path) with _connect(resolved_path) as connection: cursor = connection.execute( """ INSERT INTO rcon_historical_capture_runs ( mode, status, target_scope, started_at ) VALUES (?, 'running', ?, ?) """, (mode, target_scope, _utc_now_iso()), ) return int(cursor.lastrowid) def finalize_rcon_historical_capture_run( run_id: int, *, status: str, targets_seen: int, samples_inserted: int, duplicate_samples: int, failed_targets: int, notes: str | None = None, db_path: Path | None = None, ) -> None: """Finalize one prospective RCON capture run.""" if use_postgres_rcon_storage(explicit_sqlite_path=db_path): from .postgres_rcon_storage import finalize_capture_run finalize_capture_run( run_id, status=status, targets_seen=targets_seen, samples_inserted=samples_inserted, duplicate_samples=duplicate_samples, failed_targets=failed_targets, notes=notes, ) return resolved_path = initialize_rcon_historical_storage(db_path=db_path) with _connect(resolved_path) as connection: connection.execute( """ UPDATE rcon_historical_capture_runs SET status = ?, completed_at = ?, targets_seen = ?, samples_inserted = ?, duplicate_samples = ?, failed_targets = ?, notes = ? WHERE id = ? """, ( status, _utc_now_iso(), targets_seen, samples_inserted, duplicate_samples, failed_targets, notes, run_id, ), ) def persist_rcon_historical_sample( *, run_id: int, captured_at: str, target: Mapping[str, object], normalized_payload: Mapping[str, object], raw_payload: Mapping[str, object] | None, db_path: Path | None = None, ) -> dict[str, int]: """Persist one prospective RCON sample and refresh its checkpoint.""" if use_postgres_rcon_storage(explicit_sqlite_path=db_path): from .postgres_rcon_storage import persist_sample return persist_sample( run_id=run_id, captured_at=captured_at, target=target, normalized_payload=normalized_payload, raw_payload=raw_payload, ) resolved_path = initialize_rcon_historical_storage(db_path=db_path) with _connect(resolved_path) as connection: target_id = _upsert_target(connection, target=target) cursor = connection.execute( """ INSERT OR IGNORE INTO rcon_historical_samples ( target_id, capture_run_id, captured_at, source_kind, status, players, max_players, current_map, normalized_payload_json, raw_payload_json ) VALUES (?, ?, ?, 'rcon-live-sample', ?, ?, ?, ?, ?, ?) """, ( target_id, run_id, captured_at, normalized_payload.get("status") or "unknown", normalized_payload.get("players"), normalized_payload.get("max_players"), normalized_payload.get("current_map"), json.dumps(dict(normalized_payload), separators=(",", ":")), json.dumps(dict(raw_payload), separators=(",", ":")) if raw_payload else None, ), ) inserted = int(cursor.rowcount or 0) _upsert_checkpoint_success( connection, target_id=target_id, run_id=run_id, captured_at=captured_at, ) if inserted: _upsert_competitive_window( connection, target_id=target_id, captured_at=captured_at, normalized_payload=normalized_payload, ) return { "samples_inserted": inserted, "duplicate_samples": 0 if inserted else 1, } def mark_rcon_historical_capture_failure( *, run_id: int, target: Mapping[str, object], error_message: str, db_path: Path | None = None, ) -> None: """Persist failure metadata for one target inside a capture run.""" if use_postgres_rcon_storage(explicit_sqlite_path=db_path): from .postgres_rcon_storage import mark_capture_failure mark_capture_failure(run_id=run_id, target=target, error_message=error_message) return resolved_path = initialize_rcon_historical_storage(db_path=db_path) with _connect(resolved_path) as connection: target_id = _upsert_target(connection, target=target) connection.execute( """ INSERT INTO rcon_historical_checkpoints ( target_id, last_run_id, last_run_status, last_error, last_error_at ) VALUES (?, ?, 'failed', ?, ?) ON CONFLICT(target_id) DO UPDATE SET last_run_id = excluded.last_run_id, last_run_status = excluded.last_run_status, last_error = excluded.last_error, last_error_at = excluded.last_error_at, updated_at = CURRENT_TIMESTAMP """, (target_id, run_id, error_message, _utc_now_iso()), ) def list_rcon_historical_target_statuses( *, db_path: Path | None = None, ) -> list[dict[str, object]]: """Return per-target coverage and freshness for prospective RCON capture.""" if use_postgres_rcon_storage(explicit_sqlite_path=db_path): from .postgres_rcon_storage import list_target_statuses return list_target_statuses() resolved_path = _resolve_db_path(db_path) try: with _connect_readonly(resolved_path) as connection: rows = connection.execute( """ SELECT targets.target_key, targets.external_server_id, targets.display_name, targets.host, targets.port, targets.region, targets.source_name, checkpoints.last_successful_capture_at, checkpoints.last_sample_at, checkpoints.last_run_id, checkpoints.last_run_status, checkpoints.last_error, checkpoints.last_error_at, ( SELECT MIN(samples.captured_at) FROM rcon_historical_samples AS samples WHERE samples.target_id = targets.id ) AS first_sample_at, ( SELECT MAX(samples.captured_at) FROM rcon_historical_samples AS samples WHERE samples.target_id = targets.id ) AS latest_sample_at, ( SELECT COUNT(*) FROM rcon_historical_samples AS samples WHERE samples.target_id = targets.id ) AS sample_count FROM rcon_historical_targets AS targets LEFT JOIN rcon_historical_checkpoints AS checkpoints ON checkpoints.target_id = targets.id ORDER BY targets.display_name ASC, targets.target_key ASC """ ).fetchall() except sqlite3.OperationalError: return [] return [ { "target_key": row["target_key"], "external_server_id": row["external_server_id"], "display_name": row["display_name"], "host": row["host"], "port": row["port"], "region": row["region"], "source_name": row["source_name"], "sample_count": int(row["sample_count"] or 0), "first_sample_at": row["first_sample_at"], "last_successful_capture_at": row["last_successful_capture_at"], "last_sample_at": row["latest_sample_at"] or row["last_sample_at"], "last_run_id": row["last_run_id"], "last_run_status": row["last_run_status"], "last_error": row["last_error"], "last_error_at": row["last_error_at"], } for row in rows ] def list_recent_rcon_historical_samples( *, target_key: str | None = None, limit: int = 20, db_path: Path | None = None, ) -> list[dict[str, object]]: """Return recent prospective RCON samples for one or all configured targets.""" if use_postgres_rcon_storage(explicit_sqlite_path=db_path): from .postgres_rcon_storage import list_recent_samples return list_recent_samples(target_key=target_key, limit=limit) resolved_path = _resolve_db_path(db_path) where_clause = "" params: list[object] = [limit] if target_key: aliases = _expand_target_key_aliases(target_key) alias_placeholders = ", ".join("?" for _ in aliases) where_clause = ( "WHERE targets.target_key IN " f"({alias_placeholders}) OR targets.external_server_id IN ({alias_placeholders})" ) params = [*aliases, *aliases, limit] try: with _connect_readonly(resolved_path) as connection: rows = connection.execute( f""" SELECT targets.target_key, targets.external_server_id, targets.display_name, targets.region, samples.captured_at, samples.status, samples.players, samples.max_players, samples.current_map FROM rcon_historical_samples AS samples INNER JOIN rcon_historical_targets AS targets ON targets.id = samples.target_id {where_clause} ORDER BY samples.captured_at DESC, targets.display_name ASC LIMIT ? """, params, ).fetchall() except sqlite3.OperationalError: return [] return [ { "target_key": row["target_key"], "external_server_id": row["external_server_id"], "display_name": row["display_name"], "region": row["region"], "captured_at": row["captured_at"], "status": row["status"], "players": row["players"], "max_players": row["max_players"], "current_map": row["current_map"], } for row in rows ] def list_rcon_historical_competitive_windows( *, target_key: str | None = None, limit: int = 20, db_path: Path | None = None, ) -> list[dict[str, object]]: """Return recent RCON-backed competitive windows derived from persisted samples.""" if use_postgres_rcon_storage(explicit_sqlite_path=db_path): from .postgres_rcon_storage import list_competitive_windows return list_competitive_windows(target_key=target_key, limit=limit) resolved_path = _resolve_db_path(db_path) where_clause = "" params: list[object] = [limit] if target_key: aliases = _expand_target_key_aliases(target_key) alias_placeholders = ", ".join("?" for _ in aliases) where_clause = ( "WHERE targets.target_key IN " f"({alias_placeholders}) OR targets.external_server_id IN ({alias_placeholders})" ) params = [*aliases, *aliases, limit] try: with _connect_readonly(resolved_path) as connection: rows = connection.execute( f""" SELECT targets.target_key, targets.external_server_id, targets.display_name, targets.region, windows.session_key, windows.map_name, windows.map_pretty_name, windows.first_seen_at, windows.last_seen_at, windows.sample_count, windows.total_players, windows.peak_players, windows.last_players, windows.max_players, windows.status, windows.confidence_mode, windows.capabilities_json, windows.latest_payload_json FROM rcon_historical_competitive_windows AS windows INNER JOIN rcon_historical_targets AS targets ON targets.id = windows.target_id {where_clause} ORDER BY windows.last_seen_at DESC, targets.display_name ASC LIMIT ? """, params, ).fetchall() except sqlite3.OperationalError: return [] items: list[dict[str, object]] = [] for row in rows: sample_count = int(row["sample_count"] or 0) average_players = round((int(row["total_players"] or 0) / sample_count), 2) if sample_count > 0 else 0.0 items.append( { "target_key": row["target_key"], "external_server_id": row["external_server_id"], "display_name": row["display_name"], "region": row["region"], "session_key": row["session_key"], "map_name": row["map_name"], "map_pretty_name": row["map_pretty_name"] or row["map_name"], "first_seen_at": row["first_seen_at"], "last_seen_at": row["last_seen_at"], "duration_seconds": _calculate_duration_seconds( row["first_seen_at"], row["last_seen_at"], ), "sample_count": sample_count, "average_players": average_players, "peak_players": int(row["peak_players"] or 0), "last_players": row["last_players"], "max_players": row["max_players"], "status": row["status"], "confidence_mode": row["confidence_mode"], "capabilities": _deserialize_json_object(row["capabilities_json"]), "latest_payload": _deserialize_json_object(row["latest_payload_json"]), } ) return items def count_rcon_historical_samples_since( since: str | None, *, db_path: Path | None = None, ) -> int: """Return how many RCON samples were captured after one timestamp.""" if use_postgres_rcon_storage(explicit_sqlite_path=db_path): from .postgres_rcon_storage import count_samples_since return count_samples_since(since) if not since: return 0 resolved_path = _resolve_db_path(db_path) try: with _connect_readonly(resolved_path) as connection: row = connection.execute( """ SELECT COUNT(*) AS sample_count FROM rcon_historical_samples WHERE captured_at > ? """, (since,), ).fetchone() except sqlite3.OperationalError: return 0 return int(row["sample_count"] or 0) if row else 0 def list_rcon_historical_competitive_summary_rows( *, target_key: str | None = None, db_path: Path | None = None, ) -> list[dict[str, object]]: """Return RCON-backed per-target summary rows over competitive windows.""" if use_postgres_rcon_storage(explicit_sqlite_path=db_path): from .postgres_rcon_storage import list_competitive_summary_rows return list_competitive_summary_rows(target_key=target_key) resolved_path = _resolve_db_path(db_path) where_clause = "" params: list[object] = [] if target_key: aliases = _expand_target_key_aliases(target_key) alias_placeholders = ", ".join("?" for _ in aliases) where_clause = ( "WHERE targets.target_key IN " f"({alias_placeholders}) OR targets.external_server_id IN ({alias_placeholders})" ) params = [*aliases, *aliases] try: with _connect_readonly(resolved_path) as connection: rows = connection.execute( f""" SELECT targets.target_key, targets.external_server_id, targets.display_name, targets.region, checkpoints.last_successful_capture_at, checkpoints.last_run_status, checkpoints.last_error, checkpoints.last_error_at, COUNT(windows.id) AS window_count, COALESCE(SUM(windows.sample_count), 0) AS sample_count, MIN(windows.first_seen_at) AS first_seen_at, MAX(windows.last_seen_at) AS last_seen_at, COALESCE(MAX(windows.peak_players), 0) AS peak_players FROM rcon_historical_targets AS targets LEFT JOIN rcon_historical_checkpoints AS checkpoints ON checkpoints.target_id = targets.id LEFT JOIN rcon_historical_competitive_windows AS windows ON windows.target_id = targets.id {where_clause} GROUP BY targets.id ORDER BY targets.display_name ASC, targets.target_key ASC """, params, ).fetchall() except sqlite3.OperationalError: return [] return [ { "target_key": row["target_key"], "external_server_id": row["external_server_id"], "display_name": row["display_name"], "region": row["region"], "window_count": int(row["window_count"] or 0), "sample_count": int(row["sample_count"] or 0), "first_seen_at": row["first_seen_at"], "last_seen_at": row["last_seen_at"], "peak_players": int(row["peak_players"] or 0), "last_successful_capture_at": row["last_successful_capture_at"], "last_run_status": row["last_run_status"], "last_error": row["last_error"], "last_error_at": row["last_error_at"], } for row in rows ] def find_rcon_historical_competitive_window( *, server_key: str, ended_at: str | None, map_name: str | None = None, db_path: Path | None = None, ) -> dict[str, object] | None: """Return the closest competitive window for one server/match if coverage exists.""" if use_postgres_rcon_storage(explicit_sqlite_path=db_path): from .postgres_rcon_storage import find_competitive_window return find_competitive_window( server_key=server_key, ended_at=ended_at, map_name=map_name, ) if not ended_at: return None resolved_path = _resolve_db_path(db_path) normalized_map_name = normalize_map_name(map_name) aliases = _expand_target_key_aliases(server_key) alias_placeholders = ", ".join("?" for _ in aliases) try: with _connect_readonly(resolved_path) as connection: candidates = connection.execute( f""" SELECT windows.session_key, windows.first_seen_at, windows.last_seen_at, windows.map_name, windows.map_pretty_name, windows.sample_count, windows.total_players, windows.peak_players, windows.confidence_mode, windows.capabilities_json, windows.latest_payload_json FROM rcon_historical_competitive_windows AS windows INNER JOIN rcon_historical_targets AS targets ON targets.id = windows.target_id WHERE ( targets.target_key IN ({alias_placeholders}) OR targets.external_server_id IN ({alias_placeholders}) ) ORDER BY windows.last_seen_at DESC LIMIT 12 """, [*aliases, *aliases], ).fetchall() except sqlite3.OperationalError: return None if not candidates: return None ended_point = _parse_timestamp(ended_at) best_row: sqlite3.Row | None = None best_distance: float | None = None for row in candidates: row_map_name = normalize_map_name(row["map_pretty_name"] or row["map_name"]) if normalized_map_name and row_map_name and normalized_map_name != row_map_name: continue row_last = _parse_timestamp(row["last_seen_at"]) distance = abs((row_last - ended_point).total_seconds()) if best_distance is None or distance < best_distance: best_row = row best_distance = distance if best_row is None or best_distance is None or best_distance > 21600: return None sample_count = int(best_row["sample_count"] or 0) return { "session_key": best_row["session_key"], "first_seen_at": best_row["first_seen_at"], "last_seen_at": best_row["last_seen_at"], "duration_seconds": _calculate_duration_seconds( best_row["first_seen_at"], best_row["last_seen_at"], ), "map_name": best_row["map_name"], "map_pretty_name": best_row["map_pretty_name"] or best_row["map_name"], "sample_count": sample_count, "average_players": round((int(best_row["total_players"] or 0) / sample_count), 2) if sample_count > 0 else 0.0, "peak_players": int(best_row["peak_players"] or 0), "confidence_mode": best_row["confidence_mode"], "capabilities": _deserialize_json_object(best_row["capabilities_json"]), } def get_rcon_historical_competitive_window_by_session( *, server_key: str, session_key: str, db_path: Path | None = None, ) -> dict[str, object] | None: """Return one persisted competitive RCON window by its synthetic session key.""" if use_postgres_rcon_storage(explicit_sqlite_path=db_path): from .postgres_rcon_storage import get_competitive_window_by_session return get_competitive_window_by_session( server_key=server_key, session_key=session_key, ) normalized_session_key = str(session_key or "").strip() if not normalized_session_key: return None resolved_path = _resolve_db_path(db_path) aliases = _expand_target_key_aliases(server_key) alias_placeholders = ", ".join("?" for _ in aliases) try: with _connect_readonly(resolved_path) as connection: row = connection.execute( f""" SELECT targets.target_key, targets.external_server_id, targets.display_name, targets.region, windows.session_key, windows.map_name, windows.map_pretty_name, windows.first_seen_at, windows.last_seen_at, windows.sample_count, windows.total_players, windows.peak_players, windows.confidence_mode, windows.capabilities_json, windows.latest_payload_json FROM rcon_historical_competitive_windows AS windows INNER JOIN rcon_historical_targets AS targets ON targets.id = windows.target_id WHERE windows.session_key = ? AND ( targets.target_key IN ({alias_placeholders}) OR targets.external_server_id IN ({alias_placeholders}) ) LIMIT 1 """, [normalized_session_key, *aliases, *aliases], ).fetchone() except sqlite3.OperationalError: return None if row is None: return None sample_count = int(row["sample_count"] or 0) return { "target_key": row["target_key"], "external_server_id": row["external_server_id"], "display_name": row["display_name"], "region": row["region"], "session_key": row["session_key"], "first_seen_at": row["first_seen_at"], "last_seen_at": row["last_seen_at"], "duration_seconds": _calculate_duration_seconds( row["first_seen_at"], row["last_seen_at"], ), "map_name": row["map_name"], "map_pretty_name": row["map_pretty_name"] or row["map_name"], "sample_count": sample_count, "average_players": round((int(row["total_players"] or 0) / sample_count), 2) if sample_count > 0 else 0.0, "peak_players": int(row["peak_players"] or 0), "confidence_mode": row["confidence_mode"], "capabilities": _deserialize_json_object(row["capabilities_json"]), "latest_payload": _deserialize_json_object(row["latest_payload_json"]), } @contextmanager def _connect(db_path: Path) -> Iterator[sqlite3.Connection]: connection = connect_sqlite_writer(db_path) try: yield connection connection.commit() except Exception: connection.rollback() raise finally: connection.close() def _connect_readonly(db_path: Path): return closing(connect_sqlite_readonly(db_path)) def _resolve_db_path(db_path: Path | None) -> Path: return db_path or get_storage_path() def _expand_target_key_aliases(target_key: str) -> list[str]: normalized_target_key = str(target_key or "").strip() if not normalized_target_key: return [normalized_target_key] aliases = {normalized_target_key} try: configured_targets = load_rcon_targets() except Exception: configured_targets = () for target in configured_targets: external_server_id = str(target.external_server_id or "").strip() legacy_target_key = f"rcon:{target.host}:{target.port}" if external_server_id and external_server_id == normalized_target_key: aliases.add(legacy_target_key) aliases.add(external_server_id) elif legacy_target_key == normalized_target_key: aliases.add(legacy_target_key) if external_server_id: aliases.add(external_server_id) return sorted(alias for alias in aliases if alias) def _upsert_target(connection: sqlite3.Connection, *, target: Mapping[str, object]) -> int: target_key = str(target.get("target_key") or "").strip() if not target_key: raise ValueError("Prospective RCON targets require a non-empty target_key.") display_name = str(target.get("name") or target.get("display_name") or target_key).strip() host = str(target.get("host") or "").strip() port = int(target.get("port") or 0) if not host or port <= 0: raise ValueError("Prospective RCON targets require host and port.") connection.execute( """ INSERT INTO rcon_historical_targets ( target_key, external_server_id, display_name, host, port, region, game_port, query_port, source_name, last_configured_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(target_key) DO UPDATE SET external_server_id = excluded.external_server_id, display_name = excluded.display_name, host = excluded.host, port = excluded.port, region = excluded.region, game_port = excluded.game_port, query_port = excluded.query_port, source_name = excluded.source_name, last_configured_at = excluded.last_configured_at, updated_at = CURRENT_TIMESTAMP """, ( target_key, target.get("external_server_id"), display_name, host, port, target.get("region"), target.get("game_port"), target.get("query_port"), str(target.get("source_name") or "community-hispana-rcon"), _utc_now_iso(), ), ) row = connection.execute( "SELECT id FROM rcon_historical_targets WHERE target_key = ?", (target_key,), ).fetchone() if row is None: raise RuntimeError("Failed to resolve prospective RCON target id.") return int(row["id"]) def _upsert_checkpoint_success( connection: sqlite3.Connection, *, target_id: int, run_id: int, captured_at: str, ) -> None: connection.execute( """ INSERT INTO rcon_historical_checkpoints ( target_id, last_successful_capture_at, last_sample_at, last_run_id, last_run_status, last_error, last_error_at ) VALUES (?, ?, ?, ?, 'success', NULL, NULL) ON CONFLICT(target_id) DO UPDATE SET last_successful_capture_at = excluded.last_successful_capture_at, last_sample_at = excluded.last_sample_at, last_run_id = excluded.last_run_id, last_run_status = excluded.last_run_status, last_error = NULL, last_error_at = NULL, updated_at = CURRENT_TIMESTAMP """, (target_id, captured_at, captured_at, run_id), ) def _upsert_competitive_window( connection: sqlite3.Connection, *, target_id: int, captured_at: str, normalized_payload: Mapping[str, object], ) -> None: current_map_raw = str(normalized_payload.get("current_map") or "").strip() if not current_map_raw: return map_pretty_name = normalize_map_name(current_map_raw) or current_map_raw players = int(normalized_payload.get("players") or 0) max_players = normalized_payload.get("max_players") status = str(normalized_payload.get("status") or "unknown") latest_window = connection.execute( """ SELECT * FROM rcon_historical_competitive_windows WHERE target_id = ? ORDER BY last_seen_at DESC, id DESC LIMIT 1 """, (target_id,), ).fetchone() if latest_window and _should_extend_competitive_window( latest_window=latest_window, captured_at=captured_at, current_map=current_map_raw, ): connection.execute( """ UPDATE rcon_historical_competitive_windows SET map_name = ?, map_pretty_name = ?, last_seen_at = ?, sample_count = sample_count + 1, total_players = total_players + ?, peak_players = CASE WHEN peak_players > ? THEN peak_players ELSE ? END, last_players = ?, max_players = ?, status = ?, confidence_mode = ?, capabilities_json = ?, latest_payload_json = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? """, ( current_map_raw, map_pretty_name, captured_at, players, players, players, players, max_players, status, COMPETITIVE_MODE_APPROXIMATE, json.dumps(_build_competitive_capabilities(), ensure_ascii=True, separators=(",", ":")), json.dumps(dict(normalized_payload), ensure_ascii=True, separators=(",", ":")), latest_window["id"], ), ) return session_key = f"{target_id}:{captured_at}" connection.execute( """ INSERT INTO rcon_historical_competitive_windows ( target_id, session_key, source_kind, map_name, map_pretty_name, first_seen_at, last_seen_at, sample_count, total_players, peak_players, last_players, max_players, status, confidence_mode, capabilities_json, latest_payload_json ) VALUES (?, ?, 'rcon-historical-samples', ?, ?, ?, ?, 1, ?, ?, ?, ?, ?, ?, ?, ?) """, ( target_id, session_key, current_map_raw, map_pretty_name, captured_at, captured_at, players, players, players, max_players, status, COMPETITIVE_MODE_APPROXIMATE, json.dumps(_build_competitive_capabilities(), ensure_ascii=True, separators=(",", ":")), json.dumps(dict(normalized_payload), ensure_ascii=True, separators=(",", ":")), ), ) def _should_extend_competitive_window( *, latest_window: sqlite3.Row, captured_at: str, current_map: str, ) -> bool: latest_map = str(latest_window["map_name"] or "").strip() if normalize_map_name(latest_map) != normalize_map_name(current_map): return False latest_seen = _parse_timestamp(str(latest_window["last_seen_at"])) captured_point = _parse_timestamp(captured_at) return (captured_point - latest_seen).total_seconds() <= COMPETITIVE_WINDOW_GAP_SECONDS def _build_competitive_capabilities() -> dict[str, object]: return { "recent_matches": COMPETITIVE_MODE_APPROXIMATE, "server_summary": COMPETITIVE_MODE_EXACT, "competitive_quality": COMPETITIVE_MODE_PARTIAL, "result": "session-score", "gamestate": "session", "player_stats": "unavailable", } def _deserialize_json_object(raw_value: object) -> dict[str, object]: if isinstance(raw_value, str) and raw_value.strip(): try: parsed = json.loads(raw_value) except json.JSONDecodeError: return {} if isinstance(parsed, dict): return parsed return {} def _parse_timestamp(raw_value: str) -> datetime: timestamp = datetime.fromisoformat(raw_value.replace("Z", "+00:00")) if timestamp.tzinfo is None: timestamp = timestamp.replace(tzinfo=timezone.utc) return timestamp.astimezone(timezone.utc) def _calculate_duration_seconds(first_seen_at: str | None, last_seen_at: str | None) -> int | None: if not first_seen_at or not last_seen_at: return None return max(0, int((_parse_timestamp(last_seen_at) - _parse_timestamp(first_seen_at)).total_seconds())) def _utc_now_iso() -> str: return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")