""" Hollow-decky Decky Plugin — main.py Backend that searches Steam games by name (Steam Store + delisted cache), injects games into SLSsteam's config.yaml (packages, depots, app IDs), and provides FakeAppId, cr0wbar-fixes, collection tagging, and SLS config backups. Game Mode compatible. Core safety feature: every SLS config write is preceded by a timestamped backup. A toggleable "force refresh" option signals the frontend to trigger a Steam library refresh via its internal IPC. """ import os import sys # ── Bundle path: third-party packages live in defaults/python ──────────────── _HERE = os.path.dirname(os.path.abspath(__file__)) _BUNDLE = os.path.join(_HERE, "defaults", "python") if os.path.isdir(_BUNDLE): sys.path.insert(0, _BUNDLE) import re import json import time import shutil import asyncio import logging import threading import hashlib from pathlib import Path from datetime import datetime, timezone import requests # --------------------------------------------------------------------------- # Decky plugin module — available at runtime inside Decky Loader # --------------------------------------------------------------------------- try: import decky # type: ignore[import-untyped] except ImportError: decky = None # graceful fallback for offline dev / linting # --------------------------------------------------------------------------- # Logging # --------------------------------------------------------------------------- logger = logging.getLogger("hollow-decky") # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- BACKUP_DIR_NAME = "hollow_decky_backups" LEGACY_BACKUP_DIR_NAME = "gasket_backups" APP_CONFIG_DIR = "hollow-decky" LEGACY_APP_CONFIG_DIR = "gasket-decky" LEGACY_PLUGIN_DIR = "gasket" MAX_BACKUPS = 20 SETTINGS_DEFAULTS = { "restart_after_inject": True, "show_restart_button": True, "add_app_id_to_sls": False, } # =================================================================== # SLS Backups # =================================================================== def _backup_dir() -> Path: if decky: return Path(decky.DECKY_PLUGIN_RUNTIME_DIR) / BACKUP_DIR_NAME return Path.home() / ".config" / APP_CONFIG_DIR / BACKUP_DIR_NAME def _sls_config_path() -> Path | None: """Find SLSsteam config.yaml.""" candidates = [ Path.home() / ".config" / "SLSsteam" / "config.yaml", Path.home() / ".local/share/SLSsteam/config.yaml", Path("/home/deck/.config/SLSsteam/config.yaml"), ] for p in candidates: if p.exists(): return p return None def backup_sls() -> str: """Create a backup of SLSsteam config.yaml.""" sls_path = _sls_config_path() if not sls_path: return "SLSsteam config not found" backup_dir = _backup_dir() backup_dir.mkdir(parents=True, exist_ok=True) timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") backup_file = backup_dir / f"sls.{timestamp}.yaml" shutil.copy2(sls_path, backup_file) _prune_sls_backups() logger.info("SLS backup created: %s", backup_file.name) return backup_file.name def restore_sls(backup_name: str) -> dict: """Restore SLSsteam config.yaml from a backup.""" sls_path = _sls_config_path() if not sls_path: return {"error": "SLSsteam config not found"} backup_file = _backup_dir() / backup_name if not backup_file.exists(): return {"error": f"Backup not found: {backup_name}"} # Create a pre-restore backup first pre_restore = f"pre_restore.{datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')}.yaml" shutil.copy2(sls_path, _backup_dir() / pre_restore) shutil.copy2(backup_file, sls_path) logger.info("SLS restored from %s", backup_name) return {"restored": backup_name, "pre_restore": pre_restore} def list_sls_backups() -> list[dict]: """List available SLS config backups.""" backup_dir = _backup_dir() if not backup_dir.exists(): return [] backups = [] for f in sorted(backup_dir.glob("*.yaml"), reverse=True): stat = f.stat() # Parse timestamp from filename: sls.20260529T202720Z.yaml or pre_restore.20260529T202720Z.yaml stamp = f.stem.replace("sls.", "").replace("pre_restore.", "") backups.append({ "name": f.name, "stamp": stamp, "size": stat.st_size, "mtime": int(stat.st_mtime), }) _prune_sls_backups() return backups[:MAX_BACKUPS] def _prune_sls_backups(): """Remove old backups beyond MAX_BACKUPS, keeping the most recent.""" backup_dir = _backup_dir() if not backup_dir.exists(): return backups = sorted(backup_dir.glob("*.yaml"), key=lambda f: f.stat().st_mtime, reverse=True) for old in backups[MAX_BACKUPS:]: old.unlink(missing_ok=True) # =================================================================== # Steam API Helpers # =================================================================== # --------------------------------------------------------------------------- # Delisted games cache (from steam-tracker.com) # --------------------------------------------------------------------------- _DELISTED_CACHE: list[dict] = [] # [{appid, name, changed_at}, ...] _DELISTED_CACHE_TIME: float = 0.0 # epoch of last download _DELISTED_CACHE_TTL = 86400 # refresh every 24h def _delisted_cache_path() -> Path: if decky: return Path(decky.DECKY_PLUGIN_RUNTIME_DIR) / "delisted_cache.json" return Path.home() / ".config" / APP_CONFIG_DIR / "delisted_cache.json" def _load_delisted_cache() -> None: """Load delisted games from cache file.""" global _DELISTED_CACHE, _DELISTED_CACHE_TIME try: p = _delisted_cache_path() if p.exists(): with open(p) as f: data = json.load(f) _DELISTED_CACHE = data.get("games", []) _DELISTED_CACHE_TIME = data.get("fetched_at", 0.0) logger.info("Loaded %d delisted games from cache (fetched %s)", len(_DELISTED_CACHE), data.get("fetched_at")) except Exception as e: logger.warning("Failed to load delisted cache: %s", e) def refresh_delisted_cache(force: bool = False) -> dict: """Download delisted games list from steam-tracker.com and cache it. Returns {"count": int, "error": str|None} """ global _DELISTED_CACHE, _DELISTED_CACHE_TIME import time now = time.time() if not force and _DELISTED_CACHE and (now - _DELISTED_CACHE_TIME) < _DELISTED_CACHE_TTL: return {"count": len(_DELISTED_CACHE), "error": None} try: resp = requests.get( "https://steam-tracker.com/api?action=GetAppListV3&type=game", timeout=30, headers={"User-Agent": "Mozilla/5.0"}, ) resp.raise_for_status() data = resp.json() all_apps = data.get("removed_apps", []) # Keep only delisted games (category_id=1) with a name delisted = [ {"appid": str(a["appid"]), "name": a.get("name", ""), "changed_at": a.get("changed_at", "")} for a in all_apps if a.get("category_id") == 1 and a.get("name") ] _DELISTED_CACHE = delisted _DELISTED_CACHE_TIME = now # Save to disk try: p = _delisted_cache_path() p.parent.mkdir(parents=True, exist_ok=True) with open(p, "w") as f: json.dump({"fetched_at": now, "games": delisted}, f) except Exception as e: logger.warning("Failed to save delisted cache: %s", e) logger.info("Refreshed delisted cache: %d games", len(delisted)) return {"count": len(delisted), "error": None} except Exception as e: logger.error("Failed to refresh delisted cache: %s", e) return {"count": len(_DELISTED_CACHE), "error": str(e)} # Load cache on module import _load_delisted_cache() # --------------------------------------------------------------------------- # Search: Steam Store + Delisted cache + steamcmd.net appid lookup # --------------------------------------------------------------------------- def search_games(query: str, limit: int = 20, appid: bool = False) -> dict: """Search for games using Steam Store + delisted cache + steamcmd.net. No API key required. Search by name uses Steam Store and steam-tracker delisted games cache. Search by appid uses steamcmd.net. Query params: query (required) — search text or exact app ID limit (int, default 20) — max results for name search appid (bool, default False) — set True to match on exact game_id instead of name """ if not query or len(query.strip()) < 1: return {"error": "Query must be at least 1 character", "results": []} blacklist = [ "soundtrack", "ost", "original soundtrack", "artbook", "graphic novel", "demo", "server", "dedicated server", "tool", "sdk", "3d print model", ] # --- App ID lookup: steamcmd.net + Steam Store delisted check --- if appid: app_id = query.strip() if not app_id.isdigit(): return {"error": "App ID must be numeric", "results": []} try: details = get_app_details(app_id) if details.get("error"): # Also check delisted cache for name for g in _DELISTED_CACHE: if g["appid"] == app_id: return {"results": [{"app_id": app_id, "title": g["name"], "delisted": True}], "total": 1} return {"results": [], "total": 0} # Check if delisted via Steam Store delisted = _check_delisted(app_id) result = {"app_id": app_id, "title": details.get("title", app_id)} if delisted: result["delisted"] = True return {"results": [result], "total": 1} except Exception as e: return {"error": str(e), "results": []} # --- Name search: Steam Store + delisted cache --- if len(query.strip()) < 3: return {"error": "Query must be at least 3 characters for name search", "results": []} # Ensure delisted cache is loaded (lazy refresh if empty) if not _DELISTED_CACHE: refresh_delisted_cache() seen_ids: set[str] = set() results: list[dict] = [] query_lower = query.strip().lower() # 1. Steam Store search (listed games) try: resp = requests.get( "https://store.steampowered.com/api/storesearch/", params={"term": query.strip(), "l": "english", "cc": "us"}, timeout=15, headers={"User-Agent": "Mozilla/5.0"}, ) resp.raise_for_status() data = resp.json() for item in data.get("items", [])[:limit]: name = item.get("name", "") gid = str(item.get("id", "")) name_lower = name.lower() if any(re.search(rf"\b{re.escape(kw)}\b", name_lower) for kw in blacklist): continue if gid in seen_ids: continue seen_ids.add(gid) results.append({"app_id": gid, "title": name}) # Batch delisted check on store results if results: app_ids_to_check = [r["app_id"] for r in results[:10]] try: ids_param = ",".join(app_ids_to_check) resp2 = requests.get( f"https://store.steampowered.com/api/appdetails?appids={ids_param}&cc=us&l=en", timeout=15, headers={"User-Agent": "Mozilla/5.0"}, ) if resp2.status_code == 200: store_data = resp2.json() for r in results: aid = r["app_id"] if aid in store_data: app_entry = store_data[aid] if app_entry.get("success") is False: r["delisted"] = True elif app_entry.get("success"): info = app_entry.get("data", {}) app_type = (info.get("type") or "").lower() packages = info.get("packages") or [] is_free = info.get("is_free", False) has_price = bool(info.get("price_overview")) if app_type == "advertising": r["delisted"] = True elif not packages and not has_price and not is_free: r["delisted"] = True except Exception: pass except requests.RequestException: pass # Don't fail entirely if store search fails # 2. Delisted cache search (games not in store results) for g in _DELISTED_CACHE: if g["appid"] in seen_ids: continue if query_lower in g["name"].lower(): name_lower = g["name"].lower() if any(re.search(rf"\b{re.escape(kw)}\b", name_lower) for kw in blacklist): continue seen_ids.add(g["appid"]) results.append({"app_id": g["appid"], "title": g["name"], "delisted": True}) if len(results) >= limit + 20: # Allow extra before final sort+trim break # Sort: non-delisted first, then delisted; each group alpha by title results.sort(key=lambda r: (r.get("delisted", False), r["title"].lower())) return {"results": results[:limit], "total": len(results[:limit])} def _check_delisted(app_id: str) -> bool: """Check if a single app is delisted via Steam Store appdetails + delisted cache.""" # Check delisted cache first for g in _DELISTED_CACHE: if g["appid"] == app_id: return True # Then check Steam Store try: resp = requests.get( f"https://store.steampowered.com/api/appdetails?appids={app_id}&cc=us&l=en", timeout=10, headers={"User-Agent": "Mozilla/5.0"}, ) if resp.status_code == 200: store_data = resp.json().get(app_id, {}) if store_data.get("success") is False: return True if store_data.get("success"): info = store_data.get("data", {}) app_type = (info.get("type") or "").lower() packages = info.get("packages") or [] is_free = info.get("is_free", False) has_price = bool(info.get("price_overview")) if app_type == "advertising": return True if not packages and not has_price and not is_free: return True except Exception: pass return False def get_app_details(app_id: str) -> dict: """Fetch depot and DLC info from SteamCMD API.""" try: resp = requests.get( f"https://api.steamcmd.net/v1/info/{app_id}", timeout=15, ) resp.raise_for_status() data = resp.json() app_data = data.get("data", {}).get(app_id, {}) name = app_data.get("common", {}).get("name", "Unknown") depots = { did: {"name": info.get("name", "Main")} for did, info in app_data.get("depots", {}).items() if did.isdigit() } # DLCs are listed under 'extended' or 'dlc' dlcs = { str(dlc): {} for dlc in app_data.get("extended", {}).get("listofdlc", []) } return { "app_id": app_id, "title": name, "depots": depots, "dlc_ids": list(dlcs.keys()), "depot_count": len(depots), "dlc_count": len(dlcs), } except Exception as e: return {"error": str(e)} def get_app_price(app_id: str) -> dict | None: """Fetch price info from Steam Store API.""" try: url = f"https://store.steampowered.com/api/appdetails?appids={app_id}&cc=us&l=en" resp = requests.get(url, headers={"User-Agent": "Mozilla/5.0"}, timeout=10) if resp.status_code != 200: return None data = resp.json() app_data = data.get(app_id, {}) if not app_data.get("success"): return None info = app_data.get("data", {}) if info.get("is_free"): return { "name": info.get("name", ""), "price": "Free", "price_raw": 0, "discount": 0, "currency": "USD", } price_data = info.get("price_overview", {}) if price_data: final = price_data.get("final", 0) return { "name": info.get("name", ""), "price": price_data.get("final_formatted", ""), "price_raw": final / 100 if final else 0, "discount": price_data.get("discount_percent", 0), "currency": price_data.get("currency", "USD"), } return None except Exception: return None # =================================================================== # Settings helpers # =================================================================== def _settings_path() -> Path: if decky: return Path(decky.DECKY_PLUGIN_RUNTIME_DIR) / "settings.json" return Path.home() / ".config" / APP_CONFIG_DIR / "settings.json" def _legacy_runtime_path(filename: str): if decky: runtime = Path(decky.DECKY_PLUGIN_RUNTIME_DIR) return runtime.parent / LEGACY_PLUGIN_DIR / filename return Path.home() / ".config" / LEGACY_APP_CONFIG_DIR / filename def _legacy_settings_path(): runtime_legacy = _legacy_runtime_path("settings.json") if runtime_legacy.exists(): return runtime_legacy if decky and getattr(decky, "DECKY_SETTINGS_DIR", None): return Path(decky.DECKY_SETTINGS_DIR) / "gasket.json" return None def _load_settings() -> dict: p = _settings_path() if p.exists(): try: return json.loads(p.read_text()) except json.JSONDecodeError: pass legacy = _legacy_settings_path() if legacy and legacy.exists(): try: data = json.loads(legacy.read_text()) _save_settings(data) return data except json.JSONDecodeError: pass return {} def _save_settings(data: dict) -> None: p = _settings_path() p.parent.mkdir(parents=True, exist_ok=True) p.write_text(json.dumps(data, indent=2)) def _get_setting(key: str) -> any: settings = _load_settings() return settings.get(key, SETTINGS_DEFAULTS.get(key)) def _set_setting(key: str, value: any) -> None: settings = _load_settings() settings[key] = value _save_settings(settings) logger.info("Setting saved: %s=%r at %s", key, value, _settings_path()) def _add_to_sls(app_id: str, game_name: str, depots: list[str] | None = None) -> None: """Add package IDs (and optionally depot IDs for delisted games) to SLSsteam config.yaml. For listed games: adds packages to AdditionalPackages, optionally app_id to AdditionalApps. For delisted games: also adds depot IDs to AdditionalDepots. """ sls_path = _find_sls_config() if not sls_path: logger.info("SLSsteam config not found, skipping SLS add") return lines = sls_path.read_text().splitlines(keepends=True) def _insert(section: str, entry: str) -> bool: needle = entry.split(" #")[0].strip() # e.g. "- 936160" for line in lines: parts = line.strip().split() if len(parts) >= 2 and parts[0] == "-" and parts[1] == needle.split()[-1]: return False # already present try: idx = next(i for i, l in enumerate(lines) if l.strip() == section) lines.insert(idx + 1, entry) return True except StopIteration: lines.append(f"\n{section}\n{entry}") return True def _clean_name(name: str) -> str: return re.sub(r"\s+", " ", name or "").strip() # Check setting: also add app_id to AdditionalApps section? add_app_id = _get_setting("add_app_id_to_sls") added = [] packages = [] is_delisted = False resolved_name = _clean_name(game_name) last_error = None for attempt in range(1, 4): try: resp = requests.get( f"https://store.steampowered.com/api/appdetails?appids={app_id}&cc=us&l=en", timeout=15, ) resp.raise_for_status() data = resp.json() app_data = data.get(str(app_id), {}) if app_data.get("success") is False: # Game is delisted/private — no store page logger.info("SLS: app %s is delisted (store returned success=false)", app_id) is_delisted = True break info = app_data.get("data", {}) resolved_name = _clean_name(resolved_name or info.get("name", "")) packages = info.get("packages", []) or [] # Check for effectively unavailable games: # type "advertising" with no packages, or no packages + no price + not free app_type = (info.get("type") or "").lower() has_price = bool(info.get("price_overview")) is_free = info.get("is_free", False) if app_type == "advertising": logger.info("SLS: app %s is type=advertising, treating as delisted", app_id) is_delisted = True elif not packages and not has_price and not is_free: logger.info("SLS: app %s has no packages/price/free, treating as delisted", app_id) is_delisted = True packages = info.get("packages", []) or [] break except Exception as e: last_error = e logger.warning("SLS: package fetch attempt %d/3 failed: %s", attempt, e) if attempt < 3: time.sleep(attempt) if not resolved_name: details = get_app_details(app_id) resolved_name = _clean_name(details.get("title", "")) if not details.get("error") else "" if not resolved_name: resolved_name = str(app_id) # For delisted games: add depot IDs to AdditionalDepots if is_delisted and depots: logger.info("SLS: adding %d depots for delisted app %s", len(depots), app_id) for depot_id in depots: if _insert("AdditionalDepots:", f" - {depot_id} # {resolved_name}\n"): added.append(f"depot {depot_id}") # Add packages (excluding app_id itself, unless toggle is on) if packages: logger.info("SLS: app %s has packages %s", app_id, packages) for pkg in packages: pkg = str(pkg) if pkg != app_id: if _insert("AdditionalPackages:", f" - {pkg} # {resolved_name}\n"): added.append(f"package {pkg}") elif last_error and not is_delisted: logger.warning("SLS: package fetch failed after retries, skipping SLS entry: %s", last_error) elif not packages and not is_delisted: logger.info("SLS: app %s has no packages", app_id) # If toggle enabled, also add the app_id to AdditionalApps section if add_app_id: if _insert("AdditionalApps:", f" - {app_id} # {resolved_name}\n"): added.append(f"app {app_id}") if added: sls_path.write_text("".join(lines)) logger.info("SLSsteam: added %s", ", ".join(added)) else: logger.info("SLSsteam: already up to date") def _find_sls_config() -> Path | None: """Find SLSsteam config.yaml, returns path or None.""" candidates = [ Path.home() / ".config" / "SLSsteam" / "config.yaml", Path.home() / ".local/share/SLSsteam/config.yaml", Path("/home/deck/.config/SLSsteam/config.yaml"), ] for p in candidates: if p.exists(): return p return None def _get_fake_app_ids() -> dict: """Return list of app_ids that have FakeAppId entries in SLS config.""" sls_path = _find_sls_config() if not sls_path: logger.info("SLSsteam config not found for get_fake_app_ids") return {"ok": True, "ids": []} ids: list[str] = [] in_section = False for line in sls_path.read_text().splitlines(): stripped = line.strip() if stripped == "FakeAppIds:": in_section = True continue if in_section: # Section ends at next non-indented line or end of file if stripped and not stripped.startswith("#") and ":" in stripped: app_id = stripped.split(":")[0].strip() if app_id.isdigit(): ids.append(app_id) elif stripped and not stripped.startswith("#") and not stripped[0].isdigit(): break # next section logger.info("SLSsteam: found FakeAppIds %s", ids) return {"ok": True, "ids": ids} def _add_fake_app_id(app_id: str) -> dict: """Add app_id: 480 to SLSsteam config.yaml FakeAppIds section.""" sls_path = _find_sls_config() if not sls_path: logger.warning("SLSsteam config not found for add_fake_app_id") return {"ok": False, "error": "SLSsteam config not found"} logger.info("SLSsteam config found at %s", sls_path) lines = sls_path.read_text().splitlines(keepends=True) # Check if already present for line in lines: stripped = line.strip() if stripped == f"{app_id}: 480" or stripped == f"{app_id}:480": logger.info("SLSsteam: FakeAppId %s already present", app_id) return {"ok": True, "app_id": app_id, "fake_id": 480, "already": True} # Find FakeAppIds: section and insert after it fake_section = "FakeAppIds:" try: idx = next(i for i, l in enumerate(lines) if l.strip() == fake_section) lines.insert(idx + 1, f" {app_id}: 480\n") sls_path.write_text("".join(lines)) logger.info("SLSsteam: added FakeAppId %s: 480", app_id) return {"ok": True, "app_id": app_id, "fake_id": 480} except StopIteration: logger.warning("SLSsteam: FakeAppIds section not found in config") return {"ok": False, "error": "FakeAppIds section not found in config"} def _remove_fake_app_id(app_id: str) -> dict: """Remove app_id entry from SLSsteam config.yaml FakeAppIds section.""" sls_path = _find_sls_config() if not sls_path: logger.warning("SLSsteam config not found for remove_fake_app_id") return {"ok": False, "error": "SLSsteam config not found"} lines = sls_path.read_text().splitlines(keepends=True) new_lines = [] removed = False for line in lines: stripped = line.strip() if stripped == f"{app_id}: 480" or stripped == f"{app_id}:480": removed = True continue # skip this line new_lines.append(line) if removed: sls_path.write_text("".join(new_lines)) logger.info("SLSsteam: removed FakeAppId %s", app_id) return {"ok": True, "removed": app_id} else: logger.info("SLSsteam: FakeAppId %s not found, nothing to remove", app_id) return {"ok": True, "removed": None} # =================================================================== # Cr0wbar Fixes (https://github.com/Deadboy666/cr0wbar-fixes) # =================================================================== CR0WBAR_GITHUB = "https://api.github.com/repos/Deadboy666/cr0wbar-fixes" CR0WBAR_RAW = "https://raw.githubusercontent.com/Deadboy666/cr0wbar-fixes/main" _CR0WBAR_CACHE_DIR_NAME = "cr0wbar" def _cr0wbar_cache_dir() -> Path: if decky: return Path(decky.DECKY_PLUGIN_RUNTIME_DIR) / _CR0WBAR_CACHE_DIR_NAME return Path.home() / ".config" / APP_CONFIG_DIR / _CR0WBAR_CACHE_DIR_NAME def _cr0wbar_fix_list_path() -> Path: d = _cr0wbar_cache_dir() d.mkdir(parents=True, exist_ok=True) return d / "fix_list.json" def _cr0wbar_fixes_dir() -> Path: d = _cr0wbar_cache_dir() / "scripts" d.mkdir(parents=True, exist_ok=True) return d def fetch_cr0wbar_fix_list(force: bool = False) -> dict: """Fetch the list of available cr0wbar fixes from GitHub. Caches the result for 24 hours. Returns {app_ids: [...], count: N, cached: bool}. """ cache_path = _cr0wbar_fix_list_path() # Check cache (valid for 24 hours) if not force and cache_path.exists(): try: data = json.loads(cache_path.read_text()) fetched_at = data.get("fetched_at", 0) if time.time() - fetched_at < 86400: # 24h app_ids = data.get("app_ids", []) logger.info("Cr0wbar: using cached fix list (%d apps, age=%.0fh)", len(app_ids), (time.time() - fetched_at) / 3600) return {"app_ids": app_ids, "count": len(app_ids), "cached": True} except (json.JSONDecodeError, KeyError): pass # Fetch from GitHub API try: # Use git/trees API for efficient recursive listing resp = requests.get( f"{CR0WBAR_GITHUB}/git/trees/main?recursive=1", timeout=15, headers={"Accept": "application/vnd.github.v3+json", "User-Agent": "hollow-decky"}, ) if resp.status_code == 403: # Rate limited — try to use stale cache if available if cache_path.exists(): try: data = json.loads(cache_path.read_text()) return {"app_ids": data.get("app_ids", []), "count": len(data.get("app_ids", [])), "cached": True, "rate_limited": True} except Exception: pass return {"error": "GitHub API rate limited", "app_ids": [], "count": 0, "cached": False} resp.raise_for_status() tree_data = resp.json() # Extract app IDs from .sh filenames app_ids = [] for item in tree_data.get("tree", []): path = item.get("path", "") if path.endswith(".sh"): app_id = path[:-3] # Remove .sh extension if app_id.isdigit(): app_ids.append(app_id) logger.info("Cr0wbar: fetched %d fixes from GitHub", len(app_ids)) # Cache the result cache_data = {"app_ids": sorted(app_ids), "fetched_at": time.time()} cache_path.parent.mkdir(parents=True, exist_ok=True) cache_path.write_text(json.dumps(cache_data)) return {"app_ids": sorted(app_ids), "count": len(app_ids), "cached": False} except Exception as e: logger.error("Cr0wbar: fetch failed: %s", e) # Fallback to stale cache if cache_path.exists(): try: data = json.loads(cache_path.read_text()) return {"app_ids": data.get("app_ids", []), "count": len(data.get("app_ids", [])), "cached": True, "fallback": True} except Exception: pass return {"error": str(e), "app_ids": [], "count": 0, "cached": False} def check_cr0wbar_fix_available(app_id: str) -> dict: """Check if a cr0wbar fix is available for the given app_id. Returns {available: bool, app_id: str}. """ result = fetch_cr0wbar_fix_list() app_ids = result.get("app_ids", []) return {"available": app_id in app_ids, "app_id": app_id, "total_fixes": len(app_ids)} def download_cr0wbar_fix(app_id: str) -> dict: """Download a cr0wbar fix script for the given app_id. Returns {ok: bool, path: str, size: int} or {ok: bool, error: str}. """ # Check if fix is available fix_list = fetch_cr0wbar_fix_list() if app_id not in fix_list.get("app_ids", []): return {"ok": False, "error": f"No cr0wbar fix available for app {app_id}"} script_path = _cr0wbar_fixes_dir() / f"{app_id}.sh" # Check if already downloaded if script_path.exists(): logger.info("Cr0wbar: fix %s already downloaded at %s", app_id, script_path) return {"ok": True, "path": str(script_path), "size": script_path.stat().st_size, "cached": True} try: url = f"{CR0WBAR_RAW}/{app_id}.sh" logger.info("Cr0wbar: downloading %s", url) resp = requests.get(url, timeout=120) # Large files, generous timeout if resp.status_code == 404: return {"ok": False, "error": f"Fix script not found for app {app_id}"} resp.raise_for_status() script_path.write_bytes(resp.content) # Make executable script_path.chmod(0o755) size = script_path.stat().st_size logger.info("Cr0wbar: downloaded %s (%d bytes)", script_path.name, size) return {"ok": True, "path": str(script_path), "size": size, "cached": False} except Exception as e: logger.error("Cr0wbar: download failed for %s: %s", app_id, e) return {"ok": False, "error": str(e)} def install_cr0wbar_fix(app_id: str) -> dict: """Download and execute a cr0wbar fix script. Returns {ok: bool, error?: str, path?: str}. """ import subprocess # First, download dl_result = download_cr0wbar_fix(app_id) if not dl_result.get("ok"): return dl_result script_path = Path(dl_result["path"]) # Strip Steam-injected env vars that break system bash (e.g. LD_LIBRARY_PATH # pointing to Steam's incompatible libreadline, causing "undefined symbol: rl_trim_arg_from_keyseq") clean_env = { k: v for k, v in os.environ.items() if k not in ("LD_LIBRARY_PATH", "LD_PRELOAD", "STEAM_RUNTIME_PREFER_HOST") } # Preserve a sane PATH clean_env.setdefault("PATH", "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin") clean_env["HOME"] = str(Path.home()) try: logger.info("Cr0wbar: executing fix for %s at %s", app_id, script_path) result = subprocess.run( ["bash", str(script_path)], capture_output=True, text=True, timeout=300, # 5 minute timeout for fix execution cwd=str(script_path.parent), env=clean_env, ) if result.returncode != 0: stderr_preview = (result.stderr or "")[:500] logger.error("Cr0wbar: fix %s failed (rc=%d): %s", app_id, result.returncode, stderr_preview) return {"ok": False, "error": f"Fix script failed (exit code {result.returncode})", "stderr_preview": stderr_preview} logger.info("Cr0wbar: fix %s installed successfully", app_id) return {"ok": True, "path": str(script_path), "stdout_preview": (result.stdout or "")[:200]} except subprocess.TimeoutExpired: logger.error("Cr0wbar: fix %s timed out", app_id) return {"ok": False, "error": "Fix script timed out (5 minutes)"} except Exception as e: logger.error("Cr0wbar: fix %s execution error: %s", app_id, e, exc_info=True) return {"ok": False, "error": str(e)} _queue_path = Path() def _get_queue_path() -> Path: global _queue_path if _queue_path == Path(): if decky: _queue_path = Path(decky.DECKY_PLUGIN_RUNTIME_DIR) / "pending_inject.json" else: _queue_path = Path.home() / ".config" / APP_CONFIG_DIR / "pending_inject.json" return _queue_path def pop_pending_inject() -> dict | None: qp = _get_queue_path() if not qp.exists(): return None try: data = json.loads(qp.read_text()) if isinstance(data, list) and len(data) > 0: item = data.pop(0) if data: qp.write_text(json.dumps(data)) else: qp.unlink() return item # legacy single-item format if isinstance(data, dict) and "app_id" in data: qp.unlink() return data except Exception: pass return None def push_pending_inject(app_id: str) -> None: qp = _get_queue_path() qp.parent.mkdir(parents=True, exist_ok=True) items = [] if qp.exists(): try: existing = json.loads(qp.read_text()) if isinstance(existing, list): items = existing elif isinstance(existing, dict) and "app_id" in existing: items = [existing] except Exception: pass # Deduplicate if any(i.get("app_id") == app_id for i in items): logger.info("Queue: already has %s, skipping", app_id) return items.append({"app_id": app_id, "ts": time.time()}) qp.write_text(json.dumps(items)) logger.info("Queue: added %s (total: %d)", app_id, len(items)) # =================================================================== # HTTP bridge (store queue) # =================================================================== _http_server = None _http_port = 8091 async def _run_http_bridge() -> None: global _http_server, _http_port from http.server import HTTPServer, BaseHTTPRequestHandler class Handler(BaseHTTPRequestHandler): def do_GET(self): m = re.match(r"/queue/(\d+)", self.path) if m: push_pending_inject(m.group(1)) logger.info("Store queue: %s", m.group(1)) self._json(200, {"queued": m.group(1)}) return self.send_response(404) self.end_headers() def do_OPTIONS(self): self.send_response(204) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type") self.end_headers() def _json(self, code, data): self.send_response(code) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Content-Type", "application/json") self.end_headers() self.wfile.write(json.dumps(data).encode()) def log_message(self, format, *args): pass for port in range(8091, 8100): try: _http_server = HTTPServer(("0.0.0.0", port), Handler) _http_port = port logger.info("HTTP bridge bound to port %d", port) break except OSError: continue else: logger.error("HTTP bridge: no free port 8091-8099") return _http_server.timeout = 0.5 try: await asyncio.to_thread(_http_server.serve_forever) except asyncio.CancelledError: _http_server.shutdown() except Exception: pass # =================================================================== # Queue processing helpers # =================================================================== _queue_process_lock = threading.Lock() def _schedule_steam_restart() -> None: def _kill(): time.sleep(1.5) import subprocess cmds = [ ["pkill", "-TERM", "-x", "steam"], ] for cmd in cmds: try: r = subprocess.run(cmd, capture_output=True, timeout=5, text=True) logger.info("restart_steam: %s -> rc=%s stderr=%s", " ".join(cmd), r.returncode, (r.stderr or "").strip()) if r.returncode == 0: return except Exception as e: logger.error("restart_steam: %s failed: %s", " ".join(cmd), e) threading.Thread(target=_kill, daemon=True).start() logger.info("restart_steam: scheduled pkill in 1.5s") def _inject_game_sync(app_id: str) -> dict: """Inject a game into SLS config. No keys, no manifest download. For listed games: adds package IDs via AdditionalPackages. For delisted games: adds depot IDs via AdditionalDepots. Optionally adds app_id to AdditionalApps if toggle is on. """ logger.info("=== inject_game START for app %s ===", app_id) # Get game name and depot info from steamcmd.net details = get_app_details(app_id) if details.get("error"): return {"error": f"Could not fetch app details: {details['error']}", "app_id": app_id} game_name = details.get("title", str(app_id)) depots = details.get("depots", {}) # Check if delisted is_delisted = _check_delisted(app_id) # For delisted games, pass depot IDs; for listed games, let _add_to_sls get packages depot_ids = list(depots.keys()) if (is_delisted and depots) else None _add_to_sls(app_id, game_name, depots=depot_ids) logger.info("=== inject_game DONE for app %s (delisted=%s) ===", app_id, is_delisted) result = {"app_id": app_id, "name": game_name, "injected": True, "delisted": is_delisted} if depot_ids: result["depots"] = depot_ids return result # =================================================================== # Decky Plugin Class # =================================================================== class Plugin: # ---- Lifecycle ------------------------------------------------- async def _main(self) -> None: info = "decky" if decky else "offline" logger.info("Hollow-decky plugin loaded (%s mode)", info) # Start minimal HTTP bridge for store button + key setup try: import threading self._bridge_thread = threading.Thread( target=lambda: asyncio.run(_run_http_bridge()), daemon=True ) self._bridge_thread.start() logger.info("HTTP bridge thread started") except Exception as e: logger.warning("HTTP bridge failed: %s", e) async def _unload(self) -> None: global _http_server if _http_server: _http_server.shutdown() _http_server = None logger.info("Hollow-decky plugin unloaded") async def get_http_port(self) -> int: global _http_port return _http_port async def discover_store_tab(self) -> dict | None: """Find the Steam Store tab via CDP (used by frontend StorePatch).""" for port in range(8080, 8083): try: resp = requests.get(f"http://localhost:{port}/json", timeout=5) tabs = resp.json() for t in tabs: url = t.get("url", "") if "store.steampowered.com/app/" in url: return { "webSocketDebuggerUrl": t.get("webSocketDebuggerUrl"), "url": url, } except Exception: continue return None async def _uninstall(self) -> None: # Clean up settings _settings_path().unlink(missing_ok=True) logger.info("Hollow-decky plugin uninstalled (config cleaned)") # ---- Game Search ----------------------------------------------- async def search_games(self, query: str, limit: int = 20, appid: bool = False) -> dict: """Search for games by name or exact App ID. Uses Steam Store + delisted cache + steamcmd.net — no API key required.""" try: return search_games(query, limit=limit, appid=appid) except Exception as e: logger.error("search_games error: %s", e, exc_info=True) return {"error": str(e), "results": []} async def refresh_delisted(self, force: bool = False) -> dict: """Refresh the delisted games cache from steam-tracker.com.""" return refresh_delisted_cache(force=force) # ---- App Details ----------------------------------------------- async def get_details(self, app_id: str) -> dict: """Fetch depot/DLC info for an app_id.""" return get_app_details(app_id) # ---- Price ----------------------------------------------------- async def get_price(self, app_id: str) -> dict | None: """Fetch store price for an app_id.""" return get_app_price(app_id) # ---- Inject Game (core feature) -------------------------------- async def inject_game(self, app_id: str) -> dict: """Inject a game into SLS config. No keys, no manifest download.""" return _inject_game_sync(app_id) async def add_fake_app_id(self, app_id: str) -> dict: """Add FakeAppId entry to SLSsteam config.yaml.""" try: result = _add_fake_app_id(app_id) logger.info("add_fake_app_id result: %s", result) return result except Exception as e: logger.error("add_fake_app_id error: %s", e, exc_info=True) return {"ok": False, "error": str(e)} async def remove_fake_app_id(self, app_id: str) -> dict: """Remove FakeAppId entry from SLSsteam config.yaml.""" try: result = _remove_fake_app_id(app_id) logger.info("remove_fake_app_id result: %s", result) return result except Exception as e: logger.error("remove_fake_app_id error: %s", e, exc_info=True) return {"ok": False, "error": str(e)} async def get_fake_app_ids(self) -> dict: """Return list of app_ids with FakeAppId entries in SLS config.""" try: return _get_fake_app_ids() except Exception as e: logger.error("get_fake_app_ids error: %s", e, exc_info=True) return {"ok": True, "ids": []} async def process_queue(self) -> dict: """Process queued apps sequentially, then restart Steam once at the end.""" if not _queue_process_lock.acquire(blocking=False): logger.info("process_queue: already running") return {"busy": True, "processed": 0, "injected": 0, "errors": 0} processed = 0 injected = 0 errors = 0 try: while True: item = pop_pending_inject() if not item or not item.get("app_id"): break app_id = str(item["app_id"]) processed += 1 result = _inject_game_sync(app_id) if result.get("error"): errors += 1 logger.error("process_queue: app %s failed: %s", app_id, result.get("error")) if result.get("injected"): injected += 1 logger.info("process_queue: %d processed, %d injected, %d errors", processed, injected, errors) return {"processed": processed, "injected": injected, "errors": errors} finally: _queue_process_lock.release() # ---- Backup / Restore ------------------------------------------ async def list_backups(self) -> list[dict]: """List all available SLS config backups, newest first.""" return list_sls_backups() async def create_backup(self) -> str: """Manually create a backup. Returns the backup filename.""" return backup_sls() async def restore_backup(self, backup_name: str) -> dict: """Restore SLS config from a backup. Creates a safety backup of current first.""" try: result = restore_sls(backup_name) if "error" in result: return result return {"success": True, **result} except Exception as e: return {"error": str(e)} # ---- Settings -------------------------------------------------- async def get_setting(self, key: str, default: any = None) -> any: return _get_setting(key) if default is None else _load_settings().get(key, default) async def set_setting(self, key: str, value: any) -> None: _set_setting(key, value) async def get_pending(self) -> dict | None: """Return and clear any queued injection from the store button.""" result = pop_pending_inject() if result: logger.info("get_pending: found %s", result.get("app_id")) return result async def queue_inject(self, app_id: str) -> dict: """Queue an app_id from the store button (called via CDP console.log).""" push_pending_inject(app_id) logger.info("Store queue (CDP): %s", app_id) return {"queued": app_id} async def peek_queue(self) -> dict: """Check current queue state without popping.""" qp = _get_queue_path() if not qp.exists(): return {"items": []} try: data = json.loads(qp.read_text()) if isinstance(data, list): return {"items": [i.get("app_id") for i in data]} if isinstance(data, dict) and "app_id" in data: return {"items": [data["app_id"]]} except Exception: pass return {"items": []} async def clear_queue(self) -> dict: """Delete the pending injection queue.""" qp = _get_queue_path() if qp.exists(): qp.unlink() return {"ok": True} async def get_deck_ip(self) -> str: import socket s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) try: s.connect(("8.8.8.8", 80)) ip = s.getsockname()[0] except Exception: ip = "127.0.0.1" finally: s.close() return ip async def restart_steam(self) -> dict: """Kill Steam so it restarts and reloads library. Runs in background.""" _schedule_steam_restart() return {"ok": True} async def get_all_settings(self) -> dict: loaded = _load_settings() # Merge defaults for any missing keys return {**SETTINGS_DEFAULTS, **loaded} # ---- Cr0wbar Fixes ----------------------------------------------- async def fetch_cr0wbar_fix_list(self, force: bool = False) -> dict: """Fetch/update the list of available cr0wbar fixes from GitHub.""" try: return fetch_cr0wbar_fix_list(force=force) except Exception as e: logger.error("fetch_cr0wbar_fix_list error: %s", e, exc_info=True) return {"error": str(e), "app_ids": [], "count": 0, "cached": False} async def check_cr0wbar_fix(self, app_id: str) -> dict: """Check if a cr0wbar fix is available for the given app_id.""" try: return check_cr0wbar_fix_available(app_id) except Exception as e: logger.error("check_cr0wbar_fix error: %s", e, exc_info=True) return {"available": False, "app_id": app_id, "error": str(e)} async def install_cr0wbar_fix(self, app_id: str) -> dict: """Download and execute a cr0wbar fix script for the given app_id.""" try: return install_cr0wbar_fix(app_id) except Exception as e: logger.error("install_cr0wbar_fix error: %s", e, exc_info=True) return {"ok": False, "error": str(e)} async def download_cr0wbar_fix(self, app_id: str) -> dict: """Download a cr0wbar fix script without executing it.""" try: return download_cr0wbar_fix(app_id) except Exception as e: logger.error("download_cr0wbar_fix error: %s", e, exc_info=True) return {"ok": False, "error": str(e)}