tracker_client.py

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