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

Process historical snapshot tasks 047-051

Browse Source
This commit is contained in:
devRaGonSa
2026-03-23 13:29:06 +01:00
parent 1ea1d4b3c6
commit 68de2d955a
16 changed files with 939 additions and 227 deletions
Download Patch File Download Diff File Expand all files Collapse all files

280
backend/app/historical_snapshot_storage.py
View File

@@ -1,9 +1,8 @@
"""SQLite persistence for precomputed historical snapshots."""
"""File-based persistence for precomputed historical snapshots."""
from __future__ import annotations
import json
import sqlite3
from datetime import datetime, timezone
from pathlib import Path
@@ -13,41 +12,15 @@ from .historical_snapshots import validate_snapshot_identity
from .historical_storage import initialize_historical_storage
SNAPSHOT_DIRECTORY_NAME = "snapshots"
def initialize_historical_snapshot_storage(*, db_path: Path | None = None) -> Path:
"""Create the snapshot table used by precomputed historical payloads."""
resolved_path = initialize_historical_storage(db_path=db_path or get_storage_path())
with _connect(resolved_path) as connection:
connection.executescript(
"""
CREATE TABLE IF NOT EXISTS historical_precomputed_snapshots (
id INTEGER PRIMARY KEY AUTOINCREMENT,
server_key TEXT NOT NULL,
snapshot_type TEXT NOT NULL,
metric TEXT NOT NULL DEFAULT '',
window TEXT NOT NULL DEFAULT '',
payload_json TEXT NOT NULL,
generated_at TEXT NOT NULL,
source_range_start TEXT,
source_range_end TEXT,
is_stale INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
UNIQUE(server_key, snapshot_type, metric, window)
);
CREATE INDEX IF NOT EXISTS idx_historical_precomputed_snapshots_lookup
ON historical_precomputed_snapshots(
server_key,
snapshot_type,
metric,
window,
generated_at DESC
);
"""
)
return resolved_path
"""Create the snapshot directory used by precomputed historical payloads."""
resolved_db_path = initialize_historical_storage(db_path=db_path or get_storage_path())
snapshots_root = resolved_db_path.parent / SNAPSHOT_DIRECTORY_NAME
snapshots_root.mkdir(parents=True, exist_ok=True)
return snapshots_root
def persist_historical_snapshot(
@@ -63,60 +36,46 @@ def persist_historical_snapshot(
is_stale: bool = False,
db_path: Path | None = None,
) -> HistoricalSnapshotRecord:
"""Insert or replace one persisted historical snapshot."""
if not server_key.strip():
"""Insert or replace one persisted historical snapshot JSON file."""
normalized_server_key = server_key.strip()
if not normalized_server_key:
raise ValueError("server_key is required for historical snapshots.")
validate_snapshot_identity(snapshot_type=snapshot_type, metric=metric)
resolved_path = initialize_historical_snapshot_storage(db_path=db_path)
generated_at_value = generated_at or datetime.now(timezone.utc)
payload_json = json.dumps(payload, ensure_ascii=True, separators=(",", ":"))
normalized_metric = metric or ""
normalized_window = window or ""
snapshots_root = initialize_historical_snapshot_storage(db_path=db_path)
generated_at_value = _as_utc(generated_at or datetime.now(timezone.utc))
payload_json = json.dumps(payload, ensure_ascii=True)
snapshot_path = _build_snapshot_path(
snapshots_root=snapshots_root,
server_key=normalized_server_key,
snapshot_type=snapshot_type,
metric=metric,
)
snapshot_path.parent.mkdir(parents=True, exist_ok=True)
with _connect(resolved_path) as connection:
connection.execute(
"""
INSERT INTO historical_precomputed_snapshots (
server_key,
snapshot_type,
metric,
window,
payload_json,
generated_at,
source_range_start,
source_range_end,
is_stale
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(server_key, snapshot_type, metric, window)
DO UPDATE SET
payload_json = excluded.payload_json,
generated_at = excluded.generated_at,
source_range_start = excluded.source_range_start,
source_range_end = excluded.source_range_end,
is_stale = excluded.is_stale,
updated_at = CURRENT_TIMESTAMP
""",
(
server_key.strip(),
snapshot_type,
normalized_metric,
normalized_window,
payload_json,
_to_iso(generated_at_value),
_to_iso(source_range_start),
_to_iso(source_range_end),
1 if is_stale else 0,
),
)
snapshot_document = {
"server_key": normalized_server_key,
"snapshot_type": snapshot_type,
"metric": metric,
"window": window,
"generated_at": _to_iso(generated_at_value),
"source_range_start": _to_iso(source_range_start),
"source_range_end": _to_iso(source_range_end),
"is_stale": is_stale,
"payload": payload,
}
snapshot_path.write_text(
json.dumps(snapshot_document, ensure_ascii=True, indent=2) + "\n",
encoding="utf-8",
)
return HistoricalSnapshotRecord(
server_key=server_key.strip(),
server_key=normalized_server_key,
snapshot_type=snapshot_type,
metric=metric,
window=window,
payload_json=payload_json,
generated_at=_as_utc(generated_at_value),
generated_at=generated_at_value,
source_range_start=_as_utc(source_range_start),
source_range_end=_as_utc(source_range_end),
is_stale=is_stale,
@@ -158,44 +117,27 @@ def get_historical_snapshot(
) -> dict[str, object] | None:
"""Return one persisted snapshot and decoded payload, if present."""
validate_snapshot_identity(snapshot_type=snapshot_type, metric=metric)
resolved_path = initialize_historical_snapshot_storage(db_path=db_path)
with _connect(resolved_path) as connection:
row = connection.execute(
"""
SELECT
server_key,
snapshot_type,
metric,
window,
payload_json,
generated_at,
source_range_start,
source_range_end,
is_stale
FROM historical_precomputed_snapshots
WHERE server_key = ?
AND snapshot_type = ?
AND metric = ?
AND window = ?
""",
(server_key, snapshot_type, metric or "", window or ""),
).fetchone()
if row is None:
snapshots_root = initialize_historical_snapshot_storage(db_path=db_path)
snapshot_path = _build_snapshot_path(
snapshots_root=snapshots_root,
server_key=server_key,
snapshot_type=snapshot_type,
metric=metric,
)
if not snapshot_path.exists():
return None
payload = json.loads(row["payload_json"])
document = json.loads(snapshot_path.read_text(encoding="utf-8"))
return {
"server_key": row["server_key"],
"snapshot_type": row["snapshot_type"],
"metric": row["metric"] or None,
"window": row["window"] or None,
"generated_at": row["generated_at"],
"source_range_start": row["source_range_start"],
"source_range_end": row["source_range_end"],
"is_stale": bool(row["is_stale"]),
"payload": payload,
"server_key": document.get("server_key"),
"snapshot_type": document.get("snapshot_type"),
"metric": document.get("metric"),
"window": document.get("window"),
"generated_at": document.get("generated_at"),
"source_range_start": document.get("source_range_start"),
"source_range_end": document.get("source_range_end"),
"is_stale": bool(document.get("is_stale", False)),
"payload": document.get("payload"),
}
@@ -206,60 +148,74 @@ def list_historical_snapshots(
db_path: Path | None = None,
) -> list[dict[str, object]]:
"""List persisted snapshots for validation and operational inspection."""
resolved_path = initialize_historical_snapshot_storage(db_path=db_path)
where_parts: list[str] = []
params: list[object] = []
if server_key:
where_parts.append("server_key = ?")
params.append(server_key)
snapshots_root = initialize_historical_snapshot_storage(db_path=db_path)
if snapshot_type:
validate_snapshot_identity(snapshot_type=snapshot_type)
where_parts.append("snapshot_type = ?")
params.append(snapshot_type)
where_sql = ""
if where_parts:
where_sql = "WHERE " + " AND ".join(where_parts)
rows: list[dict[str, object]] = []
for snapshot_path in snapshots_root.glob("*/*.json"):
try:
document = json.loads(snapshot_path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
continue
with _connect(resolved_path) as connection:
rows = connection.execute(
f"""
SELECT
server_key,
snapshot_type,
metric,
window,
generated_at,
source_range_start,
source_range_end,
is_stale
FROM historical_precomputed_snapshots
{where_sql}
ORDER BY server_key ASC, snapshot_type ASC, generated_at DESC
""",
params,
).fetchall()
if server_key and document.get("server_key") != server_key:
continue
if snapshot_type and document.get("snapshot_type") != snapshot_type:
continue
return [
{
"server_key": row["server_key"],
"snapshot_type": row["snapshot_type"],
"metric": row["metric"] or None,
"window": row["window"] or None,
"generated_at": row["generated_at"],
"source_range_start": row["source_range_start"],
"source_range_end": row["source_range_end"],
"is_stale": bool(row["is_stale"]),
}
for row in rows
]
rows.append(
{
"server_key": document.get("server_key"),
"snapshot_type": document.get("snapshot_type"),
"metric": document.get("metric"),
"window": document.get("window"),
"generated_at": document.get("generated_at"),
"source_range_start": document.get("source_range_start"),
"source_range_end": document.get("source_range_end"),
"is_stale": bool(document.get("is_stale", False)),
}
)
return sorted(
rows,
key=lambda item: (
str(item.get("server_key") or ""),
str(item.get("snapshot_type") or ""),
str(item.get("metric") or ""),
str(item.get("generated_at") or ""),
),
)
def _connect(db_path: Path) -> sqlite3.Connection:
connection = sqlite3.connect(db_path)
connection.row_factory = sqlite3.Row
return connection
def _build_snapshot_path(
*,
snapshots_root: Path,
server_key: str,
snapshot_type: str,
metric: str | None,
) -> Path:
return snapshots_root / server_key / _build_snapshot_filename(
snapshot_type=snapshot_type,
metric=metric,
)
def _build_snapshot_filename(*, snapshot_type: str, metric: str | None) -> str:
if snapshot_type == "server-summary":
return "server-summary.json"
if snapshot_type == "recent-matches":
return "recent-matches.json"
if snapshot_type == "weekly-leaderboard":
metric_suffix = "matches-over-100-kills" if metric == "matches_over_100_kills" else _slugify(metric or "unknown")
return f"weekly-{metric_suffix}.json"
metric_suffix = _slugify(metric or "")
base_name = _slugify(snapshot_type)
return f"{base_name}-{metric_suffix}.json" if metric_suffix else f"{base_name}.json"
def _slugify(value: str) -> str:
return value.strip().replace("_", "-").replace(" ", "-").lower()
def _to_iso(value: datetime | None) -> str | None:

5
backend/app/historical_snapshots.py
View File

@@ -6,6 +6,7 @@ from datetime import datetime, timezone
from pathlib import Path
from .historical_storage import (
ALL_SERVERS_SLUG,
list_historical_server_summaries,
list_historical_servers,
list_recent_historical_matches,
@@ -65,11 +66,13 @@ def validate_snapshot_identity(
def list_snapshot_server_keys(*, db_path: Path | None = None) -> list[str]:
"""Return the historical server slugs that should receive persisted snapshots."""
return [
server_keys = [
str(item["slug"])
for item in list_historical_servers(db_path=db_path)
if item.get("slug")
]
server_keys.append(ALL_SERVERS_SLUG)
return server_keys
def build_historical_server_snapshots(

163
backend/app/historical_storage.py
View File

@@ -28,7 +28,15 @@ DEFAULT_HISTORICAL_SERVERS = (
scoreboard_base_url="https://scoreboard.comunidadhll.es:5443",
server_number=2,
),
HistoricalServerDefinition(
slug="comunidad-hispana-03",
display_name="Comunidad Hispana #03",
scoreboard_base_url="https://scoreboard.comunidadhll.es:3443",
server_number=3,
),
)
ALL_SERVERS_SLUG = "all-servers"
ALL_SERVERS_DISPLAY_NAME = "Totales / Todos"
DEFAULT_WEEKLY_WINDOW_DAYS = 7
DEFAULT_REFRESH_OVERLAP_HOURS = 12
SUPPORTED_WEEKLY_LEADERBOARD_METRICS = frozenset(
@@ -729,7 +737,7 @@ def list_recent_historical_matches(
resolved_path = initialize_historical_storage(db_path=db_path)
where_clause = ""
params: list[object] = []
if server_slug:
if server_slug and not _is_all_servers_selector(server_slug):
where_clause = "WHERE historical_servers.slug = ?"
params.append(server_slug)
params.append(limit)
@@ -797,6 +805,9 @@ def list_historical_server_summaries(
) -> list[dict[str, object]]:
"""Return aggregate historical metrics per server."""
resolved_path = initialize_historical_storage(db_path=db_path)
if _is_all_servers_selector(server_slug):
return [_build_all_servers_summary(db_path=resolved_path)]
where_clause = ""
params: list[object] = []
if server_slug:
@@ -1102,6 +1113,7 @@ def list_weekly_leaderboard(
) -> dict[str, object]:
"""Return ranked weekly leaderboard totals from persisted historical match stats."""
resolved_path = initialize_historical_storage(db_path=db_path)
aggregate_all_servers = _is_all_servers_selector(server_id)
current_time = datetime.now(timezone.utc)
current_week_start = _start_of_week(current_time)
previous_week_start = current_week_start - timedelta(days=DEFAULT_WEEKLY_WINDOW_DAYS)
@@ -1127,13 +1139,34 @@ def list_weekly_leaderboard(
window_start.isoformat().replace("+00:00", "Z"),
window_end.isoformat().replace("+00:00", "Z"),
]
if server_id:
if server_id and not aggregate_all_servers:
normalized_server_id = server_id.strip()
where_clauses.append(
"(historical_servers.slug = ? OR CAST(historical_servers.server_number AS TEXT) = ?)"
)
params.extend([normalized_server_id, normalized_server_id])
server_slug_expression = (
f"'{ALL_SERVERS_SLUG}'"
if aggregate_all_servers
else "historical_servers.slug"
)
server_name_expression = (
f"'{ALL_SERVERS_DISPLAY_NAME}'"
if aggregate_all_servers
else "historical_servers.display_name"
)
partition_expression = (
f"'{ALL_SERVERS_SLUG}'"
if aggregate_all_servers
else "historical_servers.slug"
)
group_by_expression = (
"historical_players.id"
if aggregate_all_servers
else "historical_servers.slug, historical_players.id"
)
metric_sum_expression = {
"kills": "COALESCE(SUM(historical_player_match_stats.kills), 0)",
"deaths": "COALESCE(SUM(historical_player_match_stats.deaths), 0)",
@@ -1149,15 +1182,15 @@ def list_weekly_leaderboard(
f"""
WITH ranked_players AS (
SELECT
historical_servers.slug AS server_slug,
historical_servers.display_name AS server_name,
{server_slug_expression} AS server_slug,
{server_name_expression} AS server_name,
historical_players.stable_player_key,
historical_players.display_name AS player_name,
historical_players.steam_id,
COUNT(DISTINCT historical_matches.id) AS matches_count,
{metric_sum_expression} AS metric_value,
ROW_NUMBER() OVER (
PARTITION BY historical_servers.slug
PARTITION BY {partition_expression}
ORDER BY
{metric_sum_expression} DESC,
COUNT(DISTINCT historical_matches.id) ASC,
@@ -1171,7 +1204,7 @@ def list_weekly_leaderboard(
INNER JOIN historical_players
ON historical_players.id = historical_player_match_stats.historical_player_id
WHERE {" AND ".join(where_clauses)}
GROUP BY historical_servers.slug, historical_players.id
GROUP BY {group_by_expression}
)
SELECT *
FROM ranked_players
@@ -2219,7 +2252,7 @@ def _count_closed_matches_in_window(
window_start.isoformat().replace("+00:00", "Z"),
window_end.isoformat().replace("+00:00", "Z"),
]
if server_id:
if server_id and not _is_all_servers_selector(server_id):
normalized_server_id = server_id.strip()
where_clauses.append(
"(historical_servers.slug = ? OR CAST(historical_servers.server_number AS TEXT) = ?)"
@@ -2253,6 +2286,122 @@ def _classify_coverage_status(
return "week-plus"
def _build_all_servers_summary(*, db_path: Path) -> dict[str, object]:
per_server_items = list_historical_server_summaries(db_path=db_path)
imported_matches_count = sum(int(item.get("matches_count") or 0) for item in per_server_items)
unique_players = _count_all_servers_unique_players(db_path=db_path)
total_kills = sum(int(item.get("total_kills") or 0) for item in per_server_items)
discovered_total_matches = sum(
_coerce_int(item.get("backfill", {}).get("discovered_total_matches")) or 0
for item in per_server_items
)
first_points = [
item.get("coverage", {}).get("first_match_at")
for item in per_server_items
if item.get("coverage", {}).get("first_match_at")
]
last_points = [
item.get("coverage", {}).get("last_match_at")
for item in per_server_items
if item.get("coverage", {}).get("last_match_at")
]
first_match_at = min(first_points) if first_points else None
last_match_at = max(last_points) if last_points else None
coverage_days = _calculate_coverage_days(first_match_at, last_match_at)
return {
"server": {
"slug": ALL_SERVERS_SLUG,
"name": ALL_SERVERS_DISPLAY_NAME,
},
"matches_count": imported_matches_count,
"imported_matches_count": imported_matches_count,
"unique_players": unique_players,
"total_kills": total_kills,
"map_count": _count_all_servers_maps(db_path=db_path),
"top_maps": _list_all_servers_top_maps(db_path=db_path, limit=3),
"coverage": {
"basis": "persisted-import-aggregate",
"status": _classify_coverage_status(imported_matches_count, coverage_days),
"imported_matches_count": imported_matches_count,
"discovered_total_matches": discovered_total_matches or None,
"first_match_at": first_match_at,
"last_match_at": last_match_at,
"coverage_days": coverage_days,
},
"backfill": {
"mode": "aggregate",
"server_count": len(per_server_items),
"discovered_total_matches": discovered_total_matches or None,
"remaining_matches_estimate": (
max(discovered_total_matches - imported_matches_count, 0)
if discovered_total_matches
else None
),
"archive_exhausted": all(
bool(item.get("backfill", {}).get("archive_exhausted"))
for item in per_server_items
),
"last_run": None,
},
"time_range": {
"start": first_match_at,
"end": last_match_at,
},
}
def _count_all_servers_unique_players(*, db_path: Path) -> int:
with _connect(db_path) as connection:
row = connection.execute(
"""
SELECT COUNT(DISTINCT historical_players.id) AS unique_players
FROM historical_player_match_stats
INNER JOIN historical_players
ON historical_players.id = historical_player_match_stats.historical_player_id
"""
).fetchone()
return int(row["unique_players"] or 0) if row is not None else 0
def _count_all_servers_maps(*, db_path: Path) -> int:
with _connect(db_path) as connection:
row = connection.execute(
"""
SELECT COUNT(DISTINCT COALESCE(map_pretty_name, map_name)) AS map_count
FROM historical_matches
"""
).fetchone()
return int(row["map_count"] or 0) if row is not None else 0
def _list_all_servers_top_maps(*, db_path: Path, limit: int) -> list[dict[str, object]]:
with _connect(db_path) as connection:
rows = connection.execute(
"""
SELECT
COALESCE(map_pretty_name, map_name, 'Mapa no disponible') AS map_name,
COUNT(*) AS matches_count
FROM historical_matches
GROUP BY COALESCE(map_pretty_name, map_name, 'Mapa no disponible')
ORDER BY matches_count DESC, map_name ASC
LIMIT ?
""",
(limit,),
).fetchall()
return [
{
"map_name": row["map_name"],
"matches_count": int(row["matches_count"] or 0),
}
for row in rows
]
def _is_all_servers_selector(value: str | None) -> bool:
return isinstance(value, str) and value.strip() == ALL_SERVERS_SLUG
def _start_of_week(value: datetime) -> datetime:
normalized = value.astimezone(timezone.utc)
midnight = normalized.replace(hour=0, minute=0, second=0, microsecond=0)

65
backend/app/payloads.py
View File

@@ -13,8 +13,10 @@ from .historical_snapshots import (
SNAPSHOT_TYPE_RECENT_MATCHES,
SNAPSHOT_TYPE_SERVER_SUMMARY,
SNAPSHOT_TYPE_WEEKLY_LEADERBOARD,
generate_and_persist_historical_snapshots,
)
from .historical_storage import (
ALL_SERVERS_SLUG,
get_historical_player_profile,
list_historical_server_summaries,
list_recent_historical_matches,
@@ -228,11 +230,20 @@ def build_weekly_leaderboard_payload(
) -> dict[str, object]:
"""Return one weekly historical leaderboard for the requested metric."""
result = list_weekly_leaderboard(limit=limit, server_id=server_id, metric=metric)
is_all_servers = server_id == ALL_SERVERS_SLUG
title_by_metric = {
"kills": "Top kills semanales por servidor",
"deaths": "Top muertes semanales por servidor",
"support": "Top puntos de soporte semanales por servidor",
"matches_over_100_kills": "Top partidas de 100+ kills semanales por servidor",
"kills": "Top kills semanales totales" if is_all_servers else "Top kills semanales por servidor",
"deaths": "Top muertes semanales totales" if is_all_servers else "Top muertes semanales por servidor",
"support": (
"Top puntos de soporte semanales totales"
if is_all_servers
else "Top puntos de soporte semanales por servidor"
),
"matches_over_100_kills": (
"Top partidas de 100+ kills semanales totales"
if is_all_servers
else "Top partidas de 100+ kills semanales por servidor"
),
}
return {
"status": "ok",
@@ -320,11 +331,28 @@ def build_weekly_leaderboard_snapshot_payload(
payload = snapshot.get("payload") if snapshot else {}
items = payload.get("items") if isinstance(payload, dict) else None
sliced_items = list(items[:limit]) if isinstance(items, list) else []
is_all_servers = server_id == ALL_SERVERS_SLUG
title_by_metric = {
"kills": "Snapshot semanal de top kills por servidor",
"deaths": "Snapshot semanal de top muertes por servidor",
"support": "Snapshot semanal de top soporte por servidor",
"matches_over_100_kills": "Snapshot semanal de partidas 100+ kills por servidor",
"kills": (
"Snapshot semanal de top kills totales"
if is_all_servers
else "Snapshot semanal de top kills por servidor"
),
"deaths": (
"Snapshot semanal de top muertes totales"
if is_all_servers
else "Snapshot semanal de top muertes por servidor"
),
"support": (
"Snapshot semanal de top soporte total"
if is_all_servers
else "Snapshot semanal de top soporte por servidor"
),
"matches_over_100_kills": (
"Snapshot semanal de partidas 100+ kills totales"
if is_all_servers
else "Snapshot semanal de partidas 100+ kills por servidor"
),
}
return {
"status": "ok",
@@ -397,7 +425,11 @@ def build_historical_server_summary_payload(
return {
"status": "ok",
"data": {
"title": "Cobertura historica importada por servidor",
"title": (
"Cobertura historica agregada de todos los servidores"
if server_slug == ALL_SERVERS_SLUG
else "Cobertura historica importada por servidor"
),
"context": "historical-server-summary",
"source": "historical-crcon-storage",
"summary_basis": "persisted-import",
@@ -433,6 +465,21 @@ def _get_historical_snapshot_record(
) -> dict[str, object] | None:
if not server_key:
return None
snapshot = get_historical_snapshot(
server_key=server_key,
snapshot_type=snapshot_type,
metric=metric,
window=window,
)
if snapshot is not None:
return snapshot
# Self-heal missing precomputed rows when raw historical data already exists.
try:
generate_and_persist_historical_snapshots(server_key=server_key)
except Exception:
return None
return get_historical_snapshot(
server_key=server_key,
snapshot_type=snapshot_type,