from __future__ import annotations import os from dataclasses import dataclass from datetime import datetime, timezone from typing import Any, Dict, List, Optional import requests BASE_WEB = "https://api.tracker.gg" PROFILE_WEB = "https://rocketleague.tracker.network/rocket-league/profile" class TrackerError(RuntimeError): pass def _dt(value: Optional[str]) -> Optional[datetime]: if not value: return None try: d = datetime.fromisoformat(value.replace("Z", "+00:00")) if d.tzinfo is None: d = d.replace(tzinfo=timezone.utc) return d.astimezone(timezone.utc) except ValueError: return None def _stat_value(stats: Dict[str, Any], name: str, default=None): return (stats.get(name) or {}).get("value", default) @dataclass class PlayerProfile: platform: str identifier: str handle: str player_id: Optional[int] last_updated: Optional[datetime] season: Optional[int] avatar_url: Optional[str] lifetime: Dict[str, Any] playlists: List[Dict[str, Any]] raw: Dict[str, Any] class TrackerClient: """Best-effort client for Tracker web endpoints observed in supplied samples. This intentionally does not solve/bypass Cloudflare or Warden challenges. A 401/403 is returned to the bot as an error instead. """ def __init__(self, timeout: int = 20): self.session = requests.Session() self.timeout = timeout self.session.headers.update( { "Accept": "application/json, text/plain, */*", "Accept-Language": "en-US,en;q=0.9", "Origin": "https://rocketleague.tracker.network", "Referer": "https://rocketleague.tracker.network/", "User-Agent": os.getenv( "TRACKER_USER_AGENT", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/153.0.0.0 Safari/537.36", ), } ) cookie = os.getenv("TRACKER_COOKIE", "").strip() if cookie: self.session.headers["Cookie"] = cookie def _prime_profile_session(self, platform: str, identifier: str) -> None: """Visit the normal public Tracker profile page first so browser cookies/state exist. This does not solve CAPTCHA/WAF challenges. It only mirrors the normal browser flow of opening a profile page before its JSON requests are made. """ page_url = f"{PROFILE_WEB}/{platform}/{identifier}" headers = { "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8", "Sec-Fetch-Dest": "document", "Sec-Fetch-Mode": "navigate", "Sec-Fetch-Site": "none", "Upgrade-Insecure-Requests": "1", } try: self.session.get(page_url, headers=headers, timeout=self.timeout, allow_redirects=True) except requests.RequestException: # The API request below is still worth attempting; it will produce the useful error. pass @staticmethod def _extract_profile_from_path(path: str) -> tuple[Optional[str], Optional[str]]: marker = "/api/v2/rocket-league/standard/profile/" if not path.startswith(marker): return None, None rest = path[len(marker):].strip("/") parts = rest.split("/") if len(parts) < 2: return None, None return parts[0], parts[1] def _get(self, path: str, params: Optional[dict] = None) -> Dict[str, Any]: url = f"{BASE_WEB}{path}" platform, identifier = self._extract_profile_from_path(path) if platform and identifier: self._prime_profile_session(platform, identifier) api_headers = { "Accept": "application/json, text/plain, */*", "Referer": ( f"{PROFILE_WEB}/{platform}/{identifier}" if platform and identifier else "https://rocketleague.tracker.network/" ), "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "same-site", } try: r = self.session.get(url, params=params, headers=api_headers, timeout=self.timeout) except requests.RequestException as exc: raise TrackerError(f"Network error contacting Tracker: {exc}") from exc if r.status_code in (401, 403): preview = " ".join((r.text or "").split())[:240] server = r.headers.get("server", "unknown") raise TrackerError( f"HTTP {r.status_code} from Tracker (server={server}). " f"Response preview: {preview or ''}" ) try: r.raise_for_status() except requests.HTTPError as exc: raise TrackerError(f"HTTP {r.status_code} for {r.url}") from exc try: return r.json() except ValueError as exc: raise TrackerError("Tracker returned non-JSON content") from exc def profile(self, platform: str, identifier: str) -> PlayerProfile: payload = self._get(f"/api/v2/rocket-league/standard/profile/{platform}/{identifier}") data = payload.get("data") or {} pinfo = data.get("platformInfo") or {} meta = data.get("metadata") or {} segments = data.get("segments") or [] lifetime: Dict[str, Any] = {} playlists: List[Dict[str, Any]] = [] for seg in segments: stats = seg.get("stats") or {} if seg.get("type") == "overview": lifetime = {k: (v or {}).get("value") for k, v in stats.items()} elif seg.get("type") == "playlist": attrs = seg.get("attributes") or {} smeta = seg.get("metadata") or {} tier = stats.get("tier") or {} div = stats.get("division") or {} rating = stats.get("rating") or {} peak = stats.get("peakRating") or {} playlists.append( { "playlist_id": attrs.get("playlistId"), "name": smeta.get("name", f"Playlist {attrs.get('playlistId')}"), "season": attrs.get("season"), "rating": rating.get("value"), "tier": (rating.get("metadata") or {}).get("tierName") or (tier.get("metadata") or {}).get("name"), "division": (div.get("metadata") or {}).get("name"), "matches_played": _stat_value(stats, "matchesPlayed"), "win_streak": _stat_value(stats, "winStreak"), "peak_rating": peak.get("value"), } ) return PlayerProfile( platform=pinfo.get("platformSlug") or platform, identifier=pinfo.get("platformUserIdentifier") or identifier, handle=pinfo.get("platformUserHandle") or identifier, player_id=meta.get("playerId"), last_updated=_dt((meta.get("lastUpdated") or {}).get("value")), season=meta.get("currentSeason"), avatar_url=pinfo.get("avatarUrl"), lifetime=lifetime, playlists=playlists, raw=payload, ) def sessions(self, platform: str, identifier: str) -> Dict[str, Any]: return self._get(f"/api/v2/rocket-league/standard/profile/{platform}/{identifier}/sessions") def mmr_history(self, player_id: int) -> Dict[str, Any]: return self._get(f"/api/v1/rocket-league/player-history/mmr/{player_id}") def overview_history(self, player_id: int) -> Dict[str, Any]: return self._get(f"/api/v1/rocket-league/player-history/overview/{player_id}", params={"type": "graph"}) def aggregated(self, platform: str, identifier: str, local_offset: int = 0) -> Dict[str, Any]: return self._get( f"/api/v1/rocket-league/matches/{platform}/{identifier}/aggregated", params={"localOffset": local_offset}, ) @staticmethod def flatten_session_matches(payload: Dict[str, Any]) -> List[Dict[str, Any]]: out: List[Dict[str, Any]] = [] for session in ((payload.get("data") or {}).get("items") or []): for m in session.get("matches") or []: meta = m.get("metadata") or {} stats = m.get("stats") or {} r = stats.get("rating") or {} rmeta = r.get("metadata") or {} out.append( { "match_id": m.get("id"), "collected_at": meta.get("dateCollected"), "playlist": meta.get("playlist"), "result": meta.get("result"), "is_grouped": bool(meta.get("isGrouped")), "rating": r.get("value"), "rating_delta": rmeta.get("ratingDelta"), "tier": rmeta.get("tier"), "division": rmeta.get("division"), "matches_played": _stat_value(stats, "matchesPlayed"), "wins": _stat_value(stats, "wins"), "goals": _stat_value(stats, "goals"), "assists": _stat_value(stats, "assists"), "saves": _stat_value(stats, "saves"), "shots": _stat_value(stats, "shots"), "mvps": _stat_value(stats, "mvps"), } ) out.sort(key=lambda x: x.get("collected_at") or "", reverse=True) return out