fix(utils): Add code to MVC model

utils/steamutils.py

@@ -0,0 +1,162 @@
+import json
+import os
+import threading
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+
+
+from flask import jsonify
+
+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