feat: load autoinstalls from PPDB API #201

Merged
Boria138 merged 1 commits from ppdb-api into main 2026-06-17 12:13:23 +00:00
11 changed files with 627 additions and 657 deletions

No files matched your search

+4
View File
@@ -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)
@@ -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"
+12 -9
View File
@@ -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
+1 -16
View File
@@ -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
+2 -4
View File
@@ -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:
+16 -1
View File
@@ -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
+35 -14
View File
@@ -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
+124 -115
View File
@@ -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)
+253 -498
View File
@@ -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")
+50
View File
@@ -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}"
+110
View File
@@ -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")