diff --git a/portprotonqt/app.py b/portprotonqt/app.py index 60fe5df2..cc2bf284 100644 --- a/portprotonqt/app.py +++ b/portprotonqt/app.py @@ -290,6 +290,7 @@ def main(): restore_prefix_path = None autoinstall_path = None theme_store_id = None + gog_launch_uri = None resolution_from_args = None if args.resolution: resolution_from_args = parse_resolution(args.resolution) @@ -315,6 +316,11 @@ def main(): elif args.file_or_url and is_autoinstall_file(args.file_or_url): autoinstall_path = normalize_launch_path(args.file_or_url) ipc_message = f"autoinstall:{quote(autoinstall_path, safe='')}" + elif args.file_or_url and args.file_or_url.startswith("gog://launch/"): + app_id = args.file_or_url.removeprefix("gog://launch/") + if app_id.isdigit(): + gog_launch_uri = args.file_or_url + ipc_message = f"gog:{app_id}" elif args.file_or_url: theme_store_id = parse_portprotonqt_theme_url(args.file_or_url) if theme_store_id is not None: @@ -381,7 +387,7 @@ def main(): logger.warning("Failed to persist PORT_DATA_PATH in PortProtonQt config") # Check if we have a portproton:// URL or launch file to handle - if args.file_or_url and not restore_prefix_path: + if args.file_or_url and not restore_prefix_path and not gog_launch_uri: if is_portproton_url(args.file_or_url): # Parse the portproton:// URL to get the full download URL download_url = parse_portproton_url(args.file_or_url) @@ -485,6 +491,10 @@ def main(): def handle_restore_prefix(): window._perform_restore(restore_prefix_path) QTimer.singleShot(0, handle_restore_prefix) + elif gog_launch_uri: + def handle_gog_launch(): + window.toggleGame(gog_launch_uri) + QTimer.singleShot(0, handle_gog_launch) # --- Handle incoming connections --- def handle_new_connection(): @@ -507,6 +517,7 @@ def main(): or msg.startswith("backup:") or msg.startswith("theme:") or msg.startswith("autoinstall:") + or msg.startswith("gog:") ): # Ensure the window is visible and not minimized window.setWindowState(window.windowState() & ~Qt.WindowState.WindowMinimized) @@ -542,6 +553,12 @@ def main(): window._perform_restore(backup_path) else: logger.warning("Invalid prefix backup via IPC: %s", backup_path) + elif msg.startswith("gog:"): + app_id = msg[4:].strip() + if app_id.isdigit(): + window.toggleGame(f"gog://launch/{app_id}") + else: + logger.warning("Invalid GOG app id via IPC: %s", app_id) elif msg.startswith("backup:"): parts = msg.split(":", 2) if len(parts) == 3: @@ -593,6 +610,7 @@ def main(): and restore_prefix_path is None and autoinstall_path is None and theme_store_id is None + and gog_launch_uri is None ) if launch_minimized: logger.info("Launching in tray") diff --git a/portprotonqt/config/portproton.py b/portprotonqt/config/portproton.py index 3706dd39..069758b6 100644 --- a/portprotonqt/config/portproton.py +++ b/portprotonqt/config/portproton.py @@ -528,10 +528,12 @@ def resolve_custom_data_dir(custom_data_root: str, exe_path: str) -> str: def create_desktop_file( exe_path: str, game_name: str | None = None, + icon_source: str | None = None, ) -> tuple[str, str, str] | None: """Create desktop entry content, destination path and icon path for a game.""" portproton_path = get_portproton_location() - if not os.path.isfile(exe_path): + is_gog_uri = re.fullmatch(r"gog://launch/\d+", exe_path) is not None + if not is_gog_uri and not os.path.isfile(exe_path): logger.error("Executable not found: %s", exe_path) return None if not portproton_path: @@ -542,9 +544,14 @@ def create_desktop_file( game_name = os.path.splitext(os.path.basename(exe_path))[0] base_path = os.path.join(portproton_path, "data") icon_name = _sanitize_icon_name(game_name) - icon_path = os.path.join(base_path, "img", f"{icon_name}.png") + has_icon_source = bool(icon_source and os.path.isfile(icon_source)) + icon_path = ( + "applications-games" if is_gog_uri and not has_icon_source + else os.path.join(base_path, "img", f"{icon_name}.png") + ) desktop_path = os.path.join(portproton_path, f"{game_name}.desktop") - os.makedirs(os.path.dirname(icon_path), exist_ok=True) + if icon_path != "applications-games": + os.makedirs(os.path.dirname(icon_path), exist_ok=True) flatpak_id = os.getenv("FLATPAK_ID") appimage_path = os.getenv("APPIMAGE", "").strip() diff --git a/portprotonqt/context_menu_manager.py b/portprotonqt/context_menu_manager.py index be040194..24c42552 100644 --- a/portprotonqt/context_menu_manager.py +++ b/portprotonqt/context_menu_manager.py @@ -459,7 +459,7 @@ class ContextMenuManager: menu.setStyleSheet(self.theme.CONTEXT_MENU_STYLE) # For non-Steam games, check if exe exists - if game_card.game_source != "steam": + if game_card.game_source == "portproton": exec_line = self._get_exec_line(game_card.name, game_card.exec_line) exe_path = self._parse_exe_path(exec_line, game_card.name) if exec_line else None if not exe_path: @@ -470,13 +470,18 @@ class ContextMenuManager: return # Normal menu for games with valid exe or from Steam - is_running = self._is_game_running(game_card) - action_text = _("Stop Game") if is_running else _("Launch Game") - action_icon = "stop" if is_running else "play" - launch_action = menu.addAction(self._get_safe_icon(action_icon), action_text) - launch_action.triggered.connect( - lambda: self._launch_game(game_card) + is_uninstalled_gog = ( + game_card.game_source == "gog" + and game_card.exec_line.startswith("gog://install/") ) + is_running = self._is_game_running(game_card) + if not is_uninstalled_gog: + action_text = _("Stop Game") if is_running else _("Launch Game") + action_icon = "stop" if is_running else "play" + launch_action = menu.addAction(self._get_safe_icon(action_icon), action_text) + launch_action.triggered.connect( + lambda: self._launch_game(game_card) + ) favorites = favorites_config.get_games() is_favorite = game_card.name in favorites @@ -485,6 +490,60 @@ class ContextMenuManager: favorite_action = menu.addAction(self._get_safe_icon(icon_name), text) favorite_action.triggered.connect(lambda: self.toggle_favorite(game_card, not is_favorite)) + if game_card.game_source == "gog": + game = { + "app_id": str(game_card.appid), + "title": game_card.name, + "cover": game_card.cover_path, + } + if game_card.exec_line.startswith("gog://launch/"): + repair_action = menu.addAction( + self._get_safe_icon("update"), _("Repair") + ) + repair_action.triggered.connect( + lambda: self.parent._repair_gog_game(game) + ) + desktop_dir = QStandardPaths.writableLocation( + QStandardPaths.StandardLocation.DesktopLocation + ) + desktop_path = self._get_shortcut_path( + game_card.name, desktop_dir + ) + desktop_exists = os.path.exists(desktop_path) + desktop_action = menu.addAction( + self._get_safe_icon("delete" if desktop_exists else "desktop"), + _("Remove from Desktop") if desktop_exists else _("Add to Desktop"), + ) + desktop_action.triggered.connect( + lambda: self.remove_from_desktop(game_card.name) + if desktop_exists + else self.add_to_desktop(game_card.name, game_card.exec_line) + ) + menu_path = self._get_menu_shortcut_path(game_card.name) + menu_exists = os.path.exists(menu_path) + menu_action = menu.addAction( + self._get_safe_icon("delete" if menu_exists else "menu"), + _("Remove from Menu") if menu_exists else _("Add to Menu"), + ) + menu_action.triggered.connect( + lambda: self.remove_from_menu(game_card.name) + if menu_exists + else self.add_to_menu(game_card.name, game_card.exec_line) + ) + delete_action = menu.addAction( + self._get_safe_icon("delete"), _("Delete") + ) + delete_action.triggered.connect( + lambda: self.parent._delete_gog_game(game) + ) + else: + import_action = menu.addAction( + self._get_safe_icon("folder"), _("Import") + ) + import_action.triggered.connect( + lambda: self.parent._import_gog_game(game) + ) + if game_card.game_source == "steam": desktop_dir = QStandardPaths.writableLocation( QStandardPaths.StandardLocation.DesktopLocation @@ -524,7 +583,7 @@ class ContextMenuManager: ) ) - if game_card.game_source != "steam": + if game_card.game_source == "portproton": desktop_dir = QStandardPaths.writableLocation(QStandardPaths.StandardLocation.DesktopLocation) desktop_path = self._get_shortcut_path(game_card.name, desktop_dir) icon_name = "delete" if os.path.exists(desktop_path) else "desktop" @@ -916,23 +975,33 @@ class ContextMenuManager: """Add the game desktop entry to the XDG applications directory.""" if not self._check_portproton(): return + is_gog_uri = str(exec_line).startswith("gog://launch/") desktop_path = self._get_desktop_path(game_name) - if not os.path.exists(desktop_path): + if not is_gog_uri and not os.path.exists(desktop_path): self.signals.show_warning_dialog.emit( _("Error"), _("No .desktop file found for '{game_name}'").format(game_name=game_name) ) return - exec_line = self._get_exec_line(game_name, exec_line) - if not exec_line: - return - exe_path = self._parse_exe_path(exec_line, game_name) - if not exe_path: - return - shortcut = create_desktop_file(exe_path, game_name) + if not is_gog_uri: + exec_line = self._get_exec_line(game_name, exec_line) + if not exec_line: + return + exec_line = self._parse_exe_path(exec_line, game_name) + if not exec_line: + return + icon_source = None + if is_gog_uri: + app_id = str(exec_line).rsplit("/", 1)[-1] + icon_source = self.parent.gog_api.get_launch_target(app_id) + shortcut = create_desktop_file(exec_line, game_name, icon_source) if not shortcut: return desktop_entry = shortcut[0] + icon_path = shortcut[2] + if icon_source and not os.path.exists(icon_path): + if not generate_thumbnail(icon_source, icon_path, size=128): + logger.error("Failed to generate thumbnail for game: %s", icon_source) applications_dir = self._get_applications_dir() os.makedirs(applications_dir, exist_ok=True) dest_path = self._get_menu_shortcut_path(game_name) @@ -975,28 +1044,35 @@ class ContextMenuManager: if not self._check_portproton(): return + is_gog_uri = str(exec_line).startswith("gog://launch/") desktop_path = self._get_desktop_path(game_name) - if not os.path.exists(desktop_path): + if not is_gog_uri and not os.path.exists(desktop_path): self.signals.show_warning_dialog.emit( _("Error"), _("No .desktop file found for '{game_name}'").format(game_name=game_name) ) return # Ensure icon exists - exec_line = self._get_exec_line(game_name, exec_line) - if not exec_line: - return - exe_path = self._parse_exe_path(exec_line, game_name) - if not exe_path: - return - shortcut = create_desktop_file(exe_path, game_name) + if not is_gog_uri: + exec_line = self._get_exec_line(game_name, exec_line) + if not exec_line: + return + exec_line = self._parse_exe_path(exec_line, game_name) + if not exec_line: + return + icon_source = None + if is_gog_uri: + app_id = str(exec_line).rsplit("/", 1)[-1] + icon_source = self.parent.gog_api.get_launch_target(app_id) + shortcut = create_desktop_file(exec_line, game_name, icon_source) if not shortcut: return desktop_entry = shortcut[0] icon_path = shortcut[2] - if not os.path.exists(icon_path): - if not generate_thumbnail(exe_path, icon_path, size=128): - logger.error("Failed to generate thumbnail for game: %s", exe_path) + thumbnail_source = icon_source or exec_line + if icon_path != "applications-games" and not os.path.exists(icon_path): + if not generate_thumbnail(thumbnail_source, icon_path, size=128): + logger.error("Failed to generate thumbnail for game: %s", thumbnail_source) desktop_dir = QStandardPaths.writableLocation(QStandardPaths.StandardLocation.DesktopLocation) os.makedirs(desktop_dir, exist_ok=True) @@ -1074,6 +1150,20 @@ class ContextMenuManager: fetch_client_icon_async(appid_str, write_shortcut) + def remove_gog_shortcuts(self, game_name: str) -> None: + """Remove desktop and application menu shortcuts for a GOG game.""" + desktop_dir = QStandardPaths.writableLocation( + QStandardPaths.StandardLocation.DesktopLocation + ) + for target_dir in (desktop_dir, self._get_applications_dir()): + shortcut_path = self._get_shortcut_path(game_name, target_dir) + try: + os.remove(shortcut_path) + except FileNotFoundError: + continue + except OSError as error: + logger.warning("Failed to remove GOG shortcut %s: %s", shortcut_path, error) + def add_steam_to_desktop(self, game_name: str, appid: int | str) -> None: """Create a desktop shortcut for an installed Steam game.""" desktop_dir = QStandardPaths.writableLocation( diff --git a/portprotonqt/detail_pages/__init__.py b/portprotonqt/detail_pages/__init__.py index 645fa6c5..e5326886 100644 --- a/portprotonqt/detail_pages/__init__.py +++ b/portprotonqt/detail_pages/__init__.py @@ -1,7 +1,10 @@ """Detail pages for PortProtonQt.""" import os +import weakref from collections.abc import Callable + +from shiboken6 import isValid from PySide6.QtWidgets import ( QWidget, QLabel, @@ -13,7 +16,7 @@ from PySide6.QtWidgets import ( QBoxLayout, QSizePolicy, ) -from PySide6.QtCore import Qt, QUrl, QTimer +from PySide6.QtCore import Qt, QUrl, QTimer, QProcess, QProcessEnvironment from PySide6.QtGui import QDesktopServices from portprotonqt.detail_pages.widgets import ( @@ -80,11 +83,15 @@ class DetailPageManager: self._autoinstall_status_controls = None self._current_detail_source: tuple[str, dict] | None = None self._resize_rebuild_pending = False + self._gog_size_requests: dict[ + str, tuple[QProcess, list[list[Callable[[], QLabel | None]]]] + ] = {} def openGameDetailPage(self, game_data: dict) -> None: """Open detailed game information page.""" self._current_detail_source = ("game", dict(game_data)) detail_page = QWidget() + detail_page.setProperty("coverCacheKey", str(game_data.get("appid", ""))) fallback_exe, fallback_icon_path = self._get_exe_icon_fallback(game_data) detail_page.setProperty("fallbackExe", fallback_exe) detail_page.setProperty("fallbackIconPath", fallback_icon_path) @@ -301,12 +308,97 @@ class DetailPageManager: ) game_info_layout.addLayout(first_row) + self._setup_gog_size_data(game_data, game_info_layout) + hltb_layout = QHBoxLayout() hltb_layout.setSpacing(10) self._setup_hltb_data(game_data.get("name", ""), hltb_layout, game_info_layout) return game_info_layout + def _setup_gog_size_data( + self, game_data: dict, game_info_layout: QVBoxLayout + ) -> None: + exec_line = str(game_data.get("exec_line", "")) + if str(game_data.get("game_source", "")).lower() != "gog": + return + if not exec_line.startswith("gog://install/"): + return + row = QHBoxLayout() + values = [] + for title_text in (_("Download"), _("Size")): + title = QLabel(title_text.upper()) + title.setStyleSheet(self.main_window.theme.LAST_LAUNCH_TITLE_STYLE) + value = QLabel("—") + value.setStyleSheet(self.main_window.theme.LAST_LAUNCH_VALUE_STYLE) + row.addWidget(title) + row.addWidget(value) + row.addSpacing(30) + values.append(weakref.ref(value)) + game_info_layout.addLayout(row) + app_id = exec_line.rsplit("/", 1)[-1] + cached = self.main_window.gog_api.get_cached_download_sizes(app_id) + if cached is not None: + self._set_gog_size_values(values, cached) + return + request = self._gog_size_requests.get(app_id) + if request is not None: + request[1].append(values) + return + try: + command = self.main_window.gog_api.build_command( + ["info", app_id, "--platform", "windows", "--skip-dlcs"] + ) + except FileNotFoundError as error: + logger.warning("Failed to get GOG download size: %s", error) + return + process = QProcess(self.main_window) + process.setProgram(command[0]) + process.setArguments(command[1:]) + environment = QProcessEnvironment.systemEnvironment() + environment.insert("GOGDL_CONFIG_PATH", str(self.main_window.gog_api.config_dir)) + process.setProcessEnvironment(environment) + self._gog_size_requests[app_id] = (process, [values]) + process.finished.connect( + lambda code, _status: self._finish_gog_size_request(app_id, code) + ) + process.start() + + def _finish_gog_size_request(self, app_id: str, code: int) -> None: + request = self._gog_size_requests.pop(app_id, None) + if request is None: + return + process, value_groups = request + if code != 0: + logger.warning("gogdl info exited with code %s", code) + process.deleteLater() + return + try: + sizes = self.main_window.gog_api.parse_download_sizes( + bytes(process.readAllStandardOutput()) + ) + except (ValueError, TypeError) as error: + logger.warning("Failed to parse GOG download size: %s", error) + process.deleteLater() + return + self.main_window.gog_api.save_download_sizes(app_id, sizes) + for value_refs in value_groups: + self._set_gog_size_values(value_refs, sizes) + process.deleteLater() + + @staticmethod + def _set_gog_size_values( + value_refs: list[Callable[[], QLabel | None]], sizes: tuple[int, int] + ) -> None: + for value_ref, size in zip(value_refs, sizes, strict=True): + label = value_ref() + if label is None or not isValid(label): + continue + divisor, unit = ( + (1024 ** 3, "GiB") if size >= 1024 ** 3 else (1024 ** 2, "MiB") + ) + label.setText(f"{size / divisor:.1f} {unit}") + def _create_playtime_row(self, last_launch: str, formatted_playtime: str) -> QHBoxLayout: """Create first row with last launch and playtime.""" first_row = QHBoxLayout() @@ -438,6 +530,11 @@ class DetailPageManager: current_exe = self._get_current_exe(exec_line) play_button = self._create_play_button(exec_line, current_exe) + if str(game_source).lower() == "gog" and exec_line.startswith("gog://install/"): + play_button.setText(_("Install")) + play_button.setIcon( + self.main_window.theme_manager.get_icon("update", as_path=True) + ) buttons_layout.addWidget(play_button) if str(game_source).lower() == "steam" and appid: @@ -471,7 +568,10 @@ class DetailPageManager: lambda: self._open_steam_game_folder(str(appid)) ) buttons_layout.addWidget(open_folder_button) - elif not self._has_game_shortcut(game_name): + elif ( + str(game_source).lower() != "gog" + and not self._has_game_shortcut(game_name) + ): add_button = self._make_action_button( _("Add a shortcut"), self.main_window.theme_manager.get_icon("addgame", as_path=True), @@ -486,6 +586,40 @@ class DetailPageManager: # Show settings button for PortProton games or Steam games using PortProtonQt if str(game_source).lower() == "portproton": self._add_portproton_buttons(buttons_layout, exec_line) + elif ( + str(game_source).lower() == "gog" + and appid + and exec_line.startswith("gog://launch/") + ): + gog_exe = self.main_window.gog_api.get_launch_target(str(appid)) + if not gog_exe: + return buttons_layout + self.main_window.gog_api.ensure_launch_parameters(str(appid)) + settings_button = self._make_action_button( + _("Settings"), + self.main_window.theme_manager.get_icon("settings", as_path=True), + ) + settings_button.clicked.connect( + lambda: self.main_window.open_exe_settings( + gog_exe, game_source="gog", + ) + ) + buttons_layout.addWidget(settings_button) + log_button = self._make_action_button( + _("Create Log"), + self.main_window.theme_manager.get_icon("edit", as_path=True), + ) + log_button.clicked.connect(lambda: self.toggleDebugLog(gog_exe, log_button)) + buttons_layout.addWidget(log_button) + self._debug_log_button = log_button + folder_button = self._make_action_button( + _("Open Folder"), + self.main_window.theme_manager.get_icon("search", as_path=True), + ) + folder_button.clicked.connect( + lambda: self._open_executable_folder(gog_exe) + ) + buttons_layout.addWidget(folder_button) elif str(game_source).lower() == "steam" and appid: try: compat_tool = get_steam_compat_tool(int(appid)) @@ -590,9 +724,13 @@ class DetailPageManager: def _open_portproton_game_folder(self, exec_line: str) -> None: """Open folder containing the PortProton game executable.""" file_to_check = self._get_file_from_exec(exec_line) - if not file_to_check or not os.path.exists(file_to_check): + self._open_executable_folder(file_to_check) + + def _open_executable_folder(self, exe_path: str | None) -> None: + """Open the folder containing an executable.""" + if not exe_path or not os.path.exists(exe_path): return - folder_path = os.path.dirname(os.path.abspath(file_to_check)) + folder_path = os.path.dirname(os.path.abspath(exe_path)) QDesktopServices.openUrl(QUrl.fromLocalFile(folder_path)) def _has_game_shortcut(self, game_name: str) -> bool: diff --git a/portprotonqt/detail_pages/utils.py b/portprotonqt/detail_pages/utils.py index 20a0af65..3c0b01b3 100644 --- a/portprotonqt/detail_pages/utils.py +++ b/portprotonqt/detail_pages/utils.py @@ -64,6 +64,7 @@ def setup_image_loading( cover_width, cover_height, lambda pixmap: relay.pixmap_ready.emit(pixmap), + app_name=str(detail_page.property("coverCacheKey") or ""), fallback_exe=fallback_exe, fallback_icon_path=fallback_icon_path, ) diff --git a/portprotonqt/detail_pages/widgets.py b/portprotonqt/detail_pages/widgets.py index 7902dc64..bbac6733 100644 --- a/portprotonqt/detail_pages/widgets.py +++ b/portprotonqt/detail_pages/widgets.py @@ -139,10 +139,10 @@ def _add_source_corner( ) -> None: """Add source corner badge to cover frame.""" source = game_source.lower() - if source not in ("steam", "portproton") or theme_manager is None: + if source not in ("steam", "gog", "portproton") or theme_manager is None: return - icon_name = "badge_steam" if source == "steam" else "badge_portproton" + icon_name = f"badge_{source}" icon = theme_manager.get_icon(icon_name, as_path=True) if not icon: return diff --git a/portprotonqt/game_card.py b/portprotonqt/game_card.py index b41354b5..c3ab7ad0 100644 --- a/portprotonqt/game_card.py +++ b/portprotonqt/game_card.py @@ -262,6 +262,7 @@ class GameCard(QFrame): self.missing_executable_path = self._get_missing_executable_path() self.steam_visible = (str(game_source).lower() == "steam" and not self.economy_mode) + self.gog_visible = (str(game_source).lower() == "gog" and not self.economy_mode) self.portproton_visible = (str(game_source).lower() == "portproton" and not self.economy_mode) self.ppdb_visible = bool(self.ppdb_id) and not self.economy_mode @@ -302,7 +303,10 @@ class GameCard(QFrame): self.coverLabel = QLabel() self.coverLabel.setAlignment(Qt.AlignmentFlag.AlignCenter) self.coverLabel.setStyleSheet(self.theme.COVER_LABEL_STYLE) - if self.missing_executable_path: + if self.missing_executable_path or ( + str(self.game_source).lower() == "gog" + and self.exec_line.startswith("gog://install/") + ): self.coverOpacity = QGraphicsOpacityEffect(self.coverLabel) self.coverOpacity.setOpacity(self.theme.missing_exe_cover_opacity) self.coverLabel.setGraphicsEffect(self.coverOpacity) @@ -343,6 +347,16 @@ class GameCard(QFrame): if self.economy_mode: self.steamLabel.setVisible(False) + gog_icon = self.theme_manager.get_icon("badge_gog", as_path=True) + self.gogLabel = SourceCorner( + icon=gog_icon, + config=corner_config, + parent=self.coverWidget, + ) + self.gogLabel.setVisible(self.gog_visible) + if self.economy_mode: + self.gogLabel.setVisible(False) + portproton_icon = self.theme_manager.get_icon("badge_portproton", as_path=True) self.portprotonLabel = SourceCorner( icon=portproton_icon, @@ -436,6 +450,7 @@ class GameCard(QFrame): width, height, self.on_cover_loaded, + app_name=str(self.appid or ""), fallback_exe=fallback_exe, fallback_icon_path=fallback_icon_path, ) @@ -568,6 +583,7 @@ class GameCard(QFrame): ribbon_size = max(ribbon_size, int(ribbon_cfg.get("min_size", 54) * self._scale)) source_ribbons = [ (self.steam_visible, self.steamLabel), + (self.gog_visible, self.gogLabel), (self.portproton_visible, self.portprotonLabel), ] @@ -589,6 +605,7 @@ class GameCard(QFrame): self.ppdbLabel.raise_() self.protondbLabel.raise_() self.portprotonLabel.raise_() + self.gogLabel.raise_() self.steamLabel.raise_() except RuntimeError: pass @@ -715,6 +732,7 @@ class GameCard(QFrame): self.display_filter = display_filter self.economy_mode = ui_config.get_economy_mode() self.steam_visible = (str(self.game_source).lower() == "steam" and not self.economy_mode) + self.gog_visible = (str(self.game_source).lower() == "gog" and not self.economy_mode) self.portproton_visible = (str(self.game_source).lower() == "portproton" and not self.economy_mode) self.ppdb_visible = bool(self.ppdb_id) and not self.economy_mode protondb_visible = is_valid_protondb_tier(self.protondb_tier) and not self.economy_mode @@ -724,6 +742,7 @@ class GameCard(QFrame): try: self.steamLabel.setVisible(self.steam_visible and not hidden_badges) + self.gogLabel.setVisible(self.gog_visible and not hidden_badges) self.portprotonLabel.setVisible(self.portproton_visible and not hidden_badges) self.ppdbLabel.setVisible(self.ppdb_visible and not hidden_badges) self.protondbLabel.setVisible(protondb_visible and not hidden_badges) diff --git a/portprotonqt/gog_api.py b/portprotonqt/gog_api.py new file mode 100644 index 00000000..3bb36bbd --- /dev/null +++ b/portprotonqt/gog_api.py @@ -0,0 +1,471 @@ +"""GOG authentication, library and gogdl process helpers.""" + +import hashlib +import os +import platform +import re +import shutil +import subprocess +from collections.abc import Callable +from concurrent.futures import ThreadPoolExecutor, as_completed +from pathlib import Path +from urllib.parse import parse_qs, urlparse + +import orjson +import requests + +from portprotonqt.localization import get_metadata_language +from portprotonqt.logger import get_logger + +logger = get_logger(__name__) +GOG_LOGIN_URL = ( + "https://auth.gog.com/auth?client_id=46899977096215655&" + "redirect_uri=https%3A%2F%2Fembed.gog.com%2Fon_login_success%3Forigin%3Dclient&" + "response_type=code&layout=galaxy" +) +GOGDL_RELEASE_URL = "https://api.github.com/repos/Heroic-Games-Launcher/heroic-gogdl/releases/latest" +GOGDL_AUTH_TIMEOUT = 60 +GOG_METADATA_WORKERS = 4 +GOG_PRODUCT_LOCALES = { + "bg": "bg-BG", "cs": "cs-CZ", "da": "da-DK", "de": "de-DE", + "el": "el-GR", "en": "en-US", "es": "es-ES", "fi": "fi-FI", + "fr": "fr-FR", "hu": "hu-HU", "it": "it-IT", "ja": "ja-JP", + "ko": "ko-KR", "nl": "nl-NL", "no": "no-NO", "pl": "pl-PL", + "pt": "pt-BR", "ro": "ro-RO", "ru": "ru-RU", "sv": "sv-SE", + "th": "th-TH", "tr": "tr-TR", "uk": "uk-UA", "zh": "zh-Hans", + "zh_hant": "zh-Hant", +} + + +class GOGAPI: + """Provide the local GOG backend used by the Qt UI.""" + + def __init__(self) -> None: + data_home = Path(os.getenv("XDG_DATA_HOME", Path.home() / ".local/share")) + self.data_dir = data_home / "PortProtonQt" / "gog" + self.auth_path = self.data_dir / "auth.json" + self.config_dir = self.data_dir / "gogdl" + self.library_path = self.data_dir / "library.json" + self.installed_path = self.data_dir / "installed.json" + self.sizes_path = self.data_dir / "sizes.json" + self.games_dir = Path.home() / "Games" + + def get_gogdl_path(self) -> str | None: + """Return an available gogdl executable.""" + bundled = self.data_dir / "bin/gogdl" + if bundled.is_file() and os.access(bundled, os.X_OK): + return str(bundled) + return shutil.which("gogdl") + + def build_command(self, arguments: list[str]) -> list[str]: + """Build a gogdl command with private configuration paths.""" + gogdl = self.get_gogdl_path() + if not gogdl: + raise FileNotFoundError("gogdl executable not found") + self.data_dir.mkdir(parents=True, exist_ok=True) + self.data_dir.chmod(0o700) + self.config_dir.mkdir(parents=True, exist_ok=True) + return [gogdl, "--auth-config-path", str(self.auth_path), *arguments] + + def ensure_gogdl(self) -> str: + """Download and verify the latest official Heroic gogdl binary.""" + existing = self.get_gogdl_path() + if existing: + return existing + architecture = {"x86_64": "x86_64", "aarch64": "arm64"}.get(platform.machine()) + if not architecture: + raise OSError(f"Unsupported gogdl architecture: {platform.machine()}") + response = requests.get(GOGDL_RELEASE_URL, timeout=15) + response.raise_for_status() + asset_name = f"gogdl_linux_{architecture}" + asset = next( + (item for item in response.json().get("assets", []) if item.get("name") == asset_name), + None, + ) + if not asset: + raise OSError(f"gogdl release asset not found: {asset_name}") + return self._download_gogdl_asset(asset) + + def _download_gogdl_asset(self, asset: dict) -> str: + target = self.data_dir / "bin/gogdl" + temporary = target.with_suffix(".part") + target.parent.mkdir(parents=True, exist_ok=True) + digest = hashlib.sha256() + try: + with requests.get(asset["browser_download_url"], stream=True, timeout=30) as response: + response.raise_for_status() + with open(temporary, "wb") as output: + for chunk in response.iter_content(chunk_size=1024 * 1024): + if chunk: + output.write(chunk) + digest.update(chunk) + expected = str(asset.get("digest", "")).removeprefix("sha256:") + if not expected or digest.hexdigest() != expected: + raise OSError("gogdl checksum verification failed") + temporary.chmod(0o755) + temporary.replace(target) + return str(target) + except Exception: + temporary.unlink(missing_ok=True) + raise + + def get_environment(self) -> dict[str, str]: + """Return the environment required by gogdl.""" + environment = os.environ.copy() + environment["GOGDL_CONFIG_PATH"] = str(self.config_dir) + return environment + + def authenticate(self, code: str) -> tuple[bool, str]: + """Exchange an OAuth code through gogdl.""" + self.ensure_gogdl() + result = subprocess.run( + self.build_command(["auth", "--code", code]), + env=self.get_environment(), capture_output=True, check=False, + timeout=GOGDL_AUTH_TIMEOUT, + ) + if result.returncode != 0: + error = result.stderr.decode(errors="replace").strip() + logger.error("gogdl authentication failed: %s", error) + return False, error or f"gogdl exited with code {result.returncode}" + try: + data = orjson.loads(result.stdout) + except orjson.JSONDecodeError: + return False, "gogdl returned an invalid authentication response" + if not isinstance(data, dict) or data.get("error"): + return False, "GOG rejected the authorization code" + if self.auth_path.is_file(): + self.auth_path.chmod(0o600) + return True, "" + + def get_credentials(self) -> dict: + """Return refreshed GOG credentials through gogdl.""" + try: + self.ensure_gogdl() + result = subprocess.run( + self.build_command(["auth"]), env=self.get_environment(), + capture_output=True, check=False, + ) + data = orjson.loads(result.stdout) if result.returncode == 0 else {} + return data if isinstance(data, dict) else {} + except (FileNotFoundError, orjson.JSONDecodeError): + return {} + + def is_authenticated(self) -> bool: + """Return whether gogdl provides usable account credentials.""" + credentials = self.get_credentials() + return bool(credentials.get("access_token") and credentials.get("user_id")) + + def refresh_library( + self, progress_callback: Callable[[int, int], None] | None = None + ) -> list[dict]: + """Fetch and cache the user's GOG library.""" + credentials = self.get_credentials() + token = credentials.get("access_token") + user_id = credentials.get("user_id") + if not token or not user_id: + return [] + entries = self._get_library_entries(str(user_id), str(token)) + gog_entries = [entry for entry in entries if entry.get("platform_id") == "gog"] + games = self._get_games_parallel(gog_entries, str(token), progress_callback) + self._save_json(self.library_path, games) + return games + + def _get_games_parallel( + self, entries: list[dict], token: str, + progress_callback: Callable[[int, int], None] | None, + ) -> list[dict]: + cached = {str(game.get("app_id")): game for game in self.load_library()} + games = [] + worker_count = min(GOG_METADATA_WORKERS, os.cpu_count() or 1) + with ThreadPoolExecutor(max_workers=worker_count) as executor: + futures = {executor.submit(self._get_game, entry, token): entry for entry in entries} + for completed, future in enumerate(as_completed(futures), 1): + entry = futures[future] + app_id = str(entry.get("external_id", "")) + try: + game = future.result() + except requests.RequestException as error: + logger.warning("Failed to load GOG metadata for %s: %s", app_id, error) + game = cached.get(app_id, {}) + if game: + games.append(game) + if progress_callback: + progress_callback(completed, len(entries)) + return games + + def load_library(self) -> list[dict]: + """Load the cached GOG library.""" + data = self._load_json(self.library_path, []) + return data if isinstance(data, list) else [] + + def load_installed(self) -> dict[str, dict]: + """Load installed GOG game records.""" + data = self._load_json(self.installed_path, {}) + return data if isinstance(data, dict) else {} + + def save_installed_game(self, app_id: str, game: dict) -> None: + """Persist one installed GOG game record.""" + installed = self.load_installed() + installed[app_id] = game + self._save_json(self.installed_path, installed) + + def remove_installed_game(self, app_id: str) -> None: + """Remove one installed GOG game record.""" + installed = self.load_installed() + installed.pop(app_id, None) + self._save_json(self.installed_path, installed) + + def get_cached_download_sizes(self, app_id: str) -> tuple[int, int] | None: + """Return cached download and disk sizes for a game.""" + cache = self._load_json(self.sizes_path, {}) + if not isinstance(cache, dict): + return None + cached = cache.get(app_id) + if not isinstance(cached, list) or len(cached) != 2: + return None + return int(cached[0]), int(cached[1]) + + def save_download_sizes(self, app_id: str, sizes: tuple[int, int]) -> None: + """Cache download and disk sizes for a game.""" + cached = self._load_json(self.sizes_path, {}) + if not isinstance(cached, dict): + cached = {} + cached[app_id] = list(sizes) + self._save_json(self.sizes_path, cached) + + def is_game_installed( + self, app_id: str, installed_games: dict[str, dict] | None = None + ) -> bool: + """Return whether a complete gogdl installation is available.""" + return self.get_installed_path(app_id, installed_games) is not None + + def get_installed_path( + self, app_id: str, installed_games: dict[str, dict] | None = None + ) -> Path | None: + """Return a recorded or discoverable gogdl installation path.""" + records = installed_games if installed_games is not None else self.load_installed() + installed = records.get(app_id, {}) + install_path = Path(str(installed.get("install_path", ""))) + if (install_path / f"goggame-{app_id}.info").is_file(): + return install_path + return self.find_install_path(app_id, self.games_dir) + + def get_install_path(self, app_id: str, title: str) -> Path: + """Return the default parent directory for GOG installations.""" + return self.games_dir + + def find_install_path(self, app_id: str, parent: Path) -> Path | None: + """Find the directory created by gogdl for an installed game.""" + metadata_name = f"goggame-{app_id}.info" + if (parent / metadata_name).is_file(): + return parent + try: + return next( + (path.parent for path in parent.glob(f"*/{metadata_name}")), + None, + ) + except OSError as error: + logger.warning("Failed to inspect GOG installation path %s: %s", parent, error) + return None + + def get_launch_target(self, app_id: str) -> str | None: + """Return the primary executable from gogdl installation metadata.""" + primary, install_path = self._get_primary_task(app_id) + if primary is None or install_path is None: + return None + target = install_path + parts = str(primary.get("path", "")).replace("\\", "/").split("/") + for part in parts: + exact_path = target / part + if exact_path.exists(): + target = exact_path + continue + try: + target = next( + child for child in target.iterdir() + if child.name.casefold() == part.casefold() + ) + except (OSError, StopIteration): + return None + return str(target) if target.is_file() else None + + def ensure_launch_parameters(self, app_id: str) -> None: + """Add GOG task arguments to the executable PPDB when absent.""" + primary, install_path = self._get_primary_task(app_id) + target = self.get_launch_target(app_id) + if primary is None or install_path is None or not target: + return + arguments = str(primary.get("arguments", "")).replace("\r", " ").replace("\n", " ").strip() + if not arguments: + return + ppdb_path = Path(f"{target}.ppdb") + try: + current = ppdb_path.read_text(encoding="utf-8") if ppdb_path.exists() else "" + base_value = arguments.replace('"', "").replace("\\", "/") + normalized = base_value + support_dir = install_path / "gog-support" / app_id / "app" + if support_dir.is_dir(): + for support_file in support_dir.rglob("*"): + if not support_file.is_file(): + continue + relative_path = os.path.relpath(support_file, Path(target).parent) + normalized = normalized.replace( + f"../{support_file.name}", relative_path.replace(os.sep, "/") + ) + safe_value = normalized.replace("$", "\\$").replace("`", "\\`") + launch_line = f'export LAUNCH_PARAMETERS="{safe_value}"' + lines = current.splitlines() + existing = next( + (line for line in lines if line.startswith("export LAUNCH_PARAMETERS=")), + "", + ) + if existing: + value = existing.removeprefix('export LAUNCH_PARAMETERS="').removesuffix('"') + comparable = re.sub(r"\\+", "/", value.replace('\\"', "").replace('"', "")) + if comparable not in (base_value, normalized): + return + lines[lines.index(existing)] = launch_line + ppdb_path.write_text("\n".join(lines) + "\n", encoding="utf-8") + return + separator = "" if not current or current.endswith("\n") else "\n" + ppdb_path.write_text( + f"{current}{separator}{launch_line}\n", + encoding="utf-8", + ) + except OSError as error: + logger.warning("Failed to add GOG launch parameters to %s: %s", ppdb_path, error) + + def _get_primary_task(self, app_id: str) -> tuple[dict | None, Path | None]: + """Return the primary GOG launch task and installation path.""" + install_path = self.get_installed_path(app_id) + if install_path is None: + return None, None + info = self._load_json(install_path / f"goggame-{app_id}.info", {}) + tasks = info.get("playTasks", []) if isinstance(info, dict) else [] + primary = next((task for task in tasks if task.get("isPrimary")), None) + if primary is None and tasks: + primary = tasks[0] + if not isinstance(primary, dict) or primary.get("type") != "FileTask": + return None, install_path + return primary, install_path + + def _get_library_entries(self, user_id: str, token: str) -> list[dict]: + entries = [] + page_token = "" + headers = {"Authorization": f"Bearer {token}"} + while True: + url = f"https://galaxy-library.gog.com/users/{user_id}/releases" + response = requests.get(url, headers=headers, params={"page_token": page_token} if page_token else {}, timeout=15) + response.raise_for_status() + data = response.json() + entries.extend(data.get("items", [])) + page_token = data.get("next_page_token", "") + if not page_token: + return entries + + def _get_game(self, entry: dict, token: str) -> dict: + if entry.get("platform_id") != "gog" or not entry.get("external_id"): + return {} + app_id = str(entry["external_id"]) + headers = {"Authorization": f"Bearer {token}"} + if entry.get("certificate"): + headers["X-GOG-Library-Cert"] = str(entry["certificate"]) + response = requests.get( + f"https://gamesdb.gog.com/platforms/gog/external_releases/{app_id}", + headers=headers, timeout=15, + ) + response.raise_for_status() + data = response.json() + game = data.get("game", {}) + if not game.get("visible_in_library", True): + return {} + description = self._localized_value(data.get("summary")) + language = get_metadata_language().lower() + try: + product_response = requests.get( + f"https://api.gog.com/products/{app_id}", + params={ + "expand": "description", + "locale": GOG_PRODUCT_LOCALES.get(language, language), + }, + timeout=15, + ) + product_response.raise_for_status() + product_description = product_response.json().get("description", {}) + localized_lead = str(product_description.get("lead") or "") + if localized_lead: + paragraphs = re.split( + r"(?:\s*){2,}", localized_lead, flags=re.IGNORECASE + ) + paragraph_count = ( + 2 if re.match(r"\s*<(?:b|strong)>", paragraphs[0], re.IGNORECASE) + else 1 + ) + description = "

".join( + paragraph.strip() for paragraph in paragraphs[:paragraph_count] + ) + except requests.RequestException as error: + logger.warning( + "Failed to load localized GOG description for %s: %s", + app_id, error, + ) + return { + "app_id": app_id, + "title": ( + self._localized_value(data.get("title")) + or self._localized_value(game.get("title")) + or app_id + ).strip(), + "description": description, + "cover": self._format_image(game.get("vertical_cover") or game.get("logo") or game.get("background")), + } + + @staticmethod + def _localized_value(values: dict | None) -> str: + if not isinstance(values, dict): + return "" + language = get_metadata_language().lower() + for key in (language, language.replace("_", "-")): + if values.get(key): + return str(values[key]) + localized = next( + (value for key, value in values.items() if key.lower().startswith(language)), + None, + ) + return str(localized or values.get("*", "")) + + @staticmethod + def _format_image(image: dict | None) -> str: + template = image.get("url_format", "") if isinstance(image, dict) else "" + return template.replace("{formatter}", "").replace("{ext}", "jpg") + + @staticmethod + def parse_download_sizes(output: bytes) -> tuple[int, int]: + """Return download and disk sizes from gogdl info output.""" + data = orjson.loads(output) + sizes = data.get("size", {}) if isinstance(data, dict) else {} + languages = data.get("languages", []) if isinstance(data, dict) else [] + selected = [sizes.get("*", {})] + if languages: + selected.append(sizes.get(str(languages[0]), {})) + download_size = sum(int(item.get("download_size", 0)) for item in selected) + disk_size = sum(int(item.get("disk_size", 0)) for item in selected) + return download_size, disk_size + + @staticmethod + def extract_auth_code(url: str) -> str: + """Extract the OAuth code from GOG's redirect URL.""" + values = parse_qs(urlparse(url).query) + return values.get("code", [""])[0] + + @staticmethod + def _load_json(path: Path, fallback): + try: + return orjson.loads(path.read_bytes()) + except (OSError, orjson.JSONDecodeError): + return fallback + + @staticmethod + def _save_json(path: Path, data) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(orjson.dumps(data, option=orjson.OPT_INDENT_2)) diff --git a/portprotonqt/main_window.py b/portprotonqt/main_window.py index 7e6e2456..033540df 100644 --- a/portprotonqt/main_window.py +++ b/portprotonqt/main_window.py @@ -19,6 +19,7 @@ from portprotonqt.animations import DetailPageAnimations from portprotonqt.custom_widgets import ClickableLabel, AutoSizeButton, NavLabel from portprotonqt.detail_pages import DetailPageManager from portprotonqt.portproton_api import PortProtonAPI, remove_empty_custom_data_dirs +from portprotonqt.gog_api import GOGAPI from portprotonqt.debug_utils import get_prefix_name from portprotonqt.input_manager import InputManager, MainWindowProtocol from portprotonqt.context_menu_manager import ContextMenuManager @@ -59,6 +60,7 @@ from portprotonqt.sound_manager import SoundManager from portprotonqt.tabs.control_hints import MainWindowControlHintsMixin from portprotonqt.tabs.autoinstall_tab import MainWindowAutoInstallTabMixin from portprotonqt.tabs.library_tab import MainWindowLibraryTabMixin +from portprotonqt.tabs.gog_tab import MainWindowGOGTabMixin from portprotonqt.tabs.settings_tab import MainWindowSettingsTabMixin from portprotonqt.tabs.system_tab import MainWindowSystemTabMixin from portprotonqt.tabs.theme_tab import MainWindowThemeTabMixin @@ -134,6 +136,7 @@ class MainWindow( MainWindowControlHintsMixin, MainWindowAutoInstallTabMixin, MainWindowLibraryTabMixin, + MainWindowGOGTabMixin, MainWindowSettingsTabMixin, MainWindowSystemTabMixin, MainWindowThemeTabMixin, @@ -358,6 +361,7 @@ class MainWindow( # Start watching dist and prefixes directories if they exist QTimer.singleShot(0, self.start_watching_directories) # Delay to ensure portproton_location is set self.downloader = Downloader(max_workers=4) + self.gog_api = GOGAPI() self.portproton_api = PortProtonAPI(self.downloader) self.autoInstallCustomDataThread = None @@ -423,7 +427,8 @@ class MainWindow( _("Wine Settings"), _("PPQT Settings"), _("System"), - _("Themes") + _("Themes"), + _("Downloads") ] for i, tabName in enumerate(tabs): btn = NavLabel(tabName) @@ -489,6 +494,7 @@ class MainWindow( if hidden_system_page: hidden_system_page.setVisible(False) self.createThemeTab() + self.createGOGDownloadsTab() self.controlHintsWidget = self.createControlHintsWidget() self.controlHintsWidget.setStyleSheet(self.theme.HINT_BAR_STYLE) @@ -931,21 +937,23 @@ class MainWindow( def start_loading(): if display_filter == "steam": self._load_steam_games_async(lambda games: self.games_loaded.emit(games)) + elif display_filter == "gog": + self._load_gog_games_async(lambda games: self.games_loaded.emit(games)) elif display_filter == "portproton": self._load_portproton_games_async(lambda games: self.games_loaded.emit(games)) elif display_filter == "favorites": - def on_all_games_favorites(portproton_games, steam_games): - games = [game for game in portproton_games + steam_games if game[0] in favorites] + def on_all_games_favorites(portproton_games, steam_games, gog_games): + games = [game for game in portproton_games + steam_games + gog_games if game[0] in favorites] self.games_loaded.emit(games) # Load games from different sources in parallel to prevent blocking - results = {'portproton': [], 'steam': []} - completed = {'portproton': False, 'steam': False} + results = {'portproton': [], 'steam': [], 'gog': []} + completed = {'portproton': False, 'steam': False, 'gog': False} def check_completion(): if all(completed.values()): QApplication.processEvents() # Keep UI responsive - on_all_games_favorites(results['portproton'], results['steam']) + on_all_games_favorites(results['portproton'], results['steam'], results['gog']) def portproton_callback(games): results['portproton'] = games @@ -959,17 +967,23 @@ class MainWindow( QApplication.processEvents() # Keep UI responsive check_completion() + def gog_callback(games): + results['gog'] = games + completed['gog'] = True + check_completion() + self._load_portproton_games_async(portproton_callback) self._load_steam_games_async(steam_callback) + self._load_gog_games_async(gog_callback) else: # For 'all' filter - load games from different sources in parallel to prevent blocking - results = {'portproton': [], 'steam': []} - completed = {'portproton': False, 'steam': False} + results = {'portproton': [], 'steam': [], 'gog': []} + completed = {'portproton': False, 'steam': False, 'gog': False} def on_all_games(): seen = set() games = [] - for game in results['portproton'] + results['steam']: + for game in results['portproton'] + results['steam'] + results['gog']: # Unique key: name + exec_line key = (game[0], game[5]) if key not in seen: @@ -995,13 +1009,48 @@ class MainWindow( QApplication.processEvents() # Keep UI responsive check_completion() + def gog_callback(games): + results['gog'] = games + completed['gog'] = True + check_completion() + # Load all sources in parallel self._load_portproton_games_async(portproton_callback) self._load_steam_games_async(steam_callback) + self._load_gog_games_async(gog_callback) # Run loading immediately to show status without delay start_loading() + def _load_gog_games_async(self, callback: Callable[[list[tuple]], None]) -> None: + api = getattr(self, "gog_api", None) or GOGAPI() + installed = api.load_installed() + games = [] + for game in api.load_library(): + app_id = str(game.get("app_id", "")) + if not app_id: + continue + action = "launch" if api.is_game_installed(app_id, installed) else "install" + games.append(( + str(game.get("title", app_id)), + str(game.get("description", "")), + str(game.get("cover", "")), + app_id, + "", + f"gog://{action}/{app_id}", + _("Never"), + format_playtime(0), + "", + "", + 0, + 0, + "gog", + "", + "", + "", + )) + callback(games) + def _load_steam_games_async(self, callback: Callable[[list[tuple]], None]): steam_games = [] installed_games = get_steam_installed_games() @@ -1461,6 +1510,15 @@ class MainWindow( def openGameDetailPage(self, game_data: dict) -> None: """Open game detail page.""" SoundManager().play("open") + exec_line = str(game_data.get("exec_line", "")) + if str(game_data.get("game_source", "")).lower() == "gog": + app_id = str(game_data.get("appid", "")) + if ( + exec_line.startswith("gog://install/") + and self.gog_api.is_game_installed(app_id) + ): + game_data = dict(game_data) + game_data["exec_line"] = f"gog://launch/{app_id}" logger.debug( "Opening detail page for %s from stacked index %d", game_data.get("name", ""), @@ -2259,12 +2317,72 @@ class MainWindow( if not QDesktopServices.openUrl(QUrl(exec_line)): logger.warning("Failed to open Steam URI: %s", exec_line) + def _handle_gog_game(self, exec_line: str, button=None) -> None: + action, app_id = exec_line.removeprefix("gog://").split("/", 1) + if action == "install" and self.gog_api.is_game_installed(app_id): + action = "launch" + if action == "launch": + self._launch_gog_game(app_id, button) + return + game = next( + (item for item in self.gog_api.load_library() + if str(item.get("app_id", "")) == app_id), + None, + ) + if game: + self._install_gog_game(game) + + def _launch_gog_game(self, app_id: str, button=None) -> None: + self.gog_api.ensure_launch_parameters(app_id) + discovered_path = self.gog_api.get_installed_path(app_id) + install_path = str(discovered_path) if discovered_path else "" + if not install_path or not self.start_sh: + QMessageBox.warning(self, _("Error"), _("GOG game or PortProton is unavailable")) + return + target = os.path.basename(self.gog_api.get_launch_target(app_id) or app_id) + if self.game_processes and self.target_exe == target: + self.stop_running_game(button) + return + if self.game_processes: + QMessageBox.warning(self, _("Error"), _("Cannot launch game while another game is running")) + return + try: + command = self.gog_api.build_command([ + "launch", install_path, app_id, "--platform", "windows", + "--wine", self.start_sh[0], + ]) + process = subprocess.Popen( + command, env=self.gog_api.get_environment(), shell=False, + preexec_fn=os.setsid, stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, text=True, errors="replace", + ) + except (OSError, ValueError) as error: + logger.error("Failed to launch GOG game %s: %s", app_id, error) + QMessageBox.warning(self, _("Error"), str(error)) + return + SoundManager().play("game_launch") + self.game_processes.append(process) + self.target_exe = target + self.current_running_button = button + self._start_launch_output_reader(process) + self.input_manager.suspend_gamepad_polling() + self.checkProcessTimer = QTimer(self) + self.checkProcessTimer.timeout.connect(self.checkTargetExe) + self.checkProcessTimer.start(500) + if button is not None: + button.setText(_("Stop")) + button.setIcon(self.theme_manager.get_icon("stop", as_path=True)) + def toggleGame(self, exec_line, button=None, game_name=None): # Handle Steam games if exec_line.startswith("steam://"): self._launch_steam_game(exec_line) return + if exec_line.startswith("gog://"): + self._handle_gog_game(exec_line, button) + return + # Handle PortProton games entry_exec_split = shlex.split(exec_line) if not entry_exec_split: diff --git a/portprotonqt/meson.build b/portprotonqt/meson.build index cb723ac6..5349b5d9 100644 --- a/portprotonqt/meson.build +++ b/portprotonqt/meson.build @@ -12,6 +12,7 @@ install_data( 'game_card.py', 'gamepad_backend.py', 'game_library_manager.py', + 'gog_api.py', 'howlongtobeat_api.py', 'image_utils.py', 'icon_extractor.py', @@ -110,6 +111,7 @@ install_data( ['tabs/__init__.py', 'tabs/autoinstall_tab.py', 'tabs/control_hints.py', + 'tabs/gog_tab.py', 'tabs/library_tab.py', 'tabs/settings_tab.py', 'tabs/system_tab.py', diff --git a/portprotonqt/tabs/gog_tab.py b/portprotonqt/tabs/gog_tab.py new file mode 100644 index 00000000..155f9047 --- /dev/null +++ b/portprotonqt/tabs/gog_tab.py @@ -0,0 +1,573 @@ +"""Shared download manager with GOG download support.""" + +import os +import re +import shutil +from datetime import datetime +from pathlib import Path + +from PySide6.QtCore import QProcess, QProcessEnvironment, QThread, QTimer, QUrl, Signal +from PySide6.QtGui import QDesktopServices +from PySide6.QtWidgets import ( + QAbstractItemView, + QFrame, + QHeaderView, + QHBoxLayout, + QLabel, + QMessageBox, + QProgressBar, + QPushButton, + QTableWidget, + QTableWidgetItem, + QVBoxLayout, + QWidget, +) + +from portprotonqt.dialogs.file_explorer import FileExplorer +from portprotonqt.gog_api import GOGAPI, GOG_LOGIN_URL +from portprotonqt.image_utils import load_pixmap_async +from portprotonqt.localization import _ +from portprotonqt.logger import get_logger + +logger = get_logger(__name__) +GOG_CANCEL_KILL_TIMEOUT_MS = 3000 + + +class GOGLibraryWorker(QThread): + """Refresh the GOG library outside the UI thread.""" + + loaded = Signal(list) + failed = Signal(str) + progress = Signal(int, int) + + def __init__(self, api: GOGAPI) -> None: + super().__init__() + self.api = api + + def run(self) -> None: + try: + self.loaded.emit(self.api.refresh_library(self.progress.emit)) + except Exception as error: + logger.exception("Failed to refresh GOG library") + self.failed.emit(str(error)) + + +class GOGAuthWorker(QThread): + """Exchange the GOG OAuth code outside the UI thread.""" + + authenticated = Signal(bool, str) + + def __init__(self, api: GOGAPI, code: str) -> None: + super().__init__() + self.api = api + self.code = code + + def run(self) -> None: + try: + authenticated, error = self.api.authenticate(self.code) + self.authenticated.emit(authenticated, error) + except Exception as error: + logger.exception("Failed to authenticate with GOG: %s", error) + self.authenticated.emit(False, str(error)) + + +class MainWindowGOGTabMixin: + """Add account actions and the shared downloads page.""" + + def createGOGDownloadsTab(self) -> None: + self.gog_process = None + self.gog_download_queue = [] + self.gog_download_output = "" + self.downloadTableHeadings = {} + page = QWidget() + page.setStyleSheet(self.theme.OTHER_PAGES_WIDGET_STYLE) + layout = QVBoxLayout(page) + layout.setSpacing(self.theme.downloadsSectionSpacing) + title = QLabel(_("Downloads")) + title.setStyleSheet(self.theme.TAB_TITLE_STYLE) + layout.addWidget(title) + self._create_active_download_card(layout) + self.downloadQueuedTable = self._create_download_table(_("Queued"), layout) + completed_header = QHBoxLayout() + completed_header.addWidget(QLabel(_("Completed"))) + clear_button = QPushButton(_("Clear List")) + clear_button.setStyleSheet(self.theme.ACTION_BUTTON_STYLE) + clear_button.clicked.connect(self._clear_completed_downloads) + completed_header.addWidget(clear_button) + completed_header.addStretch() + layout.addLayout(completed_header) + self.downloadCompletedTable = self._create_download_table("", layout) + layout.addStretch() + self.stackedWidget.addWidget(page) + + def _create_active_download_card(self, layout: QVBoxLayout) -> None: + self.downloadActiveHeading = QLabel(_("Now Downloading")) + self.downloadActiveHeading.setStyleSheet(self.theme.SETTINGS_TITLE_STYLE) + layout.addWidget(self.downloadActiveHeading) + self.downloadActiveCard = QFrame() + self.downloadActiveCard.setObjectName("downloadsActiveCard") + self.downloadActiveCard.setStyleSheet(self.theme.DOWNLOADS_ACTIVE_STYLE) + self.downloadActiveCard.setFixedHeight(self.theme.downloadsActiveCardHeight) + card_layout = QVBoxLayout(self.downloadActiveCard) + card_layout.setContentsMargins(*self.theme.downloadsActiveCardMargins) + card_layout.setSpacing(self.theme.downloadsActiveCardSpacing) + content = QHBoxLayout() + content.setSpacing(self.theme.downloadsActiveCardSpacing) + self.downloadActiveCover = QLabel() + self.downloadActiveCover.setFixedSize(*self.theme.downloadsActiveCoverSize) + content.addWidget(self.downloadActiveCover) + info = QVBoxLayout() + header = QHBoxLayout() + self.downloadActiveTitle = QLabel() + self.downloadActiveTitle.setStyleSheet(self.theme.SETTINGS_TITLE_STYLE) + header.addWidget(self.downloadActiveTitle) + header.addStretch() + self.downloadCancelButton = QPushButton(_("Cancel")) + self.downloadCancelButton.setStyleSheet(self.theme.ACTION_BUTTON_STYLE) + self.downloadCancelButton.clicked.connect(self._cancel_gog_download) + header.addWidget(self.downloadCancelButton) + info.addLayout(header) + self.downloadActiveDetails = QLabel(_("Waiting…")) + self.downloadActiveDetails.setStyleSheet(self.theme.CONTENT_STYLE) + info.addWidget(self.downloadActiveDetails) + metrics = QHBoxLayout() + self.downloadSpeedLabel = QLabel(_("Download: —")) + self.diskSpeedLabel = QLabel(_("Disk: —")) + metrics.addWidget(self.downloadSpeedLabel) + metrics.addWidget(self.diskSpeedLabel) + metrics.addStretch() + info.addLayout(metrics) + content.addLayout(info, stretch=1) + card_layout.addLayout(content) + self.downloadOverallProgress = QProgressBar() + self.downloadOverallProgress.setStyleSheet(self.theme.PROGRESS_BAR_STYLE) + card_layout.addWidget(self.downloadOverallProgress) + layout.addWidget(self.downloadActiveCard) + self.downloadActiveHeading.setVisible(False) + self.downloadActiveCard.setVisible(False) + + def _create_download_table(self, heading: str, layout: QVBoxLayout) -> QTableWidget: + if heading: + label = QLabel(heading) + label.setStyleSheet(self.theme.SETTINGS_TITLE_STYLE) + layout.addWidget(label) + columns = (_("Game Title"), _("Started at"), _("Type"), _("Store")) + table = QTableWidget(0, len(columns)) + table.setHorizontalHeaderLabels(columns) + table.setStyleSheet(self.theme.DOWNLOADS_TABLE_STYLE) + table.setAlternatingRowColors(True) + table.setEditTriggers(QAbstractItemView.EditTrigger.NoEditTriggers) + table.setSelectionMode(QAbstractItemView.SelectionMode.NoSelection) + table.setShowGrid(False) + header = table.horizontalHeader() + header.setFixedHeight(self.theme.downloadsTableHeaderHeight) + header.setSectionResizeMode(0, QHeaderView.ResizeMode.Stretch) + for column in range(1, len(columns)): + header.setSectionResizeMode(column, QHeaderView.ResizeMode.ResizeToContents) + table.verticalHeader().setVisible(False) + self._update_download_table_height(table) + layout.addWidget(table) + if heading: + self.downloadTableHeadings[table] = label + label.setVisible(False) + table.setVisible(False) + return table + + def _update_download_table_height(self, table: QTableWidget) -> None: + rows = max(1, table.rowCount()) + height = self.theme.downloadsTableHeaderHeight + rows * self.theme.downloadsTableRowHeight + table.setFixedHeight(height) + has_rows = table.rowCount() > 0 + table.setVisible(has_rows) + heading = self.downloadTableHeadings.get(table) + if heading is not None: + heading.setVisible(has_rows) + + def _clear_completed_downloads(self) -> None: + self.downloadCompletedTable.setRowCount(0) + self._update_download_table_height(self.downloadCompletedTable) + + def _start_gog_login(self) -> None: + QDesktopServices.openUrl(QUrl(GOG_LOGIN_URL)) + self.gogLoginUrlEdit.setFocus() + + def _submit_gog_login_url(self) -> None: + code = self.gog_api.extract_auth_code(self.gogLoginUrlEdit.text().strip()) + if not code: + self.gogAccountStatus.setText(_("The GOG login URL does not contain a code")) + return + if getattr(self, "gog_auth_worker", None) is not None: + return + worker = GOGAuthWorker(self.gog_api, code) + worker.authenticated.connect(self._on_gog_authenticated) + worker.finished.connect(self._on_gog_auth_worker_finished) + self.gog_auth_worker = worker + self.gogSubmitLoginButton.setEnabled(False) + worker.start() + + def _on_gog_authenticated(self, authenticated: bool, error: str) -> None: + if not authenticated: + message = error or _("Unknown error") + self.gogAccountStatus.setText(_("GOG login failed: {0}").format(message)) + return + self.gogAccountStatus.setText(_("GOG account connected")) + self._refresh_gog_library() + + def _on_gog_auth_worker_finished(self) -> None: + self.gogSubmitLoginButton.setEnabled(True) + self.gog_auth_worker = None + + def _update_gog_account_state(self) -> None: + connected = self.gog_api.auth_path.is_file() + status = ( + _("GOG account connected") + if connected else _("GOG account not connected") + ) + self.gogAccountStatus.setText(status) + self.gogRefreshButton.setEnabled(connected) + self.gogLoginButton.setEnabled(True) + + def _refresh_gog_library(self) -> None: + if getattr(self, "gog_library_worker", None) is not None: + return + self.gogRefreshButton.setEnabled(False) + self.gogAccountStatus.setText(_("Refreshing GOG library…")) + worker = GOGLibraryWorker(self.gog_api) + worker.loaded.connect(self._on_gog_library_loaded) + worker.failed.connect(self._on_gog_library_failed) + worker.progress.connect(self._on_gog_library_progress) + worker.finished.connect(self._on_gog_library_worker_finished) + self.gog_library_worker = worker + worker.start() + + def _on_gog_library_progress(self, completed: int, total: int) -> None: + self.gogAccountStatus.setText( + _("Refreshing GOG library… {0}/{1}").format(completed, total) + ) + + def _on_gog_library_loaded(self, _games: list) -> None: + self.gogAccountStatus.setText(_("GOG library updated")) + self.loadGames(force_load=True) + + def _on_gog_library_failed(self, message: str) -> None: + self.gogAccountStatus.setText(_("Failed to refresh GOG library: {0}").format(message)) + + def _on_gog_library_worker_finished(self) -> None: + self.gog_library_worker = None + self.gogRefreshButton.setEnabled(True) + + def _install_gog_game(self, game: dict) -> None: + if self.gog_process is not None: + self.gog_download_queue.append(game) + self._append_download_row(self.downloadQueuedTable, game, _("Install")) + self.switchTab(6) + return + app_id = str(game["app_id"]) + install_path = self.gog_api.get_install_path(app_id, str(game["title"])) + install_path.mkdir(parents=True, exist_ok=True) + selected_paths: list[str] = [] + file_explorer = FileExplorer( + self, + theme=self.theme, + initial_path=str(install_path), + directory_only=True, + ) + file_explorer.setWindowTitle(_("Select installation folder")) + file_explorer.file_signal.file_selected.connect(selected_paths.append) + file_explorer.exec() + if not selected_paths: + return + install_path = Path(selected_paths[0]) + manifest_path = ( + self.gog_api.config_dir / "heroic_gogdl" / "manifests" / app_id + ) + if manifest_path.is_file() and not self.gog_api.is_game_installed(app_id): + try: + manifest_path.unlink() + except OSError as error: + logger.error("Failed to reset incomplete GOG download: %s", error) + self.gogAccountStatus.setText(str(error)) + return + try: + command = self.gog_api.build_command( + ["download", app_id, "--path", str(install_path), "--platform", "windows"] + ) + except FileNotFoundError as error: + self.gogAccountStatus.setText(str(error)) + return + self._start_gog_download(game, install_path, command, _("Install")) + + def _repair_gog_game(self, game: dict) -> None: + if self.gog_process is not None: + self.gogAccountStatus.setText(_("Another GOG operation is already running")) + return + app_id = str(game["app_id"]) + install_path = self.gog_api.get_installed_path(app_id) + if install_path is None: + self.gogAccountStatus.setText(_("GOG installation not found")) + return + try: + command = self.gog_api.build_command( + ["repair", app_id, "--path", str(install_path), "--platform", "windows"] + ) + except FileNotFoundError as error: + self.gogAccountStatus.setText(str(error)) + return + self._start_gog_download(game, install_path, command, _("Repair")) + + def _import_gog_game(self, game: dict) -> None: + selected_paths: list[str] = [] + file_explorer = FileExplorer( + self, theme=self.theme, initial_path=str(Path.home() / "Games"), + directory_only=True, + ) + file_explorer.setWindowTitle(_("Select GOG game folder")) + file_explorer.file_signal.file_selected.connect(selected_paths.append) + file_explorer.exec() + if not selected_paths: + return + app_id = str(game["app_id"]) + game_path = self.gog_api.find_install_path(app_id, Path(selected_paths[0])) + if game_path is None: + self.gogAccountStatus.setText(_("Selected folder is not this GOG game")) + return + self.gog_api.save_installed_game( + app_id, {"install_path": str(game_path), "title": str(game["title"])} + ) + self.gog_api.ensure_launch_parameters(app_id) + self.gogAccountStatus.setText(_("GOG game imported")) + self.loadGames(force_load=True) + + def _delete_gog_game(self, game: dict) -> None: + app_id = str(game["app_id"]) + install_path = self.gog_api.get_installed_path(app_id) + if install_path is None: + self.gogAccountStatus.setText(_("GOG installation not found")) + return + message_box = QMessageBox(self) + message_box.setIcon(QMessageBox.Icon.Question) + message_box.setWindowTitle(_("Confirm Deletion")) + message_box.setText( + _("Delete '{0}' and all files in its installation folder?").format( + game["title"] + ) + ) + message_box.setStandardButtons( + QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No + ) + message_box.setDefaultButton(QMessageBox.StandardButton.No) + if message_box.exec() != QMessageBox.StandardButton.Yes: + return + try: + shutil.rmtree(install_path) + except OSError as error: + logger.error("Failed to delete GOG game %s: %s", app_id, error) + self.gogAccountStatus.setText(str(error)) + return + manifest_path = ( + self.gog_api.config_dir / "heroic_gogdl" / "manifests" / app_id + ) + try: + manifest_path.unlink(missing_ok=True) + except OSError as error: + logger.warning("Failed to delete GOG manifest %s: %s", app_id, error) + self.gog_api.remove_installed_game(app_id) + self.context_menu_manager.remove_gog_shortcuts(str(game["title"])) + self.gogAccountStatus.setText(_("GOG game deleted")) + self.loadGames(force_load=True) + + def _start_gog_download( + self, game: dict, install_path: Path, command: list[str], action: str + ) -> None: + app_id = str(game["app_id"]) + self.downloadActiveTitle.setText(str(game["title"])) + self.downloadActiveDetails.setText(_("Starting")) + cover_width, cover_height = self.theme.downloadsActiveCoverSize + load_pixmap_async( + str(game.get("cover", "")), cover_width, cover_height, + self.downloadActiveCover.setPixmap, + app_name=f"download-active-{app_id}", + ) + self.downloadActiveHeading.setVisible(True) + self.downloadActiveCard.setVisible(True) + process = QProcess(self) + process.setProgram(command[0]) + process.setArguments(command[1:]) + process.setProcessEnvironment(self._gog_process_environment()) + process.setProcessChannelMode(QProcess.ProcessChannelMode.MergedChannels) + process.readyReadStandardOutput.connect(self._read_gog_download_output) + process.finished.connect(self._on_gog_download_finished) + self.gog_process = process + self.gog_active_app_id = app_id + self.gog_active_game = game + self.gog_active_install_path = install_path + self.gog_active_action = action + self.gog_download_output = "" + self.downloadOverallProgress.setValue(0) + self.downloadCancelButton.setEnabled(True) + self.switchTab(6) + process.start() + + def _cancel_gog_download(self) -> None: + if self.gog_process is None: + return + process = self.gog_process + self.downloadCancelButton.setEnabled(False) + self.downloadActiveDetails.setText(_("Cancel")) + process.terminate() + QTimer.singleShot( + GOG_CANCEL_KILL_TIMEOUT_MS, + lambda: self._kill_gog_process(process), + ) + + def _kill_gog_process(self, process: QProcess) -> None: + if process.state() != QProcess.ProcessState.NotRunning: + logger.warning("gogdl did not terminate after cancellation; killing it") + process.kill() + + def _append_download_row( + self, table: QTableWidget, game: dict, action: str + ) -> tuple[int, QLabel]: + row = table.rowCount() + table.insertRow(row) + game_cell, details_label = self._create_download_game_cell(game) + table.setCellWidget(row, 0, game_cell) + values = (datetime.now().strftime("%H:%M:%S"), action, "GOG") + for column, value in enumerate(values, 1): + table.setItem(row, column, QTableWidgetItem(value)) + table.setRowHeight(row, self.theme.downloadsTableRowHeight) + if table is self.downloadQueuedTable: + details_label.setText(_("Queued")) + else: + details_label.setText(action) + self._update_download_table_height(table) + return row, details_label + + def _create_download_game_cell(self, game: dict) -> tuple[QWidget, QLabel]: + widget = QWidget() + layout = QHBoxLayout(widget) + layout.setContentsMargins(*self.theme.downloadsCellMargins) + layout.setSpacing(self.theme.downloadsCellSpacing) + cover = QLabel() + cover_width, cover_height = self.theme.downloadsCoverSize + cover.setFixedSize(cover_width, cover_height) + layout.addWidget(cover) + text_layout = QVBoxLayout() + title = QLabel(str(game["title"])) + details = QLabel(_("Waiting…")) + details.setStyleSheet(self.theme.CONTENT_STYLE) + text_layout.addWidget(title) + text_layout.addWidget(details) + layout.addLayout(text_layout) + load_pixmap_async( + str(game.get("cover", "")), cover_width, cover_height, cover.setPixmap, + app_name=f"download-{game['app_id']}", + ) + return widget, details + + def _gog_process_environment(self) -> QProcessEnvironment: + environment = QProcessEnvironment.systemEnvironment() + environment.insert("GOGDL_CONFIG_PATH", str(self.gog_api.config_dir)) + return environment + + def _read_gog_download_output(self) -> None: + if self.gog_process is None: + return + output = bytes(self.gog_process.readAllStandardOutput()).decode(errors="replace") + self.gog_download_output = (self.gog_download_output + output)[-4096:] + percentages = re.findall( + r"Progress:\s+(\d{1,3}(?:\.\d+)?)", self.gog_download_output + ) + if percentages: + self.downloadOverallProgress.setValue(int(float(percentages[-1]))) + download_speeds = re.findall( + r"Download\t-\s+(\S+)\s+MiB", self.gog_download_output + ) + disk_speeds = re.findall( + r"Disk\t-\s+(\S+)\s+MiB", self.gog_download_output + ) + downloaded = re.findall( + r"Downloaded:\s+([\d.]+\s+MiB)", self.gog_download_output + ) + eta = re.findall(r"ETA:\s+(\d{2}:\d{2}:\d{2})", self.gog_download_output) + if download_speeds: + self.downloadSpeedLabel.setText( + _("Download: {0} MiB/s").format(download_speeds[-1]) + ) + if disk_speeds: + self.diskSpeedLabel.setText( + _("Disk: {0} MiB/s").format(disk_speeds[-1]) + ) + details = [] + if downloaded: + details.append(downloaded[-1]) + if download_speeds: + details.append(f"{download_speeds[-1]} MiB/s") + if eta: + details.append(_("ETA: {0}").format(eta[-1])) + if details: + self._update_active_download_details(details) + + def _update_active_download_details(self, details: list[str]) -> None: + self.downloadActiveDetails.setText(" · ".join(details)) + + def _on_gog_download_finished( + self, code: int, _status: QProcess.ExitStatus + ) -> None: + game = self.gog_active_game + install_path = self.gog_active_install_path + app_id = str(game["app_id"]) + self.downloadActiveHeading.setVisible(False) + self.downloadActiveCard.setVisible(False) + game_path = self.gog_api.find_install_path(app_id, install_path) + if code == 0 and game_path is None: + code = 1 + completed_action = _("Failed") + if code == 0: + completed_action = ( + _("Repaired") if self.gog_active_action == _("Repair") + else _("Installed") + ) + _completed_row, completed_details = self._append_download_row( + self.downloadCompletedTable, game, completed_action, + ) + if code == 0: + self.gog_api.save_installed_game( + app_id, {"install_path": str(game_path), "title": game["title"]} + ) + self._update_installed_gog_detail(app_id) + self.loadGames(force_load=True) + else: + error = self._get_gog_download_error() + completed_details.setText(error) + logger.error("GOG download failed for %s: %s", app_id, error) + self.gog_process = None + self.downloadOverallProgress.setValue(0) + self.downloadSpeedLabel.setText(_("Download: —")) + self.diskSpeedLabel.setText(_("Disk: —")) + if self.gog_download_queue: + next_game = self.gog_download_queue.pop(0) + self.downloadQueuedTable.removeRow(0) + self._update_download_table_height(self.downloadQueuedTable) + self._install_gog_game(next_game) + + def _get_gog_download_error(self) -> str: + lines = [line.strip() for line in self.gog_download_output.splitlines()] + error_lines = [line for line in lines if "ERROR" in line.upper()] + if error_lines: + return error_lines[-1] + useful = [line for line in lines if line and "[PROGRESS]" not in line] + return useful[-1] if useful else _("Download process failed") + + def _update_installed_gog_detail(self, app_id: str) -> None: + install_uri = f"gog://install/{app_id}" + if self.current_exec_line != install_uri: + return + if self.stackedWidget.currentWidget() is not self.currentDetailPage: + return + source = self.detail_page_manager._current_detail_source + if source is None or source[0] != "game": + return + source[1]["exec_line"] = f"gog://launch/{app_id}" + self.current_exec_line = source[1]["exec_line"] + self.detail_page_manager._reopen_current_detail_page() diff --git a/portprotonqt/tabs/library_tab.py b/portprotonqt/tabs/library_tab.py index b6c38ab2..61c4f7a1 100644 --- a/portprotonqt/tabs/library_tab.py +++ b/portprotonqt/tabs/library_tab.py @@ -277,8 +277,8 @@ class MainWindowLibraryTabMixin(_MainWindowTypingBase): self.gamesSortCombo.activated.connect(self._delay_library_controls_hover_close) controls_layout.addWidget(self.gamesSortCombo) - self.filter_keys = ["all", "steam", "portproton", "favorites"] - self.filter_labels = [_("All"), "Steam", "PortProton", _("Favorites")] + self.filter_keys = ["all", "steam", "gog", "portproton", "favorites"] + self.filter_labels = [_("All"), "Steam", "GOG", "PortProton", _("Favorites")] self.gamesDisplayCombo = self._create_library_combo(self.filter_labels, _("Display Filter:")) self._set_combo_current_key( self.gamesDisplayCombo, @@ -384,7 +384,7 @@ class MainWindowLibraryTabMixin(_MainWindowTypingBase): def _get_games_without_exe(self) -> list[tuple]: missing_games = [] for game in self.game_library_manager.games: - if len(game) <= 12 or game[12] == "steam": + if len(game) <= 12 or game[12] != "portproton": continue exec_line = game[5] if len(game) > 5 else "" exe_path = extract_exec_target_path(exec_line) if exec_line else None diff --git a/portprotonqt/tabs/settings_tab.py b/portprotonqt/tabs/settings_tab.py index afe18555..df35e8d2 100644 --- a/portprotonqt/tabs/settings_tab.py +++ b/portprotonqt/tabs/settings_tab.py @@ -119,6 +119,32 @@ class MainWindowSettingsTabMixin(_MainWindowTypingBase): genForm.setRowWrapPolicy(QFormLayout.RowWrapPolicy.DontWrapRows) scrollLayout.addWidget(genFrame) + gogFrame, gogForm = create_section(_("GOG Account"), self.theme) + gogForm.setRowWrapPolicy(QFormLayout.RowWrapPolicy.DontWrapRows) + scrollLayout.addWidget(gogFrame) + self.gogAccountStatus = QLabel() + self.gogAccountStatus.setStyleSheet(self.theme.SETTINGS_TITLE_STYLE) + self.gogLoginUrlEdit = CustomLineEdit(theme=self.theme) + self.gogLoginUrlEdit.setPlaceholderText(_("Paste the final GOG login URL")) + self.gogLoginUrlEdit.setStyleSheet(self.theme.LINE_EDIT_STYLE) + self.gogLoginButton = AutoSizeButton(_("Open login page")) + self.gogLoginButton.setStyleSheet(self.theme.ACTION_BUTTON_STYLE) + self.gogLoginButton.clicked.connect(self._start_gog_login) + self.gogSubmitLoginButton = AutoSizeButton(_("Confirm URL")) + self.gogSubmitLoginButton.setStyleSheet(self.theme.ACTION_BUTTON_STYLE) + self.gogSubmitLoginButton.clicked.connect(self._submit_gog_login_url) + self.gogRefreshButton = AutoSizeButton(_("Refresh library")) + self.gogRefreshButton.setStyleSheet(self.theme.ACTION_BUTTON_STYLE) + self.gogRefreshButton.clicked.connect(self._refresh_gog_library) + gog_buttons = QHBoxLayout() + gog_buttons.addWidget(self.gogLoginButton) + gog_buttons.addWidget(self.gogSubmitLoginButton) + gog_buttons.addWidget(self.gogRefreshButton) + gog_buttons.addStretch() + gogForm.addRow(self.gogAccountStatus, gog_buttons) + gogForm.addRow(self.gogLoginUrlEdit) + self._update_gog_account_state() + self.timeDetailCombo = CustomComboBox(theme=self.theme) self.timeDetailCombo.view().window().setWindowFlags( Qt.WindowType.Popup | Qt.WindowType.FramelessWindowHint diff --git a/portprotonqt/themes/standart/styles/base.py b/portprotonqt/themes/standart/styles/base.py index d1d6b0a2..8c972ef3 100644 --- a/portprotonqt/themes/standart/styles/base.py +++ b/portprotonqt/themes/standart/styles/base.py @@ -46,6 +46,55 @@ PROGRESS_BAR_STYLE = f""" }} """ +DOWNLOADS_SUMMARY_STYLE = f""" + QFrame#downloadsSummary {{ + background: {color_surface_elevated}; + border: {border_thin} {color_border_faint}; + border-radius: {border_radius_small}; + }} +""" + +DOWNLOADS_ACTIVE_STYLE = f""" + QFrame#downloadsActiveCard {{ + background: {color_surface_elevated}; + border: {border_thin} {color_border_faint}; + border-radius: {border_radius_large}; + }} +""" + +DOWNLOADS_SECTION_STYLE = f""" + QFrame#downloadsSection {{ + background: {color_surface_elevated}; + border: {border_thin} {color_border_faint}; + border-radius: {border_radius_small}; + }} +""" + +DOWNLOADS_TABLE_STYLE = f""" + QTableWidget {{ + background: {color_surface_elevated}; + alternate-background-color: {color_surface}; + border: {border_none}; + color: {color_text}; + gridline-color: {color_border_faint}; + font-family: '{font_family}'; + font-size: {font_size_value}; + }} + QHeaderView::section {{ + background: {color_surface}; + color: {color_text}; + border: {border_none}; + padding: 8px; + font-family: '{font_family}'; + font-size: {font_size_value}; + font-weight: 600; + }} + QTableWidget::item {{ + border-bottom: {border_thin} {color_border_faint}; + padding: 8px; + }} +""" + # STATUS BAR STYLE STATUS_BAR_STYLE = f""" QStatusBar {{ diff --git a/portprotonqt/themes/standart/styles/constants.py b/portprotonqt/themes/standart/styles/constants.py index cebb2595..6c5403ea 100644 --- a/portprotonqt/themes/standart/styles/constants.py +++ b/portprotonqt/themes/standart/styles/constants.py @@ -22,6 +22,17 @@ wineSettingsSetSpacing = 10 themeStorePageSpacing = 10 themeStoreCardDefaultWidth = 280 themeStoreDetailCarouselMinHeight = 300 +downloadsSummaryHeight = 96 +downloadsTableHeaderHeight = 38 +downloadsTableRowHeight = 68 +downloadsCoverSize = (72, 48) +downloadsSectionSpacing = 14 +downloadsCellMargins = (8, 6, 8, 6) +downloadsCellSpacing = 10 +downloadsActiveCardHeight = 190 +downloadsActiveCoverSize = (180, 120) +downloadsActiveCardMargins = (16, 14, 16, 14) +downloadsActiveCardSpacing = 14 mangoHudSwitchesColumns = 4 mangoHudSwitchesVerticalSpacing = 10 mangoHudFpsColumns = 4 diff --git a/tests/test_detail_pages.py b/tests/test_detail_pages.py index ecffc66e..336b5149 100644 --- a/tests/test_detail_pages.py +++ b/tests/test_detail_pages.py @@ -36,6 +36,22 @@ def test_detail_page_exe_fallback_uses_image_cache( ) +def test_gog_size_result_ignores_deleted_labels(monkeypatch: MonkeyPatch) -> None: + label = MagicMock() + monkeypatch.setattr("portprotonqt.detail_pages.isValid", lambda _label: True) + + DetailPageManager._set_gog_size_values( + [lambda: None, lambda: None], (100, 200) + ) + DetailPageManager._set_gog_size_values( + [lambda: label, lambda: label], (1024 ** 2, 1024 ** 3) + ) + + assert [call.args[0] for call in label.setText.call_args_list] == [ + "1.0 MiB", "1.0 GiB", + ] + + def test_detail_page_loads_exe_fallback_without_cover(monkeypatch: MonkeyPatch) -> None: detail_page = MagicMock() detail_page.property.side_effect = { diff --git a/tests/test_gog_api.py b/tests/test_gog_api.py new file mode 100644 index 00000000..49efa95f --- /dev/null +++ b/tests/test_gog_api.py @@ -0,0 +1,197 @@ +from pathlib import Path +from unittest.mock import Mock + +import orjson +import pytest + +from portprotonqt.gog_api import GOGAPI, GOG_PRODUCT_LOCALES +from portprotonqt.localization import LOCALE_MAP + + +def test_default_games_directory_is_user_games_folder() -> None: + api = GOGAPI() + + assert api.games_dir == Path.home() / "Games" + + +def test_extract_auth_code() -> None: + url = "https://embed.gog.com/on_login_success?origin=client&code=test%20code" + + assert GOGAPI.extract_auth_code(url) == "test code" + + +def test_load_installed_uses_isolated_data_dir(tmp_path: Path) -> None: + api = GOGAPI() + api.installed_path = tmp_path / "installed.json" + api.installed_path.write_bytes(orjson.dumps({"123": {"title": "Game"}})) + + assert api.load_installed() == {"123": {"title": "Game"}} + + +def test_remove_installed_game_keeps_other_records(tmp_path: Path) -> None: + api = GOGAPI() + api.installed_path = tmp_path / "installed.json" + api.installed_path.write_bytes(orjson.dumps({"123": {}, "456": {}})) + + api.remove_installed_game("123") + + assert api.load_installed() == {"456": {}} + + +def test_get_launch_target_reads_primary_task(tmp_path: Path) -> None: + api = GOGAPI() + api.installed_path = tmp_path / "installed.json" + game_dir = tmp_path / "game" + executable = game_dir / "BIN/Game.exe" + executable.parent.mkdir(parents=True) + executable.touch() + api.installed_path.write_bytes(orjson.dumps({"123": {"install_path": str(game_dir)}})) + info = { + "playTasks": [{ + "type": "FileTask", "path": "bin\\game.exe", + "arguments": '-conf "..\\game.conf"', "isPrimary": True, + }] + } + (game_dir / "goggame-123.info").write_bytes(orjson.dumps(info)) + + assert api.get_launch_target("123") == str(executable) + + api.ensure_launch_parameters("123") + assert 'export LAUNCH_PARAMETERS="-conf ../game.conf"' in Path( + f"{executable}.ppdb" + ).read_text() + + +def test_is_game_installed_requires_gogdl_metadata(tmp_path: Path) -> None: + api = GOGAPI() + api.installed_path = tmp_path / "installed.json" + game_dir = tmp_path / "game" + game_dir.mkdir() + api.installed_path.write_bytes(orjson.dumps({"123": {"install_path": str(game_dir)}})) + + assert api.is_game_installed("123") is False + + (game_dir / "goggame-123.info").write_text("{}") + assert api.is_game_installed("123") is True + + +def test_find_install_path_returns_gogdl_directory(tmp_path: Path) -> None: + api = GOGAPI() + game_dir = tmp_path / "Game created by gogdl" + game_dir.mkdir() + (game_dir / "goggame-123.info").write_text("{}") + + assert api.find_install_path("123", tmp_path) == game_dir + + +def test_is_game_installed_discovers_default_gogdl_directory(tmp_path: Path) -> None: + api = GOGAPI() + api.installed_path = tmp_path / "installed.json" + api.games_dir = tmp_path / "games" + game_dir = api.games_dir / "Game created by gogdl" + game_dir.mkdir(parents=True) + (game_dir / "goggame-123.info").write_text("{}") + + assert api.is_game_installed("123") is True + + +def test_localized_value_prefers_current_language(monkeypatch) -> None: + monkeypatch.setattr("portprotonqt.gog_api.get_metadata_language", lambda: "ru") + + assert GOGAPI._localized_value({"*": "English", "ru-RU": "Русский"}) == "Русский" + + +def test_product_locales_cover_supported_languages() -> None: + assert set(GOG_PRODUCT_LOCALES) == {language.lower() for language in LOCALE_MAP} + + +def test_parse_download_sizes_includes_language_data() -> None: + output = ( + b'{"size":{"*":{"download_size":100,"disk_size":200},' + b'"en-US":{"download_size":30,"disk_size":40}},' + b'"languages":["en-US"]}' + ) + + assert GOGAPI.parse_download_sizes(output) == (130, 240) + + +def test_download_sizes_cache(tmp_path: Path) -> None: + api = GOGAPI() + api.sizes_path = tmp_path / "sizes.json" + + assert api.get_cached_download_sizes("123") is None + api.save_download_sizes("123", (100, 200)) + + assert api.get_cached_download_sizes("123") == (100, 200) + + +@pytest.mark.parametrize( + ("lead", "expected"), + [ + ( + "Description française.

Deuxième paragraphe.", + "Description française.", + ), + ( + "Avertissement.

Description.

Troisième paragraphe.", + "Avertissement.

Description.", + ), + ], +) +def test_get_game_loads_localized_product_description( + monkeypatch: pytest.MonkeyPatch, lead: str, expected: str +) -> None: + gamesdb_response = Mock() + gamesdb_response.json.return_value = { + "game": {"title": {"*": "Game"}, "visible_in_library": True}, + "summary": {"*": "English description"}, + } + product_response = Mock() + product_response.json.return_value = { + "description": {"lead": lead}, + } + request = Mock(side_effect=[gamesdb_response, product_response]) + monkeypatch.setattr("portprotonqt.gog_api.get_metadata_language", lambda: "fr") + monkeypatch.setattr("portprotonqt.gog_api.requests.get", request) + + game = GOGAPI()._get_game( + {"platform_id": "gog", "external_id": "1207666073"}, "token" + ) + + assert game["description"] == expected + assert request.call_args_list[1].kwargs["params"]["locale"] == "fr-FR" + + +def test_is_authenticated_requires_token_and_user_id(monkeypatch) -> None: + api = GOGAPI() + + monkeypatch.setattr(api, "get_credentials", lambda: {"access_token": "token"}) + assert api.is_authenticated() is False + + monkeypatch.setattr( + api, "get_credentials", lambda: {"access_token": "token", "user_id": "123"} + ) + assert api.is_authenticated() is True + + +def test_ensure_gogdl_downloads_matching_architecture(tmp_path: Path, monkeypatch) -> None: + api = GOGAPI() + api.data_dir = tmp_path + asset = {"name": "gogdl_linux_x86_64"} + response = Mock() + response.json.return_value = {"assets": [asset]} + monkeypatch.setattr("portprotonqt.gog_api.platform.machine", lambda: "x86_64") + monkeypatch.setattr("portprotonqt.gog_api.requests.get", lambda *args, **kwargs: response) + monkeypatch.setattr(api, "_download_gogdl_asset", lambda selected: str(selected["name"])) + + assert api.ensure_gogdl() == "gogdl_linux_x86_64" + + +def test_authenticate_returns_gogdl_error(monkeypatch) -> None: + api = GOGAPI() + result = Mock(returncode=1, stdout=b"", stderr=b"authorization failed") + monkeypatch.setattr(api, "ensure_gogdl", lambda: "/bin/gogdl") + monkeypatch.setattr(api, "build_command", lambda arguments: ["gogdl", *arguments]) + monkeypatch.setattr("portprotonqt.gog_api.subprocess.run", lambda *args, **kwargs: result) + + assert api.authenticate("code") == (False, "authorization failed") diff --git a/tests/test_main_window.py b/tests/test_main_window.py index 298659b8..2ffd34d0 100644 --- a/tests/test_main_window.py +++ b/tests/test_main_window.py @@ -1,10 +1,12 @@ """Tests for main window library data processing.""" import shlex +from collections.abc import Callable from pathlib import Path from queue import Queue from types import SimpleNamespace from typing import Any, cast +from unittest.mock import MagicMock from pytest import MonkeyPatch from PySide6.QtCore import Qt @@ -15,6 +17,7 @@ from portprotonqt.game_library_manager import GameLibraryManager from portprotonqt.main_window import MainWindow from portprotonqt.portproton_api import remove_empty_custom_data_dirs import portprotonqt.tabs.autoinstall_tab as autoinstall_tab_module +import portprotonqt.tabs.gog_tab as gog_tab_module import portprotonqt.tabs.library_tab as library_tab_module from portprotonqt.tabs import ( MainWindowAutoInstallTabMixin, @@ -24,6 +27,7 @@ from portprotonqt.tabs import ( MainWindowWineTabMixin, ) from portprotonqt.tabs.autoinstall_tab import MainWindowAutoInstallTabMixin as AutoInstallMixin +from portprotonqt.tabs.gog_tab import MainWindowGOGTabMixin as GOGMixin from portprotonqt.tabs.library_tab import MainWindowLibraryTabMixin as LibraryMixin from portprotonqt.tabs.settings_tab import MainWindowSettingsTabMixin as SettingsMixin from portprotonqt.tabs.theme_store import THEME_STORE_ITEM, ThemeStoreMixin @@ -152,6 +156,119 @@ def test_main_window_inherits_all_tab_mixins() -> None: assert issubclass(MainWindow, mixin) +def test_gog_account_state_detects_saved_auth( + tmp_path: Path, monkeypatch: MonkeyPatch +) -> None: + class Control: + def __init__(self) -> None: + self.text = "" + self.enabled = False + + def setText(self, text: str) -> None: + self.text = text + + def setEnabled(self, enabled: bool) -> None: + self.enabled = enabled + + auth_path = tmp_path / "auth.json" + auth_path.write_text("{}") + window = SimpleNamespace( + gog_api=SimpleNamespace(auth_path=auth_path), + gogAccountStatus=Control(), + gogRefreshButton=Control(), + gogLoginButton=Control(), + ) + monkeypatch.setattr(gog_tab_module, "_", lambda text: text) + + GOGMixin._update_gog_account_state(window) + + assert window.gogAccountStatus.text == "GOG account connected" + assert window.gogRefreshButton.enabled is True + + +def test_repair_gog_game_uses_repair_command(tmp_path: Path) -> None: + install_path = tmp_path / "Game" + started: list[tuple] = [] + api = SimpleNamespace( + get_installed_path=lambda _app_id: install_path, + build_command=lambda arguments: ["gogdl", *arguments], + ) + window = SimpleNamespace( + gog_process=None, + gog_api=api, + gogAccountStatus=SimpleNamespace(setText=lambda _text: None), + _start_gog_download=lambda *arguments: started.append(arguments), + ) + + GOGMixin._repair_gog_game(window, {"app_id": "123", "title": "Game"}) + + assert started[0][2] == [ + "gogdl", "repair", "123", "--path", str(install_path), + "--platform", "windows", + ] + + +def test_cancel_gog_download_terminates_then_kills(monkeypatch: MonkeyPatch) -> None: + process = SimpleNamespace( + terminate=MagicMock(), + kill=MagicMock(), + state=lambda: gog_tab_module.QProcess.ProcessState.Running, + ) + window = SimpleNamespace( + gog_process=process, + downloadCancelButton=SimpleNamespace(setEnabled=MagicMock()), + downloadActiveDetails=SimpleNamespace(setText=MagicMock()), + _kill_gog_process=lambda active: GOGMixin._kill_gog_process(window, active), + ) + monkeypatch.setattr(gog_tab_module.QTimer, "singleShot", lambda _delay, callback: callback()) + + GOGMixin._cancel_gog_download(window) + + process.terminate.assert_called_once() + process.kill.assert_called_once() + + +def test_import_gog_game_saves_selected_installation( + tmp_path: Path, monkeypatch: MonkeyPatch +) -> None: + selected_path = tmp_path / "Selected" + game_path = selected_path / "Game" + saved: list[tuple] = [] + + class Explorer: + def __init__(self, *_args, **_kwargs) -> None: + self.callback: Callable[[str], None] = lambda _path: None + self.file_signal = SimpleNamespace( + file_selected=SimpleNamespace(connect=self._connect) + ) + + def _connect(self, callback: Callable[[str], None]) -> None: + self.callback = callback + + def setWindowTitle(self, _title: str) -> None: + return + + def exec(self) -> None: + self.callback(str(selected_path)) + + api = SimpleNamespace( + find_install_path=lambda _app_id, _path: game_path, + save_installed_game=lambda *arguments: saved.append(arguments), + ensure_launch_parameters=lambda _app_id: None, + ) + window = SimpleNamespace( + theme=object(), + gog_api=api, + gogAccountStatus=SimpleNamespace(setText=lambda _text: None), + loadGames=lambda **_kwargs: None, + ) + monkeypatch.setattr(gog_tab_module, "FileExplorer", Explorer) + + GOGMixin._import_gog_game(window, {"app_id": "123", "title": "Game"}) + + assert saved == [("123", {"install_path": str(game_path), "title": "Game"})] + + def test_library_controls_animation_ignores_game_card_scale_duration() -> None: theme = SimpleNamespace(GAME_CARD_ANIMATION={"scale_anim_duration": 10}) @@ -732,7 +849,7 @@ def test_remove_empty_custom_data_dirs_keeps_non_empty_dirs(tmp_config_dir: Path assert kept_dir.exists() -def test_get_games_without_exe_skips_existing_and_steam(tmp_path: Path) -> None: +def test_get_games_without_exe_only_includes_portproton(tmp_path: Path) -> None: exe_path = tmp_path / "Game.exe" exe_path.write_text("", encoding="utf-8") window = MainWindow.__new__(MainWindow) @@ -756,6 +873,7 @@ def test_get_games_without_exe_skips_existing_and_steam(tmp_path: Path) -> None: "portproton", ), ("Steam", "", "", "", "", "steam://rungameid/1", "", "", "", "", 0, 0, "steam"), + ("GOG", "", "", "", "", "gog://install/1", "", "", "", "", 0, 0, "gog"), ] ) diff --git a/tests/test_shortcuts.py b/tests/test_shortcuts.py index 0d3c23ef..10ee7f55 100644 --- a/tests/test_shortcuts.py +++ b/tests/test_shortcuts.py @@ -188,6 +188,35 @@ class TestCreateDesktopFile: result = create_desktop_file("/nonexistent/game.exe") assert result is None + def test_gog_launch_uri_uses_common_desktop_entry( + self, tmp_path: Path, monkeypatch: Any + ) -> None: + _patch_location(monkeypatch, tmp_path) + + result = create_desktop_file( + "gog://launch/123", game_name="GOG Game" + ) + + assert result is not None + entry, _, icon_path = result + assert 'Exec=portprotonqt --silent "gog://launch/123"' in entry + assert icon_path == "applications-games" + + def test_gog_launch_uri_uses_executable_icon( + self, tmp_path: Path, monkeypatch: Any + ) -> None: + _patch_location(monkeypatch, tmp_path) + exe = _make_exe(tmp_path) + + result = create_desktop_file( + "gog://launch/123", game_name="GOG Game", icon_source=exe + ) + + assert result is not None + entry, _, icon_path = result + assert icon_path.endswith("data/img/GOG_Game.png") + assert f"Icon={icon_path}" in entry + def test_no_portproton_returns_none(self, monkeypatch: Any) -> None: monkeypatch.setattr( "portprotonqt.config.portproton.get_portproton_location",