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

feat: resolve scoreboard links and enrich match details

Browse Source 
Implement trusted scoreboard origins, persisted scoreboard links, RCON-to-scoreboard correlation, match link UX priority, enriched internal match details, and historical UI regression validation.
This commit is contained in:
devRaGonSa
2026-05-19 11:13:17 +02:00
committed by GitHub
parent af4de29a0a
commit 95bb09118f
19 changed files with 1851 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

101
backend/app/historical_storage.py
View File

@@ -6,7 +6,6 @@ import sqlite3
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Mapping
from urllib.parse import urlparse
from .config import (
get_historical_refresh_overlap_hours,
@@ -17,28 +16,22 @@ from .config import (
from .historical_models import HistoricalServerDefinition
from .monthly_mvp import build_monthly_mvp_rankings
from .monthly_mvp_v2 import build_monthly_mvp_v2_rankings
from .scoreboard_origins import (
list_trusted_public_scoreboard_origins,
resolve_trusted_scoreboard_match_url,
)
from .sqlite_utils import connect_sqlite_writer
DEFAULT_HISTORICAL_SERVERS = (
DEFAULT_HISTORICAL_SERVERS = tuple(
HistoricalServerDefinition(
slug="comunidad-hispana-01",
display_name="Comunidad Hispana #01",
scoreboard_base_url="https://scoreboard.comunidadhll.es",
server_number=1,
),
HistoricalServerDefinition(
slug="comunidad-hispana-02",
display_name="Comunidad Hispana #02",
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,
),
slug=origin.slug,
display_name=origin.display_name,
scoreboard_base_url=origin.base_url,
server_number=origin.server_number,
source_kind=origin.source_kind,
)
for origin in list_trusted_public_scoreboard_origins()
)
ALL_SERVERS_SLUG = "all-servers"
ALL_SERVERS_DISPLAY_NAME = "Todos"
@@ -768,6 +761,7 @@ def list_recent_historical_matches(
historical_matches.allied_score,
historical_matches.axis_score,
historical_matches.raw_payload_ref,
historical_servers.slug,
historical_servers.scoreboard_base_url,
COUNT(historical_player_match_stats.id) AS player_count
FROM historical_matches
@@ -809,7 +803,7 @@ def list_recent_historical_matches(
"player_count": int(row["player_count"] or 0),
"match_url": _resolve_safe_match_url(
row["raw_payload_ref"],
row["scoreboard_base_url"],
row["server_slug"],
),
}
)
@@ -832,6 +826,7 @@ def get_historical_match_detail(
row = connection.execute(
"""
SELECT
historical_matches.id AS match_pk,
historical_servers.slug AS server_slug,
historical_servers.display_name AS server_name,
historical_matches.external_match_id,
@@ -842,6 +837,7 @@ def get_historical_match_detail(
historical_matches.allied_score,
historical_matches.axis_score,
historical_matches.raw_payload_ref,
historical_servers.slug,
historical_servers.scoreboard_base_url,
COUNT(historical_player_match_stats.id) AS player_count,
SUM(COALESCE(historical_player_match_stats.time_seconds, 0)) AS total_time_seconds
@@ -857,6 +853,33 @@ def get_historical_match_detail(
""",
(normalized_server_slug, normalized_match_id),
).fetchone()
player_rows = []
if row is not None:
player_rows = connection.execute(
"""
SELECT
historical_players.display_name,
historical_players.stable_player_key,
historical_player_match_stats.team_side,
historical_player_match_stats.level,
historical_player_match_stats.kills,
historical_player_match_stats.deaths,
historical_player_match_stats.teamkills,
historical_player_match_stats.combat,
historical_player_match_stats.offense,
historical_player_match_stats.defense,
historical_player_match_stats.support,
historical_player_match_stats.time_seconds
FROM historical_player_match_stats
INNER JOIN historical_players
ON historical_players.id = historical_player_match_stats.historical_player_id
WHERE historical_player_match_stats.historical_match_id = ?
ORDER BY
COALESCE(historical_player_match_stats.kills, 0) DESC,
historical_players.display_name ASC
""",
(row["match_pk"],),
).fetchall()
if row is None:
return None
started_at = row["started_at"]
@@ -885,10 +908,27 @@ def get_historical_match_detail(
},
"player_count": int(row["player_count"] or 0),
"total_time_seconds": _coerce_int(row["total_time_seconds"]),
"players": [
{
"name": player_row["display_name"],
"stable_player_key": player_row["stable_player_key"],
"team_side": player_row["team_side"],
"level": _coerce_int(player_row["level"]),
"kills": _coerce_int(player_row["kills"]),
"deaths": _coerce_int(player_row["deaths"]),
"teamkills": _coerce_int(player_row["teamkills"]),
"combat": _coerce_int(player_row["combat"]),
"offense": _coerce_int(player_row["offense"]),
"defense": _coerce_int(player_row["defense"]),
"support": _coerce_int(player_row["support"]),
"time_seconds": _coerce_int(player_row["time_seconds"]),
}
for player_row in player_rows
],
"capture_basis": "public-scoreboard-match",
"match_url": _resolve_safe_match_url(
row["raw_payload_ref"],
row["scoreboard_base_url"],
row["server_slug"],
),
}
@@ -3198,23 +3238,8 @@ def _is_all_servers_selector(value: str | None) -> bool:
return isinstance(value, str) and value.strip() == ALL_SERVERS_SLUG
def _resolve_safe_match_url(raw_payload_ref: object, scoreboard_base_url: object) -> str | None:
candidate = _stringify(raw_payload_ref)
base_url = _stringify(scoreboard_base_url)
if not candidate or not base_url:
return None
candidate_parts = urlparse(candidate)
base_parts = urlparse(base_url)
if candidate_parts.scheme not in {"http", "https"}:
return None
if candidate_parts.scheme != base_parts.scheme or candidate_parts.netloc != base_parts.netloc:
return None
if not candidate_parts.path.startswith("/games/"):
return None
if candidate_parts.username or candidate_parts.password:
return None
return candidate
def _resolve_safe_match_url(raw_payload_ref: object, server_slug: object) -> str | None:
return resolve_trusted_scoreboard_match_url(raw_payload_ref, server_slug)
def _calculate_match_duration_seconds(started_at: object, ended_at: object) -> int | None:

8
backend/app/payloads.py
View File

@@ -39,7 +39,6 @@ from .historical_snapshots import (
)
from .historical_storage import (
ALL_SERVERS_SLUG,
DEFAULT_HISTORICAL_SERVERS,
get_historical_match_detail,
get_historical_player_profile,
list_historical_server_summaries,
@@ -50,6 +49,7 @@ from .historical_storage import (
)
from .rcon_historical_read_model import get_rcon_historical_match_detail
from .normalizers import normalize_map_name
from .scoreboard_origins import get_trusted_public_scoreboard_origin
from .storage import list_latest_snapshots, list_server_history, list_snapshot_history
@@ -1910,7 +1910,5 @@ def _resolve_community_history_url(external_server_id: object) -> str | None:
normalized_server_id = str(external_server_id or "").strip()
if not normalized_server_id:
return None
for server in DEFAULT_HISTORICAL_SERVERS:
if server.slug == normalized_server_id:
return f"{server.scoreboard_base_url}/games"
return None
origin = get_trusted_public_scoreboard_origin(normalized_server_id)
return f"{origin.base_url}/games" if origin else None

14
backend/app/rcon_historical_read_model.py
View File

@@ -6,6 +6,7 @@ from datetime import datetime, timezone
from .historical_storage import ALL_SERVERS_SLUG
from .normalizers import normalize_map_name
from .rcon_scoreboard_correlation import resolve_rcon_scoreboard_match_url
from .rcon_historical_storage import (
find_rcon_historical_competitive_window,
get_rcon_historical_competitive_window_by_session,
@@ -136,6 +137,8 @@ def get_rcon_historical_match_detail(
)
if item is None:
return None
player_count = int(round(float(item.get("average_players") or 0)))
server_slug = item["external_server_id"] or item["target_key"]
return {
"server": {
"slug": item["target_key"],
@@ -160,9 +163,18 @@ def get_rcon_historical_match_detail(
"player_count": int(round(float(item.get("average_players") or 0))),
"peak_players": item.get("peak_players"),
"sample_count": item.get("sample_count"),
"players": [],
"capture_basis": "rcon-competitive-window",
"capabilities": item.get("capabilities"),
"match_url": None,
"match_url": resolve_rcon_scoreboard_match_url(
server_slug=server_slug,
map_name=item.get("map_pretty_name") or item.get("map_name"),
started_at=item["first_seen_at"],
ended_at=item["last_seen_at"],
duration_seconds=item.get("duration_seconds"),
player_count=player_count,
peak_players=item.get("peak_players"),
),
}

218
backend/app/rcon_scoreboard_correlation.py Normal file
View File

@@ -0,0 +1,218 @@
"""Correlate RCON competitive windows with trusted persisted scoreboard matches."""
from __future__ import annotations
import sqlite3
from datetime import datetime, timezone
from pathlib import Path
from .config import get_storage_path
from .normalizers import normalize_map_name
from .scoreboard_origins import resolve_trusted_scoreboard_match_url
from .sqlite_utils import connect_sqlite_readonly
MIN_CONFIDENCE_SCORE = 5
MAX_CANDIDATES = 200
def resolve_rcon_scoreboard_match_url(
*,
server_slug: object,
map_name: object,
started_at: object,
ended_at: object,
duration_seconds: object = None,
player_count: object = None,
peak_players: object = None,
db_path: Path | None = None,
) -> str | None:
"""Return a trusted scoreboard URL for an RCON window only on strong evidence."""
normalized_server_slug = str(server_slug or "").strip()
normalized_map = normalize_map_name(map_name)
rcon_start = _parse_timestamp(started_at)
rcon_end = _parse_timestamp(ended_at)
if not normalized_server_slug or not normalized_map or not rcon_start or not rcon_end:
return None
if rcon_end < rcon_start:
rcon_start, rcon_end = rcon_end, rcon_start
candidates = _list_persisted_scoreboard_candidates(
server_slug=normalized_server_slug,
db_path=db_path or get_storage_path(),
)
scored_candidates = [
scored
for candidate in candidates
if (scored := _score_candidate(
candidate,
normalized_map=normalized_map,
rcon_start=rcon_start,
rcon_end=rcon_end,
duration_seconds=_coerce_int(duration_seconds),
player_count=_coerce_int(player_count),
peak_players=_coerce_int(peak_players),
))
is not None
]
if not scored_candidates:
return None
scored_candidates.sort(key=lambda item: item["score"], reverse=True)
best = scored_candidates[0]
if int(best["score"]) < MIN_CONFIDENCE_SCORE:
return None
if len(scored_candidates) > 1 and int(scored_candidates[1]["score"]) >= int(best["score"]):
return None
return str(best["match_url"])
def _list_persisted_scoreboard_candidates(
*,
server_slug: str,
db_path: Path,
) -> list[dict[str, object]]:
try:
with connect_sqlite_readonly(db_path) as connection:
rows = connection.execute(
"""
SELECT
historical_matches.external_match_id,
historical_matches.started_at,
historical_matches.ended_at,
historical_matches.map_name,
historical_matches.map_pretty_name,
historical_matches.raw_payload_ref,
historical_servers.slug AS server_slug,
COUNT(historical_player_match_stats.id) AS player_count
FROM historical_matches
INNER JOIN historical_servers
ON historical_servers.id = historical_matches.historical_server_id
LEFT JOIN historical_player_match_stats
ON historical_player_match_stats.historical_match_id = historical_matches.id
WHERE historical_servers.slug = ?
AND historical_matches.raw_payload_ref IS NOT NULL
GROUP BY historical_matches.id
ORDER BY COALESCE(historical_matches.ended_at, historical_matches.started_at) DESC
LIMIT ?
""",
(server_slug, MAX_CANDIDATES),
).fetchall()
except sqlite3.Error:
return []
items: list[dict[str, object]] = []
for row in rows:
match_url = resolve_trusted_scoreboard_match_url(
row["raw_payload_ref"],
row["server_slug"],
)
if not match_url:
continue
items.append(
{
"external_match_id": row["external_match_id"],
"started_at": row["started_at"],
"ended_at": row["ended_at"],
"map_name": row["map_name"],
"map_pretty_name": row["map_pretty_name"],
"player_count": row["player_count"],
"match_url": match_url,
}
)
return items
def _score_candidate(
candidate: dict[str, object],
*,
normalized_map: str,
rcon_start: datetime,
rcon_end: datetime,
duration_seconds: int | None,
player_count: int | None,
peak_players: int | None,
) -> dict[str, object] | None:
candidate_map = normalize_map_name(
candidate.get("map_pretty_name") or candidate.get("map_name")
)
if candidate_map != normalized_map:
return None
candidate_start = _parse_timestamp(candidate.get("started_at"))
candidate_end = _parse_timestamp(candidate.get("ended_at"))
if not candidate_start or not candidate_end:
return None
if candidate_end < candidate_start:
candidate_start, candidate_end = candidate_end, candidate_start
score = 0
overlap_seconds = _overlap_seconds(rcon_start, rcon_end, candidate_start, candidate_end)
rcon_midpoint = rcon_start + (rcon_end - rcon_start) / 2
if overlap_seconds > 0:
score += 3
if candidate_start <= rcon_midpoint <= candidate_end:
score += 2
closest_edge_distance = min(
abs((rcon_start - candidate_start).total_seconds()),
abs((rcon_start - candidate_end).total_seconds()),
abs((rcon_end - candidate_start).total_seconds()),
abs((rcon_end - candidate_end).total_seconds()),
)
if closest_edge_distance <= 1800:
score += 2
elif closest_edge_distance <= 3600:
score += 1
candidate_duration = int((candidate_end - candidate_start).total_seconds())
if duration_seconds and candidate_duration > 0:
if abs(candidate_duration - duration_seconds) <= 1800:
score += 1
elif overlap_seconds > 0 and duration_seconds <= candidate_duration:
score += 1
candidate_players = _coerce_int(candidate.get("player_count"))
reference_players = peak_players or player_count
if candidate_players and reference_players:
if abs(candidate_players - reference_players) <= 20:
score += 1
elif candidate_players >= int(reference_players * 0.75):
score += 1
if score <= 0:
return None
return {
"score": score,
"match_url": candidate["match_url"],
}
def _overlap_seconds(
first_start: datetime,
first_end: datetime,
second_start: datetime,
second_end: datetime,
) -> int:
return max(0, int((min(first_end, second_end) - max(first_start, second_start)).total_seconds()))
def _parse_timestamp(value: object) -> datetime | None:
if not isinstance(value, str) or not value.strip():
return None
try:
parsed = datetime.fromisoformat(value.strip().replace("Z", "+00:00"))
except ValueError:
return None
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=timezone.utc)
return parsed.astimezone(timezone.utc)
def _coerce_int(value: object) -> int | None:
if value is None:
return None
try:
return int(round(float(value)))
except (TypeError, ValueError):
return None

76
backend/app/scoreboard_origins.py Normal file
View File

@@ -0,0 +1,76 @@
"""Trusted public scoreboard origins for active community servers."""
from __future__ import annotations
from dataclasses import dataclass
from urllib.parse import urlparse
@dataclass(frozen=True, slots=True)
class TrustedScoreboardOrigin:
"""Public scoreboard origin trusted for one active community server."""
slug: str
display_name: str
base_url: str
server_number: int
source_kind: str = "crcon-scoreboard-json"
TRUSTED_PUBLIC_SCOREBOARD_ORIGINS = (
TrustedScoreboardOrigin(
slug="comunidad-hispana-01",
display_name="Comunidad Hispana #01",
base_url="https://scoreboard.comunidadhll.es",
server_number=1,
),
TrustedScoreboardOrigin(
slug="comunidad-hispana-02",
display_name="Comunidad Hispana #02",
base_url="https://scoreboard.comunidadhll.es:5443",
server_number=2,
),
)
def list_trusted_public_scoreboard_origins() -> tuple[TrustedScoreboardOrigin, ...]:
"""Return trusted public scoreboard origins for active default servers."""
return TRUSTED_PUBLIC_SCOREBOARD_ORIGINS
def get_trusted_public_scoreboard_origin(
server_slug: object,
) -> TrustedScoreboardOrigin | None:
"""Return the trusted public scoreboard origin for one active server."""
normalized_slug = str(server_slug or "").strip()
if not normalized_slug:
return None
for origin in TRUSTED_PUBLIC_SCOREBOARD_ORIGINS:
if origin.slug == normalized_slug:
return origin
return None
def resolve_trusted_scoreboard_match_url(
raw_payload_ref: object,
server_slug: object,
) -> str | None:
"""Return a match URL only when it belongs to the trusted server origin."""
origin = get_trusted_public_scoreboard_origin(server_slug)
candidate = str(raw_payload_ref or "").strip()
if origin is None or not candidate:
return None
candidate_parts = urlparse(candidate)
origin_parts = urlparse(origin.base_url)
if candidate_parts.scheme not in {"http", "https"}:
return None
if candidate_parts.scheme != origin_parts.scheme:
return None
if candidate_parts.netloc != origin_parts.netloc:
return None
if candidate_parts.username or candidate_parts.password:
return None
if not candidate_parts.path.startswith("/games/"):
return None
return candidate