Delete tracker_client.py

tracker_client.py

@@ -1,238 +0,0 @@
-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 '<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},
-        )
-
-    @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