feat: added gog integration #229
No files matched your search
+19
-1
@@ -290,6 +290,7 @@ def main():
|
||||
restore_prefix_path = None
|
||||
autoinstall_path = None
|
||||
theme_store_id = None
|
||||
gog_launch_uri = None
|
||||
resolution_from_args = None
|
||||
if args.resolution:
|
||||
resolution_from_args = parse_resolution(args.resolution)
|
||||
@@ -315,6 +316,11 @@ def main():
|
||||
elif args.file_or_url and is_autoinstall_file(args.file_or_url):
|
||||
autoinstall_path = normalize_launch_path(args.file_or_url)
|
||||
ipc_message = f"autoinstall:{quote(autoinstall_path, safe='')}"
|
||||
elif args.file_or_url and args.file_or_url.startswith("gog://launch/"):
|
||||
app_id = args.file_or_url.removeprefix("gog://launch/")
|
||||
if app_id.isdigit():
|
||||
gog_launch_uri = args.file_or_url
|
||||
ipc_message = f"gog:{app_id}"
|
||||
elif args.file_or_url:
|
||||
theme_store_id = parse_portprotonqt_theme_url(args.file_or_url)
|
||||
if theme_store_id is not None:
|
||||
@@ -381,7 +387,7 @@ def main():
|
||||
logger.warning("Failed to persist PORT_DATA_PATH in PortProtonQt config")
|
||||
|
||||
# Check if we have a portproton:// URL or launch file to handle
|
||||
if args.file_or_url and not restore_prefix_path:
|
||||
if args.file_or_url and not restore_prefix_path and not gog_launch_uri:
|
||||
if is_portproton_url(args.file_or_url):
|
||||
# Parse the portproton:// URL to get the full download URL
|
||||
download_url = parse_portproton_url(args.file_or_url)
|
||||
@@ -485,6 +491,10 @@ def main():
|
||||
def handle_restore_prefix():
|
||||
window._perform_restore(restore_prefix_path)
|
||||
QTimer.singleShot(0, handle_restore_prefix)
|
||||
elif gog_launch_uri:
|
||||
def handle_gog_launch():
|
||||
window.toggleGame(gog_launch_uri)
|
||||
QTimer.singleShot(0, handle_gog_launch)
|
||||
|
||||
# --- Handle incoming connections ---
|
||||
def handle_new_connection():
|
||||
@@ -507,6 +517,7 @@ def main():
|
||||
or msg.startswith("backup:")
|
||||
or msg.startswith("theme:")
|
||||
or msg.startswith("autoinstall:")
|
||||
or msg.startswith("gog:")
|
||||
):
|
||||
# Ensure the window is visible and not minimized
|
||||
window.setWindowState(window.windowState() & ~Qt.WindowState.WindowMinimized)
|
||||
@@ -542,6 +553,12 @@ def main():
|
||||
window._perform_restore(backup_path)
|
||||
else:
|
||||
logger.warning("Invalid prefix backup via IPC: %s", backup_path)
|
||||
elif msg.startswith("gog:"):
|
||||
app_id = msg[4:].strip()
|
||||
if app_id.isdigit():
|
||||
window.toggleGame(f"gog://launch/{app_id}")
|
||||
else:
|
||||
logger.warning("Invalid GOG app id via IPC: %s", app_id)
|
||||
elif msg.startswith("backup:"):
|
||||
parts = msg.split(":", 2)
|
||||
if len(parts) == 3:
|
||||
@@ -593,6 +610,7 @@ def main():
|
||||
and restore_prefix_path is None
|
||||
and autoinstall_path is None
|
||||
and theme_store_id is None
|
||||
and gog_launch_uri is None
|
||||
)
|
||||
if launch_minimized:
|
||||
logger.info("Launching in tray")
|
||||
|
||||
@@ -528,10 +528,12 @@ def resolve_custom_data_dir(custom_data_root: str, exe_path: str) -> str:
|
||||
def create_desktop_file(
|
||||
exe_path: str,
|
||||
game_name: str | None = None,
|
||||
icon_source: str | None = None,
|
||||
) -> tuple[str, str, str] | None:
|
||||
"""Create desktop entry content, destination path and icon path for a game."""
|
||||
portproton_path = get_portproton_location()
|
||||
if not os.path.isfile(exe_path):
|
||||
is_gog_uri = re.fullmatch(r"gog://launch/\d+", exe_path) is not None
|
||||
if not is_gog_uri and not os.path.isfile(exe_path):
|
||||
logger.error("Executable not found: %s", exe_path)
|
||||
return None
|
||||
if not portproton_path:
|
||||
@@ -542,9 +544,14 @@ def create_desktop_file(
|
||||
game_name = os.path.splitext(os.path.basename(exe_path))[0]
|
||||
base_path = os.path.join(portproton_path, "data")
|
||||
icon_name = _sanitize_icon_name(game_name)
|
||||
icon_path = os.path.join(base_path, "img", f"{icon_name}.png")
|
||||
has_icon_source = bool(icon_source and os.path.isfile(icon_source))
|
||||
icon_path = (
|
||||
"applications-games" if is_gog_uri and not has_icon_source
|
||||
else os.path.join(base_path, "img", f"{icon_name}.png")
|
||||
)
|
||||
desktop_path = os.path.join(portproton_path, f"{game_name}.desktop")
|
||||
os.makedirs(os.path.dirname(icon_path), exist_ok=True)
|
||||
if icon_path != "applications-games":
|
||||
os.makedirs(os.path.dirname(icon_path), exist_ok=True)
|
||||
|
||||
flatpak_id = os.getenv("FLATPAK_ID")
|
||||
appimage_path = os.getenv("APPIMAGE", "").strip()
|
||||
|
||||
@@ -459,7 +459,7 @@ class ContextMenuManager:
|
||||
menu.setStyleSheet(self.theme.CONTEXT_MENU_STYLE)
|
||||
|
||||
# For non-Steam games, check if exe exists
|
||||
if game_card.game_source != "steam":
|
||||
if game_card.game_source == "portproton":
|
||||
exec_line = self._get_exec_line(game_card.name, game_card.exec_line)
|
||||
exe_path = self._parse_exe_path(exec_line, game_card.name) if exec_line else None
|
||||
if not exe_path:
|
||||
@@ -470,13 +470,18 @@ class ContextMenuManager:
|
||||
return
|
||||
|
||||
# Normal menu for games with valid exe or from Steam
|
||||
is_running = self._is_game_running(game_card)
|
||||
action_text = _("Stop Game") if is_running else _("Launch Game")
|
||||
action_icon = "stop" if is_running else "play"
|
||||
launch_action = menu.addAction(self._get_safe_icon(action_icon), action_text)
|
||||
launch_action.triggered.connect(
|
||||
lambda: self._launch_game(game_card)
|
||||
is_uninstalled_gog = (
|
||||
game_card.game_source == "gog"
|
||||
and game_card.exec_line.startswith("gog://install/")
|
||||
)
|
||||
is_running = self._is_game_running(game_card)
|
||||
if not is_uninstalled_gog:
|
||||
action_text = _("Stop Game") if is_running else _("Launch Game")
|
||||
action_icon = "stop" if is_running else "play"
|
||||
launch_action = menu.addAction(self._get_safe_icon(action_icon), action_text)
|
||||
launch_action.triggered.connect(
|
||||
lambda: self._launch_game(game_card)
|
||||
)
|
||||
|
||||
favorites = favorites_config.get_games()
|
||||
is_favorite = game_card.name in favorites
|
||||
@@ -485,6 +490,60 @@ class ContextMenuManager:
|
||||
favorite_action = menu.addAction(self._get_safe_icon(icon_name), text)
|
||||
favorite_action.triggered.connect(lambda: self.toggle_favorite(game_card, not is_favorite))
|
||||
|
||||
if game_card.game_source == "gog":
|
||||
game = {
|
||||
"app_id": str(game_card.appid),
|
||||
"title": game_card.name,
|
||||
"cover": game_card.cover_path,
|
||||
}
|
||||
if game_card.exec_line.startswith("gog://launch/"):
|
||||
repair_action = menu.addAction(
|
||||
self._get_safe_icon("update"), _("Repair")
|
||||
)
|
||||
repair_action.triggered.connect(
|
||||
lambda: self.parent._repair_gog_game(game)
|
||||
)
|
||||
desktop_dir = QStandardPaths.writableLocation(
|
||||
QStandardPaths.StandardLocation.DesktopLocation
|
||||
)
|
||||
desktop_path = self._get_shortcut_path(
|
||||
game_card.name, desktop_dir
|
||||
)
|
||||
desktop_exists = os.path.exists(desktop_path)
|
||||
desktop_action = menu.addAction(
|
||||
self._get_safe_icon("delete" if desktop_exists else "desktop"),
|
||||
_("Remove from Desktop") if desktop_exists else _("Add to Desktop"),
|
||||
)
|
||||
desktop_action.triggered.connect(
|
||||
lambda: self.remove_from_desktop(game_card.name)
|
||||
if desktop_exists
|
||||
else self.add_to_desktop(game_card.name, game_card.exec_line)
|
||||
)
|
||||
menu_path = self._get_menu_shortcut_path(game_card.name)
|
||||
menu_exists = os.path.exists(menu_path)
|
||||
menu_action = menu.addAction(
|
||||
self._get_safe_icon("delete" if menu_exists else "menu"),
|
||||
_("Remove from Menu") if menu_exists else _("Add to Menu"),
|
||||
)
|
||||
menu_action.triggered.connect(
|
||||
lambda: self.remove_from_menu(game_card.name)
|
||||
if menu_exists
|
||||
else self.add_to_menu(game_card.name, game_card.exec_line)
|
||||
)
|
||||
delete_action = menu.addAction(
|
||||
self._get_safe_icon("delete"), _("Delete")
|
||||
)
|
||||
delete_action.triggered.connect(
|
||||
lambda: self.parent._delete_gog_game(game)
|
||||
)
|
||||
else:
|
||||
import_action = menu.addAction(
|
||||
self._get_safe_icon("folder"), _("Import")
|
||||
)
|
||||
import_action.triggered.connect(
|
||||
lambda: self.parent._import_gog_game(game)
|
||||
)
|
||||
|
||||
if game_card.game_source == "steam":
|
||||
desktop_dir = QStandardPaths.writableLocation(
|
||||
QStandardPaths.StandardLocation.DesktopLocation
|
||||
@@ -524,7 +583,7 @@ class ContextMenuManager:
|
||||
)
|
||||
)
|
||||
|
||||
if game_card.game_source != "steam":
|
||||
if game_card.game_source == "portproton":
|
||||
desktop_dir = QStandardPaths.writableLocation(QStandardPaths.StandardLocation.DesktopLocation)
|
||||
desktop_path = self._get_shortcut_path(game_card.name, desktop_dir)
|
||||
icon_name = "delete" if os.path.exists(desktop_path) else "desktop"
|
||||
@@ -916,23 +975,33 @@ class ContextMenuManager:
|
||||
"""Add the game desktop entry to the XDG applications directory."""
|
||||
if not self._check_portproton():
|
||||
return
|
||||
is_gog_uri = str(exec_line).startswith("gog://launch/")
|
||||
desktop_path = self._get_desktop_path(game_name)
|
||||
if not os.path.exists(desktop_path):
|
||||
if not is_gog_uri and not os.path.exists(desktop_path):
|
||||
self.signals.show_warning_dialog.emit(
|
||||
_("Error"),
|
||||
_("No .desktop file found for '{game_name}'").format(game_name=game_name)
|
||||
)
|
||||
return
|
||||
exec_line = self._get_exec_line(game_name, exec_line)
|
||||
if not exec_line:
|
||||
return
|
||||
exe_path = self._parse_exe_path(exec_line, game_name)
|
||||
if not exe_path:
|
||||
return
|
||||
shortcut = create_desktop_file(exe_path, game_name)
|
||||
if not is_gog_uri:
|
||||
exec_line = self._get_exec_line(game_name, exec_line)
|
||||
if not exec_line:
|
||||
return
|
||||
exec_line = self._parse_exe_path(exec_line, game_name)
|
||||
if not exec_line:
|
||||
return
|
||||
icon_source = None
|
||||
if is_gog_uri:
|
||||
app_id = str(exec_line).rsplit("/", 1)[-1]
|
||||
icon_source = self.parent.gog_api.get_launch_target(app_id)
|
||||
shortcut = create_desktop_file(exec_line, game_name, icon_source)
|
||||
if not shortcut:
|
||||
return
|
||||
desktop_entry = shortcut[0]
|
||||
icon_path = shortcut[2]
|
||||
if icon_source and not os.path.exists(icon_path):
|
||||
if not generate_thumbnail(icon_source, icon_path, size=128):
|
||||
logger.error("Failed to generate thumbnail for game: %s", icon_source)
|
||||
applications_dir = self._get_applications_dir()
|
||||
os.makedirs(applications_dir, exist_ok=True)
|
||||
dest_path = self._get_menu_shortcut_path(game_name)
|
||||
@@ -975,28 +1044,35 @@ class ContextMenuManager:
|
||||
|
||||
if not self._check_portproton():
|
||||
return
|
||||
is_gog_uri = str(exec_line).startswith("gog://launch/")
|
||||
desktop_path = self._get_desktop_path(game_name)
|
||||
if not os.path.exists(desktop_path):
|
||||
if not is_gog_uri and not os.path.exists(desktop_path):
|
||||
self.signals.show_warning_dialog.emit(
|
||||
_("Error"),
|
||||
_("No .desktop file found for '{game_name}'").format(game_name=game_name)
|
||||
)
|
||||
return
|
||||
# Ensure icon exists
|
||||
exec_line = self._get_exec_line(game_name, exec_line)
|
||||
if not exec_line:
|
||||
return
|
||||
exe_path = self._parse_exe_path(exec_line, game_name)
|
||||
if not exe_path:
|
||||
return
|
||||
shortcut = create_desktop_file(exe_path, game_name)
|
||||
if not is_gog_uri:
|
||||
exec_line = self._get_exec_line(game_name, exec_line)
|
||||
if not exec_line:
|
||||
return
|
||||
exec_line = self._parse_exe_path(exec_line, game_name)
|
||||
if not exec_line:
|
||||
return
|
||||
icon_source = None
|
||||
if is_gog_uri:
|
||||
app_id = str(exec_line).rsplit("/", 1)[-1]
|
||||
icon_source = self.parent.gog_api.get_launch_target(app_id)
|
||||
shortcut = create_desktop_file(exec_line, game_name, icon_source)
|
||||
if not shortcut:
|
||||
return
|
||||
desktop_entry = shortcut[0]
|
||||
icon_path = shortcut[2]
|
||||
if not os.path.exists(icon_path):
|
||||
if not generate_thumbnail(exe_path, icon_path, size=128):
|
||||
logger.error("Failed to generate thumbnail for game: %s", exe_path)
|
||||
thumbnail_source = icon_source or exec_line
|
||||
if icon_path != "applications-games" and not os.path.exists(icon_path):
|
||||
if not generate_thumbnail(thumbnail_source, icon_path, size=128):
|
||||
logger.error("Failed to generate thumbnail for game: %s", thumbnail_source)
|
||||
|
||||
desktop_dir = QStandardPaths.writableLocation(QStandardPaths.StandardLocation.DesktopLocation)
|
||||
os.makedirs(desktop_dir, exist_ok=True)
|
||||
@@ -1074,6 +1150,20 @@ class ContextMenuManager:
|
||||
|
||||
fetch_client_icon_async(appid_str, write_shortcut)
|
||||
|
||||
def remove_gog_shortcuts(self, game_name: str) -> None:
|
||||
"""Remove desktop and application menu shortcuts for a GOG game."""
|
||||
desktop_dir = QStandardPaths.writableLocation(
|
||||
QStandardPaths.StandardLocation.DesktopLocation
|
||||
)
|
||||
for target_dir in (desktop_dir, self._get_applications_dir()):
|
||||
shortcut_path = self._get_shortcut_path(game_name, target_dir)
|
||||
try:
|
||||
os.remove(shortcut_path)
|
||||
except FileNotFoundError:
|
||||
continue
|
||||
except OSError as error:
|
||||
logger.warning("Failed to remove GOG shortcut %s: %s", shortcut_path, error)
|
||||
|
||||
def add_steam_to_desktop(self, game_name: str, appid: int | str) -> None:
|
||||
"""Create a desktop shortcut for an installed Steam game."""
|
||||
desktop_dir = QStandardPaths.writableLocation(
|
||||
|
||||
@@ -1,7 +1,10 @@
|
||||
"""Detail pages for PortProtonQt."""
|
||||
|
||||
import os
|
||||
import weakref
|
||||
from collections.abc import Callable
|
||||
|
||||
from shiboken6 import isValid
|
||||
from PySide6.QtWidgets import (
|
||||
QWidget,
|
||||
QLabel,
|
||||
@@ -13,7 +16,7 @@ from PySide6.QtWidgets import (
|
||||
QBoxLayout,
|
||||
QSizePolicy,
|
||||
)
|
||||
from PySide6.QtCore import Qt, QUrl, QTimer
|
||||
from PySide6.QtCore import Qt, QUrl, QTimer, QProcess, QProcessEnvironment
|
||||
from PySide6.QtGui import QDesktopServices
|
||||
|
||||
from portprotonqt.detail_pages.widgets import (
|
||||
@@ -80,11 +83,15 @@ class DetailPageManager:
|
||||
self._autoinstall_status_controls = None
|
||||
self._current_detail_source: tuple[str, dict] | None = None
|
||||
self._resize_rebuild_pending = False
|
||||
self._gog_size_requests: dict[
|
||||
str, tuple[QProcess, list[list[Callable[[], QLabel | None]]]]
|
||||
] = {}
|
||||
|
||||
def openGameDetailPage(self, game_data: dict) -> None:
|
||||
"""Open detailed game information page."""
|
||||
self._current_detail_source = ("game", dict(game_data))
|
||||
detail_page = QWidget()
|
||||
detail_page.setProperty("coverCacheKey", str(game_data.get("appid", "")))
|
||||
fallback_exe, fallback_icon_path = self._get_exe_icon_fallback(game_data)
|
||||
detail_page.setProperty("fallbackExe", fallback_exe)
|
||||
detail_page.setProperty("fallbackIconPath", fallback_icon_path)
|
||||
@@ -301,12 +308,97 @@ class DetailPageManager:
|
||||
)
|
||||
game_info_layout.addLayout(first_row)
|
||||
|
||||
self._setup_gog_size_data(game_data, game_info_layout)
|
||||
|
||||
hltb_layout = QHBoxLayout()
|
||||
hltb_layout.setSpacing(10)
|
||||
self._setup_hltb_data(game_data.get("name", ""), hltb_layout, game_info_layout)
|
||||
|
||||
return game_info_layout
|
||||
|
||||
def _setup_gog_size_data(
|
||||
self, game_data: dict, game_info_layout: QVBoxLayout
|
||||
) -> None:
|
||||
exec_line = str(game_data.get("exec_line", ""))
|
||||
if str(game_data.get("game_source", "")).lower() != "gog":
|
||||
return
|
||||
if not exec_line.startswith("gog://install/"):
|
||||
return
|
||||
row = QHBoxLayout()
|
||||
values = []
|
||||
for title_text in (_("Download"), _("Size")):
|
||||
title = QLabel(title_text.upper())
|
||||
title.setStyleSheet(self.main_window.theme.LAST_LAUNCH_TITLE_STYLE)
|
||||
value = QLabel("—")
|
||||
value.setStyleSheet(self.main_window.theme.LAST_LAUNCH_VALUE_STYLE)
|
||||
row.addWidget(title)
|
||||
row.addWidget(value)
|
||||
row.addSpacing(30)
|
||||
values.append(weakref.ref(value))
|
||||
game_info_layout.addLayout(row)
|
||||
app_id = exec_line.rsplit("/", 1)[-1]
|
||||
cached = self.main_window.gog_api.get_cached_download_sizes(app_id)
|
||||
if cached is not None:
|
||||
self._set_gog_size_values(values, cached)
|
||||
return
|
||||
request = self._gog_size_requests.get(app_id)
|
||||
if request is not None:
|
||||
request[1].append(values)
|
||||
return
|
||||
try:
|
||||
command = self.main_window.gog_api.build_command(
|
||||
["info", app_id, "--platform", "windows", "--skip-dlcs"]
|
||||
)
|
||||
except FileNotFoundError as error:
|
||||
logger.warning("Failed to get GOG download size: %s", error)
|
||||
return
|
||||
process = QProcess(self.main_window)
|
||||
process.setProgram(command[0])
|
||||
process.setArguments(command[1:])
|
||||
environment = QProcessEnvironment.systemEnvironment()
|
||||
environment.insert("GOGDL_CONFIG_PATH", str(self.main_window.gog_api.config_dir))
|
||||
process.setProcessEnvironment(environment)
|
||||
self._gog_size_requests[app_id] = (process, [values])
|
||||
process.finished.connect(
|
||||
lambda code, _status: self._finish_gog_size_request(app_id, code)
|
||||
)
|
||||
process.start()
|
||||
|
||||
def _finish_gog_size_request(self, app_id: str, code: int) -> None:
|
||||
request = self._gog_size_requests.pop(app_id, None)
|
||||
if request is None:
|
||||
return
|
||||
process, value_groups = request
|
||||
if code != 0:
|
||||
logger.warning("gogdl info exited with code %s", code)
|
||||
process.deleteLater()
|
||||
return
|
||||
try:
|
||||
sizes = self.main_window.gog_api.parse_download_sizes(
|
||||
bytes(process.readAllStandardOutput())
|
||||
)
|
||||
except (ValueError, TypeError) as error:
|
||||
logger.warning("Failed to parse GOG download size: %s", error)
|
||||
process.deleteLater()
|
||||
return
|
||||
self.main_window.gog_api.save_download_sizes(app_id, sizes)
|
||||
for value_refs in value_groups:
|
||||
self._set_gog_size_values(value_refs, sizes)
|
||||
process.deleteLater()
|
||||
|
||||
@staticmethod
|
||||
def _set_gog_size_values(
|
||||
value_refs: list[Callable[[], QLabel | None]], sizes: tuple[int, int]
|
||||
) -> None:
|
||||
for value_ref, size in zip(value_refs, sizes, strict=True):
|
||||
label = value_ref()
|
||||
if label is None or not isValid(label):
|
||||
continue
|
||||
divisor, unit = (
|
||||
(1024 ** 3, "GiB") if size >= 1024 ** 3 else (1024 ** 2, "MiB")
|
||||
)
|
||||
label.setText(f"{size / divisor:.1f} {unit}")
|
||||
|
||||
def _create_playtime_row(self, last_launch: str, formatted_playtime: str) -> QHBoxLayout:
|
||||
"""Create first row with last launch and playtime."""
|
||||
first_row = QHBoxLayout()
|
||||
@@ -438,6 +530,11 @@ class DetailPageManager:
|
||||
|
||||
current_exe = self._get_current_exe(exec_line)
|
||||
play_button = self._create_play_button(exec_line, current_exe)
|
||||
if str(game_source).lower() == "gog" and exec_line.startswith("gog://install/"):
|
||||
play_button.setText(_("Install"))
|
||||
play_button.setIcon(
|
||||
self.main_window.theme_manager.get_icon("update", as_path=True)
|
||||
)
|
||||
buttons_layout.addWidget(play_button)
|
||||
|
||||
if str(game_source).lower() == "steam" and appid:
|
||||
@@ -471,7 +568,10 @@ class DetailPageManager:
|
||||
lambda: self._open_steam_game_folder(str(appid))
|
||||
)
|
||||
buttons_layout.addWidget(open_folder_button)
|
||||
elif not self._has_game_shortcut(game_name):
|
||||
elif (
|
||||
str(game_source).lower() != "gog"
|
||||
and not self._has_game_shortcut(game_name)
|
||||
):
|
||||
add_button = self._make_action_button(
|
||||
_("Add a shortcut"),
|
||||
self.main_window.theme_manager.get_icon("addgame", as_path=True),
|
||||
@@ -486,6 +586,40 @@ class DetailPageManager:
|
||||
# Show settings button for PortProton games or Steam games using PortProtonQt
|
||||
if str(game_source).lower() == "portproton":
|
||||
self._add_portproton_buttons(buttons_layout, exec_line)
|
||||
elif (
|
||||
str(game_source).lower() == "gog"
|
||||
and appid
|
||||
and exec_line.startswith("gog://launch/")
|
||||
):
|
||||
gog_exe = self.main_window.gog_api.get_launch_target(str(appid))
|
||||
if not gog_exe:
|
||||
return buttons_layout
|
||||
self.main_window.gog_api.ensure_launch_parameters(str(appid))
|
||||
settings_button = self._make_action_button(
|
||||
_("Settings"),
|
||||
self.main_window.theme_manager.get_icon("settings", as_path=True),
|
||||
)
|
||||
settings_button.clicked.connect(
|
||||
lambda: self.main_window.open_exe_settings(
|
||||
gog_exe, game_source="gog",
|
||||
)
|
||||
)
|
||||
buttons_layout.addWidget(settings_button)
|
||||
log_button = self._make_action_button(
|
||||
_("Create Log"),
|
||||
self.main_window.theme_manager.get_icon("edit", as_path=True),
|
||||
)
|
||||
log_button.clicked.connect(lambda: self.toggleDebugLog(gog_exe, log_button))
|
||||
buttons_layout.addWidget(log_button)
|
||||
self._debug_log_button = log_button
|
||||
folder_button = self._make_action_button(
|
||||
_("Open Folder"),
|
||||
self.main_window.theme_manager.get_icon("search", as_path=True),
|
||||
)
|
||||
folder_button.clicked.connect(
|
||||
lambda: self._open_executable_folder(gog_exe)
|
||||
)
|
||||
buttons_layout.addWidget(folder_button)
|
||||
elif str(game_source).lower() == "steam" and appid:
|
||||
try:
|
||||
compat_tool = get_steam_compat_tool(int(appid))
|
||||
@@ -590,9 +724,13 @@ class DetailPageManager:
|
||||
def _open_portproton_game_folder(self, exec_line: str) -> None:
|
||||
"""Open folder containing the PortProton game executable."""
|
||||
file_to_check = self._get_file_from_exec(exec_line)
|
||||
if not file_to_check or not os.path.exists(file_to_check):
|
||||
self._open_executable_folder(file_to_check)
|
||||
|
||||
def _open_executable_folder(self, exe_path: str | None) -> None:
|
||||
"""Open the folder containing an executable."""
|
||||
if not exe_path or not os.path.exists(exe_path):
|
||||
return
|
||||
folder_path = os.path.dirname(os.path.abspath(file_to_check))
|
||||
folder_path = os.path.dirname(os.path.abspath(exe_path))
|
||||
QDesktopServices.openUrl(QUrl.fromLocalFile(folder_path))
|
||||
|
||||
def _has_game_shortcut(self, game_name: str) -> bool:
|
||||
|
||||
@@ -64,6 +64,7 @@ def setup_image_loading(
|
||||
cover_width,
|
||||
cover_height,
|
||||
lambda pixmap: relay.pixmap_ready.emit(pixmap),
|
||||
app_name=str(detail_page.property("coverCacheKey") or ""),
|
||||
fallback_exe=fallback_exe,
|
||||
fallback_icon_path=fallback_icon_path,
|
||||
)
|
||||
|
||||
@@ -139,10 +139,10 @@ def _add_source_corner(
|
||||
) -> None:
|
||||
"""Add source corner badge to cover frame."""
|
||||
source = game_source.lower()
|
||||
if source not in ("steam", "portproton") or theme_manager is None:
|
||||
if source not in ("steam", "gog", "portproton") or theme_manager is None:
|
||||
return
|
||||
|
||||
icon_name = "badge_steam" if source == "steam" else "badge_portproton"
|
||||
icon_name = f"badge_{source}"
|
||||
icon = theme_manager.get_icon(icon_name, as_path=True)
|
||||
if not icon:
|
||||
return
|
||||
|
||||
@@ -262,6 +262,7 @@ class GameCard(QFrame):
|
||||
self.missing_executable_path = self._get_missing_executable_path()
|
||||
|
||||
self.steam_visible = (str(game_source).lower() == "steam" and not self.economy_mode)
|
||||
self.gog_visible = (str(game_source).lower() == "gog" and not self.economy_mode)
|
||||
self.portproton_visible = (str(game_source).lower() == "portproton" and not self.economy_mode)
|
||||
self.ppdb_visible = bool(self.ppdb_id) and not self.economy_mode
|
||||
|
||||
@@ -302,7 +303,10 @@ class GameCard(QFrame):
|
||||
self.coverLabel = QLabel()
|
||||
self.coverLabel.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||||
self.coverLabel.setStyleSheet(self.theme.COVER_LABEL_STYLE)
|
||||
if self.missing_executable_path:
|
||||
if self.missing_executable_path or (
|
||||
str(self.game_source).lower() == "gog"
|
||||
and self.exec_line.startswith("gog://install/")
|
||||
):
|
||||
self.coverOpacity = QGraphicsOpacityEffect(self.coverLabel)
|
||||
self.coverOpacity.setOpacity(self.theme.missing_exe_cover_opacity)
|
||||
self.coverLabel.setGraphicsEffect(self.coverOpacity)
|
||||
@@ -343,6 +347,16 @@ class GameCard(QFrame):
|
||||
if self.economy_mode:
|
||||
self.steamLabel.setVisible(False)
|
||||
|
||||
gog_icon = self.theme_manager.get_icon("badge_gog", as_path=True)
|
||||
self.gogLabel = SourceCorner(
|
||||
icon=gog_icon,
|
||||
config=corner_config,
|
||||
parent=self.coverWidget,
|
||||
)
|
||||
self.gogLabel.setVisible(self.gog_visible)
|
||||
if self.economy_mode:
|
||||
self.gogLabel.setVisible(False)
|
||||
|
||||
portproton_icon = self.theme_manager.get_icon("badge_portproton", as_path=True)
|
||||
self.portprotonLabel = SourceCorner(
|
||||
icon=portproton_icon,
|
||||
@@ -436,6 +450,7 @@ class GameCard(QFrame):
|
||||
width,
|
||||
height,
|
||||
self.on_cover_loaded,
|
||||
app_name=str(self.appid or ""),
|
||||
fallback_exe=fallback_exe,
|
||||
fallback_icon_path=fallback_icon_path,
|
||||
)
|
||||
@@ -568,6 +583,7 @@ class GameCard(QFrame):
|
||||
ribbon_size = max(ribbon_size, int(ribbon_cfg.get("min_size", 54) * self._scale))
|
||||
source_ribbons = [
|
||||
(self.steam_visible, self.steamLabel),
|
||||
(self.gog_visible, self.gogLabel),
|
||||
(self.portproton_visible, self.portprotonLabel),
|
||||
]
|
||||
|
||||
@@ -589,6 +605,7 @@ class GameCard(QFrame):
|
||||
self.ppdbLabel.raise_()
|
||||
self.protondbLabel.raise_()
|
||||
self.portprotonLabel.raise_()
|
||||
self.gogLabel.raise_()
|
||||
self.steamLabel.raise_()
|
||||
except RuntimeError:
|
||||
pass
|
||||
@@ -715,6 +732,7 @@ class GameCard(QFrame):
|
||||
self.display_filter = display_filter
|
||||
self.economy_mode = ui_config.get_economy_mode()
|
||||
self.steam_visible = (str(self.game_source).lower() == "steam" and not self.economy_mode)
|
||||
self.gog_visible = (str(self.game_source).lower() == "gog" and not self.economy_mode)
|
||||
self.portproton_visible = (str(self.game_source).lower() == "portproton" and not self.economy_mode)
|
||||
self.ppdb_visible = bool(self.ppdb_id) and not self.economy_mode
|
||||
protondb_visible = is_valid_protondb_tier(self.protondb_tier) and not self.economy_mode
|
||||
@@ -724,6 +742,7 @@ class GameCard(QFrame):
|
||||
|
||||
try:
|
||||
self.steamLabel.setVisible(self.steam_visible and not hidden_badges)
|
||||
self.gogLabel.setVisible(self.gog_visible and not hidden_badges)
|
||||
self.portprotonLabel.setVisible(self.portproton_visible and not hidden_badges)
|
||||
self.ppdbLabel.setVisible(self.ppdb_visible and not hidden_badges)
|
||||
self.protondbLabel.setVisible(protondb_visible and not hidden_badges)
|
||||
|
||||
@@ -0,0 +1,471 @@
|
||||
"""GOG authentication, library and gogdl process helpers."""
|
||||
|
||||
import hashlib
|
||||
import os
|
||||
import platform
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
from collections.abc import Callable
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
from pathlib import Path
|
||||
from urllib.parse import parse_qs, urlparse
|
||||
|
||||
import orjson
|
||||
import requests
|
||||
|
||||
from portprotonqt.localization import get_metadata_language
|
||||
from portprotonqt.logger import get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
GOG_LOGIN_URL = (
|
||||
"https://auth.gog.com/auth?client_id=46899977096215655&"
|
||||
"redirect_uri=https%3A%2F%2Fembed.gog.com%2Fon_login_success%3Forigin%3Dclient&"
|
||||
"response_type=code&layout=galaxy"
|
||||
)
|
||||
GOGDL_RELEASE_URL = "https://api.github.com/repos/Heroic-Games-Launcher/heroic-gogdl/releases/latest"
|
||||
GOGDL_AUTH_TIMEOUT = 60
|
||||
GOG_METADATA_WORKERS = 4
|
||||
GOG_PRODUCT_LOCALES = {
|
||||
"bg": "bg-BG", "cs": "cs-CZ", "da": "da-DK", "de": "de-DE",
|
||||
"el": "el-GR", "en": "en-US", "es": "es-ES", "fi": "fi-FI",
|
||||
"fr": "fr-FR", "hu": "hu-HU", "it": "it-IT", "ja": "ja-JP",
|
||||
"ko": "ko-KR", "nl": "nl-NL", "no": "no-NO", "pl": "pl-PL",
|
||||
"pt": "pt-BR", "ro": "ro-RO", "ru": "ru-RU", "sv": "sv-SE",
|
||||
"th": "th-TH", "tr": "tr-TR", "uk": "uk-UA", "zh": "zh-Hans",
|
||||
"zh_hant": "zh-Hant",
|
||||
}
|
||||
|
||||
|
||||
class GOGAPI:
|
||||
"""Provide the local GOG backend used by the Qt UI."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
data_home = Path(os.getenv("XDG_DATA_HOME", Path.home() / ".local/share"))
|
||||
self.data_dir = data_home / "PortProtonQt" / "gog"
|
||||
self.auth_path = self.data_dir / "auth.json"
|
||||
self.config_dir = self.data_dir / "gogdl"
|
||||
self.library_path = self.data_dir / "library.json"
|
||||
self.installed_path = self.data_dir / "installed.json"
|
||||
self.sizes_path = self.data_dir / "sizes.json"
|
||||
self.games_dir = Path.home() / "Games"
|
||||
|
||||
def get_gogdl_path(self) -> str | None:
|
||||
"""Return an available gogdl executable."""
|
||||
bundled = self.data_dir / "bin/gogdl"
|
||||
if bundled.is_file() and os.access(bundled, os.X_OK):
|
||||
return str(bundled)
|
||||
return shutil.which("gogdl")
|
||||
|
||||
def build_command(self, arguments: list[str]) -> list[str]:
|
||||
"""Build a gogdl command with private configuration paths."""
|
||||
gogdl = self.get_gogdl_path()
|
||||
if not gogdl:
|
||||
raise FileNotFoundError("gogdl executable not found")
|
||||
self.data_dir.mkdir(parents=True, exist_ok=True)
|
||||
self.data_dir.chmod(0o700)
|
||||
self.config_dir.mkdir(parents=True, exist_ok=True)
|
||||
return [gogdl, "--auth-config-path", str(self.auth_path), *arguments]
|
||||
|
||||
def ensure_gogdl(self) -> str:
|
||||
"""Download and verify the latest official Heroic gogdl binary."""
|
||||
existing = self.get_gogdl_path()
|
||||
if existing:
|
||||
return existing
|
||||
architecture = {"x86_64": "x86_64", "aarch64": "arm64"}.get(platform.machine())
|
||||
if not architecture:
|
||||
raise OSError(f"Unsupported gogdl architecture: {platform.machine()}")
|
||||
response = requests.get(GOGDL_RELEASE_URL, timeout=15)
|
||||
response.raise_for_status()
|
||||
asset_name = f"gogdl_linux_{architecture}"
|
||||
asset = next(
|
||||
(item for item in response.json().get("assets", []) if item.get("name") == asset_name),
|
||||
None,
|
||||
)
|
||||
if not asset:
|
||||
raise OSError(f"gogdl release asset not found: {asset_name}")
|
||||
return self._download_gogdl_asset(asset)
|
||||
|
||||
def _download_gogdl_asset(self, asset: dict) -> str:
|
||||
target = self.data_dir / "bin/gogdl"
|
||||
temporary = target.with_suffix(".part")
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
digest = hashlib.sha256()
|
||||
try:
|
||||
with requests.get(asset["browser_download_url"], stream=True, timeout=30) as response:
|
||||
response.raise_for_status()
|
||||
with open(temporary, "wb") as output:
|
||||
for chunk in response.iter_content(chunk_size=1024 * 1024):
|
||||
if chunk:
|
||||
output.write(chunk)
|
||||
digest.update(chunk)
|
||||
expected = str(asset.get("digest", "")).removeprefix("sha256:")
|
||||
if not expected or digest.hexdigest() != expected:
|
||||
raise OSError("gogdl checksum verification failed")
|
||||
temporary.chmod(0o755)
|
||||
temporary.replace(target)
|
||||
return str(target)
|
||||
except Exception:
|
||||
temporary.unlink(missing_ok=True)
|
||||
raise
|
||||
|
||||
def get_environment(self) -> dict[str, str]:
|
||||
"""Return the environment required by gogdl."""
|
||||
environment = os.environ.copy()
|
||||
environment["GOGDL_CONFIG_PATH"] = str(self.config_dir)
|
||||
return environment
|
||||
|
||||
def authenticate(self, code: str) -> tuple[bool, str]:
|
||||
"""Exchange an OAuth code through gogdl."""
|
||||
self.ensure_gogdl()
|
||||
result = subprocess.run(
|
||||
self.build_command(["auth", "--code", code]),
|
||||
env=self.get_environment(), capture_output=True, check=False,
|
||||
timeout=GOGDL_AUTH_TIMEOUT,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
error = result.stderr.decode(errors="replace").strip()
|
||||
logger.error("gogdl authentication failed: %s", error)
|
||||
return False, error or f"gogdl exited with code {result.returncode}"
|
||||
try:
|
||||
data = orjson.loads(result.stdout)
|
||||
except orjson.JSONDecodeError:
|
||||
return False, "gogdl returned an invalid authentication response"
|
||||
if not isinstance(data, dict) or data.get("error"):
|
||||
return False, "GOG rejected the authorization code"
|
||||
if self.auth_path.is_file():
|
||||
self.auth_path.chmod(0o600)
|
||||
return True, ""
|
||||
|
||||
def get_credentials(self) -> dict:
|
||||
"""Return refreshed GOG credentials through gogdl."""
|
||||
try:
|
||||
self.ensure_gogdl()
|
||||
result = subprocess.run(
|
||||
self.build_command(["auth"]), env=self.get_environment(),
|
||||
capture_output=True, check=False,
|
||||
)
|
||||
data = orjson.loads(result.stdout) if result.returncode == 0 else {}
|
||||
return data if isinstance(data, dict) else {}
|
||||
except (FileNotFoundError, orjson.JSONDecodeError):
|
||||
return {}
|
||||
|
||||
def is_authenticated(self) -> bool:
|
||||
"""Return whether gogdl provides usable account credentials."""
|
||||
credentials = self.get_credentials()
|
||||
return bool(credentials.get("access_token") and credentials.get("user_id"))
|
||||
|
||||
def refresh_library(
|
||||
self, progress_callback: Callable[[int, int], None] | None = None
|
||||
) -> list[dict]:
|
||||
"""Fetch and cache the user's GOG library."""
|
||||
credentials = self.get_credentials()
|
||||
token = credentials.get("access_token")
|
||||
user_id = credentials.get("user_id")
|
||||
if not token or not user_id:
|
||||
return []
|
||||
entries = self._get_library_entries(str(user_id), str(token))
|
||||
gog_entries = [entry for entry in entries if entry.get("platform_id") == "gog"]
|
||||
games = self._get_games_parallel(gog_entries, str(token), progress_callback)
|
||||
self._save_json(self.library_path, games)
|
||||
return games
|
||||
|
||||
def _get_games_parallel(
|
||||
self, entries: list[dict], token: str,
|
||||
progress_callback: Callable[[int, int], None] | None,
|
||||
) -> list[dict]:
|
||||
cached = {str(game.get("app_id")): game for game in self.load_library()}
|
||||
games = []
|
||||
worker_count = min(GOG_METADATA_WORKERS, os.cpu_count() or 1)
|
||||
with ThreadPoolExecutor(max_workers=worker_count) as executor:
|
||||
futures = {executor.submit(self._get_game, entry, token): entry for entry in entries}
|
||||
for completed, future in enumerate(as_completed(futures), 1):
|
||||
entry = futures[future]
|
||||
app_id = str(entry.get("external_id", ""))
|
||||
try:
|
||||
game = future.result()
|
||||
except requests.RequestException as error:
|
||||
logger.warning("Failed to load GOG metadata for %s: %s", app_id, error)
|
||||
game = cached.get(app_id, {})
|
||||
if game:
|
||||
games.append(game)
|
||||
if progress_callback:
|
||||
progress_callback(completed, len(entries))
|
||||
return games
|
||||
|
||||
def load_library(self) -> list[dict]:
|
||||
"""Load the cached GOG library."""
|
||||
data = self._load_json(self.library_path, [])
|
||||
return data if isinstance(data, list) else []
|
||||
|
||||
def load_installed(self) -> dict[str, dict]:
|
||||
"""Load installed GOG game records."""
|
||||
data = self._load_json(self.installed_path, {})
|
||||
return data if isinstance(data, dict) else {}
|
||||
|
||||
def save_installed_game(self, app_id: str, game: dict) -> None:
|
||||
"""Persist one installed GOG game record."""
|
||||
installed = self.load_installed()
|
||||
installed[app_id] = game
|
||||
self._save_json(self.installed_path, installed)
|
||||
|
||||
def remove_installed_game(self, app_id: str) -> None:
|
||||
"""Remove one installed GOG game record."""
|
||||
installed = self.load_installed()
|
||||
installed.pop(app_id, None)
|
||||
self._save_json(self.installed_path, installed)
|
||||
|
||||
def get_cached_download_sizes(self, app_id: str) -> tuple[int, int] | None:
|
||||
"""Return cached download and disk sizes for a game."""
|
||||
cache = self._load_json(self.sizes_path, {})
|
||||
if not isinstance(cache, dict):
|
||||
return None
|
||||
cached = cache.get(app_id)
|
||||
if not isinstance(cached, list) or len(cached) != 2:
|
||||
return None
|
||||
return int(cached[0]), int(cached[1])
|
||||
|
||||
def save_download_sizes(self, app_id: str, sizes: tuple[int, int]) -> None:
|
||||
"""Cache download and disk sizes for a game."""
|
||||
cached = self._load_json(self.sizes_path, {})
|
||||
if not isinstance(cached, dict):
|
||||
cached = {}
|
||||
cached[app_id] = list(sizes)
|
||||
self._save_json(self.sizes_path, cached)
|
||||
|
||||
def is_game_installed(
|
||||
self, app_id: str, installed_games: dict[str, dict] | None = None
|
||||
) -> bool:
|
||||
"""Return whether a complete gogdl installation is available."""
|
||||
return self.get_installed_path(app_id, installed_games) is not None
|
||||
|
||||
def get_installed_path(
|
||||
self, app_id: str, installed_games: dict[str, dict] | None = None
|
||||
) -> Path | None:
|
||||
"""Return a recorded or discoverable gogdl installation path."""
|
||||
records = installed_games if installed_games is not None else self.load_installed()
|
||||
installed = records.get(app_id, {})
|
||||
install_path = Path(str(installed.get("install_path", "")))
|
||||
if (install_path / f"goggame-{app_id}.info").is_file():
|
||||
return install_path
|
||||
return self.find_install_path(app_id, self.games_dir)
|
||||
|
||||
def get_install_path(self, app_id: str, title: str) -> Path:
|
||||
"""Return the default parent directory for GOG installations."""
|
||||
return self.games_dir
|
||||
|
||||
def find_install_path(self, app_id: str, parent: Path) -> Path | None:
|
||||
"""Find the directory created by gogdl for an installed game."""
|
||||
metadata_name = f"goggame-{app_id}.info"
|
||||
if (parent / metadata_name).is_file():
|
||||
return parent
|
||||
try:
|
||||
return next(
|
||||
(path.parent for path in parent.glob(f"*/{metadata_name}")),
|
||||
None,
|
||||
)
|
||||
except OSError as error:
|
||||
logger.warning("Failed to inspect GOG installation path %s: %s", parent, error)
|
||||
return None
|
||||
|
||||
def get_launch_target(self, app_id: str) -> str | None:
|
||||
"""Return the primary executable from gogdl installation metadata."""
|
||||
primary, install_path = self._get_primary_task(app_id)
|
||||
if primary is None or install_path is None:
|
||||
return None
|
||||
target = install_path
|
||||
parts = str(primary.get("path", "")).replace("\\", "/").split("/")
|
||||
for part in parts:
|
||||
exact_path = target / part
|
||||
if exact_path.exists():
|
||||
target = exact_path
|
||||
continue
|
||||
try:
|
||||
target = next(
|
||||
child for child in target.iterdir()
|
||||
if child.name.casefold() == part.casefold()
|
||||
)
|
||||
except (OSError, StopIteration):
|
||||
return None
|
||||
return str(target) if target.is_file() else None
|
||||
|
||||
def ensure_launch_parameters(self, app_id: str) -> None:
|
||||
"""Add GOG task arguments to the executable PPDB when absent."""
|
||||
primary, install_path = self._get_primary_task(app_id)
|
||||
target = self.get_launch_target(app_id)
|
||||
if primary is None or install_path is None or not target:
|
||||
return
|
||||
arguments = str(primary.get("arguments", "")).replace("\r", " ").replace("\n", " ").strip()
|
||||
if not arguments:
|
||||
return
|
||||
ppdb_path = Path(f"{target}.ppdb")
|
||||
try:
|
||||
current = ppdb_path.read_text(encoding="utf-8") if ppdb_path.exists() else ""
|
||||
base_value = arguments.replace('"', "").replace("\\", "/")
|
||||
normalized = base_value
|
||||
support_dir = install_path / "gog-support" / app_id / "app"
|
||||
if support_dir.is_dir():
|
||||
for support_file in support_dir.rglob("*"):
|
||||
if not support_file.is_file():
|
||||
continue
|
||||
relative_path = os.path.relpath(support_file, Path(target).parent)
|
||||
normalized = normalized.replace(
|
||||
f"../{support_file.name}", relative_path.replace(os.sep, "/")
|
||||
)
|
||||
safe_value = normalized.replace("$", "\\$").replace("`", "\\`")
|
||||
launch_line = f'export LAUNCH_PARAMETERS="{safe_value}"'
|
||||
lines = current.splitlines()
|
||||
existing = next(
|
||||
(line for line in lines if line.startswith("export LAUNCH_PARAMETERS=")),
|
||||
"",
|
||||
)
|
||||
if existing:
|
||||
value = existing.removeprefix('export LAUNCH_PARAMETERS="').removesuffix('"')
|
||||
comparable = re.sub(r"\\+", "/", value.replace('\\"', "").replace('"', ""))
|
||||
if comparable not in (base_value, normalized):
|
||||
return
|
||||
lines[lines.index(existing)] = launch_line
|
||||
ppdb_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
||||
return
|
||||
separator = "" if not current or current.endswith("\n") else "\n"
|
||||
ppdb_path.write_text(
|
||||
f"{current}{separator}{launch_line}\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
except OSError as error:
|
||||
logger.warning("Failed to add GOG launch parameters to %s: %s", ppdb_path, error)
|
||||
|
||||
def _get_primary_task(self, app_id: str) -> tuple[dict | None, Path | None]:
|
||||
"""Return the primary GOG launch task and installation path."""
|
||||
install_path = self.get_installed_path(app_id)
|
||||
if install_path is None:
|
||||
return None, None
|
||||
info = self._load_json(install_path / f"goggame-{app_id}.info", {})
|
||||
tasks = info.get("playTasks", []) if isinstance(info, dict) else []
|
||||
primary = next((task for task in tasks if task.get("isPrimary")), None)
|
||||
if primary is None and tasks:
|
||||
primary = tasks[0]
|
||||
if not isinstance(primary, dict) or primary.get("type") != "FileTask":
|
||||
return None, install_path
|
||||
return primary, install_path
|
||||
|
||||
def _get_library_entries(self, user_id: str, token: str) -> list[dict]:
|
||||
entries = []
|
||||
page_token = ""
|
||||
headers = {"Authorization": f"Bearer {token}"}
|
||||
while True:
|
||||
url = f"https://galaxy-library.gog.com/users/{user_id}/releases"
|
||||
response = requests.get(url, headers=headers, params={"page_token": page_token} if page_token else {}, timeout=15)
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
entries.extend(data.get("items", []))
|
||||
page_token = data.get("next_page_token", "")
|
||||
if not page_token:
|
||||
return entries
|
||||
|
||||
def _get_game(self, entry: dict, token: str) -> dict:
|
||||
if entry.get("platform_id") != "gog" or not entry.get("external_id"):
|
||||
return {}
|
||||
app_id = str(entry["external_id"])
|
||||
headers = {"Authorization": f"Bearer {token}"}
|
||||
if entry.get("certificate"):
|
||||
headers["X-GOG-Library-Cert"] = str(entry["certificate"])
|
||||
response = requests.get(
|
||||
f"https://gamesdb.gog.com/platforms/gog/external_releases/{app_id}",
|
||||
headers=headers, timeout=15,
|
||||
)
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
game = data.get("game", {})
|
||||
if not game.get("visible_in_library", True):
|
||||
return {}
|
||||
description = self._localized_value(data.get("summary"))
|
||||
language = get_metadata_language().lower()
|
||||
try:
|
||||
product_response = requests.get(
|
||||
f"https://api.gog.com/products/{app_id}",
|
||||
params={
|
||||
"expand": "description",
|
||||
"locale": GOG_PRODUCT_LOCALES.get(language, language),
|
||||
},
|
||||
timeout=15,
|
||||
)
|
||||
product_response.raise_for_status()
|
||||
product_description = product_response.json().get("description", {})
|
||||
localized_lead = str(product_description.get("lead") or "")
|
||||
if localized_lead:
|
||||
paragraphs = re.split(
|
||||
r"(?:<br\s*/?>\s*){2,}", localized_lead, flags=re.IGNORECASE
|
||||
)
|
||||
paragraph_count = (
|
||||
2 if re.match(r"\s*<(?:b|strong)>", paragraphs[0], re.IGNORECASE)
|
||||
else 1
|
||||
)
|
||||
description = "<br><br>".join(
|
||||
paragraph.strip() for paragraph in paragraphs[:paragraph_count]
|
||||
)
|
||||
except requests.RequestException as error:
|
||||
logger.warning(
|
||||
"Failed to load localized GOG description for %s: %s",
|
||||
app_id, error,
|
||||
)
|
||||
return {
|
||||
"app_id": app_id,
|
||||
"title": (
|
||||
self._localized_value(data.get("title"))
|
||||
or self._localized_value(game.get("title"))
|
||||
or app_id
|
||||
).strip(),
|
||||
"description": description,
|
||||
"cover": self._format_image(game.get("vertical_cover") or game.get("logo") or game.get("background")),
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def _localized_value(values: dict | None) -> str:
|
||||
if not isinstance(values, dict):
|
||||
return ""
|
||||
language = get_metadata_language().lower()
|
||||
for key in (language, language.replace("_", "-")):
|
||||
if values.get(key):
|
||||
return str(values[key])
|
||||
localized = next(
|
||||
(value for key, value in values.items() if key.lower().startswith(language)),
|
||||
None,
|
||||
)
|
||||
return str(localized or values.get("*", ""))
|
||||
|
||||
@staticmethod
|
||||
def _format_image(image: dict | None) -> str:
|
||||
template = image.get("url_format", "") if isinstance(image, dict) else ""
|
||||
return template.replace("{formatter}", "").replace("{ext}", "jpg")
|
||||
|
||||
@staticmethod
|
||||
def parse_download_sizes(output: bytes) -> tuple[int, int]:
|
||||
"""Return download and disk sizes from gogdl info output."""
|
||||
data = orjson.loads(output)
|
||||
sizes = data.get("size", {}) if isinstance(data, dict) else {}
|
||||
languages = data.get("languages", []) if isinstance(data, dict) else []
|
||||
selected = [sizes.get("*", {})]
|
||||
if languages:
|
||||
selected.append(sizes.get(str(languages[0]), {}))
|
||||
download_size = sum(int(item.get("download_size", 0)) for item in selected)
|
||||
disk_size = sum(int(item.get("disk_size", 0)) for item in selected)
|
||||
return download_size, disk_size
|
||||
|
||||
@staticmethod
|
||||
def extract_auth_code(url: str) -> str:
|
||||
"""Extract the OAuth code from GOG's redirect URL."""
|
||||
values = parse_qs(urlparse(url).query)
|
||||
return values.get("code", [""])[0]
|
||||
|
||||
@staticmethod
|
||||
def _load_json(path: Path, fallback):
|
||||
try:
|
||||
return orjson.loads(path.read_bytes())
|
||||
except (OSError, orjson.JSONDecodeError):
|
||||
return fallback
|
||||
|
||||
@staticmethod
|
||||
def _save_json(path: Path, data) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_bytes(orjson.dumps(data, option=orjson.OPT_INDENT_2))
|
||||
+127
-9
@@ -19,6 +19,7 @@ from portprotonqt.animations import DetailPageAnimations
|
||||
from portprotonqt.custom_widgets import ClickableLabel, AutoSizeButton, NavLabel
|
||||
from portprotonqt.detail_pages import DetailPageManager
|
||||
from portprotonqt.portproton_api import PortProtonAPI, remove_empty_custom_data_dirs
|
||||
from portprotonqt.gog_api import GOGAPI
|
||||
from portprotonqt.debug_utils import get_prefix_name
|
||||
from portprotonqt.input_manager import InputManager, MainWindowProtocol
|
||||
from portprotonqt.context_menu_manager import ContextMenuManager
|
||||
@@ -59,6 +60,7 @@ from portprotonqt.sound_manager import SoundManager
|
||||
from portprotonqt.tabs.control_hints import MainWindowControlHintsMixin
|
||||
from portprotonqt.tabs.autoinstall_tab import MainWindowAutoInstallTabMixin
|
||||
from portprotonqt.tabs.library_tab import MainWindowLibraryTabMixin
|
||||
from portprotonqt.tabs.gog_tab import MainWindowGOGTabMixin
|
||||
from portprotonqt.tabs.settings_tab import MainWindowSettingsTabMixin
|
||||
from portprotonqt.tabs.system_tab import MainWindowSystemTabMixin
|
||||
from portprotonqt.tabs.theme_tab import MainWindowThemeTabMixin
|
||||
@@ -134,6 +136,7 @@ class MainWindow(
|
||||
MainWindowControlHintsMixin,
|
||||
MainWindowAutoInstallTabMixin,
|
||||
MainWindowLibraryTabMixin,
|
||||
MainWindowGOGTabMixin,
|
||||
MainWindowSettingsTabMixin,
|
||||
MainWindowSystemTabMixin,
|
||||
MainWindowThemeTabMixin,
|
||||
@@ -358,6 +361,7 @@ class MainWindow(
|
||||
# Start watching dist and prefixes directories if they exist
|
||||
QTimer.singleShot(0, self.start_watching_directories) # Delay to ensure portproton_location is set
|
||||
self.downloader = Downloader(max_workers=4)
|
||||
self.gog_api = GOGAPI()
|
||||
self.portproton_api = PortProtonAPI(self.downloader)
|
||||
self.autoInstallCustomDataThread = None
|
||||
|
||||
@@ -423,7 +427,8 @@ class MainWindow(
|
||||
_("Wine Settings"),
|
||||
_("PPQT Settings"),
|
||||
_("System"),
|
||||
_("Themes")
|
||||
_("Themes"),
|
||||
_("Downloads")
|
||||
]
|
||||
for i, tabName in enumerate(tabs):
|
||||
btn = NavLabel(tabName)
|
||||
@@ -489,6 +494,7 @@ class MainWindow(
|
||||
if hidden_system_page:
|
||||
hidden_system_page.setVisible(False)
|
||||
self.createThemeTab()
|
||||
self.createGOGDownloadsTab()
|
||||
|
||||
self.controlHintsWidget = self.createControlHintsWidget()
|
||||
self.controlHintsWidget.setStyleSheet(self.theme.HINT_BAR_STYLE)
|
||||
@@ -931,21 +937,23 @@ class MainWindow(
|
||||
def start_loading():
|
||||
if display_filter == "steam":
|
||||
self._load_steam_games_async(lambda games: self.games_loaded.emit(games))
|
||||
elif display_filter == "gog":
|
||||
self._load_gog_games_async(lambda games: self.games_loaded.emit(games))
|
||||
elif display_filter == "portproton":
|
||||
self._load_portproton_games_async(lambda games: self.games_loaded.emit(games))
|
||||
elif display_filter == "favorites":
|
||||
def on_all_games_favorites(portproton_games, steam_games):
|
||||
games = [game for game in portproton_games + steam_games if game[0] in favorites]
|
||||
def on_all_games_favorites(portproton_games, steam_games, gog_games):
|
||||
games = [game for game in portproton_games + steam_games + gog_games if game[0] in favorites]
|
||||
self.games_loaded.emit(games)
|
||||
|
||||
# Load games from different sources in parallel to prevent blocking
|
||||
results = {'portproton': [], 'steam': []}
|
||||
completed = {'portproton': False, 'steam': False}
|
||||
results = {'portproton': [], 'steam': [], 'gog': []}
|
||||
completed = {'portproton': False, 'steam': False, 'gog': False}
|
||||
|
||||
def check_completion():
|
||||
if all(completed.values()):
|
||||
QApplication.processEvents() # Keep UI responsive
|
||||
on_all_games_favorites(results['portproton'], results['steam'])
|
||||
on_all_games_favorites(results['portproton'], results['steam'], results['gog'])
|
||||
|
||||
def portproton_callback(games):
|
||||
results['portproton'] = games
|
||||
@@ -959,17 +967,23 @@ class MainWindow(
|
||||
QApplication.processEvents() # Keep UI responsive
|
||||
check_completion()
|
||||
|
||||
def gog_callback(games):
|
||||
results['gog'] = games
|
||||
completed['gog'] = True
|
||||
check_completion()
|
||||
|
||||
self._load_portproton_games_async(portproton_callback)
|
||||
self._load_steam_games_async(steam_callback)
|
||||
self._load_gog_games_async(gog_callback)
|
||||
else:
|
||||
# For 'all' filter - load games from different sources in parallel to prevent blocking
|
||||
results = {'portproton': [], 'steam': []}
|
||||
completed = {'portproton': False, 'steam': False}
|
||||
results = {'portproton': [], 'steam': [], 'gog': []}
|
||||
completed = {'portproton': False, 'steam': False, 'gog': False}
|
||||
|
||||
def on_all_games():
|
||||
seen = set()
|
||||
games = []
|
||||
for game in results['portproton'] + results['steam']:
|
||||
for game in results['portproton'] + results['steam'] + results['gog']:
|
||||
# Unique key: name + exec_line
|
||||
key = (game[0], game[5])
|
||||
if key not in seen:
|
||||
@@ -995,13 +1009,48 @@ class MainWindow(
|
||||
QApplication.processEvents() # Keep UI responsive
|
||||
check_completion()
|
||||
|
||||
def gog_callback(games):
|
||||
results['gog'] = games
|
||||
completed['gog'] = True
|
||||
check_completion()
|
||||
|
||||
# Load all sources in parallel
|
||||
self._load_portproton_games_async(portproton_callback)
|
||||
self._load_steam_games_async(steam_callback)
|
||||
self._load_gog_games_async(gog_callback)
|
||||
|
||||
# Run loading immediately to show status without delay
|
||||
start_loading()
|
||||
|
||||
def _load_gog_games_async(self, callback: Callable[[list[tuple]], None]) -> None:
|
||||
api = getattr(self, "gog_api", None) or GOGAPI()
|
||||
installed = api.load_installed()
|
||||
games = []
|
||||
for game in api.load_library():
|
||||
app_id = str(game.get("app_id", ""))
|
||||
if not app_id:
|
||||
continue
|
||||
action = "launch" if api.is_game_installed(app_id, installed) else "install"
|
||||
games.append((
|
||||
str(game.get("title", app_id)),
|
||||
str(game.get("description", "")),
|
||||
str(game.get("cover", "")),
|
||||
app_id,
|
||||
"",
|
||||
f"gog://{action}/{app_id}",
|
||||
_("Never"),
|
||||
format_playtime(0),
|
||||
"",
|
||||
"",
|
||||
0,
|
||||
0,
|
||||
"gog",
|
||||
"",
|
||||
"",
|
||||
"",
|
||||
))
|
||||
callback(games)
|
||||
|
||||
def _load_steam_games_async(self, callback: Callable[[list[tuple]], None]):
|
||||
steam_games = []
|
||||
installed_games = get_steam_installed_games()
|
||||
@@ -1461,6 +1510,15 @@ class MainWindow(
|
||||
def openGameDetailPage(self, game_data: dict) -> None:
|
||||
"""Open game detail page."""
|
||||
SoundManager().play("open")
|
||||
exec_line = str(game_data.get("exec_line", ""))
|
||||
if str(game_data.get("game_source", "")).lower() == "gog":
|
||||
app_id = str(game_data.get("appid", ""))
|
||||
if (
|
||||
exec_line.startswith("gog://install/")
|
||||
and self.gog_api.is_game_installed(app_id)
|
||||
):
|
||||
game_data = dict(game_data)
|
||||
game_data["exec_line"] = f"gog://launch/{app_id}"
|
||||
logger.debug(
|
||||
"Opening detail page for %s from stacked index %d",
|
||||
game_data.get("name", ""),
|
||||
@@ -2259,12 +2317,72 @@ class MainWindow(
|
||||
if not QDesktopServices.openUrl(QUrl(exec_line)):
|
||||
logger.warning("Failed to open Steam URI: %s", exec_line)
|
||||
|
||||
def _handle_gog_game(self, exec_line: str, button=None) -> None:
|
||||
action, app_id = exec_line.removeprefix("gog://").split("/", 1)
|
||||
if action == "install" and self.gog_api.is_game_installed(app_id):
|
||||
action = "launch"
|
||||
if action == "launch":
|
||||
self._launch_gog_game(app_id, button)
|
||||
return
|
||||
game = next(
|
||||
(item for item in self.gog_api.load_library()
|
||||
if str(item.get("app_id", "")) == app_id),
|
||||
None,
|
||||
)
|
||||
if game:
|
||||
self._install_gog_game(game)
|
||||
|
||||
def _launch_gog_game(self, app_id: str, button=None) -> None:
|
||||
self.gog_api.ensure_launch_parameters(app_id)
|
||||
discovered_path = self.gog_api.get_installed_path(app_id)
|
||||
install_path = str(discovered_path) if discovered_path else ""
|
||||
if not install_path or not self.start_sh:
|
||||
QMessageBox.warning(self, _("Error"), _("GOG game or PortProton is unavailable"))
|
||||
return
|
||||
target = os.path.basename(self.gog_api.get_launch_target(app_id) or app_id)
|
||||
if self.game_processes and self.target_exe == target:
|
||||
self.stop_running_game(button)
|
||||
return
|
||||
if self.game_processes:
|
||||
QMessageBox.warning(self, _("Error"), _("Cannot launch game while another game is running"))
|
||||
return
|
||||
try:
|
||||
command = self.gog_api.build_command([
|
||||
"launch", install_path, app_id, "--platform", "windows",
|
||||
"--wine", self.start_sh[0],
|
||||
])
|
||||
process = subprocess.Popen(
|
||||
command, env=self.gog_api.get_environment(), shell=False,
|
||||
preexec_fn=os.setsid, stdout=subprocess.PIPE,
|
||||
stderr=subprocess.STDOUT, text=True, errors="replace",
|
||||
)
|
||||
except (OSError, ValueError) as error:
|
||||
logger.error("Failed to launch GOG game %s: %s", app_id, error)
|
||||
QMessageBox.warning(self, _("Error"), str(error))
|
||||
return
|
||||
SoundManager().play("game_launch")
|
||||
self.game_processes.append(process)
|
||||
self.target_exe = target
|
||||
self.current_running_button = button
|
||||
self._start_launch_output_reader(process)
|
||||
self.input_manager.suspend_gamepad_polling()
|
||||
self.checkProcessTimer = QTimer(self)
|
||||
self.checkProcessTimer.timeout.connect(self.checkTargetExe)
|
||||
self.checkProcessTimer.start(500)
|
||||
if button is not None:
|
||||
button.setText(_("Stop"))
|
||||
button.setIcon(self.theme_manager.get_icon("stop", as_path=True))
|
||||
|
||||
def toggleGame(self, exec_line, button=None, game_name=None):
|
||||
# Handle Steam games
|
||||
if exec_line.startswith("steam://"):
|
||||
self._launch_steam_game(exec_line)
|
||||
return
|
||||
|
||||
if exec_line.startswith("gog://"):
|
||||
self._handle_gog_game(exec_line, button)
|
||||
return
|
||||
|
||||
# Handle PortProton games
|
||||
entry_exec_split = shlex.split(exec_line)
|
||||
if not entry_exec_split:
|
||||
|
||||
@@ -12,6 +12,7 @@ install_data(
|
||||
'game_card.py',
|
||||
'gamepad_backend.py',
|
||||
'game_library_manager.py',
|
||||
'gog_api.py',
|
||||
'howlongtobeat_api.py',
|
||||
'image_utils.py',
|
||||
'icon_extractor.py',
|
||||
@@ -110,6 +111,7 @@ install_data(
|
||||
['tabs/__init__.py',
|
||||
'tabs/autoinstall_tab.py',
|
||||
'tabs/control_hints.py',
|
||||
'tabs/gog_tab.py',
|
||||
'tabs/library_tab.py',
|
||||
'tabs/settings_tab.py',
|
||||
'tabs/system_tab.py',
|
||||
|
||||
@@ -0,0 +1,573 @@
|
||||
"""Shared download manager with GOG download support."""
|
||||
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
from PySide6.QtCore import QProcess, QProcessEnvironment, QThread, QTimer, QUrl, Signal
|
||||
from PySide6.QtGui import QDesktopServices
|
||||
from PySide6.QtWidgets import (
|
||||
QAbstractItemView,
|
||||
QFrame,
|
||||
QHeaderView,
|
||||
QHBoxLayout,
|
||||
QLabel,
|
||||
QMessageBox,
|
||||
QProgressBar,
|
||||
QPushButton,
|
||||
QTableWidget,
|
||||
QTableWidgetItem,
|
||||
QVBoxLayout,
|
||||
QWidget,
|
||||
)
|
||||
|
||||
from portprotonqt.dialogs.file_explorer import FileExplorer
|
||||
from portprotonqt.gog_api import GOGAPI, GOG_LOGIN_URL
|
||||
from portprotonqt.image_utils import load_pixmap_async
|
||||
from portprotonqt.localization import _
|
||||
from portprotonqt.logger import get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
GOG_CANCEL_KILL_TIMEOUT_MS = 3000
|
||||
|
||||
|
||||
class GOGLibraryWorker(QThread):
|
||||
"""Refresh the GOG library outside the UI thread."""
|
||||
|
||||
loaded = Signal(list)
|
||||
failed = Signal(str)
|
||||
progress = Signal(int, int)
|
||||
|
||||
def __init__(self, api: GOGAPI) -> None:
|
||||
super().__init__()
|
||||
self.api = api
|
||||
|
||||
def run(self) -> None:
|
||||
try:
|
||||
self.loaded.emit(self.api.refresh_library(self.progress.emit))
|
||||
except Exception as error:
|
||||
logger.exception("Failed to refresh GOG library")
|
||||
self.failed.emit(str(error))
|
||||
|
||||
|
||||
class GOGAuthWorker(QThread):
|
||||
"""Exchange the GOG OAuth code outside the UI thread."""
|
||||
|
||||
authenticated = Signal(bool, str)
|
||||
|
||||
def __init__(self, api: GOGAPI, code: str) -> None:
|
||||
super().__init__()
|
||||
self.api = api
|
||||
self.code = code
|
||||
|
||||
def run(self) -> None:
|
||||
try:
|
||||
authenticated, error = self.api.authenticate(self.code)
|
||||
self.authenticated.emit(authenticated, error)
|
||||
except Exception as error:
|
||||
logger.exception("Failed to authenticate with GOG: %s", error)
|
||||
self.authenticated.emit(False, str(error))
|
||||
|
||||
|
||||
class MainWindowGOGTabMixin:
|
||||
"""Add account actions and the shared downloads page."""
|
||||
|
||||
def createGOGDownloadsTab(self) -> None:
|
||||
self.gog_process = None
|
||||
self.gog_download_queue = []
|
||||
self.gog_download_output = ""
|
||||
self.downloadTableHeadings = {}
|
||||
page = QWidget()
|
||||
page.setStyleSheet(self.theme.OTHER_PAGES_WIDGET_STYLE)
|
||||
layout = QVBoxLayout(page)
|
||||
layout.setSpacing(self.theme.downloadsSectionSpacing)
|
||||
title = QLabel(_("Downloads"))
|
||||
title.setStyleSheet(self.theme.TAB_TITLE_STYLE)
|
||||
layout.addWidget(title)
|
||||
self._create_active_download_card(layout)
|
||||
self.downloadQueuedTable = self._create_download_table(_("Queued"), layout)
|
||||
completed_header = QHBoxLayout()
|
||||
completed_header.addWidget(QLabel(_("Completed")))
|
||||
clear_button = QPushButton(_("Clear List"))
|
||||
clear_button.setStyleSheet(self.theme.ACTION_BUTTON_STYLE)
|
||||
clear_button.clicked.connect(self._clear_completed_downloads)
|
||||
completed_header.addWidget(clear_button)
|
||||
completed_header.addStretch()
|
||||
layout.addLayout(completed_header)
|
||||
self.downloadCompletedTable = self._create_download_table("", layout)
|
||||
layout.addStretch()
|
||||
self.stackedWidget.addWidget(page)
|
||||
|
||||
def _create_active_download_card(self, layout: QVBoxLayout) -> None:
|
||||
self.downloadActiveHeading = QLabel(_("Now Downloading"))
|
||||
self.downloadActiveHeading.setStyleSheet(self.theme.SETTINGS_TITLE_STYLE)
|
||||
layout.addWidget(self.downloadActiveHeading)
|
||||
self.downloadActiveCard = QFrame()
|
||||
self.downloadActiveCard.setObjectName("downloadsActiveCard")
|
||||
self.downloadActiveCard.setStyleSheet(self.theme.DOWNLOADS_ACTIVE_STYLE)
|
||||
self.downloadActiveCard.setFixedHeight(self.theme.downloadsActiveCardHeight)
|
||||
card_layout = QVBoxLayout(self.downloadActiveCard)
|
||||
card_layout.setContentsMargins(*self.theme.downloadsActiveCardMargins)
|
||||
card_layout.setSpacing(self.theme.downloadsActiveCardSpacing)
|
||||
content = QHBoxLayout()
|
||||
content.setSpacing(self.theme.downloadsActiveCardSpacing)
|
||||
self.downloadActiveCover = QLabel()
|
||||
self.downloadActiveCover.setFixedSize(*self.theme.downloadsActiveCoverSize)
|
||||
content.addWidget(self.downloadActiveCover)
|
||||
info = QVBoxLayout()
|
||||
header = QHBoxLayout()
|
||||
self.downloadActiveTitle = QLabel()
|
||||
self.downloadActiveTitle.setStyleSheet(self.theme.SETTINGS_TITLE_STYLE)
|
||||
header.addWidget(self.downloadActiveTitle)
|
||||
header.addStretch()
|
||||
self.downloadCancelButton = QPushButton(_("Cancel"))
|
||||
self.downloadCancelButton.setStyleSheet(self.theme.ACTION_BUTTON_STYLE)
|
||||
self.downloadCancelButton.clicked.connect(self._cancel_gog_download)
|
||||
header.addWidget(self.downloadCancelButton)
|
||||
info.addLayout(header)
|
||||
self.downloadActiveDetails = QLabel(_("Waiting…"))
|
||||
self.downloadActiveDetails.setStyleSheet(self.theme.CONTENT_STYLE)
|
||||
info.addWidget(self.downloadActiveDetails)
|
||||
metrics = QHBoxLayout()
|
||||
self.downloadSpeedLabel = QLabel(_("Download: —"))
|
||||
self.diskSpeedLabel = QLabel(_("Disk: —"))
|
||||
metrics.addWidget(self.downloadSpeedLabel)
|
||||
metrics.addWidget(self.diskSpeedLabel)
|
||||
metrics.addStretch()
|
||||
info.addLayout(metrics)
|
||||
content.addLayout(info, stretch=1)
|
||||
card_layout.addLayout(content)
|
||||
self.downloadOverallProgress = QProgressBar()
|
||||
self.downloadOverallProgress.setStyleSheet(self.theme.PROGRESS_BAR_STYLE)
|
||||
card_layout.addWidget(self.downloadOverallProgress)
|
||||
layout.addWidget(self.downloadActiveCard)
|
||||
self.downloadActiveHeading.setVisible(False)
|
||||
self.downloadActiveCard.setVisible(False)
|
||||
|
||||
def _create_download_table(self, heading: str, layout: QVBoxLayout) -> QTableWidget:
|
||||
if heading:
|
||||
label = QLabel(heading)
|
||||
label.setStyleSheet(self.theme.SETTINGS_TITLE_STYLE)
|
||||
layout.addWidget(label)
|
||||
columns = (_("Game Title"), _("Started at"), _("Type"), _("Store"))
|
||||
table = QTableWidget(0, len(columns))
|
||||
table.setHorizontalHeaderLabels(columns)
|
||||
table.setStyleSheet(self.theme.DOWNLOADS_TABLE_STYLE)
|
||||
table.setAlternatingRowColors(True)
|
||||
table.setEditTriggers(QAbstractItemView.EditTrigger.NoEditTriggers)
|
||||
table.setSelectionMode(QAbstractItemView.SelectionMode.NoSelection)
|
||||
table.setShowGrid(False)
|
||||
header = table.horizontalHeader()
|
||||
header.setFixedHeight(self.theme.downloadsTableHeaderHeight)
|
||||
header.setSectionResizeMode(0, QHeaderView.ResizeMode.Stretch)
|
||||
for column in range(1, len(columns)):
|
||||
header.setSectionResizeMode(column, QHeaderView.ResizeMode.ResizeToContents)
|
||||
table.verticalHeader().setVisible(False)
|
||||
self._update_download_table_height(table)
|
||||
layout.addWidget(table)
|
||||
if heading:
|
||||
self.downloadTableHeadings[table] = label
|
||||
label.setVisible(False)
|
||||
table.setVisible(False)
|
||||
return table
|
||||
|
||||
def _update_download_table_height(self, table: QTableWidget) -> None:
|
||||
rows = max(1, table.rowCount())
|
||||
height = self.theme.downloadsTableHeaderHeight + rows * self.theme.downloadsTableRowHeight
|
||||
table.setFixedHeight(height)
|
||||
has_rows = table.rowCount() > 0
|
||||
table.setVisible(has_rows)
|
||||
heading = self.downloadTableHeadings.get(table)
|
||||
if heading is not None:
|
||||
heading.setVisible(has_rows)
|
||||
|
||||
def _clear_completed_downloads(self) -> None:
|
||||
self.downloadCompletedTable.setRowCount(0)
|
||||
self._update_download_table_height(self.downloadCompletedTable)
|
||||
|
||||
def _start_gog_login(self) -> None:
|
||||
QDesktopServices.openUrl(QUrl(GOG_LOGIN_URL))
|
||||
self.gogLoginUrlEdit.setFocus()
|
||||
|
||||
def _submit_gog_login_url(self) -> None:
|
||||
code = self.gog_api.extract_auth_code(self.gogLoginUrlEdit.text().strip())
|
||||
if not code:
|
||||
self.gogAccountStatus.setText(_("The GOG login URL does not contain a code"))
|
||||
return
|
||||
if getattr(self, "gog_auth_worker", None) is not None:
|
||||
return
|
||||
worker = GOGAuthWorker(self.gog_api, code)
|
||||
worker.authenticated.connect(self._on_gog_authenticated)
|
||||
worker.finished.connect(self._on_gog_auth_worker_finished)
|
||||
self.gog_auth_worker = worker
|
||||
self.gogSubmitLoginButton.setEnabled(False)
|
||||
worker.start()
|
||||
|
||||
def _on_gog_authenticated(self, authenticated: bool, error: str) -> None:
|
||||
if not authenticated:
|
||||
message = error or _("Unknown error")
|
||||
self.gogAccountStatus.setText(_("GOG login failed: {0}").format(message))
|
||||
return
|
||||
self.gogAccountStatus.setText(_("GOG account connected"))
|
||||
self._refresh_gog_library()
|
||||
|
||||
def _on_gog_auth_worker_finished(self) -> None:
|
||||
self.gogSubmitLoginButton.setEnabled(True)
|
||||
self.gog_auth_worker = None
|
||||
|
||||
def _update_gog_account_state(self) -> None:
|
||||
connected = self.gog_api.auth_path.is_file()
|
||||
status = (
|
||||
_("GOG account connected")
|
||||
if connected else _("GOG account not connected")
|
||||
)
|
||||
self.gogAccountStatus.setText(status)
|
||||
self.gogRefreshButton.setEnabled(connected)
|
||||
self.gogLoginButton.setEnabled(True)
|
||||
|
||||
def _refresh_gog_library(self) -> None:
|
||||
if getattr(self, "gog_library_worker", None) is not None:
|
||||
return
|
||||
self.gogRefreshButton.setEnabled(False)
|
||||
self.gogAccountStatus.setText(_("Refreshing GOG library…"))
|
||||
worker = GOGLibraryWorker(self.gog_api)
|
||||
worker.loaded.connect(self._on_gog_library_loaded)
|
||||
worker.failed.connect(self._on_gog_library_failed)
|
||||
worker.progress.connect(self._on_gog_library_progress)
|
||||
worker.finished.connect(self._on_gog_library_worker_finished)
|
||||
self.gog_library_worker = worker
|
||||
worker.start()
|
||||
|
||||
def _on_gog_library_progress(self, completed: int, total: int) -> None:
|
||||
self.gogAccountStatus.setText(
|
||||
_("Refreshing GOG library… {0}/{1}").format(completed, total)
|
||||
)
|
||||
|
||||
def _on_gog_library_loaded(self, _games: list) -> None:
|
||||
self.gogAccountStatus.setText(_("GOG library updated"))
|
||||
self.loadGames(force_load=True)
|
||||
|
||||
def _on_gog_library_failed(self, message: str) -> None:
|
||||
self.gogAccountStatus.setText(_("Failed to refresh GOG library: {0}").format(message))
|
||||
|
||||
def _on_gog_library_worker_finished(self) -> None:
|
||||
self.gog_library_worker = None
|
||||
self.gogRefreshButton.setEnabled(True)
|
||||
|
||||
def _install_gog_game(self, game: dict) -> None:
|
||||
if self.gog_process is not None:
|
||||
self.gog_download_queue.append(game)
|
||||
self._append_download_row(self.downloadQueuedTable, game, _("Install"))
|
||||
self.switchTab(6)
|
||||
return
|
||||
app_id = str(game["app_id"])
|
||||
install_path = self.gog_api.get_install_path(app_id, str(game["title"]))
|
||||
install_path.mkdir(parents=True, exist_ok=True)
|
||||
selected_paths: list[str] = []
|
||||
file_explorer = FileExplorer(
|
||||
self,
|
||||
theme=self.theme,
|
||||
initial_path=str(install_path),
|
||||
directory_only=True,
|
||||
)
|
||||
file_explorer.setWindowTitle(_("Select installation folder"))
|
||||
file_explorer.file_signal.file_selected.connect(selected_paths.append)
|
||||
file_explorer.exec()
|
||||
if not selected_paths:
|
||||
return
|
||||
install_path = Path(selected_paths[0])
|
||||
manifest_path = (
|
||||
self.gog_api.config_dir / "heroic_gogdl" / "manifests" / app_id
|
||||
)
|
||||
if manifest_path.is_file() and not self.gog_api.is_game_installed(app_id):
|
||||
try:
|
||||
manifest_path.unlink()
|
||||
except OSError as error:
|
||||
logger.error("Failed to reset incomplete GOG download: %s", error)
|
||||
self.gogAccountStatus.setText(str(error))
|
||||
return
|
||||
try:
|
||||
command = self.gog_api.build_command(
|
||||
["download", app_id, "--path", str(install_path), "--platform", "windows"]
|
||||
)
|
||||
except FileNotFoundError as error:
|
||||
self.gogAccountStatus.setText(str(error))
|
||||
return
|
||||
self._start_gog_download(game, install_path, command, _("Install"))
|
||||
|
||||
def _repair_gog_game(self, game: dict) -> None:
|
||||
if self.gog_process is not None:
|
||||
self.gogAccountStatus.setText(_("Another GOG operation is already running"))
|
||||
return
|
||||
app_id = str(game["app_id"])
|
||||
install_path = self.gog_api.get_installed_path(app_id)
|
||||
if install_path is None:
|
||||
self.gogAccountStatus.setText(_("GOG installation not found"))
|
||||
return
|
||||
try:
|
||||
command = self.gog_api.build_command(
|
||||
["repair", app_id, "--path", str(install_path), "--platform", "windows"]
|
||||
)
|
||||
except FileNotFoundError as error:
|
||||
self.gogAccountStatus.setText(str(error))
|
||||
return
|
||||
self._start_gog_download(game, install_path, command, _("Repair"))
|
||||
|
||||
def _import_gog_game(self, game: dict) -> None:
|
||||
selected_paths: list[str] = []
|
||||
file_explorer = FileExplorer(
|
||||
self, theme=self.theme, initial_path=str(Path.home() / "Games"),
|
||||
directory_only=True,
|
||||
)
|
||||
file_explorer.setWindowTitle(_("Select GOG game folder"))
|
||||
file_explorer.file_signal.file_selected.connect(selected_paths.append)
|
||||
file_explorer.exec()
|
||||
if not selected_paths:
|
||||
return
|
||||
app_id = str(game["app_id"])
|
||||
game_path = self.gog_api.find_install_path(app_id, Path(selected_paths[0]))
|
||||
if game_path is None:
|
||||
self.gogAccountStatus.setText(_("Selected folder is not this GOG game"))
|
||||
return
|
||||
self.gog_api.save_installed_game(
|
||||
app_id, {"install_path": str(game_path), "title": str(game["title"])}
|
||||
)
|
||||
self.gog_api.ensure_launch_parameters(app_id)
|
||||
self.gogAccountStatus.setText(_("GOG game imported"))
|
||||
self.loadGames(force_load=True)
|
||||
|
||||
def _delete_gog_game(self, game: dict) -> None:
|
||||
app_id = str(game["app_id"])
|
||||
install_path = self.gog_api.get_installed_path(app_id)
|
||||
if install_path is None:
|
||||
self.gogAccountStatus.setText(_("GOG installation not found"))
|
||||
return
|
||||
message_box = QMessageBox(self)
|
||||
message_box.setIcon(QMessageBox.Icon.Question)
|
||||
message_box.setWindowTitle(_("Confirm Deletion"))
|
||||
message_box.setText(
|
||||
_("Delete '{0}' and all files in its installation folder?").format(
|
||||
game["title"]
|
||||
)
|
||||
)
|
||||
message_box.setStandardButtons(
|
||||
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No
|
||||
)
|
||||
message_box.setDefaultButton(QMessageBox.StandardButton.No)
|
||||
if message_box.exec() != QMessageBox.StandardButton.Yes:
|
||||
return
|
||||
try:
|
||||
shutil.rmtree(install_path)
|
||||
except OSError as error:
|
||||
logger.error("Failed to delete GOG game %s: %s", app_id, error)
|
||||
self.gogAccountStatus.setText(str(error))
|
||||
return
|
||||
manifest_path = (
|
||||
self.gog_api.config_dir / "heroic_gogdl" / "manifests" / app_id
|
||||
)
|
||||
try:
|
||||
manifest_path.unlink(missing_ok=True)
|
||||
except OSError as error:
|
||||
logger.warning("Failed to delete GOG manifest %s: %s", app_id, error)
|
||||
self.gog_api.remove_installed_game(app_id)
|
||||
self.context_menu_manager.remove_gog_shortcuts(str(game["title"]))
|
||||
self.gogAccountStatus.setText(_("GOG game deleted"))
|
||||
self.loadGames(force_load=True)
|
||||
|
||||
def _start_gog_download(
|
||||
self, game: dict, install_path: Path, command: list[str], action: str
|
||||
) -> None:
|
||||
app_id = str(game["app_id"])
|
||||
self.downloadActiveTitle.setText(str(game["title"]))
|
||||
self.downloadActiveDetails.setText(_("Starting"))
|
||||
cover_width, cover_height = self.theme.downloadsActiveCoverSize
|
||||
load_pixmap_async(
|
||||
str(game.get("cover", "")), cover_width, cover_height,
|
||||
self.downloadActiveCover.setPixmap,
|
||||
app_name=f"download-active-{app_id}",
|
||||
)
|
||||
self.downloadActiveHeading.setVisible(True)
|
||||
self.downloadActiveCard.setVisible(True)
|
||||
process = QProcess(self)
|
||||
process.setProgram(command[0])
|
||||
process.setArguments(command[1:])
|
||||
process.setProcessEnvironment(self._gog_process_environment())
|
||||
process.setProcessChannelMode(QProcess.ProcessChannelMode.MergedChannels)
|
||||
process.readyReadStandardOutput.connect(self._read_gog_download_output)
|
||||
process.finished.connect(self._on_gog_download_finished)
|
||||
self.gog_process = process
|
||||
self.gog_active_app_id = app_id
|
||||
self.gog_active_game = game
|
||||
self.gog_active_install_path = install_path
|
||||
self.gog_active_action = action
|
||||
self.gog_download_output = ""
|
||||
self.downloadOverallProgress.setValue(0)
|
||||
self.downloadCancelButton.setEnabled(True)
|
||||
self.switchTab(6)
|
||||
process.start()
|
||||
|
||||
def _cancel_gog_download(self) -> None:
|
||||
if self.gog_process is None:
|
||||
return
|
||||
process = self.gog_process
|
||||
self.downloadCancelButton.setEnabled(False)
|
||||
self.downloadActiveDetails.setText(_("Cancel"))
|
||||
process.terminate()
|
||||
QTimer.singleShot(
|
||||
GOG_CANCEL_KILL_TIMEOUT_MS,
|
||||
lambda: self._kill_gog_process(process),
|
||||
)
|
||||
|
||||
def _kill_gog_process(self, process: QProcess) -> None:
|
||||
if process.state() != QProcess.ProcessState.NotRunning:
|
||||
logger.warning("gogdl did not terminate after cancellation; killing it")
|
||||
process.kill()
|
||||
|
||||
def _append_download_row(
|
||||
self, table: QTableWidget, game: dict, action: str
|
||||
) -> tuple[int, QLabel]:
|
||||
row = table.rowCount()
|
||||
table.insertRow(row)
|
||||
game_cell, details_label = self._create_download_game_cell(game)
|
||||
table.setCellWidget(row, 0, game_cell)
|
||||
values = (datetime.now().strftime("%H:%M:%S"), action, "GOG")
|
||||
for column, value in enumerate(values, 1):
|
||||
table.setItem(row, column, QTableWidgetItem(value))
|
||||
table.setRowHeight(row, self.theme.downloadsTableRowHeight)
|
||||
if table is self.downloadQueuedTable:
|
||||
details_label.setText(_("Queued"))
|
||||
else:
|
||||
details_label.setText(action)
|
||||
self._update_download_table_height(table)
|
||||
return row, details_label
|
||||
|
||||
def _create_download_game_cell(self, game: dict) -> tuple[QWidget, QLabel]:
|
||||
widget = QWidget()
|
||||
layout = QHBoxLayout(widget)
|
||||
layout.setContentsMargins(*self.theme.downloadsCellMargins)
|
||||
layout.setSpacing(self.theme.downloadsCellSpacing)
|
||||
cover = QLabel()
|
||||
cover_width, cover_height = self.theme.downloadsCoverSize
|
||||
cover.setFixedSize(cover_width, cover_height)
|
||||
layout.addWidget(cover)
|
||||
text_layout = QVBoxLayout()
|
||||
title = QLabel(str(game["title"]))
|
||||
details = QLabel(_("Waiting…"))
|
||||
details.setStyleSheet(self.theme.CONTENT_STYLE)
|
||||
text_layout.addWidget(title)
|
||||
text_layout.addWidget(details)
|
||||
layout.addLayout(text_layout)
|
||||
load_pixmap_async(
|
||||
str(game.get("cover", "")), cover_width, cover_height, cover.setPixmap,
|
||||
app_name=f"download-{game['app_id']}",
|
||||
)
|
||||
return widget, details
|
||||
|
||||
def _gog_process_environment(self) -> QProcessEnvironment:
|
||||
environment = QProcessEnvironment.systemEnvironment()
|
||||
environment.insert("GOGDL_CONFIG_PATH", str(self.gog_api.config_dir))
|
||||
return environment
|
||||
|
||||
def _read_gog_download_output(self) -> None:
|
||||
if self.gog_process is None:
|
||||
return
|
||||
output = bytes(self.gog_process.readAllStandardOutput()).decode(errors="replace")
|
||||
self.gog_download_output = (self.gog_download_output + output)[-4096:]
|
||||
percentages = re.findall(
|
||||
r"Progress:\s+(\d{1,3}(?:\.\d+)?)", self.gog_download_output
|
||||
)
|
||||
if percentages:
|
||||
self.downloadOverallProgress.setValue(int(float(percentages[-1])))
|
||||
download_speeds = re.findall(
|
||||
r"Download\t-\s+(\S+)\s+MiB", self.gog_download_output
|
||||
)
|
||||
disk_speeds = re.findall(
|
||||
r"Disk\t-\s+(\S+)\s+MiB", self.gog_download_output
|
||||
)
|
||||
downloaded = re.findall(
|
||||
r"Downloaded:\s+([\d.]+\s+MiB)", self.gog_download_output
|
||||
)
|
||||
eta = re.findall(r"ETA:\s+(\d{2}:\d{2}:\d{2})", self.gog_download_output)
|
||||
if download_speeds:
|
||||
self.downloadSpeedLabel.setText(
|
||||
_("Download: {0} MiB/s").format(download_speeds[-1])
|
||||
)
|
||||
if disk_speeds:
|
||||
self.diskSpeedLabel.setText(
|
||||
_("Disk: {0} MiB/s").format(disk_speeds[-1])
|
||||
)
|
||||
details = []
|
||||
if downloaded:
|
||||
details.append(downloaded[-1])
|
||||
if download_speeds:
|
||||
details.append(f"{download_speeds[-1]} MiB/s")
|
||||
if eta:
|
||||
details.append(_("ETA: {0}").format(eta[-1]))
|
||||
if details:
|
||||
self._update_active_download_details(details)
|
||||
|
||||
def _update_active_download_details(self, details: list[str]) -> None:
|
||||
self.downloadActiveDetails.setText(" · ".join(details))
|
||||
|
||||
def _on_gog_download_finished(
|
||||
self, code: int, _status: QProcess.ExitStatus
|
||||
) -> None:
|
||||
game = self.gog_active_game
|
||||
install_path = self.gog_active_install_path
|
||||
app_id = str(game["app_id"])
|
||||
self.downloadActiveHeading.setVisible(False)
|
||||
self.downloadActiveCard.setVisible(False)
|
||||
game_path = self.gog_api.find_install_path(app_id, install_path)
|
||||
if code == 0 and game_path is None:
|
||||
code = 1
|
||||
completed_action = _("Failed")
|
||||
if code == 0:
|
||||
completed_action = (
|
||||
_("Repaired") if self.gog_active_action == _("Repair")
|
||||
else _("Installed")
|
||||
)
|
||||
_completed_row, completed_details = self._append_download_row(
|
||||
self.downloadCompletedTable, game, completed_action,
|
||||
)
|
||||
if code == 0:
|
||||
self.gog_api.save_installed_game(
|
||||
app_id, {"install_path": str(game_path), "title": game["title"]}
|
||||
)
|
||||
self._update_installed_gog_detail(app_id)
|
||||
self.loadGames(force_load=True)
|
||||
else:
|
||||
error = self._get_gog_download_error()
|
||||
completed_details.setText(error)
|
||||
logger.error("GOG download failed for %s: %s", app_id, error)
|
||||
self.gog_process = None
|
||||
self.downloadOverallProgress.setValue(0)
|
||||
self.downloadSpeedLabel.setText(_("Download: —"))
|
||||
self.diskSpeedLabel.setText(_("Disk: —"))
|
||||
if self.gog_download_queue:
|
||||
next_game = self.gog_download_queue.pop(0)
|
||||
self.downloadQueuedTable.removeRow(0)
|
||||
self._update_download_table_height(self.downloadQueuedTable)
|
||||
self._install_gog_game(next_game)
|
||||
|
||||
def _get_gog_download_error(self) -> str:
|
||||
lines = [line.strip() for line in self.gog_download_output.splitlines()]
|
||||
error_lines = [line for line in lines if "ERROR" in line.upper()]
|
||||
if error_lines:
|
||||
return error_lines[-1]
|
||||
useful = [line for line in lines if line and "[PROGRESS]" not in line]
|
||||
return useful[-1] if useful else _("Download process failed")
|
||||
|
||||
def _update_installed_gog_detail(self, app_id: str) -> None:
|
||||
install_uri = f"gog://install/{app_id}"
|
||||
if self.current_exec_line != install_uri:
|
||||
return
|
||||
if self.stackedWidget.currentWidget() is not self.currentDetailPage:
|
||||
return
|
||||
source = self.detail_page_manager._current_detail_source
|
||||
if source is None or source[0] != "game":
|
||||
return
|
||||
source[1]["exec_line"] = f"gog://launch/{app_id}"
|
||||
self.current_exec_line = source[1]["exec_line"]
|
||||
self.detail_page_manager._reopen_current_detail_page()
|
||||
@@ -277,8 +277,8 @@ class MainWindowLibraryTabMixin(_MainWindowTypingBase):
|
||||
self.gamesSortCombo.activated.connect(self._delay_library_controls_hover_close)
|
||||
controls_layout.addWidget(self.gamesSortCombo)
|
||||
|
||||
self.filter_keys = ["all", "steam", "portproton", "favorites"]
|
||||
self.filter_labels = [_("All"), "Steam", "PortProton", _("Favorites")]
|
||||
self.filter_keys = ["all", "steam", "gog", "portproton", "favorites"]
|
||||
self.filter_labels = [_("All"), "Steam", "GOG", "PortProton", _("Favorites")]
|
||||
self.gamesDisplayCombo = self._create_library_combo(self.filter_labels, _("Display Filter:"))
|
||||
self._set_combo_current_key(
|
||||
self.gamesDisplayCombo,
|
||||
@@ -384,7 +384,7 @@ class MainWindowLibraryTabMixin(_MainWindowTypingBase):
|
||||
def _get_games_without_exe(self) -> list[tuple]:
|
||||
missing_games = []
|
||||
for game in self.game_library_manager.games:
|
||||
if len(game) <= 12 or game[12] == "steam":
|
||||
if len(game) <= 12 or game[12] != "portproton":
|
||||
continue
|
||||
exec_line = game[5] if len(game) > 5 else ""
|
||||
exe_path = extract_exec_target_path(exec_line) if exec_line else None
|
||||
|
||||
@@ -119,6 +119,32 @@ class MainWindowSettingsTabMixin(_MainWindowTypingBase):
|
||||
genForm.setRowWrapPolicy(QFormLayout.RowWrapPolicy.DontWrapRows)
|
||||
scrollLayout.addWidget(genFrame)
|
||||
|
||||
gogFrame, gogForm = create_section(_("GOG Account"), self.theme)
|
||||
gogForm.setRowWrapPolicy(QFormLayout.RowWrapPolicy.DontWrapRows)
|
||||
scrollLayout.addWidget(gogFrame)
|
||||
self.gogAccountStatus = QLabel()
|
||||
self.gogAccountStatus.setStyleSheet(self.theme.SETTINGS_TITLE_STYLE)
|
||||
self.gogLoginUrlEdit = CustomLineEdit(theme=self.theme)
|
||||
self.gogLoginUrlEdit.setPlaceholderText(_("Paste the final GOG login URL"))
|
||||
self.gogLoginUrlEdit.setStyleSheet(self.theme.LINE_EDIT_STYLE)
|
||||
self.gogLoginButton = AutoSizeButton(_("Open login page"))
|
||||
self.gogLoginButton.setStyleSheet(self.theme.ACTION_BUTTON_STYLE)
|
||||
self.gogLoginButton.clicked.connect(self._start_gog_login)
|
||||
self.gogSubmitLoginButton = AutoSizeButton(_("Confirm URL"))
|
||||
self.gogSubmitLoginButton.setStyleSheet(self.theme.ACTION_BUTTON_STYLE)
|
||||
self.gogSubmitLoginButton.clicked.connect(self._submit_gog_login_url)
|
||||
self.gogRefreshButton = AutoSizeButton(_("Refresh library"))
|
||||
self.gogRefreshButton.setStyleSheet(self.theme.ACTION_BUTTON_STYLE)
|
||||
self.gogRefreshButton.clicked.connect(self._refresh_gog_library)
|
||||
gog_buttons = QHBoxLayout()
|
||||
gog_buttons.addWidget(self.gogLoginButton)
|
||||
gog_buttons.addWidget(self.gogSubmitLoginButton)
|
||||
gog_buttons.addWidget(self.gogRefreshButton)
|
||||
gog_buttons.addStretch()
|
||||
gogForm.addRow(self.gogAccountStatus, gog_buttons)
|
||||
gogForm.addRow(self.gogLoginUrlEdit)
|
||||
self._update_gog_account_state()
|
||||
|
||||
self.timeDetailCombo = CustomComboBox(theme=self.theme)
|
||||
self.timeDetailCombo.view().window().setWindowFlags(
|
||||
Qt.WindowType.Popup | Qt.WindowType.FramelessWindowHint
|
||||
|
||||
@@ -46,6 +46,55 @@ PROGRESS_BAR_STYLE = f"""
|
||||
}}
|
||||
"""
|
||||
|
||||
DOWNLOADS_SUMMARY_STYLE = f"""
|
||||
QFrame#downloadsSummary {{
|
||||
background: {color_surface_elevated};
|
||||
border: {border_thin} {color_border_faint};
|
||||
border-radius: {border_radius_small};
|
||||
}}
|
||||
"""
|
||||
|
||||
DOWNLOADS_ACTIVE_STYLE = f"""
|
||||
QFrame#downloadsActiveCard {{
|
||||
background: {color_surface_elevated};
|
||||
border: {border_thin} {color_border_faint};
|
||||
border-radius: {border_radius_large};
|
||||
}}
|
||||
"""
|
||||
|
||||
DOWNLOADS_SECTION_STYLE = f"""
|
||||
QFrame#downloadsSection {{
|
||||
background: {color_surface_elevated};
|
||||
border: {border_thin} {color_border_faint};
|
||||
border-radius: {border_radius_small};
|
||||
}}
|
||||
"""
|
||||
|
||||
DOWNLOADS_TABLE_STYLE = f"""
|
||||
QTableWidget {{
|
||||
background: {color_surface_elevated};
|
||||
alternate-background-color: {color_surface};
|
||||
border: {border_none};
|
||||
color: {color_text};
|
||||
gridline-color: {color_border_faint};
|
||||
font-family: '{font_family}';
|
||||
font-size: {font_size_value};
|
||||
}}
|
||||
QHeaderView::section {{
|
||||
background: {color_surface};
|
||||
color: {color_text};
|
||||
border: {border_none};
|
||||
padding: 8px;
|
||||
font-family: '{font_family}';
|
||||
font-size: {font_size_value};
|
||||
font-weight: 600;
|
||||
}}
|
||||
QTableWidget::item {{
|
||||
border-bottom: {border_thin} {color_border_faint};
|
||||
padding: 8px;
|
||||
}}
|
||||
"""
|
||||
|
||||
# STATUS BAR STYLE
|
||||
STATUS_BAR_STYLE = f"""
|
||||
QStatusBar {{
|
||||
|
||||
@@ -22,6 +22,17 @@ wineSettingsSetSpacing = 10
|
||||
themeStorePageSpacing = 10
|
||||
themeStoreCardDefaultWidth = 280
|
||||
themeStoreDetailCarouselMinHeight = 300
|
||||
downloadsSummaryHeight = 96
|
||||
downloadsTableHeaderHeight = 38
|
||||
downloadsTableRowHeight = 68
|
||||
downloadsCoverSize = (72, 48)
|
||||
downloadsSectionSpacing = 14
|
||||
downloadsCellMargins = (8, 6, 8, 6)
|
||||
downloadsCellSpacing = 10
|
||||
downloadsActiveCardHeight = 190
|
||||
downloadsActiveCoverSize = (180, 120)
|
||||
downloadsActiveCardMargins = (16, 14, 16, 14)
|
||||
downloadsActiveCardSpacing = 14
|
||||
mangoHudSwitchesColumns = 4
|
||||
mangoHudSwitchesVerticalSpacing = 10
|
||||
mangoHudFpsColumns = 4
|
||||
|
||||
@@ -36,6 +36,22 @@ def test_detail_page_exe_fallback_uses_image_cache(
|
||||
)
|
||||
|
||||
|
||||
def test_gog_size_result_ignores_deleted_labels(monkeypatch: MonkeyPatch) -> None:
|
||||
label = MagicMock()
|
||||
monkeypatch.setattr("portprotonqt.detail_pages.isValid", lambda _label: True)
|
||||
|
||||
DetailPageManager._set_gog_size_values(
|
||||
[lambda: None, lambda: None], (100, 200)
|
||||
)
|
||||
DetailPageManager._set_gog_size_values(
|
||||
[lambda: label, lambda: label], (1024 ** 2, 1024 ** 3)
|
||||
)
|
||||
|
||||
assert [call.args[0] for call in label.setText.call_args_list] == [
|
||||
"1.0 MiB", "1.0 GiB",
|
||||
]
|
||||
|
||||
|
||||
def test_detail_page_loads_exe_fallback_without_cover(monkeypatch: MonkeyPatch) -> None:
|
||||
detail_page = MagicMock()
|
||||
detail_page.property.side_effect = {
|
||||
|
||||
@@ -0,0 +1,197 @@
|
||||
from pathlib import Path
|
||||
from unittest.mock import Mock
|
||||
|
||||
import orjson
|
||||
import pytest
|
||||
|
||||
from portprotonqt.gog_api import GOGAPI, GOG_PRODUCT_LOCALES
|
||||
from portprotonqt.localization import LOCALE_MAP
|
||||
|
||||
|
||||
def test_default_games_directory_is_user_games_folder() -> None:
|
||||
api = GOGAPI()
|
||||
|
||||
assert api.games_dir == Path.home() / "Games"
|
||||
|
||||
|
||||
def test_extract_auth_code() -> None:
|
||||
url = "https://embed.gog.com/on_login_success?origin=client&code=test%20code"
|
||||
|
||||
assert GOGAPI.extract_auth_code(url) == "test code"
|
||||
|
||||
|
||||
def test_load_installed_uses_isolated_data_dir(tmp_path: Path) -> None:
|
||||
api = GOGAPI()
|
||||
api.installed_path = tmp_path / "installed.json"
|
||||
api.installed_path.write_bytes(orjson.dumps({"123": {"title": "Game"}}))
|
||||
|
||||
assert api.load_installed() == {"123": {"title": "Game"}}
|
||||
|
||||
|
||||
def test_remove_installed_game_keeps_other_records(tmp_path: Path) -> None:
|
||||
api = GOGAPI()
|
||||
api.installed_path = tmp_path / "installed.json"
|
||||
api.installed_path.write_bytes(orjson.dumps({"123": {}, "456": {}}))
|
||||
|
||||
api.remove_installed_game("123")
|
||||
|
||||
assert api.load_installed() == {"456": {}}
|
||||
|
||||
|
||||
def test_get_launch_target_reads_primary_task(tmp_path: Path) -> None:
|
||||
api = GOGAPI()
|
||||
api.installed_path = tmp_path / "installed.json"
|
||||
game_dir = tmp_path / "game"
|
||||
executable = game_dir / "BIN/Game.exe"
|
||||
executable.parent.mkdir(parents=True)
|
||||
executable.touch()
|
||||
api.installed_path.write_bytes(orjson.dumps({"123": {"install_path": str(game_dir)}}))
|
||||
info = {
|
||||
"playTasks": [{
|
||||
"type": "FileTask", "path": "bin\\game.exe",
|
||||
"arguments": '-conf "..\\game.conf"', "isPrimary": True,
|
||||
}]
|
||||
}
|
||||
(game_dir / "goggame-123.info").write_bytes(orjson.dumps(info))
|
||||
|
||||
assert api.get_launch_target("123") == str(executable)
|
||||
|
||||
api.ensure_launch_parameters("123")
|
||||
assert 'export LAUNCH_PARAMETERS="-conf ../game.conf"' in Path(
|
||||
f"{executable}.ppdb"
|
||||
).read_text()
|
||||
|
||||
|
||||
def test_is_game_installed_requires_gogdl_metadata(tmp_path: Path) -> None:
|
||||
api = GOGAPI()
|
||||
api.installed_path = tmp_path / "installed.json"
|
||||
game_dir = tmp_path / "game"
|
||||
game_dir.mkdir()
|
||||
api.installed_path.write_bytes(orjson.dumps({"123": {"install_path": str(game_dir)}}))
|
||||
|
||||
assert api.is_game_installed("123") is False
|
||||
|
||||
(game_dir / "goggame-123.info").write_text("{}")
|
||||
assert api.is_game_installed("123") is True
|
||||
|
||||
|
||||
def test_find_install_path_returns_gogdl_directory(tmp_path: Path) -> None:
|
||||
api = GOGAPI()
|
||||
game_dir = tmp_path / "Game created by gogdl"
|
||||
game_dir.mkdir()
|
||||
(game_dir / "goggame-123.info").write_text("{}")
|
||||
|
||||
assert api.find_install_path("123", tmp_path) == game_dir
|
||||
|
||||
|
||||
def test_is_game_installed_discovers_default_gogdl_directory(tmp_path: Path) -> None:
|
||||
api = GOGAPI()
|
||||
api.installed_path = tmp_path / "installed.json"
|
||||
api.games_dir = tmp_path / "games"
|
||||
game_dir = api.games_dir / "Game created by gogdl"
|
||||
game_dir.mkdir(parents=True)
|
||||
(game_dir / "goggame-123.info").write_text("{}")
|
||||
|
||||
assert api.is_game_installed("123") is True
|
||||
|
||||
|
||||
def test_localized_value_prefers_current_language(monkeypatch) -> None:
|
||||
monkeypatch.setattr("portprotonqt.gog_api.get_metadata_language", lambda: "ru")
|
||||
|
||||
assert GOGAPI._localized_value({"*": "English", "ru-RU": "Русский"}) == "Русский"
|
||||
|
||||
|
||||
def test_product_locales_cover_supported_languages() -> None:
|
||||
assert set(GOG_PRODUCT_LOCALES) == {language.lower() for language in LOCALE_MAP}
|
||||
|
||||
|
||||
def test_parse_download_sizes_includes_language_data() -> None:
|
||||
output = (
|
||||
b'{"size":{"*":{"download_size":100,"disk_size":200},'
|
||||
b'"en-US":{"download_size":30,"disk_size":40}},'
|
||||
b'"languages":["en-US"]}'
|
||||
)
|
||||
|
||||
assert GOGAPI.parse_download_sizes(output) == (130, 240)
|
||||
|
||||
|
||||
def test_download_sizes_cache(tmp_path: Path) -> None:
|
||||
api = GOGAPI()
|
||||
api.sizes_path = tmp_path / "sizes.json"
|
||||
|
||||
assert api.get_cached_download_sizes("123") is None
|
||||
api.save_download_sizes("123", (100, 200))
|
||||
|
||||
assert api.get_cached_download_sizes("123") == (100, 200)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("lead", "expected"),
|
||||
[
|
||||
(
|
||||
"Description française.<br><br>Deuxième paragraphe.",
|
||||
"Description française.",
|
||||
),
|
||||
(
|
||||
"<b>Avertissement.</b><br><br>Description.<br><br>Troisième paragraphe.",
|
||||
"<b>Avertissement.</b><br><br>Description.",
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_get_game_loads_localized_product_description(
|
||||
monkeypatch: pytest.MonkeyPatch, lead: str, expected: str
|
||||
) -> None:
|
||||
gamesdb_response = Mock()
|
||||
gamesdb_response.json.return_value = {
|
||||
"game": {"title": {"*": "Game"}, "visible_in_library": True},
|
||||
"summary": {"*": "English description"},
|
||||
}
|
||||
product_response = Mock()
|
||||
product_response.json.return_value = {
|
||||
"description": {"lead": lead},
|
||||
}
|
||||
request = Mock(side_effect=[gamesdb_response, product_response])
|
||||
monkeypatch.setattr("portprotonqt.gog_api.get_metadata_language", lambda: "fr")
|
||||
monkeypatch.setattr("portprotonqt.gog_api.requests.get", request)
|
||||
|
||||
game = GOGAPI()._get_game(
|
||||
{"platform_id": "gog", "external_id": "1207666073"}, "token"
|
||||
)
|
||||
|
||||
assert game["description"] == expected
|
||||
assert request.call_args_list[1].kwargs["params"]["locale"] == "fr-FR"
|
||||
|
||||
|
||||
def test_is_authenticated_requires_token_and_user_id(monkeypatch) -> None:
|
||||
api = GOGAPI()
|
||||
|
||||
monkeypatch.setattr(api, "get_credentials", lambda: {"access_token": "token"})
|
||||
assert api.is_authenticated() is False
|
||||
|
||||
monkeypatch.setattr(
|
||||
api, "get_credentials", lambda: {"access_token": "token", "user_id": "123"}
|
||||
)
|
||||
assert api.is_authenticated() is True
|
||||
|
||||
|
||||
def test_ensure_gogdl_downloads_matching_architecture(tmp_path: Path, monkeypatch) -> None:
|
||||
api = GOGAPI()
|
||||
api.data_dir = tmp_path
|
||||
asset = {"name": "gogdl_linux_x86_64"}
|
||||
response = Mock()
|
||||
response.json.return_value = {"assets": [asset]}
|
||||
monkeypatch.setattr("portprotonqt.gog_api.platform.machine", lambda: "x86_64")
|
||||
monkeypatch.setattr("portprotonqt.gog_api.requests.get", lambda *args, **kwargs: response)
|
||||
monkeypatch.setattr(api, "_download_gogdl_asset", lambda selected: str(selected["name"]))
|
||||
|
||||
assert api.ensure_gogdl() == "gogdl_linux_x86_64"
|
||||
|
||||
|
||||
def test_authenticate_returns_gogdl_error(monkeypatch) -> None:
|
||||
api = GOGAPI()
|
||||
result = Mock(returncode=1, stdout=b"", stderr=b"authorization failed")
|
||||
monkeypatch.setattr(api, "ensure_gogdl", lambda: "/bin/gogdl")
|
||||
monkeypatch.setattr(api, "build_command", lambda arguments: ["gogdl", *arguments])
|
||||
monkeypatch.setattr("portprotonqt.gog_api.subprocess.run", lambda *args, **kwargs: result)
|
||||
|
||||
assert api.authenticate("code") == (False, "authorization failed")
|
||||
+119
-1
@@ -1,10 +1,12 @@
|
||||
"""Tests for main window library data processing."""
|
||||
|
||||
import shlex
|
||||
from collections.abc import Callable
|
||||
from pathlib import Path
|
||||
from queue import Queue
|
||||
from types import SimpleNamespace
|
||||
from typing import Any, cast
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from pytest import MonkeyPatch
|
||||
from PySide6.QtCore import Qt
|
||||
@@ -15,6 +17,7 @@ from portprotonqt.game_library_manager import GameLibraryManager
|
||||
from portprotonqt.main_window import MainWindow
|
||||
from portprotonqt.portproton_api import remove_empty_custom_data_dirs
|
||||
import portprotonqt.tabs.autoinstall_tab as autoinstall_tab_module
|
||||
import portprotonqt.tabs.gog_tab as gog_tab_module
|
||||
import portprotonqt.tabs.library_tab as library_tab_module
|
||||
from portprotonqt.tabs import (
|
||||
MainWindowAutoInstallTabMixin,
|
||||
@@ -24,6 +27,7 @@ from portprotonqt.tabs import (
|
||||
MainWindowWineTabMixin,
|
||||
)
|
||||
from portprotonqt.tabs.autoinstall_tab import MainWindowAutoInstallTabMixin as AutoInstallMixin
|
||||
from portprotonqt.tabs.gog_tab import MainWindowGOGTabMixin as GOGMixin
|
||||
from portprotonqt.tabs.library_tab import MainWindowLibraryTabMixin as LibraryMixin
|
||||
from portprotonqt.tabs.settings_tab import MainWindowSettingsTabMixin as SettingsMixin
|
||||
from portprotonqt.tabs.theme_store import THEME_STORE_ITEM, ThemeStoreMixin
|
||||
@@ -152,6 +156,119 @@ def test_main_window_inherits_all_tab_mixins() -> None:
|
||||
assert issubclass(MainWindow, mixin)
|
||||
|
||||
|
||||
def test_gog_account_state_detects_saved_auth(
|
||||
tmp_path: Path, monkeypatch: MonkeyPatch
|
||||
) -> None:
|
||||
class Control:
|
||||
def __init__(self) -> None:
|
||||
self.text = ""
|
||||
self.enabled = False
|
||||
|
||||
def setText(self, text: str) -> None:
|
||||
self.text = text
|
||||
|
||||
def setEnabled(self, enabled: bool) -> None:
|
||||
self.enabled = enabled
|
||||
|
||||
auth_path = tmp_path / "auth.json"
|
||||
auth_path.write_text("{}")
|
||||
window = SimpleNamespace(
|
||||
gog_api=SimpleNamespace(auth_path=auth_path),
|
||||
gogAccountStatus=Control(),
|
||||
gogRefreshButton=Control(),
|
||||
gogLoginButton=Control(),
|
||||
)
|
||||
monkeypatch.setattr(gog_tab_module, "_", lambda text: text)
|
||||
|
||||
GOGMixin._update_gog_account_state(window)
|
||||
|
||||
assert window.gogAccountStatus.text == "GOG account connected"
|
||||
assert window.gogRefreshButton.enabled is True
|
||||
|
||||
|
||||
def test_repair_gog_game_uses_repair_command(tmp_path: Path) -> None:
|
||||
install_path = tmp_path / "Game"
|
||||
started: list[tuple] = []
|
||||
api = SimpleNamespace(
|
||||
get_installed_path=lambda _app_id: install_path,
|
||||
build_command=lambda arguments: ["gogdl", *arguments],
|
||||
)
|
||||
window = SimpleNamespace(
|
||||
gog_process=None,
|
||||
gog_api=api,
|
||||
gogAccountStatus=SimpleNamespace(setText=lambda _text: None),
|
||||
_start_gog_download=lambda *arguments: started.append(arguments),
|
||||
)
|
||||
|
||||
GOGMixin._repair_gog_game(window, {"app_id": "123", "title": "Game"})
|
||||
|
||||
assert started[0][2] == [
|
||||
"gogdl", "repair", "123", "--path", str(install_path),
|
||||
"--platform", "windows",
|
||||
]
|
||||
|
||||
|
||||
def test_cancel_gog_download_terminates_then_kills(monkeypatch: MonkeyPatch) -> None:
|
||||
process = SimpleNamespace(
|
||||
terminate=MagicMock(),
|
||||
kill=MagicMock(),
|
||||
state=lambda: gog_tab_module.QProcess.ProcessState.Running,
|
||||
)
|
||||
window = SimpleNamespace(
|
||||
gog_process=process,
|
||||
downloadCancelButton=SimpleNamespace(setEnabled=MagicMock()),
|
||||
downloadActiveDetails=SimpleNamespace(setText=MagicMock()),
|
||||
_kill_gog_process=lambda active: GOGMixin._kill_gog_process(window, active),
|
||||
)
|
||||
monkeypatch.setattr(gog_tab_module.QTimer, "singleShot", lambda _delay, callback: callback())
|
||||
|
||||
GOGMixin._cancel_gog_download(window)
|
||||
|
||||
process.terminate.assert_called_once()
|
||||
process.kill.assert_called_once()
|
||||
|
||||
|
||||
def test_import_gog_game_saves_selected_installation(
|
||||
tmp_path: Path, monkeypatch: MonkeyPatch
|
||||
) -> None:
|
||||
selected_path = tmp_path / "Selected"
|
||||
game_path = selected_path / "Game"
|
||||
saved: list[tuple] = []
|
||||
|
||||
class Explorer:
|
||||
def __init__(self, *_args, **_kwargs) -> None:
|
||||
self.callback: Callable[[str], None] = lambda _path: None
|
||||
self.file_signal = SimpleNamespace(
|
||||
file_selected=SimpleNamespace(connect=self._connect)
|
||||
)
|
||||
|
||||
def _connect(self, callback: Callable[[str], None]) -> None:
|
||||
self.callback = callback
|
||||
|
||||
def setWindowTitle(self, _title: str) -> None:
|
||||
return
|
||||
|
||||
def exec(self) -> None:
|
||||
self.callback(str(selected_path))
|
||||
|
||||
api = SimpleNamespace(
|
||||
find_install_path=lambda _app_id, _path: game_path,
|
||||
save_installed_game=lambda *arguments: saved.append(arguments),
|
||||
ensure_launch_parameters=lambda _app_id: None,
|
||||
)
|
||||
window = SimpleNamespace(
|
||||
theme=object(),
|
||||
gog_api=api,
|
||||
gogAccountStatus=SimpleNamespace(setText=lambda _text: None),
|
||||
loadGames=lambda **_kwargs: None,
|
||||
)
|
||||
monkeypatch.setattr(gog_tab_module, "FileExplorer", Explorer)
|
||||
|
||||
GOGMixin._import_gog_game(window, {"app_id": "123", "title": "Game"})
|
||||
|
||||
assert saved == [("123", {"install_path": str(game_path), "title": "Game"})]
|
||||
|
||||
|
||||
def test_library_controls_animation_ignores_game_card_scale_duration() -> None:
|
||||
theme = SimpleNamespace(GAME_CARD_ANIMATION={"scale_anim_duration": 10})
|
||||
|
||||
@@ -732,7 +849,7 @@ def test_remove_empty_custom_data_dirs_keeps_non_empty_dirs(tmp_config_dir: Path
|
||||
assert kept_dir.exists()
|
||||
|
||||
|
||||
def test_get_games_without_exe_skips_existing_and_steam(tmp_path: Path) -> None:
|
||||
def test_get_games_without_exe_only_includes_portproton(tmp_path: Path) -> None:
|
||||
exe_path = tmp_path / "Game.exe"
|
||||
exe_path.write_text("", encoding="utf-8")
|
||||
window = MainWindow.__new__(MainWindow)
|
||||
@@ -756,6 +873,7 @@ def test_get_games_without_exe_skips_existing_and_steam(tmp_path: Path) -> None:
|
||||
"portproton",
|
||||
),
|
||||
("Steam", "", "", "", "", "steam://rungameid/1", "", "", "", "", 0, 0, "steam"),
|
||||
("GOG", "", "", "", "", "gog://install/1", "", "", "", "", 0, 0, "gog"),
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
@@ -188,6 +188,35 @@ class TestCreateDesktopFile:
|
||||
result = create_desktop_file("/nonexistent/game.exe")
|
||||
assert result is None
|
||||
|
||||
def test_gog_launch_uri_uses_common_desktop_entry(
|
||||
self, tmp_path: Path, monkeypatch: Any
|
||||
) -> None:
|
||||
_patch_location(monkeypatch, tmp_path)
|
||||
|
||||
result = create_desktop_file(
|
||||
"gog://launch/123", game_name="GOG Game"
|
||||
)
|
||||
|
||||
assert result is not None
|
||||
entry, _, icon_path = result
|
||||
assert 'Exec=portprotonqt --silent "gog://launch/123"' in entry
|
||||
assert icon_path == "applications-games"
|
||||
|
||||
def test_gog_launch_uri_uses_executable_icon(
|
||||
self, tmp_path: Path, monkeypatch: Any
|
||||
) -> None:
|
||||
_patch_location(monkeypatch, tmp_path)
|
||||
exe = _make_exe(tmp_path)
|
||||
|
||||
result = create_desktop_file(
|
||||
"gog://launch/123", game_name="GOG Game", icon_source=exe
|
||||
)
|
||||
|
||||
assert result is not None
|
||||
entry, _, icon_path = result
|
||||
assert icon_path.endswith("data/img/GOG_Game.png")
|
||||
assert f"Icon={icon_path}" in entry
|
||||
|
||||
def test_no_portproton_returns_none(self, monkeypatch: Any) -> None:
|
||||
monkeypatch.setattr(
|
||||
"portprotonqt.config.portproton.get_portproton_location",
|
||||
|
||||
Reference in new issue
Block a user