srcoder.moe/utils/steam.py

import json
import os
import threading
import time
import urllib.error
import urllib.parse
import urllib.request

from flask import Blueprint, jsonify

steam_bp = Blueprint("steam", __name__)

STEAM_API_KEY = os.environ.get("STEAM_API_KEY", "").strip()
STEAM_ID = os.environ.get("STEAM_ID", "76561199137306791").strip()
ACTIVITY_TTL = 60
RECENT_COUNT = 5
TOP_COUNT = 5

API_BASE = "https://api.steampowered.com"
STORE_APP = "https://store.steampowered.com/app"
PROFILE_URL = f"https://steamcommunity.com/profiles/{STEAM_ID}"

_CACHE_LOCK = threading.Lock()
_CACHE: dict = {"at": 0.0, "payload": None}

PERSONA_STATES = {
    0: "не в сети",
    1: "в сети",
    2: "занят",
    3: "отошёл",
    4: "спит",
    5: "ищет обмен",
    6: "ищет поиграть",
    7: "невидим",
}


def _get(path, **params):
    params["key"] = STEAM_API_KEY
    url = f"{API_BASE}{path}?{urllib.parse.urlencode(params)}"
    req = urllib.request.Request(url, headers={"Accept": "application/json"})
    with urllib.request.urlopen(req, timeout=10) as resp:
        return json.load(resp)


def _header(appid):
    return f"https://cdn.akamai.steamstatic.com/steam/apps/{appid}/header.jpg"


def _store_url(appid):
    return f"{STORE_APP}/{appid}/"


def _serialize(game):
    appid = int(game["appid"])
    minutes = int(game.get("playtime_forever") or 0)
    two_weeks = game.get("playtime_2weeks")
    last_played = int(game.get("rtime_last_played") or 0)
    return {
        "appid": appid,
        "name": game.get("name") or f"App #{appid}",
        "hours_total": round(minutes / 60, 1),
        "hours_2weeks": round(int(two_weeks) / 60, 1) if two_weeks else None,
        "last_played": last_played or None,
        "header": _header(appid),
        "url": _store_url(appid),
    }


def _fetch():
    recent_resp = _get(
        "/IPlayerService/GetRecentlyPlayedGames/v1/",
        steamid=STEAM_ID,
        count=RECENT_COUNT,
    )
    owned_resp = _get(
        "/IPlayerService/GetOwnedGames/v1/",
        steamid=STEAM_ID,
        include_appinfo=1,
        include_played_free_games=1,
    )
    summary_resp = _get("/ISteamUser/GetPlayerSummaries/v2/", steamids=STEAM_ID)

    recent = [
        _serialize(game)
        for game in (recent_resp.get("response") or {}).get("games") or []
    ]

    owned = owned_resp.get("response") or {}
    played = [
        game
        for game in owned.get("games") or []
        if int(game.get("playtime_forever") or 0) > 0
    ]
    played.sort(key=lambda game: int(game.get("playtime_forever") or 0), reverse=True)
    top = [_serialize(game) for game in played[:TOP_COUNT]]

    players = (summary_resp.get("response") or {}).get("players") or []
    player = players[0] if players else {}

    game_id = str(player.get("gameid") or "")
    game_name = player.get("gameextrainfo") or ""
    now = None
    if game_id:
        appid = int(game_id)
        if not game_name:
            known = next((g["name"] for g in recent + top if g["appid"] == appid), None)
            game_name = known or ""
        now = {
            "appid": appid,
            "name": game_name,
            "header": _header(appid),
            "url": _store_url(appid),
        }

    state = int(player.get("personastate") or 0)
    return {
        "ok": True,
        "updated_at": int(time.time()),
        "now": now,
        "state": PERSONA_STATES.get(state, ""),
        "profile": {
            "url": player.get("profileurl") or PROFILE_URL,
            "name": player.get("personaname") or "",
            "public": int(player.get("communityvisibilitystate") or 0) == 3,
        },
        "recent": recent,
        "top": top,
        "game_count": int(owned.get("game_count") or 0) or None,
    }


def _fetch_activity():
    try:
        return _fetch()
    except urllib.error.HTTPError as exc:
        if exc.code in (401, 403):
            raise RuntimeError("Steam API: неверный или недействительный ключ") from exc
        if exc.code == 429:
            raise RuntimeError("Steam API: лимит запросов, попробуйте позже") from exc
        raise RuntimeError(f"Steam API: HTTP {exc.code}") from exc
    except urllib.error.URLError as exc:
        raise RuntimeError(f"нет связи с Steam API: {exc.reason}") from exc


def _serve():
    now = time.monotonic()
    with _CACHE_LOCK:
        if _CACHE["payload"] is not None and now - _CACHE["at"] < ACTIVITY_TTL:
            return jsonify(_CACHE["payload"]), 200

        try:
            payload = _fetch_activity()
        except (RuntimeError, OSError, ValueError, KeyError, AttributeError) as exc:
            if _CACHE["payload"] is not None:
                stale = dict(_CACHE["payload"])
                stale["stale"] = True
                return jsonify(stale), 200
            return jsonify(ok=False, error=str(exc)), 502

        _CACHE["at"] = now
        _CACHE["payload"] = payload
        return jsonify(payload), 200


@steam_bp.route("/api/steam/activity")
def steam_activity():
    if not STEAM_API_KEY:
        return jsonify(ok=False, error="STEAM_API_KEY is not set"), 503
    return _serve()