diff --git a/ai/tasks/done/TASK-043-historical-snapshots-generation-runner.md b/ai/tasks/done/TASK-043-historical-snapshots-generation-runner.md new file mode 100644 index 0000000..9280d32 --- /dev/null +++ b/ai/tasks/done/TASK-043-historical-snapshots-generation-runner.md @@ -0,0 +1,59 @@ +# TASK-043-historical-snapshots-generation-runner + +## Goal +Implementar un generador/runner de snapshots históricos que calcule y refresque periódicamente resumen, tops y partidas recientes para cada servidor de la comunidad. + +## Context +Una vez existe almacenamiento de snapshots, hace falta una capa operativa que genere esos snapshots a partir del histórico persistido. El objetivo es que la página histórica no dependa de cálculos pesados en tiempo real, sino de datos precalculados actualizados periódicamente. + +## Steps +1. Revisar los agregados históricos actuales y la nueva capa de snapshots. +2. Implementar un runner o flujo de generación que produzca snapshots para cada servidor. +3. Generar, como mínimo: + - resumen de servidor + - top kills 7d + - top muertes 7d + - top partidas con más de 100 kills por jugador 7d + - top puntos de soporte 7d + - partidas recientes +4. Añadir metadatos de generación y estado. +5. Definir una frecuencia razonable de refresco, por ejemplo 10 o 15 minutos. +6. Documentar cómo ejecutar el runner manualmente y cómo operarlo de forma periódica. +7. No migrar todavía toda la UI en esta task. +8. 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 +- backend/README.md +- backend/app/config.py +- backend/app/historical_storage.py +- backend/app/historical_snapshots.py +- backend/app/historical_snapshot_storage.py +- backend/app/historical_runner.py + +## Expected Files to Modify +- backend/README.md +- backend/app/config.py +- backend/app/historical_runner.py +- backend/app/historical_snapshots.py +- backend/app/historical_snapshot_storage.py +- opcionalmente nuevos módulos auxiliares de generación + +## Constraints +- No romper el live status. +- No crear UI nueva en esta task. +- No hacer cambios destructivos. +- Mantener el trabajo centrado en generación periódica de snapshots. + +## Validation +- Existe un runner que genera snapshots de resumen, tops y partidas recientes. +- Los snapshots se pueden refrescar periódicamente. +- La documentación explica cómo ejecutarlo. +- Los cambios quedan committeados y se hace push al remoto si el entorno lo permite. + +## Change Budget +- Preferir menos de 6 archivos modificados o creados. +- Preferir menos de 240 líneas cambiadas. diff --git a/backend/README.md b/backend/README.md index ade7aaf..56a487b 100644 --- a/backend/README.md +++ b/backend/README.md @@ -70,6 +70,7 @@ Variables opcionales: - `HLL_HISTORICAL_CRCON_REQUEST_RETRIES` - `HLL_HISTORICAL_CRCON_RETRY_DELAY_SECONDS` - `HLL_HISTORICAL_REFRESH_INTERVAL_SECONDS` +- `HLL_HISTORICAL_SNAPSHOT_REFRESH_INTERVAL_SECONDS` - `HLL_HISTORICAL_REFRESH_MAX_RETRIES` - `HLL_HISTORICAL_REFRESH_RETRY_DELAY_SECONDS` @@ -558,23 +559,34 @@ Los reintentos de cada request JSON pueden ajustarse sin tocar codigo con: - `HLL_HISTORICAL_CRCON_RETRY_DELAY_SECONDS` El runner `python -m app.historical_runner` deja ese refresh incremental listo -para ejecucion local repetida sin depender de infraestructura externa. Por +para ejecucion local repetida sin depender de infraestructura externa y +regenera snapshots historicos precalculados tras cada refresh correcto. Por defecto: -- refresca cada `1800` segundos +- refresca y recompone snapshots cada `900` segundos - reintenta hasta `2` veces tras un fallo - espera `30` segundos entre reintentos - reutiliza el registro de `historical_ingestion_runs` para dejar trazabilidad de ultimo refresh, resultado y errores basicos +- persiste por servidor: + - `server-summary` + - `weekly-leaderboard` para `kills`, `deaths`, `support` y `matches_over_100_kills` + - `recent-matches` Flags utiles del runner: - `--server comunidad-hispana-01` para limitar a un servidor -- `--interval 900` para aumentar frecuencia local +- `--interval 900` para fijar la frecuencia recomendada de snapshots - `--retries 1` para reducir reintentos - `--retry-delay 10` para bajar la espera entre fallos - `--max-runs 1` para una validacion puntual sin bucle indefinido +Variables utiles del runner: + +- `HLL_HISTORICAL_SNAPSHOT_REFRESH_INTERVAL_SECONDS` +- `HLL_HISTORICAL_REFRESH_MAX_RETRIES` +- `HLL_HISTORICAL_REFRESH_RETRY_DELAY_SECONDS` + Al inicializar la persistencia local, el backend normaliza tambien la identidad historica ya guardada: diff --git a/backend/app/config.py b/backend/app/config.py index 2d6bcce..390cb42 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -16,6 +16,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_SNAPSHOT_REFRESH_INTERVAL_SECONDS = 900 DEFAULT_HISTORICAL_REFRESH_MAX_RETRIES = 2 DEFAULT_HISTORICAL_REFRESH_RETRY_DELAY_SECONDS = 30 DEFAULT_ALLOWED_ORIGINS = ( @@ -145,12 +146,17 @@ def get_historical_crcon_retry_delay_seconds() -> float: def get_historical_refresh_interval_seconds() -> int: """Return the default interval used by the historical refresh loop.""" configured_value = os.getenv( - "HLL_HISTORICAL_REFRESH_INTERVAL_SECONDS", - str(DEFAULT_HISTORICAL_REFRESH_INTERVAL_SECONDS), + "HLL_HISTORICAL_SNAPSHOT_REFRESH_INTERVAL_SECONDS", + os.getenv( + "HLL_HISTORICAL_REFRESH_INTERVAL_SECONDS", + str(DEFAULT_HISTORICAL_SNAPSHOT_REFRESH_INTERVAL_SECONDS), + ), ) interval_seconds = int(configured_value) if interval_seconds <= 0: - raise ValueError("HLL_HISTORICAL_REFRESH_INTERVAL_SECONDS must be positive.") + raise ValueError( + "HLL_HISTORICAL_SNAPSHOT_REFRESH_INTERVAL_SECONDS must be positive." + ) return interval_seconds diff --git a/backend/app/historical_runner.py b/backend/app/historical_runner.py index 17eb7b0..ac18ae2 100644 --- a/backend/app/historical_runner.py +++ b/backend/app/historical_runner.py @@ -5,6 +5,7 @@ from __future__ import annotations import argparse import json import time +from datetime import datetime, timezone from typing import Any from .config import ( @@ -13,6 +14,8 @@ from .config import ( get_historical_refresh_retry_delay_seconds, ) from .historical_ingestion import run_incremental_refresh +from .historical_snapshot_storage import persist_historical_snapshot_batch +from .historical_snapshots import build_all_historical_snapshots def run_periodic_historical_refresh( @@ -25,7 +28,7 @@ def run_periodic_historical_refresh( page_size: int | None = None, max_runs: int | None = None, ) -> None: - """Run periodic historical refreshes until interrupted or the limit is reached.""" + """Run periodic historical refreshes and rebuild persisted snapshots.""" completed_runs = 0 print( "Starting historical refresh loop " @@ -65,16 +68,18 @@ def _run_refresh_with_retries( while True: attempt += 1 try: - result = run_incremental_refresh( + refresh_result = run_incremental_refresh( server_slug=server_slug, max_pages=max_pages, page_size=page_size, ) + snapshot_result = generate_historical_snapshots(server_slug=server_slug) return { "status": "ok", "attempts_used": attempt, "max_retries": max_retries, - "result": result, + "refresh_result": refresh_result, + "snapshot_result": snapshot_result, } except Exception as exc: if attempt > max_retries: @@ -88,16 +93,42 @@ def _run_refresh_with_retries( time.sleep(retry_delay_seconds) +def generate_historical_snapshots( + *, + server_slug: str | None = None, +) -> dict[str, Any]: + """Build and persist precomputed snapshots for one server or all servers.""" + generated_at = datetime.now(timezone.utc) + snapshots = build_all_historical_snapshots( + server_key=server_slug, + generated_at=generated_at, + ) + persisted_records = persist_historical_snapshot_batch(snapshots) + snapshots_by_server: dict[str, int] = {} + for record in persisted_records: + snapshots_by_server.setdefault(record.server_key, 0) + snapshots_by_server[record.server_key] += 1 + + return { + "generated_at": generated_at.isoformat().replace("+00:00", "Z"), + "server_slug": server_slug, + "snapshot_count": len(persisted_records), + "servers_processed": len(snapshots_by_server), + "snapshots_by_server": snapshots_by_server, + "refresh_interval_seconds": get_historical_refresh_interval_seconds(), + } + + def main() -> None: """Allow local scheduled historical refresh execution without external infra.""" parser = argparse.ArgumentParser( - description="Run periodic historical CRCON refreshes for HLL Vietnam.", + description="Run periodic historical refreshes and regenerate snapshots for HLL Vietnam.", ) parser.add_argument( "--interval", type=int, default=get_historical_refresh_interval_seconds(), - help="Seconds to wait between incremental refresh runs.", + help="Seconds to wait between refresh-plus-snapshot runs.", ) parser.add_argument( "--retries", diff --git a/backend/app/historical_snapshot_storage.py b/backend/app/historical_snapshot_storage.py index 225d625..77a980c 100644 --- a/backend/app/historical_snapshot_storage.py +++ b/backend/app/historical_snapshot_storage.py @@ -123,6 +123,31 @@ def persist_historical_snapshot( ) +def persist_historical_snapshot_batch( + snapshots: list[dict[str, object]], + *, + db_path: Path | None = None, +) -> list[HistoricalSnapshotRecord]: + """Persist a batch of snapshots generated in one runner cycle.""" + records: list[HistoricalSnapshotRecord] = [] + for snapshot in snapshots: + records.append( + persist_historical_snapshot( + server_key=str(snapshot["server_key"]), + snapshot_type=str(snapshot["snapshot_type"]), + payload=snapshot["payload"], + metric=snapshot.get("metric"), + window=snapshot.get("window"), + generated_at=snapshot.get("generated_at"), + source_range_start=snapshot.get("source_range_start"), + source_range_end=snapshot.get("source_range_end"), + is_stale=bool(snapshot.get("is_stale", False)), + db_path=db_path, + ) + ) + return records + + def get_historical_snapshot( *, server_key: str, diff --git a/backend/app/historical_snapshots.py b/backend/app/historical_snapshots.py index a078dc3..a7c1bbb 100644 --- a/backend/app/historical_snapshots.py +++ b/backend/app/historical_snapshots.py @@ -2,6 +2,15 @@ from __future__ import annotations +from datetime import datetime, timezone +from pathlib import Path + +from .historical_storage import ( + list_historical_server_summaries, + list_historical_servers, + list_recent_historical_matches, + list_weekly_leaderboard, +) SNAPSHOT_TYPE_SERVER_SUMMARY = "server-summary" SNAPSHOT_TYPE_WEEKLY_LEADERBOARD = "weekly-leaderboard" @@ -23,9 +32,17 @@ SUPPORTED_LEADERBOARD_METRICS = frozenset( "matches_over_100_kills", } ) +SNAPSHOT_LEADERBOARD_METRICS = ( + "kills", + "deaths", + "matches_over_100_kills", + "support", +) DEFAULT_SNAPSHOT_WINDOW = "all-time" DEFAULT_WEEKLY_SNAPSHOT_WINDOW = "7d" +DEFAULT_WEEKLY_LEADERBOARD_LIMIT = 10 +DEFAULT_RECENT_MATCHES_LIMIT = 20 def validate_snapshot_identity( @@ -44,3 +61,184 @@ def validate_snapshot_identity( if metric is not None: raise ValueError(f"Metric is only supported for {SNAPSHOT_TYPE_WEEKLY_LEADERBOARD}.") + + +def list_snapshot_server_keys(*, db_path: Path | None = None) -> list[str]: + """Return the historical server slugs that should receive persisted snapshots.""" + return [ + str(item["slug"]) + for item in list_historical_servers(db_path=db_path) + if item.get("slug") + ] + + +def build_historical_server_snapshots( + *, + server_key: str, + generated_at: datetime | None = None, + leaderboard_limit: int = DEFAULT_WEEKLY_LEADERBOARD_LIMIT, + recent_matches_limit: int = DEFAULT_RECENT_MATCHES_LIMIT, + db_path: Path | None = None, +) -> list[dict[str, object]]: + """Build all precomputed historical snapshots required for one server.""" + generated_at_value = _as_utc(generated_at or datetime.now(timezone.utc)) + snapshots = [_build_server_summary_snapshot(server_key, generated_at_value, db_path=db_path)] + + for metric in SNAPSHOT_LEADERBOARD_METRICS: + snapshots.append( + _build_weekly_leaderboard_snapshot( + server_key, + metric, + generated_at_value, + limit=leaderboard_limit, + db_path=db_path, + ) + ) + + snapshots.append( + _build_recent_matches_snapshot( + server_key, + generated_at_value, + limit=recent_matches_limit, + db_path=db_path, + ) + ) + return snapshots + + +def build_all_historical_snapshots( + *, + server_key: str | None = None, + generated_at: datetime | None = None, + leaderboard_limit: int = DEFAULT_WEEKLY_LEADERBOARD_LIMIT, + recent_matches_limit: int = DEFAULT_RECENT_MATCHES_LIMIT, + db_path: Path | None = None, +) -> list[dict[str, object]]: + """Build the full snapshot set for one server or for all configured servers.""" + target_server_keys = [server_key] if server_key else list_snapshot_server_keys(db_path=db_path) + snapshots: list[dict[str, object]] = [] + for target_server_key in target_server_keys: + snapshots.extend( + build_historical_server_snapshots( + server_key=target_server_key, + generated_at=generated_at, + leaderboard_limit=leaderboard_limit, + recent_matches_limit=recent_matches_limit, + db_path=db_path, + ) + ) + return snapshots + + +def _build_server_summary_snapshot( + server_key: str, + generated_at: datetime, + *, + db_path: Path | None = None, +) -> dict[str, object]: + summary_items = list_historical_server_summaries(server_slug=server_key, db_path=db_path) + summary_item = summary_items[0] if summary_items else {} + time_range = summary_item.get("time_range") if isinstance(summary_item, dict) else {} + return { + "server_key": server_key, + "snapshot_type": SNAPSHOT_TYPE_SERVER_SUMMARY, + "metric": None, + "window": DEFAULT_SNAPSHOT_WINDOW, + "generated_at": generated_at, + "source_range_start": _parse_optional_timestamp(time_range.get("start")), + "source_range_end": _parse_optional_timestamp(time_range.get("end")), + "is_stale": False, + "payload": { + "server_key": server_key, + "generated_at": _to_iso(generated_at), + "item": summary_item, + }, + } + + +def _build_weekly_leaderboard_snapshot( + server_key: str, + metric: str, + generated_at: datetime, + *, + limit: int, + db_path: Path | None = None, +) -> dict[str, object]: + leaderboard_result = list_weekly_leaderboard( + limit=limit, + server_id=server_key, + metric=metric, + db_path=db_path, + ) + return { + "server_key": server_key, + "snapshot_type": SNAPSHOT_TYPE_WEEKLY_LEADERBOARD, + "metric": metric, + "window": DEFAULT_WEEKLY_SNAPSHOT_WINDOW, + "generated_at": generated_at, + "source_range_start": _parse_optional_timestamp(leaderboard_result.get("window_start")), + "source_range_end": _parse_optional_timestamp(leaderboard_result.get("window_end")), + "is_stale": False, + "payload": { + "server_key": server_key, + "metric": metric, + "limit": limit, + "generated_at": _to_iso(generated_at), + **leaderboard_result, + }, + } + + +def _build_recent_matches_snapshot( + server_key: str, + generated_at: datetime, + *, + limit: int, + db_path: Path | None = None, +) -> dict[str, object]: + items = list_recent_historical_matches( + limit=limit, + server_slug=server_key, + db_path=db_path, + ) + closed_points = [ + _parse_optional_timestamp(item.get("closed_at")) + for item in items + if isinstance(item, dict) and item.get("closed_at") + ] + return { + "server_key": server_key, + "snapshot_type": SNAPSHOT_TYPE_RECENT_MATCHES, + "metric": None, + "window": DEFAULT_SNAPSHOT_WINDOW, + "generated_at": generated_at, + "source_range_start": min(closed_points) if closed_points else None, + "source_range_end": max(closed_points) if closed_points else None, + "is_stale": False, + "payload": { + "server_key": server_key, + "limit": limit, + "generated_at": _to_iso(generated_at), + "items": items, + }, + } + + +def _parse_optional_timestamp(value: object) -> datetime | None: + if not isinstance(value, str) or not value.strip(): + return None + normalized = value.strip().replace("Z", "+00:00") + parsed = datetime.fromisoformat(normalized) + if parsed.tzinfo is None: + return parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + + +def _as_utc(value: datetime) -> datetime: + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + + +def _to_iso(value: datetime) -> str: + return _as_utc(value).isoformat().replace("+00:00", "Z")