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

Make backend RCON-first with safe fallback

Browse Source
This commit is contained in:
devRaGonSa
2026-03-25 14:11:52 +01:00
parent 4dc1be8261
commit 787e753f77
15 changed files with 884 additions and 261 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
backend/app/config.py
View File

@@ -10,8 +10,8 @@ DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 8000
DEFAULT_STORAGE_FILENAME = "hll_vietnam_dev.sqlite3"
DEFAULT_REFRESH_INTERVAL_SECONDS = 300
DEFAULT_LIVE_DATA_SOURCE = "a2s"
DEFAULT_HISTORICAL_DATA_SOURCE = "public-scoreboard"
DEFAULT_LIVE_DATA_SOURCE = "rcon"
DEFAULT_HISTORICAL_DATA_SOURCE = "rcon"
DEFAULT_RCON_TIMEOUT_SECONDS = 10.0
DEFAULT_HISTORICAL_CRCON_PAGE_SIZE = 50
DEFAULT_HISTORICAL_CRCON_TIMEOUT_SECONDS = 15.0
@@ -30,7 +30,7 @@ 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_INTERVAL_SECONDS = 600
DEFAULT_RCON_HISTORICAL_CAPTURE_MAX_RETRIES = 2
DEFAULT_RCON_HISTORICAL_CAPTURE_RETRY_DELAY_SECONDS = 15
DEFAULT_SQLITE_WRITER_TIMEOUT_SECONDS = 30.0

202
backend/app/data_sources.py
View File

@@ -18,6 +18,7 @@ from .server_targets import A2SServerTarget, load_a2s_targets
LIVE_SOURCE_A2S = "a2s"
SOURCE_KIND_PUBLIC_SCOREBOARD = "public-scoreboard"
SOURCE_KIND_RCON = "rcon"
@@ -50,7 +51,7 @@ class LiveDataSource(Protocol):
def collect_snapshots(self, *, persist: bool) -> dict[str, object]:
"""Collect one live snapshot batch."""
def build_target_index(self) -> dict[str | None, A2SServerTarget]:
def build_target_index(self) -> dict[str | None, object]:
"""Return optional server connection metadata keyed by external id."""
@@ -75,6 +76,102 @@ class A2SLiveDataSource:
}
@dataclass(frozen=True, slots=True)
class RconFirstLiveDataSource:
"""Live source arbitration with RCON as primary and A2S as controlled fallback."""
primary_source: RconLiveDataSource = RconLiveDataSource()
fallback_source: A2SLiveDataSource = A2SLiveDataSource()
source_kind: str = SOURCE_KIND_RCON
def collect_snapshots(self, *, persist: bool) -> dict[str, object]:
attempts: list[dict[str, object]] = []
fallback_reason: str | None = None
try:
primary_payload = self.primary_source.collect_snapshots(persist=persist)
except Exception as error: # noqa: BLE001 - source arbitration keeps fallback controlled
attempts.append(
build_source_attempt(
source=SOURCE_KIND_RCON,
role="primary",
status="error",
reason="rcon-live-request-failed",
message=str(error),
)
)
fallback_reason = "rcon-live-request-failed"
else:
primary_success_count = int(primary_payload.get("success_count") or 0)
primary_snapshots = list(primary_payload.get("snapshots") or [])
if primary_success_count > 0 and primary_snapshots:
attempts.append(
build_source_attempt(
source=SOURCE_KIND_RCON,
role="primary",
status="success",
)
)
return attach_source_policy(
primary_payload,
build_source_policy(
primary_source=SOURCE_KIND_RCON,
selected_source=SOURCE_KIND_RCON,
source_attempts=attempts,
),
)
attempts.append(
build_source_attempt(
source=SOURCE_KIND_RCON,
role="primary",
status="empty",
reason="rcon-live-returned-no-usable-snapshots",
message=f"success_count={primary_success_count}",
)
)
fallback_reason = "rcon-live-returned-no-usable-snapshots"
try:
fallback_payload = self.fallback_source.collect_snapshots(persist=persist)
except Exception as error: # noqa: BLE001 - keep combined failure explicit
attempts.append(
build_source_attempt(
source=LIVE_SOURCE_A2S,
role="fallback",
status="error",
reason="a2s-live-fallback-failed",
message=str(error),
)
)
raise RuntimeError(
"RCON-first live collection failed and A2S fallback also failed."
) from error
attempts.append(
build_source_attempt(
source=LIVE_SOURCE_A2S,
role="fallback",
status="success",
)
)
return attach_source_policy(
fallback_payload,
build_source_policy(
primary_source=SOURCE_KIND_RCON,
selected_source=LIVE_SOURCE_A2S,
fallback_used=True,
fallback_reason=fallback_reason,
source_attempts=attempts,
),
)
def build_target_index(self) -> dict[str | None, object]:
target_index = dict(self.fallback_source.build_target_index())
target_index.update(self.primary_source.build_target_index())
return target_index
@dataclass(frozen=True, slots=True)
class RconHistoricalDataSource:
"""Minimal persisted historical read model over prospective RCON capture."""
@@ -123,7 +220,7 @@ class RconHistoricalDataSource:
def get_historical_data_source() -> HistoricalDataSource:
"""Select the historical provider configured for the current environment."""
source_kind = get_historical_data_source_kind()
if source_kind == "public-scoreboard":
if source_kind == SOURCE_KIND_PUBLIC_SCOREBOARD:
return PublicScoreboardHistoricalDataSource()
if source_kind == SOURCE_KIND_RCON:
return RconHistoricalDataSource()
@@ -136,5 +233,104 @@ def get_live_data_source() -> LiveDataSource:
if source_kind == LIVE_SOURCE_A2S:
return A2SLiveDataSource()
if source_kind == SOURCE_KIND_RCON:
return RconLiveDataSource()
return RconFirstLiveDataSource()
raise ValueError(f"Unsupported live data source: {source_kind}")
def get_rcon_historical_read_model() -> RconHistoricalDataSource | None:
"""Return the minimal persisted RCON historical read model when selected."""
if get_historical_data_source_kind() != SOURCE_KIND_RCON:
return None
return RconHistoricalDataSource()
def resolve_historical_ingestion_data_source() -> tuple[HistoricalDataSource, dict[str, object]]:
"""Resolve the writer-oriented historical provider with safe fallback semantics."""
configured_kind = get_historical_data_source_kind()
if configured_kind == SOURCE_KIND_PUBLIC_SCOREBOARD:
return (
PublicScoreboardHistoricalDataSource(),
build_source_policy(
primary_source=SOURCE_KIND_PUBLIC_SCOREBOARD,
selected_source=SOURCE_KIND_PUBLIC_SCOREBOARD,
source_attempts=[
build_source_attempt(
source=SOURCE_KIND_PUBLIC_SCOREBOARD,
role="primary",
status="success",
)
],
),
)
if configured_kind == SOURCE_KIND_RCON:
return (
PublicScoreboardHistoricalDataSource(),
build_source_policy(
primary_source=SOURCE_KIND_RCON,
selected_source=SOURCE_KIND_PUBLIC_SCOREBOARD,
fallback_used=True,
fallback_reason="rcon-historical-ingestion-not-supported-yet",
source_attempts=[
build_source_attempt(
source=SOURCE_KIND_RCON,
role="primary",
status="unsupported",
reason="rcon-historical-ingestion-not-supported-yet",
),
build_source_attempt(
source=SOURCE_KIND_PUBLIC_SCOREBOARD,
role="fallback",
status="success",
),
],
),
)
raise ValueError(f"Unsupported historical data source: {configured_kind}")
def build_source_attempt(
*,
source: str,
role: str,
status: str,
reason: str | None = None,
message: str | None = None,
) -> dict[str, object]:
"""Build one normalized trace entry for source arbitration."""
return {
"source": source,
"role": role,
"status": status,
"reason": reason,
"message": message,
}
def build_source_policy(
*,
primary_source: str,
selected_source: str,
fallback_used: bool = False,
fallback_reason: str | None = None,
source_attempts: list[dict[str, object]] | None = None,
) -> dict[str, object]:
"""Build one small source-policy block for API responses and worker output."""
return {
"primary_source": primary_source,
"selected_source": selected_source,
"fallback_used": fallback_used,
"fallback_reason": fallback_reason,
"source_attempts": list(source_attempts or []),
}
def attach_source_policy(
payload: dict[str, object],
source_policy: dict[str, object],
) -> dict[str, object]:
"""Attach normalized source-policy metadata to an existing payload."""
enriched = dict(payload)
enriched.update(source_policy)
return enriched

5
backend/app/historical_ingestion.py
View File

@@ -12,7 +12,7 @@ from .config import (
get_historical_crcon_page_size,
get_historical_refresh_overlap_hours,
)
from .data_sources import HistoricalDataSource, get_historical_data_source
from .data_sources import HistoricalDataSource, resolve_historical_ingestion_data_source
from .historical_snapshots import generate_and_persist_historical_snapshots
from .historical_storage import (
finalize_backfill_progress,
@@ -119,7 +119,7 @@ def _run_ingestion(
) -> dict[str, object]:
initialize_historical_storage()
stats = IngestionStats()
data_source = get_historical_data_source()
data_source, source_policy = resolve_historical_ingestion_data_source()
selected_servers = _select_servers(server_slug)
processed_servers: list[dict[str, object]] = []
active_runs: dict[str, int] = {}
@@ -219,6 +219,7 @@ def _run_ingestion(
"status": "ok",
"mode": mode,
"source_provider": data_source.source_kind,
"source_policy": source_policy,
"page_size": page_size or get_historical_crcon_page_size(),
"start_page": start_page,
"detail_workers": detail_workers or get_historical_crcon_detail_workers(),

79
backend/app/historical_runner.py
View File

@@ -13,12 +13,14 @@ from .config import (
get_historical_refresh_interval_seconds,
get_historical_refresh_max_retries,
get_historical_refresh_retry_delay_seconds,
get_historical_data_source_kind,
)
from .historical_ingestion import run_incremental_refresh
from .historical_snapshots import (
generate_and_persist_historical_snapshots,
generate_and_persist_priority_historical_snapshots,
)
from .rcon_historical_worker import run_rcon_historical_capture
from .writer_lock import backend_writer_lock, build_writer_lock_holder
HOURLY_INTERVAL_SECONDS = 3600
@@ -95,20 +97,42 @@ def _run_refresh_with_retries(
f"app.historical_runner refresh:{server_slug or 'all-servers'}"
)
):
refresh_result = run_incremental_refresh(
server_slug=server_slug,
max_pages=max_pages,
page_size=page_size,
rebuild_snapshots=False,
)
snapshot_result = generate_historical_snapshots(
server_slug=server_slug,
run_number=run_number,
rcon_capture_result = _run_primary_rcon_capture()
should_run_classic_fallback, classic_fallback_reason = (
_resolve_classic_fallback_policy(
server_slug=server_slug,
run_number=run_number,
rcon_capture_result=rcon_capture_result,
)
)
if should_run_classic_fallback:
refresh_result = run_incremental_refresh(
server_slug=server_slug,
max_pages=max_pages,
page_size=page_size,
rebuild_snapshots=False,
)
snapshot_result = generate_historical_snapshots(
server_slug=server_slug,
run_number=run_number,
)
else:
refresh_result = {
"status": "skipped",
"reason": "rcon-primary-cycle-no-classic-fallback-needed",
}
snapshot_result = {
"status": "skipped",
"reason": "rcon-primary-cycle-no-classic-fallback-needed",
"generation_policy": "classic-historical-fallback-only",
}
return {
"status": "ok",
"attempts_used": attempt,
"max_retries": max_retries,
"rcon_capture_result": rcon_capture_result,
"classic_fallback_used": should_run_classic_fallback,
"classic_fallback_reason": classic_fallback_reason,
"refresh_result": refresh_result,
"snapshot_result": snapshot_result,
}
@@ -164,6 +188,43 @@ def _describe_snapshot_scope(server_slug: str | None) -> list[str]:
return [*DEFAULT_HISTORICAL_SERVER_SCOPE, "all-servers"]
def _run_primary_rcon_capture() -> dict[str, Any]:
if get_historical_data_source_kind() != "rcon":
return {
"status": "skipped",
"reason": "historical-data-source-configured-without-rcon-primary",
}
return run_rcon_historical_capture()
def _resolve_classic_fallback_policy(
*,
server_slug: str | None,
run_number: int,
rcon_capture_result: dict[str, Any],
) -> tuple[bool, str]:
if get_historical_data_source_kind() != "rcon":
return True, "public-scoreboard-configured-as-primary-historical-source"
if not _rcon_capture_has_usable_results(rcon_capture_result):
return True, "rcon-historical-capture-failed-or-returned-no-usable-targets"
if server_slug:
return True, "manual-server-scope-still-needs-classic-historical-fallback"
if run_number % get_historical_full_snapshot_every_runs() == 0:
return True, "periodic-classic-fallback-for-competitive-historical-coverage"
return False, "rcon-primary-cycle-succeeded-without-needing-classic-fallback"
def _rcon_capture_has_usable_results(rcon_capture_result: dict[str, Any]) -> bool:
if rcon_capture_result.get("status") != "ok":
return False
targets = rcon_capture_result.get("targets")
return isinstance(targets, list) and len(targets) > 0
def main() -> None:
"""Allow local scheduled historical refresh execution without external infra."""
parser = argparse.ArgumentParser(

432
backend/app/payloads.py
View File

@@ -9,7 +9,15 @@ from .config import (
get_live_data_source_kind,
get_refresh_interval_seconds,
)
from .data_sources import RconHistoricalDataSource, get_historical_data_source, get_live_data_source
from .data_sources import (
LIVE_SOURCE_A2S,
SOURCE_KIND_PUBLIC_SCOREBOARD,
SOURCE_KIND_RCON,
build_source_attempt,
build_source_policy,
get_live_data_source,
get_rcon_historical_read_model,
)
from .historical_snapshot_storage import get_historical_snapshot
from .historical_snapshots import (
DEFAULT_MONTHLY_SNAPSHOT_WINDOW,
@@ -29,6 +37,7 @@ from .historical_snapshots import (
)
from .historical_storage import (
ALL_SERVERS_SLUG,
DEFAULT_HISTORICAL_SERVERS,
get_historical_player_profile,
list_historical_server_summaries,
list_monthly_leaderboard,
@@ -102,21 +111,28 @@ def build_servers_payload() -> dict[str, object]:
max_snapshot_age_seconds,
)
refresh_errors: list[dict[str, object]] = []
refresh_source_policy = build_source_policy(
primary_source=get_live_data_source_kind(),
selected_source="none",
fallback_reason=None,
source_attempts=[],
)
if refresh_attempted:
refreshed_items, refresh_errors = _try_collect_real_time_snapshot()
refreshed_items, refresh_errors, refresh_source_policy = _try_collect_real_time_snapshot()
if refreshed_items:
refreshed_snapshot_at = _resolve_last_snapshot_at(refreshed_items)
refreshed_snapshot_age_seconds = _calculate_snapshot_age_seconds(refreshed_snapshot_at)
return _build_servers_response(
items=refreshed_items,
response_source="real-time-a2s-refresh",
response_source=_build_live_response_source(refresh_source_policy),
last_snapshot_at=refreshed_snapshot_at,
snapshot_age_seconds=refreshed_snapshot_age_seconds,
max_snapshot_age_seconds=max_snapshot_age_seconds,
refresh_attempted=True,
refresh_status="success",
refresh_errors=refresh_errors,
source_policy=refresh_source_policy,
)
if persisted_items:
@@ -133,6 +149,11 @@ def build_servers_payload() -> dict[str, object]:
refresh_attempted=refresh_attempted,
refresh_status=refresh_status,
refresh_errors=refresh_errors,
source_policy=_infer_live_source_policy_from_items(
persisted_items,
refresh_attempted=refresh_attempted,
refresh_errors=refresh_errors,
),
)
return {
@@ -150,6 +171,7 @@ def build_servers_payload() -> dict[str, object]:
"refresh_attempted": refresh_attempted,
"refresh_status": "failed" if refresh_attempted else "not-needed",
"refresh_errors": refresh_errors,
**refresh_source_policy,
"items": [],
},
}
@@ -219,14 +241,6 @@ 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",
@@ -239,6 +253,9 @@ def build_weekly_top_kills_payload(
"window_start": result["window_start"],
"window_end": result["window_end"],
"limit": limit,
**_resolve_historical_fallback_policy(
fallback_reason="rcon-historical-read-model-does-not-support-weekly-top-kills",
),
"items": result["items"],
},
}
@@ -252,16 +269,6 @@ 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)
@@ -301,6 +308,9 @@ def build_historical_leaderboard_payload(
"previous_month_closed_matches": result.get("previous_month_closed_matches"),
"sufficient_sample": result.get("sufficient_sample"),
"limit": limit,
**_resolve_historical_fallback_policy(
fallback_reason="rcon-historical-read-model-does-not-support-competitive-leaderboards",
),
"items": result["items"],
},
}
@@ -343,25 +353,37 @@ def build_recent_historical_matches_payload(
) -> 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):
data_source = get_rcon_historical_read_model()
if data_source is not None:
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,
},
}
if items:
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,
**build_source_policy(
primary_source=SOURCE_KIND_RCON,
selected_source=SOURCE_KIND_RCON,
source_attempts=[
build_source_attempt(
source=SOURCE_KIND_RCON,
role="primary",
status="success",
)
],
),
"items": items,
"capabilities": capabilities,
},
}
items = list_recent_historical_matches(limit=limit, server_slug=server_slug)
return {
"status": "ok",
@@ -371,6 +393,9 @@ def build_recent_historical_matches_payload(
"source": "historical-crcon-storage",
"limit": limit,
"server_slug": server_slug,
**_resolve_historical_fallback_policy(
fallback_reason="rcon-historical-read-model-has-no-recent-activity",
),
"items": items,
},
}
@@ -382,14 +407,6 @@ 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,
@@ -405,6 +422,9 @@ def build_monthly_mvp_payload(
),
"context": "historical-monthly-mvp",
"source": "historical-precomputed-snapshots",
**_resolve_historical_fallback_policy(
fallback_reason="rcon-historical-read-model-does-not-support-monthly-mvp-yet",
),
},
}
@@ -416,15 +436,6 @@ 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,
@@ -442,6 +453,9 @@ def build_player_event_payload(
),
"context": "historical-player-events",
"source": "historical-precomputed-player-event-snapshots",
**_resolve_historical_fallback_policy(
fallback_reason="rcon-historical-read-model-does-not-support-player-events-yet",
),
},
}
@@ -452,14 +466,6 @@ 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,
@@ -475,6 +481,9 @@ def build_monthly_mvp_v2_payload(
),
"context": "historical-monthly-mvp-v2",
"source": "historical-precomputed-snapshots",
**_resolve_historical_fallback_policy(
fallback_reason="rcon-historical-read-model-does-not-support-monthly-mvp-v2-yet",
),
},
}
@@ -484,13 +493,6 @@ 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,
@@ -506,6 +508,9 @@ def build_historical_server_summary_snapshot_payload(
"source": "historical-precomputed-snapshots",
"server_slug": server_slug,
"found": snapshot is not None and isinstance(item, dict),
**_resolve_historical_fallback_policy(
fallback_reason="rcon-historical-read-model-does-not-support-historical-snapshots-yet",
),
**_build_historical_snapshot_metadata(snapshot),
"item": item if isinstance(item, dict) else None,
},
@@ -520,16 +525,6 @@ 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
@@ -591,6 +586,9 @@ def build_leaderboard_snapshot_payload(
"sufficient_sample": payload.get("sufficient_sample") if isinstance(payload, dict) else None,
"snapshot_limit": payload.get("limit") if isinstance(payload, dict) else None,
"limit": limit,
**_resolve_historical_fallback_policy(
fallback_reason="rcon-historical-read-model-does-not-support-historical-snapshots-yet",
),
"items": sliced_items,
},
}
@@ -632,14 +630,6 @@ 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,
@@ -659,6 +649,9 @@ def build_recent_historical_matches_snapshot_payload(
**_build_historical_snapshot_metadata(snapshot),
"snapshot_limit": payload.get("limit") if isinstance(payload, dict) else None,
"limit": limit,
**_resolve_historical_fallback_policy(
fallback_reason="rcon-historical-read-model-does-not-support-historical-snapshots-yet",
),
"items": sliced_items,
},
}
@@ -670,14 +663,6 @@ 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,
@@ -723,6 +708,9 @@ def build_monthly_mvp_snapshot_payload(
),
"snapshot_limit": payload.get("limit") if isinstance(payload, dict) else None,
"limit": limit,
**_resolve_historical_fallback_policy(
fallback_reason="rcon-historical-read-model-does-not-support-historical-snapshots-yet",
),
"items": sliced_items,
},
}
@@ -734,14 +722,6 @@ 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,
@@ -789,6 +769,9 @@ def build_monthly_mvp_v2_snapshot_payload(
),
"snapshot_limit": payload.get("limit") if isinstance(payload, dict) else None,
"limit": limit,
**_resolve_historical_fallback_policy(
fallback_reason="rcon-historical-read-model-does-not-support-historical-snapshots-yet",
),
"items": sliced_items,
},
}
@@ -801,15 +784,6 @@ 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,
@@ -839,6 +813,9 @@ def build_player_event_snapshot_payload(
"month_key": payload.get("month_key") if isinstance(payload, dict) else None,
"snapshot_limit": payload.get("limit") if isinstance(payload, dict) else None,
"limit": limit,
**_resolve_historical_fallback_policy(
fallback_reason="rcon-historical-read-model-does-not-support-historical-snapshots-yet",
),
"items": sliced_items,
},
}
@@ -850,28 +827,43 @@ def build_historical_server_summary_payload(
) -> 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):
data_source = get_rcon_historical_read_model()
if data_source is not None:
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,
},
}
if items and any(
item.get("coverage", {}).get("status") != "empty"
for item in items
):
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,
**build_source_policy(
primary_source=SOURCE_KIND_RCON,
selected_source=SOURCE_KIND_RCON,
source_attempts=[
build_source_attempt(
source=SOURCE_KIND_RCON,
role="primary",
status="success",
)
],
),
"items": items,
"capabilities": capabilities,
},
}
items = list_historical_server_summaries(server_slug=server_slug)
return {
"status": "ok",
@@ -886,6 +878,9 @@ def build_historical_server_summary_payload(
"summary_basis": "persisted-import",
"weekly_ranking_window_days": 7,
"server_slug": server_slug,
**_resolve_historical_fallback_policy(
fallback_reason="rcon-historical-read-model-has-no-summary-coverage",
),
"items": items,
},
}
@@ -893,13 +888,6 @@ 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",
@@ -909,6 +897,9 @@ def build_historical_player_profile_payload(player_id: str) -> dict[str, object]
"source": "historical-crcon-storage",
"player_id": player_id,
"found": profile is not None,
**_resolve_historical_fallback_policy(
fallback_reason="rcon-historical-read-model-does-not-support-player-profile-yet",
),
"profile": profile,
},
}
@@ -931,45 +922,6 @@ 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 {
@@ -1087,11 +1039,14 @@ def _enrich_server_item(
) -> dict[str, object]:
enriched = dict(item)
enriched["current_map"] = normalize_map_name(enriched.get("current_map"))
history_url = _resolve_community_history_url(enriched.get("external_server_id"))
enriched["community_history_url"] = history_url
enriched["community_history_available"] = bool(history_url)
external_server_id = enriched.get("external_server_id")
snapshot_origin = enriched.get("snapshot_origin")
target = target_index.get(external_server_id)
if not target or snapshot_origin != "real-a2s":
if not target or snapshot_origin not in {"real-a2s", "real-rcon"}:
enriched["host"] = None
enriched["query_port"] = None
enriched["game_port"] = None
@@ -1129,12 +1084,26 @@ def _should_refresh_snapshot(
return snapshot_age_seconds > max_snapshot_age_seconds
def _try_collect_real_time_snapshot() -> tuple[list[dict[str, object]], list[dict[str, object]]]:
def _try_collect_real_time_snapshot() -> tuple[
list[dict[str, object]],
list[dict[str, object]],
dict[str, object],
]:
payload = get_live_data_source().collect_snapshots(persist=False)
snapshots = payload.get("snapshots")
items = _select_primary_snapshot_items(_enrich_server_items(list(snapshots or [])))
errors = payload.get("errors")
return items, list(errors or [])
return (
items,
list(errors or []),
{
"primary_source": payload.get("primary_source"),
"selected_source": payload.get("selected_source"),
"fallback_used": bool(payload.get("fallback_used")),
"fallback_reason": payload.get("fallback_reason"),
"source_attempts": list(payload.get("source_attempts") or []),
},
)
def _build_servers_response(
@@ -1147,6 +1116,7 @@ def _build_servers_response(
refresh_attempted: bool,
refresh_status: str,
refresh_errors: list[dict[str, object]],
source_policy: dict[str, object],
) -> dict[str, object]:
freshness = (
"fresh"
@@ -1168,6 +1138,7 @@ def _build_servers_response(
"refresh_attempted": refresh_attempted,
"refresh_status": refresh_status,
"refresh_errors": refresh_errors,
**source_policy,
"items": items,
},
}
@@ -1191,3 +1162,100 @@ def _to_snapshot_age_minutes(snapshot_age_seconds: int | None) -> int | None:
return None
return snapshot_age_seconds // 60
def _resolve_historical_fallback_policy(*, fallback_reason: str) -> dict[str, object]:
if get_historical_data_source_kind() != SOURCE_KIND_RCON:
return build_source_policy(
primary_source=SOURCE_KIND_PUBLIC_SCOREBOARD,
selected_source=SOURCE_KIND_PUBLIC_SCOREBOARD,
source_attempts=[
build_source_attempt(
source=SOURCE_KIND_PUBLIC_SCOREBOARD,
role="primary",
status="success",
)
],
)
return build_source_policy(
primary_source=SOURCE_KIND_RCON,
selected_source=SOURCE_KIND_PUBLIC_SCOREBOARD,
fallback_used=True,
fallback_reason=fallback_reason,
source_attempts=[
build_source_attempt(
source=SOURCE_KIND_RCON,
role="primary",
status="unsupported",
reason=fallback_reason,
),
build_source_attempt(
source=SOURCE_KIND_PUBLIC_SCOREBOARD,
role="fallback",
status="success",
),
],
)
def _infer_live_source_policy_from_items(
items: list[dict[str, object]],
*,
refresh_attempted: bool,
refresh_errors: list[dict[str, object]],
) -> dict[str, object]:
selected_source = "persisted-snapshot"
fallback_used = False
fallback_reason = None
snapshot_origins = {
str(item.get("snapshot_origin") or "").strip()
for item in items
if item.get("snapshot_origin")
}
if "real-rcon" in snapshot_origins:
selected_source = SOURCE_KIND_RCON
elif "real-a2s" in snapshot_origins:
selected_source = LIVE_SOURCE_A2S
if get_live_data_source_kind() == SOURCE_KIND_RCON:
fallback_used = True
fallback_reason = "persisted-live-snapshot-came-from-a2s"
attempt_status = "success" if items else ("error" if refresh_attempted else "cached")
attempt_reason = None if items else "no-live-snapshot-items"
if refresh_errors and attempt_reason is None:
attempt_reason = "live-refresh-errors-present"
return build_source_policy(
primary_source=get_live_data_source_kind(),
selected_source=selected_source,
fallback_used=fallback_used,
fallback_reason=fallback_reason,
source_attempts=[
build_source_attempt(
source=selected_source,
role="served-response",
status=attempt_status,
reason=attempt_reason,
)
],
)
def _build_live_response_source(source_policy: dict[str, object]) -> str:
selected_source = str(source_policy.get("selected_source") or "")
if selected_source == SOURCE_KIND_RCON:
return "real-time-rcon-refresh"
if selected_source == LIVE_SOURCE_A2S:
return "real-time-a2s-fallback"
return "real-time-refresh"
def _resolve_community_history_url(external_server_id: object) -> str | None:
normalized_server_id = str(external_server_id or "").strip()
if not normalized_server_id:
return None
for server in DEFAULT_HISTORICAL_SERVERS:
if server.slug == normalized_server_id:
return f"{server.scoreboard_base_url}/games"
return None

61
backend/app/player_event_source.py
View File

@@ -2,9 +2,16 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Protocol
from .config import get_historical_data_source_kind
from .data_sources import (
SOURCE_KIND_PUBLIC_SCOREBOARD,
SOURCE_KIND_RCON,
build_source_attempt,
build_source_policy,
)
from .player_event_models import PlayerEventRecord
from .providers.player_event_source_provider import PublicScoreboardPlayerEventSource
@@ -52,11 +59,53 @@ class RconPlayerEventSource:
}
def get_player_event_source() -> PlayerEventSource:
"""Select the event adapter that best matches the configured historical source."""
@dataclass(frozen=True, slots=True)
class PlayerEventSourceSelection:
"""Resolved player-event adapter plus source-policy metadata."""
source: PlayerEventSource
source_policy: dict[str, object]
def resolve_player_event_source() -> PlayerEventSourceSelection:
"""Select the event adapter with safe fallback when raw RCON events are unavailable."""
source_kind = get_historical_data_source_kind()
if source_kind == "public-scoreboard":
return PublicScoreboardPlayerEventSource()
if source_kind == "rcon":
return RconPlayerEventSource()
if source_kind == SOURCE_KIND_PUBLIC_SCOREBOARD:
return PlayerEventSourceSelection(
source=PublicScoreboardPlayerEventSource(),
source_policy=build_source_policy(
primary_source=SOURCE_KIND_PUBLIC_SCOREBOARD,
selected_source="public-scoreboard-match-summary",
source_attempts=[
build_source_attempt(
source=SOURCE_KIND_PUBLIC_SCOREBOARD,
role="primary",
status="success",
)
],
),
)
if source_kind == SOURCE_KIND_RCON:
return PlayerEventSourceSelection(
source=PublicScoreboardPlayerEventSource(),
source_policy=build_source_policy(
primary_source=SOURCE_KIND_RCON,
selected_source="public-scoreboard-match-summary",
fallback_used=True,
fallback_reason="rcon-player-events-not-implemented-yet",
source_attempts=[
build_source_attempt(
source=SOURCE_KIND_RCON,
role="primary",
status="unsupported",
reason="rcon-player-events-not-implemented-yet",
),
build_source_attempt(
source="public-scoreboard-match-summary",
role="fallback",
status="success",
),
],
),
)
raise ValueError(f"Unsupported player event source: {source_kind}")

11
backend/app/player_event_worker.py
View File

@@ -16,9 +16,9 @@ from .config import (
get_player_event_refresh_overlap_hours,
get_player_event_refresh_retry_delay_seconds,
)
from .data_sources import get_historical_data_source
from .data_sources import resolve_historical_ingestion_data_source
from .historical_storage import list_historical_servers
from .player_event_source import get_player_event_source
from .player_event_source import resolve_player_event_source
from .player_event_storage import (
finalize_player_event_ingestion_run,
finalize_player_event_progress,
@@ -62,8 +62,9 @@ def run_player_event_refresh(
)
):
initialize_player_event_storage()
data_source = get_historical_data_source()
event_source = get_player_event_source()
data_source, data_source_policy = resolve_historical_ingestion_data_source()
event_source_selection = resolve_player_event_source()
event_source = event_source_selection.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 = (
@@ -153,7 +154,9 @@ def run_player_event_refresh(
"status": "ok",
"mode": "refresh",
"source_provider": data_source.source_kind,
"source_policy": data_source_policy,
"event_adapter": event_source.source_kind,
"event_source_policy": event_source_selection.source_policy,
"page_size": resolved_page_size,
"detail_workers": resolved_detail_workers,
"overlap_hours": resolved_overlap_hours,