refactor: simplify migration implementation

1 files changed, 102 insertions, 183 deletions

diff --git a/py_modules/lsfg_vk/steam_service.py b/py_modules/lsfg_vk/steam_service.py
index 50d722b..201441e 100644
--- a/py_modules/lsfg_vk/steam_service.py
+++ b/py_modules/lsfg_vk/steam_service.py
@@ -10,7 +10,6 @@ from .constants import (
     WRAPPER_FILENAME,
 )
 
-
 _FLATPAK_APP_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9-]*(?:\.[A-Za-z0-9][A-Za-z0-9-]*)+$")
 _WRAPPER_TOKEN = f"~/{WRAPPER_FILENAME}"
 
@@ -25,107 +24,89 @@ def _split_command(value: Optional[str]) -> Optional[list[str]]:
 
 
 def _is_managed_wrapper(value: str) -> bool:
-    """Recognize the wrapper Target while keeping arbitrary launchers as host games."""
     if value in {_WRAPPER_TOKEN, f"$HOME/{WRAPPER_FILENAME}"}:
         return True
     path = Path(value)
     return path.is_absolute() and path.name == WRAPPER_FILENAME
 
 
-def classify_shortcut_transport(
-    executable: Optional[str],
-    launch_options: Optional[str] = None,
-) -> Dict[str, object]:
-    """Classify direct Flatpak invocations, including the managed wrapper Target."""
+def classify_shortcut_transport(executable: Optional[str], launch_options: Optional[str] = None) -> Dict[str, object]:
     executable_tokens = _split_command(executable)
     option_tokens = _split_command(launch_options)
     if executable_tokens is None or option_tokens is None or not executable_tokens:
         return {"kind": "host"}
-
     direct_flatpak = executable_tokens[0] == "/usr/bin/flatpak"
     managed_wrapper = len(executable_tokens) == 1 and _is_managed_wrapper(executable_tokens[0])
     if not direct_flatpak and not managed_wrapper:
         return {"kind": "host"}
-
     arguments = [*executable_tokens[1:], *option_tokens]
     if not arguments or arguments[0] != "run":
         return {"kind": "host"}
-
     for argument in arguments[1:]:
-        if argument == "--":
+        if argument == "--" or argument.startswith("-"):
             continue
-        if argument.startswith("-"):
-            continue
-        if _FLATPAK_APP_ID.fullmatch(argument):
-            return {"kind": "flatpak", "flatpakAppId": argument}
-        return {"kind": "host"}
+        return (
+            {"kind": "flatpak", "flatpakAppId": argument}
+            if _FLATPAK_APP_ID.fullmatch(argument)
+            else {"kind": "host"}
+        )
     return {"kind": "host"}
 
 
+def _first_string(values: Dict[str, object], *keys: str) -> Optional[str]:
+    return next((values[key] for key in keys if isinstance(values.get(key), str)), None)
+
+
 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
+        "858280", "961940", "1054830", "1113280", "1245040", "1420170",
+        "1493710", "1580130", "1887720", "2180100", "228980", "2348590",
+        "2805730", "3029110", "3127680", "3658110", "4183110", "4185400",
+        "4427310", "4628710", "4628740", "4690330", "993090", "1070560",
+        "1391110", "1628350",
     }
 
     def _steam_roots(self):
-        candidates = (
+        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",
-        )
-        seen = set()
-
-        for candidate in candidates:
+        ):
             yield from self._unique_existing_root(candidate, seen)
 
     def _steam_library_roots(self):
         seen = set()
-        for candidate in self._steam_roots():
-            yield from self._unique_existing_root(candidate, seen)
-
+        for root in self._steam_roots():
+            yield from self._unique_existing_root(root, seen)
             for library_file in (
-                candidate / "steamapps/libraryfolders.vdf",
-                candidate / "config/libraryfolders.vdf",
+                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)
@@ -142,16 +123,12 @@ class SteamService(BaseService):
                     value, offset = read_object(offset)
                 elif value_type == 1:
                     value, offset = read_string(offset)
-                elif value_type == 2:
-                    if offset + 4 > len(data):
+                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 + 4], "little", signed=True)
-                    offset += 4
-                elif value_type == 7:
-                    if offset + 8 > len(data):
-                        raise ValueError("truncated binary VDF 64-bit integer")
-                    value = int.from_bytes(data[offset:offset + 8], "little", signed=True)
-                    offset += 8
+                    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
@@ -170,53 +147,28 @@ class SteamService(BaseService):
         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
-        executable = next(
-            (
-                shortcut.get(key)
-                for key in ("Exe", "exe", "executable")
-                if isinstance(shortcut.get(key), str)
-            ),
-            None,
-        )
-        launch_options = next(
-            (
-                shortcut.get(key)
-                for key in ("LaunchOptions", "launchoptions", "launch_options", "arguments")
-                if isinstance(shortcut.get(key), str)
-            ),
-            None,
-        )
-        start_dir = next(
-            (
-                shortcut.get(key)
-                for key in ("StartDir", "startdir", "start_dir")
-                if isinstance(shortcut.get(key), str)
-            ),
-            None,
-        )
+        executable = _first_string(shortcut, "Exe", "exe", "executable")
+        arguments = _first_string(shortcut, "LaunchOptions", "launchoptions", "launch_options", "arguments")
+        start_dir = _first_string(shortcut, "StartDir", "startdir", "start_dir")
         game: Dict[str, object] = {
-            "appid": str(appid & 0xffffffff),
+            "appid": str(appid & 0xFFFFFFFF),
             "name": name,
             "nonSteam": True,
-            "transport": classify_shortcut_transport(executable, launch_options),
+            "transport": classify_shortcut_transport(executable, arguments),
         }
-        if executable is not None:
-            game["executable"] = executable
-        if launch_options is not None:
-            game["arguments"] = launch_options
-        if start_dir is not None:
-            game["startDir"] = start_dir
+        for key, value in (("executable", executable), ("arguments", arguments), ("startDir", start_dir)):
+            if value is not None:
+                game[key] = value
         return game
 
     def _shortcut_games(self):
         games = {}
-        for steam_root in self._steam_roots():
-            for shortcuts_file in sorted((steam_root / "userdata").glob("*/config/shortcuts.vdf")):
+        for root in self._steam_roots():
+            for path in sorted((root / "userdata").glob("*/config/shortcuts.vdf")):
                 try:
-                    root = self._read_shortcuts(shortcuts_file.read_bytes())
+                    shortcuts = self._read_shortcuts(path.read_bytes()).get("shortcuts", {})
                 except (OSError, ValueError):
                     continue
-                shortcuts = root.get("shortcuts", {})
                 if not isinstance(shortcuts, dict):
                     continue
                 for shortcut in shortcuts.values():
@@ -225,25 +177,12 @@ class SteamService(BaseService):
                         games.setdefault(game["appid"], game)
         return list(games.values())
 
-    @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
+        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]]:
@@ -253,7 +192,6 @@ class SteamService(BaseService):
         )
         if section is None:
             return None
-
         depth = 1
         in_string = False
         escaped = False
@@ -266,9 +204,7 @@ class SteamService(BaseService):
                     escaped = True
                 elif character == '"':
                     in_string = False
-                continue
-
-            if character == '"':
+            elif character == '"':
                 in_string = True
             elif character == "{":
                 depth += 1
@@ -283,98 +219,82 @@ class SteamService(BaseService):
         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<key>[^"]+)"[ \t]+"(?P<value>(?:\\.|[^"\\])*)"'
-        )
-        for match in pattern.finditer(content, body_start, body_end):
-            if match.group("key") == key:
-                return match.group("value")
-        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_branch = self._branch_or_default(
-            self._section_value(content, "UserConfig", "BetaKey")
-        )
-        current_branch = self._branch_or_default(
+        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_branch != STEAM_LOSSLESS_SCALING_BRANCH
-            or current_branch != STEAM_LOSSLESS_SCALING_BRANCH
-        )
+        needs_switch = selected != STEAM_LOSSLESS_SCALING_BRANCH or current != STEAM_LOSSLESS_SCALING_BRANCH
         return {
             "installed": True,
             "manifest_path": str(manifest_path),
-            "selected_branch": selected_branch,
-            "current_branch": current_branch,
+            "selected_branch": selected,
+            "current_branch": current,
             "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
-            ),
+            "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]:
-        """Find the branch-specific upstream DLL in any Steam library."""
         if self.get_branch_status().get("needs_switch"):
             return None
-        for library_root in self._steam_library_roots():
-            dll_path = library_root / "steamapps/common/Lossless Scaling/lsfg-vk.dll"
-            if dll_path.is_file():
-                return str(dll_path)
-        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_path = self._manifest_path()
-            if manifest_path is None:
+            manifest = self._manifest_path()
+            if manifest 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,
+                    **self._missing_branch_fields(),
                 )
-
-            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 = "Select lsfg-vk in Lossless Scaling's Steam Properties > Betas"
+            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),
-                installed=False,
-                manifest_path=None,
-                selected_branch=None,
-                current_branch=None,
-                target_branch=STEAM_LOSSLESS_SCALING_BRANCH,
-                needs_switch=False,
-                restart_required=False,
-            )
+            return self._error_response(dict, str(error), **self._missing_branch_fields())
 
     def get_installed_games(self) -> Dict[str, object]:
-        """Return installed Steam app IDs and names for the Game Mode selector."""
         try:
             games: Dict[str, Dict[str, object]] = {}
-            for library_root in self._steam_library_roots():
-                for manifest in (library_root / "steamapps").glob("appmanifest_*.acf"):
+            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
@@ -385,10 +305,9 @@ class SteamService(BaseService):
                     appid = match.group(1)
                     if appid in self.GAME_SELECTOR_EXCLUDED_APPIDS:
                         continue
-                    name = self._section_value(content, "AppState", "name") or f"App {appid}"
                     games[appid] = {
                         "appid": appid,
-                        "name": name,
+                        "name": self._section_value(content, "AppState", "name") or f"App {appid}",
                         "nonSteam": False,
                         "transport": {"kind": "host"},
                     }