rgonsal/comunidadhll
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

RCON model

Browse Source
This commit is contained in:
devRaGonSa
2026-03-25 11:46:28 +01:00
parent 43e1d612af
commit c70073dbe1
19 changed files with 2058 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
backend/app/config.py
View File

@@ -19,6 +19,7 @@ DEFAULT_HISTORICAL_CRCON_DETAIL_WORKERS = 8
DEFAULT_HISTORICAL_CRCON_REQUEST_RETRIES = 3
DEFAULT_HISTORICAL_CRCON_RETRY_DELAY_SECONDS = 0.5
DEFAULT_HISTORICAL_REFRESH_INTERVAL_SECONDS = 1800
DEFAULT_HISTORICAL_REFRESH_OVERLAP_HOURS = 12
DEFAULT_HISTORICAL_SNAPSHOT_REFRESH_INTERVAL_SECONDS = 900
DEFAULT_HISTORICAL_REFRESH_MAX_RETRIES = 2
DEFAULT_HISTORICAL_REFRESH_RETRY_DELAY_SECONDS = 30
@@ -26,8 +27,12 @@ DEFAULT_HISTORICAL_FULL_SNAPSHOT_EVERY_RUNS = 4
DEFAULT_HISTORICAL_WEEKLY_FALLBACK_MIN_MATCHES = 3
DEFAULT_HISTORICAL_WEEKLY_FALLBACK_MAX_WEEKDAY = 2
DEFAULT_PLAYER_EVENT_REFRESH_INTERVAL_SECONDS = 1800
DEFAULT_PLAYER_EVENT_REFRESH_OVERLAP_HOURS = 12
DEFAULT_PLAYER_EVENT_REFRESH_MAX_RETRIES = 2
DEFAULT_PLAYER_EVENT_REFRESH_RETRY_DELAY_SECONDS = 30
DEFAULT_RCON_HISTORICAL_CAPTURE_INTERVAL_SECONDS = 120
DEFAULT_RCON_HISTORICAL_CAPTURE_MAX_RETRIES = 2
DEFAULT_RCON_HISTORICAL_CAPTURE_RETRY_DELAY_SECONDS = 15
DEFAULT_ALLOWED_ORIGINS = (
"null",
"http://127.0.0.1:5500",
@@ -172,6 +177,19 @@ def get_historical_refresh_interval_seconds() -> int:
return interval_seconds
def get_historical_refresh_overlap_hours() -> int:
"""Return the overlap window used by incremental historical refreshes."""
configured_value = os.getenv(
"HLL_HISTORICAL_REFRESH_OVERLAP_HOURS",
str(DEFAULT_HISTORICAL_REFRESH_OVERLAP_HOURS),
)
overlap_hours = int(configured_value)
if overlap_hours < 0:
raise ValueError("HLL_HISTORICAL_REFRESH_OVERLAP_HOURS must be zero or positive.")
return overlap_hours
def get_live_data_source_kind() -> str:
"""Return the live provider kind selected for the current environment."""
source_kind = os.getenv("HLL_BACKEND_LIVE_DATA_SOURCE", DEFAULT_LIVE_DATA_SOURCE).strip()
@@ -284,6 +302,18 @@ def get_player_event_refresh_interval_seconds() -> int:
return interval_seconds
def get_player_event_refresh_overlap_hours() -> int:
"""Return the overlap window used by player event refresh runs."""
configured_value = os.getenv(
"HLL_PLAYER_EVENT_REFRESH_OVERLAP_HOURS",
str(DEFAULT_PLAYER_EVENT_REFRESH_OVERLAP_HOURS),
)
overlap_hours = int(configured_value)
if overlap_hours < 0:
raise ValueError("HLL_PLAYER_EVENT_REFRESH_OVERLAP_HOURS must be zero or positive.")
return overlap_hours
def get_player_event_refresh_max_retries() -> int:
"""Return the retry count used by the player event refresh loop."""
configured_value = os.getenv(
@@ -310,6 +340,44 @@ def get_player_event_refresh_retry_delay_seconds() -> int:
return retry_delay_seconds
def get_rcon_historical_capture_interval_seconds() -> int:
"""Return the default interval used by the prospective RCON capture loop."""
configured_value = os.getenv(
"HLL_RCON_HISTORICAL_CAPTURE_INTERVAL_SECONDS",
str(DEFAULT_RCON_HISTORICAL_CAPTURE_INTERVAL_SECONDS),
)
interval_seconds = int(configured_value)
if interval_seconds <= 0:
raise ValueError("HLL_RCON_HISTORICAL_CAPTURE_INTERVAL_SECONDS must be positive.")
return interval_seconds
def get_rcon_historical_capture_max_retries() -> int:
"""Return the retry count used by the prospective RCON capture loop."""
configured_value = os.getenv(
"HLL_RCON_HISTORICAL_CAPTURE_MAX_RETRIES",
str(DEFAULT_RCON_HISTORICAL_CAPTURE_MAX_RETRIES),
)
max_retries = int(configured_value)
if max_retries < 0:
raise ValueError("HLL_RCON_HISTORICAL_CAPTURE_MAX_RETRIES must be zero or positive.")
return max_retries
def get_rcon_historical_capture_retry_delay_seconds() -> int:
"""Return the wait time between failed prospective RCON capture attempts."""
configured_value = os.getenv(
"HLL_RCON_HISTORICAL_CAPTURE_RETRY_DELAY_SECONDS",
str(DEFAULT_RCON_HISTORICAL_CAPTURE_RETRY_DELAY_SECONDS),
)
retry_delay_seconds = int(configured_value)
if retry_delay_seconds < 0:
raise ValueError(
"HLL_RCON_HISTORICAL_CAPTURE_RETRY_DELAY_SECONDS must be zero or positive."
)
return retry_delay_seconds
def get_a2s_targets_payload() -> str | None:
"""Return the optional JSON payload that overrides local A2S targets."""
raw_payload = os.getenv(DEFAULT_A2S_TARGETS_ENV_VAR)

36
backend/app/data_sources.py
View File

@@ -9,6 +9,11 @@ from .collector import collect_server_snapshots
from .config import get_historical_data_source_kind, get_live_data_source_kind
from .providers.public_scoreboard_provider import PublicScoreboardHistoricalDataSource
from .providers.rcon_provider import RconLiveDataSource
from .rcon_historical_read_model import (
describe_rcon_historical_read_model,
list_rcon_historical_recent_activity,
list_rcon_historical_server_summaries,
)
from .server_targets import A2SServerTarget, load_a2s_targets
@@ -72,15 +77,19 @@ class A2SLiveDataSource:
@dataclass(frozen=True, slots=True)
class RconHistoricalDataSource:
"""Placeholder historical provider for future production RCON integration."""
"""Minimal persisted historical read model over prospective RCON capture."""
source_kind: str = SOURCE_KIND_RCON
def fetch_public_info(self, *, base_url: str) -> dict[str, object]:
raise RuntimeError("Historical RCON provider is not implemented yet.")
raise RuntimeError(
"RCON historical read mode does not support CRCON ingestion operations."
)
def fetch_match_page(self, *, base_url: str, page: int, limit: int) -> dict[str, object]:
raise RuntimeError("Historical RCON provider is not implemented yet.")
raise RuntimeError(
"RCON historical read mode does not support CRCON ingestion operations."
)
def fetch_match_details(
self,
@@ -89,7 +98,26 @@ class RconHistoricalDataSource:
match_ids: list[str],
max_workers: int,
) -> list[dict[str, object]]:
raise RuntimeError("Historical RCON provider is not implemented yet.")
raise RuntimeError(
"RCON historical read mode does not support CRCON ingestion operations."
)
def list_server_summaries(self, *, server_key: str | None = None) -> list[dict[str, object]]:
"""Return coverage and freshness from persisted prospective RCON samples."""
return list_rcon_historical_server_summaries(server_key=server_key)
def list_recent_activity(
self,
*,
server_key: str | None = None,
limit: int = 20,
) -> list[dict[str, object]]:
"""Return recent persisted RCON activity without on-demand network calls."""
return list_rcon_historical_recent_activity(server_key=server_key, limit=limit)
def describe_capabilities(self) -> dict[str, object]:
"""Describe the supported RCON historical read surface."""
return describe_rcon_historical_read_model()
def get_historical_data_source() -> HistoricalDataSource:

23
backend/app/historical_ingestion.py
View File

@@ -10,6 +10,7 @@ from typing import Iterable
from .config import (
get_historical_crcon_detail_workers,
get_historical_crcon_page_size,
get_historical_refresh_overlap_hours,
)
from .data_sources import HistoricalDataSource, get_historical_data_source
from .historical_snapshots import generate_and_persist_historical_snapshots
@@ -75,6 +76,7 @@ def run_incremental_refresh(
page_size: int | None = None,
start_page: int | None = None,
detail_workers: int | None = None,
overlap_hours: int | None = None,
rebuild_snapshots: bool = True,
) -> dict[str, object]:
"""Refresh recent historical pages without replaying the whole archive."""
@@ -85,6 +87,7 @@ def run_incremental_refresh(
page_size=page_size,
start_page=start_page,
detail_workers=detail_workers,
overlap_hours=overlap_hours,
incremental=True,
rebuild_snapshots=rebuild_snapshots,
)
@@ -98,6 +101,7 @@ def _run_ingestion(
page_size: int | None,
start_page: int | None,
detail_workers: int | None,
overlap_hours: int | None,
incremental: bool,
rebuild_snapshots: bool,
) -> dict[str, object]:
@@ -107,6 +111,13 @@ def _run_ingestion(
selected_servers = _select_servers(server_slug)
processed_servers: list[dict[str, object]] = []
active_runs: dict[str, int] = {}
resolved_overlap_hours = (
get_historical_refresh_overlap_hours()
if overlap_hours is None
else overlap_hours
)
if resolved_overlap_hours < 0:
raise ValueError("--overlap-hours must be zero or positive.")
try:
for server in selected_servers:
@@ -118,7 +129,10 @@ def _run_ingestion(
run_id=run_id,
)
cutoff = (
get_refresh_cutoff_for_server(str(server["slug"]))
get_refresh_cutoff_for_server(
str(server["slug"]),
overlap_hours=resolved_overlap_hours,
)
if incremental
else None
)
@@ -196,6 +210,7 @@ def _run_ingestion(
"page_size": page_size or get_historical_crcon_page_size(),
"start_page": start_page,
"detail_workers": detail_workers or get_historical_crcon_detail_workers(),
"overlap_hours": resolved_overlap_hours if incremental else None,
"servers": processed_servers,
"coverage": list_historical_coverage_report(server_slug=server_slug),
"snapshot_result": snapshot_result,
@@ -408,6 +423,11 @@ def build_arg_parser() -> argparse.ArgumentParser:
type=int,
help="parallel worker count for per-match detail requests",
)
parser.add_argument(
"--overlap-hours",
type=int,
help="override the incremental overlap window in hours",
)
return parser
@@ -431,6 +451,7 @@ def main(argv: Iterable[str] | None = None) -> int:
page_size=args.page_size,
start_page=args.start_page,
detail_workers=args.detail_workers,
overlap_hours=args.overlap_hours,
)
print(json.dumps(result, indent=2))

13
backend/app/historical_storage.py
View File

@@ -8,6 +8,7 @@ from pathlib import Path
from typing import Mapping
from .config import (
get_historical_refresh_overlap_hours,
get_historical_weekly_fallback_max_weekday,
get_historical_weekly_fallback_min_matches,
get_storage_path,
@@ -40,7 +41,6 @@ DEFAULT_HISTORICAL_SERVERS = (
ALL_SERVERS_SLUG = "all-servers"
ALL_SERVERS_DISPLAY_NAME = "Todos"
DEFAULT_WEEKLY_WINDOW_DAYS = 7
DEFAULT_REFRESH_OVERLAP_HOURS = 12
SUPPORTED_WEEKLY_LEADERBOARD_METRICS = frozenset(
{
"kills",
@@ -707,10 +707,17 @@ def upsert_historical_match(
def get_refresh_cutoff_for_server(
server_slug: str,
*,
overlap_hours: int = DEFAULT_REFRESH_OVERLAP_HOURS,
overlap_hours: int | None = None,
db_path: Path | None = None,
) -> str | None:
"""Return the timestamp used to stop incremental scans once older pages appear."""
resolved_overlap_hours = (
get_historical_refresh_overlap_hours()
if overlap_hours is None
else overlap_hours
)
if resolved_overlap_hours < 0:
raise ValueError("overlap_hours must be zero or positive.")
resolved_path = initialize_historical_storage(db_path=db_path)
with _connect(resolved_path) as connection:
server_row = _resolve_historical_server(connection, server_slug)
@@ -726,7 +733,7 @@ def get_refresh_cutoff_for_server(
if not latest_seen_at:
return None
cutoff = _parse_timestamp(latest_seen_at) - timedelta(hours=overlap_hours)
cutoff = _parse_timestamp(latest_seen_at) - timedelta(hours=resolved_overlap_hours)
return cutoff.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")

184
backend/app/payloads.py
View File

@@ -9,7 +9,7 @@ from .config import (
get_live_data_source_kind,
get_refresh_interval_seconds,
)
from .data_sources import get_live_data_source
from .data_sources import RconHistoricalDataSource, get_historical_data_source, get_live_data_source
from .historical_snapshot_storage import get_historical_snapshot
from .historical_snapshots import (
DEFAULT_MONTHLY_SNAPSHOT_WINDOW,
@@ -219,6 +219,14 @@ def build_weekly_top_kills_payload(
server_id: str | None = None,
) -> dict[str, object]:
"""Return weekly top kills grouped by real community server."""
rcon_payload = _build_rcon_historical_unsupported_payload(
title="Top kills semanales por servidor",
context="historical-top-kills",
server_key=server_id,
limit=limit,
)
if rcon_payload is not None:
return rcon_payload
result = list_weekly_top_kills(limit=limit, server_id=server_id)
return {
"status": "ok",
@@ -244,6 +252,16 @@ def build_historical_leaderboard_payload(
timeframe: str = "weekly",
) -> dict[str, object]:
"""Return one historical leaderboard for the requested timeframe and metric."""
rcon_payload = _build_rcon_historical_unsupported_payload(
title="Ranking historico",
context="historical-leaderboard",
server_key=server_id,
limit=limit,
metric=metric,
timeframe=timeframe,
)
if rcon_payload is not None:
return rcon_payload
normalized_timeframe = timeframe.strip().lower() if isinstance(timeframe, str) else "weekly"
if normalized_timeframe == "monthly":
result = list_monthly_leaderboard(limit=limit, server_id=server_id, metric=metric)
@@ -324,6 +342,26 @@ def build_recent_historical_matches_payload(
server_slug: str | None = None,
) -> dict[str, object]:
"""Return recent historical matches from persisted CRCON data."""
if get_historical_data_source_kind() == "rcon":
data_source = get_historical_data_source()
if isinstance(data_source, RconHistoricalDataSource):
items = data_source.list_recent_activity(server_key=server_slug, limit=limit)
capabilities = data_source.describe_capabilities()
return {
"status": "ok",
"data": {
"title": "Actividad reciente capturada por RCON",
"context": "historical-recent-matches",
"source": "rcon-historical-read-model",
"historical_data_source": "rcon",
"supported": True,
"coverage_basis": "prospective-rcon-samples",
"limit": limit,
"server_slug": server_slug,
"items": items,
"capabilities": capabilities,
},
}
items = list_recent_historical_matches(limit=limit, server_slug=server_slug)
return {
"status": "ok",
@@ -344,6 +382,14 @@ def build_monthly_mvp_payload(
server_id: str | None = None,
) -> dict[str, object]:
"""Return the precomputed monthly MVP payload through the stable API surface."""
rcon_payload = _build_rcon_historical_unsupported_payload(
title="Top MVP mensual",
context="historical-monthly-mvp",
server_key=server_id,
limit=limit,
)
if rcon_payload is not None:
return rcon_payload
snapshot_payload = build_monthly_mvp_snapshot_payload(
limit=limit,
server_id=server_id,
@@ -370,6 +416,15 @@ def build_player_event_payload(
view: str = "most-killed",
) -> dict[str, object]:
"""Return one V2 player-event payload through the stable API surface."""
rcon_payload = _build_rcon_historical_unsupported_payload(
title="Metricas V2 mensuales",
context="historical-player-events",
server_key=server_id,
limit=limit,
metric=view,
)
if rcon_payload is not None:
return rcon_payload
snapshot_payload = build_player_event_snapshot_payload(
limit=limit,
server_id=server_id,
@@ -397,6 +452,14 @@ def build_monthly_mvp_v2_payload(
server_id: str | None = None,
) -> dict[str, object]:
"""Return the precomputed monthly MVP V2 payload through the stable API surface."""
rcon_payload = _build_rcon_historical_unsupported_payload(
title="Top MVP mensual V2",
context="historical-monthly-mvp-v2",
server_key=server_id,
limit=limit,
)
if rcon_payload is not None:
return rcon_payload
snapshot_payload = build_monthly_mvp_v2_snapshot_payload(
limit=limit,
server_id=server_id,
@@ -421,6 +484,13 @@ def build_historical_server_summary_snapshot_payload(
server_slug: str | None = None,
) -> dict[str, object]:
"""Return one precomputed summary snapshot without recalculating aggregates."""
rcon_payload = _build_rcon_historical_unsupported_payload(
title="Snapshot historico de resumen por servidor",
context="historical-server-summary-snapshot",
server_key=server_slug,
)
if rcon_payload is not None:
return rcon_payload
snapshot = _get_historical_snapshot_record(
server_key=server_slug,
snapshot_type=SNAPSHOT_TYPE_SERVER_SUMMARY,
@@ -450,6 +520,16 @@ def build_leaderboard_snapshot_payload(
timeframe: str = "weekly",
) -> dict[str, object]:
"""Return one precomputed leaderboard snapshot for the requested timeframe."""
rcon_payload = _build_rcon_historical_unsupported_payload(
title="Snapshot historico de leaderboard",
context="historical-leaderboard-snapshot",
server_key=server_id,
limit=limit,
metric=metric,
timeframe=timeframe,
)
if rcon_payload is not None:
return rcon_payload
normalized_timeframe = timeframe.strip().lower() if isinstance(timeframe, str) else "weekly"
if normalized_timeframe == "monthly":
snapshot_type = SNAPSHOT_TYPE_MONTHLY_LEADERBOARD
@@ -552,6 +632,14 @@ def build_recent_historical_matches_snapshot_payload(
server_slug: str | None = None,
) -> dict[str, object]:
"""Return one precomputed recent-matches snapshot."""
rcon_payload = _build_rcon_historical_unsupported_payload(
title="Snapshot historico de actividad reciente por servidor",
context="historical-recent-matches-snapshot",
server_key=server_slug,
limit=limit,
)
if rcon_payload is not None:
return rcon_payload
snapshot = _get_historical_snapshot_record(
server_key=server_slug,
snapshot_type=SNAPSHOT_TYPE_RECENT_MATCHES,
@@ -582,6 +670,14 @@ def build_monthly_mvp_snapshot_payload(
server_id: str | None = None,
) -> dict[str, object]:
"""Return one precomputed monthly MVP snapshot."""
rcon_payload = _build_rcon_historical_unsupported_payload(
title="Snapshot top MVP mensual",
context="historical-monthly-mvp-snapshot",
server_key=server_id,
limit=limit,
)
if rcon_payload is not None:
return rcon_payload
snapshot = _get_historical_snapshot_record(
server_key=server_id,
snapshot_type=SNAPSHOT_TYPE_MONTHLY_MVP,
@@ -638,6 +734,14 @@ def build_monthly_mvp_v2_snapshot_payload(
server_id: str | None = None,
) -> dict[str, object]:
"""Return one precomputed monthly MVP V2 snapshot."""
rcon_payload = _build_rcon_historical_unsupported_payload(
title="Snapshot top MVP mensual V2",
context="historical-monthly-mvp-v2-snapshot",
server_key=server_id,
limit=limit,
)
if rcon_payload is not None:
return rcon_payload
snapshot = _get_historical_snapshot_record(
server_key=server_id,
snapshot_type=SNAPSHOT_TYPE_MONTHLY_MVP_V2,
@@ -697,6 +801,15 @@ def build_player_event_snapshot_payload(
view: str = "most-killed",
) -> dict[str, object]:
"""Return one precomputed V2 player-event snapshot."""
rcon_payload = _build_rcon_historical_unsupported_payload(
title="Snapshot metricas V2 mensuales",
context="historical-player-events-snapshot",
server_key=server_id,
limit=limit,
metric=view,
)
if rcon_payload is not None:
return rcon_payload
snapshot_type = _resolve_player_event_snapshot_type(view)
snapshot = _get_historical_snapshot_record(
server_key=server_id,
@@ -736,6 +849,29 @@ def build_historical_server_summary_payload(
server_slug: str | None = None,
) -> dict[str, object]:
"""Return aggregated historical metrics per server."""
if get_historical_data_source_kind() == "rcon":
data_source = get_historical_data_source()
if isinstance(data_source, RconHistoricalDataSource):
items = data_source.list_server_summaries(server_key=server_slug)
capabilities = data_source.describe_capabilities()
return {
"status": "ok",
"data": {
"title": (
"Cobertura historica minima por RCON"
if server_slug != ALL_SERVERS_SLUG
else "Cobertura historica minima RCON agregada"
),
"context": "historical-server-summary",
"source": "rcon-historical-read-model",
"historical_data_source": "rcon",
"summary_basis": "prospective-rcon-samples",
"server_slug": server_slug,
"supported": True,
"items": items,
"capabilities": capabilities,
},
}
items = list_historical_server_summaries(server_slug=server_slug)
return {
"status": "ok",
@@ -757,6 +893,13 @@ def build_historical_server_summary_payload(
def build_historical_player_profile_payload(player_id: str) -> dict[str, object]:
"""Return aggregate historical metrics for one player identity."""
rcon_payload = _build_rcon_historical_unsupported_payload(
title="Perfil historico de jugador",
context="historical-player-profile",
server_key=player_id,
)
if rcon_payload is not None:
return rcon_payload
profile = get_historical_player_profile(player_id)
return {
"status": "ok",
@@ -788,6 +931,45 @@ def _get_historical_snapshot_record(
)
def _build_rcon_historical_unsupported_payload(
*,
title: str,
context: str,
server_key: str | None = None,
limit: int | None = None,
metric: str | None = None,
timeframe: str | None = None,
) -> dict[str, object] | None:
if get_historical_data_source_kind() != "rcon":
return None
data_source = get_historical_data_source()
if not isinstance(data_source, RconHistoricalDataSource):
return None
return {
"status": "ok",
"data": {
"title": title,
"context": context,
"source": "rcon-historical-read-model",
"historical_data_source": "rcon",
"supported": False,
"server_slug": server_key,
"limit": limit,
"metric": metric,
"timeframe": timeframe,
"items": [],
"item": None,
"profile": None,
"found": False,
"missing_reason": "rcon-minimal-read-model-does-not-support-this-endpoint-yet",
"supported_historical_source_for_full_contract": "public-scoreboard",
"capabilities": data_source.describe_capabilities(),
},
}
def _build_historical_snapshot_metadata(snapshot: dict[str, object] | None) -> dict[str, object]:
if snapshot is None:
return {

27
backend/app/player_event_storage.py
View File

@@ -4,10 +4,10 @@ from __future__ import annotations
import sqlite3
from collections.abc import Iterable
from datetime import datetime, timezone
from datetime import datetime, timedelta, timezone
from pathlib import Path
from .config import get_storage_path
from .config import get_player_event_refresh_overlap_hours, get_storage_path
from .player_event_models import PlayerEventRecord
@@ -386,9 +386,17 @@ def get_player_event_resume_page(
def get_player_event_refresh_cutoff_for_server(
server_slug: str,
*,
overlap_hours: int | None = None,
db_path: Path | None = None,
) -> str | None:
"""Return the latest occurred_at already persisted for one server."""
resolved_overlap_hours = (
get_player_event_refresh_overlap_hours()
if overlap_hours is None
else overlap_hours
)
if resolved_overlap_hours < 0:
raise ValueError("overlap_hours must be zero or positive.")
resolved_path = initialize_player_event_storage(db_path=db_path)
with _connect(resolved_path) as connection:
row = connection.execute(
@@ -399,7 +407,12 @@ def get_player_event_refresh_cutoff_for_server(
""",
(server_slug,),
).fetchone()
return str(row["latest_occurred_at"]) if row and row["latest_occurred_at"] else None
latest_occurred_at = str(row["latest_occurred_at"]) if row and row["latest_occurred_at"] else None
if not latest_occurred_at:
return None
cutoff = _parse_timestamp(latest_occurred_at) - timedelta(hours=resolved_overlap_hours)
return cutoff.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
def _connect(db_path: Path) -> sqlite3.Connection:
@@ -411,3 +424,11 @@ def _connect(db_path: Path) -> sqlite3.Connection:
def _utc_now_iso() -> str:
return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
def _parse_timestamp(value: str) -> datetime:
normalized = value.strip().replace("Z", "+00:00")
parsed = datetime.fromisoformat(normalized)
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=timezone.utc)
return parsed

21
backend/app/player_event_worker.py
View File

@@ -13,6 +13,7 @@ from .config import (
get_historical_crcon_page_size,
get_player_event_refresh_interval_seconds,
get_player_event_refresh_max_retries,
get_player_event_refresh_overlap_hours,
get_player_event_refresh_retry_delay_seconds,
)
from .data_sources import get_historical_data_source
@@ -51,6 +52,7 @@ def run_player_event_refresh(
page_size: int | None = None,
start_page: int | None = None,
detail_workers: int | None = None,
overlap_hours: int | None = None,
) -> dict[str, object]:
"""Refresh recent player event summaries from the configured historical source."""
initialize_player_event_storage()
@@ -58,6 +60,13 @@ def run_player_event_refresh(
event_source = get_player_event_source()
resolved_page_size = page_size or get_historical_crcon_page_size()
resolved_detail_workers = detail_workers or get_historical_crcon_detail_workers()
resolved_overlap_hours = (
get_player_event_refresh_overlap_hours()
if overlap_hours is None
else overlap_hours
)
if resolved_overlap_hours < 0:
raise ValueError("--overlap-hours must be zero or positive.")
selected_servers = _select_servers(server_slug)
processed_servers: list[dict[str, object]] = []
active_runs: dict[str, int] = {}
@@ -70,7 +79,10 @@ def run_player_event_refresh(
target_server_slug=current_server_slug,
)
active_runs[current_server_slug] = run_id
cutoff = get_player_event_refresh_cutoff_for_server(current_server_slug)
cutoff = get_player_event_refresh_cutoff_for_server(
current_server_slug,
overlap_hours=resolved_overlap_hours,
)
mark_player_event_progress_started(
server_slug=current_server_slug,
mode="refresh",
@@ -138,6 +150,7 @@ def run_player_event_refresh(
"event_adapter": event_source.source_kind,
"page_size": resolved_page_size,
"detail_workers": resolved_detail_workers,
"overlap_hours": resolved_overlap_hours,
"scope": event_source.describe_scope(),
"servers": processed_servers,
}
@@ -394,6 +407,11 @@ def build_arg_parser() -> argparse.ArgumentParser:
type=int,
help="parallel worker count for per-match detail requests",
)
parser.add_argument(
"--overlap-hours",
type=int,
help="override the incremental overlap window in hours",
)
parser.add_argument(
"--interval",
type=int,
@@ -432,6 +450,7 @@ def main(argv: Iterable[str] | None = None) -> int:
page_size=args.page_size,
start_page=args.start_page,
detail_workers=args.detail_workers,
overlap_hours=args.overlap_hours,
)
print(json.dumps(result, indent=2))
return 0

8
backend/app/providers/rcon_provider.py
View File

@@ -4,7 +4,11 @@ from __future__ import annotations
from dataclasses import dataclass
from ..rcon_client import RconServerTarget, load_rcon_targets, query_live_server_state
from ..rcon_client import (
RconServerTarget,
load_rcon_targets,
query_live_server_sample,
)
from ..snapshots import build_snapshot_batch, utc_now
from ..storage import persist_snapshot_batch
@@ -26,7 +30,7 @@ class RconLiveDataSource:
for target in configured_targets:
try:
normalized_records.append(query_live_server_state(target))
normalized_records.append(query_live_server_sample(target)["normalized"])
except Exception as error: # noqa: BLE001 - keep provider failures controlled
errors.append(
{

54
backend/app/rcon_client.py
View File

@@ -192,6 +192,16 @@ def query_live_server_state(
timeout_seconds: float | None = None,
) -> dict[str, object]:
"""Query one HLL server via RCON and normalize it to the live snapshot shape."""
sample = query_live_server_sample(target, timeout_seconds=timeout_seconds)
return dict(sample["normalized"])
def query_live_server_sample(
target: RconServerTarget,
*,
timeout_seconds: float | None = None,
) -> dict[str, object]:
"""Query one HLL server and return both normalized and raw session data."""
resolved_timeout = timeout_seconds or get_rcon_request_timeout_seconds()
with HllRconConnection(timeout_seconds=resolved_timeout) as connection:
connection.connect(host=target.host, port=target.port, password=target.password)
@@ -202,19 +212,43 @@ def query_live_server_state(
resolved_external_id = target.external_server_id or f"rcon:{target.host}:{target.port}"
return {
"external_server_id": resolved_external_id,
"server_name": _string_or_none(session.get("serverName")) or target.name,
"status": "online",
"players": _coerce_optional_int(session.get("playerCount")),
"max_players": _coerce_optional_int(session.get("maxPlayerCount")),
"current_map": _string_or_none(session.get("mapId")) or _string_or_none(session.get("mapName")),
"region": target.region,
"source_name": target.source_name,
"snapshot_origin": "real-rcon",
"source_ref": f"rcon://{target.host}:{target.port}",
"target": {
"target_key": build_rcon_target_key(target),
"name": target.name,
"host": target.host,
"port": target.port,
"external_server_id": target.external_server_id,
"region": target.region,
"game_port": target.game_port,
"query_port": target.query_port,
"source_name": target.source_name,
},
"normalized": {
"external_server_id": resolved_external_id,
"server_name": _string_or_none(session.get("serverName")) or target.name,
"status": "online",
"players": _coerce_optional_int(session.get("playerCount")),
"max_players": _coerce_optional_int(session.get("maxPlayerCount")),
"current_map": (
_string_or_none(session.get("mapId")) or _string_or_none(session.get("mapName"))
),
"region": target.region,
"source_name": target.source_name,
"snapshot_origin": "real-rcon",
"source_ref": f"rcon://{target.host}:{target.port}",
},
"raw_session": session,
}
def build_rcon_target_key(target: RconServerTarget) -> str:
"""Build a stable local key for one configured RCON target."""
external_server_id = _string_or_none(target.external_server_id)
if external_server_id:
return external_server_id
return f"rcon:{target.host}:{target.port}"
def _coerce_rcon_target(raw_target: dict[str, object]) -> RconServerTarget:
name = str(raw_target.get("name") or "Unnamed RCON target").strip()
host = str(raw_target.get("host") or "").strip()

198
backend/app/rcon_historical_read_model.py Normal file
View File

@@ -0,0 +1,198 @@
"""Read-only minimal HTTP model over prospective RCON historical persistence."""
from __future__ import annotations
from datetime import datetime, timezone
from .historical_storage import ALL_SERVERS_SLUG
from .normalizers import normalize_map_name
from .rcon_historical_storage import (
list_rcon_historical_target_statuses,
list_recent_rcon_historical_samples,
)
def list_rcon_historical_server_summaries(
*,
server_key: str | None = None,
) -> list[dict[str, object]]:
"""Return per-target coverage and freshness from prospective RCON storage."""
items = list_rcon_historical_target_statuses()
if server_key and server_key != ALL_SERVERS_SLUG:
normalized = server_key.strip()
items = [
item
for item in items
if item["target_key"] == normalized or item["external_server_id"] == normalized
]
summaries = [_build_server_summary(item) for item in items]
if server_key == ALL_SERVERS_SLUG:
return [_build_all_servers_summary(summaries)]
return summaries
def list_rcon_historical_recent_activity(
*,
server_key: str | None = None,
limit: int = 20,
) -> list[dict[str, object]]:
"""Return recent persisted RCON activity samples for one or all targets."""
normalized_server_key = None if server_key == ALL_SERVERS_SLUG else server_key
items = list_recent_rcon_historical_samples(target_key=normalized_server_key, limit=limit)
return [
{
**item,
"current_map": normalize_map_name(item.get("current_map")),
"minutes_since_capture": _minutes_since_timestamp(item.get("captured_at")),
}
for item in items
]
def describe_rcon_historical_read_model() -> dict[str, object]:
"""Describe what the minimal RCON historical read model currently supports."""
return {
"source": "rcon-historical-read-model",
"supported_endpoints": [
"/api/historical/server-summary",
"/api/historical/recent-matches",
],
"unsupported_endpoints": [
"/api/historical/weekly-top-kills",
"/api/historical/weekly-leaderboard",
"/api/historical/leaderboard",
"/api/historical/monthly-mvp",
"/api/historical/monthly-mvp-v2",
"/api/historical/player-events",
"/api/historical/player-profile",
"/api/historical/snapshots/*",
],
"capabilities": [
"coverage by configured RCON target",
"recent persisted live activity",
"freshness and last successful capture metadata",
],
"limitations": [
"No retroactive backfill of closed matches.",
"No weekly or monthly competitive leaderboards.",
"No MVP or player-event parity with public-scoreboard.",
"No precomputed historical snapshots for the RCON read model yet.",
],
}
def _build_server_summary(item: dict[str, object]) -> dict[str, object]:
sample_count = int(item.get("sample_count") or 0)
first_last_points = list_rcon_historical_recent_activity(
server_key=str(item["target_key"]),
limit=1,
)
last_sample_at = item.get("last_sample_at")
latest_activity = first_last_points[0] if first_last_points else None
return {
"server": {
"slug": item["target_key"],
"name": item["display_name"],
"external_server_id": item["external_server_id"],
"region": item["region"],
},
"coverage": {
"basis": "prospective-rcon-samples",
"status": "available" if sample_count > 0 else "empty",
"sample_count": sample_count,
"first_sample_at": item.get("first_sample_at"),
"last_sample_at": last_sample_at,
"coverage_hours": _calculate_coverage_hours(item.get("first_sample_at"), last_sample_at),
},
"freshness": {
"last_successful_capture_at": item.get("last_successful_capture_at"),
"minutes_since_last_capture": _minutes_since_timestamp(last_sample_at),
"last_run_id": item.get("last_run_id"),
"last_run_status": item.get("last_run_status"),
"last_error": item.get("last_error"),
"last_error_at": item.get("last_error_at"),
},
"activity": {
"latest_players": latest_activity.get("players") if latest_activity else None,
"latest_max_players": latest_activity.get("max_players") if latest_activity else None,
"latest_map": latest_activity.get("current_map") if latest_activity else None,
"latest_status": latest_activity.get("status") if latest_activity else None,
},
"time_range": {
"start": None,
"end": last_sample_at,
},
}
def _build_all_servers_summary(items: list[dict[str, object]]) -> dict[str, object]:
total_samples = sum(int(item["coverage"].get("sample_count") or 0) for item in items)
last_points = [
item["time_range"].get("end")
for item in items
if item["time_range"].get("end")
]
last_capture_at = max(last_points) if last_points else None
return {
"server": {
"slug": ALL_SERVERS_SLUG,
"name": "Todos",
"external_server_id": None,
"region": None,
},
"coverage": {
"basis": "prospective-rcon-samples-aggregate",
"status": "available" if total_samples > 0 else "empty",
"sample_count": total_samples,
"first_sample_at": None,
"last_sample_at": last_capture_at,
"coverage_hours": None,
},
"freshness": {
"last_successful_capture_at": last_capture_at,
"minutes_since_last_capture": _minutes_since_timestamp(last_capture_at),
"last_run_id": None,
"last_run_status": None,
"last_error": None,
"last_error_at": None,
},
"activity": {
"latest_players": None,
"latest_max_players": None,
"latest_map": None,
"latest_status": None,
},
"time_range": {
"start": None,
"end": last_capture_at,
},
"server_count": len(items),
}
def _minutes_since_timestamp(timestamp: str | None) -> int | None:
if not timestamp:
return None
captured_at = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
if captured_at.tzinfo is None:
captured_at = captured_at.replace(tzinfo=timezone.utc)
delta = datetime.now(timezone.utc) - captured_at.astimezone(timezone.utc)
return max(0, int(delta.total_seconds() // 60))
def _calculate_coverage_hours(
first_sample_at: str | None,
last_sample_at: str | None,
) -> float | None:
if not first_sample_at or not last_sample_at:
return None
first_point = datetime.fromisoformat(first_sample_at.replace("Z", "+00:00"))
last_point = datetime.fromisoformat(last_sample_at.replace("Z", "+00:00"))
if first_point.tzinfo is None:
first_point = first_point.replace(tzinfo=timezone.utc)
if last_point.tzinfo is None:
last_point = last_point.replace(tzinfo=timezone.utc)
delta = last_point.astimezone(timezone.utc) - first_point.astimezone(timezone.utc)
return round(delta.total_seconds() / 3600, 2)

453
backend/app/rcon_historical_storage.py Normal file
View File

@@ -0,0 +1,453 @@
"""Separate storage and run tracking for prospective RCON historical capture."""
from __future__ import annotations
import json
import sqlite3
from collections.abc import Mapping
from datetime import datetime, timezone
from pathlib import Path
from .config import get_storage_path
def initialize_rcon_historical_storage(*, db_path: Path | None = None) -> Path:
"""Create the SQLite structures used by prospective RCON capture."""
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);
"""
)
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."""
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."""
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."""
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,
)
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."""
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."""
resolved_path = initialize_rcon_historical_storage(db_path=db_path)
with _connect(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()
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."""
resolved_path = initialize_rcon_historical_storage(db_path=db_path)
where_clause = ""
params: list[object] = [limit]
if target_key:
where_clause = "WHERE targets.target_key = ? OR targets.external_server_id = ?"
params = [target_key, target_key, limit]
with _connect(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()
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 _connect(db_path: Path) -> sqlite3.Connection:
connection = sqlite3.connect(db_path)
connection.row_factory = sqlite3.Row
connection.execute("PRAGMA foreign_keys = ON")
return connection
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 _utc_now_iso() -> str:
return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")

299
backend/app/rcon_historical_worker.py Normal file
View File

@@ -0,0 +1,299 @@
"""Dedicated prospective RCON historical capture worker."""
from __future__ import annotations
import argparse
import json
import time
from dataclasses import dataclass
from typing import Iterable
from .config import (
get_rcon_historical_capture_interval_seconds,
get_rcon_historical_capture_max_retries,
get_rcon_historical_capture_retry_delay_seconds,
)
from .rcon_client import build_rcon_target_key, load_rcon_targets, query_live_server_sample
from .rcon_historical_storage import (
finalize_rcon_historical_capture_run,
initialize_rcon_historical_storage,
list_rcon_historical_target_statuses,
mark_rcon_historical_capture_failure,
persist_rcon_historical_sample,
start_rcon_historical_capture_run,
)
from .snapshots import utc_now
@dataclass(slots=True)
class RconHistoricalCaptureStats:
targets_seen: int = 0
samples_inserted: int = 0
duplicate_samples: int = 0
failed_targets: int = 0
def run_rcon_historical_capture(
*,
target_key: str | None = None,
) -> dict[str, object]:
"""Capture one prospective RCON sample for one or all configured targets."""
initialize_rcon_historical_storage()
selected_targets = _select_targets(target_key)
captured_at = utc_now().isoformat().replace("+00:00", "Z")
target_scope = target_key or "all-configured-rcon-targets"
run_id = start_rcon_historical_capture_run(mode="capture", target_scope=target_scope)
stats = RconHistoricalCaptureStats()
items: list[dict[str, object]] = []
errors: list[dict[str, object]] = []
try:
for target in selected_targets:
target_metadata = _serialize_target(target)
stats.targets_seen += 1
try:
sample = query_live_server_sample(target)
delta = persist_rcon_historical_sample(
run_id=run_id,
captured_at=captured_at,
target=target_metadata,
normalized_payload=sample["normalized"],
raw_payload=sample["raw_session"],
)
stats.samples_inserted += int(delta["samples_inserted"])
stats.duplicate_samples += int(delta["duplicate_samples"])
items.append(
{
"target_key": target_metadata["target_key"],
"external_server_id": target.external_server_id,
"captured_at": captured_at,
"sample_inserted": bool(delta["samples_inserted"]),
"normalized": sample["normalized"],
}
)
except Exception as exc: # noqa: BLE001 - controlled worker failures
stats.failed_targets += 1
mark_rcon_historical_capture_failure(
run_id=run_id,
target=target_metadata,
error_message=str(exc),
)
errors.append(
{
"target_key": target_metadata["target_key"],
"name": target.name,
"host": target.host,
"port": target.port,
"message": str(exc),
}
)
status = "success" if not errors else ("partial" if items else "failed")
finalize_rcon_historical_capture_run(
run_id,
status=status,
targets_seen=stats.targets_seen,
samples_inserted=stats.samples_inserted,
duplicate_samples=stats.duplicate_samples,
failed_targets=stats.failed_targets,
notes=None if not errors else json.dumps(errors, separators=(",", ":")),
)
except Exception as exc:
finalize_rcon_historical_capture_run(
run_id,
status="failed",
targets_seen=stats.targets_seen,
samples_inserted=stats.samples_inserted,
duplicate_samples=stats.duplicate_samples,
failed_targets=max(1, stats.failed_targets),
notes=str(exc),
)
raise
return {
"status": "ok" if items else "error",
"run_status": status,
"captured_at": captured_at,
"target_scope": target_scope,
"targets": items,
"errors": errors,
"storage_status": list_rcon_historical_target_statuses(),
"totals": {
"targets_seen": stats.targets_seen,
"samples_inserted": stats.samples_inserted,
"duplicate_samples": stats.duplicate_samples,
"failed_targets": stats.failed_targets,
},
}
def run_periodic_rcon_historical_capture(
*,
interval_seconds: int,
max_retries: int,
retry_delay_seconds: int,
target_key: str | None = None,
max_runs: int | None = None,
) -> None:
"""Run prospective RCON capture in a local loop."""
completed_runs = 0
print(
json.dumps(
{
"event": "rcon-historical-capture-loop-started",
"interval_seconds": interval_seconds,
"max_retries": max_retries,
"retry_delay_seconds": retry_delay_seconds,
"target_scope": target_key or "all-configured-rcon-targets",
},
indent=2,
)
)
print("Press Ctrl+C to stop.")
try:
while max_runs is None or completed_runs < max_runs:
completed_runs += 1
payload = _run_capture_with_retries(
max_retries=max_retries,
retry_delay_seconds=retry_delay_seconds,
target_key=target_key,
)
print(json.dumps({"run": completed_runs, **payload}, indent=2))
if max_runs is not None and completed_runs >= max_runs:
break
time.sleep(interval_seconds)
except KeyboardInterrupt:
print("\nRCON historical capture loop stopped by user.")
def _run_capture_with_retries(
*,
max_retries: int,
retry_delay_seconds: int,
target_key: str | None,
) -> dict[str, object]:
attempt = 0
while True:
attempt += 1
try:
return {
"status": "ok",
"attempts_used": attempt,
"capture_result": run_rcon_historical_capture(target_key=target_key),
}
except Exception as exc:
if attempt > max_retries:
return {
"status": "error",
"attempts_used": attempt,
"error": str(exc),
}
if retry_delay_seconds > 0:
time.sleep(retry_delay_seconds)
def _select_targets(target_key: str | None) -> list[object]:
configured_targets = list(load_rcon_targets())
if not configured_targets:
raise RuntimeError("No RCON targets configured in HLL_BACKEND_RCON_TARGETS.")
if target_key is None:
return configured_targets
normalized = target_key.strip()
selected = [
target
for target in configured_targets
if build_rcon_target_key(target) == normalized
]
if not selected:
raise ValueError(f"Unknown RCON target key: {target_key}")
return selected
def _serialize_target(target: object) -> dict[str, object]:
return {
"target_key": build_rcon_target_key(target),
"external_server_id": target.external_server_id,
"name": target.name,
"host": target.host,
"port": target.port,
"region": target.region,
"game_port": target.game_port,
"query_port": target.query_port,
"source_name": target.source_name,
}
def build_arg_parser() -> argparse.ArgumentParser:
"""Create the CLI parser for manual or periodic prospective RCON capture."""
parser = argparse.ArgumentParser(
description="Prospective RCON historical capture for HLL Vietnam.",
)
parser.add_argument(
"mode",
choices=("capture", "loop"),
help="capture runs once; loop keeps collecting periodically",
)
parser.add_argument(
"--target",
dest="target_key",
help="optional target key; defaults to all configured RCON targets",
)
parser.add_argument(
"--interval",
type=int,
default=get_rcon_historical_capture_interval_seconds(),
help="seconds to wait between loop runs",
)
parser.add_argument(
"--retries",
type=int,
default=get_rcon_historical_capture_max_retries(),
help="retry attempts after a failed capture",
)
parser.add_argument(
"--retry-delay",
type=int,
default=get_rcon_historical_capture_retry_delay_seconds(),
help="seconds to wait between failed attempts",
)
parser.add_argument(
"--max-runs",
type=int,
help="optional safety cap for loop mode",
)
return parser
def main(argv: Iterable[str] | None = None) -> int:
"""Run the prospective RCON historical capture CLI."""
parser = build_arg_parser()
args = parser.parse_args(list(argv) if argv is not None else None)
if args.mode == "capture":
result = run_rcon_historical_capture(target_key=args.target_key)
print(json.dumps(result, indent=2))
return 0
if args.interval <= 0:
raise ValueError("--interval must be a positive integer.")
if args.retries < 0:
raise ValueError("--retries must be zero or positive.")
if args.retry_delay < 0:
raise ValueError("--retry-delay must be zero or positive.")
if args.max_runs is not None and args.max_runs <= 0:
raise ValueError("--max-runs must be positive when provided.")
run_periodic_rcon_historical_capture(
interval_seconds=args.interval,
max_retries=args.retries,
retry_delay_seconds=args.retry_delay,
target_key=args.target_key,
max_runs=args.max_runs,
)
return 0
if __name__ == "__main__":
raise SystemExit(main())