Steam profile API for server-side applications

# Save as profile_batch_view.py; this reads saved JSON and makes no requests.
import json
import math
import re
import sys


def need(condition, message):
    if not condition:
        raise ValueError(message)


def batch_view(data, wanted):
    need(1 <= len(wanted) <= 64 and all(
        isinstance(sid, str) and re.fullmatch(r"[0-9]{17}", sid) for sid in wanted),
        "Supply 1..64 numeric SteamID64 strings from the request")
    wanted = list(dict.fromkeys(wanted))
    need(isinstance(data, dict), "Expected a detailed batch object")
    profiles, missing, meta = (data.get(key) for key in ("profiles", "missing", "meta"))
    need(isinstance(profiles, dict) and isinstance(missing, list)
         and isinstance(meta, dict), "Detailed response unavailable; check HTTP status and capability")
    need(type(meta.get("version")) is int and meta["version"] == 1,
         "Unsupported details version")
    need(isinstance(meta.get("request_id"), str)
         and re.fullmatch(r"[0-9a-f]{32}", meta["request_id"]), "Invalid request ID")
    for key in ("requested", "returned", "charged_units"):
        need(type(meta.get(key)) is int and meta[key] >= 0, "Invalid batch count")
    need(meta.get("quota_admitted") is True and meta["requested"] == len(wanted)
         and meta["returned"] == len(profiles) and meta["charged_units"] == len(wanted),
         "Request identity or admitted quota does not match")
    need(set(profiles) <= set(wanted) and len(missing) <= len(wanted),
         "Unexpected result IDs")
    cards = {}
    for sid, profile in profiles.items():
        need(isinstance(profile, dict) and profile.get("steamid") == sid,
             "Profile identity does not match")
        need(profile.get("personaname") is None or isinstance(profile["personaname"], str),
             "Invalid profile name")
        need(profile.get("source", "unknown") in
             ("steam_webapi", "steam_public_profile_xml", "unknown"), "Invalid profile source")
        stamp = profile.get("source_observed_at")
        try:
            known_time = type(stamp) in (int, float) and math.isfinite(stamp) and stamp > 0
        except OverflowError:
            known_time = False
        cards[sid] = {"steamid": sid, "personaname": profile.get("personaname"),
                      "source": profile.get("source", "unknown"),
                      "source_observed_at": stamp if known_time else None}
    failures, retry_ids, waits = {}, [], []
    for row in missing:
        need(isinstance(row, dict), "Invalid missing result")
        sid, error = row.get("steamid"), row.get("error")
        need(isinstance(sid, str) and sid in wanted and sid not in cards and sid not in failures,
             "Duplicate, overlapping or unexpected missing ID")
        need(type(row.get("retryable")) is bool and isinstance(error, dict)
             and isinstance(error.get("code"), str) and bool(error["code"])
             and type(error.get("status")) is int and 400 <= error["status"] <= 599,
             "Invalid per-ID error")
        delay = error.get("retry_after")
        need(delay is None or type(delay) is int and 1 <= delay <= 86400,
             "Invalid retry delay")
        if row["retryable"]:
            need(error["status"] in (429, 502, 503, 504), "Unexpected retry classification")
            retry_ids.append(sid)
            if delay is not None:
                waits.append(delay)
        failures[sid] = {"code": error["code"], "status": error["status"],
                         "retryable": row["retryable"]}
    need(set(cards) | set(failures) == set(wanted), "Batch does not cover the requested IDs")
    need(type(meta.get("complete")) is bool and meta["complete"] == (not failures),
         "Inconsistent completion flag")
    return {"profiles": cards, "missing": failures, "retry_candidates": retry_ids,
            "per_id_minimum_delay": max(waits) if waits else None,
            "charged_units": meta["charged_units"], "request_id": meta["request_id"]}


def reject_constant(value):
    raise ValueError("Non-finite JSON constant: " + value)


if __name__ == "__main__":
    try:
        with open("profiles.json", "rb") as saved:
            raw = saved.read(1048577)
        need(len(raw) <= 1048576, "Saved response exceeds the 1 MiB size limit")
        result = batch_view(json.loads(raw, parse_constant=reject_constant), sys.argv[1:])
        print(json.dumps(result, ensure_ascii=True, allow_nan=False, indent=2))
    except (OSError, ValueError, RecursionError) as error:
        raise SystemExit(str(error))