public site and containers for that.

progression_backend/steam_api.py

@@ -0,0 +1,94 @@
+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