refactor(context_menu): clean code

Signed-off-by: Boris Yumankulov <boria138@altlinux.org>

portprotonqt/context_menu_manager.py

@@ -5,16 +5,14 @@ import shutil
 import subprocess
 import threading
 import logging
-import re
 import orjson
-import vdf
 from PySide6.QtWidgets import QMessageBox, QDialog, QMenu, QFileDialog
 from PySide6.QtCore import QUrl, QPoint, QObject, Signal, Qt
 from PySide6.QtGui import QDesktopServices
 from portprotonqt.localization import _
 from portprotonqt.config_utils import parse_desktop_entry, read_favorites, save_favorites
-from portprotonqt.steam_api import is_game_in_steam, add_to_steam, remove_from_steam, get_steam_home, get_last_steam_user, convert_steam_id
+from portprotonqt.steam_api import is_game_in_steam, add_to_steam, remove_from_steam
-from portprotonqt.egs_api import add_egs_to_steam, get_egs_executable
+from portprotonqt.egs_api import add_egs_to_steam, get_egs_executable, remove_egs_from_steam
 from portprotonqt.dialogs import AddGameDialog, generate_thumbnail
 
 logger = logging.getLogger(__name__)
@@ -893,124 +891,7 @@ Icon={icon_path}
                 )
 
         if game_source == "epic":
-            # For EGS games, construct the script path used in Steam shortcuts.vdf
+            remove_egs_from_steam(game_name, self.portproton_location, on_remove_from_steam_result)
-            if not self.portproton_location:
-                self.signals.show_warning_dialog.emit(
-                    _("Error"),
-                    _("PortProton directory not found")
-                )
-                return
-            steam_scripts_dir = os.path.join(self.portproton_location, "steam_scripts")
-            safe_game_name = re.sub(r'[<>:"/\\|?*]', '_', game_name.strip())
-            script_path = os.path.join(steam_scripts_dir, f"{safe_game_name}_egs.sh")
-            quoted_script_path = f'"{script_path}"'
-
-            # Directly remove the shortcut by matching AppName and Exe
-            try:
-                steam_home = get_steam_home()
-                if not steam_home:
-                    self.signals.show_warning_dialog.emit(_("Error"), _("Steam directory not found"))
-                    return
-
-                last_user = get_last_steam_user(steam_home)
-                if not last_user or 'SteamID' not in last_user:
-                    self.signals.show_warning_dialog.emit(_("Error"), _("Failed to get Steam user ID"))
-                    return
-
-                userdata_dir = os.path.join(steam_home, "userdata")
-                user_id = last_user['SteamID']
-                unsigned_id = convert_steam_id(user_id)
-                user_dir = os.path.join(userdata_dir, str(unsigned_id))
-                steam_shortcuts_path = os.path.join(user_dir, "config", "shortcuts.vdf")
-                backup_path = f"{steam_shortcuts_path}.backup"
-
-                if not os.path.exists(steam_shortcuts_path):
-                    self.signals.show_warning_dialog.emit(
-                        _("Error"),
-                        _("Steam shortcuts file not found")
-                    )
-                    return
-
-                # Backup shortcuts.vdf
-                try:
-                    shutil.copy2(steam_shortcuts_path, backup_path)
-                    logger.info("Created backup of shortcuts.vdf at %s", backup_path)
-                except Exception as e:
-                    self.signals.show_warning_dialog.emit(
-                        _("Error"),
-                        _("Failed to create backup of shortcuts.vdf: {error}").format(error=str(e))
-                    )
-                    return
-
-                # Load shortcuts.vdf
-                try:
-                    with open(steam_shortcuts_path, 'rb') as f:
-                        shortcuts_data = vdf.binary_load(f)
-                except Exception as e:
-                    self.signals.show_warning_dialog.emit(
-                        _("Error"),
-                        _("Failed to load shortcuts.vdf: {error}").format(error=str(e))
-                    )
-                    return
-
-                shortcuts = shortcuts_data.get("shortcuts", {})
-                modified = False
-                new_shortcuts = {}
-                index = 0
-
-                for _key, entry in shortcuts.items():
-                    if entry.get("AppName") == game_name and entry.get("Exe") == quoted_script_path:
-                        modified = True
-                        logger.info("Removing EGS game '%s' from Steam shortcuts", game_name)
-                        continue
-                    new_shortcuts[str(index)] = entry
-                    index += 1
-
-                if not modified:
-                    self.signals.show_warning_dialog.emit(
-                        _("Error"),
-                        _("Game '{game_name}' not found in Steam shortcuts").format(game_name=game_name)
-                    )
-                    return
-
-                # Save updated shortcuts.vdf
-                try:
-                    with open(steam_shortcuts_path, 'wb') as f:
-                        vdf.binary_dump({"shortcuts": new_shortcuts}, f)
-                    logger.info("Updated shortcuts.vdf, removed '%s'", game_name)
-                    on_remove_from_steam_result((True, "Game '{game_name}' was removed from Steam. Please restart Steam for changes to take effect."))
-                except Exception as e:
-                    self.signals.show_warning_dialog.emit(
-                        _("Error"),
-                        _("Failed to update shortcuts.vdf: {error}").format(error=str(e))
-                    )
-                    if os.path.exists(backup_path):
-                        try:
-                            shutil.copy2(backup_path, steam_shortcuts_path)
-                            logger.info("Restored shortcuts.vdf from backup")
-                        except Exception as restore_err:
-                            logger.error("Failed to restore shortcuts.vdf: %s", restore_err)
-                    on_remove_from_steam_result((False, "Failed to update shortcuts.vdf: {error}"))
-                    return
-
-                # Optionally, remove the script file
-                if os.path.exists(script_path):
-                    try:
-                        os.remove(script_path)
-                        logger.info("Removed EGS script: %s", script_path)
-                    except OSError as e:
-                        logger.warning(f"Failed to remove EGS script '{script_path}': {str(e)}")
-
-            except Exception as e:
-                self.signals.show_warning_dialog.emit(
-                    _("Error"),
-                    _("Failed to remove EGS game '{game_name}' from Steam: {error}").format(
-                        game_name=game_name, error=str(e)
-                    )
-                )
-                on_remove_from_steam_result((False, "Failed to remove EGS game '{game_name}' from Steam: {error}"))
-                return
-
         else:
             # For non-EGS games, use steam_api
             exec_line = self._get_exec_line(game_name, exec_line)

portprotonqt/egs_api.py

@@ -27,18 +27,12 @@ from PySide6.QtGui import QPixmap
 logger = get_logger(__name__)
 downloader = Downloader()
 
-def get_egs_executable(app_name: str, legendary_config_path: str) -> str | None:
+def read_installed_json(legendary_config_path: str) -> dict | None:
-    """Получает путь к исполняемому файлу EGS-игры из installed.json с использованием orjson."""
+    """Читает installed.json и возвращает словарь с данными или None в случае ошибки."""
     installed_json_path = os.path.join(legendary_config_path, "installed.json")
     try:
         with open(installed_json_path, "rb") as f:
-            installed_data = orjson.loads(f.read())
+            return orjson.loads(f.read())
-            if app_name in installed_data:
-                install_path = installed_data[app_name].get("install_path", "").decode('utf-8') if isinstance(installed_data[app_name].get("install_path"), bytes) else installed_data[app_name].get("install_path", "")
-                executable = installed_data[app_name].get("executable", "").decode('utf-8') if isinstance(installed_data[app_name].get("executable"), bytes) else installed_data[app_name].get("executable", "")
-                if install_path and executable:
-                    return os.path.join(install_path, executable)
-            return None
     except FileNotFoundError:
         logger.error(f"installed.json not found at {installed_json_path}")
         return None
@@ -49,6 +43,17 @@ def get_egs_executable(app_name: str, legendary_config_path: str) -> str | None:
         logger.error(f"Error reading installed.json: {e}")
         return None
 
+def get_egs_executable(app_name: str, legendary_config_path: str) -> str | None:
+    """Получает путь к исполняемому файлу EGS-игры из installed.json."""
+    installed_data = read_installed_json(legendary_config_path)
+    if installed_data is None or app_name not in installed_data:
+        return None
+    install_path = installed_data[app_name].get("install_path", "").decode('utf-8') if isinstance(installed_data[app_name].get("install_path"), bytes) else installed_data[app_name].get("install_path", "")
+    executable = installed_data[app_name].get("executable", "").decode('utf-8') if isinstance(installed_data[app_name].get("executable"), bytes) else installed_data[app_name].get("executable", "")
+    if install_path and executable:
+        return os.path.join(install_path, executable)
+    return None
+
 def get_cache_dir() -> Path:
     """Returns the path to the cache directory, creating it if necessary."""
     xdg_cache_home = os.getenv(
@@ -59,6 +64,108 @@ def get_cache_dir() -> Path:
     cache_dir.mkdir(parents=True, exist_ok=True)
     return cache_dir
 
+def remove_egs_from_steam(game_name: str, portproton_dir: str, callback: Callable[[tuple[bool, str]], None]) -> None:
+    """
+    Removes an EGS game from Steam by modifying shortcuts.vdf and deleting the launch script.
+    Calls the callback with (success, message).
+
+    Args:
+        game_name: The display name of the game.
+        portproton_dir: Path to the PortProton directory.
+        callback: Callback function to handle the result (success, message).
+    """
+    if not portproton_dir:
+        logger.error("PortProton directory not found")
+        callback((False, "PortProton directory not found"))
+        return
+
+    steam_scripts_dir = os.path.join(portproton_dir, "steam_scripts")
+    safe_game_name = re.sub(r'[<>:"/\\|?*]', '_', game_name.strip())
+    script_path = os.path.join(steam_scripts_dir, f"{safe_game_name}_egs.sh")
+    quoted_script_path = f'"{script_path}"'
+
+    steam_home = get_steam_home()
+    if not steam_home:
+        logger.error("Steam directory not found")
+        callback((False, "Steam directory not found"))
+        return
+
+    last_user = get_last_steam_user(steam_home)
+    if not last_user or 'SteamID' not in last_user:
+        logger.error("Failed to retrieve Steam user ID")
+        callback((False, "Failed to get Steam user ID"))
+        return
+
+    userdata_dir = os.path.join(steam_home, "userdata")
+    user_id = last_user['SteamID']
+    unsigned_id = convert_steam_id(user_id)
+    user_dir = os.path.join(userdata_dir, str(unsigned_id))
+    steam_shortcuts_path = os.path.join(user_dir, "config", "shortcuts.vdf")
+    backup_path = f"{steam_shortcuts_path}.backup"
+
+    if not os.path.exists(steam_shortcuts_path):
+        logger.error("Steam shortcuts file not found")
+        callback((False, "Steam shortcuts file not found"))
+        return
+
+    try:
+        shutil.copy2(steam_shortcuts_path, backup_path)
+        logger.info("Created backup of shortcuts.vdf at %s", backup_path)
+    except Exception as e:
+        logger.error(f"Failed to create backup of shortcuts.vdf: {e}")
+        callback((False, f"Failed to create backup of shortcuts.vdf: {e}"))
+        return
+
+    try:
+        with open(steam_shortcuts_path, 'rb') as f:
+            shortcuts_data = vdf.binary_load(f)
+    except Exception as e:
+        logger.error(f"Failed to load shortcuts.vdf: {e}")
+        callback((False, f"Failed to load shortcuts.vdf: {e}"))
+        return
+
+    shortcuts = shortcuts_data.get("shortcuts", {})
+    modified = False
+    new_shortcuts = {}
+    index = 0
+
+    for _key, entry in shortcuts.items():
+        if entry.get("AppName") == game_name and entry.get("Exe") == quoted_script_path:
+            modified = True
+            logger.info("Removing EGS game '%s' from Steam shortcuts", game_name)
+            continue
+        new_shortcuts[str(index)] = entry
+        index += 1
+
+    if not modified:
+        logger.error("Game '%s' not found in Steam shortcuts", game_name)
+        callback((False, f"Game '{game_name}' not found in Steam shortcuts"))
+        return
+
+    try:
+        with open(steam_shortcuts_path, 'wb') as f:
+            vdf.binary_dump({"shortcuts": new_shortcuts}, f)
+        logger.info("Updated shortcuts.vdf, removed '%s'", game_name)
+    except Exception as e:
+        logger.error(f"Failed to update shortcuts.vdf: {e}")
+        if os.path.exists(backup_path):
+            try:
+                shutil.copy2(backup_path, steam_shortcuts_path)
+                logger.info("Restored shortcuts.vdf from backup")
+            except Exception as restore_err:
+                logger.error(f"Failed to restore shortcuts.vdf: {restore_err}")
+        callback((False, f"Failed to update shortcuts.vdf: {e}"))
+        return
+
+    if os.path.exists(script_path):
+        try:
+            os.remove(script_path)
+            logger.info("Removed EGS script: %s", script_path)
+        except OSError as e:
+            logger.warning(f"Failed to remove EGS script '{script_path}': {str(e)}")
+
+    callback((True, f"Game '{game_name}' was removed from Steam. Please restart Steam for changes to take effect."))
+
 def add_egs_to_steam(app_name: str, game_title: str, legendary_path: str, callback: Callable[[tuple[bool, str]], None]) -> None:
     """
     Asynchronously adds an EGS game to Steam via shortcuts.vdf with PortProton tag.