From 489cca7ca7815547086a185801b282324ab7bf12 Mon Sep 17 00:00:00 2001 From: Boris Yumankulov Date: Tue, 16 Jun 2026 19:47:55 +0500 Subject: [PATCH] feat: load autoinstalls from PPDB API Signed-off-by: Boris Yumankulov --- AGENTS.md | 4 + .../share/portproton/scripts/functions_helper | 20 + build-aux/share/portproton/scripts/start.sh | 21 +- portprotonqt/detail_pages/__init__.py | 17 +- portprotonqt/detail_pages/utils.py | 6 +- portprotonqt/image_utils.py | 17 +- portprotonqt/input_manager.py | 49 +- portprotonqt/main_window.py | 239 +++--- portprotonqt/portproton_api.py | 751 ++++++------------ tests/test_main_window.py | 50 ++ tests/test_portproton_api.py | 110 +++ 11 files changed, 627 insertions(+), 657 deletions(-) create mode 100644 tests/test_main_window.py create mode 100644 tests/test_portproton_api.py diff --git a/AGENTS.md b/AGENTS.md index d2f1dbaf..55fed5df 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -169,6 +169,8 @@ QColor(self.theme.color_disabled_text) - **ALWAYS update or remove comments that reference removed dependencies, patterns, or context** - **NEVER leave stub/no-op functions** (e.g., `def func(): pass` with comment "removed") - **When removing a feature, delete the function entirely, not stub it** +- When removing async/download code, preserve callback completion paths in callers +- Add or update regression tests for callback-driven UI loading paths touched by the change - Never invent modules - Do not move files unless requested - Do not create new files for organization (unless task requires a new module) @@ -357,7 +359,9 @@ tests/ ├── test_steam_cache.py # exiftool skip, cache eviction, delete_cached_app_files ├── test_base_config.py # BaseConfig read/write, caching, versioning ├── test_cli.py # normalize_launch_path, URL/resolution parsing +├── test_main_window.py # Main window data processing and callback regressions ├── test_portproton_config.py # exec_line parsing, launcher tail, extensions +├── test_portproton_api.py # PPDB API helpers, autoinstall localization fallback ├── test_migration.py # Desktop shortcut migration, prefix backup, squashfs ├── test_icon_extractor.py # NE/PE icon extraction, DIB decoding, thumbnails ├── test_dbus_tools.py # D-Bus tools (notifications, idle inhibit, power profiles) diff --git a/build-aux/share/portproton/scripts/functions_helper b/build-aux/share/portproton/scripts/functions_helper index 45cae682..f79e681f 100755 --- a/build-aux/share/portproton/scripts/functions_helper +++ b/build-aux/share/portproton/scripts/functions_helper @@ -4616,10 +4616,30 @@ portwine_create_shortcut () { portwine_output_yad_shortcut } +pw_download_autoinstall_ppdb () { + [[ -z "${PW_PPDB_FILE:-}" ]] && return 0 + [[ -z "${PW_EXE_FILE:-}" ]] && return 0 + [[ ! -e "${PW_EXE_FILE}" ]] && return 0 + + local ppdb_path="${PW_EXE_FILE}.ppdb" + local tmp_ppdb_path="${ppdb_path}.tmp" + + print_info "Download PPDB file for $(basename "${PW_EXE_FILE}")..." + if try_download "${PW_PPDB_FILE}" "${tmp_ppdb_path}" no_mirror ; then + mv -f "${tmp_ppdb_path}" "${ppdb_path}" + return 0 + fi + + try_remove_file "${tmp_ppdb_path}" + print_warning "Failed to download PPDB file for $(basename "${PW_EXE_FILE}")" + return 0 +} + portwine_output_yad_shortcut () { [[ ! -e "${PW_EXE_FILE}" ]] && return 1 + pw_download_autoinstall_ppdb pw_create_gui_png unset name_desktop check_variables PW_SHORTCUT_MENU "FALSE" diff --git a/build-aux/share/portproton/scripts/start.sh b/build-aux/share/portproton/scripts/start.sh index 7c96b5f1..d85a9861 100755 --- a/build-aux/share/portproton/scripts/start.sh +++ b/build-aux/share/portproton/scripts/start.sh @@ -250,7 +250,6 @@ get_wine_and_pfx () { case "$1" in --help) help_info () { - files_from_autoinstall=$(ls "${PORT_SCRIPTS_PATH}/pw_autoinstall") echo -e "Usage: [--repair] [--reinstall] [--autoinstall] --repair Forces all scripts to be updated to a working state @@ -278,9 +277,7 @@ case "$1" in --clear_pfx Clear specified prefix, requires WINE version and prefix name --mangohud-preview Starts MangoHud preview in vkcube (optional argument: inline MangoHud config) --initial Initial setup command ---autoinstall --autoinstall and the name of what needs to be installed is given in the list below: - -$(echo $files_from_autoinstall | awk '{for (i = 1; i <= NF; i++) {if (i % 10 == 0) {print ""} printf "%s ", $i}}') +--autoinstall Runs the specified .ppai autoinstall script Usage examples: portproton cli --launch /path/to/game.exe @@ -294,7 +291,7 @@ Usage examples: portproton cli --winetricks-list-all WINE_LG DEFAULT portproton cli --winetricks-install WINE_LG DEFAULT --no-force corefonts portproton cli --mangohud-preview \"fps,frametime,cpu_temp,gpu_temp\" - portproton cli --autoinstall [script_name_from_pw_autoinstall] + portproton cli --autoinstall /path/to/script.ppai " } help_info @@ -333,12 +330,18 @@ Usage examples: export PW_USE_D3D_EXTRAS=1 export WINE_LARGE_ADDRESS_AWARE=0 trap "stop_portproton" SIGTERM SIGINT - if [[ "$2" =~ "/" ]] && [[ -f "$2" ]] ; then + if [[ "$2" =~ ^https?://.*\.ppai($|\?) ]] ; then + autoinstall_script="${PORT_WINE_TMP_PATH}/$(basename "${2%%\?*}")" + if try_download "$2" "${autoinstall_script}" no_mirror ; then + # shellcheck source=/dev/null + . "${autoinstall_script}" + try_remove_file "${autoinstall_script}" + else + fatal "Not found autoinstal script: $2" + fi + elif [[ -f "$2" ]] ; then # shellcheck source=/dev/null . "${2}" - elif [[ -f "${PORT_SCRIPTS_PATH}/pw_autoinstall/${2}" ]] ; then - # shellcheck source=/dev/null - . "${PORT_SCRIPTS_PATH}/pw_autoinstall/${2}" else fatal "Not found autoinstal script: $2" fi diff --git a/portprotonqt/detail_pages/__init__.py b/portprotonqt/detail_pages/__init__.py index 604b7c40..945ccde0 100644 --- a/portprotonqt/detail_pages/__init__.py +++ b/portprotonqt/detail_pages/__init__.py @@ -829,22 +829,7 @@ class DetailPageManager: def _get_enhanced_description(self, script_name: str, description: str) -> str: """Get enhanced description from metadata if available.""" - if not script_name: - return description - - import locale - - try: - current_locale = locale.getlocale()[0] or "en" - except Exception as e: - logger.debug("Failed to get current locale: %s", e) - current_locale = "en" - - lang_code = "ru" if current_locale and "ru" in current_locale.lower() else "en" - metadata_description = self.portproton_api.get_autoinstall_description( - script_name, lang_code - ) - return metadata_description if metadata_description else description + return description def _create_autoinstall_buttons_layout( self, script_name: str, name: str, buttons_layout: FlowLayout diff --git a/portprotonqt/detail_pages/utils.py b/portprotonqt/detail_pages/utils.py index d81cd999..26d79b81 100644 --- a/portprotonqt/detail_pages/utils.py +++ b/portprotonqt/detail_pages/utils.py @@ -12,7 +12,6 @@ from portprotonqt.image_utils import load_pixmap_async, round_corners, set_anima from portprotonqt.config import ( extract_exec_target_path, favorites_config, - get_portproton_scripts_path, parse_desktop_entry, ) from portprotonqt.logger import get_logger @@ -386,11 +385,10 @@ def _check_desktop_file( def _get_autoinstall_exe_name(script_name: str) -> str: - scripts_path = get_portproton_scripts_path() - if not scripts_path: + script_path = script_name + if not os.path.isfile(script_path): return "" - script_path = os.path.join(scripts_path, "pw_autoinstall", script_name) install_exe = "" try: with open(script_path, encoding="utf-8") as script_file: diff --git a/portprotonqt/image_utils.py b/portprotonqt/image_utils.py index 9cfe4b55..0397cfe5 100644 --- a/portprotonqt/image_utils.py +++ b/portprotonqt/image_utils.py @@ -1,4 +1,5 @@ import os +import re from weakref import WeakKeyDictionary from PySide6.QtGui import QPen, QColor, QPixmap, QPainter, QPainterPath, QImageReader, QMovie, QBitmap from PySide6.QtCore import ( @@ -353,6 +354,17 @@ def _get_cover_url_suffix(cover: str) -> str: return ".jpg" +def _get_ppdb_autoinstall_image_name(cover: str) -> str: + match = re.search(r"/game_(\d+)_(icon|library)_", cover) + if not match: + return "" + game_id, image_type = match.groups() + suffix = _get_cover_url_suffix(cover) + if image_type == "icon": + return f"{game_id}_compat{suffix}" + return f"{game_id}{suffix}" + + def load_pixmap_async( cover: str, width: int, @@ -502,7 +514,10 @@ def load_pixmap_async( if cover and cover.startswith(("http://", "https://")): try: - local_path = os.path.join(image_folder, f"{app_name}{_get_cover_url_suffix(cover)}") + filename = _get_ppdb_autoinstall_image_name(cover) + if not filename: + filename = f"{app_name}{_get_cover_url_suffix(cover)}" + local_path = os.path.join(image_folder, filename) if os.path.exists(local_path): pixmap = QPixmap(local_path) # Check if the pixmap loaded successfully diff --git a/portprotonqt/input_manager.py b/portprotonqt/input_manager.py index 16e05faa..b11282fb 100644 --- a/portprotonqt/input_manager.py +++ b/portprotonqt/input_manager.py @@ -3198,7 +3198,7 @@ class InputManager(QObject): if code in (PAD_DPAD_X, PAD_DPAD_Y): current_index = self._parent.stackedWidget.currentIndex() if current_index in (0, 1): - if self._handle_library_toolbar_navigation(code, value): + if self._handle_toolbar_navigation(code, value): return if code == PAD_DPAD_Y and value < 0 and self._focus_tab_from_search(current_index): return @@ -3725,8 +3725,7 @@ class InputManager(QObject): filter_widgets[0].setFocus(Qt.FocusReason.OtherFocusReason) return True - def _focus_first_library_card(self) -> bool: - container = getattr(self._parent, "gamesListWidget", None) + def _focus_first_game_card(self, container: QWidget | None) -> bool: if container is None: return False for card in container.findChildren(GameCard): @@ -3740,16 +3739,38 @@ class InputManager(QObject): return True return False - def _handle_library_toolbar_navigation(self, code: int, value: int) -> bool: - if value == 0 or self._parent.stackedWidget.currentIndex() != 0: + def _focus_first_library_card(self) -> bool: + container = getattr(self._parent, "gamesListWidget", None) + return self._focus_first_game_card(container) + + def _get_autoinstall_toolbar_widgets(self) -> list[QWidget]: + widgets = [] + for attr_name in ("autoInstallSearchLineEdit", "autoInstallRefreshButton"): + widget = getattr(self._parent, attr_name, None) + if isinstance(widget, QWidget) and widget.isVisible() and widget.isEnabled(): + widgets.append(widget) + return widgets + + def _focus_first_autoinstall_card(self) -> bool: + container = getattr(self._parent, "autoInstallContainer", None) + return self._focus_first_game_card(container) + + def _handle_toolbar_navigation(self, code: int, value: int) -> bool: + current_index = self._parent.stackedWidget.currentIndex() + if value == 0 or current_index not in (0, 1): return False - toolbar_widgets = self._get_library_toolbar_widgets() - filter_widgets = self._get_library_filter_widgets() + if current_index == 0: + toolbar_widgets = self._get_library_toolbar_widgets() + focus_first_card = self._focus_first_library_card + else: + toolbar_widgets = self._get_autoinstall_toolbar_widgets() + focus_first_card = self._focus_first_autoinstall_card focused = self._focused_widget() + filter_widgets = self._get_library_filter_widgets() if current_index == 0 else [] if focused in filter_widgets: if code == PAD_DPAD_X: - current_index = filter_widgets.index(cast(QWidget, focused)) - next_index = current_index + (1 if value > 0 else -1) + widget_index = filter_widgets.index(cast(QWidget, focused)) + next_index = widget_index + (1 if value > 0 else -1) if 0 <= next_index < len(filter_widgets): filter_widgets[next_index].setFocus(Qt.FocusReason.OtherFocusReason) return True @@ -3760,12 +3781,12 @@ class InputManager(QObject): if isinstance(controls_button, QWidget): controls_button.setFocus(Qt.FocusReason.OtherFocusReason) return True - return self._focus_first_library_card() + return focus_first_card() if focused not in toolbar_widgets: return False if code == PAD_DPAD_X: - current_index = toolbar_widgets.index(cast(QWidget, focused)) - next_index = current_index + (1 if value > 0 else -1) + widget_index = toolbar_widgets.index(cast(QWidget, focused)) + next_index = widget_index + (1 if value > 0 else -1) if 0 <= next_index < len(toolbar_widgets): toolbar_widgets[next_index].setFocus(Qt.FocusReason.OtherFocusReason) return True @@ -3775,8 +3796,8 @@ class InputManager(QObject): controls_button = getattr(self._parent, "libraryControlsButton", None) if focused is controls_button and self._focus_library_filter_controls(): return True - return self._focus_first_library_card() - tab_button = self._parent.tabButtons.get(0) + return focus_first_card() + tab_button = self._parent.tabButtons.get(current_index) if tab_button is not None and tab_button.isVisible(): tab_button.setFocus(Qt.FocusReason.OtherFocusReason) return True diff --git a/portprotonqt/main_window.py b/portprotonqt/main_window.py index 08317bc6..5765fe59 100644 --- a/portprotonqt/main_window.py +++ b/portprotonqt/main_window.py @@ -303,6 +303,7 @@ class MainWindow(MainWindowControlHintsMixin, MainWindowSystemTabMixin, MainWind QTimer.singleShot(0, self.start_watching_directories) # Delay to ensure portproton_location is set self.downloader = Downloader(max_workers=4) self.portproton_api = PortProtonAPI(self.downloader) + self.autoInstallCustomDataThread = None self.installing = False self.install_process = None @@ -1041,7 +1042,6 @@ class MainWindow(MainWindowControlHintsMixin, MainWindowSystemTabMixin, MainWind self, file_path: str, callback: Callable[[tuple | None], None], - assets_checked: bool = False, ): entry = parse_desktop_entry(file_path) if not entry: @@ -1070,7 +1070,6 @@ class MainWindow(MainWindowControlHintsMixin, MainWindowSystemTabMixin, MainWind generated_img_icon = "" themed_launch_icon = "" economy_mode = ui_config.get_economy_mode() - if game_exe: exe_name = os.path.splitext(os.path.basename(game_exe))[0] user_game_folder = os.path.join(user_custom_folder, exe_name) @@ -1080,18 +1079,6 @@ class MainWindow(MainWindowControlHintsMixin, MainWindowSystemTabMixin, MainWind game_exe, entry.get("Icon", ""), desktop_name ) - # Check if local game folder is empty and download assets if it is - if not themed_launch_icon and not economy_mode and not assets_checked and not os.listdir(user_game_folder): - logger.debug(f"Local folder for {exe_name} is empty, checking repository") - def on_assets_downloaded(results): - if results["cover"]: - logger.info(f"Downloaded assets for {exe_name}: {results}") - if results["metadata"]: - logger.info(f"Downloaded metadata for {exe_name}: {results['metadata']}") - self._process_desktop_file_async(file_path, callback, assets_checked=True) - self.portproton_api.download_game_assets_async(exe_name, timeout=5, callback=on_assets_downloaded) - return - user_files = set(os.listdir(user_game_folder)) if os.path.exists(user_game_folder) else set() for ext in COVER_IMAGE_EXTENSIONS: candidate = f"cover{ext}" @@ -1833,14 +1820,39 @@ class MainWindow(MainWindowControlHintsMixin, MainWindowSystemTabMixin, MainWind self.autoInstallSearchLineEdit = CustomLineEdit(self, theme=self.theme) icon: QIcon = cast(QIcon, self.theme_manager.get_icon("search")) action_pos = QLineEdit.ActionPosition.LeadingPosition - self.autoInstallSearchLineEdit.addAction(icon, action_pos) + self.autoInstallSearchIconAction = self.autoInstallSearchLineEdit.addAction(icon, action_pos) self.autoInstallSearchLineEdit.setMaximumWidth(200) self.autoInstallSearchLineEdit.setPlaceholderText(_("Search ...")) self.autoInstallSearchLineEdit.setClearButtonEnabled(True) self.autoInstallSearchLineEdit.setStyleSheet(self.theme.SEARCH_EDIT_STYLE) self.autoInstallSearchLineEdit.textChanged.connect(self.filterAutoInstallGames) + self.autoInstallSearchLineEdit.focusInEvent = self._wrap_autoinstall_search_focus_event( + self.autoInstallSearchLineEdit.focusInEvent, + True, + ) + self.autoInstallSearchLineEdit.focusOutEvent = self._wrap_autoinstall_search_focus_event( + self.autoInstallSearchLineEdit.focusOutEvent, + False, + ) + self.autoInstallSearchLineEdit.resizeEvent = self._wrap_autoinstall_search_resize_event( + self.autoInstallSearchLineEdit.resizeEvent + ) searchLayout.addWidget(self.autoInstallSearchLineEdit) + + self.autoInstallRefreshButton = AutoSizeButton(icon=self.theme_manager.get_icon("update", as_path=True)) + button_style = getattr( + self.theme, + "LIBRARY_CONTROLS_BUTTON_STYLE", + self.theme.ADDGAME_BACK_BUTTON_STYLE, + ) + self.autoInstallRefreshButton.setStyleSheet(button_style) + self.autoInstallRefreshButton.setFocusPolicy(Qt.FocusPolicy.StrongFocus) + self.autoInstallRefreshButton.clicked.connect(self._refresh_autoinstall_games) + self._register_gamepad_tooltip(self.autoInstallRefreshButton, _("Refresh Grid")) + searchLayout.addWidget(self.autoInstallRefreshButton) + autoInstallLayout.addWidget(searchWidget) + self._setup_autoinstall_search_animation() self.autoInstallScrollArea = AutoHideScrollArea(theme=self.theme) self.autoInstallScrollArea.setWidgetResizable(True) @@ -1886,7 +1898,6 @@ class MainWindow(MainWindowControlHintsMixin, MainWindowSystemTabMixin, MainWind def on_autoinstall_games_loaded(games: list[tuple]): self.autoInstallLoaded = True self.autoInstallLoading = False - economy_mode = ui_config.get_economy_mode() # Clear while self.autoInstallContainerLayout.count(): @@ -1909,21 +1920,14 @@ class MainWindow(MainWindowControlHintsMixin, MainWindowSystemTabMixin, MainWind logger.warning(f"Invalid exec_line for autoinstall: {exec_line}") return game_data["game_source"] = "portproton" + ppai_target = exec_line[12:].strip() + if ppai_target.startswith(("http://", "https://")): + self._open_autoinstall_card_after_script_download(game_data, ppai_target) + return self.detail_page_manager.openAutoInstallDetailPage(game_data) - def get_autoinstall_theme_cover(exe_name: str) -> str | None: - theme_cover = self.theme_manager.get_theme_image(exe_name, self.current_theme_name) - if isinstance(theme_cover, str) and "autoinstall_covers" in theme_cover: - return theme_cover - if auto_layout_mode != "list": - return None - classic_cover = self.theme_manager.get_theme_image(exe_name, "classic") - if isinstance(classic_cover, str) and "autoinstall_covers" in classic_cover: - return classic_cover - return None - # Create cards - for game_tuple in games: + for game_tuple in sorted(games, key=lambda item: str(item[0]).casefold()): name = game_tuple[0] description = game_tuple[1] cover_path = game_tuple[2] @@ -1932,17 +1936,12 @@ class MainWindow(MainWindowControlHintsMixin, MainWindowSystemTabMixin, MainWind exec_line = game_tuple[5] game_source = game_tuple[12] exe_name = game_tuple[13] - theme_cover = get_autoinstall_theme_cover(exe_name) - has_theme_cover = theme_cover is not None - if auto_layout_mode == "list" and has_theme_cover: - cover_path = theme_cover - elif not cover_path: - if has_theme_cover: - cover_path = theme_cover - elif economy_mode: - classic_cover = self.theme_manager.get_theme_image(exe_name, "classic") - if isinstance(classic_cover, str) and "autoinstall_covers" in classic_cover: - cover_path = classic_cover + compact_cover = game_tuple[14] if len(game_tuple) > 14 else "" + full_cover = game_tuple[15] if len(game_tuple) > 15 else cover_path + if auto_layout_mode == "list": + cover_path = compact_cover or full_cover + else: + cover_path = full_cover or compact_cover card = GameCard( name, description, cover_path, appid, controller_support, @@ -1971,38 +1970,6 @@ class MainWindow(MainWindowControlHintsMixin, MainWindowSystemTabMixin, MainWind self.allAutoInstallCards.append(card) self.autoInstallContainerLayout.addWidget(card) - # Load missing covers and metadata in batch - exe_names_to_load = [game_tuple[13] for game_tuple in games] - - def batch_cover_callback(exe_name, local_path): - if local_path and exe_name in self.autoInstallGameCards: - card = self.autoInstallGameCards[exe_name] - if card.list_layout: - theme_cover = get_autoinstall_theme_cover(exe_name) - if theme_cover is not None: - return - card.cover_path = local_path - cover_width = 64 if card.list_layout else self.auto_card_width - cover_height = cover_width if card.list_layout else int(self.auto_card_width * 1.5) - load_pixmap_async(local_path, cover_width, cover_height, card.on_cover_loaded) - - def batch_metadata_callback(exe_name, local_path): - logger.debug(f"Metadata callback called for {exe_name}: {local_path}") - if local_path and os.path.exists(local_path): - try: - self._update_card_name_from_metadata(exe_name, local_path) - logger.info(f"Updated metadata for {exe_name}") - except Exception as e: - logger.error(f"Error updating card metadata for {exe_name}: {e}") - - if exe_names_to_load and not economy_mode: - self.portproton_api.download_autoinstall_assets_batch_async( - exe_names_to_load, - timeout=5, - cover_callback=batch_cover_callback, - metadata_callback=batch_metadata_callback - ) - self.autoInstallContainer.updateGeometry() self.autoInstallScrollArea.updateGeometry() self.filterAutoInstallGames() @@ -2011,19 +1978,99 @@ class MainWindow(MainWindowControlHintsMixin, MainWindowSystemTabMixin, MainWind self.stackedWidget.addWidget(autoInstallPage) - def _start_autoinstall_load(self) -> None: + def _open_autoinstall_card_after_script_download( + self, + game_data: dict, + ppai_url: str, + ) -> None: + def on_script_ready(script_path: str) -> None: + self.autoInstallScriptLoadThread = None + if script_path: + game_data["exec_line"] = f"autoinstall:{script_path}" + self.detail_page_manager.openAutoInstallDetailPage(game_data) + if script_path: + self.autoInstallCustomDataThread = self.portproton_api.start_autoinstall_custom_data_write( + script_path, + game_data, + ) + self.autoInstallCustomDataThread.finished.connect( + lambda: setattr(self, "autoInstallCustomDataThread", None) + ) + + self.autoInstallScriptLoadThread = self.portproton_api.start_autoinstall_script_download( + ppai_url, + on_script_ready, + ) + + def _setup_autoinstall_search_animation(self) -> None: + self.autoInstallSearchAnimation = ExpandingSearchAnimation( + self.autoInstallSearchLineEdit, + self.theme, + self.searchDebounceTimer.interval(), + ) + collapsed_width = self.autoInstallRefreshButton.sizeHint().width() + expanded_width = self.autoInstallSearchLineEdit.maximumWidth() + self.autoInstallSearchAnimation.setup(collapsed_width, expanded_width) + QTimer.singleShot(0, self._center_collapsed_autoinstall_search_icon) + + def _wrap_autoinstall_search_focus_event(self, original_event: Callable, expand: bool) -> Callable: + def handle_focus_event(event): + original_event(event) + if not hasattr(self, "autoInstallSearchAnimation"): + return + if expand: + self.autoInstallSearchAnimation.expand() + else: + self.autoInstallSearchAnimation.collapse() + QTimer.singleShot(0, self._center_collapsed_autoinstall_search_icon) + return handle_focus_event + + def _wrap_autoinstall_search_resize_event(self, original_event: Callable) -> Callable: + def handle_resize_event(event): + original_event(event) + self._center_collapsed_autoinstall_search_icon() + return handle_resize_event + + def _center_collapsed_autoinstall_search_icon(self) -> None: + animation = getattr(self, "autoInstallSearchAnimation", None) + if animation is None or self.autoInstallSearchLineEdit.maximumWidth() != animation.collapsed_width: + return + for button in self.autoInstallSearchLineEdit.findChildren(QToolButton): + if button.defaultAction() is self.autoInstallSearchIconAction: + size = button.sizeHint() + x = (self.autoInstallSearchLineEdit.width() - size.width()) // 2 + y = (self.autoInstallSearchLineEdit.height() - size.height()) // 2 + button.setGeometry(x, y, size.width(), size.height()) + return + + def _start_autoinstall_load(self, force_refresh: bool = False) -> None: if self.autoInstallLoaded or self.autoInstallLoading: return if not hasattr(self, "_on_autoinstall_games_loaded"): return self.autoInstallLoading = True self.autoInstallLoadThread = self.portproton_api.start_autoinstall_games_load( - self._on_autoinstall_games_loaded + self._on_autoinstall_games_loaded, + force_refresh=force_refresh, ) - if self.autoInstallLoadThread: - def on_thread_finished(): - self.autoInstallLoadThread = None # Release reference - self.autoInstallLoadThread.finished.connect(on_thread_finished) + if not self.autoInstallLoadThread: + self.autoInstallLoading = False + if hasattr(self, "autoInstallRefreshButton"): + self.autoInstallRefreshButton.setEnabled(True) + return + + def on_thread_finished(): + self.autoInstallLoadThread = None # Release reference + if hasattr(self, "autoInstallRefreshButton"): + self.autoInstallRefreshButton.setEnabled(True) + self.autoInstallLoadThread.finished.connect(on_thread_finished) + + def _refresh_autoinstall_games(self) -> None: + if self.autoInstallLoading: + return + self.autoInstallRefreshButton.setEnabled(False) + self.autoInstallLoaded = False + self._start_autoinstall_load(force_refresh=True) def on_auto_slider_released(self): """Handles auto-install slider release to update card size.""" @@ -4354,44 +4401,6 @@ class MainWindow(MainWindowControlHintsMixin, MainWindowSystemTabMixin, MainWind except Exception as e: logger.warning(f"Failed to cleanup debug log manager: {e}") - def _update_card_name_from_metadata(self, exe_name: str, metadata_path: str): - """Update card name and description from metadata file.""" - # Read the translated metadata using the existing function - language_code = get_metadata_language() - translations = read_metadata_translations(metadata_path, language_code) - - # Update the card with the new name and description if available - if exe_name in self.autoInstallGameCards: - card = self.autoInstallGameCards[exe_name] - - # Defensive check: Ensure card is not a list or other unexpected type - if isinstance(card, list): - logger.error(f"Card for {exe_name} is unexpectedly a list: {card}") - return - - # Additional defensive checks for card validity - if not hasattr(card, 'nameLabel'): - logger.warning(f"Card for {exe_name} doesn't have nameLabel attribute") - return - - if not (hasattr(card, 'nameLabel') and hasattr(card.nameLabel, 'setText')): - logger.warning(f"Card nameLabel for {exe_name} doesn't have setText method") - return - - if translations and 'name' in translations and translations['name'] and translations['name'] != _('Unknown Game'): - # Update the card's internal name reference - card.name = translations['name'] - # Update the display label - if hasattr(card, 'nameLabel') and card.nameLabel: - card.nameLabel.setText(translations['name']) - - # Update description if available - if translations and 'description' in translations and translations['description']: - card.description = translations['description'] - if hasattr(card, 'descriptionLabel') and card.descriptionLabel: - card.descriptionLabel.setText(translations['description']) - - def goBackDetailPage(self, page): """Bridge method to detail page manager.""" result = self.detail_page_manager.goBackDetailPage(page) diff --git a/portprotonqt/portproton_api.py b/portprotonqt/portproton_api.py index 12f3328d..9ce63f46 100644 --- a/portprotonqt/portproton_api.py +++ b/portprotonqt/portproton_api.py @@ -3,7 +3,6 @@ import subprocess import requests import urllib.parse import time -import glob import re import hashlib import queue @@ -11,7 +10,7 @@ import shutil import locale from collections.abc import Callable from typing import Any -from PySide6.QtCore import QThread, Signal, QUrl, QObject, Qt, SignalInstance +from PySide6.QtCore import QThread, Signal, QUrl, QObject from PySide6.QtGui import QDesktopServices from PySide6.QtWidgets import QApplication from portprotonqt.downloader import Downloader, get_requests_session @@ -19,7 +18,6 @@ from portprotonqt.logger import get_logger from portprotonqt.config import ( extract_exec_target_path, get_portproton_location, - get_portproton_scripts_path, get_portproton_start_command, ) from portprotonqt.localization import _ @@ -28,7 +26,7 @@ from portprotonqt.config.cache import CacheManager from portprotonqt.image_utils import COVER_IMAGE_EXTENSIONS logger = get_logger(__name__) -AUTOINSTALL_CACHE_DURATION = 3600 # 1 hour for autoinstall cache +AUTOINSTALL_API_URL = "https://ppdb.linux-gaming.ru/api/games/autoinstall" HEAD_FAILURE_RETRY_DELAY = 60 # 1 minute cooldown for failed HEAD checks HEAD_CACHE_DURATION = 24 * 60 * 60 HEAD_CACHE_NAME = "head_cache" @@ -106,9 +104,8 @@ def extract_exe_name(exec_line: str) -> str: class PortProtonAPI: - """API to fetch game assets (cover, metadata) and forum topics from the PortProtonQt repository.""" + """API helpers for PPDB and autoinstall data.""" def __init__(self, downloader: Downloader | None = None): - self.base_url = "https://git.linux-gaming.ru/Linux-Gaming/PortProtonQt-Custom-Metadata/raw/branch/main" self.downloader = downloader or Downloader(max_workers=4) self.xdg_data_home = os.getenv("XDG_DATA_HOME", os.path.join(os.path.expanduser("~"), ".local", "share")) self.custom_data_dir = os.path.join(self.xdg_data_home, "PortProtonQt", "custom_data") @@ -119,13 +116,6 @@ class PortProtonAPI: self._head_negative_cache: set[str] = set() self._head_failure_cache: dict[str, float] = {} self._head_disk_cache_cleaned = False - self._pending_asset_notifier: QObject | None = None - self._pending_asset_batch_thread: QThread | None = None - - def _get_game_dir(self, exe_name: str) -> str: - game_dir = os.path.join(self.custom_data_dir, exe_name) - os.makedirs(game_dir, exist_ok=True) - return game_dir def _load_head_cache_entry(self, cache_manager: CacheManager, cache_key: str) -> bool | None: cached = cache_manager.load_json(HEAD_CACHE_NAME) @@ -195,525 +185,289 @@ class PortProtonAPI: self._head_failure_cache[url] = time.time() return False - def download_game_assets_async(self, exe_name: str, timeout: int = 5, callback: Callable[[dict[str, str | None]], None] | None = None) -> None: - game_dir = self._get_game_dir(exe_name) - cover_extensions = [".png"] - cover_url_base = f"{self.base_url}/{exe_name}/cover" - metadata_url = f"{self.base_url}/{exe_name}/metadata.txt" - - results: dict[str, str | None] = {"cover": None, "metadata": None} - pending_downloads = 0 - - def on_cover_downloaded(local_path: str | None, ext: str): - nonlocal pending_downloads - if local_path: - logger.info(f"Async cover downloaded for {exe_name}: {local_path}") - results["cover"] = local_path - else: - logger.debug(f"No cover downloaded for {exe_name} with extension {ext}") - pending_downloads -= 1 - check_completion() - - def on_metadata_downloaded(local_path: str | None): - nonlocal pending_downloads - if local_path: - logger.info(f"Async metadata downloaded for {exe_name}: {local_path}") - results["metadata"] = local_path - else: - logger.debug(f"No metadata downloaded for {exe_name}") - pending_downloads -= 1 - check_completion() - - def check_completion(): - if pending_downloads == 0 and callback: - callback(results) - - metadata_found = False - local_metadata_path = os.path.join(game_dir, "metadata.txt") - if os.path.exists(local_metadata_path): - logger.debug(f"Metadata already exists locally for {exe_name}: {local_metadata_path}") - results["metadata"] = local_metadata_path - metadata_found = True - elif self._check_file_exists(metadata_url, timeout): - metadata_found = True - pending_downloads += 1 - self.downloader.download_async( - metadata_url, - local_metadata_path, - timeout=timeout, - callback=on_metadata_downloaded - ) - - if not metadata_found: - logger.debug(f"No metadata found for {exe_name}, skipping cover checks") - if callback: - callback(results) - return - - for ext in cover_extensions: - cover_url = f"{cover_url_base}{ext}" - if self._check_file_exists(cover_url, timeout): - local_cover_path = os.path.join(game_dir, f"cover{ext}") - pending_downloads += 1 - self.downloader.download_async( - cover_url, - local_cover_path, - timeout=timeout, - callback=lambda path, ext=ext: on_cover_downloaded(path, ext) - ) - break - - if pending_downloads == 0: - logger.debug(f"No assets found for {exe_name}") - if callback: - callback(results) - - def download_autoinstall_cover_async(self, exe_name: str, timeout: int = 5, callback: Callable[[str | None], None] | None = None) -> None: - """Download only autoinstall cover image (PNG only, no metadata).""" - xdg_data_home = os.getenv("XDG_DATA_HOME", - os.path.join(os.path.expanduser("~"), ".local", "share")) - autoinstall_root = os.path.join(xdg_data_home, "PortProtonQt", "custom_data", "autoinstall") - user_game_folder = os.path.join(autoinstall_root, exe_name) - - if not os.path.isdir(user_game_folder): - try: - os.mkdir(user_game_folder) - except FileExistsError: - pass - - local_cover_path = os.path.join(user_game_folder, "cover.png") - - # Check if the cover already exists locally before attempting download - if os.path.exists(local_cover_path): - logger.debug(f"Async autoinstall cover already exists locally for {exe_name}: {local_cover_path}") - if callback: - callback(local_cover_path) - return - - cover_url = f"{self.base_url}/{exe_name}/cover.png" - - def on_cover_downloaded(local_path: str | None): - if local_path: - logger.info(f"Async autoinstall cover downloaded for {exe_name}: {local_path}") - else: - logger.debug(f"No autoinstall cover downloaded for {exe_name}") - if callback: - callback(local_path) - - self.downloader.download_async( - cover_url, - local_cover_path, - timeout=timeout, - callback=on_cover_downloaded - ) - - def download_autoinstall_metadata_async(self, exe_name: str, timeout: int = 5, callback: Callable[[str | None], None] | None = None) -> None: - """Download autoinstall metadata.txt file.""" - xdg_data_home = os.getenv("XDG_DATA_HOME", - os.path.join(os.path.expanduser("~"), ".local", "share")) - autoinstall_root = os.path.join(xdg_data_home, "PortProtonQt", "custom_data", "autoinstall") - user_game_folder = os.path.join(autoinstall_root, exe_name) - - if not os.path.isdir(user_game_folder): - try: - os.makedirs(user_game_folder, exist_ok=True) - except FileExistsError: - pass - - local_metadata_path = os.path.join(user_game_folder, "metadata.txt") - - # Check if the file already exists locally before attempting download - if os.path.exists(local_metadata_path): - logger.debug(f"Async autoinstall metadata already exists locally for {exe_name}: {local_metadata_path}") - if callback: - callback(local_metadata_path) - return - - metadata_url = f"{self.base_url}/{exe_name}/metadata.txt" - - def on_metadata_downloaded(local_path: str | None): - if local_path: - logger.info(f"Async autoinstall metadata downloaded for {exe_name}: {local_path}") - else: - logger.debug(f"No autoinstall metadata downloaded for {exe_name}") - if callback: - callback(local_path) - - self.downloader.download_async( - metadata_url, - local_metadata_path, - timeout=timeout, - callback=on_metadata_downloaded - ) - - def download_autoinstall_assets_batch_async( - self, - exe_names: list[str], - timeout: int = 5, - cover_callback: Callable[[str, str | None], None] | None = None, - metadata_callback: Callable[[str, str | None], None] | None = None - ) -> None: - """Download covers and metadata for multiple autoinstall games in parallel. - - Args: - exe_names: List of executable names to download assets for - timeout: Request timeout in seconds - cover_callback: Callback(exe_name, local_path) for cover downloads - metadata_callback: Callback(exe_name, local_path) for metadata downloads - """ - xdg_data_home = os.getenv( - "XDG_DATA_HOME", - os.path.join(os.path.expanduser("~"), ".local", "share"), - ) - autoinstall_root = os.path.join( - xdg_data_home, "PortProtonQt", "custom_data", "autoinstall" - ) - - class AssetNotifier(QObject): - cover_ready = Signal(str, object) - metadata_ready = Signal(str, object) - - notifier = AssetNotifier() - if cover_callback: - notifier.cover_ready.connect(cover_callback, Qt.ConnectionType.DirectConnection) - if metadata_callback: - notifier.metadata_ready.connect(metadata_callback, Qt.ConnectionType.DirectConnection) - self._pending_asset_notifier = notifier - - pending_tasks: list[tuple[str, str, SignalInstance, str, str]] = [] - - for exe_name in exe_names: - user_game_folder = os.path.join(autoinstall_root, exe_name) - try: - os.makedirs(user_game_folder, exist_ok=True) - except FileExistsError: - pass - - local_cover_path = os.path.join(user_game_folder, "cover.png") - local_metadata_path = os.path.join(user_game_folder, "metadata.txt") - - cover_exists = os.path.exists(local_cover_path) - metadata_exists = os.path.exists(local_metadata_path) - - if cover_exists: - logger.debug(f"Batch cover already exists for {exe_name}: {local_cover_path}") - notifier.cover_ready.emit(exe_name, local_cover_path) - else: - cover_url = f"{self.base_url}/{exe_name}/cover.png" - pending_tasks.append((cover_url, local_cover_path, notifier.cover_ready, exe_name, "cover")) - - if metadata_exists: - logger.debug(f"Batch metadata already exists for {exe_name}: {local_metadata_path}") - notifier.metadata_ready.emit(exe_name, local_metadata_path) - else: - metadata_url = f"{self.base_url}/{exe_name}/metadata.txt" - pending_tasks.append((metadata_url, local_metadata_path, notifier.metadata_ready, exe_name, "metadata")) - - if not pending_tasks: - return - - pending_tasks_by_url = {task[0]: task for task in pending_tasks} - - class BatchDownloadWorker(QThread): - item_ready = Signal(str, object) - finished = Signal() - downloader: Downloader - urls: list[str] - local_paths: list[str] - timeout: int - - def run(self): - self.downloader.download_parallel( - self.urls, - self.local_paths, - timeout=self.timeout, - on_result=lambda url, result: self.item_ready.emit(url, result), - ) - self.finished.emit() - - worker = BatchDownloadWorker() - worker.downloader = self.downloader - worker.urls = [task[0] for task in pending_tasks] - worker.local_paths = [task[1] for task in pending_tasks] - worker.timeout = timeout - - def on_item_ready(url: str, result: str | None): - task = pending_tasks_by_url.get(url) - if task is None: - return - _url, _local_path, signal, exe_name, asset_type = task - if result: - logger.info(f"{asset_type.capitalize()} downloaded for {exe_name}: {result}") - else: - logger.debug(f"No {asset_type} found for {exe_name}") - signal.emit(exe_name, result) - - def on_batch_finished(): - self._pending_asset_batch_thread = None - - worker.item_ready.connect(on_item_ready) - worker.finished.connect(on_batch_finished) - worker.finished.connect(worker.deleteLater) - self._pending_asset_batch_thread = worker - worker.start() - - def get_autoinstall_description(self, exe_name: str, lang_code: str = "en") -> str | None: - """Read description from downloaded metadata.txt file for autoinstall game. - - Args: - exe_name: The executable name/script name - lang_code: Language code ("en" or "ru" for description) - - Returns: - Description string or None if not found - """ - xdg_data_home = os.getenv("XDG_DATA_HOME", - os.path.join(os.path.expanduser("~"), ".local", "share")) - autoinstall_root = os.path.join(xdg_data_home, "PortProtonQt", "custom_data", "autoinstall") - metadata_path = os.path.join(autoinstall_root, exe_name, "metadata.txt") - - if not os.path.exists(metadata_path): - return None - - try: - with open(metadata_path, encoding='utf-8') as f: - content = f.read() - - # Parse the metadata content to extract description - # Format: description_en=... or description_ru=... - if lang_code == "ru": - pattern = r'^description_ru=(.*)$' - else: - pattern = r'^description_en=(.*)$' - - match = re.search(pattern, content, re.MULTILINE) - if match: - description = match.group(1).strip() - # Handle potential quoted strings - if description.startswith('"') and description.endswith('"'): - description = description[1:-1] - return description - else: - # Try fallback to the other language if the requested one is not found - fallback_lang = "ru" if lang_code == "en" else "en" - fallback_pattern = rf'^description_{fallback_lang}=(.*)$' - fallback_match = re.search(fallback_pattern, content, re.MULTILINE) - if fallback_match: - description = fallback_match.group(1).strip() - if description.startswith('"') and description.endswith('"'): - description = description[1:-1] - return description - except Exception as e: - logger.error(f"Error reading metadata for {exe_name}: {e}") - - return None - - def parse_autoinstall_script(self, file_path: str) -> tuple[str | None, str | None]: - """Extract display_name from # name comment and exe_name from autoinstall bash script.""" - try: - with open(file_path, encoding='utf-8') as f: - content = f.read() - - # Skip emulators - if re.search(r'#\s*type\s*:\s*emulators', content, re.IGNORECASE): - return None, None - - display_name = None - exe_name = None - - # Extract display_name from "# name:" comment - name_match = re.search(r'#\s*name\s*:\s*(.+)', content, re.IGNORECASE) - if name_match: - display_name = name_match.group(1).strip() - - # --- pw_create_unique_exe --- - pw_match = re.search(r'pw_create_unique_exe(?:\s+["\']([^"\']+)["\'])?', content) - if pw_match: - arg = pw_match.group(1) - if arg: - exe_name = arg.strip() - if not exe_name.lower().endswith(".exe"): - exe_name += ".exe" - else: - export_match = re.search( - r'export\s+PORTWINE_CREATE_SHORTCUT_NAME\s*=\s*["\']([^"\']+)["\']', - content, re.IGNORECASE) - if export_match: - exe_name = f"{export_match.group(1).strip()}.exe" - - else: - portwine_match = None - for line in content.splitlines(): - stripped = line.strip() - if stripped.startswith("#"): - continue - if "PW_EXE_FILE" in stripped and "=" in stripped: - portwine_match = stripped - break - - if portwine_match: - exe_expr = portwine_match.split("=", 1)[1].strip().strip("'\" ") - exe_candidates = re.findall(r'[-\w\s/\\\.]+\.exe', exe_expr) - if exe_candidates: - exe_name = os.path.basename(exe_candidates[-1].strip()) - - - # Fallback - if not display_name and exe_name: - display_name = exe_name - - return display_name, exe_name - - except Exception as e: - logger.error(f"Failed to parse {file_path}: {e}") - return None, None - - def _compute_scripts_signature(self, auto_dir: str) -> str: - """Compute a hash-based signature of the autoinstall scripts to detect changes.""" - if not os.path.exists(auto_dir): - return "" - scripts = sorted(glob.glob(os.path.join(auto_dir, "*"))) - # Simple hash: concatenate sorted filenames and hash - filenames_str = "".join(sorted([os.path.basename(s) for s in scripts])) - return hashlib.md5(filenames_str.encode()).hexdigest() - def _load_autoinstall_cache(self): - """Load cached autoinstall games if fresh and scripts unchanged.""" + """Load cached autoinstall games.""" if self._autoinstall_cache is not None: return self._autoinstall_cache cache_manager = CacheManager() if cache_manager.exists("autoinstall_games_cache"): try: data = cache_manager.load_json("autoinstall_games_cache") - if data: - mod_time = cache_manager.get_file_age("autoinstall_games_cache") - if mod_time is not None and mod_time < AUTOINSTALL_CACHE_DURATION: - start_time = time.time() - cached_signature = data.get("scripts_signature", "") - scripts_path = get_portproton_scripts_path() - auto_dir = os.path.join(scripts_path, "pw_autoinstall") if scripts_path else "" - current_signature = self._compute_scripts_signature(auto_dir) - if time.time() - start_time > 3: - logger.warning("Cache loading took too long, skipping cache") - return None - if cached_signature != current_signature: - logger.info("Scripts signature mismatch; invalidating cache") - return None - self._autoinstall_cache = data["games"] - logger.info(f"Loaded {len(self._autoinstall_cache)} cached autoinstall games") - return self._autoinstall_cache + if isinstance(data, dict): + if data.get("api_url") != AUTOINSTALL_API_URL: + return None + games = data["games"] + if not self._autoinstall_cache_uses_urls(games): + return None + self._autoinstall_cache = games + logger.info(f"Loaded {len(self._autoinstall_cache)} cached autoinstall games") + return self._autoinstall_cache except Exception as e: logger.error(f"Failed to load autoinstall cache: {e}") return None + def _autoinstall_cache_uses_urls(self, games: list) -> bool: + for game in games: + if len(game) <= 5 or not isinstance(game[5], str): + return False + if not game[5].startswith("autoinstall:http"): + return False + return True + def _save_autoinstall_cache(self, games): - """Save parsed autoinstall games to cache with scripts signature.""" + """Save autoinstall games to cache.""" try: cache_manager = CacheManager() - scripts_path = get_portproton_scripts_path() - auto_dir = os.path.join(scripts_path, "pw_autoinstall") if scripts_path else "" - scripts_signature = self._compute_scripts_signature(auto_dir) - data = {"games": games, "scripts_signature": scripts_signature, "timestamp": time.time()} + data = {"games": games, "api_url": AUTOINSTALL_API_URL, "timestamp": time.time()} cache_manager.save_json("autoinstall_games_cache", data) - logger.debug(f"Saved {len(games)} autoinstall games to cache with signature {scripts_signature}") + logger.debug(f"Saved {len(games)} autoinstall games to cache") except Exception as e: logger.error(f"Failed to save autoinstall cache: {e}") - def start_autoinstall_games_load(self, callback: Callable[[list[tuple]], None]) -> QThread | None: + def clear_autoinstall_cache(self) -> None: + """Clear cached autoinstall API data.""" + self._autoinstall_cache = None + cache_manager = CacheManager() + cache_manager.remove("autoinstall_games_cache") + + def _get_autoinstall_lang_code(self) -> str: + try: + current_locale = locale.getlocale()[0] or "en" + except (AttributeError, IndexError, TypeError): + current_locale = "en" + for lang_code in ("ru", "es", "pt"): + if lang_code in current_locale.lower(): + return lang_code + return "en" + + def _get_autoinstall_field(self, game: dict, field: str, lang_code: str) -> str: + value = game.get(f"{field}_{lang_code}") or game.get(f"{field}_en") or game.get(field) + return value if isinstance(value, str) else "" + + def _get_custom_game_dir(self, exe_name: str) -> str: + game_dir = os.path.join(self.custom_data_dir, exe_name) + os.makedirs(game_dir, exist_ok=True) + return game_dir + + def _clean_metadata_value(self, value: str) -> str: + return value.replace("\r", " ").replace("\n", " ").strip() + + def _write_autoinstall_metadata(self, game_data: dict, game_dir: str) -> None: + metadata_path = os.path.join(game_dir, "metadata.txt") + name = game_data.get("name", "") + description = game_data.get("description", "") + try: + with open(metadata_path, "w", encoding="utf-8") as metadata_file: + if isinstance(name, str) and name.strip(): + metadata_file.write(f"name={self._clean_metadata_value(name)}\n") + if isinstance(description, str) and description.strip(): + clean_description = self._clean_metadata_value(description) + metadata_file.write(f"description={clean_description}\n") + except OSError as e: + logger.warning("Failed to write autoinstall metadata %s: %s", metadata_path, e) + + def _get_custom_cover_path(self, game_dir: str, cover_url: str) -> str: + ext = os.path.splitext(urllib.parse.urlparse(cover_url).path)[1].lower() + if ext not in COVER_IMAGE_EXTENSIONS: + ext = ".png" + return os.path.join(game_dir, f"cover{ext}") + + def _cache_autoinstall_cover(self, cover_path: str, game_dir: str) -> str: + if not cover_path.startswith(("http://", "https://")): + return "" + local_path = self._get_custom_cover_path(game_dir, cover_path) + downloaded_path = self.downloader.download(cover_path, local_path, timeout=10) + return downloaded_path or "" + + def _extract_exe_name_from_script_line(self, line: str) -> str: + for part in line.replace("\\", "/").split("/"): + clean_part = part.strip().strip('"\' }') + if clean_part.lower().endswith(".exe"): + return clean_part + return "" + + def _get_autoinstall_exe_name(self, script_path: str) -> str: + install_exe = "" + try: + with open(script_path, encoding="utf-8") as script_file: + for line in script_file: + if "PW_EXE_FILE" in line: + exe_name = self._extract_exe_name_from_script_line(line) + if exe_name: + return exe_name + if "PW_AUTOINSTALL_EXE" not in line: + continue + exe_name = self._extract_exe_name_from_script_line(line) + if exe_name: + install_exe = exe_name + except OSError as e: + logger.warning("Failed to read autoinstall script %s: %s", script_path, e) + return install_exe + + def write_autoinstall_custom_data(self, script_path: str, game_data: dict) -> None: + if not game_data: + return + exe_file = self._get_autoinstall_exe_name(script_path) + if not exe_file: + return + + exe_name = os.path.splitext(os.path.basename(exe_file))[0] + game_dir = self._get_custom_game_dir(exe_name) + self._write_autoinstall_metadata(game_data, game_dir) + cover_path = game_data.get("cover_path", "") + if isinstance(cover_path, str): + self._cache_autoinstall_cover(cover_path, game_dir) + + def _get_autoinstall_script_path(self, ppai_url: str) -> str: + script_name = os.path.basename(urllib.parse.urlparse(ppai_url).path) + if not script_name.endswith(".ppai"): + return "" + game_match = re.search(r"game_(\d+)_", script_name) + cache_key = game_match.group(1) if game_match else hashlib.sha256(ppai_url.encode()).hexdigest() + cache_dir = os.path.join(self.custom_data_dir, "autoinstall", cache_key) + os.makedirs(cache_dir, exist_ok=True) + return os.path.join(cache_dir, script_name) + + def download_autoinstall_script(self, ppai_url: str) -> str: + script_path = self._get_autoinstall_script_path(ppai_url) + if not script_path: + return "" + + temp_path = f"{script_path}.tmp" + try: + session = get_requests_session() + response = session.get(ppai_url, timeout=10) + response.raise_for_status() + with open(temp_path, "w", encoding="utf-8") as script_file: + script_file.write(response.text) + os.replace(temp_path, script_path) + return script_path + except (OSError, requests.RequestException) as e: + logger.warning("Failed to download autoinstall script %s: %s", ppai_url, e) + if os.path.exists(script_path): + return script_path + return "" + + def start_autoinstall_script_download( + self, + ppai_url: str, + callback: Callable[[str], None], + ) -> QThread: + class AutoinstallScriptWorker(QThread): + finished = Signal(str) + api: "PortProtonAPI" + ppai_url: str + + def run(self): + script_path = self.api.download_autoinstall_script(self.ppai_url) + self.finished.emit(script_path) + + worker = AutoinstallScriptWorker() + worker.api = self + worker.ppai_url = ppai_url + worker.finished.connect(callback) + worker.start() + return worker + + def start_autoinstall_custom_data_write( + self, + script_path: str, + game_data: dict, + ) -> QThread: + class AutoinstallCustomDataWorker(QThread): + finished = Signal() + api: "PortProtonAPI" + script_path: str + game_data: dict + + def run(self): + self.api.write_autoinstall_custom_data(self.script_path, self.game_data) + self.finished.emit() + + worker = AutoinstallCustomDataWorker() + worker.api = self + worker.script_path = script_path + worker.game_data = game_data + worker.start() + return worker + + def _create_autoinstall_game_tuple(self, game: dict, lang_code: str) -> tuple | None: + game_id = game.get("id") + if not isinstance(game_id, int): + return None + + display_name = self._get_autoinstall_field(game, "name", lang_code) + if not display_name: + return None + + ppai_url = game.get("ppai_url") + if not isinstance(ppai_url, str): + return None + + description = self._get_autoinstall_field(game, "description", lang_code) + compact_icon = game.get("icon_compact_url") or "" + full_icon = game.get("icon_full_url") or "" + exe_name = f"game_{game_id}" + + return ( + display_name, description, full_icon, "", + "", f"autoinstall:{ppai_url}", "Never", "0h 0m", "", "", 0, 0, + "autoinstall", exe_name, compact_icon, full_icon + ) + + def start_autoinstall_games_load( + self, + callback: Callable[[list[tuple]], None], + force_refresh: bool = False, + ) -> QThread | None: """Start loading auto-install games in a background thread. Returns the thread for management.""" class AutoinstallWorker(QThread): finished = Signal(list) api: "PortProtonAPI" portproton_location: str | None + force_refresh: bool def run(self): - # Check cache in this background thread, not in main thread - start_time = time.time() - cached_games = self.api._load_autoinstall_cache() - # If cache loading took too long (>2 seconds), skip cache and load directly - if time.time() - start_time > 2: - logger.warning("Cache loading took too long, proceeding without cache") - cached_games = None + if not self.force_refresh: + cached_games = self.api._load_autoinstall_cache() + if cached_games is not None: + self.finished.emit(cached_games) + return - if cached_games is not None: - self.finished.emit(cached_games) - return - - # No cache: Load games from scratch games = [] - scripts_path = get_portproton_scripts_path() - if not scripts_path: - logger.info("PortProton scripts path is not resolved for autoinstall games") + try: + session = get_requests_session() + response = session.get(AUTOINSTALL_API_URL, timeout=10) + response.raise_for_status() + data = response.json() + except (ValueError, requests.RequestException) as e: + logger.warning("Failed to load autoinstall API: %s", e) + if self.force_refresh: + cached_games = self.api._load_autoinstall_cache() + if cached_games is not None: + self.finished.emit(cached_games) + return self.finished.emit(games) return - auto_dir = os.path.join(scripts_path, "pw_autoinstall") - if not os.path.exists(auto_dir): - logger.info("PortProton autoinstall scripts directory not found: %s", auto_dir) + api_games = data.get("games", []) if isinstance(data, dict) else [] + if not isinstance(api_games, list): + logger.warning("Invalid autoinstall API response") + if self.force_refresh: + cached_games = self.api._load_autoinstall_cache() + if cached_games is not None: + self.finished.emit(cached_games) + return self.finished.emit(games) return - scripts = sorted(glob.glob(os.path.join(auto_dir, "*"))) - if not scripts: - logger.info("PortProton autoinstall scripts not found in: %s", auto_dir) - self.finished.emit(games) - return - - xdg_data_home = os.getenv( - "XDG_DATA_HOME", - os.path.join(os.path.expanduser("~"), ".local", "share"), - ) - base_autoinstall_dir = os.path.join( - xdg_data_home, "PortProtonQt", "custom_data", "autoinstall" - ) - os.makedirs(base_autoinstall_dir, exist_ok=True) - - for script_path in scripts: - display_name, exe_name = self.api.parse_autoinstall_script(script_path) - script_name = os.path.splitext(os.path.basename(script_path))[0] - - if not (display_name and exe_name): + lang_code = self.api._get_autoinstall_lang_code() + for game in api_games: + if not isinstance(game, dict): continue - - exe_name = os.path.splitext(exe_name)[0] - user_game_folder = os.path.join(base_autoinstall_dir, exe_name) - os.makedirs(user_game_folder, exist_ok=True) - - # Find cover - cover_path = "" - user_files = ( - set(os.listdir(user_game_folder)) - if os.path.exists(user_game_folder) - else set() - ) - for ext in COVER_IMAGE_EXTENSIONS: - candidate = f"cover{ext}" - if candidate in user_files: - cover_path = os.path.join(user_game_folder, candidate) - break - - if not cover_path: - logger.debug(f"No local cover found for autoinstall {exe_name}") - - # Try to get the description from metadata file - description = "" - # Look for metadata in the expected location - try: - current_locale = locale.getlocale()[0] or 'en' - except (AttributeError, IndexError, TypeError): - current_locale = 'en' - lang_code = 'ru' if current_locale and 'ru' in current_locale.lower() else 'en' - - # Try to read description from downloaded metadata - metadata_description = self.api.get_autoinstall_description(exe_name, lang_code) - if metadata_description: - description = metadata_description - - game_tuple = ( - display_name, description, cover_path, "", - "", f"autoinstall:{script_name}", "Never", "0h 0m", "", "", 0, 0, "autoinstall", exe_name - ) - games.append(game_tuple) + game_tuple = self.api._create_autoinstall_game_tuple(game, lang_code) + if game_tuple is not None: + games.append(game_tuple) self.api._save_autoinstall_cache(games) self.api._autoinstall_cache = games @@ -722,6 +476,7 @@ class PortProtonAPI: worker = AutoinstallWorker() worker.api = self worker.portproton_location = self.portproton_location + worker.force_refresh = force_refresh worker.finished.connect(lambda games: callback(games)) worker.start() logger.info("Started background load of autoinstall games") diff --git a/tests/test_main_window.py b/tests/test_main_window.py new file mode 100644 index 00000000..cf0b7c5d --- /dev/null +++ b/tests/test_main_window.py @@ -0,0 +1,50 @@ +"""Tests for main window library data processing.""" + +from pathlib import Path +from typing import Any + +from portprotonqt.main_window import MainWindow + + +def test_process_portproton_desktop_calls_callback_without_asset_download( + tmp_config_dir: Path, + tmp_path: Path, + monkeypatch: Any, +) -> None: + exe_path = tmp_path / "Game.exe" + exe_path.write_text("", encoding="utf-8") + desktop_path = tmp_path / "Game.desktop" + desktop_path.write_text( + "[Desktop Entry]\n" + "Name=Test Game\n" + f"Exec=portproton {exe_path}\n" + "Icon=\n", + encoding="utf-8", + ) + + window = MainWindow.__new__(MainWindow) + window.portproton_location = str(tmp_path) + results: list[tuple | None] = [] + + def fake_steam_info(_name: str, _exec_line: str, callback: Any) -> None: + callback({}) + + monkeypatch.setattr( + "portprotonqt.main_window.generate_thumbnail", + lambda *_args, **_kwargs: False, + ) + monkeypatch.setattr("portprotonqt.main_window.get_steam_game_info_async", fake_steam_info) + monkeypatch.setattr("portprotonqt.main_window.get_last_launch", lambda _exe_name: "Never") + monkeypatch.setattr( + "portprotonqt.main_window.get_last_launch_timestamp", + lambda _exe_name: 0, + ) + monkeypatch.setattr("portprotonqt.main_window.get_playtime_for_exe", lambda *_args: None) + monkeypatch.setattr("portprotonqt.main_window.ui_config.get_economy_mode", lambda: False) + + window._process_desktop_file_async(str(desktop_path), results.append) + + assert len(results) == 1 + assert results[0] is not None + assert results[0][0] == "Test Game" + assert results[0][5] == f"portproton {exe_path}" diff --git a/tests/test_portproton_api.py b/tests/test_portproton_api.py new file mode 100644 index 00000000..fe3d4013 --- /dev/null +++ b/tests/test_portproton_api.py @@ -0,0 +1,110 @@ +"""Tests for PortProton API helpers.""" + +from pathlib import Path +from typing import Any, cast + +from portprotonqt.portproton_api import PortProtonAPI + + +class DummyDownloader: + def __init__(self) -> None: + self.downloads: list[tuple[str, str, int]] = [] + + def download(self, url: str, local_path: str, timeout: int = 5) -> str: + self.downloads.append((url, local_path, timeout)) + Path(local_path).write_text("cover", encoding="utf-8") + return local_path + + +def test_autoinstall_description_falls_back_to_english(tmp_config_dir: Path) -> None: + api = PortProtonAPI() + game = { + "name_ru": None, + "description_en": "Launcher for the VK Play game library.", + } + + assert api._get_autoinstall_field(game, "description", "ru") == "Launcher for the VK Play game library." + + +def test_autoinstall_name_falls_back_to_plain_field(tmp_config_dir: Path) -> None: + api = PortProtonAPI() + game = { + "name": "VK Play", + "name_en": None, + "name_ru": None, + } + + assert api._get_autoinstall_field(game, "name", "ru") == "VK Play" + + +def test_autoinstall_script_writes_custom_metadata_for_target_exe( + tmp_config_dir: Path, + tmp_path: Path, +) -> None: + api = PortProtonAPI(downloader=cast(Any, DummyDownloader())) + game_data = { + "name": "Test Game", + "description": "English description", + } + script_path = tmp_path / "game_42_test.ppai" + script_path.write_text( + 'export PW_AUTOINSTALL_EXE="${PORT_DATA_PATH}/data/prefixes/GAME/drive_c/Game.exe"\n', + encoding="utf-8", + ) + + api.write_autoinstall_custom_data(str(script_path), game_data) + + metadata_path = tmp_config_dir.parent / "data" / "PortProtonQt" / "custom_data" + metadata_path = metadata_path / "Game" / "metadata.txt" + assert metadata_path.exists() + assert "name=Test Game" in metadata_path.read_text(encoding="utf-8") + assert "description=English description" in metadata_path.read_text(encoding="utf-8") + + +def test_autoinstall_script_caches_cover_for_target_exe( + tmp_config_dir: Path, + tmp_path: Path, +) -> None: + downloader = DummyDownloader() + api = PortProtonAPI(downloader=cast(Any, downloader)) + game_data = { + "name": "Cover Game", + "description": "Description", + "cover_path": "https://ppdb.linux-gaming.ru/covers/game_43.webp", + } + script_path = tmp_path / "game_43_test.ppai" + script_path.write_text( + 'PW_EXE_FILE="$WINEPREFIX/drive_c/users/steamuser/AppData/Local/VKPlayLoader.exe"\n', + encoding="utf-8", + ) + + api.write_autoinstall_custom_data(str(script_path), game_data) + + cover_path = tmp_config_dir.parent / "data" / "PortProtonQt" / "custom_data" + cover_path = cover_path / "VKPlayLoader" / "cover.webp" + assert cover_path.exists() + assert downloader.downloads[0][1] == str(cover_path) + assert downloader.downloads[0][0] == game_data["cover_path"] + + +def test_autoinstall_script_uses_cached_card_data( + tmp_config_dir: Path, + tmp_path: Path, +) -> None: + api = PortProtonAPI(downloader=cast(Any, DummyDownloader())) + game_data = { + "name": "Cached Game", + "description": "Cached description", + } + script_path = tmp_path / "game_44_test.ppai" + script_path.write_text( + 'PW_EXE_FILE="$WINEPREFIX/drive_c/CachedLauncher.exe"\n', + encoding="utf-8", + ) + + api.write_autoinstall_custom_data(str(script_path), game_data) + + metadata_path = tmp_config_dir.parent / "data" / "PortProtonQt" / "custom_data" + metadata_path = metadata_path / "CachedLauncher" / "metadata.txt" + assert metadata_path.exists() + assert "name=Cached Game" in metadata_path.read_text(encoding="utf-8") -- 2.54.0