feat: added gog integration #229

Merged
Boria138 merged 1 commits from gog into main 2026-07-22 12:51:56 +00:00
19 changed files with 1934 additions and 51 deletions

No files matched your search

+19 -1
View File
@@ -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")
+10 -3
View File
@@ -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()
+117 -27
View File
@@ -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(
+142 -4
View File
@@ -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:
+1
View File
@@ -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,
)
+2 -2
View File
@@ -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
+20 -1
View File
@@ -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)
+471
View File
@@ -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
View File
@@ -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:
+2
View File
@@ -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',
+573
View File
@@ -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()
+3 -3
View File
@@ -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
+26
View File
@@ -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
+16
View File
@@ -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 = {
+197
View File
@@ -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
View File
@@ -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"),
]
)
+29
View File
@@ -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",