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

Complete TASK-070 data source abstraction

Browse Source
This commit is contained in:
devRaGonSa
2026-03-24 10:34:47 +01:00
parent f58ba29054
commit 784fc45034
6 changed files with 415 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
backend/app/config.py
View File

@@ -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(

257
backend/app/data_sources.py Normal file
View File

@@ -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")

141
backend/app/historical_ingestion.py
View File

@@ -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)

15
backend/app/payloads.py
View File

@@ -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")