83 lines
2.6 KiB
Python
83 lines
2.6 KiB
Python
"""Async wrapper for Steam's anonymous GetPublishedFileDetails endpoint."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Dict, List, TypedDict
|
|
|
|
import httpx
|
|
|
|
|
|
class CollectionEntry(TypedDict):
|
|
"""One element of fetch_collection_details's response.
|
|
|
|
result == 1 → valid collection; children populated.
|
|
result != 1 → not a collection / deleted / private; children typically [].
|
|
"""
|
|
result: int
|
|
children: List[str]
|
|
|
|
STEAM_URL = (
|
|
"https://api.steampowered.com/ISteamRemoteStorage/GetPublishedFileDetails/v1/"
|
|
)
|
|
|
|
|
|
async def fetch_workshop_details(
|
|
client: httpx.AsyncClient,
|
|
workshop_ids: List[str],
|
|
) -> Dict[str, dict]:
|
|
if not workshop_ids:
|
|
return {}
|
|
data: Dict[str, str] = {"itemcount": str(len(workshop_ids))}
|
|
for i, wid in enumerate(workshop_ids):
|
|
data[f"publishedfileids[{i}]"] = wid
|
|
r = await client.post(STEAM_URL, data=data)
|
|
r.raise_for_status()
|
|
body = r.json()
|
|
out: Dict[str, dict] = {}
|
|
for item in body.get("response", {}).get("publishedfiledetails", []) or []:
|
|
out[item["publishedfileid"]] = item
|
|
return out
|
|
|
|
|
|
COLLECTION_URL = (
|
|
"https://api.steampowered.com/ISteamRemoteStorage/GetCollectionDetails/v1/"
|
|
)
|
|
|
|
|
|
async def fetch_collection_details(
|
|
client: httpx.AsyncClient,
|
|
collection_ids: List[str],
|
|
) -> Dict[str, CollectionEntry]:
|
|
"""Resolve candidate collection IDs to their child wsids.
|
|
|
|
Returns a dict keyed by collection_id with shape:
|
|
{ "result": int, "children": List[str] }
|
|
|
|
Anonymous endpoint; no API key needed. result==1 means valid collection;
|
|
result!=1 means the ID isn't a collection (could be a mod, deleted, or
|
|
private). Caller decides what to do with non-1 results - see Spec B+F
|
|
§10 Q3 "Partial expansion failure" and Q4 "Flakiness".
|
|
"""
|
|
if not collection_ids:
|
|
return {}
|
|
data: Dict[str, str] = {"collectioncount": str(len(collection_ids))}
|
|
for i, cid in enumerate(collection_ids):
|
|
data[f"publishedfileids[{i}]"] = cid
|
|
r = await client.post(COLLECTION_URL, data=data)
|
|
r.raise_for_status()
|
|
body = r.json()
|
|
out: Dict[str, CollectionEntry] = {}
|
|
for item in body.get("response", {}).get("collectiondetails", []) or []:
|
|
cid = item.get("publishedfileid")
|
|
if not cid:
|
|
continue
|
|
out[cid] = {
|
|
"result": int(item.get("result") or 0),
|
|
"children": [
|
|
c.get("publishedfileid", "")
|
|
for c in (item.get("children") or [])
|
|
if c.get("publishedfileid")
|
|
],
|
|
}
|
|
return out
|