Add annual ranking snapshot generator

ai/tasks/done/TASK-168-add-annual-ranking-snapshot-generator.md

@@ -0,0 +1,103 @@
+---
+id: TASK-168
+title: Add annual ranking snapshot generator
+status: done
+type: backend
+team: Arquitecto de Base de Datos
+supporting_teams:
+  - Backend Senior
+  - Arquitecto Python
+roadmap_item: foundation
+priority: medium
+---
+
+# TASK-168 - Add annual ranking snapshot generator
+
+## Goal
+
+Implementar un generador backend para snapshots anuales top 20 de jugadores sobre datos RCON materializados.
+
+## Context
+
+TASK-166 definió el diseño de persistencia, y TASK-167 ya añadió las tablas:
+
+- `rcon_annual_ranking_snapshots`
+- `rcon_annual_ranking_snapshot_items`
+
+Ahora se necesita el generador que rellene esas tablas desde
+`rcon_materialized_matches` y `rcon_match_player_stats` con `source_basis = admin-log-match-ended`.
+
+Debe mantenerse separado de frontend y sin introducir una API pública todavía.
+
+## Files to Read First
+
+- `AGENTS.md`
+- `ai/repo-context.md`
+- `ai/architecture-index.md`
+- `docs/annual-ranking-snapshot-schema-plan.md`
+- `backend/app/rcon_admin_log_materialization.py`
+- `backend/app/rcon_historical_leaderboards.py`
+- `backend/app/rcon_historical_player_stats.py`
+- `backend/app/postgres_rcon_storage.py`
+- `backend/app/sqlite_to_postgres_migration.py`
+
+## Expected Files to Modify
+
+- `backend/app/rcon_annual_rankings.py` (crear)
+- `backend/app/rcon_admin_log_materialization.py` (si hace falta reutilizar inicialización o alcance)
+- `backend/app/postgres_rcon_storage.py` (solo si detecta ajustes de soporte SQLite/Postgres)
+- `backend/app/sqlite_to_postgres_migration.py` (solo si detecta ajustes migratorios necesarios)
+
+## Steps
+
+1. Seguir los archivos de `Files to Read First` y respetar el contrato de datos existente.
+2. Implementar `generate_annual_ranking_snapshot` con:
+   - `year`, `server_key`, `metric`, `limit`, `replace_existing`.
+   - Soporte inicial para `metric='kills'`.
+3. Agregar una ruta CLI opcional (`python -m app.rcon_annual_rankings generate --year ...`).
+4. Reusar conexión SQLite/Postgres con `connect_sqlite_readonly`/`connect_postgres_compat` y placeholders `?`.
+5. Validar la función con datos existentes y con BD vacía sin crash.
+6. Preparar la Outcome con validación, comportamiento con vacío y limitaciones.
+
+## Constraints
+
+- No crear API pública en esta task.
+- Mantener el cambio pequeño y verificable.
+- No tocar frontend ni archivos `frontend/assets/js/partida-actual.js`.
+- No tocar `frontend/assets/img/clans/bxb.png`.
+- No reactivar Elo/MMR.
+- No reintroducir Comunidad Hispana #03.
+- No ampliar workers históricos salvo que sea imprescindible y justificable.
+- No incluir métricas adicionales a `kills` salvo que no amplíe el alcance de forma razonable y segura.
+
+## Validation
+
+- `python -m compileall backend/app`
+- `scripts/run-integration-tests.ps1` cuando aplique.
+- Validación local con SQLite:
+  - inicializar storage,
+  - ejecutar generador con un año de prueba (si hay datos),
+  - verificar que se inserta el snapshot y hasta `limit` items.
+- Verificar comportamiento sin datos: debe terminar sin excepción y crear snapshot vacío o sin items cuando corresponda.
+- `git diff --name-only` y confirmar que no se tocó frontend.
+
+## Outcome
+
+- Archivos modificados:
+  - `backend/app/rcon_annual_rankings.py` (nuevo)
+- Función/comando generado:
+  - `generate_annual_ranking_snapshot(year: int, server_key: str | None = None, metric: str = "kills", limit: int = 20, replace_existing: bool = True, db_path: Path | None = None) -> dict[str, object]`.
+  - CLI opcional: `python -m app.rcon_annual_rankings generate --year <year> [--server-key <server>] [--metric kills] [--limit <n>] [--replace-existing]`.
+- Validaciones ejecutadas y resultados:
+  - `python -m compileall backend/app` (sin errores).
+  - `scripts/run-integration-tests.ps1` (pasado).
+  - `initialize_rcon_materialized_storage()` + `generate_annual_ranking_snapshot(year=2026, metric='kills', limit=10)` con datos existentes: snapshot creado (`source_matches_count` > 0), `items <= 10`.
+  - `generate_annual_ranking_snapshot(year=1990, metric='kills', limit=20)` con ventana sin datos: snapshot creado y `items: []` sin excepción.
+  - `replace_existing=False` devuelve `skipped_regeneration: True` y reutiliza snapshot existente.
+  - CLI manual ejecutada: `python -m app.rcon_annual_rankings generate --year 1990 --metric kills --limit 3`.
+- Limitaciones conocidas:
+  - Soporte inicial exclusivamente para `metric='kills'`.
+  - No se implementó API pública ni refresco programado.
+  - No se modificón workers históricos.
+  - No se tocó frontend ni archivos preexistentes `frontend/assets/js/partida-actual.js` y `frontend/assets/img/clans/bxb.png`.
+- Siguiente task recomendada: API pública de ranking anual snapshot.

ai/tasks/pending/TASK-168-add-annual-ranking-snapshot-generator.md

@@ -1,91 +0,0 @@
----
-id: TASK-168
-title: Add annual ranking snapshot generator
-status: pending
-type: backend
-team: Arquitecto de Base de Datos
-supporting_teams:
-  - Backend Senior
-  - Arquitecto Python
-roadmap_item: foundation
-priority: medium
----
-
-# TASK-168 - Add annual ranking snapshot generator
-
-## Goal
-
-Implementar un generador backend para snapshots anuales top 20 de jugadores sobre datos RCON materializados.
-
-## Context
-
-TASK-166 definió el diseño de persistencia, y TASK-167 ya añadió las tablas:
-
-- `rcon_annual_ranking_snapshots`
-- `rcon_annual_ranking_snapshot_items`
-
-Ahora se necesita el generador que rellene esas tablas desde
-`rcon_materialized_matches` y `rcon_match_player_stats` con `source_basis = admin-log-match-ended`.
-
-Debe mantenerse separado de frontend y sin introducir una API pública todavía.
-
-## Files to Read First
-
-- `AGENTS.md`
-- `ai/repo-context.md`
-- `ai/architecture-index.md`
-- `docs/annual-ranking-snapshot-schema-plan.md`
-- `backend/app/rcon_admin_log_materialization.py`
-- `backend/app/rcon_historical_leaderboards.py`
-- `backend/app/rcon_historical_player_stats.py`
-- `backend/app/postgres_rcon_storage.py`
-- `backend/app/sqlite_to_postgres_migration.py`
-
-## Expected Files to Modify
-
-- `backend/app/rcon_annual_rankings.py` (crear)
-- `backend/app/rcon_admin_log_materialization.py` (si hace falta reutilizar inicialización o alcance)
-- `backend/app/postgres_rcon_storage.py` (solo si detecta ajustes de soporte SQLite/Postgres)
-- `backend/app/sqlite_to_postgres_migration.py` (solo si detecta ajustes migratorios necesarios)
-
-## Steps
-
-1. Seguir los archivos de `Files to Read First` y respetar el contrato de datos existente.
-2. Implementar `generate_annual_ranking_snapshot` con:
-   - `year`, `server_key`, `metric`, `limit`, `replace_existing`.
-   - Soporte inicial para `metric='kills'`.
-3. Agregar una ruta CLI opcional (`python -m app.rcon_annual_rankings generate --year ...`).
-4. Reusar conexión SQLite/Postgres con `connect_sqlite_readonly`/`connect_postgres_compat` y placeholders `?`.
-5. Validar la función con datos existentes y con BD vacía sin crash.
-6. Preparar la Outcome con validación, comportamiento con vacío y limitaciones.
-
-## Constraints
-
-- No crear API pública en esta task.
-- Mantener el cambio pequeño y verificable.
-- No tocar frontend ni archivos `frontend/assets/js/partida-actual.js`.
-- No tocar `frontend/assets/img/clans/bxb.png`.
-- No reactivar Elo/MMR.
-- No reintroducir Comunidad Hispana #03.
-- No ampliar workers históricos salvo que sea imprescindible y justificable.
-- No incluir métricas adicionales a `kills` salvo que no amplíe el alcance de forma razonable y segura.
-
-## Validation
-
-- `python -m compileall backend/app`
-- `scripts/run-integration-tests.ps1` cuando aplique.
-- Validación local con SQLite:
-  - inicializar storage,
-  - ejecutar generador con un año de prueba (si hay datos),
-  - verificar que se inserta el snapshot y hasta `limit` items.
-- Verificar comportamiento sin datos: debe terminar sin excepción y crear snapshot vacío o sin items cuando corresponda.
-- `git diff --name-only` y confirmar que no se tocó frontend.
-
-## Outcome
-
-- Archivos modificados.
-- Función/comando generado.
-- Validaciones ejecutadas y resultados (incluyendo caso vacío).
-- Limitaciones conocidas.
-- Siguiente task recomendada: API pública de ranking anual snapshot.
-

backend/app/rcon_annual_rankings.py

@@ -0,0 +1,421 @@
+"""Annual ranking snapshot generator over materialized RCON match stats."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from datetime import datetime, timezone
+from pathlib import Path
+
+from .config import get_storage_path, use_postgres_rcon_storage
+from .historical_storage import ALL_SERVERS_SLUG
+from .rcon_admin_log_materialization import MATCH_RESULT_SOURCE, initialize_rcon_materialized_storage
+from .sqlite_utils import connect_sqlite_writer
+
+
+def generate_annual_ranking_snapshot(
+    *,
+    year: int,
+    server_key: str | None = None,
+    metric: str = "kills",
+    limit: int = 20,
+    replace_existing: bool = True,
+    db_path: Path | None = None,
+) -> dict[str, object]:
+    """Generate and persist an annual top-k ranking snapshot for materialized RCON data."""
+    normalized_server_key = _normalize_server_key(server_key)
+    normalized_metric = _normalize_metric(metric)
+    normalized_limit = max(1, int(limit))
+    window_start, window_end = _annual_window(year)
+
+    resolved_path = initialize_rcon_materialized_storage(db_path=db_path)
+    scope_sql, scope_params = _build_scope_sql(normalized_server_key)
+    if use_postgres_rcon_storage(explicit_sqlite_path=db_path):
+        from .postgres_rcon_storage import connect_postgres_compat
+
+        connection_scope = connect_postgres_compat()
+    else:
+        connection_scope = connect_sqlite_writer(resolved_path)
+
+    with connection_scope as connection:
+        source_matches_count = _count_matches_in_window(
+            connection=connection,
+            start=window_start,
+            end=window_end,
+            scope_sql=scope_sql,
+            scope_params=scope_params,
+        )
+        existing_snapshot_id = _find_existing_snapshot(
+            connection=connection,
+            year=year,
+            server_key=normalized_server_key,
+            metric=normalized_metric,
+        )
+        if existing_snapshot_id is not None and not replace_existing:
+            snapshot = _get_snapshot(connection=connection, snapshot_id=existing_snapshot_id)
+            items = _list_items(connection=connection, snapshot_id=existing_snapshot_id)
+            return {
+                "status": "ok",
+                "snapshot": snapshot,
+                "items": items,
+                "source_matches_count": source_matches_count,
+                "ranked_players": len(items),
+                "skipped_regeneration": True,
+            }
+
+        ranking_rows = _fetch_annual_ranking_rows(
+            connection=connection,
+            start=window_start,
+            end=window_end,
+            metric=normalized_metric,
+            limit=normalized_limit,
+            scope_sql=scope_sql,
+            scope_params=scope_params,
+        )
+
+        snapshot_id = _delete_existing_snapshot(
+            connection=connection,
+            year=year,
+            server_key=normalized_server_key,
+            metric=normalized_metric,
+        )
+        snapshot_id = _insert_snapshot(
+            connection=connection,
+            year=year,
+            server_key=normalized_server_key,
+            metric=normalized_metric,
+            limit=normalized_limit,
+            source_matches_count=source_matches_count,
+            window_start=window_start,
+            window_end=window_end,
+        )
+
+        _insert_items(
+            connection=connection,
+            snapshot_id=snapshot_id,
+            rows=ranking_rows,
+            limit=normalized_limit,
+        )
+
+        snapshot = _get_snapshot(connection=connection, snapshot_id=snapshot_id)
+        items = _list_items(connection=connection, snapshot_id=snapshot_id)
+
+    return {
+        "status": "ok",
+        "snapshot": snapshot,
+        "items": items,
+        "source_matches_count": source_matches_count,
+        "ranked_players": len(items),
+    }
+
+
+def _normalize_server_key(server_key: str | None) -> str:
+    normalized = str(server_key or "").strip()
+    normalized_lower = normalized.lower()
+    if not normalized or normalized_lower in {ALL_SERVERS_SLUG, "all"}:
+        return ALL_SERVERS_SLUG
+    return normalized
+
+
+def _normalize_metric(metric: str) -> str:
+    normalized = str(metric or "kills").strip().lower()
+    if normalized != "kills":
+        raise ValueError("Only metric 'kills' is supported for annual ranking generation.")
+    return normalized
+
+
+def _annual_window(year: int) -> tuple[str, str]:
+    start = datetime(year, 1, 1, 0, 0, 0, tzinfo=timezone.utc).isoformat().replace("+00:00", "Z")
+    end = datetime(year + 1, 1, 1, 0, 0, 0, tzinfo=timezone.utc).isoformat().replace("+00:00", "Z")
+    return start, end
+
+
+def _fetch_annual_ranking_rows(
+    *,
+    connection: object,
+    start: str,
+    end: str,
+    metric: str,
+    limit: int,
+    scope_sql: str,
+    scope_params: list[object],
+) -> list[dict[str, object]]:
+    # For now metric support is intentionally narrowed to kills only.
+    rows = connection.execute(
+        f"""
+        SELECT
+            stats.player_id,
+            COALESCE(MAX(stats.player_name), stats.player_id) AS player_name,
+            SUM(COALESCE(stats.kills, 0)) AS kills,
+            SUM(COALESCE(stats.deaths, 0)) AS deaths,
+            SUM(COALESCE(stats.teamkills, 0)) AS teamkills,
+            COUNT(DISTINCT stats.match_key) AS matches_considered,
+            SUM(COALESCE(stats.kills, 0)) AS metric_value
+        FROM rcon_match_player_stats AS stats
+        INNER JOIN rcon_materialized_matches AS matches
+            ON matches.target_key = stats.target_key
+           AND matches.match_key = stats.match_key
+        WHERE matches.source_basis = ?
+          AND COALESCE(CAST(matches.ended_at AS TEXT), CAST(matches.started_at AS TEXT)) >= ?
+          AND COALESCE(CAST(matches.ended_at AS TEXT), CAST(matches.started_at AS TEXT)) < ?
+          {scope_sql}
+          AND TRIM(COALESCE(stats.player_name, '')) != ''
+        GROUP BY stats.player_id
+        HAVING SUM(COALESCE(stats.kills, 0)) > 0
+        ORDER BY metric_value DESC, matches_considered DESC, player_name ASC
+        LIMIT ?
+        """,
+        [MATCH_RESULT_SOURCE, start, end, *scope_params, limit],
+    ).fetchall()
+    if metric != "kills":
+        return []
+    return [dict(row) for row in rows]
+
+
+def _count_matches_in_window(
+    *,
+    connection: object,
+    start: str,
+    end: str,
+    scope_sql: str,
+    scope_params: list[object],
+) -> int:
+    row = connection.execute(
+        f"""
+        SELECT COUNT(*) AS source_matches_count
+        FROM (
+            SELECT matches.target_key, matches.match_key
+            FROM rcon_materialized_matches AS matches
+            WHERE matches.source_basis = ?
+              AND COALESCE(CAST(matches.ended_at AS TEXT), CAST(matches.started_at AS TEXT)) >= ?
+              AND COALESCE(CAST(matches.ended_at AS TEXT), CAST(matches.started_at AS TEXT)) < ?
+              {scope_sql}
+            GROUP BY matches.target_key, matches.match_key
+        ) AS source_matches
+        """,
+        [MATCH_RESULT_SOURCE, start, end, *scope_params],
+    ).fetchone()
+    return int(row["source_matches_count"] or 0) if row else 0
+
+
+def _find_existing_snapshot(
+    *,
+    connection: object,
+    year: int,
+    server_key: str,
+    metric: str,
+) -> int | None:
+    row = connection.execute(
+        """
+        SELECT id
+        FROM rcon_annual_ranking_snapshots
+        WHERE year = ? AND server_key = ? AND metric = ?
+        LIMIT 1
+        """,
+        [year, server_key, metric],
+    ).fetchone()
+    return int(row["id"]) if row else None
+
+
+def _delete_existing_snapshot(
+    *,
+    connection: object,
+    year: int,
+    server_key: str,
+    metric: str,
+) -> int | None:
+    existing_id = _find_existing_snapshot(
+        connection=connection,
+        year=year,
+        server_key=server_key,
+        metric=metric,
+    )
+    if existing_id is None:
+        return None
+    connection.execute(
+        "DELETE FROM rcon_annual_ranking_snapshot_items WHERE snapshot_id = ?",
+        (existing_id,),
+    )
+    connection.execute(
+        "DELETE FROM rcon_annual_ranking_snapshots WHERE id = ?",
+        (existing_id,),
+    )
+    return existing_id
+
+
+def _insert_snapshot(
+    *,
+    connection: object,
+    year: int,
+    server_key: str,
+    metric: str,
+    limit: int,
+    source_matches_count: int,
+    window_start: str,
+    window_end: str,
+) -> int:
+    cursor = connection.execute(
+        """
+        INSERT INTO rcon_annual_ranking_snapshots (
+            year,
+            server_key,
+            metric,
+            limit_size,
+            source_basis,
+            window_start,
+            window_end,
+            source_matches_count
+        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+        RETURNING id
+        """,
+        [
+            year,
+            server_key,
+            metric,
+            limit,
+            MATCH_RESULT_SOURCE,
+            window_start,
+            window_end,
+            source_matches_count,
+        ],
+    )
+    row = cursor.fetchone()
+    if row is not None and row["id"] is not None:
+        return int(row["id"])
+
+    existing = _find_existing_snapshot(
+        connection=connection,
+        year=year,
+        server_key=server_key,
+        metric=metric,
+    )
+    if existing is None:
+        raise RuntimeError("Unable to resolve annual snapshot id after insert.")
+    return existing
+
+
+def _insert_items(
+    *,
+    connection: object,
+    snapshot_id: int,
+    rows: list[dict[str, object]],
+    limit: int,
+) -> None:
+    for index, row in enumerate(rows[:limit], start=1):
+        kills = int(row.get("kills") or 0)
+        deaths = int(row.get("deaths") or 0)
+        metric_value = int(row.get("metric_value") or 0)
+        connection.execute(
+            """
+            INSERT INTO rcon_annual_ranking_snapshot_items (
+                snapshot_id,
+                ranking_position,
+                player_id,
+                player_name,
+                metric_value,
+                matches_considered,
+                kills,
+                deaths,
+                teamkills,
+                kd_ratio
+            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            """,
+            [
+                snapshot_id,
+                index,
+                str(row.get("player_id")),
+                str(row.get("player_name")),
+                metric_value,
+                int(row.get("matches_considered") or 0),
+                kills,
+                deaths,
+                int(row.get("teamkills") or 0),
+                float(kills / deaths) if deaths else float(kills),
+            ],
+        )
+
+
+def _get_snapshot(*, connection: object, snapshot_id: int) -> dict[str, object] | None:
+    row = connection.execute(
+        """
+        SELECT
+            id,
+            year,
+            server_key,
+            metric,
+            limit_size,
+            source_basis,
+            window_start,
+            window_end,
+            status,
+            source_matches_count,
+            generated_at
+        FROM rcon_annual_ranking_snapshots
+        WHERE id = ?
+        LIMIT 1
+        """,
+        [snapshot_id],
+    ).fetchone()
+    if not row:
+        return None
+    return dict(row)
+
+
+def _list_items(*, connection: object, snapshot_id: int) -> list[dict[str, object]]:
+    rows = connection.execute(
+        """
+        SELECT
+            ranking_position,
+            player_id,
+            player_name,
+            metric_value,
+            matches_considered,
+            kills,
+            deaths,
+            teamkills,
+            kd_ratio
+        FROM rcon_annual_ranking_snapshot_items
+        WHERE snapshot_id = ?
+        ORDER BY ranking_position ASC
+        """,
+        [snapshot_id],
+    ).fetchall()
+    return [dict(row) for row in rows]
+
+
+def _build_scope_sql(server_key: str, *, table_alias: str = "matches") -> tuple[str, list[object]]:
+    if server_key == ALL_SERVERS_SLUG:
+        return "", []
+    return f"AND ({table_alias}.target_key = ? OR {table_alias}.external_server_id = ?)", [server_key, server_key]
+
+
+def _main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description="Generate annual ranking snapshots.")
+    subparsers = parser.add_subparsers(dest="command")
+    generate_parser = subparsers.add_parser("generate")
+    generate_parser.add_argument("--year", type=int, required=True)
+    generate_parser.add_argument("--server-key", default=None)
+    generate_parser.add_argument("--metric", default="kills")
+    generate_parser.add_argument("--limit", type=int, default=20)
+    generate_parser.add_argument("--replace-existing", action="store_true", default=True)
+    parser.set_defaults(command="generate")
+    args = parser.parse_args(argv)
+
+    if args.command == "generate":
+        payload = generate_annual_ranking_snapshot(
+            year=args.year,
+            server_key=args.server_key,
+            metric=args.metric,
+            limit=args.limit,
+            replace_existing=args.replace_existing,
+            db_path=get_storage_path(),
+        )
+        print(json.dumps({"status": "ok", "data": payload}, ensure_ascii=True, indent=2))
+        return 0
+
+    parser.print_help()
+    return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(_main())