76 lines
2.2 KiB
Python
76 lines
2.2 KiB
Python
"""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
|