la-li-lu-le-lo/steam-sync.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
steam-sync.py — tiny local helper for la-li-lu-le-lo, the MGS2 Master Collection
100% companion.

It does two things:
  1. serves the folder it lives in over http://127.0.0.1:8765/  (so the page gets a
     real origin and localStorage saves reliably)
  2. exposes /api/steam?profile=<vanity|steamid64> which fetches that profile's
     public MGS2 Master Collection achievements from Steam and returns them as JSON

No API key. No third-party packages. Only one outbound request, to the public
Steam Community page for the profile you ask for.

    python3 steam-sync.py            # then open http://127.0.0.1:8765/
    python3 steam-sync.py --open     # ...and open the browser for you
    PORT=9000 python3 steam-sync.py  # different port
"""
import http.server
import json
import os
import re
import socket
import socketserver
import sys
import urllib.error
import urllib.parse
import urllib.request
import webbrowser
from xml.etree import ElementTree as ET

APPID = "2131640"
GAME = "METAL GEAR SOLID 2: Sons of Liberty - Master Collection Version"
PORT = int(os.environ.get("PORT", "8765"))
ROOT = os.path.dirname(os.path.abspath(__file__))
UA = ("Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
      "(KHTML, like Gecko) Chrome/120.0 Safari/537.36")


def steam_url(profile):
    profile = profile.strip()
    if re.fullmatch(r"\d{17}", profile):
        return "https://steamcommunity.com/profiles/%s/stats/%s/?tab=achievements&xml=1" % (profile, APPID)
    return "https://steamcommunity.com/id/%s/stats/%s/?tab=achievements&xml=1" % (
        urllib.parse.quote(profile), APPID)


def fetch_steam(profile):
    url = steam_url(profile)
    req = urllib.request.Request(url, headers={"User-Agent": UA, "Accept-Language": "en"})
    try:
        with urllib.request.urlopen(req, timeout=25) as r:
            raw = r.read()
    except urllib.error.HTTPError as e:
        return {"ok": False, "error": "Steam returned HTTP %s." % e.code,
                "hint": "Try again in a moment; Steam rate-limits rapid requests."}
    except Exception as e:
        return {"ok": False, "error": "Could not reach Steam: %s" % e,
                "hint": "Check your internet connection."}

    try:
        root = ET.fromstring(raw)
    except ET.ParseError:
        return {"ok": False, "error": "Steam sent something unreadable.",
                "hint": "The profile may be private or Steam may be rate-limiting."}

    err = root.find("error")
    if err is not None and (err.text or "").strip():
        msg = err.text.strip()
        hint = "Check the spelling, or paste the whole profile URL."
        if "private" in msg.lower():
            hint = "Steam > your profile > Edit Profile > Privacy Settings > set 'Game details' to Public."
        return {"ok": False, "error": msg, "hint": hint}

    privacy = (root.findtext("privacyState") or "").strip()
    if privacy and privacy != "public":
        return {"ok": False,
                "error": "That profile is %s." % privacy,
                "hint": "Steam > Edit Profile > Privacy Settings > set 'Game details' to Public, then sync again."}

    ach_nodes = root.findall("./achievements/achievement")
    if not ach_nodes:
        return {"ok": False,
                "error": "No MGS2 Master Collection stats on that account.",
                "hint": "The account must own the game (appid %s) and have launched it at least once." % APPID}

    out = []
    for a in ach_nodes:
        out.append({
            "api": (a.findtext("apiname") or "").strip(),
            "name": (a.findtext("name") or "").strip(),
            "desc": (a.findtext("description") or "").strip(),
            "unlocked": a.get("closed") == "1",
            "icon": (a.findtext("iconOpen") or "").strip(),
            "iconLocked": (a.findtext("iconClosed") or "").strip(),
        })

    return {
        "ok": True,
        "game": GAME,
        "appid": APPID,
        "steamid64": (root.findtext("./player/steamID64") or "").strip(),
        # This XML carries no persona name, only steamID64 and customURL.
        "persona": ((root.findtext("./player/steamID")
                     or root.findtext("./player/customURL")
                     or profile) or "").strip(),
        "unlockedCount": sum(1 for a in out if a["unlocked"]),
        "total": len(out),
        "achievements": out,
    }


class Handler(http.server.SimpleHTTPRequestHandler):
    def __init__(self, *args, **kw):
        super().__init__(*args, directory=ROOT, **kw)

    def log_message(self, fmt, *args):
        # Never let a closed/dead output pipe (e.g. when run in the background)
        # take the whole server down with it.
        if "/api/" not in (self.path or ""):
            return
        try:
            sys.stderr.write("  steam-sync: %s %s\n" % (self.command, self.path.split("?")[0]))
            sys.stderr.flush()
        except Exception:
            pass

    def _cors(self):
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Headers", "*")
        self.send_header("Access-Control-Allow-Methods", "GET, OPTIONS")

    def do_OPTIONS(self):
        self.send_response(204)
        self._cors()
        self.end_headers()

    def do_GET(self):
        if self.path.split("?")[0] != "/api/steam":
            return super().do_GET()

        qs = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query)
        profile = (qs.get("profile") or [""])[0].strip()
        if not profile:
            payload = {"ok": False, "error": "No profile given.",
                       "hint": "Enter a vanity name or SteamID64."}
        else:
            try:
                payload = fetch_steam(profile)
            except Exception as e:  # never 500 on the user
                payload = {"ok": False, "error": "Unexpected error: %s" % e, "hint": ""}

        body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        self._cors()
        self.end_headers()
        self.wfile.write(body)


class Server(socketserver.ThreadingTCPServer):
    allow_reuse_address = True
    daemon_threads = True


def main():
    # A stalled connection to Steam can never hang a handler thread forever.
    socket.setdefaulttimeout(30)
    if not os.path.exists(os.path.join(ROOT, "index.html")):
        print("!! index.html is not in %s — keep steam-sync.py next to it." % ROOT)
    try:
        httpd = Server(("127.0.0.1", PORT), Handler)
    except OSError as e:
        print("!! Could not bind port %d: %s" % (PORT, e))
        print("   Another copy may already be running, or set PORT=<other> and retry.")
        return 1
    url = "http://127.0.0.1:%d/" % PORT
    print("=" * 66)
    print(" la-li-lu-le-lo — MGS2 100% companion (local helper)")
    print("=" * 66)
    print("    serving : %s" % ROOT)
    print("    page    : %s" % url)
    print("    api     : %sapi/steam?profile=<vanity or SteamID64>" % url)
    print("    stop    : Ctrl+C")
    print("-" * 66)
    if "--open" in sys.argv:
        webbrowser.open(url)
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        print("\n  stopped.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())