diff --git a/ai/architecture-index.md b/ai/architecture-index.md index 45b0eaf..e79d092 100644 --- a/ai/architecture-index.md +++ b/ai/architecture-index.md @@ -72,5 +72,6 @@ Community website repository with a static landing in the current phase and a pl - The persisted historical domain model for CRCON matches, players and ingestion runs is documented in `docs/historical-domain-model.md`. - The V1 monthly MVP scoring proposal for persisted historical player metrics is documented in `docs/monthly-mvp-ranking-scoring-design.md`. - The audited boundary between direct live RCON and future event-driven RCON metrics is documented in `docs/rcon-data-capability-audit.md`. +- The first V2 player-event foundation now lives in dedicated `player_event_*` backend modules and starts from CRCON match-detail summaries, not from live RCON. - Frontend data consumption should remain progressive, endpoint by endpoint, with static fallbacks preserved during migration. - The frontend integration strategy is documented in `docs/frontend-data-consumption-plan.md`. diff --git a/ai/tasks/done/TASK-077-player-kill-event-source-adapter.md b/ai/tasks/done/TASK-077-player-kill-event-source-adapter.md new file mode 100644 index 0000000..64464cc --- /dev/null +++ b/ai/tasks/done/TASK-077-player-kill-event-source-adapter.md @@ -0,0 +1,77 @@ +# TASK-077-player-kill-event-source-adapter + +## Goal +Implementar una primera capa/adaptador de fuente de eventos centrada en eventos de kill/death para preparar la V2 del sistema de métricas de jugador. + +## Context +Las auditorías previas concluyen que el cliente RCON live actual no basta por sí solo para obtener directamente todos los agregados avanzados visibles en ecosistemas tipo CRCON/HLL Records. Para construir métricas como: +- killer -> victim +- most_killed +- death_by +- kills por arma +- teamkills por evento + +hace falta una fuente de eventos o una adaptación técnica que permita modelar esos datos en bruto antes de agregarlos. + +La V2 debe empezar por una primera capa mínima centrada en eventos de kill/death, sin abarcar todavía todo el resto de acciones tácticas. + +## Steps +1. Revisar las auditorías y documentos de diseño V2: + - docs/rcon-data-capability-audit.md + - docs/crcon-advanced-metrics-origin-audit.md + - docs/player-event-pipeline-v2-design.md +2. Identificar la fuente o punto técnico más viable dentro del proyecto actual para empezar a capturar eventos de kill/death. +3. Diseñar e implementar un adaptador/fuente mínima para eventos de jugador que, como mínimo, pueda producir un formato común con campos base como: + - server + - match + - timestamp + - killer + - victim + - weapon si existe + - teamkill si existe +4. Mantener esta capa desacoplada de la persistencia final. +5. Documentar claramente: + - qué datos se capturan realmente en esta fase + - qué datos siguen fuera de alcance +6. No implementar todavía snapshots/UI del MVP V2. +7. No romper live RCON ni el histórico actual. +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 +- ai/repo-context.md +- ai/architecture-index.md +- docs/rcon-data-capability-audit.md +- docs/crcon-advanced-metrics-origin-audit.md +- docs/player-event-pipeline-v2-design.md +- backend/README.md +- backend/app/rcon_client.py +- backend/app/providers/rcon_provider.py +- backend/app/data_sources.py + +## Expected Files to Modify +- backend/README.md +- opcionalmente nuevos módulos, por ejemplo: + - backend/app/player_event_source.py + - backend/app/providers/player_event_source_provider.py + - backend/app/event_sources/*.py +- opcionalmente ai/architecture-index.md si conviene enlazar la nueva pieza + +## Constraints +- No implementar todavía agregados finales ni UI. +- No romper el proveedor RCON live actual. +- No hacer cambios destructivos. +- Mantener el trabajo centrado en una fuente mínima de eventos kill/death. + +## Validation +- Existe una capa/adaptador clara para eventos de kill/death. +- Produce un formato mínimo reutilizable por la persistencia posterior. +- Queda documentado qué cubre y qué no cubre todavía. +- 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. diff --git a/ai/tasks/done/TASK-078-player-event-raw-ledger-storage.md b/ai/tasks/done/TASK-078-player-event-raw-ledger-storage.md new file mode 100644 index 0000000..1f81d77 --- /dev/null +++ b/ai/tasks/done/TASK-078-player-event-raw-ledger-storage.md @@ -0,0 +1,65 @@ +# TASK-078-player-event-raw-ledger-storage + +## Goal +Crear la persistencia raw mínima para eventos de jugador V2, permitiendo almacenar de forma fiable eventos de kill/death con claves estables, deduplicación y base para agregados posteriores. + +## Context +Una vez exista una fuente/adaptador de eventos, hace falta una capa de persistencia en bruto. Esta capa debe ser lo bastante simple para operar ya, pero lo bastante sólida para soportar luego agregados como: +- most_killed +- death_by +- kills por arma +- teamkills por jugador +- killer/victim mensual + +## Steps +1. Revisar el diseño V2 del pipeline de eventos. +2. Diseñar una tabla o conjunto mínimo de tablas para el ledger raw de eventos de jugador. +3. Incluir como base, si la fuente lo permite: + - server key + - match reference + - event timestamp + - killer identity + - victim identity + - weapon + - teamkill flag + - event source reference o clave de deduplicación +4. Implementar inicialización segura del storage de eventos. +5. Implementar inserción idempotente o deduplicada. +6. Mantener separada esta capa del histórico actual `historical_*`. +7. Documentar la estructura y su propósito. +8. No implementar todavía agregados finales ni UI. +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 +- docs/player-event-pipeline-v2-design.md +- backend/README.md +- backend/app/historical_models.py +- backend/app/historical_storage.py +- cualquier módulo nuevo de la task anterior relacionado con eventos + +## Expected Files to Modify +- backend/README.md +- opcionalmente nuevos módulos, por ejemplo: + - backend/app/player_event_storage.py + - backend/app/player_event_models.py +- opcionalmente docs/decisions.md si conviene fijar una decisión de persistencia + +## Constraints +- No romper el histórico actual. +- No mezclar esta persistencia con snapshots ni UI. +- No hacer cambios destructivos. +- Mantener el trabajo centrado en el ledger raw de eventos. + +## Validation +- Existe una persistencia raw mínima para eventos de jugador. +- Soporta deduplicación o idempotencia razonable. +- Queda separada del histórico actual. +- 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. diff --git a/ai/tasks/done/TASK-079-player-event-ingestion-worker.md b/ai/tasks/done/TASK-079-player-event-ingestion-worker.md new file mode 100644 index 0000000..e1d99f2 --- /dev/null +++ b/ai/tasks/done/TASK-079-player-event-ingestion-worker.md @@ -0,0 +1,57 @@ +# TASK-079-player-event-ingestion-worker + +## Goal +Añadir un worker/proceso incremental para ingerir eventos de jugador V2 de forma segura, reanudable y compatible con la operación actual del proyecto. + +## Context +Con una fuente de eventos y un ledger raw, hace falta un proceso de ingestión incremental que capture eventos sin depender del request path del usuario y sin romper el pipeline actual de histórico y snapshots. + +## Steps +1. Revisar la nueva capa de fuente de eventos y la persistencia raw. +2. Diseñar un worker incremental específico para eventos de jugador. +3. Añadir soporte para: + - ejecución manual + - reintentos básicos + - checkpoint o mecanismo de reanudación si aplica + - deduplicación segura +4. Mantenerlo desacoplado del `historical-runner` actual si eso reduce riesgo operacional. +5. Documentar cómo se ejecuta y cuál es su alcance. +6. No implementar todavía snapshots finales ni UI del MVP V2. +7. No romper el stack Docker actual ni el flujo histórico existente. +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 +- docs/player-event-pipeline-v2-design.md +- backend/README.md +- backend/app/historical_runner.py +- backend/app/config.py +- módulos de eventos añadidos en tasks previas + +## Expected Files to Modify +- backend/app/config.py +- backend/README.md +- opcionalmente docker-compose.yml si hace falta dejar preparado el worker +- opcionalmente nuevos módulos, por ejemplo: + - backend/app/player_event_worker.py + - backend/app/player_event_ingestion.py + +## Constraints +- No romper el runner histórico actual. +- No meter todavía lógica de scoring V2. +- No tocar producto visible. +- No hacer cambios destructivos. +- Mantener el trabajo centrado en ingestión incremental de eventos. + +## Validation +- Existe un worker o flujo incremental para eventos de jugador. +- Puede ejecutarse de forma segura y reanudable. +- No rompe el resto del sistema. +- 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. diff --git a/ai/tasks/done/TASK-080-player-event-derived-aggregates.md b/ai/tasks/done/TASK-080-player-event-derived-aggregates.md new file mode 100644 index 0000000..d513af3 --- /dev/null +++ b/ai/tasks/done/TASK-080-player-event-derived-aggregates.md @@ -0,0 +1,65 @@ +# TASK-080-player-event-derived-aggregates + +## Goal +Construir los primeros agregados derivados V2 a partir del ledger raw de eventos de jugador, centrados en duelos y armas, para preparar una futura V2 del MVP. + +## Context +Con eventos raw ya capturados, el siguiente paso es derivar métricas útiles y reutilizables. Las primeras más valiosas para la V2 del MVP son: +- most_killed +- death_by +- killer/victim +- kills por arma +- teamkills por jugador + +## Steps +1. Revisar la estructura del ledger raw y el diseño V2. +2. Diseñar e implementar agregados básicos por: + - partida + - mes + - servidor + - all-servers cuando aplique +3. Incluir al menos: + - most_killed + - death_by + - net duel summaries + - kills por arma + - teamkills derivados por jugador +4. Mantener los agregados separados de la UI y del MVP V1. +5. Documentar qué métricas derivadas quedan ya disponibles para una futura V2 del score. +6. No implementar todavía la fórmula final del MVP V2 ni su UI. +7. No hacer cambios destructivos. +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 +- docs/player-event-pipeline-v2-design.md +- docs/crcon-advanced-metrics-origin-audit.md +- backend/README.md +- módulos de eventos raw ya implementados +- backend/app/historical_storage.py +- docs/monthly-mvp-ranking-scoring-design.md + +## Expected Files to Modify +- backend/README.md +- opcionalmente nuevos módulos, por ejemplo: + - backend/app/player_event_aggregates.py + - backend/app/player_event_queries.py +- opcionalmente docs/decisions.md si hace falta fijar alguna decisión de agregación + +## Constraints +- No tocar todavía la UI del MVP V2. +- No romper el MVP V1. +- No hacer cambios destructivos. +- Mantener el trabajo centrado en agregados base de duelos y armas. + +## Validation +- Existen agregados derivados básicos reutilizables para V2. +- Queda clara su disponibilidad por servidor/mes/all-servers cuando aplique. +- 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. diff --git a/backend/README.md b/backend/README.md index d85ceac..372ff81 100644 --- a/backend/README.md +++ b/backend/README.md @@ -998,6 +998,91 @@ sin tocar endpoints ni payloads. Esta separacion mantiene el backend simple y deja una base clara para futuras tasks sin introducir integraciones reales todavia. +## Fuente y ledger de eventos de jugador V2 + +La repo incluye ahora una primera base V2 separada del historico `historical_*` +para preparar metricas avanzadas de duelos, armas y teamkills sin tocar todavia +la UI ni el scoring final. + +Fuente minima elegida en esta fase: + +- detalle de partida `GET /api/get_map_scoreboard?map_id={id}` del scoreboard CRCON + +Importante: + +- esta fuente no es un feed raw por kill +- el adaptador actual normaliza solo senales parciales ya visibles en el + resumen de partida: + - `most_killed` + - `death_by` + - `weapons` + - `death_by_weapons` + - `teamkills` +- `occurred_at` usa el timestamp de cierre o inicio de la partida, no el + instante exacto del kill +- el ledger raw es append-only y deduplica por `event_id` +- la persistencia queda separada de `historical_matches` y + `historical_player_match_stats` aunque comparte el mismo SQLite de desarrollo + +Contrato minimo normalizado por evento: + +- `event_id` +- `event_type` +- `occurred_at` +- `server_slug` +- `external_match_id` +- `source_kind` +- `source_ref` +- `killer_player_key` +- `victim_player_key` +- `weapon_name` +- `kill_category` +- `is_teamkill` +- `event_value` + +Tablas nuevas: + +- `player_event_raw_ledger` +- `player_event_ingestion_runs` +- `player_event_backfill_progress` + +Comandos manuales desde `backend/`: + +```powershell +python -m app.player_event_worker refresh +python -m app.player_event_worker refresh --server comunidad-hispana-01 --max-pages 1 +python -m app.player_event_worker loop --interval 1800 +``` + +Variables opcionales del worker: + +- `HLL_PLAYER_EVENT_REFRESH_INTERVAL_SECONDS` +- `HLL_PLAYER_EVENT_REFRESH_MAX_RETRIES` +- `HLL_PLAYER_EVENT_REFRESH_RETRY_DELAY_SECONDS` + +Politica operativa minima: + +- el worker corre fuera del request path HTTP +- reusa la capa historica `public-scoreboard` solo como fuente de detalle +- persiste checkpoints por servidor y pagina +- la reejecucion es segura porque el ledger usa insercion idempotente por + `event_id` + +Agregados V2 ya disponibles desde codigo: + +- `list_most_killed()` +- `list_death_by()` +- `list_net_duel_summaries()` +- `list_weapon_kills()` +- `list_teamkill_summaries()` + +Limitaciones actuales de esta fase: + +- no existe todavia un ledger raw por kill individual +- los agregados de duelos y armas son parciales, porque dependen del mejor + resumen disponible por jugador en CRCON y no de todos los encounters del match +- la V2 no expone aun endpoints HTTP ni snapshots propios + ## Alcance Esta fase no implementa: diff --git a/backend/app/config.py b/backend/app/config.py index 1b9f5ff..b3803c0 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -25,6 +25,9 @@ DEFAULT_HISTORICAL_REFRESH_RETRY_DELAY_SECONDS = 30 DEFAULT_HISTORICAL_FULL_SNAPSHOT_EVERY_RUNS = 4 DEFAULT_HISTORICAL_WEEKLY_FALLBACK_MIN_MATCHES = 3 DEFAULT_HISTORICAL_WEEKLY_FALLBACK_MAX_WEEKDAY = 2 +DEFAULT_PLAYER_EVENT_REFRESH_INTERVAL_SECONDS = 1800 +DEFAULT_PLAYER_EVENT_REFRESH_MAX_RETRIES = 2 +DEFAULT_PLAYER_EVENT_REFRESH_RETRY_DELAY_SECONDS = 30 DEFAULT_ALLOWED_ORIGINS = ( "null", "http://127.0.0.1:5500", @@ -269,6 +272,44 @@ def get_historical_weekly_fallback_max_weekday() -> int: return max_weekday +def get_player_event_refresh_interval_seconds() -> int: + """Return the default interval used by the player event refresh loop.""" + configured_value = os.getenv( + "HLL_PLAYER_EVENT_REFRESH_INTERVAL_SECONDS", + str(DEFAULT_PLAYER_EVENT_REFRESH_INTERVAL_SECONDS), + ) + interval_seconds = int(configured_value) + if interval_seconds <= 0: + raise ValueError("HLL_PLAYER_EVENT_REFRESH_INTERVAL_SECONDS must be positive.") + return interval_seconds + + +def get_player_event_refresh_max_retries() -> int: + """Return the retry count used by the player event refresh loop.""" + configured_value = os.getenv( + "HLL_PLAYER_EVENT_REFRESH_MAX_RETRIES", + str(DEFAULT_PLAYER_EVENT_REFRESH_MAX_RETRIES), + ) + max_retries = int(configured_value) + if max_retries < 0: + raise ValueError("HLL_PLAYER_EVENT_REFRESH_MAX_RETRIES must be zero or positive.") + return max_retries + + +def get_player_event_refresh_retry_delay_seconds() -> int: + """Return the wait time between player event refresh retries.""" + configured_value = os.getenv( + "HLL_PLAYER_EVENT_REFRESH_RETRY_DELAY_SECONDS", + str(DEFAULT_PLAYER_EVENT_REFRESH_RETRY_DELAY_SECONDS), + ) + retry_delay_seconds = int(configured_value) + if retry_delay_seconds < 0: + raise ValueError( + "HLL_PLAYER_EVENT_REFRESH_RETRY_DELAY_SECONDS must be zero or positive." + ) + return retry_delay_seconds + + def get_a2s_targets_payload() -> str | None: """Return the optional JSON payload that overrides local A2S targets.""" raw_payload = os.getenv(DEFAULT_A2S_TARGETS_ENV_VAR) diff --git a/backend/app/player_event_aggregates.py b/backend/app/player_event_aggregates.py new file mode 100644 index 0000000..956e4cc --- /dev/null +++ b/backend/app/player_event_aggregates.py @@ -0,0 +1,254 @@ +"""Derived duel and weapon aggregates computed from the raw player event ledger.""" + +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +from .config import get_storage_path +from .player_event_storage import initialize_player_event_storage + + +def list_most_killed( + *, + server_slug: str | None = None, + month: str | None = None, + external_match_id: str | None = None, + limit: int = 10, + db_path: Path | None = None, +) -> list[dict[str, object]]: + """Return strongest killer -> victim summaries from the raw ledger.""" + return _query_pair_summary( + event_type="player_kill_summary", + server_slug=server_slug, + month=month, + external_match_id=external_match_id, + limit=limit, + db_path=db_path, + ) + + +def list_death_by( + *, + server_slug: str | None = None, + month: str | None = None, + external_match_id: str | None = None, + limit: int = 10, + db_path: Path | None = None, +) -> list[dict[str, object]]: + """Return strongest killer -> victim summaries from the victim perspective.""" + return _query_pair_summary( + event_type="player_death_summary", + server_slug=server_slug, + month=month, + external_match_id=external_match_id, + limit=limit, + db_path=db_path, + ) + + +def list_net_duel_summaries( + *, + server_slug: str | None = None, + month: str | None = None, + external_match_id: str | None = None, + limit: int = 10, + db_path: Path | None = None, +) -> list[dict[str, object]]: + """Return partial net duel summaries using the strongest encounter signals available.""" + resolved_path = initialize_player_event_storage(db_path=db_path) + where_sql, params = _build_common_where( + event_type="player_kill_summary", + server_slug=server_slug, + month=month, + external_match_id=external_match_id, + ) + with _connect(resolved_path) as connection: + rows = connection.execute( + f""" + WITH duel_pairs AS ( + SELECT + CASE + WHEN COALESCE(killer_player_key, '') <= COALESCE(victim_player_key, '') + THEN killer_player_key + ELSE victim_player_key + END AS player_a_key, + CASE + WHEN COALESCE(killer_player_key, '') <= COALESCE(victim_player_key, '') + THEN killer_display_name + ELSE victim_display_name + END AS player_a_name, + CASE + WHEN COALESCE(killer_player_key, '') <= COALESCE(victim_player_key, '') + THEN victim_player_key + ELSE killer_player_key + END AS player_b_key, + CASE + WHEN COALESCE(killer_player_key, '') <= COALESCE(victim_player_key, '') + THEN victim_display_name + ELSE killer_display_name + END AS player_b_name, + CASE + WHEN COALESCE(killer_player_key, '') <= COALESCE(victim_player_key, '') + THEN event_value + ELSE -event_value + END AS net_value, + event_value + FROM player_event_raw_ledger + WHERE {where_sql} + AND killer_player_key IS NOT NULL + AND victim_player_key IS NOT NULL + ) + SELECT + player_a_key, + player_a_name, + player_b_key, + player_b_name, + COALESCE(SUM(event_value), 0) AS total_encounters, + COALESCE(SUM(net_value), 0) AS net_duel_value + FROM duel_pairs + GROUP BY player_a_key, player_a_name, player_b_key, player_b_name + ORDER BY ABS(net_duel_value) DESC, total_encounters DESC, player_a_name ASC, player_b_name ASC + LIMIT ? + """, + [*params, limit], + ).fetchall() + return [dict(row) for row in rows] + + +def list_weapon_kills( + *, + server_slug: str | None = None, + month: str | None = None, + external_match_id: str | None = None, + limit: int = 10, + db_path: Path | None = None, +) -> list[dict[str, object]]: + """Return partial weapon summaries derived from top kill events.""" + resolved_path = initialize_player_event_storage(db_path=db_path) + where_sql, params = _build_common_where( + event_type="player_kill_summary", + server_slug=server_slug, + month=month, + external_match_id=external_match_id, + ) + with _connect(resolved_path) as connection: + rows = connection.execute( + f""" + SELECT + killer_player_key, + killer_display_name, + COALESCE(weapon_name, 'unknown') AS weapon_name, + COALESCE(SUM(event_value), 0) AS total_kills + FROM player_event_raw_ledger + WHERE {where_sql} + AND killer_player_key IS NOT NULL + GROUP BY killer_player_key, killer_display_name, COALESCE(weapon_name, 'unknown') + ORDER BY total_kills DESC, killer_display_name ASC, weapon_name ASC + LIMIT ? + """, + [*params, limit], + ).fetchall() + return [dict(row) for row in rows] + + +def list_teamkill_summaries( + *, + server_slug: str | None = None, + month: str | None = None, + external_match_id: str | None = None, + limit: int = 10, + db_path: Path | None = None, +) -> list[dict[str, object]]: + """Return derived teamkill totals per player from the raw ledger.""" + resolved_path = initialize_player_event_storage(db_path=db_path) + where_sql, params = _build_common_where( + event_type="player_teamkill_summary", + server_slug=server_slug, + month=month, + external_match_id=external_match_id, + ) + with _connect(resolved_path) as connection: + rows = connection.execute( + f""" + SELECT + killer_player_key, + killer_display_name, + COALESCE(SUM(event_value), 0) AS total_teamkills + FROM player_event_raw_ledger + WHERE {where_sql} + AND killer_player_key IS NOT NULL + GROUP BY killer_player_key, killer_display_name + ORDER BY total_teamkills DESC, killer_display_name ASC + LIMIT ? + """, + [*params, limit], + ).fetchall() + return [dict(row) for row in rows] + + +def _query_pair_summary( + *, + event_type: str, + server_slug: str | None, + month: str | None, + external_match_id: str | None, + limit: int, + db_path: Path | None, +) -> list[dict[str, object]]: + resolved_path = initialize_player_event_storage(db_path=db_path) + where_sql, params = _build_common_where( + event_type=event_type, + server_slug=server_slug, + month=month, + external_match_id=external_match_id, + ) + with _connect(resolved_path) as connection: + rows = connection.execute( + f""" + SELECT + killer_player_key, + killer_display_name, + victim_player_key, + victim_display_name, + COALESCE(SUM(event_value), 0) AS total_kills + FROM player_event_raw_ledger + WHERE {where_sql} + AND killer_player_key IS NOT NULL + AND victim_player_key IS NOT NULL + GROUP BY killer_player_key, killer_display_name, victim_player_key, victim_display_name + ORDER BY total_kills DESC, killer_display_name ASC, victim_display_name ASC + LIMIT ? + """, + [*params, limit], + ).fetchall() + return [dict(row) for row in rows] + + +def _build_common_where( + *, + event_type: str, + server_slug: str | None, + month: str | None, + external_match_id: str | None, +) -> tuple[str, list[object]]: + clauses = ["event_type = ?"] + params: list[object] = [event_type] + + if server_slug and server_slug != "all-servers": + clauses.append("server_slug = ?") + params.append(server_slug.strip()) + if month: + clauses.append("substr(COALESCE(occurred_at, ''), 1, 7) = ?") + params.append(month.strip()) + if external_match_id: + clauses.append("external_match_id = ?") + params.append(external_match_id.strip()) + + return " AND ".join(clauses), params + + +def _connect(db_path: Path) -> sqlite3.Connection: + connection = sqlite3.connect(db_path or get_storage_path()) + connection.row_factory = sqlite3.Row + return connection diff --git a/backend/app/player_event_models.py b/backend/app/player_event_models.py new file mode 100644 index 0000000..182ed65 --- /dev/null +++ b/backend/app/player_event_models.py @@ -0,0 +1,32 @@ +"""Normalized player event models for the V2 event pipeline foundation.""" + +from __future__ import annotations + +from dataclasses import asdict, dataclass + + +@dataclass(frozen=True, slots=True) +class PlayerEventRecord: + """Minimal normalized player event contract reused across source and storage.""" + + event_id: str + event_type: str + occurred_at: str | None + server_slug: str + external_match_id: str + source_kind: str + source_ref: str | None + raw_event_ref: str | None + killer_player_key: str | None + killer_display_name: str | None + victim_player_key: str | None + victim_display_name: str | None + weapon_name: str | None + weapon_category: str | None + kill_category: str | None + is_teamkill: bool + event_value: int = 1 + + def to_dict(self) -> dict[str, object]: + """Return the event as a plain dictionary.""" + return asdict(self) diff --git a/backend/app/player_event_source.py b/backend/app/player_event_source.py new file mode 100644 index 0000000..3f933f2 --- /dev/null +++ b/backend/app/player_event_source.py @@ -0,0 +1,62 @@ +"""Player event source selection and contracts for the V2 pipeline.""" + +from __future__ import annotations + +from typing import Protocol + +from .config import get_historical_data_source_kind +from .player_event_models import PlayerEventRecord +from .providers.player_event_source_provider import PublicScoreboardPlayerEventSource + + +class PlayerEventSource(Protocol): + """Contract for adapters that normalize player event signals.""" + + source_kind: str + + def extract_match_events( + self, + *, + server_slug: str, + match_payload: dict[str, object], + source_ref: str | None = None, + ) -> list[PlayerEventRecord]: + """Normalize one match payload into reusable player event records.""" + + def describe_scope(self) -> dict[str, object]: + """Describe what the adapter can and cannot capture today.""" + + +class RconPlayerEventSource: + """Placeholder adapter for a future raw RCON/log feed.""" + + source_kind = "rcon-events" + + def extract_match_events( + self, + *, + server_slug: str, + match_payload: dict[str, object], + source_ref: str | None = None, + ) -> list[PlayerEventRecord]: + raise RuntimeError("Raw RCON player event extraction is not implemented yet.") + + def describe_scope(self) -> dict[str, object]: + return { + "source_kind": self.source_kind, + "supports_raw_kill_events": False, + "captures": [], + "limitations": [ + "No raw RCON event or log feed is integrated in this repository yet.", + ], + } + + +def get_player_event_source() -> PlayerEventSource: + """Select the event adapter that best matches the configured historical source.""" + source_kind = get_historical_data_source_kind() + if source_kind == "public-scoreboard": + return PublicScoreboardPlayerEventSource() + if source_kind == "rcon": + return RconPlayerEventSource() + raise ValueError(f"Unsupported player event source: {source_kind}") diff --git a/backend/app/player_event_storage.py b/backend/app/player_event_storage.py new file mode 100644 index 0000000..9074b9a --- /dev/null +++ b/backend/app/player_event_storage.py @@ -0,0 +1,413 @@ +"""Raw storage and run tracking for the V2 player event pipeline.""" + +from __future__ import annotations + +import sqlite3 +from collections.abc import Iterable +from datetime import datetime, timezone +from pathlib import Path + +from .config import get_storage_path +from .player_event_models import PlayerEventRecord + + +def initialize_player_event_storage(*, db_path: Path | None = None) -> Path: + """Create the append-only player event ledger and its worker metadata tables.""" + resolved_path = db_path or get_storage_path() + resolved_path.parent.mkdir(parents=True, exist_ok=True) + + with _connect(resolved_path) as connection: + connection.executescript( + """ + CREATE TABLE IF NOT EXISTS player_event_raw_ledger ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + event_id TEXT NOT NULL UNIQUE, + event_type TEXT NOT NULL, + occurred_at TEXT, + server_slug TEXT NOT NULL, + external_match_id TEXT NOT NULL, + source_kind TEXT NOT NULL, + source_ref TEXT, + raw_event_ref TEXT, + killer_player_key TEXT, + killer_display_name TEXT, + victim_player_key TEXT, + victim_display_name TEXT, + weapon_name TEXT, + weapon_category TEXT, + kill_category TEXT, + is_teamkill INTEGER NOT NULL DEFAULT 0, + event_value INTEGER NOT NULL DEFAULT 1, + inserted_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP + ); + + CREATE TABLE IF NOT EXISTS player_event_ingestion_runs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + mode TEXT NOT NULL, + status TEXT NOT NULL, + target_server_slug TEXT, + started_at TEXT NOT NULL, + completed_at TEXT, + pages_processed INTEGER NOT NULL DEFAULT 0, + matches_seen INTEGER NOT NULL DEFAULT 0, + matches_fetched INTEGER NOT NULL DEFAULT 0, + events_inserted INTEGER NOT NULL DEFAULT 0, + duplicate_events INTEGER NOT NULL DEFAULT 0, + notes TEXT, + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP + ); + + CREATE TABLE IF NOT EXISTS player_event_backfill_progress ( + server_slug TEXT NOT NULL, + mode TEXT NOT NULL, + next_page INTEGER NOT NULL DEFAULT 1, + last_completed_page INTEGER, + cutoff_occurred_at TEXT, + discovered_total_matches INTEGER, + archive_exhausted INTEGER NOT NULL DEFAULT 0, + last_run_id INTEGER, + last_run_status TEXT, + last_run_started_at TEXT, + last_run_completed_at TEXT, + last_error TEXT, + updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (server_slug, mode) + ); + + CREATE INDEX IF NOT EXISTS idx_player_event_raw_server_match + ON player_event_raw_ledger(server_slug, external_match_id); + + CREATE INDEX IF NOT EXISTS idx_player_event_raw_occurred_at + ON player_event_raw_ledger(occurred_at DESC); + + CREATE INDEX IF NOT EXISTS idx_player_event_raw_killer_victim + ON player_event_raw_ledger(killer_player_key, victim_player_key); + """ + ) + + return resolved_path + + +def upsert_player_events( + events: Iterable[PlayerEventRecord], + *, + db_path: Path | None = None, +) -> dict[str, int]: + """Insert normalized events idempotently into the raw ledger.""" + resolved_path = initialize_player_event_storage(db_path=db_path) + inserted = 0 + duplicates = 0 + with _connect(resolved_path) as connection: + for event in events: + cursor = connection.execute( + """ + INSERT OR IGNORE INTO player_event_raw_ledger ( + event_id, + event_type, + occurred_at, + server_slug, + external_match_id, + source_kind, + source_ref, + raw_event_ref, + killer_player_key, + killer_display_name, + victim_player_key, + victim_display_name, + weapon_name, + weapon_category, + kill_category, + is_teamkill, + event_value + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + event.event_id, + event.event_type, + event.occurred_at, + event.server_slug, + event.external_match_id, + event.source_kind, + event.source_ref, + event.raw_event_ref, + event.killer_player_key, + event.killer_display_name, + event.victim_player_key, + event.victim_display_name, + event.weapon_name, + event.weapon_category, + event.kill_category, + 1 if event.is_teamkill else 0, + max(1, int(event.event_value)), + ), + ) + if int(cursor.rowcount or 0) > 0: + inserted += 1 + else: + duplicates += 1 + return { + "events_inserted": inserted, + "duplicate_events": duplicates, + } + + +def start_player_event_ingestion_run( + *, + mode: str, + target_server_slug: str | None = None, + db_path: Path | None = None, +) -> int: + """Persist one player event ingestion attempt.""" + resolved_path = initialize_player_event_storage(db_path=db_path) + with _connect(resolved_path) as connection: + cursor = connection.execute( + """ + INSERT INTO player_event_ingestion_runs ( + mode, + status, + target_server_slug, + started_at + ) VALUES (?, 'running', ?, ?) + """, + (mode, target_server_slug, _utc_now_iso()), + ) + return int(cursor.lastrowid) + + +def finalize_player_event_ingestion_run( + run_id: int, + *, + status: str, + pages_processed: int, + matches_seen: int, + matches_fetched: int, + events_inserted: int, + duplicate_events: int, + notes: str | None = None, + db_path: Path | None = None, +) -> None: + """Update one player event ingestion attempt with final counters.""" + resolved_path = initialize_player_event_storage(db_path=db_path) + with _connect(resolved_path) as connection: + connection.execute( + """ + UPDATE player_event_ingestion_runs + SET status = ?, + completed_at = ?, + pages_processed = ?, + matches_seen = ?, + matches_fetched = ?, + events_inserted = ?, + duplicate_events = ?, + notes = ? + WHERE id = ? + """, + ( + status, + _utc_now_iso(), + pages_processed, + matches_seen, + matches_fetched, + events_inserted, + duplicate_events, + notes, + run_id, + ), + ) + + +def mark_player_event_progress_started( + *, + server_slug: str, + mode: str, + run_id: int, + cutoff_occurred_at: str | None, + db_path: Path | None = None, +) -> None: + """Persist the start state for one server ingestion attempt.""" + resolved_path = initialize_player_event_storage(db_path=db_path) + with _connect(resolved_path) as connection: + connection.execute( + """ + INSERT INTO player_event_backfill_progress ( + server_slug, + mode, + next_page, + cutoff_occurred_at, + archive_exhausted, + last_run_id, + last_run_status, + last_run_started_at, + last_run_completed_at, + last_error + ) VALUES (?, ?, 1, ?, 0, ?, 'running', ?, NULL, NULL) + ON CONFLICT(server_slug, mode) DO UPDATE SET + cutoff_occurred_at = excluded.cutoff_occurred_at, + last_run_id = excluded.last_run_id, + last_run_status = excluded.last_run_status, + last_run_started_at = excluded.last_run_started_at, + last_run_completed_at = NULL, + last_error = NULL, + updated_at = CURRENT_TIMESTAMP + """, + (server_slug, mode, cutoff_occurred_at, run_id, _utc_now_iso()), + ) + + +def mark_player_event_progress_page_completed( + *, + server_slug: str, + mode: str, + page_number: int, + discovered_total_matches: int | None, + run_id: int, + db_path: Path | None = None, +) -> None: + """Advance the resume checkpoint after one page completes successfully.""" + resolved_path = initialize_player_event_storage(db_path=db_path) + with _connect(resolved_path) as connection: + connection.execute( + """ + INSERT INTO player_event_backfill_progress ( + server_slug, + mode, + next_page, + last_completed_page, + discovered_total_matches, + archive_exhausted, + last_run_id, + last_run_status, + last_run_started_at, + last_run_completed_at, + last_error + ) VALUES (?, ?, ?, ?, ?, 0, ?, 'running', ?, NULL, NULL) + ON CONFLICT(server_slug, mode) DO UPDATE SET + next_page = excluded.next_page, + last_completed_page = excluded.last_completed_page, + discovered_total_matches = COALESCE( + excluded.discovered_total_matches, + player_event_backfill_progress.discovered_total_matches + ), + archive_exhausted = 0, + last_run_id = excluded.last_run_id, + last_run_status = excluded.last_run_status, + last_run_started_at = excluded.last_run_started_at, + last_run_completed_at = NULL, + last_error = NULL, + updated_at = CURRENT_TIMESTAMP + """, + ( + server_slug, + mode, + page_number + 1, + page_number, + discovered_total_matches, + run_id, + _utc_now_iso(), + ), + ) + + +def finalize_player_event_progress( + *, + server_slug: str, + mode: str, + run_id: int, + status: str, + archive_exhausted: bool = False, + error_message: str | None = None, + db_path: Path | None = None, +) -> None: + """Persist the final state of one server event ingestion attempt.""" + resolved_path = initialize_player_event_storage(db_path=db_path) + with _connect(resolved_path) as connection: + connection.execute( + """ + INSERT INTO player_event_backfill_progress ( + server_slug, + mode, + next_page, + archive_exhausted, + last_run_id, + last_run_status, + last_run_started_at, + last_run_completed_at, + last_error + ) VALUES (?, ?, 1, ?, ?, ?, ?, ?, ?) + ON CONFLICT(server_slug, mode) DO UPDATE SET + archive_exhausted = CASE + WHEN excluded.last_run_status = 'success' AND excluded.archive_exhausted = 1 + THEN 1 + ELSE player_event_backfill_progress.archive_exhausted + END, + last_run_id = excluded.last_run_id, + last_run_status = excluded.last_run_status, + last_run_started_at = COALESCE( + player_event_backfill_progress.last_run_started_at, + excluded.last_run_started_at + ), + last_run_completed_at = excluded.last_run_completed_at, + last_error = excluded.last_error, + updated_at = CURRENT_TIMESTAMP + """, + ( + server_slug, + mode, + 1 if archive_exhausted else 0, + run_id, + status, + _utc_now_iso(), + _utc_now_iso(), + error_message, + ), + ) + + +def get_player_event_resume_page( + server_slug: str, + *, + mode: str = "bootstrap", + db_path: Path | None = None, +) -> int: + """Return the saved resume page for a bootstrap-like event backfill.""" + resolved_path = initialize_player_event_storage(db_path=db_path) + with _connect(resolved_path) as connection: + row = connection.execute( + """ + SELECT next_page + FROM player_event_backfill_progress + WHERE server_slug = ? AND mode = ? + """, + (server_slug, mode), + ).fetchone() + return max(1, int(row["next_page"])) if row and row["next_page"] else 1 + + +def get_player_event_refresh_cutoff_for_server( + server_slug: str, + *, + db_path: Path | None = None, +) -> str | None: + """Return the latest occurred_at already persisted for one server.""" + resolved_path = initialize_player_event_storage(db_path=db_path) + with _connect(resolved_path) as connection: + row = connection.execute( + """ + SELECT MAX(occurred_at) AS latest_occurred_at + FROM player_event_raw_ledger + WHERE server_slug = ? + """, + (server_slug,), + ).fetchone() + return str(row["latest_occurred_at"]) if row and row["latest_occurred_at"] else None + + +def _connect(db_path: Path) -> sqlite3.Connection: + connection = sqlite3.connect(db_path) + connection.row_factory = sqlite3.Row + connection.execute("PRAGMA foreign_keys = ON") + return connection + + +def _utc_now_iso() -> str: + return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") diff --git a/backend/app/player_event_worker.py b/backend/app/player_event_worker.py new file mode 100644 index 0000000..d53eea3 --- /dev/null +++ b/backend/app/player_event_worker.py @@ -0,0 +1,462 @@ +"""Incremental worker for the V2 player event ingestion pipeline.""" + +from __future__ import annotations + +import argparse +import json +import time +from dataclasses import dataclass +from typing import Iterable + +from .config import ( + get_historical_crcon_detail_workers, + get_historical_crcon_page_size, + get_player_event_refresh_interval_seconds, + get_player_event_refresh_max_retries, + get_player_event_refresh_retry_delay_seconds, +) +from .data_sources import get_historical_data_source +from .historical_storage import list_historical_servers +from .player_event_source import get_player_event_source +from .player_event_storage import ( + finalize_player_event_ingestion_run, + finalize_player_event_progress, + get_player_event_refresh_cutoff_for_server, + get_player_event_resume_page, + initialize_player_event_storage, + mark_player_event_progress_page_completed, + mark_player_event_progress_started, + start_player_event_ingestion_run, + upsert_player_events, +) + + +@dataclass(slots=True) +class PlayerEventIngestionStats: + pages_processed: int = 0 + matches_seen: int = 0 + matches_fetched: int = 0 + events_inserted: int = 0 + duplicate_events: int = 0 + + def apply(self, delta: dict[str, int]) -> None: + self.events_inserted += int(delta.get("events_inserted", 0)) + self.duplicate_events += int(delta.get("duplicate_events", 0)) + + +def run_player_event_refresh( + *, + server_slug: str | None = None, + max_pages: int | None = None, + page_size: int | None = None, + start_page: int | None = None, + detail_workers: int | None = None, +) -> dict[str, object]: + """Refresh recent player event summaries from the configured historical source.""" + initialize_player_event_storage() + data_source = get_historical_data_source() + event_source = get_player_event_source() + resolved_page_size = page_size or get_historical_crcon_page_size() + resolved_detail_workers = detail_workers or get_historical_crcon_detail_workers() + selected_servers = _select_servers(server_slug) + processed_servers: list[dict[str, object]] = [] + active_runs: dict[str, int] = {} + + try: + for server in selected_servers: + current_server_slug = str(server["slug"]) + run_id = start_player_event_ingestion_run( + mode="refresh", + target_server_slug=current_server_slug, + ) + active_runs[current_server_slug] = run_id + cutoff = get_player_event_refresh_cutoff_for_server(current_server_slug) + mark_player_event_progress_started( + server_slug=current_server_slug, + mode="refresh", + run_id=run_id, + cutoff_occurred_at=cutoff, + ) + server_stats = _ingest_server( + server=server, + run_id=run_id, + data_source=data_source, + event_source=event_source, + page_size=resolved_page_size, + max_pages=max_pages, + start_page=_resolve_start_page( + server_slug=current_server_slug, + start_page=start_page, + ), + detail_workers=resolved_detail_workers, + cutoff=cutoff, + ) + finalize_player_event_ingestion_run( + run_id, + status="success", + pages_processed=server_stats["pages_processed"], + matches_seen=server_stats["matches_seen"], + matches_fetched=server_stats["matches_fetched"], + events_inserted=server_stats["events_inserted"], + duplicate_events=server_stats["duplicate_events"], + notes=f"source={data_source.source_kind};adapter={event_source.source_kind}", + ) + finalize_player_event_progress( + server_slug=current_server_slug, + mode="refresh", + run_id=run_id, + status="success", + archive_exhausted=bool(server_stats["archive_exhausted"]), + ) + processed_servers.append(server_stats) + active_runs.pop(current_server_slug, None) + except Exception as exc: + for active_server_slug, run_id in active_runs.items(): + finalize_player_event_ingestion_run( + run_id, + status="failed", + pages_processed=0, + matches_seen=0, + matches_fetched=0, + events_inserted=0, + duplicate_events=0, + notes=str(exc), + ) + finalize_player_event_progress( + server_slug=active_server_slug, + mode="refresh", + run_id=run_id, + status="failed", + error_message=str(exc), + ) + raise + + return { + "status": "ok", + "mode": "refresh", + "source_provider": data_source.source_kind, + "event_adapter": event_source.source_kind, + "page_size": resolved_page_size, + "detail_workers": resolved_detail_workers, + "scope": event_source.describe_scope(), + "servers": processed_servers, + } + + +def run_periodic_player_event_refresh( + *, + interval_seconds: int, + max_retries: int, + retry_delay_seconds: int, + server_slug: str | None = None, + max_pages: int | None = None, + page_size: int | None = None, + detail_workers: int | None = None, + max_runs: int | None = None, +) -> None: + """Run the refresh worker repeatedly with bounded retries.""" + completed_runs = 0 + print( + json.dumps( + { + "event": "player-event-refresh-loop-started", + "interval_seconds": interval_seconds, + "max_retries": max_retries, + "retry_delay_seconds": retry_delay_seconds, + "server_scope": [server_slug] if server_slug else [server["slug"] for server in list_historical_servers()], + }, + indent=2, + ) + ) + print("Press Ctrl+C to stop.") + + try: + while max_runs is None or completed_runs < max_runs: + completed_runs += 1 + payload = _run_refresh_with_retries( + max_retries=max_retries, + retry_delay_seconds=retry_delay_seconds, + server_slug=server_slug, + max_pages=max_pages, + page_size=page_size, + detail_workers=detail_workers, + ) + print(json.dumps({"run": completed_runs, **payload}, indent=2)) + if max_runs is not None and completed_runs >= max_runs: + break + time.sleep(interval_seconds) + except KeyboardInterrupt: + print("\nPlayer event refresh loop stopped by user.") + + +def _run_refresh_with_retries( + *, + max_retries: int, + retry_delay_seconds: int, + server_slug: str | None, + max_pages: int | None, + page_size: int | None, + detail_workers: int | None, +) -> dict[str, object]: + attempt = 0 + while True: + attempt += 1 + try: + return { + "status": "ok", + "attempts_used": attempt, + "refresh_result": run_player_event_refresh( + server_slug=server_slug, + max_pages=max_pages, + page_size=page_size, + detail_workers=detail_workers, + ), + } + except Exception as exc: + if attempt > max_retries: + return { + "status": "error", + "attempts_used": attempt, + "error": str(exc), + } + if retry_delay_seconds > 0: + time.sleep(retry_delay_seconds) + + +def _ingest_server( + *, + server: dict[str, object], + run_id: int, + data_source: object, + event_source: object, + page_size: int, + max_pages: int | None, + start_page: int, + detail_workers: int, + cutoff: str | None, +) -> dict[str, object]: + page_limit = max_pages or 1000000 + local_stats = PlayerEventIngestionStats() + discovered_total_matches: int | None = None + archive_exhausted = False + + for page_number in range(start_page, start_page + page_limit): + payload = data_source.fetch_match_page( + base_url=str(server["scoreboard_base_url"]), + page=page_number, + limit=page_size, + ) + if discovered_total_matches is None: + discovered_total_matches = _coerce_int(payload.get("total")) + page_matches = _coerce_match_list(payload.get("maps")) + if not page_matches: + archive_exhausted = True + break + + local_stats.pages_processed += 1 + stop_after_page = False + match_ids_to_fetch: list[str] = [] + + for match_summary in page_matches: + local_stats.matches_seen += 1 + reference_timestamp = _pick_match_timestamp(match_summary) + if cutoff and reference_timestamp and reference_timestamp < cutoff: + stop_after_page = True + continue + + match_id = _stringify(match_summary.get("id")) + if match_id: + match_ids_to_fetch.append(match_id) + + detail_payloads = data_source.fetch_match_details( + base_url=str(server["scoreboard_base_url"]), + match_ids=match_ids_to_fetch, + max_workers=detail_workers, + ) + local_stats.matches_fetched += len(detail_payloads) + for detail_payload in detail_payloads: + match_id = _stringify(detail_payload.get("id")) or "unknown" + source_ref = ( + f"{server['scoreboard_base_url']}/api/get_map_scoreboard?map_id={match_id}" + ) + normalized_events = event_source.extract_match_events( + server_slug=str(server["slug"]), + match_payload=detail_payload, + source_ref=source_ref, + ) + local_stats.apply(upsert_player_events(normalized_events)) + + mark_player_event_progress_page_completed( + server_slug=str(server["slug"]), + mode="refresh", + page_number=page_number, + discovered_total_matches=discovered_total_matches, + run_id=run_id, + ) + + if stop_after_page: + break + + return { + "server_slug": server["slug"], + "source_provider": data_source.source_kind, + "event_adapter": event_source.source_kind, + "pages_processed": local_stats.pages_processed, + "matches_seen": local_stats.matches_seen, + "matches_fetched": local_stats.matches_fetched, + "events_inserted": local_stats.events_inserted, + "duplicate_events": local_stats.duplicate_events, + "cutoff": cutoff, + "archive_exhausted": archive_exhausted, + "discovered_total_matches": discovered_total_matches, + } + + +def _resolve_start_page(*, server_slug: str, start_page: int | None) -> int: + if start_page is not None: + return max(1, start_page) + return get_player_event_resume_page(server_slug, mode="refresh") + + +def _select_servers(server_slug: str | None) -> list[dict[str, object]]: + servers = list_historical_servers() + if server_slug is None: + return servers + normalized = server_slug.strip() + selected = [server for server in servers if server["slug"] == normalized] + if not selected: + raise ValueError(f"Unknown historical server slug: {server_slug}") + return selected + + +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 _pick_match_timestamp(match_payload: dict[str, object]) -> str | None: + for key in ("end", "start", "creation_time"): + value = match_payload.get(key) + if isinstance(value, str) and value.strip(): + return value.strip() + return None + + +def _stringify(value: object) -> str | None: + if value is None: + return None + text = str(value).strip() + return text or None + + +def _coerce_int(value: object) -> int | None: + if value in (None, ""): + return None + try: + return int(value) + except (TypeError, ValueError): + return None + + +def build_arg_parser() -> argparse.ArgumentParser: + """Create the CLI parser for manual or periodic player event ingestion.""" + parser = argparse.ArgumentParser( + description="Player event refresh worker for HLL Vietnam.", + ) + parser.add_argument( + "mode", + choices=("refresh", "loop"), + help="refresh runs once; loop keeps the worker running periodically", + ) + parser.add_argument( + "--server", + dest="server_slug", + help="optional historical server slug", + ) + parser.add_argument( + "--max-pages", + type=int, + help="optional page cap for local validation", + ) + parser.add_argument( + "--page-size", + type=int, + help="override CRCON page size", + ) + parser.add_argument( + "--start-page", + type=int, + help="override the saved resume page", + ) + parser.add_argument( + "--detail-workers", + type=int, + help="parallel worker count for per-match detail requests", + ) + parser.add_argument( + "--interval", + type=int, + default=get_player_event_refresh_interval_seconds(), + help="seconds to wait between loop runs", + ) + parser.add_argument( + "--retries", + type=int, + default=get_player_event_refresh_max_retries(), + help="retry attempts after a failed refresh", + ) + parser.add_argument( + "--retry-delay", + type=int, + default=get_player_event_refresh_retry_delay_seconds(), + help="seconds to wait between failed attempts", + ) + parser.add_argument( + "--max-runs", + type=int, + help="optional safety cap for loop mode", + ) + return parser + + +def main(argv: Iterable[str] | None = None) -> int: + """Run the player event worker CLI.""" + parser = build_arg_parser() + args = parser.parse_args(list(argv) if argv is not None else None) + + if args.mode == "refresh": + result = run_player_event_refresh( + server_slug=args.server_slug, + max_pages=args.max_pages, + page_size=args.page_size, + start_page=args.start_page, + detail_workers=args.detail_workers, + ) + print(json.dumps(result, indent=2)) + return 0 + + if args.interval <= 0: + raise ValueError("--interval must be a positive integer.") + if args.retries < 0: + raise ValueError("--retries must be zero or positive.") + if args.retry_delay < 0: + raise ValueError("--retry-delay must be zero or positive.") + if args.max_runs is not None and args.max_runs <= 0: + raise ValueError("--max-runs must be positive when provided.") + + run_periodic_player_event_refresh( + interval_seconds=args.interval, + max_retries=args.retries, + retry_delay_seconds=args.retry_delay, + server_slug=args.server_slug, + max_pages=args.max_pages, + page_size=args.page_size, + detail_workers=args.detail_workers, + max_runs=args.max_runs, + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/backend/app/providers/player_event_source_provider.py b/backend/app/providers/player_event_source_provider.py new file mode 100644 index 0000000..023c06f --- /dev/null +++ b/backend/app/providers/player_event_source_provider.py @@ -0,0 +1,328 @@ +"""Player event adapter backed by public CRCON scoreboard match details.""" + +from __future__ import annotations + +import hashlib +from collections.abc import Mapping +from dataclasses import dataclass + +from ..player_event_models import PlayerEventRecord + + +@dataclass(frozen=True, slots=True) +class _PlayerIdentity: + stable_player_key: str + display_name: str | None + + +@dataclass(frozen=True, slots=True) +class PublicScoreboardPlayerEventSource: + """Normalize partial duel and weapon signals from CRCON match detail payloads.""" + + source_kind: str = "public-scoreboard-match-summary" + + def extract_match_events( + self, + *, + server_slug: str, + match_payload: dict[str, object], + source_ref: str | None = None, + ) -> list[PlayerEventRecord]: + match_id = _stringify(match_payload.get("id")) + if not match_id: + return [] + + occurred_at = _pick_match_timestamp(match_payload) + player_rows = _coerce_player_rows(match_payload.get("player_stats")) + if not player_rows: + return [] + + identity_index = _build_identity_index(player_rows) + events: list[PlayerEventRecord] = [] + + for player_row in player_rows: + actor = _build_player_identity(player_row) + if actor is None: + continue + + top_victim_name, top_victim_count = _extract_named_count(player_row.get("most_killed")) + top_weapon_name, _ = _extract_named_count(player_row.get("weapons")) + top_kill_type_name, _ = _extract_named_count(player_row.get("kills_by_type")) + victim = _find_identity_by_name(identity_index, top_victim_name) + if victim is not None and top_victim_count > 0: + events.append( + _build_event( + event_type="player_kill_summary", + occurred_at=occurred_at, + server_slug=server_slug, + match_id=match_id, + source_kind=self.source_kind, + source_ref=source_ref, + raw_event_ref=f"match:{match_id}:player:{actor.stable_player_key}:most-killed", + killer=actor, + victim=victim, + weapon_name=top_weapon_name, + kill_category=top_kill_type_name, + is_teamkill=False, + event_value=top_victim_count, + ) + ) + + top_killer_name, top_killer_count = _extract_named_count(player_row.get("death_by")) + death_weapon_name, _ = _extract_named_count(player_row.get("death_by_weapons")) + killer = _find_identity_by_name(identity_index, top_killer_name) + if killer is not None and top_killer_count > 0: + events.append( + _build_event( + event_type="player_death_summary", + occurred_at=occurred_at, + server_slug=server_slug, + match_id=match_id, + source_kind=self.source_kind, + source_ref=source_ref, + raw_event_ref=f"match:{match_id}:player:{actor.stable_player_key}:death-by", + killer=killer, + victim=actor, + weapon_name=death_weapon_name, + kill_category=None, + is_teamkill=False, + event_value=top_killer_count, + ) + ) + + teamkills = _coerce_int(player_row.get("teamkills")) or 0 + if teamkills > 0: + events.append( + _build_event( + event_type="player_teamkill_summary", + occurred_at=occurred_at, + server_slug=server_slug, + match_id=match_id, + source_kind=self.source_kind, + source_ref=source_ref, + raw_event_ref=f"match:{match_id}:player:{actor.stable_player_key}:teamkills", + killer=actor, + victim=None, + weapon_name=top_weapon_name, + kill_category=top_kill_type_name, + is_teamkill=True, + event_value=teamkills, + ) + ) + + return events + + def describe_scope(self) -> dict[str, object]: + return { + "source_kind": self.source_kind, + "supports_raw_kill_events": False, + "captures": [ + "Top victim per player from most_killed", + "Top killer per player from death_by", + "Top weapon hints from weapons and death_by_weapons", + "Aggregated teamkills per player and match", + ], + "limitations": [ + "The current source is match-summary data, not a true per-kill event feed.", + "occurred_at uses the match end/start timestamp, not the exact kill timestamp.", + "Only the strongest encounter and weapon signals available in the CRCON detail payload are normalized.", + "Full killer->victim ledgers, complete weapon breakdowns, and exact per-event teamkills still require a dedicated raw event/log source.", + ], + } + + +def _build_identity_index(player_rows: list[dict[str, object]]) -> dict[str, _PlayerIdentity]: + identity_index: dict[str, _PlayerIdentity] = {} + for player_row in player_rows: + identity = _build_player_identity(player_row) + if identity is None or not identity.display_name: + continue + identity_index[_normalize_name(identity.display_name)] = identity + return identity_index + + +def _build_player_identity(player_row: dict[str, object]) -> _PlayerIdentity | None: + display_name = _stringify(player_row.get("player")) or _stringify(player_row.get("name")) + source_player_id = _stringify(player_row.get("player_id")) or _stringify(player_row.get("id")) + steam_id = _extract_steam_id(player_row.get("steaminfo")) + stable_player_key = _build_stable_player_key(steam_id=steam_id, source_player_id=source_player_id) + if stable_player_key is None: + return None + return _PlayerIdentity( + stable_player_key=stable_player_key, + display_name=display_name or stable_player_key, + ) + + +def _find_identity_by_name( + identity_index: dict[str, _PlayerIdentity], + player_name: str | None, +) -> _PlayerIdentity | None: + if not player_name: + return None + return identity_index.get(_normalize_name(player_name)) + + +def _build_event( + *, + event_type: str, + occurred_at: str | None, + server_slug: str, + match_id: str, + source_kind: str, + source_ref: str | None, + raw_event_ref: str, + killer: _PlayerIdentity | None, + victim: _PlayerIdentity | None, + weapon_name: str | None, + kill_category: str | None, + is_teamkill: bool, + event_value: int, +) -> PlayerEventRecord: + event_id = _build_event_id( + event_type=event_type, + occurred_at=occurred_at, + server_slug=server_slug, + match_id=match_id, + killer_player_key=killer.stable_player_key if killer else None, + victim_player_key=victim.stable_player_key if victim else None, + weapon_name=weapon_name, + is_teamkill=is_teamkill, + event_value=event_value, + ) + return PlayerEventRecord( + event_id=event_id, + event_type=event_type, + occurred_at=occurred_at, + server_slug=server_slug, + external_match_id=match_id, + source_kind=source_kind, + source_ref=source_ref, + raw_event_ref=raw_event_ref, + killer_player_key=killer.stable_player_key if killer else None, + killer_display_name=killer.display_name if killer else None, + victim_player_key=victim.stable_player_key if victim else None, + victim_display_name=victim.display_name if victim else None, + weapon_name=weapon_name, + weapon_category=None, + kill_category=kill_category, + is_teamkill=is_teamkill, + event_value=max(1, event_value), + ) + + +def _build_event_id( + *, + event_type: str, + occurred_at: str | None, + server_slug: str, + match_id: str, + killer_player_key: str | None, + victim_player_key: str | None, + weapon_name: str | None, + is_teamkill: bool, + event_value: int, +) -> str: + raw_key = "|".join( + [ + event_type, + occurred_at or "", + server_slug, + match_id, + killer_player_key or "", + victim_player_key or "", + weapon_name or "", + "1" if is_teamkill else "0", + str(event_value), + ] + ) + return hashlib.sha1(raw_key.encode("utf-8")).hexdigest() + + +def _pick_match_timestamp(match_payload: Mapping[str, object]) -> str | None: + for key in ("end", "start", "creation_time"): + value = _stringify(match_payload.get(key)) + if value: + return value + return None + + +def _extract_named_count(value: object) -> tuple[str | None, int]: + if isinstance(value, str): + return _stringify(value), 1 + if isinstance(value, Mapping): + nested_name = None + nested_player = value.get("player") + if isinstance(nested_player, Mapping): + nested_name = _stringify(nested_player.get("name")) or _stringify(nested_player.get("player")) + name = ( + _stringify(value.get("name")) + or _stringify(value.get("player")) + or _stringify(value.get("victim")) + or _stringify(value.get("killer")) + or nested_name + ) + count = ( + _coerce_int(value.get("count")) + or _coerce_int(value.get("kills")) + or _coerce_int(value.get("deaths")) + or _coerce_int(value.get("value")) + or _coerce_int(value.get("total")) + or 1 + ) + return name, max(1, count) + if isinstance(value, list): + for item in value: + name, count = _extract_named_count(item) + if name: + return name, count + return None, 0 + + +def _extract_steam_id(value: object) -> str | None: + if isinstance(value, Mapping): + profile = value.get("profile") + if isinstance(profile, Mapping): + steam_id = _stringify(profile.get("steamid")) + if steam_id: + return steam_id + return _stringify(value.get("id")) + return None + + +def _build_stable_player_key( + *, + steam_id: str | None, + source_player_id: str | None, +) -> str | None: + if steam_id: + return f"steam:{steam_id}" + if source_player_id: + return f"crcon-player:{source_player_id}" + return None + + +def _coerce_player_rows(value: object) -> list[dict[str, object]]: + if not isinstance(value, list): + return [] + return [item for item in value if isinstance(item, dict)] + + +def _normalize_name(value: str) -> str: + return value.strip().casefold() + + +def _stringify(value: object) -> str | None: + if value is None: + return None + text = str(value).strip() + return text or None + + +def _coerce_int(value: object) -> int | None: + if value in (None, ""): + return None + try: + return int(value) + except (TypeError, ValueError): + return None