Spaces:
Sleeping
Sleeping
| 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) | |
| 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 | |
| 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 '<empty>'}" | |
| ) | |
| 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}, | |
| ) | |
| 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 | |