95 lines
3.5 KiB
Python
95 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Iterable
|
|
from dataclasses import dataclass
|
|
|
|
import requests
|
|
from requests.adapters import HTTPAdapter
|
|
from urllib3.util.retry import Retry
|
|
|
|
COLLECTION_DETAILS_URL = (
|
|
"https://api.steampowered.com/ISteamRemoteStorage/GetCollectionDetails/v1/"
|
|
)
|
|
PUBLISHED_FILE_DETAILS_URL = (
|
|
"https://api.steampowered.com/ISteamRemoteStorage/GetPublishedFileDetails/v1/"
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class WorkshopDetails:
|
|
workshop_id: str
|
|
title: str
|
|
time_updated: int
|
|
file_size: int
|
|
consumer_app_id: int
|
|
result: int
|
|
|
|
|
|
def _chunks(values: list[str], size: int) -> Iterable[list[str]]:
|
|
for index in range(0, len(values), size):
|
|
yield values[index : index + size]
|
|
|
|
|
|
class SteamWorkshopClient:
|
|
def __init__(self, timeout: int = 30):
|
|
self.timeout = timeout
|
|
self.session = requests.Session()
|
|
retries = Retry(
|
|
total=4,
|
|
backoff_factor=0.75,
|
|
status_forcelist=(429, 500, 502, 503, 504),
|
|
allowed_methods=frozenset({"POST"}),
|
|
)
|
|
self.session.mount("https://", HTTPAdapter(max_retries=retries))
|
|
self.session.headers.update({"User-Agent": "ProgressionMods/1.0"})
|
|
|
|
def get_collections(self, collection_ids: list[str]) -> dict[str, list[str]]:
|
|
payload: dict[str, str | int] = {"collectioncount": len(collection_ids)}
|
|
for index, collection_id in enumerate(collection_ids):
|
|
payload[f"publishedfileids[{index}]"] = collection_id
|
|
|
|
response = self.session.post(
|
|
COLLECTION_DETAILS_URL, data=payload, timeout=self.timeout
|
|
)
|
|
response.raise_for_status()
|
|
details = response.json().get("response", {}).get("collectiondetails", [])
|
|
|
|
result: dict[str, list[str]] = {}
|
|
for detail in details:
|
|
collection_id = str(detail.get("publishedfileid", ""))
|
|
if int(detail.get("result", 0)) != 1:
|
|
raise RuntimeError(f"Steam rejected collection {collection_id}")
|
|
children = sorted(
|
|
detail.get("children", []),
|
|
key=lambda child: int(child.get("sortorder", 0)),
|
|
)
|
|
result[collection_id] = [
|
|
str(child["publishedfileid"]) for child in children
|
|
]
|
|
return result
|
|
|
|
def get_item_details(self, workshop_ids: list[str]) -> dict[str, WorkshopDetails]:
|
|
result: dict[str, WorkshopDetails] = {}
|
|
for batch in _chunks(workshop_ids, 100):
|
|
payload: dict[str, str | int] = {"itemcount": len(batch)}
|
|
for index, workshop_id in enumerate(batch):
|
|
payload[f"publishedfileids[{index}]"] = workshop_id
|
|
response = self.session.post(
|
|
PUBLISHED_FILE_DETAILS_URL, data=payload, timeout=self.timeout
|
|
)
|
|
response.raise_for_status()
|
|
details = (
|
|
response.json().get("response", {}).get("publishedfiledetails", [])
|
|
)
|
|
for detail in details:
|
|
workshop_id = str(detail.get("publishedfileid", ""))
|
|
result[workshop_id] = WorkshopDetails(
|
|
workshop_id=workshop_id,
|
|
title=str(detail.get("title", "")),
|
|
time_updated=int(detail.get("time_updated", 0) or 0),
|
|
file_size=int(detail.get("file_size", 0) or 0),
|
|
consumer_app_id=int(detail.get("consumer_app_id", 0) or 0),
|
|
result=int(detail.get("result", 0) or 0),
|
|
)
|
|
return result
|