"""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 {}