feat: load autoinstalls from PPDB API #201
No files matched your search
@@ -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"
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
@@ -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
@@ -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")
|
||||
|
||||
@@ -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}"
|
||||
@@ -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")
|
||||
Reference in new issue
Block a user