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

Process historical bootstrap and leaderboard tasks

Browse Source
This commit is contained in:
devRaGonSa
2026-03-23 15:24:49 +01:00
parent f35ce58a84
commit d18cb6d05b
13 changed files with 936 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
backend/app/historical_snapshot_storage.py
View File

@@ -3,7 +3,6 @@
from __future__ import annotations
import json
import sqlite3
from datetime import datetime, timezone
from pathlib import Path
@@ -58,8 +57,6 @@ def persist_historical_snapshot(
if _should_preserve_existing_snapshot(
incoming_payload=payload,
snapshot_type=snapshot_type,
server_key=normalized_server_key,
db_path=db_path,
existing_document=existing_document,
):
preserved_payload = existing_document.get("payload") if existing_document else payload
@@ -222,8 +219,6 @@ def _should_preserve_existing_snapshot(
*,
incoming_payload: dict[str, object] | list[object],
snapshot_type: str,
server_key: str,
db_path: Path | None,
existing_document: dict[str, object] | None,
) -> bool:
if not _is_effectively_empty_snapshot_payload(snapshot_type, incoming_payload):
@@ -233,7 +228,7 @@ def _should_preserve_existing_snapshot(
existing_document.get("payload"),
):
return True
return _historical_data_exists(server_key=server_key, db_path=db_path)
return False
def _is_effectively_empty_snapshot_payload(
@@ -254,40 +249,11 @@ def _is_effectively_empty_snapshot_payload(
items = payload.get("items")
return not isinstance(items, list) or len(items) == 0
if snapshot_type == "weekly-leaderboard":
if snapshot_type in {"weekly-leaderboard", "monthly-leaderboard"}:
items = payload.get("items")
return not isinstance(items, list) or len(items) == 0
return False
def _historical_data_exists(*, server_key: str, db_path: Path | None) -> bool:
resolved_db_path = initialize_historical_storage(db_path=db_path or get_storage_path())
where_clause = ""
params: list[object] = []
if server_key != "all-servers":
where_clause = "WHERE historical_servers.slug = ?"
params.append(server_key)
connection = sqlite3.connect(resolved_db_path)
connection.row_factory = sqlite3.Row
try:
row = connection.execute(
f"""
SELECT COUNT(*) AS matches_count
FROM historical_matches
INNER JOIN historical_servers
ON historical_servers.id = historical_matches.historical_server_id
{where_clause}
""",
params,
).fetchone()
finally:
connection.close()
return bool(row and int(row["matches_count"] or 0) > 0)
def _read_snapshot_document(snapshot_path: Path) -> dict[str, object] | None:
if not snapshot_path.exists():
return None
@@ -318,6 +284,9 @@ def _build_snapshot_filename(*, snapshot_type: str, metric: str | None) -> str:
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"
if snapshot_type == "monthly-leaderboard":
metric_suffix = "matches-over-100-kills" if metric == "matches_over_100_kills" else _slugify(metric or "unknown")
return f"monthly-{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"

64
backend/app/historical_snapshots.py
View File

@@ -9,18 +9,21 @@ from .historical_storage import (
ALL_SERVERS_SLUG,
list_historical_server_summaries,
list_historical_servers,
list_monthly_leaderboard,
list_recent_historical_matches,
list_weekly_leaderboard,
)
SNAPSHOT_TYPE_SERVER_SUMMARY = "server-summary"
SNAPSHOT_TYPE_WEEKLY_LEADERBOARD = "weekly-leaderboard"
SNAPSHOT_TYPE_MONTHLY_LEADERBOARD = "monthly-leaderboard"
SNAPSHOT_TYPE_RECENT_MATCHES = "recent-matches"
SUPPORTED_SNAPSHOT_TYPES = frozenset(
{
SNAPSHOT_TYPE_SERVER_SUMMARY,
SNAPSHOT_TYPE_WEEKLY_LEADERBOARD,
SNAPSHOT_TYPE_MONTHLY_LEADERBOARD,
SNAPSHOT_TYPE_RECENT_MATCHES,
}
)
@@ -49,6 +52,7 @@ SNAPSHOT_LEADERBOARD_METRICS = (
DEFAULT_SNAPSHOT_WINDOW = "all-time"
DEFAULT_WEEKLY_SNAPSHOT_WINDOW = "7d"
DEFAULT_MONTHLY_SNAPSHOT_WINDOW = "month"
DEFAULT_WEEKLY_LEADERBOARD_LIMIT = 10
DEFAULT_RECENT_MATCHES_LIMIT = 20
@@ -62,13 +66,18 @@ def validate_snapshot_identity(
if snapshot_type not in SUPPORTED_SNAPSHOT_TYPES:
raise ValueError(f"Unsupported historical snapshot type: {snapshot_type}")
if snapshot_type == SNAPSHOT_TYPE_WEEKLY_LEADERBOARD:
if snapshot_type in {
SNAPSHOT_TYPE_WEEKLY_LEADERBOARD,
SNAPSHOT_TYPE_MONTHLY_LEADERBOARD,
}:
if metric not in SUPPORTED_LEADERBOARD_METRICS:
raise ValueError(f"Unsupported historical snapshot metric: {metric}")
return
if metric is not None:
raise ValueError(f"Metric is only supported for {SNAPSHOT_TYPE_WEEKLY_LEADERBOARD}.")
raise ValueError(
"Metric is only supported for weekly-leaderboard and monthly-leaderboard."
)
def list_snapshot_server_keys(*, db_path: Path | None = None) -> list[str]:
@@ -104,6 +113,15 @@ def build_historical_server_snapshots(
db_path=db_path,
)
)
snapshots.append(
_build_monthly_leaderboard_snapshot(
server_key,
metric,
generated_at_value,
limit=leaderboard_limit,
db_path=db_path,
)
)
snapshots.append(
_build_recent_matches_snapshot(
@@ -141,6 +159,15 @@ def build_priority_historical_snapshots(
db_path=db_path,
)
)
snapshots.append(
_build_monthly_leaderboard_snapshot(
server_key,
metric,
generated_at_value,
limit=leaderboard_limit,
db_path=db_path,
)
)
snapshots.append(
_build_recent_matches_snapshot(
server_key,
@@ -305,6 +332,39 @@ def _build_weekly_leaderboard_snapshot(
}
def _build_monthly_leaderboard_snapshot(
server_key: str,
metric: str,
generated_at: datetime,
*,
limit: int,
db_path: Path | None = None,
) -> dict[str, object]:
leaderboard_result = list_monthly_leaderboard(
limit=limit,
server_id=server_key,
metric=metric,
db_path=db_path,
)
return {
"server_key": server_key,
"snapshot_type": SNAPSHOT_TYPE_MONTHLY_LEADERBOARD,
"metric": metric,
"window": DEFAULT_MONTHLY_SNAPSHOT_WINDOW,
"generated_at": generated_at,
"source_range_start": _parse_optional_timestamp(leaderboard_result.get("window_start")),
"source_range_end": _parse_optional_timestamp(leaderboard_result.get("window_end")),
"is_stale": False,
"payload": {
"server_key": server_key,
"metric": metric,
"limit": limit,
"generated_at": _to_iso(generated_at),
**leaderboard_result,
},
}
def _build_recent_matches_snapshot(
server_key: str,
generated_at: datetime,

228
backend/app/historical_storage.py
View File

@@ -47,6 +47,7 @@ SUPPORTED_WEEKLY_LEADERBOARD_METRICS = frozenset(
"matches_over_100_kills",
}
)
SUPPORTED_MONTHLY_LEADERBOARD_METRICS = SUPPORTED_WEEKLY_LEADERBOARD_METRICS
def initialize_historical_storage(*, db_path: Path | None = None) -> Path:
@@ -1289,6 +1290,165 @@ def list_weekly_top_kills(
}
def list_monthly_leaderboard(
*,
limit: int = 10,
server_id: str | None = None,
metric: str = "kills",
db_path: Path | None = None,
) -> dict[str, object]:
"""Return ranked monthly 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_month_start = _start_of_month(current_time)
previous_month_start = _start_of_previous_month(current_month_start)
normalized_metric = metric.strip() if isinstance(metric, str) else ""
if normalized_metric not in SUPPORTED_MONTHLY_LEADERBOARD_METRICS:
raise ValueError(f"Unsupported monthly leaderboard metric: {metric}")
monthly_window = _select_monthly_window(
server_id=server_id,
current_time=current_time,
current_month_start=current_month_start,
previous_month_start=previous_month_start,
db_path=resolved_path,
)
window_start = monthly_window["window_start"]
window_end = monthly_window["window_end"]
where_clauses = [
"historical_matches.ended_at IS NOT NULL",
"historical_matches.ended_at >= ?",
"historical_matches.ended_at < ?",
]
params: list[object] = [
window_start.isoformat().replace("+00:00", "Z"),
window_end.isoformat().replace("+00:00", "Z"),
]
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)",
"support": "COALESCE(SUM(historical_player_match_stats.support), 0)",
"matches_over_100_kills": (
"COALESCE(SUM(CASE WHEN COALESCE(historical_player_match_stats.kills, 0) >= 100 "
"THEN 1 ELSE 0 END), 0)"
),
}[normalized_metric]
with _connect(resolved_path) as connection:
rows = connection.execute(
f"""
WITH ranked_players AS (
SELECT
{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 {partition_expression}
ORDER BY
{metric_sum_expression} DESC,
COUNT(DISTINCT historical_matches.id) ASC,
historical_players.display_name ASC
) AS ranking_position
FROM historical_player_match_stats
INNER JOIN historical_matches
ON historical_matches.id = historical_player_match_stats.historical_match_id
INNER JOIN historical_servers
ON historical_servers.id = historical_matches.historical_server_id
INNER JOIN historical_players
ON historical_players.id = historical_player_match_stats.historical_player_id
WHERE {" AND ".join(where_clauses)}
GROUP BY {group_by_expression}
)
SELECT *
FROM ranked_players
WHERE ranking_position <= ?
ORDER BY server_slug ASC, ranking_position ASC
""",
[*params, limit],
).fetchall()
window_days = _calculate_window_days(window_start=window_start, window_end=window_end)
items: list[dict[str, object]] = []
for row in rows:
items.append(
{
"server": {
"slug": row["server_slug"],
"name": row["server_name"],
},
"time_range": {
"start": window_start.isoformat().replace("+00:00", "Z"),
"end": window_end.isoformat().replace("+00:00", "Z"),
"window_days": window_days,
},
"player": {
"stable_player_key": row["stable_player_key"],
"name": row["player_name"],
"steam_id": row["steam_id"],
},
"metric": normalized_metric,
"ranking_position": int(row["ranking_position"]),
"metric_value": int(row["metric_value"] or 0),
"matches_considered": int(row["matches_count"] or 0),
}
)
return {
"timeframe": "monthly",
"metric": normalized_metric,
"window_start": window_start.isoformat().replace("+00:00", "Z"),
"window_end": window_end.isoformat().replace("+00:00", "Z"),
"window_days": window_days,
"window_kind": monthly_window["window_kind"],
"window_label": monthly_window["window_label"],
"uses_fallback": monthly_window["uses_fallback"],
"selection_reason": monthly_window["selection_reason"],
"current_month_start": current_month_start.isoformat().replace("+00:00", "Z"),
"current_month_closed_matches": monthly_window["current_month_closed_matches"],
"previous_month_closed_matches": monthly_window["previous_month_closed_matches"],
"sufficient_sample": {
"minimum_closed_matches": monthly_window["minimum_closed_matches"],
"current_month_closed_matches": monthly_window["current_month_closed_matches"],
"current_month_has_sufficient_sample": monthly_window["current_month_has_sufficient_sample"],
"is_early_month": monthly_window["is_early_month"],
},
"items": items,
}
def _connect(db_path: Path) -> sqlite3.Connection:
connection = sqlite3.connect(db_path)
connection.row_factory = sqlite3.Row
@@ -2236,6 +2396,59 @@ def _select_weekly_window(
}
def _select_monthly_window(
*,
server_id: str | None,
current_time: datetime,
current_month_start: datetime,
previous_month_start: datetime,
db_path: Path,
) -> dict[str, object]:
current_month_closed_matches = _count_closed_matches_in_window(
server_id=server_id,
window_start=current_month_start,
window_end=current_time,
db_path=db_path,
)
previous_month_closed_matches = _count_closed_matches_in_window(
server_id=server_id,
window_start=previous_month_start,
window_end=current_month_start,
db_path=db_path,
)
is_early_month = current_time.day <= 3
uses_fallback = current_month_closed_matches <= 0 and previous_month_closed_matches > 0
if uses_fallback:
return {
"window_start": previous_month_start,
"window_end": current_month_start,
"window_kind": "previous-closed-month-fallback",
"window_label": "Mes cerrado anterior",
"uses_fallback": True,
"selection_reason": "no-current-month-matches",
"minimum_closed_matches": 1,
"current_month_closed_matches": current_month_closed_matches,
"previous_month_closed_matches": previous_month_closed_matches,
"current_month_has_sufficient_sample": False,
"is_early_month": is_early_month,
}
return {
"window_start": current_month_start,
"window_end": current_time,
"window_kind": "current-month",
"window_label": "Mes actual",
"uses_fallback": False,
"selection_reason": "current-month",
"minimum_closed_matches": 1,
"current_month_closed_matches": current_month_closed_matches,
"previous_month_closed_matches": previous_month_closed_matches,
"current_month_has_sufficient_sample": current_month_closed_matches > 0,
"is_early_month": is_early_month,
}
def _count_closed_matches_in_window(
*,
server_id: str | None,
@@ -2408,6 +2621,21 @@ def _start_of_week(value: datetime) -> datetime:
return midnight - timedelta(days=midnight.weekday())
def _start_of_month(value: datetime) -> datetime:
normalized = value.astimezone(timezone.utc)
return normalized.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
def _start_of_previous_month(value: datetime) -> datetime:
previous_day = value - timedelta(days=1)
return _start_of_month(previous_day)
def _calculate_window_days(*, window_start: datetime, window_end: datetime) -> int:
delta = window_end - window_start
return max(1, int((delta.total_seconds() + 86399) // 86400))
def _get_nested(payload: Mapping[str, object], *path: str) -> object:
current: object = payload
for key in path:

157
backend/app/payloads.py
View File

@@ -8,8 +8,10 @@ from .collector import collect_server_snapshots
from .config import get_refresh_interval_seconds
from .historical_snapshot_storage import get_historical_snapshot
from .historical_snapshots import (
DEFAULT_MONTHLY_SNAPSHOT_WINDOW,
DEFAULT_SNAPSHOT_WINDOW,
DEFAULT_WEEKLY_SNAPSHOT_WINDOW,
SNAPSHOT_TYPE_MONTHLY_LEADERBOARD,
SNAPSHOT_TYPE_RECENT_MATCHES,
SNAPSHOT_TYPE_SERVER_SUMMARY,
SNAPSHOT_TYPE_WEEKLY_LEADERBOARD,
@@ -18,6 +20,7 @@ from .historical_storage import (
ALL_SERVERS_SLUG,
get_historical_player_profile,
list_historical_server_summaries,
list_monthly_leaderboard,
list_recent_historical_matches,
list_weekly_leaderboard,
list_weekly_top_kills,
@@ -221,36 +224,38 @@ def build_weekly_top_kills_payload(
}
def build_weekly_leaderboard_payload(
def build_historical_leaderboard_payload(
*,
limit: int = 10,
server_id: str | None = None,
metric: str = "kills",
timeframe: str = "weekly",
) -> dict[str, object]:
"""Return one weekly historical leaderboard for the requested metric."""
result = list_weekly_leaderboard(limit=limit, server_id=server_id, metric=metric)
"""Return one historical leaderboard for the requested timeframe and metric."""
normalized_timeframe = timeframe.strip().lower() if isinstance(timeframe, str) else "weekly"
if normalized_timeframe == "monthly":
result = list_monthly_leaderboard(limit=limit, server_id=server_id, metric=metric)
summary_basis = "closed-matches-calendar-month"
context = "historical-monthly-leaderboard"
else:
normalized_timeframe = "weekly"
result = list_weekly_leaderboard(limit=limit, server_id=server_id, metric=metric)
summary_basis = "closed-matches-calendar-week"
context = "historical-weekly-leaderboard"
is_all_servers = server_id == ALL_SERVERS_SLUG
title_by_metric = {
"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",
"data": {
"title": title_by_metric.get(metric, "Ranking semanal por servidor"),
"context": "historical-weekly-leaderboard",
"title": _build_leaderboard_title(
metric=metric,
timeframe=normalized_timeframe,
is_all_servers=is_all_servers,
),
"context": context,
"timeframe": normalized_timeframe,
"metric": metric,
"summary_basis": "closed-matches-calendar-week",
"summary_basis": summary_basis,
"window_days": result.get("window_days", 7),
"window_start": result["window_start"],
"window_end": result["window_end"],
@@ -261,6 +266,9 @@ def build_weekly_leaderboard_payload(
"current_week_start": result.get("current_week_start"),
"current_week_closed_matches": result.get("current_week_closed_matches"),
"previous_week_closed_matches": result.get("previous_week_closed_matches"),
"current_month_start": result.get("current_month_start"),
"current_month_closed_matches": result.get("current_month_closed_matches"),
"previous_month_closed_matches": result.get("previous_month_closed_matches"),
"sufficient_sample": result.get("sufficient_sample"),
"limit": limit,
"items": result["items"],
@@ -268,6 +276,21 @@ def build_weekly_leaderboard_payload(
}
def build_weekly_leaderboard_payload(
*,
limit: int = 10,
server_id: str | None = None,
metric: str = "kills",
) -> dict[str, object]:
"""Return one weekly historical leaderboard for the requested metric."""
return build_historical_leaderboard_payload(
limit=limit,
server_id=server_id,
metric=metric,
timeframe="weekly",
)
def build_recent_historical_matches_payload(
*,
limit: int = 20,
@@ -314,52 +337,48 @@ def build_historical_server_summary_snapshot_payload(
}
def build_weekly_leaderboard_snapshot_payload(
def build_leaderboard_snapshot_payload(
*,
limit: int = 10,
server_id: str | None = None,
metric: str = "kills",
timeframe: str = "weekly",
) -> dict[str, object]:
"""Return one precomputed weekly leaderboard snapshot."""
"""Return one precomputed leaderboard snapshot for the requested timeframe."""
normalized_timeframe = timeframe.strip().lower() if isinstance(timeframe, str) else "weekly"
if normalized_timeframe == "monthly":
snapshot_type = SNAPSHOT_TYPE_MONTHLY_LEADERBOARD
window = DEFAULT_MONTHLY_SNAPSHOT_WINDOW
context = "historical-monthly-leaderboard-snapshot"
else:
normalized_timeframe = "weekly"
snapshot_type = SNAPSHOT_TYPE_WEEKLY_LEADERBOARD
window = DEFAULT_WEEKLY_SNAPSHOT_WINDOW
context = "historical-weekly-leaderboard-snapshot"
snapshot = _get_historical_snapshot_record(
server_key=server_id,
snapshot_type=SNAPSHOT_TYPE_WEEKLY_LEADERBOARD,
snapshot_type=snapshot_type,
metric=metric,
window=DEFAULT_WEEKLY_SNAPSHOT_WINDOW,
window=window,
)
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 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",
"data": {
"title": title_by_metric.get(metric, "Snapshot semanal por servidor"),
"context": "historical-weekly-leaderboard-snapshot",
"title": _build_leaderboard_title(
metric=metric,
timeframe=normalized_timeframe,
is_all_servers=is_all_servers,
snapshot=True,
),
"context": context,
"source": "historical-precomputed-snapshots",
"server_slug": server_id,
"timeframe": normalized_timeframe,
"metric": metric,
"found": snapshot is not None,
**_build_historical_snapshot_metadata(snapshot),
@@ -377,6 +396,13 @@ def build_weekly_leaderboard_snapshot_payload(
"previous_week_closed_matches": (
payload.get("previous_week_closed_matches") if isinstance(payload, dict) else None
),
"current_month_start": payload.get("current_month_start") if isinstance(payload, dict) else None,
"current_month_closed_matches": (
payload.get("current_month_closed_matches") if isinstance(payload, dict) else None
),
"previous_month_closed_matches": (
payload.get("previous_month_closed_matches") if isinstance(payload, dict) else None
),
"sufficient_sample": payload.get("sufficient_sample") if isinstance(payload, dict) else None,
"snapshot_limit": payload.get("limit") if isinstance(payload, dict) else None,
"limit": limit,
@@ -385,6 +411,21 @@ def build_weekly_leaderboard_snapshot_payload(
}
def build_weekly_leaderboard_snapshot_payload(
*,
limit: int = 10,
server_id: str | None = None,
metric: str = "kills",
) -> dict[str, object]:
"""Return one precomputed weekly leaderboard snapshot."""
return build_leaderboard_snapshot_payload(
limit=limit,
server_id=server_id,
metric=metric,
timeframe="weekly",
)
def build_recent_historical_matches_snapshot_payload(
*,
limit: int = 20,
@@ -499,6 +540,26 @@ def _build_historical_snapshot_metadata(snapshot: dict[str, object] | None) -> d
}
def _build_leaderboard_title(
*,
metric: str,
timeframe: str,
is_all_servers: bool,
snapshot: bool = False,
) -> str:
timeframe_label = "mensual" if timeframe == "monthly" else "semanal"
scope_label = "totales" if is_all_servers else "por servidor"
prefix = "Snapshot " if snapshot else ""
title_by_metric = {
"kills": f"{prefix}Top kills {timeframe_label} {scope_label}",
"deaths": f"{prefix}Top muertes {timeframe_label} {scope_label}",
"support": f"{prefix}Top puntos de soporte {timeframe_label} {scope_label}",
"matches_over_100_kills": f"{prefix}Top partidas de 100+ kills {timeframe_label} {scope_label}",
}
fallback_label = f"{prefix}Ranking {timeframe_label} por servidor".strip()
return title_by_metric.get(metric, fallback_label)
def _enrich_server_items(items: list[dict[str, object]]) -> list[dict[str, object]]:
target_index = {
target.external_server_id: target

40
backend/app/routes.py
View File

@@ -10,9 +10,11 @@ from .payloads import (
build_discord_payload,
build_error_payload,
build_health_payload,
build_historical_leaderboard_payload,
build_historical_server_summary_snapshot_payload,
build_historical_player_profile_payload,
build_historical_server_summary_payload,
build_leaderboard_snapshot_payload,
build_recent_historical_matches_snapshot_payload,
build_recent_historical_matches_payload,
build_server_detail_history_payload,
@@ -54,6 +56,25 @@ def resolve_get_payload(path: str) -> tuple[HTTPStatus | None, dict[str, object]
server_id = parse_qs(parsed.query).get("server", [None])[0]
return HTTPStatus.OK, build_weekly_top_kills_payload(limit=limit, server_id=server_id)
if parsed.path == "/api/historical/leaderboard":
limit = _parse_limit(parsed.query)
if limit is None:
return HTTPStatus.BAD_REQUEST, build_error_payload("Invalid limit parameter")
params = parse_qs(parsed.query)
server_id = params.get("server", [None])[0]
metric = params.get("metric", ["kills"])[0]
timeframe = params.get("timeframe", ["weekly"])[0]
if metric not in {"kills", "deaths", "support", "matches_over_100_kills"}:
return HTTPStatus.BAD_REQUEST, build_error_payload("Invalid metric parameter")
if timeframe not in {"weekly", "monthly"}:
return HTTPStatus.BAD_REQUEST, build_error_payload("Invalid timeframe parameter")
return HTTPStatus.OK, build_historical_leaderboard_payload(
limit=limit,
server_id=server_id,
metric=metric,
timeframe=timeframe,
)
if parsed.path == "/api/historical/weekly-leaderboard":
limit = _parse_limit(parsed.query)
if limit is None:
@@ -69,6 +90,25 @@ def resolve_get_payload(path: str) -> tuple[HTTPStatus | None, dict[str, object]
metric=metric,
)
if parsed.path == "/api/historical/snapshots/leaderboard":
limit = _parse_limit(parsed.query)
if limit is None:
return HTTPStatus.BAD_REQUEST, build_error_payload("Invalid limit parameter")
params = parse_qs(parsed.query)
server_id = params.get("server", [None])[0]
metric = params.get("metric", ["kills"])[0]
timeframe = params.get("timeframe", ["weekly"])[0]
if metric not in {"kills", "deaths", "support", "matches_over_100_kills"}:
return HTTPStatus.BAD_REQUEST, build_error_payload("Invalid metric parameter")
if timeframe not in {"weekly", "monthly"}:
return HTTPStatus.BAD_REQUEST, build_error_payload("Invalid timeframe parameter")
return HTTPStatus.OK, build_leaderboard_snapshot_payload(
limit=limit,
server_id=server_id,
metric=metric,
timeframe=timeframe,
)
if parsed.path == "/api/historical/snapshots/weekly-leaderboard":
limit = _parse_limit(parsed.query)
if limit is None: