steam-backlog-enforcer/steam_backlog_enforcer/_steam_api_client.py

"""HTTP client for the Steam Web API.

Split to keep both files under the 250-line cap.
"""

from __future__ import annotations

from concurrent.futures import ThreadPoolExecutor, as_completed
import logging
import threading
import time
from typing import TYPE_CHECKING, Any

import requests

from steam_backlog_enforcer._achievement_skip import (
    is_no_stats_answer,
    record_scan,
    skippable_app_ids,
)
from steam_backlog_enforcer._progress import current_progress
from steam_backlog_enforcer._steam_models import (
    MAX_WORKERS,
    STEAM_API_BASE,
    AchievementInfo,
    GameInfo,
)

if TYPE_CHECKING:
    from collections.abc import Callable

logger = logging.getLogger(__name__)


class SteamAPIError(Exception):
    """Raised when the Steam API returns an error."""


class SteamAPIClient:
    """Client for interacting with the Steam Web API."""

    def __init__(self, api_key: str, steam_id: str) -> None:
        """Initialize the Steam API client.

        Args:
            api_key: Steam Web API key.
            steam_id: Steam64 ID of the user.
        """
        self.api_key = api_key
        self.steam_id = steam_id
        self.session = requests.Session()
        adapter = requests.adapters.HTTPAdapter(
            pool_maxsize=MAX_WORKERS,
            pool_connections=MAX_WORKERS,
        )
        self.session.mount("https://", adapter)
        self.session.headers["Accept"] = "application/json"
        self._rate_lock = threading.Lock()
        self._request_times: list[float] = []
        self._max_rps = 18
        # Games Steam answered "no stats" for during the current scan.
        self._no_stats: set[int] = set()

    def _rate_limit(self) -> None:
        """Enforce rate limit across threads."""
        while True:
            with self._rate_lock:
                now = time.time()
                self._request_times = [t for t in self._request_times if now - t < 1.0]
                if len(self._request_times) < self._max_rps:
                    self._request_times.append(now)
                    return
            time.sleep(0.06)

    def _get(self, url: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
        """Rate-limited GET request."""
        self._rate_limit()
        if params is None:
            params = {}
        params["key"] = self.api_key
        try:
            resp = self.session.get(url, params=params, timeout=30)
            resp.raise_for_status()
            result: dict[str, Any] = resp.json()
        except requests.RequestException as e:
            msg = f"Steam API request failed: {e}"
            raise SteamAPIError(msg) from e
        return result

    def get_owned_games(self) -> list[dict[str, Any]]:
        """Fetch all games owned by the user."""
        url = f"{STEAM_API_BASE}/IPlayerService/GetOwnedGames/v1/"
        data = self._get(
            url,
            {
                "steamid": self.steam_id,
                "include_appinfo": "true",
                "include_played_free_games": "true",
                "format": "json",
            },
        )
        games: list[dict[str, Any]] = data.get("response", {}).get("games", [])
        logger.info("Found %d owned games.", len(games))
        return games

    def get_achievement_details(
        self, app_id: int, *, strict: bool = False
    ) -> list[AchievementInfo]:
        """Fetch per-achievement detail for a game.

        ``strict``: only Steam's "no stats" answer is empty; failures raise.
        """
        url = f"{STEAM_API_BASE}/ISteamUserStats/GetPlayerAchievements/v1/"
        try:
            data = self._get(
                url,
                {
                    "steamid": self.steam_id,
                    "appid": str(app_id),
                    "l": "english",
                    "format": "json",
                },
            )
        except SteamAPIError as exc:
            if strict and not is_no_stats_answer(exc):
                raise
            return []

        stats = data.get("playerstats", {})
        if not stats.get("success", False):
            return []

        raw: list[dict[str, Any]] = stats.get("achievements", [])
        return [
            AchievementInfo(
                api_name=a.get("apiname", ""),
                display_name=a.get("name", a.get("apiname", "")),
                achieved=bool(a.get("achieved", 0)),
                unlock_time=a.get("unlocktime", 0),
            )
            for a in raw
        ]

    def _fetch_one_game(self, game_dict: dict[str, Any]) -> GameInfo | None:
        """Fetch achievement data for one game. Thread-safe."""
        app_id = game_dict["appid"]

        achievements = self.get_achievement_details(app_id, strict=True)
        if not achievements:
            self._no_stats.add(app_id)  # set.add is atomic under the GIL
            return None

        name = game_dict.get("name", f"Unknown ({app_id})")
        total = len(achievements)
        unlocked = sum(1 for a in achievements if a.achieved)

        return GameInfo(
            app_id=app_id,
            name=name,
            total_achievements=total,
            unlocked_achievements=unlocked,
            playtime_minutes=game_dict.get("playtime_forever", 0),
            achievements=achievements,
        )

    def build_game_list(
        self,
        progress_callback: Callable[[int, int], None] | None = None,
    ) -> list[GameInfo]:
        """Build full game list with achievement data (parallel).

        Games recently found to have no achievements are not asked about
        again (see :mod:`._achievement_skip`).
        """
        owned = self.get_owned_games()
        now = time.time()
        skip = skippable_app_ids(self.steam_id, owned, now)
        to_fetch = [g for g in owned if g["appid"] not in skip]
        if skip:
            logger.info("Skipping %d games known to have no achievements.", len(skip))
        self._no_stats = set()
        games = self._fetch_games(to_fetch, progress_callback)
        # Only games Steam actually answered for: a timeout or a 5xx must not
        # be remembered as "no achievements" and hidden from scans for a month.
        found = {g.app_id for g in games}
        answered = [g for g in to_fetch if g["appid"] in found | self._no_stats]
        record_scan(self.steam_id, answered, found, now)
        games.sort(key=lambda g: g.name.lower())
        return games

    def _fetch_games(
        self,
        owned: list[dict[str, Any]],
        progress_callback: Callable[[int, int], None] | None,
    ) -> list[GameInfo]:
        """Fetch achievements for *owned* in parallel, reporting each game."""
        games: list[GameInfo] = []
        done_count = 0
        total = len(owned)
        lock = threading.Lock()
        progress = current_progress()
        progress.phase("Scanning Steam achievements", total)

        with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool:
            futures = {pool.submit(self._fetch_one_game, g): g for g in owned}
            try:
                for future in as_completed(futures):
                    try:
                        result = future.result()
                    except (
                        KeyError,
                        TypeError,
                        ValueError,
                        SteamAPIError,
                        requests.RequestException,
                    ):
                        result = None
                    with lock:
                        done_count += 1
                        progress.advance(
                            str(futures[future].get("name", "")), step=done_count
                        )
                        if progress_callback:
                            progress_callback(done_count, total)
                    if result is not None:
                        games.append(result)
            except BaseException:
                # A cancelled job must not wait for the whole queue to drain:
                # drop what has not started, leaving only requests in flight.
                pool.shutdown(wait=False, cancel_futures=True)
                raise
        return games

    def refresh_single_game(
        self, app_id: int, name: str, playtime: int = 0
    ) -> GameInfo | None:
        """Re-fetch achievement data for one game."""
        achievements = self.get_achievement_details(app_id)
        if not achievements:
            return None
        total = len(achievements)
        unlocked = sum(1 for a in achievements if a.achieved)
        return GameInfo(
            app_id=app_id,
            name=name,
            total_achievements=total,
            unlocked_achievements=unlocked,
            playtime_minutes=playtime,
            achievements=achievements,
        )