156 lines
6.7 KiB
Python
156 lines
6.7 KiB
Python
"""Backfill public scoreboard candidates for RCON match link correlation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from typing import Iterable
|
|
|
|
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
|
|
DEFAULT_PAGE_SIZE = 100
|
|
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)
|
|
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")
|
|
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 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)
|
|
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 _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") from exc
|
|
if parsed.tzinfo is None:
|
|
parsed = parsed.replace(tzinfo=timezone.utc)
|
|
return parsed.astimezone(timezone.utc)
|
|
|
|
|
|
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)
|
|
except (TypeError, ValueError):
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|