From fa5a2a936edd3f14ae78a587277da78c52d009fb Mon Sep 17 00:00:00 2001 From: devRaGonSa <97627393+devRaGonSa@users.noreply.github.com> Date: Wed, 20 May 2026 13:20:05 +0200 Subject: [PATCH] fix: avoid elo imports in scoreboard candidate backfill --- backend/app/scoreboard_candidate_backfill.py | 227 +++++++++---------- 1 file changed, 103 insertions(+), 124 deletions(-) diff --git a/backend/app/scoreboard_candidate_backfill.py b/backend/app/scoreboard_candidate_backfill.py index e604698..03b72ff 100644 --- a/backend/app/scoreboard_candidate_backfill.py +++ b/backend/app/scoreboard_candidate_backfill.py @@ -1,9 +1,4 @@ -"""Backfill public scoreboard candidates for RCON match link correlation. - -This command intentionally reuses the existing historical public-scoreboard -archive ingestion path. The RCON materialized detail endpoint can only expose a -safe public match URL when a matching scoreboard row exists in historical_matches. -""" +"""Backfill public scoreboard candidates for RCON match link correlation.""" from __future__ import annotations @@ -12,7 +7,8 @@ import json from datetime import datetime, timezone from typing import Iterable -from .historical_ingestion import run_incremental_refresh +from .historical_storage import initialize_historical_storage, list_historical_servers, upsert_historical_match +from .providers.public_scoreboard_provider import PublicScoreboardHistoricalDataSource from .scoreboard_origins import list_trusted_public_scoreboard_origins DEFAULT_MAX_PAGES = 20 @@ -23,134 +19,131 @@ DEFAULT_DETAIL_WORKERS = 4 def main(argv: Iterable[str] | None = None) -> int: parser = build_arg_parser() args = parser.parse_args(list(argv) if argv is not None else None) - - from_timestamp = _parse_timestamp(args.from_timestamp, option_name="--from") - to_timestamp = _parse_timestamp(args.to_timestamp, option_name="--to") - if to_timestamp <= from_timestamp: + start_at = _parse_timestamp(args.start_at, option_name="--from") + end_at = _parse_timestamp(args.end_at, option_name="--to") + if end_at <= start_at: parser.error("--to must be later than --from") - - _validate_server(args.server_slug, parser) - overlap_hours = _calculate_overlap_hours(from_timestamp) - - progress_events: list[dict[str, object]] = [] - - def _capture_progress(payload: dict[str, object]) -> None: - progress_events.append(payload) - print(json.dumps(payload, ensure_ascii=False)) - - result = run_incremental_refresh( - server_slug=args.server_slug, - max_pages=args.max_pages, - page_size=args.page_size, - detail_workers=args.detail_workers, - overlap_hours=overlap_hours, - rebuild_snapshots=False, - progress_callback=_capture_progress, - ) - - totals = dict(result.get("totals") or {}) - report = { - "status": result.get("status"), - "server": args.server_slug, - "requested_window": { - "from": _format_timestamp(from_timestamp), - "to": _format_timestamp(to_timestamp), - }, - "ingestion_policy": { - "overlap_hours": overlap_hours, - "max_pages": args.max_pages, - "page_size": args.page_size, - "detail_workers": args.detail_workers, - "implementation": "historical-ingestion-public-scoreboard-candidates", - }, - "candidates_seen": _coerce_int(totals.get("matches_seen")), - "candidates_inserted": _coerce_int(totals.get("matches_inserted")), - "candidates_updated": _coerce_int(totals.get("matches_updated")), - "player_rows_inserted": _coerce_int(totals.get("player_rows_inserted")), - "player_rows_updated": _coerce_int(totals.get("player_rows_updated")), - "skipped_unsafe_urls": 0, - "errors": _extract_errors(progress_events, result), - "raw_result": result, - } + server = _resolve_server(args.server_slug, parser) + report = run_backfill(server=server, start_at=start_at, end_at=end_at, max_pages=args.max_pages, page_size=args.page_size, detail_workers=args.detail_workers) print(json.dumps(report, ensure_ascii=False, indent=2)) - return 0 if str(result.get("status")) == "ok" else 1 + return 0 if not report["errors"] else 1 + + +def run_backfill(*, server: dict[str, object], start_at: datetime, end_at: datetime, max_pages: int, page_size: int, detail_workers: int) -> dict[str, object]: + initialize_historical_storage() + provider = PublicScoreboardHistoricalDataSource() + server_slug = str(server["slug"]) + base_url = str(server["scoreboard_base_url"]) + counters = {"pages_processed": 0, "candidates_seen": 0, "candidates_inserted": 0, "candidates_updated": 0, "player_rows_inserted": 0, "player_rows_updated": 0} + errors: list[dict[str, object]] = [] + stopped_after_window = False + for page in range(1, max_pages + 1): + try: + page_payload = provider.fetch_match_page(base_url=base_url, page=page, limit=page_size) + except Exception as exc: + errors.append({"stage": "fetch_match_page", "page": page, "message": str(exc)}) + break + matches = _coerce_match_list(page_payload.get("maps")) + if not matches: + break + counters["pages_processed"] += 1 + ids: list[str] = [] + for match in matches: + counters["candidates_seen"] += 1 + ref_time = _parse_optional_timestamp(_pick_match_timestamp(match)) + if ref_time and ref_time < start_at: + stopped_after_window = True + continue + if ref_time and ref_time >= end_at: + continue + match_id = _stringify(match.get("id")) + if match_id: + ids.append(match_id) + if ids: + try: + details = provider.fetch_match_details(base_url=base_url, match_ids=ids, max_workers=detail_workers) + except Exception as exc: + errors.append({"stage": "fetch_match_details", "page": page, "message": str(exc)}) + details = [] + for detail in details: + try: + delta = upsert_historical_match(server_slug=server_slug, match_payload=detail) + except Exception as exc: + errors.append({"stage": "upsert_historical_match", "match_id": _stringify(detail.get("id")), "message": str(exc)}) + continue + counters["candidates_inserted"] += _coerce_int(delta.get("matches_inserted")) + counters["candidates_updated"] += _coerce_int(delta.get("matches_updated")) + counters["player_rows_inserted"] += _coerce_int(delta.get("player_rows_inserted")) + counters["player_rows_updated"] += _coerce_int(delta.get("player_rows_updated")) + if stopped_after_window: + break + return {"status": "ok" if not errors else "partial", "server": server_slug, "scoreboard_base_url": base_url, "requested_window": {"from": _format_timestamp(start_at), "to": _format_timestamp(end_at)}, "stopped_after_window": stopped_after_window, "skipped_unsafe_urls": 0, "errors": errors, **counters} def build_arg_parser() -> argparse.ArgumentParser: - parser = argparse.ArgumentParser( - description="Backfill public scoreboard match candidates for RCON link correlation.", - ) - parser.add_argument( - "--server", - dest="server_slug", - required=True, - help="trusted historical server slug, e.g. comunidad-hispana-02", - ) - parser.add_argument( - "--from", - dest="from_timestamp", - required=True, - help="inclusive UTC-ish ISO timestamp used as the lower backfill bound", - ) - parser.add_argument( - "--to", - dest="to_timestamp", - required=True, - help="exclusive UTC-ish ISO timestamp kept in the report for traceability", - ) - parser.add_argument( - "--max-pages", - type=int, - default=DEFAULT_MAX_PAGES, - help=f"maximum public-scoreboard pages to scan, default {DEFAULT_MAX_PAGES}", - ) - parser.add_argument( - "--page-size", - type=int, - default=DEFAULT_PAGE_SIZE, - help=f"public-scoreboard page size, default {DEFAULT_PAGE_SIZE}", - ) - parser.add_argument( - "--detail-workers", - type=int, - default=DEFAULT_DETAIL_WORKERS, - help=f"parallel detail workers, default {DEFAULT_DETAIL_WORKERS}", - ) + parser = argparse.ArgumentParser(description="Backfill public scoreboard match candidates for RCON link correlation.") + parser.add_argument("--server", dest="server_slug", required=True) + parser.add_argument("--from", dest="start_at", required=True) + parser.add_argument("--to", dest="end_at", required=True) + parser.add_argument("--max-pages", type=int, default=DEFAULT_MAX_PAGES) + parser.add_argument("--page-size", type=int, default=DEFAULT_PAGE_SIZE) + parser.add_argument("--detail-workers", type=int, default=DEFAULT_DETAIL_WORKERS) return parser -def _validate_server(server_slug: str, parser: argparse.ArgumentParser) -> None: - trusted_slugs = {origin.slug for origin in list_trusted_public_scoreboard_origins()} - if server_slug not in trusted_slugs: - parser.error( - f"unknown or untrusted server '{server_slug}'. " - f"Allowed values: {', '.join(sorted(trusted_slugs))}" - ) +def _resolve_server(server_slug: str, parser: argparse.ArgumentParser) -> dict[str, object]: + trusted = {origin.slug for origin in list_trusted_public_scoreboard_origins()} + if server_slug not in trusted: + parser.error(f"unknown or untrusted server '{server_slug}'") + for server in list_historical_servers(): + if server.get("slug") == server_slug: + return server + parser.error(f"trusted server '{server_slug}' is not present in historical storage") + raise AssertionError("unreachable") def _parse_timestamp(value: str, *, option_name: str) -> datetime: try: parsed = datetime.fromisoformat(value.strip().replace("Z", "+00:00")) except ValueError as exc: - raise argparse.ArgumentTypeError( - f"{option_name} must be an ISO timestamp, got {value!r}" - ) from exc + raise argparse.ArgumentTypeError(f"{option_name} must be an ISO timestamp") from exc if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=timezone.utc) return parsed.astimezone(timezone.utc) -def _calculate_overlap_hours(from_timestamp: datetime) -> int: - now = datetime.now(timezone.utc) - delta_seconds = max(0, int((now - from_timestamp).total_seconds())) - # Add one hour so the requested lower bound is safely included after integer rounding. - return max(1, (delta_seconds // 3600) + 1) +def _parse_optional_timestamp(value: object) -> datetime | None: + if not isinstance(value, str) or not value.strip(): + return None + try: + return _parse_timestamp(value, option_name="timestamp") + except argparse.ArgumentTypeError: + return None def _format_timestamp(value: datetime) -> str: return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") +def _coerce_match_list(payload: object) -> list[dict[str, object]]: + return [item for item in payload if isinstance(item, dict)] if isinstance(payload, list) else [] + + +def _pick_match_timestamp(match: dict[str, object]) -> object: + for key in ("end", "start", "creation_time"): + value = match.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: try: return int(value or 0) @@ -158,19 +151,5 @@ def _coerce_int(value: object) -> int: return 0 -def _extract_errors( - progress_events: list[dict[str, object]], - result: dict[str, object], -) -> list[dict[str, object]]: - errors: list[dict[str, object]] = [] - if str(result.get("status")) != "ok": - errors.append({"scope": "result", "message": str(result)}) - for event in progress_events: - text = str(event.get("event") or "") - if "error" in text or "failed" in text: - errors.append(event) - return errors - - if __name__ == "__main__": raise SystemExit(main())