diff options
Diffstat (limited to 'py_modules/lsfg_vk/steam_service.py')
| -rw-r--r-- | py_modules/lsfg_vk/steam_service.py | 290 |
1 files changed, 290 insertions, 0 deletions
diff --git a/py_modules/lsfg_vk/steam_service.py b/py_modules/lsfg_vk/steam_service.py new file mode 100644 index 0000000..93f2593 --- /dev/null +++ b/py_modules/lsfg_vk/steam_service.py @@ -0,0 +1,290 @@ +import re +from pathlib import Path +from typing import Dict, Optional, Tuple + +from .base_service import BaseService +from .constants import ( + STEAM_LOSSLESS_SCALING_APP_ID, + STEAM_LOSSLESS_SCALING_BRANCH, +) + + +class SteamService(BaseService): + DEFAULT_BRANCH = "public" + MANIFEST_FILENAME = f"appmanifest_{STEAM_LOSSLESS_SCALING_APP_ID}.acf" + # Valve compatibility tools, runtimes, Steamworks redistributables, and LSFG. + GAME_SELECTOR_EXCLUDED_APPIDS = { + "858280", # Proton 3.7 + "961940", # Proton 3.16 + "1054830", # Proton 4.2 + "1113280", # Proton 4.11 + "1245040", # Proton 5.0 + "1420170", # Proton 5.13 + "1493710", # Proton Experimental + "1580130", # Proton 6.3 + "1887720", # Proton 7 + "2180100", # Proton Hotfix + "228980", # Steamworks Common Redistributables + "2348590", # Proton 8 + "2805730", # Proton 9 + "3029110", # Lepton + "3127680", # fex + "3658110", # Proton 10 + "4183110", # Steam Linux Runtime 4.0 + "4185400", # Steam Linux Runtime 4.0 for arm64 + "4427310", # Proton Experimental (ARM64) + "4628710", # Proton 11 / Proton Next + "4628740", # Proton 11 (ARM64) + "4690330", # Legacy Steam Runtime + "993090", # Lossless Scaling + "1070560", # Steam Linux Runtime 1.0 + "1391110", # Steam Linux Runtime 2.0 + "1628350", # Steam Linux Runtime 3.0 + } + + def _steam_roots(self): + seen = set() + for candidate in ( + self.user_home / ".local/share/Steam", + self.user_home / ".steam/steam", + self.user_home / ".steam/root", + self.user_home / ".var/app/com.valvesoftware.Steam/.local/share/Steam", + ): + yield from self._unique_existing_root(candidate, seen) + + def _steam_library_roots(self): + seen = set() + for root in self._steam_roots(): + yield from self._unique_existing_root(root, seen) + for library_file in ( + root / "steamapps/libraryfolders.vdf", + root / "config/libraryfolders.vdf", + ): + try: + content = library_file.read_text(encoding="utf-8") + except OSError: + continue + for raw_path in re.findall(r'(?m)^\s*"path"\s+"((?:\\.|[^"])*)"', content): + path = raw_path.replace(r'\"', '"').replace(r'\\', '\\') + yield from self._unique_existing_root(Path(path), seen) + + @staticmethod + def _unique_existing_root(path: Path, seen: set[str]): + if not path.exists(): + return + try: + resolved = str(path.resolve()) + except OSError: + resolved = str(path) + if resolved not in seen: + seen.add(resolved) + yield path + + @staticmethod + def _read_shortcuts(data: bytes) -> Dict[str, object]: + def read_string(offset: int) -> Tuple[str, int]: + end = data.index(b"\0", offset) + return data[offset:end].decode("utf-8", errors="replace"), end + 1 + + def read_object(offset: int = 0) -> Tuple[Dict[str, object], int]: + values = {} + while offset < len(data): + value_type, offset = data[offset], offset + 1 + if value_type == 8: + return values, offset + key, offset = read_string(offset) + if value_type == 0: + value, offset = read_object(offset) + elif value_type == 1: + value, offset = read_string(offset) + elif value_type in (2, 7): + width = 4 if value_type == 2 else 8 + if offset + width > len(data): + raise ValueError("truncated binary VDF integer") + value = int.from_bytes(data[offset:offset + width], "little", signed=True) + offset += width + else: + raise ValueError(f"unsupported binary VDF type {value_type}") + values[key] = value + raise ValueError("unterminated binary VDF object") + + values, offset = read_object() + if offset != len(data): + raise ValueError("trailing binary VDF data") + return values + + @staticmethod + def _shortcut_game(shortcut: object) -> Optional[Dict[str, object]]: + if not isinstance(shortcut, dict): + return None + appid = shortcut.get("appid") + name = shortcut.get("AppName") or shortcut.get("appname") + if not isinstance(appid, int) or appid == 0 or not isinstance(name, str) or not name: + return None + game = { + "appid": str(appid & 0xFFFFFFFF), + "name": name, + "nonSteam": True, + } + executable = shortcut.get("Exe") or shortcut.get("exe") + if isinstance(executable, str) and executable.strip().strip('"') in {"flatpak", "/usr/bin/flatpak"}: + game["isFlatpakShortcut"] = True + return game + + def _shortcut_games(self): + games = {} + for root in self._steam_roots(): + for path in sorted((root / "userdata").glob("*/config/shortcuts.vdf")): + try: + shortcuts = self._read_shortcuts(path.read_bytes()).get("shortcuts", {}) + except (OSError, ValueError): + continue + if not isinstance(shortcuts, dict): + continue + for shortcut in shortcuts.values(): + game = self._shortcut_game(shortcut) + if game and game["appid"] not in self.GAME_SELECTOR_EXCLUDED_APPIDS: + games.setdefault(game["appid"], game) + return list(games.values()) + + def _manifest_path(self) -> Optional[Path]: + return next(( + path + for root in self._steam_library_roots() + if (path := root / "steamapps" / self.MANIFEST_FILENAME).is_file() + ), None) + + @staticmethod + def _section_bounds(content: str, section_name: str) -> Optional[Tuple[int, int, str]]: + section = re.search( + rf'(?m)^(?P<indent>[ \t]*)"{re.escape(section_name)}"[ \t\r\n]*\{{', + content, + ) + if section is None: + return None + depth = 1 + in_string = False + escaped = False + for index in range(section.end(), len(content)): + character = content[index] + if in_string: + if escaped: + escaped = False + elif character == "\\": + escaped = True + elif character == '"': + in_string = False + elif character == '"': + in_string = True + elif character == "{": + depth += 1 + elif character == "}": + depth -= 1 + if depth == 0: + return section.end(), index, section.group("indent") + return None + + @classmethod + def _section_value(cls, content: str, section_name: str, key: str) -> Optional[str]: + bounds = cls._section_bounds(content, section_name) + if bounds is None: + return None + start, end, _ = bounds + pattern = re.compile(r'(?m)^[ \t]*"(?P<key>[^"]+)"[ \t]+"(?P<value>(?:\\.|[^"\\])*)"') + return next(( + match.group("value") + for match in pattern.finditer(content, start, end) + if match.group("key") == key + ), None) + + @classmethod + def _branch_or_default(cls, branch: Optional[str]) -> str: + return branch or cls.DEFAULT_BRANCH + + def _status_fields(self, manifest_path: Path, content: str) -> Dict[str, object]: + selected = self._branch_or_default(self._section_value(content, "UserConfig", "BetaKey")) + current = self._branch_or_default( + self._section_value(content, "MountedConfig", "BetaKey") + or self._section_value(content, "UserConfig", "BetaKey") + ) + needs_switch = selected != STEAM_LOSSLESS_SCALING_BRANCH or current != STEAM_LOSSLESS_SCALING_BRANCH + return { + "installed": True, + "manifest_path": str(manifest_path), + "selected_branch": selected, + "current_branch": current, + "target_branch": STEAM_LOSSLESS_SCALING_BRANCH, + "needs_switch": needs_switch, + "restart_required": selected == STEAM_LOSSLESS_SCALING_BRANCH and current != STEAM_LOSSLESS_SCALING_BRANCH, + } + + @staticmethod + def _missing_branch_fields() -> Dict[str, object]: + return { + "installed": False, + "manifest_path": None, + "selected_branch": None, + "current_branch": None, + "target_branch": STEAM_LOSSLESS_SCALING_BRANCH, + "needs_switch": False, + "restart_required": False, + } + + def find_lsfg_vk_dll(self) -> Optional[str]: + if self.get_branch_status().get("needs_switch"): + return None + return next(( + str(path) + for root in self._steam_library_roots() + if (path := root / "steamapps/common/Lossless Scaling/lsfg-vk.dll").is_file() + ), None) + + def get_branch_status(self) -> Dict[str, object]: + try: + manifest = self._manifest_path() + if manifest is None: + return self._success_response( + dict, + "Lossless Scaling is not installed through Steam", + **self._missing_branch_fields(), + ) + fields = self._status_fields(manifest, manifest.read_text(encoding="utf-8")) + message = ( + "Lossless Scaling is using the lsfg-vk Steam branch" + if not fields["needs_switch"] + else "lsfg-vk is selected; restart Steam to finish the branch switch" + if fields["restart_required"] + else "Select lsfg-vk in Lossless Scaling's Steam Properties > Betas" + ) + return self._success_response(dict, message, **fields) + except Exception as error: + return self._error_response(dict, str(error), **self._missing_branch_fields()) + + def get_installed_games(self) -> Dict[str, object]: + try: + games: Dict[str, Dict[str, object]] = {} + for root in self._steam_library_roots(): + for manifest in (root / "steamapps").glob("appmanifest_*.acf"): + match = re.fullmatch(r"appmanifest_(\d+)\.acf", manifest.name) + if not match: + continue + try: + content = manifest.read_text(encoding="utf-8") + except OSError: + continue + appid = match.group(1) + if appid in self.GAME_SELECTOR_EXCLUDED_APPIDS: + continue + games[appid] = { + "appid": appid, + "name": self._section_value(content, "AppState", "name") or f"App {appid}", + "nonSteam": False, + } + for game in self._shortcut_games(): + games.setdefault(str(game["appid"]), game) + return self._success_response( + dict, + games=sorted(games.values(), key=lambda game: str(game["name"]).lower()), + ) + except Exception as error: + return self._error_response(dict, str(error), games=[]) |