rain
a63606b5c8
The Pages deployment is now the primary copy, so two things had to change: - Hosted pages fall back to steam-sync.py on 127.0.0.1:8765 instead of refusing to sync. steam-sync.py now answers Chrome's Private Network Access preflight (Access-Control-Allow-Private-Network), which is what a public https page needs before it may call into localhost. - The build sources lived in /tmp and would have been lost. They now live in build/, and build.py outputs to the repo root. deploy.sh rebuilds, commits and pushes in one step. Editing index.html directly is no longer the workflow - edit build/ and run ./deploy.sh.
202 lines
7.6 KiB
Python
202 lines
7.6 KiB
Python
#!/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")
|
|
# Chrome asks for this when a public https page (e.g. the Pages deployment)
|
|
# calls into 127.0.0.1. Without it the request is blocked before it is sent.
|
|
self.send_header("Access-Control-Allow-Private-Network", "true")
|
|
|
|
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())
|