diff --git a/ai/tasks/done/TASK-070-data-source-abstraction-for-historical-and-live.md b/ai/tasks/done/TASK-070-data-source-abstraction-for-historical-and-live.md new file mode 100644 index 0000000..774a011 --- /dev/null +++ b/ai/tasks/done/TASK-070-data-source-abstraction-for-historical-and-live.md @@ -0,0 +1,75 @@ +# TASK-070-data-source-abstraction-for-historical-and-live + +## Goal +Introducir una capa de abstracción de fuente de datos para que el backend pueda trabajar con distintas fuentes según entorno sin cambiar el contrato de producto ni la UI. + +## Context +El proyecto hoy funciona con una fuente pública basada en scoreboard/CRCON público. Se quiere mantener ese modo para desarrollo, pero preparar producción para funcionar con RCON como fuente principal. Antes de implementar el proveedor RCON, hace falta definir una abstracción clara que desacople el backend de una única fuente. + +## Steps +1. Revisar la arquitectura actual de ingestión histórica, snapshots y panel live. +2. Identificar los puntos donde hoy se asume una fuente concreta. +3. Diseñar una interfaz o capa de proveedor de datos para cubrir como mínimo: + - histórico / ingestión + - estado actual live de servidores +4. Definir un contrato claro para proveedores de datos, por ejemplo: + - public-scoreboard provider + - rcon provider +5. Mantener intactos los contratos de salida del backend hacia la UI. +6. Preparar la selección del proveedor por configuración/entorno. +7. Documentar la arquitectura resultante. +8. No cambiar todavía el producto visible más de lo imprescindible. +9. Al completar la implementación: + - dejar el repositorio consistente + - hacer commit + - hacer push al remoto si el entorno lo permite + +## Files to Read First +- AGENTS.md +- ai/repo-context.md +- ai/architecture-index.md +- backend/README.md +- backend/app/config.py +- backend/app/historical_ingestion.py +- backend/app/historical_runner.py +- backend/app/payloads.py +- backend/app/routes.py +- backend/app/main.py + +## Expected Files to Modify +- backend/app/config.py +- backend/app/historical_ingestion.py +- backend/app/payloads.py +- opcionalmente nuevos módulos, por ejemplo: + - backend/app/data_sources.py + - backend/app/source_provider.py +- backend/README.md +- opcionalmente ai/architecture-index.md + +## Constraints +- No romper el modo actual de desarrollo. +- No acoplar la UI a una fuente concreta. +- No hacer cambios destructivos. +- Mantener el trabajo centrado en la abstracción de fuente de datos. + +## Validation +- Existe una abstracción clara de proveedor de datos. +- El backend puede seleccionar fuente por configuración. +- El flujo actual no se rompe. +- Los cambios quedan committeados y se hace push si el entorno lo permite. + +## Change Budget +- Preferir menos de 6 archivos modificados o creados. +- Preferir menos de 240 líneas cambiadas. + +## Outcome +- Se introdujo `backend/app/data_sources.py` como capa de seleccion y contrato para proveedores live e historicos. +- `historical_ingestion.py` dejo de depender directamente de requests CRCON embebidas y ahora resuelve su proveedor por configuracion. +- `payloads.py` dejo de invocar el colector A2S de forma acoplada y ahora consume un proveedor live seleccionado por entorno. +- Se mantuvieron intactos los contratos HTTP del backend hacia la UI. +- Quedaron preparadas las rutas de seleccion para `public-scoreboard`, `a2s` y el placeholder futuro `rcon`. + +## Validation Notes +- `python -m compileall backend/app` completo sin errores. +- `git diff --name-only` quedo limitado al alcance backend y documentacion de la task. +- El modo por defecto sigue apuntando a `a2s` para live y `public-scoreboard` para historico, por lo que el flujo de desarrollo no cambia. diff --git a/backend/README.md b/backend/README.md index af9f912..5bd5fac 100644 --- a/backend/README.md +++ b/backend/README.md @@ -64,6 +64,8 @@ Variables opcionales: - `HLL_BACKEND_PORT` - `HLL_BACKEND_ALLOWED_ORIGINS` - `HLL_BACKEND_REFRESH_INTERVAL_SECONDS` +- `HLL_BACKEND_LIVE_DATA_SOURCE` +- `HLL_BACKEND_HISTORICAL_DATA_SOURCE` - `HLL_HISTORICAL_CRCON_PAGE_SIZE` - `HLL_HISTORICAL_CRCON_TIMEOUT_SECONDS` - `HLL_HISTORICAL_CRCON_DETAIL_WORKERS` @@ -83,6 +85,8 @@ Variables especialmente relevantes para Docker y Compose: - `HLL_BACKEND_PORT` - `HLL_BACKEND_STORAGE_PATH` - `HLL_BACKEND_ALLOWED_ORIGINS` +- `HLL_BACKEND_LIVE_DATA_SOURCE` +- `HLL_BACKEND_HISTORICAL_DATA_SOURCE` - `HLL_HISTORICAL_CRCON_PAGE_SIZE` - `HLL_HISTORICAL_CRCON_TIMEOUT_SECONDS` - `HLL_HISTORICAL_CRCON_DETAIL_WORKERS` @@ -218,6 +222,46 @@ Los endpoints historicos leen la persistencia local SQLite creada por el colector. Si todavia no hay snapshots guardados, responden `status: "ok"` con `items: []` para mantener un contrato simple en desarrollo. +## Seleccion de fuente de datos + +El backend separa ahora la fuente de datos del contrato HTTP del producto. +Esto permite cambiar proveedores por entorno sin tocar `routes.py`, payloads de +UI ni el formato consumido por frontend. + +Variables nuevas: + +- `HLL_BACKEND_LIVE_DATA_SOURCE` +- `HLL_BACKEND_HISTORICAL_DATA_SOURCE` + +Valores soportados en esta fase: + +- live: + - `a2s` como modo actual de desarrollo + - `rcon` reservado para la futura integracion productiva +- historico: + - `public-scoreboard` como modo actual de desarrollo + - `rcon` reservado para la futura integracion productiva + +Defaults actuales: + +- `HLL_BACKEND_LIVE_DATA_SOURCE=a2s` +- `HLL_BACKEND_HISTORICAL_DATA_SOURCE=public-scoreboard` + +La seleccion efectiva se resuelve en `app/data_sources.py`: + +- `get_live_data_source()` entrega el proveedor usado por `payloads.py` + cuando `/api/servers` necesita un refresh real +- `get_historical_data_source()` entrega el proveedor usado por + `historical_ingestion.py` para bootstrap y refresh incremental + +En esta task solo queda implementado el proveedor ya operativo de desarrollo: + +- live `a2s` +- historico `public-scoreboard` + +La opcion `rcon` queda preparada como placeholder explicito y hoy responde con +error controlado si se selecciona antes de implementar su adapter. + ## Criterio de estructura - `__init__.py` declara el paquete `app` y reexporta las utilidades publicas @@ -227,6 +271,7 @@ colector. Si todavia no hay snapshots guardados, responden `status: "ok"` con - `a2s_client.py` encapsula una consulta minima A2S_INFO por UDP para probar servidores reales sin acoplar todavia el backend a una fuente mas compleja. - `config.py` centraliza host, puerto y allowlist minima de origenes locales. +- `data_sources.py` define los contratos y la seleccion por entorno para live e historico. - `historical_ingestion.py` consulta la capa JSON publica de CRCON para bootstrap y refresh incremental. - `historical_models.py` fija las entidades historicas minimas del dominio. - `historical_snapshots.py` fija los tipos y selectores validos de snapshots historicos precalculados. diff --git a/backend/app/config.py b/backend/app/config.py index 6324cb9..5e2219c 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -10,6 +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_HISTORICAL_CRCON_PAGE_SIZE = 50 DEFAULT_HISTORICAL_CRCON_TIMEOUT_SECONDS = 15.0 DEFAULT_HISTORICAL_CRCON_DETAIL_WORKERS = 8 @@ -164,6 +166,27 @@ def get_historical_refresh_interval_seconds() -> int: return interval_seconds +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() + if source_kind not in {"a2s", "rcon"}: + raise ValueError("HLL_BACKEND_LIVE_DATA_SOURCE must be 'a2s' or 'rcon'.") + return source_kind + + +def get_historical_data_source_kind() -> str: + """Return the historical provider kind selected for the current environment.""" + source_kind = os.getenv( + "HLL_BACKEND_HISTORICAL_DATA_SOURCE", + DEFAULT_HISTORICAL_DATA_SOURCE, + ).strip() + if source_kind not in {"public-scoreboard", "rcon"}: + raise ValueError( + "HLL_BACKEND_HISTORICAL_DATA_SOURCE must be 'public-scoreboard' or 'rcon'." + ) + return source_kind + + def get_historical_refresh_max_retries() -> int: """Return the retry count used by the historical refresh loop.""" configured_value = os.getenv( diff --git a/backend/app/data_sources.py b/backend/app/data_sources.py new file mode 100644 index 0000000..4e1aa46 --- /dev/null +++ b/backend/app/data_sources.py @@ -0,0 +1,257 @@ +"""Data source provider contracts for live and historical backend flows.""" + +from __future__ import annotations + +import json +import time +from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass +from typing import Protocol +from urllib.error import HTTPError, URLError +from urllib.parse import urlencode +from urllib.request import Request, urlopen + +from .collector import collect_server_snapshots +from .config import ( + get_historical_crcon_request_retries, + get_historical_crcon_request_timeout_seconds, + get_historical_crcon_retry_delay_seconds, + get_historical_data_source_kind, + get_live_data_source_kind, +) +from .server_targets import A2SServerTarget, load_a2s_targets + + +HISTORICAL_SOURCE_PUBLIC_SCOREBOARD = "public-scoreboard" +LIVE_SOURCE_A2S = "a2s" +SOURCE_KIND_RCON = "rcon" + +PUBLIC_INFO_ENDPOINT = "/api/get_public_info" +MATCH_LIST_ENDPOINT = "/api/get_scoreboard_maps" +MATCH_DETAIL_ENDPOINT = "/api/get_map_scoreboard" + + +class HistoricalDataSource(Protocol): + """Contract for historical providers used by ingestion flows.""" + + source_kind: str + + def fetch_public_info(self, *, base_url: str) -> dict[str, object]: + """Fetch provider metadata for one historical source.""" + + def fetch_match_page(self, *, base_url: str, page: int, limit: int) -> dict[str, object]: + """Fetch one page of historical matches.""" + + def fetch_match_details( + self, + *, + base_url: str, + match_ids: list[str], + max_workers: int, + ) -> list[dict[str, object]]: + """Fetch detailed payloads for one batch of matches.""" + + +class LiveDataSource(Protocol): + """Contract for live providers used by API payload builders.""" + + source_kind: str + + def collect_snapshots(self, *, persist: bool) -> dict[str, object]: + """Collect one live snapshot batch.""" + + def build_target_index(self) -> dict[str | None, A2SServerTarget]: + """Return optional server connection metadata keyed by external id.""" + + +@dataclass(frozen=True, slots=True) +class PublicScoreboardHistoricalDataSource: + """Historical provider backed by the public CRCON scoreboard JSON API.""" + + source_kind: str = HISTORICAL_SOURCE_PUBLIC_SCOREBOARD + + def fetch_public_info(self, *, base_url: str) -> dict[str, object]: + return self._fetch_dict_payload(base_url, PUBLIC_INFO_ENDPOINT) + + def fetch_match_page(self, *, base_url: str, page: int, limit: int) -> dict[str, object]: + return self._fetch_dict_payload( + base_url, + MATCH_LIST_ENDPOINT, + {"page": page, "limit": limit}, + context=f"page={page}", + ) + + def fetch_match_details( + self, + *, + base_url: str, + match_ids: list[str], + max_workers: int, + ) -> list[dict[str, object]]: + if not match_ids: + return [] + if max_workers <= 1: + return [ + self._fetch_match_detail(base_url=base_url, match_id=match_id) + for match_id in match_ids + ] + + with ThreadPoolExecutor(max_workers=max_workers) as executor: + futures = [ + executor.submit(self._fetch_match_detail, base_url=base_url, match_id=match_id) + for match_id in match_ids + ] + return [future.result() for future in futures] + + def _fetch_match_detail(self, *, base_url: str, match_id: str) -> dict[str, object]: + return self._fetch_dict_payload( + base_url, + MATCH_DETAIL_ENDPOINT, + {"map_id": match_id}, + context=f"match={match_id}", + ) + + def _fetch_json( + self, + *, + base_url: str, + endpoint: str, + query: dict[str, object] | None = None, + ) -> object: + url = f"{base_url}{endpoint}" + if query: + url = f"{url}?{urlencode(query)}" + + request = Request( + url, + headers={ + "Accept": "application/json", + "User-Agent": "HLL-Vietnam-Historical-Ingestion/0.1", + }, + ) + try: + with urlopen( + request, + timeout=get_historical_crcon_request_timeout_seconds(), + ) as response: + return json.loads(response.read().decode("utf-8")) + except HTTPError as exc: + raise RuntimeError(f"Historical provider request failed: {url} ({exc.code})") from exc + except URLError as exc: + raise RuntimeError(f"Historical provider request failed: {url} ({exc.reason})") from exc + + def _fetch_dict_payload( + self, + base_url: str, + endpoint: str, + query: dict[str, object] | None = None, + *, + context: str = "", + retries: int | None = None, + ) -> dict[str, object]: + resolved_retries = retries or get_historical_crcon_request_retries() + base_retry_delay_seconds = get_historical_crcon_retry_delay_seconds() + last_error: Exception | None = None + for attempt in range(1, resolved_retries + 1): + try: + payload = _unwrap_result( + self._fetch_json(base_url=base_url, endpoint=endpoint, query=query) + ) + except Exception as exc: # pragma: no cover - network path + last_error = exc + else: + if isinstance(payload, dict): + return payload + last_error = ValueError( + f"Unexpected payload type for {base_url}{endpoint} {context}".strip() + ) + + if attempt < resolved_retries: + time.sleep(base_retry_delay_seconds * attempt) + + assert last_error is not None + raise last_error + + +@dataclass(frozen=True, slots=True) +class A2SLiveDataSource: + """Live provider backed by the existing A2S collector flow.""" + + source_kind: str = LIVE_SOURCE_A2S + + def collect_snapshots(self, *, persist: bool) -> dict[str, object]: + return collect_server_snapshots( + source_mode="a2s", + allow_controlled_fallback=False, + persist=persist, + ) + + def build_target_index(self) -> dict[str | None, A2SServerTarget]: + return { + target.external_server_id: target + for target in load_a2s_targets() + if target.external_server_id + } + + +@dataclass(frozen=True, slots=True) +class RconHistoricalDataSource: + """Placeholder historical provider for future production RCON integration.""" + + 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.") + + def fetch_match_page(self, *, base_url: str, page: int, limit: int) -> dict[str, object]: + raise RuntimeError("Historical RCON provider is not implemented yet.") + + def fetch_match_details( + self, + *, + base_url: str, + match_ids: list[str], + max_workers: int, + ) -> list[dict[str, object]]: + raise RuntimeError("Historical RCON provider is not implemented yet.") + + +@dataclass(frozen=True, slots=True) +class RconLiveDataSource: + """Placeholder live provider for future production RCON integration.""" + + source_kind: str = SOURCE_KIND_RCON + + def collect_snapshots(self, *, persist: bool) -> dict[str, object]: + raise RuntimeError("Live RCON provider is not implemented yet.") + + def build_target_index(self) -> dict[str | None, A2SServerTarget]: + return {} + + +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 == HISTORICAL_SOURCE_PUBLIC_SCOREBOARD: + return PublicScoreboardHistoricalDataSource() + if source_kind == SOURCE_KIND_RCON: + return RconHistoricalDataSource() + raise ValueError(f"Unsupported historical data source: {source_kind}") + + +def get_live_data_source() -> LiveDataSource: + """Select the live provider configured for the current environment.""" + source_kind = get_live_data_source_kind() + if source_kind == LIVE_SOURCE_A2S: + return A2SLiveDataSource() + if source_kind == SOURCE_KIND_RCON: + return RconLiveDataSource() + raise ValueError(f"Unsupported live data source: {source_kind}") + + +def _unwrap_result(payload: object) -> object: + if not isinstance(payload, dict): + return payload + if "result" not in payload: + return payload + return payload.get("result") diff --git a/backend/app/historical_ingestion.py b/backend/app/historical_ingestion.py index 59b104a..83a37d4 100644 --- a/backend/app/historical_ingestion.py +++ b/backend/app/historical_ingestion.py @@ -4,21 +4,14 @@ from __future__ import annotations import argparse import json -import time -from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass from typing import Iterable -from urllib.error import HTTPError, URLError -from urllib.parse import urlencode -from urllib.request import Request, urlopen from .config import ( get_historical_crcon_detail_workers, get_historical_crcon_page_size, - get_historical_crcon_request_retries, - get_historical_crcon_request_timeout_seconds, - get_historical_crcon_retry_delay_seconds, ) +from .data_sources import HistoricalDataSource, get_historical_data_source from .historical_snapshots import generate_and_persist_historical_snapshots from .historical_storage import ( finalize_backfill_progress, @@ -35,11 +28,6 @@ from .historical_storage import ( ) -PUBLIC_INFO_ENDPOINT = "/api/get_public_info" -MATCH_LIST_ENDPOINT = "/api/get_scoreboard_maps" -MATCH_DETAIL_ENDPOINT = "/api/get_map_scoreboard" - - @dataclass(slots=True) class IngestionStats: """Mutable counters for one ingestion execution.""" @@ -115,6 +103,7 @@ def _run_ingestion( ) -> dict[str, object]: initialize_historical_storage() stats = IngestionStats() + data_source = get_historical_data_source() selected_servers = _select_servers(server_slug) processed_servers: list[dict[str, object]] = [] active_runs: dict[str, int] = {} @@ -143,6 +132,7 @@ def _run_ingestion( mode=mode, run_id=run_id, stats=stats, + data_source=data_source, max_pages=max_pages, page_size=page_size, start_page=resolved_start_page, @@ -202,6 +192,7 @@ def _run_ingestion( return { "status": "ok", "mode": mode, + "source_provider": data_source.source_kind, "page_size": page_size or get_historical_crcon_page_size(), "start_page": start_page, "detail_workers": detail_workers or get_historical_crcon_detail_workers(), @@ -225,6 +216,7 @@ def _ingest_server( mode: str, run_id: int, stats: IngestionStats, + data_source: HistoricalDataSource, max_pages: int | None, page_size: int | None, start_page: int, @@ -236,14 +228,14 @@ def _ingest_server( page_limit = max_pages or 1000000 start_page = max(1, start_page) local_stats = IngestionStats() - public_info = _fetch_public_info(str(server["scoreboard_base_url"])) + public_info = data_source.fetch_public_info(base_url=str(server["scoreboard_base_url"])) discovered_total_matches: int | None = None last_page_processed: int | None = None archive_exhausted = False for page_number in range(start_page, start_page + page_limit): - payload = _fetch_match_page( - str(server["scoreboard_base_url"]), + payload = data_source.fetch_match_page( + base_url=str(server["scoreboard_base_url"]), page=page_number, limit=resolved_page_size, ) @@ -273,9 +265,9 @@ def _ingest_server( if match_id: match_ids_to_fetch.append(match_id) - for detail_payload in _fetch_match_details( - str(server["scoreboard_base_url"]), - match_ids_to_fetch, + for detail_payload in data_source.fetch_match_details( + base_url=str(server["scoreboard_base_url"]), + match_ids=match_ids_to_fetch, max_workers=resolved_detail_workers, ): delta = upsert_historical_match( @@ -301,6 +293,7 @@ def _ingest_server( "server_slug": server["slug"], "public_name": _extract_public_name(public_info), "server_number": public_info.get("server_number") or server.get("server_number"), + "source_provider": data_source.source_kind, "pages_processed": local_stats.pages_processed, "matches_seen": local_stats.matches_seen, "discovered_total_matches": discovered_total_matches, @@ -340,122 +333,12 @@ def _select_servers(server_slug: str | None) -> list[dict[str, object]]: return selected -def _fetch_public_info(base_url: str) -> dict[str, object]: - return _fetch_dict_payload(base_url, PUBLIC_INFO_ENDPOINT) - - -def _fetch_match_page(base_url: str, *, page: int, limit: int) -> dict[str, object]: - return _fetch_dict_payload( - base_url, - MATCH_LIST_ENDPOINT, - {"page": page, "limit": limit}, - context=f"page={page}", - ) - - -def _fetch_match_detail(base_url: str, *, match_id: str) -> dict[str, object]: - return _fetch_dict_payload( - base_url, - MATCH_DETAIL_ENDPOINT, - {"map_id": match_id}, - context=f"match={match_id}", - ) - - -def _fetch_match_details( - base_url: str, - match_ids: list[str], - *, - max_workers: int, -) -> list[dict[str, object]]: - if not match_ids: - return [] - if max_workers <= 1: - return [ - _fetch_match_detail(base_url, match_id=match_id) - for match_id in match_ids - ] - - with ThreadPoolExecutor(max_workers=max_workers) as executor: - futures = [ - executor.submit(_fetch_match_detail, base_url, match_id=match_id) - for match_id in match_ids - ] - return [future.result() for future in futures] - - -def _fetch_json( - base_url: str, - endpoint: str, - query: dict[str, object] | None = None, -) -> object: - url = f"{base_url}{endpoint}" - if query: - url = f"{url}?{urlencode(query)}" - - request = Request( - url, - headers={ - "Accept": "application/json", - "User-Agent": "HLL-Vietnam-Historical-Ingestion/0.1", - }, - ) - try: - with urlopen( - request, - timeout=get_historical_crcon_request_timeout_seconds(), - ) as response: - return json.loads(response.read().decode("utf-8")) - except HTTPError as exc: - raise RuntimeError(f"Historical CRCON request failed: {url} ({exc.code})") from exc - except URLError as exc: - raise RuntimeError(f"Historical CRCON request failed: {url} ({exc.reason})") from exc - - -def _fetch_dict_payload( - base_url: str, - endpoint: str, - query: dict[str, object] | None = None, - *, - context: str = "", - retries: int | None = None, -) -> dict[str, object]: - resolved_retries = retries or get_historical_crcon_request_retries() - base_retry_delay_seconds = get_historical_crcon_retry_delay_seconds() - last_error: Exception | None = None - for attempt in range(1, resolved_retries + 1): - try: - payload = _unwrap_result(_fetch_json(base_url, endpoint, query)) - except Exception as exc: # pragma: no cover - network path - last_error = exc - else: - if isinstance(payload, dict): - return payload - last_error = ValueError( - f"Unexpected payload type for {base_url}{endpoint} {context}".strip() - ) - - if attempt < resolved_retries: - time.sleep(base_retry_delay_seconds * attempt) - - assert last_error is not None - raise last_error - - def _coerce_match_list(payload: object) -> list[dict[str, object]]: if not isinstance(payload, list): return [] return [item for item in payload if isinstance(item, dict)] -def _unwrap_result(payload: object) -> object: - if not isinstance(payload, dict): - return payload - if "result" not in payload: - return payload - return payload.get("result") - - def _pick_match_timestamp(match_payload: dict[str, object]) -> str | None: for key in ("end", "start", "creation_time"): value = match_payload.get(key) diff --git a/backend/app/payloads.py b/backend/app/payloads.py index e076681..2d0f1c4 100644 --- a/backend/app/payloads.py +++ b/backend/app/payloads.py @@ -4,8 +4,8 @@ from __future__ import annotations from datetime import datetime, timezone -from .collector import collect_server_snapshots from .config import get_refresh_interval_seconds +from .data_sources import get_live_data_source from .historical_snapshot_storage import get_historical_snapshot from .historical_snapshots import ( DEFAULT_MONTHLY_SNAPSHOT_WINDOW, @@ -27,7 +27,6 @@ from .historical_storage import ( list_weekly_top_kills, ) from .normalizers import normalize_map_name -from .server_targets import load_a2s_targets from .storage import list_latest_snapshots, list_server_history, list_snapshot_history @@ -679,11 +678,7 @@ def _build_monthly_mvp_title(*, is_all_servers: bool, snapshot: bool = False) -> def _enrich_server_items(items: list[dict[str, object]]) -> list[dict[str, object]]: - target_index = { - target.external_server_id: target - for target in load_a2s_targets() - if target.external_server_id - } + target_index = get_live_data_source().build_target_index() enriched_items: list[dict[str, object]] = [] for item in items: enriched_items.append(_enrich_server_item(item, target_index)) @@ -748,11 +743,7 @@ def _should_refresh_snapshot( def _try_collect_real_time_snapshot() -> tuple[list[dict[str, object]], list[dict[str, object]]]: - payload = collect_server_snapshots( - source_mode="a2s", - allow_controlled_fallback=False, - persist=True, - ) + payload = get_live_data_source().collect_snapshots(persist=True) snapshots = payload.get("snapshots") items = _select_primary_snapshot_items(_enrich_server_items(list(snapshots or []))) errors = payload.get("errors")