Rewrite server: FastAPI + auth (users/roles/tokens) + admin WebUI + server-authoritative covers/metadata

app/server_client.py

@@ -0,0 +1,112 @@
+"""Client-side HTTP access to the psppt server (auth + content fetches).
+
+The server is authoritative for covers and metadata. When a server is
+configured (URL + per-user token), the cover/metadata paths use it and show an
+error rather than falling back to libretro/IGDB/etc.
+"""
+import json
+import logging
+import urllib.error
+import urllib.parse
+import urllib.request
+
+from . import config
+
+log = logging.getLogger("server_client")
+
+_CFG_FILE = config.STATE_DIR / "server.json"
+_cache = {"loaded": False, "cfg": {}}
+
+
+def _load():
+    if not _cache["loaded"]:
+        cfg = {}
+        try:
+            cfg = json.loads(_CFG_FILE.read_text(encoding="utf-8"))
+        except (OSError, ValueError):
+            cfg = {}
+        _cache["cfg"] = cfg
+        _cache["loaded"] = True
+    return _cache["cfg"]
+
+
+def _save(cfg):
+    _CFG_FILE.parent.mkdir(parents=True, exist_ok=True)
+    _CFG_FILE.write_text(json.dumps(cfg, indent=2), encoding="utf-8")
+    _cache["cfg"] = cfg
+    _cache["loaded"] = True
+
+
+def configured():
+    cfg = _load()
+    return bool(cfg.get("url") and cfg.get("token"))
+
+
+def url():
+    return (_load().get("url") or config.STORE_SERVER_URL or "").rstrip("/")
+
+
+def token():
+    return _load().get("token") or config.STORE_TOKEN or ""
+
+
+def username():
+    return _load().get("username") or ""
+
+
+def logout():
+    _save({"url": url(), "token": "", "username": ""})
+
+
+def login(server_url, user, password):
+    """Authenticate against the server and store the per-user token."""
+    base = (server_url or "").rstrip("/")
+    req = urllib.request.Request(
+        f"{base}/api/auth/login",
+        data=json.dumps({"username": user, "password": password}).encode("utf-8"),
+        headers={"Content-Type": "application/json", "User-Agent": "psppt/0.1"},
+        method="POST",
+    )
+    with urllib.request.urlopen(req, timeout=15) as resp:
+        data = json.loads(resp.read())
+    _save({"url": base, "token": data.get("token") or "", "username": data.get("username") or user})
+    return data
+
+
+def _get(path, timeout=15):
+    req = urllib.request.Request(
+        f"{url()}{path}",
+        headers={"Authorization": f"Bearer {token()}", "User-Agent": "psppt/0.1"},
+    )
+    with urllib.request.urlopen(req, timeout=timeout) as resp:
+        return resp.read()
+
+
+def _lookup_id(system, name):
+    """Return the server release id for a name (first match), or None."""
+    q = urllib.parse.quote(name or "")
+    data = json.loads(_get(f"/api/releases?system={system}&q={q}&limit=1", timeout=15))
+    items = data.get("items") or []
+    return items[0]["id"] if items else None
+
+
+def fetch_cover(system, name):
+    """Return cover bytes for a release, or None (no cover on the server)."""
+    rid = _lookup_id(system, name)
+    if rid is None:
+        return None
+    try:
+        return _get(f"/api/cover/{rid}", timeout=30)
+    except urllib.error.HTTPError as exc:
+        if exc.code == 404:
+            return None
+        raise
+
+
+def fetch_metadata(system, name):
+    """Return metadata dict for a release, or None."""
+    rid = _lookup_id(system, name)
+    if rid is None:
+        return None
+    detail = json.loads(_get(f"/api/releases/{rid}", timeout=15))
+    return detail.get("meta") or {}