From 89e64a7dbddb8f713dd2ca37131924e8bd8ab51f Mon Sep 17 00:00:00 2001 From: xXJSONDeruloXx Date: Sat, 5 Sep 2026 23:58:16 -0400 Subject: feat: select Lossless Scaling lsfg-vk branch --- py_modules/lsfg_vk/steam_service.py | 273 ++++++++++++++++++++++++++++++++++++ 1 file changed, 273 insertions(+) create mode 100644 py_modules/lsfg_vk/steam_service.py (limited to 'py_modules/lsfg_vk/steam_service.py') diff --git a/py_modules/lsfg_vk/steam_service.py b/py_modules/lsfg_vk/steam_service.py new file mode 100644 index 0000000..3278f3c --- /dev/null +++ b/py_modules/lsfg_vk/steam_service.py @@ -0,0 +1,273 @@ +import os +import re +import tempfile +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" + + def _steam_library_roots(self): + candidates = ( + 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", + ) + seen = set() + + for candidate in candidates: + yield from self._unique_existing_root(candidate, seen) + + library_file = candidate / "steamapps/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 in seen: + return + seen.add(resolved) + yield path + + def _manifest_path(self) -> Optional[Path]: + for library_root in self._steam_library_roots(): + manifest = library_root / "steamapps" / self.MANIFEST_FILENAME + if manifest.is_file(): + return manifest + return None + + @staticmethod + def _section_bounds(content: str, section_name: str) -> Optional[Tuple[int, int, str]]: + section = re.search( + rf'(?m)^(?P[ \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 + continue + + if 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 + body_start, body_end, _ = bounds + pattern = re.compile( + r'(?m)^[ \t]*"(?P[^"]+)"[ \t]+"(?P(?:\\.|[^"\\])*)"' + ) + for match in pattern.finditer(content, body_start, body_end): + if match.group("key") == key: + return match.group("value") + return None + + @classmethod + def _set_section_value(cls, content: str, section_name: str, key: str, value: str) -> str: + bounds = cls._section_bounds(content, section_name) + if bounds is None: + if section_name != "UserConfig": + raise ValueError(f"Steam manifest is missing the {section_name} section") + app_state = cls._section_bounds(content, "AppState") + if app_state is None: + raise ValueError("Steam manifest is missing the AppState section") + _, app_state_end, app_state_indent = app_state + prefix = content[:app_state_end] + if not prefix.endswith(("\n", "\r")): + prefix += "\n" + entry_indent = app_state_indent + "\t" + section = ( + f'{entry_indent}"UserConfig"\n' + f'{entry_indent}{{\n' + f'{entry_indent}\t"{key}"\t"{value}"\n' + f'{entry_indent}}}\n' + ) + return prefix + section + content[app_state_end:] + + body_start, body_end, section_indent = bounds + pattern = re.compile( + rf'(?m)^[ \t]*"{re.escape(key)}"[ \t]+"(?P(?:\\.|[^"\\])*)"' + ) + match = pattern.search(content, body_start, body_end) + if match is not None: + return content[: match.start("value")] + value + content[match.end("value") :] + + prefix = content[:body_end] + if not prefix.endswith(("\n", "\r")): + prefix += "\n" + entry_indent = section_indent + "\t" + return prefix + f'{entry_indent}"{key}"\t"{value}"\n' + content[body_end:] + + @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_branch = self._branch_or_default( + self._section_value(content, "UserConfig", "BetaKey") + ) + current_branch = self._branch_or_default( + self._section_value(content, "MountedConfig", "BetaKey") + or self._section_value(content, "UserConfig", "BetaKey") + ) + needs_switch = ( + selected_branch != STEAM_LOSSLESS_SCALING_BRANCH + or current_branch != STEAM_LOSSLESS_SCALING_BRANCH + ) + return { + "installed": True, + "manifest_path": str(manifest_path), + "selected_branch": selected_branch, + "current_branch": current_branch, + "target_branch": STEAM_LOSSLESS_SCALING_BRANCH, + "needs_switch": needs_switch, + "restart_required": ( + selected_branch == STEAM_LOSSLESS_SCALING_BRANCH + and current_branch != STEAM_LOSSLESS_SCALING_BRANCH + ), + } + + def get_branch_status(self) -> Dict[str, object]: + try: + manifest_path = self._manifest_path() + if manifest_path is None: + return self._success_response( + dict, + "Lossless Scaling is not installed through Steam", + installed=False, + manifest_path=None, + selected_branch=None, + current_branch=None, + target_branch=STEAM_LOSSLESS_SCALING_BRANCH, + needs_switch=False, + restart_required=False, + ) + + content = manifest_path.read_text(encoding="utf-8") + fields = self._status_fields(manifest_path, content) + if not fields["needs_switch"]: + message = "Lossless Scaling is using the lsfg-vk Steam branch" + elif fields["restart_required"]: + message = "lsfg-vk is selected; restart Steam to finish the branch switch" + else: + message = "Lossless Scaling is not using the lsfg-vk Steam branch" + return self._success_response(dict, message, **fields) + except Exception as error: + return self._error_response( + dict, + str(error), + installed=False, + manifest_path=None, + selected_branch=None, + current_branch=None, + target_branch=STEAM_LOSSLESS_SCALING_BRANCH, + needs_switch=False, + restart_required=False, + ) + + def select_branch(self) -> Dict[str, object]: + try: + manifest_path = self._manifest_path() + if manifest_path is None: + raise FileNotFoundError("Lossless Scaling is not installed through Steam") + + content = manifest_path.read_text(encoding="utf-8") + fields = self._status_fields(manifest_path, content) + if not fields["needs_switch"]: + return self._success_response( + dict, + "Lossless Scaling is already using the lsfg-vk Steam branch", + changed=False, + **fields, + ) + + updated = self._set_section_value( + content, + "UserConfig", + "BetaKey", + STEAM_LOSSLESS_SCALING_BRANCH, + ) + if updated != content: + file_mode = manifest_path.stat().st_mode & 0o777 + temporary_path = None + try: + with tempfile.NamedTemporaryFile( + mode="w", + encoding="utf-8", + dir=manifest_path.parent, + prefix=f".{manifest_path.name}.", + delete=False, + ) as temporary_file: + temporary_path = Path(temporary_file.name) + temporary_file.write(updated) + temporary_file.flush() + os.fsync(temporary_file.fileno()) + temporary_path.chmod(file_mode) + os.replace(temporary_path, manifest_path) + except Exception: + if temporary_path is not None: + temporary_path.unlink(missing_ok=True) + raise + + new_fields = dict(fields) + new_fields["selected_branch"] = STEAM_LOSSLESS_SCALING_BRANCH + new_fields["needs_switch"] = new_fields["current_branch"] != STEAM_LOSSLESS_SCALING_BRANCH + new_fields["restart_required"] = new_fields["needs_switch"] + return self._success_response( + dict, + "lsfg-vk selected for Lossless Scaling; restart Steam to download it", + changed=updated != content, + **new_fields, + ) + except Exception as error: + return self._error_response( + dict, + str(error), + changed=False, + installed=False, + manifest_path=None, + selected_branch=None, + current_branch=None, + target_branch=STEAM_LOSSLESS_SCALING_BRANCH, + needs_switch=False, + restart_required=False, + ) -- cgit v1.2.3