Fetch a Steam inventory as JSON

steam_inventory.py

#!/usr/bin/env python3
"""
Steam inventory dumper with JSON output.

Output formats:
  - Default: Pretty-printed JSON with nested structure
  - --flat: JSONL (one item per line) for streaming/grep

Examples:
    ./steam_inventory.py <steamid64>
    ./steam_inventory.py <steamid64> --flat | grep -i knife
    ./steam_inventory.py <steamid64> | jq '[.games[].items[] | select(.count > 1)]'
"""

import argparse
import json
import re
import signal
import sys
import time
from collections import Counter
from dataclasses import asdict, dataclass, field
from functools import cache, wraps
from typing import Any, Callable, TypeVar

import requests

T = TypeVar("T")

# Constants
DEFAULT_RETRIES = 5
RETRY_DELAY = 3.0
REQUEST_TIMEOUT = 20
PAGE_SIZE = 2000
INTER_REQUEST_DELAY = 0.3

USER_AGENT = (
    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
)

# Mutable runtime config
_retries = DEFAULT_RETRIES
_verbose = False


@dataclass(slots=True)
class Item:
    """A single inventory item."""

    name: str
    count: int
    appid: int
    game: str


@dataclass(slots=True)
class GameInventory:
    """Inventory for a single game."""

    appid: int
    name: str
    items: list[Item] = field(default_factory=list)

    @property
    def total_items(self) -> int:
        return sum(item.count for item in self.items)

    def to_dict(self) -> dict[str, Any]:
        """Custom serialization (items only have name/count in nested output)."""
        return {
            "appid": self.appid,
            "name": self.name,
            "total_items": self.total_items,
            "unique_items": len(self.items),
            "items": [{"name": i.name, "count": i.count} for i in self.items],
        }


def print_error(msg: str) -> None:
    """Print JSON error to stderr."""
    print(json.dumps({"error": msg}), file=sys.stderr)


@cache
def get_session() -> requests.Session:
    """Get or create the singleton requests session."""
    session = requests.Session()
    session.headers["User-Agent"] = USER_AGENT
    return session


def with_retry(func: Callable[..., T | None]) -> Callable[..., T | None]:
    """Decorator to retry a function on failure with exponential backoff."""

    @wraps(func)
    def wrapper(*args: Any, **kwargs: Any) -> T | None:
        for attempt in range(_retries):
            try:
                if (result := func(*args, **kwargs)) is not None:
                    return result
                if _verbose:
                    print(
                        json.dumps(
                            {
                                "status": "retrying",
                                "attempt": attempt + 1,
                                "reason": "empty result",
                            }
                        ),
                        file=sys.stderr,
                    )
            except Exception as e:
                if _verbose:
                    print(
                        json.dumps(
                            {
                                "status": "retrying",
                                "attempt": attempt + 1,
                                "reason": str(e),
                            }
                        ),
                        file=sys.stderr,
                    )
            if attempt < _retries - 1:
                time.sleep(RETRY_DELAY * (2**attempt))
        return None

    return wrapper


@cache
def get_game_name(appid: int) -> str:
    """Fetch game name from Steam API with caching and fallback."""

    @with_retry
    def fetch() -> str | None:
        url = f"https://store.steampowered.com/api/appdetails?appids={appid}"
        data = get_session().get(url, timeout=REQUEST_TIMEOUT).json()
        return data[str(appid)]["data"]["name"]

    return fetch() or f"AppID {appid}"


@with_retry
def fetch_inventory_page(
    steamid: str, appid: int, contextid: int, start_assetid: str | None = None
) -> dict[str, Any] | None:
    """Fetch a single page of inventory items."""
    url = f"https://steamcommunity.com/inventory/{steamid}/{appid}/{contextid}"
    params: dict[str, Any] = {"l": "english", "count": PAGE_SIZE}
    if start_assetid:
        params["start_assetid"] = start_assetid

    resp = get_session().get(url, params=params, timeout=REQUEST_TIMEOUT)
    return resp.json() if resp.status_code == 200 and resp.text else None


def get_items_in_context(steamid: str, appid: int, contextid: int) -> Counter[str]:
    """Fetch all items from a specific inventory context."""
    items: Counter[str] = Counter()
    start_assetid: str | None = None

    while True:
        data = fetch_inventory_page(steamid, appid, contextid, start_assetid)
        if not data or not data.get("success") or not data.get("assets"):
            break

        desc_map = {
            f"{d['classid']}_{d['instanceid']}": d for d in data.get("descriptions", [])
        }

        for asset in data["assets"]:
            if desc := desc_map.get(f"{asset['classid']}_{asset['instanceid']}"):
                name = desc.get("market_hash_name") or desc.get("name", "Unknown")
                items[name] += int(asset.get("amount", 1))

        if not data.get("more_items"):
            break
        start_assetid = data.get("last_assetid")
        time.sleep(INTER_REQUEST_DELAY)

    return items


@with_retry
def discover_games(steamid: str) -> dict[int, list[int]] | None:
    """Discover which games have items and their active context IDs.

    Returns: dict mapping appid -> list of context IDs with items, or None on failure.
    """
    url = f"https://steamcommunity.com/profiles/{steamid}/inventory/"
    resp = get_session().get(url, timeout=REQUEST_TIMEOUT)

    if resp.status_code != 200 or not resp.text:
        return None

    if not (match := re.search(r"g_rgAppContextData\s*=\s*(\{.*?\});", resp.text)):
        return None

    data = json.loads(match.group(1))
    result: dict[int, list[int]] = {}
    for appid_str, appdata in data.items():
        if appdata.get("asset_count", 0) > 0:
            contexts = [
                int(ctx_id)
                for ctx_id, ctx_data in appdata.get("rgContexts", {}).items()
                if ctx_data.get("asset_count", 0) > 0
            ]
            if contexts:
                result[int(appid_str)] = contexts
    return result or None


def fetch_inventory(steamid: str) -> list[GameInventory]:
    """Fetch complete inventory for a Steam user."""
    if not (games := discover_games(steamid)):
        return []

    inventories: list[GameInventory] = []

    for appid in sorted(games):
        game_name = get_game_name(appid)
        combined: Counter[str] = Counter()
        contexts = games[appid]

        for i, ctx in enumerate(contexts):
            combined.update(get_items_in_context(steamid, appid, ctx))
            if i < len(contexts) - 1:
                time.sleep(INTER_REQUEST_DELAY)

        if combined:
            sorted_items = [
                Item(name=n, count=c, appid=appid, game=game_name)
                for n, c in sorted(combined.items(), key=lambda x: (-x[1], x[0]))
            ]
            inventories.append(
                GameInventory(appid=appid, name=game_name, items=sorted_items)
            )

    return inventories


def output_nested(steamid: str, inventories: list[GameInventory]) -> None:
    """Output in pretty-printed nested JSON format."""
    result = {
        "steamid": steamid,
        "games": [inv.to_dict() for inv in inventories],
        "summary": {
            "total_items": sum(inv.total_items for inv in inventories),
            "total_games": len(inventories),
        },
    }
    print(json.dumps(result, ensure_ascii=False, indent=2))


def output_flat(inventories: list[GameInventory]) -> None:
    """Output in JSONL format (one item per line)."""
    for inv in inventories:
        for item in inv.items:
            print(json.dumps(asdict(item), ensure_ascii=False))


def main() -> int:
    """Main entry point."""
    global _retries, _verbose
    if hasattr(signal, "SIGPIPE"):
        signal.signal(signal.SIGPIPE, signal.SIG_DFL)

    parser = argparse.ArgumentParser(
        description="Dump Steam inventory as JSON",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s <steamid64>
  %(prog)s <steamid64> --flat | grep M4A4
  %(prog)s <steamid64> | jq '.summary'
        """,
    )
    parser.add_argument("steamid", help="SteamID64 (17-digit number)")
    parser.add_argument(
        "--flat", "-f", action="store_true", help="JSONL output (one item per line)"
    )
    parser.add_argument(
        "--verbose", "-v", action="store_true", help="Show retry attempts on stderr"
    )
    parser.add_argument(
        "--retries",
        "-r",
        type=int,
        default=DEFAULT_RETRIES,
        help=f"Retries (default: {DEFAULT_RETRIES})",
    )

    args = parser.parse_args()

    if not (args.steamid.isdigit() and len(args.steamid) == 17):
        print_error("SteamID64 must be a 17-digit number")
        return 1

    _retries, _verbose = args.retries, args.verbose

    try:
        inventories = fetch_inventory(args.steamid)
    except KeyboardInterrupt:
        print_error("Interrupted")
        return 130

    if not inventories:
        print_error("No public inventory found")
        return 1

    output_flat(inventories) if args.flat else output_nested(args.steamid, inventories)

    return 0


if __name__ == "__main__":
    sys.exit(main())