"""Steam Web API client for player profile lookups. Mirrors the shape of l4d2web/services/steam_workshop.py:17-43: - single thread-local requests.Session - 30s timeout - HTTPS only Difference: GetPlayerSummaries requires an API key in the querystring, unlike the anonymous workshop endpoints. """ from __future__ import annotations import threading from dataclasses import dataclass from typing import Iterable import requests GET_PLAYER_SUMMARIES_URL = ( "https://api.steampowered.com/ISteamUser/GetPlayerSummaries/v0002/" ) REQUEST_TIMEOUT_SECONDS = 30.0 MAX_IDS_PER_CALL = 100 _session_local = threading.local() def _session() -> requests.Session: sess = getattr(_session_local, "session", None) if sess is None: sess = requests.Session() _session_local.session = sess return sess def _session_get(url: str, params: dict, timeout: float = REQUEST_TIMEOUT_SECONDS): """Indirection seam so tests can monkeypatch a fake here.""" return _session().get(url, params=params, timeout=timeout) @dataclass(slots=True, frozen=True) class SteamProfile: steam_id_64: str persona_name: str avatar_url: str def fetch_profiles_batch( steam_ids: Iterable[str], *, api_key: str ) -> list[SteamProfile]: """Resolve a batch of SteamID64 strings to persona name + avatar URL. Steam's API caps each call at 100 IDs; this helper chunks transparently. IDs that Steam can't resolve (private, deleted) are simply absent from the response and from the returned list. """ ids = list(steam_ids) out: list[SteamProfile] = [] for i in range(0, len(ids), MAX_IDS_PER_CALL): chunk = ids[i : i + MAX_IDS_PER_CALL] params = {"key": api_key, "steamids": ",".join(chunk)} resp = _session_get(GET_PLAYER_SUMMARIES_URL, params=params) resp.raise_for_status() payload = resp.json() or {} players = (payload.get("response") or {}).get("players") or [] for p in players: out.append( SteamProfile( steam_id_64=str(p["steamid"]), persona_name=str(p.get("personaname", "")), avatar_url=str(p.get("avatarmedium", "")), ) ) return out