feat: Added EGS integration #247

Merged
Boria138 merged 1 commits from egs into main 2026-08-29 05:37:07 +00:00
19 changed files with 1827 additions and 68 deletions

No files matched your search

+1 -1
View File
@@ -6,7 +6,7 @@
<img src="build-aux/share/icons/hicolor/scalable/apps/ru.linux_gaming.PortProtonQt.svg" width="64">
<h1 align="center">PortProtonQt</h1>
<a href="https://git.linux-gaming.ru/Linux-Gaming/PortProtonQt/releases"><img src="https://img.shields.io/badge/Downloads-6382-green?style=flat-square" alt="Total Downloads"></a>
<p align="center">A modern and convenient interface for managing and launching games from PortProton, Steam, and GOG. It brings libraries together in one place and simplifies launching Windows games on Linux.</p>
<p align="center">A modern and convenient interface for managing and launching games from PortProton, Steam, Epic Games Store, and GOG. It brings libraries together in one place and simplifies launching Windows games on Linux.</p>
</div>
### Installation (devel)
+1 -1
View File
@@ -6,7 +6,7 @@
<img src="build-aux/share/icons/hicolor/scalable/apps/ru.linux_gaming.PortProtonQt.svg" width="64">
<h1 align="center">PortProtonQt</h1>
<a href="https://git.linux-gaming.ru/Linux-Gaming/PortProtonQt/releases"><img src="https://img.shields.io/badge/%D0%97%D0%B0%D0%B3%D1%80%D1%83%D0%B7%D0%BA%D0%B8-6382-green?style=flat-square" alt="Всего загрузок"></a>
<p align="center">Современный и удобный интерфейс для управления и запуска игр из PortProton, Steam и GOG. Объединяет библиотеки в одном месте и упрощает запуск Windows-игр на Linux.</p>
<p align="center">Современный и удобный интерфейс для управления и запуска игр из PortProton, Steam, Epic Games Store и GOG. Объединяет библиотеки в одном месте и упрощает запуск Windows-игр на Linux.</p>
</div>
### Установка (devel)
+10
View File
@@ -342,6 +342,10 @@ def main():
if app_id.isdigit():
gog_launch_uri = args.file_or_url
ipc_message = f"gog:{app_id}"
elif args.file_or_url and args.file_or_url.startswith("egs://launch/"):
app_id = args.file_or_url.removeprefix("egs://launch/")
if app_id and "/" not in app_id:
ipc_message = f"egs:{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:
@@ -582,6 +586,12 @@ def main():
window.toggleGame(f"gog://launch/{app_id}")
else:
logger.warning("Invalid GOG app id via IPC: %s", app_id)
elif msg.startswith("egs:"):
app_id = msg[4:].strip()
if app_id and "/" not in app_id:
window.toggleGame(f"egs://launch/{app_id}")
else:
logger.warning("Invalid Epic app id via IPC: %s", app_id)
elif msg.startswith("backup:"):
parts = msg.split(":", 2)
if len(parts) == 3:
+42 -4
View File
@@ -471,12 +471,12 @@ class ContextMenuManager:
return
# Normal menu for games with valid exe or from Steam
is_uninstalled_gog = (
game_card.game_source == "gog"
and game_card.exec_line.startswith("gog://install/")
is_uninstalled_store_game = (
game_card.game_source in ("gog", "egs")
and game_card.exec_line.startswith(("gog://install/", "egs://install/"))
)
is_running = self._is_game_running(game_card)
if not is_uninstalled_gog:
if not is_uninstalled_store_game:
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)
@@ -545,6 +545,44 @@ class ContextMenuManager:
lambda: self.parent._import_gog_game(game)
)
if game_card.game_source == "egs":
app_id = str(game_card.appid)
if game_card.exec_line.startswith("egs://launch/"):
update_action = menu.addAction(
self._get_safe_icon("update"), _("Update")
)
update_action.triggered.connect(
lambda: self.parent._update_egs_game(app_id)
)
repair_action = menu.addAction(
self._get_safe_icon("update"), _("Repair")
)
repair_action.triggered.connect(
lambda: self.parent._repair_egs_game(app_id)
)
eos_enabled = self.parent._is_egs_overlay_enabled(app_id)
eos_action = menu.addAction(
self._get_safe_icon("update"),
_("Disable EOS Overlay")
if eos_enabled else _("Enable EOS Overlay"),
)
eos_action.triggered.connect(
lambda: self.parent._enable_egs_overlay(app_id)
)
delete_action = menu.addAction(
self._get_safe_icon("delete"), _("Delete")
)
delete_action.triggered.connect(
lambda: self.parent._delete_egs_game(app_id)
)
else:
import_action = menu.addAction(
self._get_safe_icon("folder"), _("Import")
)
import_action.triggered.connect(
lambda: self.parent._import_egs_game(app_id)
)
if game_card.game_source == "steam":
desktop_dir = QStandardPaths.writableLocation(
QStandardPaths.StandardLocation.DesktopLocation
+92 -10
View File
@@ -86,6 +86,9 @@ class DetailPageManager:
self._gog_size_requests: dict[
str, tuple[QProcess, list[list[Callable[[], QLabel | None]]]]
] = {}
self._egs_size_requests: dict[
str, tuple[QProcess, list[list[Callable[[], QLabel | None]]]]
] = {}
def openGameDetailPage(self, game_data: dict) -> None:
"""Open detailed game information page."""
@@ -310,6 +313,7 @@ class DetailPageManager:
game_info_layout.addLayout(first_row)
self._setup_gog_size_data(game_data, game_info_layout)
self._setup_egs_size_data(game_data, game_info_layout)
hltb_layout = QHBoxLayout()
hltb_layout.setSpacing(10)
@@ -365,6 +369,75 @@ class DetailPageManager:
)
process.start()
def _setup_egs_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() != "egs":
return
if not exec_line.startswith("egs://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]
api = self.main_window.egs_api
cached = api.get_cached_download_sizes(app_id)
if cached is not None:
self._set_gog_size_values(values, cached)
return
request = self._egs_size_requests.get(app_id)
if request is not None:
request[1].append(values)
return
try:
command = api.build_command(["info", app_id, "--json", "--platform", "Windows"])
except (FileNotFoundError, OSError) as error:
logger.warning("Failed to get EGS download size: %s", error)
return
process = QProcess(self.main_window)
process.setProgram(command[0])
process.setArguments(command[1:])
environment = QProcessEnvironment.systemEnvironment()
environment.insert("LEGENDARY_CONFIG_PATH", str(api.config_dir))
process.setProcessEnvironment(environment)
self._egs_size_requests[app_id] = (process, [values])
process.finished.connect(
lambda code, _status: self._finish_egs_size_request(app_id, code)
)
process.start()
def _finish_egs_size_request(self, app_id: str, code: int) -> None:
request = self._egs_size_requests.pop(app_id, None)
if request is None:
return
process, value_groups = request
if code != 0:
logger.warning("legendary info exited with code %s", code)
process.deleteLater()
return
try:
sizes = self.main_window.egs_api.parse_download_sizes(
bytes(process.readAllStandardOutput().data())
)
except (ValueError, TypeError) as error:
logger.warning("Failed to parse EGS download size: %s", error)
process.deleteLater()
return
self.main_window.egs_api.save_download_sizes(app_id, sizes)
for value_refs in value_groups:
self._set_gog_size_values(value_refs, sizes)
process.deleteLater()
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:
@@ -531,7 +604,9 @@ 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/"):
source = str(game_source).lower()
install_uri = exec_line.startswith(f"{source}://install/")
if source in {"gog", "egs"} and install_uri:
play_button.setText(_("Install"))
play_button.setIcon(
self.main_window.theme_manager.get_icon("update", as_path=True)
@@ -570,7 +645,7 @@ class DetailPageManager:
)
buttons_layout.addWidget(open_folder_button)
elif (
str(game_source).lower() != "gog"
str(game_source).lower() not in {"gog", "egs"}
and not self._has_game_shortcut(game_name)
):
add_button = self._make_action_button(
@@ -588,21 +663,26 @@ class DetailPageManager:
if str(game_source).lower() == "portproton":
self._add_portproton_buttons(buttons_layout, exec_line)
elif (
str(game_source).lower() == "gog"
source in {"gog", "egs"}
and appid
and exec_line.startswith("gog://launch/")
and exec_line.startswith(f"{source}://launch/")
):
gog_exe = self.main_window.gog_api.get_launch_target(str(appid))
if not gog_exe:
store_api = (
self.main_window.gog_api if source == "gog"
else self.main_window.egs_api
)
store_exe = store_api.get_launch_target(str(appid))
if not store_exe:
return buttons_layout
self.main_window.gog_api.ensure_launch_parameters(str(appid))
if source == "gog":
store_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",
store_exe, game_source=source,
)
)
buttons_layout.addWidget(settings_button)
@@ -610,7 +690,9 @@ class DetailPageManager:
_("Create Log"),
self.main_window.theme_manager.get_icon("edit", as_path=True),
)
log_button.clicked.connect(lambda: self.toggleDebugLog(gog_exe, log_button))
log_button.clicked.connect(
lambda: self.toggleDebugLog(store_exe, log_button)
)
buttons_layout.addWidget(log_button)
self._debug_log_button = log_button
folder_button = self._make_action_button(
@@ -618,7 +700,7 @@ class DetailPageManager:
self.main_window.theme_manager.get_icon("search", as_path=True),
)
folder_button.clicked.connect(
lambda: self._open_executable_folder(gog_exe)
lambda: self._open_executable_folder(store_exe)
)
buttons_layout.addWidget(folder_button)
elif str(game_source).lower() == "steam" and appid:
+1 -1
View File
@@ -140,7 +140,7 @@ def _add_source_corner(
) -> None:
"""Add source corner badge to cover frame."""
source = game_source.lower()
if source not in ("steam", "gog", "portproton") or theme_manager is None:
if source not in ("steam", "gog", "egs", "portproton") or theme_manager is None:
return
icon_name = f"badge_{source}"
+438
View File
@@ -0,0 +1,438 @@
"""Epic Games Store integration through Legendary."""
import hashlib
import os
import platform
import shutil
import subprocess
import time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from typing import Any
import orjson
import requests
from portprotonqt.localization import get_store_content_languages
from portprotonqt.logger import get_logger
logger = get_logger(__name__)
EGS_LOGIN_URL = "https://legendary.gl/epiclogin"
LEGENDARY_RELEASE_URL = (
"https://api.github.com/repos/Heroic-Games-Launcher/legendary/releases/latest"
)
LEGENDARY_TIMEOUT = 120
EGS_METADATA_WORKERS = 4
EGS_METADATA_TIMEOUT = 15
EGS_DESCRIPTION_CACHE_VERSION = 2
LEGENDARY_UPDATE_INTERVAL = 30 * 24 * 60 * 60
EGS_API_USER_AGENT = (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 EpicGamesLauncher"
)
class EGSAPI:
"""Provide EGS authentication, library and installation helpers."""
def __init__(self) -> None:
data_home = Path(os.getenv("XDG_DATA_HOME", Path.home() / ".local/share"))
self.data_dir = data_home / "PortProtonQt" / "egs"
self.config_dir = self.data_dir / "legendary"
self.user_path = self.config_dir / "user.json"
self.library_path = self.data_dir / "library.json"
self.sizes_path = self.data_dir / "sizes.json"
self.version_path = self.data_dir / "bin/legendary.version"
self.games_dir = Path.home() / "Games"
def get_legendary_path(self) -> str | None:
"""Return an available Legendary executable."""
bundled = self.data_dir / "bin/legendary"
if bundled.is_file() and os.access(bundled, os.X_OK):
return str(bundled)
return shutil.which("legendary")
def ensure_legendary(self) -> str:
"""Download the current official Heroic Legendary binary if needed."""
existing = self.get_legendary_path()
if existing:
return existing
asset, release_tag = self._get_latest_legendary_release()
return self._download_legendary(asset, release_tag)
def update_legendary(self) -> str:
"""Update the bundled Legendary binary when a newer release exists."""
bundled = self.data_dir / "bin/legendary"
if bundled.is_file() and os.access(bundled, os.X_OK):
try:
last_check = self.version_path.stat().st_mtime
except OSError:
last_check = 0
if time.time() - last_check < LEGENDARY_UPDATE_INTERVAL:
return str(bundled)
asset, release_tag = self._get_latest_legendary_release()
if (
bundled.is_file()
and self._get_legendary_release_tag() == release_tag
):
self.version_path.touch()
return str(bundled)
return self._download_legendary(asset, release_tag)
def _get_latest_legendary_release(self) -> tuple[dict, str]:
"""Return the matching Legendary asset and release tag."""
architecture = {"x86_64": "x86_64", "aarch64": "arm64"}.get(
platform.machine()
)
if not architecture:
raise OSError(f"Unsupported Legendary architecture: {platform.machine()}")
response = requests.get(LEGENDARY_RELEASE_URL, timeout=15)
response.raise_for_status()
release = response.json()
asset_name = f"legendary_linux_{architecture}"
asset = next(
(item for item in release.get("assets", []) if item.get("name") == asset_name),
None,
)
if not asset:
raise OSError(f"Legendary release asset not found: {asset_name}")
release_tag = str(release.get("tag_name", ""))
if not release_tag:
raise OSError("Legendary release tag not found")
return asset, release_tag
def _get_legendary_release_tag(self) -> str:
try:
return self.version_path.read_text(encoding="utf-8").strip()
except OSError:
return ""
def _download_legendary(self, asset: dict, release_tag: str) -> str:
target = self.data_dir / "bin/legendary"
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("Legendary checksum verification failed")
temporary.chmod(0o755)
temporary.replace(target)
self.version_path.write_text(release_tag, encoding="utf-8")
return str(target)
except Exception:
temporary.unlink(missing_ok=True)
raise
def build_command(self, arguments: list[str]) -> list[str]:
"""Build a Legendary command using private configuration paths."""
executable = self.get_legendary_path()
if not executable:
raise FileNotFoundError("Legendary executable not found")
self.config_dir.mkdir(parents=True, exist_ok=True)
self.config_dir.chmod(0o700)
command_arguments = list(arguments)
if "-y" in command_arguments:
command_arguments.remove("-y")
command_arguments.insert(0, "-y")
return [executable, *command_arguments]
def get_environment(self) -> dict[str, str]:
"""Return the environment required by Legendary."""
environment = os.environ.copy()
environment["LEGENDARY_CONFIG_PATH"] = str(self.config_dir)
return environment
def authenticate(self, code: str) -> tuple[bool, str]:
"""Authenticate an Epic account with an authorization code."""
self.ensure_legendary()
result = subprocess.run(
self.build_command(["auth", "--code", code]),
env=self.get_environment(), capture_output=True, check=False,
timeout=LEGENDARY_TIMEOUT,
)
if result.returncode == 0 and self.user_path.is_file():
self.user_path.chmod(0o600)
return True, ""
error = result.stderr.decode(errors="replace").strip()
logger.error("Legendary authentication failed: %s", error)
return False, error or f"Legendary exited with code {result.returncode}"
@staticmethod
def extract_auth_code(text: str) -> str:
"""Extract an Epic authorization code from copied text or JSON."""
value = text.strip()
try:
data = orjson.loads(value)
except orjson.JSONDecodeError:
return value if "://" not in value else ""
if not isinstance(data, dict):
return ""
return str(data.get("authorizationCode", "")).strip()
def logout(self) -> bool:
"""Remove the Epic login and cached library."""
try:
if self.get_legendary_path() and self.user_path.is_file():
subprocess.run(
self.build_command(["auth", "--delete"]),
env=self.get_environment(), capture_output=True, check=False,
timeout=LEGENDARY_TIMEOUT,
)
self.user_path.unlink(missing_ok=True)
self.library_path.unlink(missing_ok=True)
return True
except (OSError, subprocess.SubprocessError) as error:
logger.error("Failed to disconnect Epic account: %s", error)
return False
def get_account_name(self) -> str:
"""Return the Epic display name from Legendary state."""
user = self._load_json(self.user_path, {})
return str(user.get("displayName", "")) if isinstance(user, dict) else ""
def refresh_library(self) -> list[dict]:
"""Refresh and cache the Epic library through Legendary."""
self.update_legendary()
result = subprocess.run(
self.build_command(["list", "--json", "--third-party"]),
env=self.get_environment(), capture_output=True, check=False,
timeout=LEGENDARY_TIMEOUT,
)
if result.returncode != 0:
error = result.stderr.decode(errors="replace").strip()
raise OSError(error or "Failed to refresh Epic library")
raw_games = orjson.loads(result.stdout)
if not isinstance(raw_games, list):
raise OSError("Legendary returned an invalid Epic library")
games = [self._normalize_game(game) for game in raw_games]
games = [game for game in games if game.get("app_id")]
self._enrich_descriptions(games)
self._save_json(self.library_path, games)
return games
def load_library(self) -> list[dict]:
"""Load the cached Epic library."""
data = self._load_json(self.library_path, [])
return data if isinstance(data, list) else []
def clear_library_cache(self) -> None:
"""Remove cached Epic library metadata."""
try:
self.library_path.unlink(missing_ok=True)
except OSError as error:
logger.warning("Failed to clear Epic library cache: %s", error)
def load_installed(self) -> dict[str, dict]:
"""Load Legendary installed game records."""
data = self._load_json(self.config_dir / "installed.json", {})
return data if isinstance(data, dict) else {}
def is_game_installed(self, app_id: str) -> bool:
"""Return whether Legendary records a usable installation."""
return self.get_launch_target(app_id) is not None
@staticmethod
def is_eos_overlay_enabled(prefix_path: Path) -> bool:
"""Return whether the EOS overlay is enabled in a Wine prefix."""
try:
registry = (prefix_path / "user.reg").read_text(
encoding="utf-8", errors="replace"
)
except OSError:
return False
section_start = registry.find(r"[Software\\Epic Games\\EOS]")
if section_start < 0:
return False
section_end = registry.find("\n[", section_start + 1)
section = registry[section_start:section_end if section_end >= 0 else None]
return '"OverlayPath"=' in section
def get_launch_target(self, app_id: str) -> str | None:
"""Return the installed game's Windows executable."""
installed = self.load_installed().get(app_id, {})
install_path = Path(str(installed.get("install_path", "")))
executable = str(installed.get("executable", "")).replace("\\", "/")
target = install_path.joinpath(*executable.split("/"))
return str(target) if executable and target.is_file() else None
def get_cached_download_sizes(self, app_id: str) -> tuple[int, int] | None:
"""Return cached EGS download and disk sizes."""
cache = self._load_json(self.sizes_path, {})
value = cache.get(app_id) if isinstance(cache, dict) else None
if not isinstance(value, list) or len(value) != 2:
return None
sizes = int(value[0]), int(value[1])
return None if sizes[0] > 0 and sizes[1] <= 0 else sizes
def save_download_sizes(self, app_id: str, sizes: tuple[int, int]) -> None:
"""Cache EGS download and disk sizes."""
cache = self._load_json(self.sizes_path, {})
if not isinstance(cache, dict):
cache = {}
cache[app_id] = list(sizes)
self._save_json(self.sizes_path, cache)
@staticmethod
def parse_download_sizes(output: bytes) -> tuple[int, int]:
"""Extract download and install sizes from Legendary JSON."""
data = orjson.loads(output)
values: dict[str, int] = {}
def visit(value: Any) -> None:
if isinstance(value, dict):
for key, item in value.items():
if key in {"download_size", "install_size", "disk_size"}:
try:
values[key] = max(values.get(key, 0), int(item))
except (TypeError, ValueError):
continue
visit(item)
elif isinstance(value, list):
for item in value:
visit(item)
visit(data)
install_size = values.get("install_size") or values.get("disk_size", 0)
return values.get("download_size", 0), install_size
def _enrich_descriptions(self, games: list[dict]) -> None:
cached = {str(game.get("app_id")): game for game in self.load_library()}
pending = []
preferred_locale = get_store_content_languages()[0]
for game in games:
old_game = cached.get(game["app_id"], {})
old_description = str(old_game.get("description", ""))
if (
old_description
and old_description != game["title"]
and old_game.get("description_preference") == preferred_locale
and old_game.get("description_cache_version")
== EGS_DESCRIPTION_CACHE_VERSION
):
game["description"] = old_description
game["description_locale"] = str(old_game.get("description_locale", ""))
game["description_preference"] = preferred_locale
game["description_cache_version"] = EGS_DESCRIPTION_CACHE_VERSION
else:
pending.append(game)
if not pending:
return
worker_count = min(EGS_METADATA_WORKERS, os.cpu_count() or 1)
with ThreadPoolExecutor(max_workers=worker_count) as executor:
descriptions = executor.map(self._get_store_description, pending)
for game, result in zip(pending, descriptions, strict=True):
description, description_locale = result
if description:
game["description"] = description
game["description_locale"] = description_locale
game["description_preference"] = preferred_locale
game["description_cache_version"] = EGS_DESCRIPTION_CACHE_VERSION
def _get_store_description(self, game: dict) -> tuple[str, str]:
slug = self._get_product_slug(str(game.get("namespace", "")), game["title"])
for language in get_store_content_languages():
try:
response = requests.get(
"https://store-content.ak.epicgames.com/api/"
f"{language}/content/products/{slug}",
headers={"User-Agent": EGS_API_USER_AGENT},
timeout=EGS_METADATA_TIMEOUT,
)
if response.status_code == requests.codes.not_found:
continue
response.raise_for_status()
description = self._parse_store_description(response.json())
if description:
return description, language
except requests.RequestException as error:
logger.debug(
"Epic description unavailable for %s: %s", game["app_id"], error
)
return "", ""
return "", ""
@staticmethod
def _parse_store_description(data: Any) -> str:
if not isinstance(data, dict) or not isinstance(data.get("pages"), list):
return ""
page = next(
(item for item in data["pages"]
if isinstance(item, dict) and item.get("type") == "productHome"),
{},
)
about = page.get("data", {}).get("about", {})
short_description = str(about.get("shortDescription") or "").strip()
if short_description:
return short_description
description = str(about.get("description") or "").strip()
return description.split("\n\n", 1)[0].strip()
def _get_product_slug(self, namespace: str, title: str) -> str:
fallback = "-".join("".join(
character if character.isascii() and character.isalnum() else " "
for character in title.lower()
).split())
if not namespace:
return fallback
query = {
"query": "query ProductSlug($namespace: String!) { Catalog { "
"catalogNs(namespace: $namespace) { mappings(pageType: "
"\"productHome\") { pageSlug pageType } } } }",
"variables": {"namespace": namespace},
}
try:
response = requests.post(
"https://launcher.store.epicgames.com/graphql",
json=query,
headers={
"Content-Type": "application/json",
"User-Agent": EGS_API_USER_AGENT,
},
timeout=EGS_METADATA_TIMEOUT,
)
response.raise_for_status()
mappings = response.json()["data"]["Catalog"]["catalogNs"]["mappings"]
return str(mappings[0].get("pageSlug") or fallback) if mappings else fallback
except (KeyError, IndexError, TypeError, requests.RequestException) as error:
logger.debug("Epic product slug unavailable for %s: %s", namespace, error)
return fallback
@staticmethod
def _normalize_game(game: dict) -> dict:
metadata = game.get("metadata", {})
images = metadata.get("keyImages", []) if isinstance(metadata, dict) else []
cover = next(
(str(image.get("url", "")) for image in images
if image.get("type") in ("DieselGameBoxTall", "OfferImageTall")),
"",
)
return {
"app_id": str(game.get("app_name") or game.get("appName") or ""),
"title": str(game.get("app_title") or metadata.get("title") or ""),
"description": str(metadata.get("description", "")),
"cover": cover,
"namespace": str(metadata.get("namespace", "")),
"description_locale": "",
"description_preference": "",
"description_cache_version": 0,
}
@staticmethod
def _load_json(path: Path, default: Any) -> Any:
try:
return orjson.loads(path.read_bytes())
except (OSError, orjson.JSONDecodeError):
return default
@staticmethod
def _save_json(path: Path, data: Any) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(orjson.dumps(data, option=orjson.OPT_INDENT_2))
+16 -2
View File
@@ -273,6 +273,7 @@ class GameCard(QFrame):
self.steam_visible = str(game_source).lower() == "steam"
self.gog_visible = str(game_source).lower() == "gog"
self.egs_visible = str(game_source).lower() == "egs"
self.portproton_visible = str(game_source).lower() == "portproton"
self.ppdb_visible = bool(self.ppdb_id) and not self.economy_mode
@@ -321,8 +322,8 @@ class GameCard(QFrame):
self.coverLabel.setAlignment(Qt.AlignmentFlag.AlignCenter)
self.coverLabel.setStyleSheet(self.theme.COVER_LABEL_STYLE)
if self.missing_executable_path or (
str(self.game_source).lower() == "gog"
and self.exec_line.startswith("gog://install/")
str(self.game_source).lower() in ("gog", "egs")
and self.exec_line.startswith(("gog://install/", "egs://install/"))
):
self.coverOpacity = QGraphicsOpacityEffect(self.coverLabel)
self.coverOpacity.setOpacity(self.theme.missing_exe_cover_opacity)
@@ -370,6 +371,14 @@ class GameCard(QFrame):
)
self.gogLabel.setVisible(self.gog_visible)
egs_icon = self.theme_manager.get_icon("badge_egs", as_path=True)
self.egsLabel = SourceCorner(
icon=egs_icon,
config=corner_config,
parent=self.coverWidget,
)
self.egsLabel.setVisible(self.egs_visible)
portproton_icon = self.theme_manager.get_icon("badge_portproton", as_path=True)
self.portprotonLabel = SourceCorner(
icon=portproton_icon,
@@ -478,6 +487,7 @@ class GameCard(QFrame):
for label, icon_name in (
(self.steamLabel, "badge_steam"),
(self.gogLabel, "badge_gog"),
(self.egsLabel, "badge_egs"),
(self.portprotonLabel, "badge_portproton"),
):
icon = self.theme_manager.get_icon(
@@ -657,6 +667,7 @@ class GameCard(QFrame):
source_ribbons = [
(self.steam_visible, self.steamLabel),
(self.gog_visible, self.gogLabel),
(self.egs_visible, self.egsLabel),
(self.portproton_visible, self.portprotonLabel),
]
@@ -679,6 +690,7 @@ class GameCard(QFrame):
self.protondbLabel.raise_()
self.portprotonLabel.raise_()
self.gogLabel.raise_()
self.egsLabel.raise_()
self.steamLabel.raise_()
except RuntimeError:
pass
@@ -817,6 +829,7 @@ class GameCard(QFrame):
self.economy_mode = ui_config.get_economy_mode()
self.steam_visible = str(self.game_source).lower() == "steam"
self.gog_visible = str(self.game_source).lower() == "gog"
self.egs_visible = str(self.game_source).lower() == "egs"
self.portproton_visible = str(self.game_source).lower() == "portproton"
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
@@ -827,6 +840,7 @@ class GameCard(QFrame):
try:
self.steamLabel.setVisible(self.steam_visible)
self.gogLabel.setVisible(self.gog_visible)
self.egsLabel.setVisible(self.egs_visible)
self.portprotonLabel.setVisible(self.portproton_visible)
self.ppdbLabel.setVisible(self.ppdb_visible and not hidden_badges)
self.protondbLabel.setVisible(protondb_visible and not hidden_badges)
+16 -3
View File
@@ -23,6 +23,7 @@ downloader = Downloader()
logger = get_logger(__name__)
COVER_IMAGE_EXTENSIONS = (".png", ".apng", ".jpg", ".jpeg", ".gif", ".webp", ".jxl", ".svg")
DEFAULT_ANIMATION_DELAY_MS = 100
REMOTE_COVER_DOWNLOAD_TIMEOUT = 15
# Global queue and thread pool for image loading
image_load_queue = Queue()
@@ -489,7 +490,11 @@ def load_pixmap_async(
finish_with(pixmap)
finish_with(load_fallback_pixmap())
downloader.download_async(cover, local_path, timeout=5, callback=on_downloaded)
downloader.download_async(
cover, local_path,
timeout=REMOTE_COVER_DOWNLOAD_TIMEOUT,
callback=on_downloaded,
)
return
except Exception as e:
logger.error(f"URL processing error {cover}: {e}")
@@ -524,7 +529,11 @@ def load_pixmap_async(
logger.info("Downloading SGDB cover for %s -> %s", app_name or "unknown", filename)
finish_with(load_fallback_pixmap())
downloader.download_async(cover, local_path, timeout=5, callback=on_downloaded)
downloader.download_async(
cover, local_path,
timeout=REMOTE_COVER_DOWNLOAD_TIMEOUT,
callback=on_downloaded,
)
return
except Exception as e:
@@ -555,7 +564,11 @@ def load_pixmap_async(
finish_with(pixmap)
finish_with(load_fallback_pixmap())
downloader.download_async(cover, local_path, timeout=5, callback=on_downloaded)
downloader.download_async(
cover, local_path,
timeout=REMOTE_COVER_DOWNLOAD_TIMEOUT,
callback=on_downloaded,
)
return
except Exception as e:
logger.error("Error processing remote image URL %s: %s", cover, str(e))
+19
View File
@@ -36,6 +36,8 @@ LOCALE_MAP = {
'el': 'greek',
}
STORE_CONTENT_LOCALES = {'es': 'es-ES', 'pt': 'pt-BR', 'zh': 'zh-CN'}
# Try local locale directory first, fallback to system for development
_local_localedir = Path(__file__).parent / "locales"
_system_prefix = "/app" if os.getenv("FLATPAK_ID") else os.getenv("SHARUN_DIR", "/usr")
@@ -87,6 +89,23 @@ def get_metadata_language() -> str:
return 'en'
def get_store_content_languages() -> tuple[str, ...]:
"""Return preferred store content locales with English fallback."""
language = get_metadata_language().lower().split('_', 1)[0] or 'en'
system_locale = get_system_locale().lower().replace('-', '_')
if language == 'zh' and any(
variant in system_locale for variant in ('hant', '_tw', '_hk', '_mo')
):
preferred = 'zh-TW'
else:
preferred = STORE_CONTENT_LOCALES.get(language, language)
languages = [preferred]
if language not in languages:
languages.append(language)
if 'en-US' not in languages:
languages.append('en-US')
return tuple(languages)
def read_metadata_translations(metadata_file, language_code):
"""
Read translations from metadata.txt for the specified language.
+280 -17
View File
@@ -8,9 +8,14 @@ import subprocess
import tempfile
import re
import time
from datetime import datetime
from pathlib import Path
from queue import Empty, Queue
from threading import Thread
from typing import TYPE_CHECKING
import psutil
from portprotonqt.logger import get_logger
from portprotonqt.icon_extractor import generate_thumbnail, get_exe_icon_cache_path
from portprotonqt.dialogs import FileExplorer, ExeSettingsDialog
@@ -21,6 +26,7 @@ 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.egs_api import EGSAPI
from portprotonqt.debug_utils import get_prefix_name
from portprotonqt.input_manager import InputManager, MainWindowProtocol
from portprotonqt.context_menu_manager import ContextMenuManager
@@ -67,7 +73,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 GOGLibraryWorker, MainWindowGOGTabMixin
from portprotonqt.tabs.download_tab import GOGLibraryWorker, MainWindowDownloadTabMixin
from portprotonqt.tabs.settings_tab import MainWindowSettingsTabMixin
from portprotonqt.tabs.system_tab import MainWindowSystemTabMixin
from portprotonqt.tabs.theme_tab import MainWindowThemeTabMixin
@@ -76,12 +82,13 @@ from portprotonqt.tabs.workers import MainWindowWorkersMixin
from PySide6.QtWidgets import (QLineEdit, QMainWindow, QWidget, QVBoxLayout, QLabel, QHBoxLayout, QStackedWidget, QComboBox,
QMessageBox, QApplication, QPushButton, QCheckBox)
from PySide6.QtCore import Qt, QEvent, QEventLoop, QUrl, Signal, QTimer, Slot, QProcess, QThread, QFileSystemWatcher, QObject
from PySide6.QtCore import Qt, QEvent, QEventLoop, QUrl, Signal, QTimer, Slot, QProcess, QProcessEnvironment, QThread, QFileSystemWatcher, QObject
from PySide6.QtGui import QColor, QDesktopServices, QHideEvent, QShowEvent, QGuiApplication
from typing import cast
from collections.abc import Callable
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime
STORE_LAUNCH_GRACE_SECONDS = 10
if TYPE_CHECKING:
from portprotonqt.appimage_updater import AppImageUpdateWorker
@@ -144,7 +151,7 @@ class MainWindow(
MainWindowControlHintsMixin,
MainWindowAutoInstallTabMixin,
MainWindowLibraryTabMixin,
MainWindowGOGTabMixin,
MainWindowDownloadTabMixin,
MainWindowSettingsTabMixin,
MainWindowSystemTabMixin,
MainWindowThemeTabMixin,
@@ -375,6 +382,7 @@ class MainWindow(
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.egs_api = EGSAPI()
self.portproton_api = PortProtonAPI(self.downloader)
self.autoInstallCustomDataThread = None
@@ -961,18 +969,21 @@ class MainWindow(
if callable(source_loader):
source_loader(lambda games: self.games_loaded.emit(games))
elif display_filter == "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]
def on_all_games_favorites():
games = [
game for source in results.values() for game in source
if game[0] in favorites
]
self.games_loaded.emit(games)
# Load games from different sources in parallel to prevent blocking
results = {'portproton': [], 'steam': [], 'gog': []}
completed = {'portproton': False, 'steam': False, 'gog': False}
results = {'portproton': [], 'steam': [], 'gog': [], 'egs': []}
completed = dict.fromkeys(results, False)
def check_completion():
if all(completed.values()):
QApplication.processEvents() # Keep UI responsive
on_all_games_favorites(results['portproton'], results['steam'], results['gog'])
on_all_games_favorites()
def portproton_callback(games):
results['portproton'] = games
@@ -991,23 +1002,30 @@ class MainWindow(
completed['gog'] = True
check_completion()
def egs_callback(games):
results['egs'] = games
completed['egs'] = True
check_completion()
self._load_portproton_games_async(portproton_callback)
self._load_steam_games_async(steam_callback)
self._load_gog_games_async(gog_callback)
self._load_egs_games_async(egs_callback)
else:
# For 'all' filter - load games from different sources in parallel to prevent blocking
results = {'portproton': [], 'steam': [], 'gog': []}
completed = {'portproton': False, 'steam': False, 'gog': False}
results = {'portproton': [], 'steam': [], 'gog': [], 'egs': []}
completed = dict.fromkeys(results, False)
def on_all_games():
seen = set()
games = []
for game in results['portproton'] + results['steam'] + results['gog']:
for source_games in results.values():
for game in source_games:
# Unique key: name + exec_line
key = (game[0], game[5])
if key not in seen:
seen.add(key)
games.append(game)
key = (game[0], game[5])
if key not in seen:
seen.add(key)
games.append(game)
QApplication.processEvents() # Keep UI responsive
self.games_loaded.emit(games)
@@ -1033,10 +1051,16 @@ class MainWindow(
completed['gog'] = True
check_completion()
def egs_callback(games):
results['egs'] = games
completed['egs'] = 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)
self._load_egs_games_async(egs_callback)
# Run loading immediately to show status without delay
start_loading()
@@ -1104,6 +1128,54 @@ class MainWindow(
lambda info, current_game=game: on_game_info(current_game, info),
)
def _load_egs_games_async(self, callback: Callable[[list[tuple]], None]) -> None:
library = [game for game in self.egs_api.load_library() if game.get("app_id")]
only_installed = game_config.get_only_installed()
games = []
if not library:
callback(games)
return
from portprotonqt.time_utils import get_statistics_path
statistics_file = get_statistics_path()
processed_count = 0
def on_game_info(game: dict, steam_info: dict) -> None:
nonlocal processed_count
app_id = str(game["app_id"])
installed = self.egs_api.is_game_installed(app_id)
if not only_installed or installed:
target = self.egs_api.get_launch_target(app_id) if installed else None
description = str(game.get("description", ""))
if not description or description == str(game.get("title", "")):
description = str(steam_info.get("description", ""))
playtime = get_playtime_for_exe(
statistics_file, target, exact_path=True
) if target else 0
playtime = playtime or 0
games.append((
str(game.get("title", app_id)), description,
str(game.get("cover", "")), app_id,
steam_info.get("controller_support", ""),
f"egs://{'launch' if installed else 'install'}/{app_id}",
get_last_launch(f"egs-{app_id}") if target else _("Never"),
format_playtime(playtime), steam_info.get("protondb_tier", ""),
steam_info.get("anticheat_status", ""),
get_last_launch_timestamp(f"egs-{app_id}") if target else 0,
playtime, "egs", steam_info.get("anticheat_slug", ""),
steam_info.get("ppdb_id", ""), steam_info.get("ppdb_rating", ""),
steam_info.get("appid", ""),
))
processed_count += 1
if processed_count == len(library):
callback(games)
for game in library:
get_steam_game_info_async(
str(game.get("title", game["app_id"])),
f"egs://launch/{game['app_id']}",
lambda info, current_game=game: on_game_info(current_game, info),
)
def _upgrade_legacy_gog_library(
self, api: GOGAPI, library: list[dict], callback: Callable[[list], None]
) -> bool:
@@ -1593,6 +1665,14 @@ class MainWindow(
):
game_data = dict(game_data)
game_data["exec_line"] = f"gog://launch/{app_id}"
if str(game_data.get("game_source", "")).lower() == "egs":
app_id = str(game_data.get("appid", ""))
if (
exec_line.startswith("egs://install/")
and self.egs_api.is_game_installed(app_id)
):
game_data = dict(game_data)
game_data["exec_line"] = f"egs://launch/{app_id}"
logger.debug(
"Opening detail page for %s from stacked index %d",
game_data.get("name", ""),
@@ -2138,10 +2218,26 @@ class MainWindow(
if exec_line.startswith("gog://launch/"):
app_id = exec_line.rsplit("/", 1)[-1]
return str(self.gog_api.get_launch_target(app_id) or "")
if exec_line.startswith("egs://launch/"):
app_id = exec_line.rsplit("/", 1)[-1]
return str(self.egs_api.get_launch_target(app_id) or "")
return extract_exec_target_path(exec_line) or ""
def _has_running_game_process(self) -> bool:
return any(proc.poll() is None for proc in self.game_processes)
if any(proc.poll() is None for proc in self.game_processes):
return True
target = str(self.target_exe or "").lower()
if target:
for process in psutil.process_iter(attrs=["name"]):
try:
if str(process.info.get("name") or "").lower() == target:
return True
except (psutil.NoSuchProcess, psutil.AccessDenied):
continue
started = getattr(self, "game_start_time", None)
if started is None:
return False
return (datetime.now() - started).total_seconds() < STORE_LAUNCH_GRACE_SECONDS
def _start_launch_output_reader(self, process: subprocess.Popen[str]) -> None:
"""Start background reader for PortProton launch output."""
@@ -2459,6 +2555,169 @@ class MainWindow(
if game:
self._install_gog_game(game)
def _handle_egs_game(self, exec_line: str, button=None) -> None:
action, app_id = exec_line.removeprefix("egs://").split("/", 1)
if action == "install" and self.egs_api.is_game_installed(app_id):
action = "launch"
if action == "launch":
self._launch_egs_game(app_id, button)
return
self._install_egs_game(app_id)
def _install_egs_game(self, app_id: str) -> None:
self._install_egs_download(app_id)
def _repair_egs_game(self, app_id: str) -> None:
self._start_egs_operation(
app_id, ["repair", app_id, "--skip-sdl", "-y"], _("Repair")
)
def _update_egs_game(self, app_id: str) -> None:
self._start_egs_operation(
app_id, ["update", app_id, "--platform", "Windows", "--skip-sdl", "-y"],
_("Update")
)
def _delete_egs_game(self, app_id: str) -> None:
self._start_egs_operation(
app_id, ["uninstall", app_id, "-y"], _("Delete")
)
def _import_egs_game(self, app_id: str) -> None:
selected_paths: list[str] = []
explorer = FileExplorer(
self, theme=self.theme, initial_path=str(Path.home() / "Games"),
directory_only=True,
)
explorer.setWindowTitle(_("Select Epic game folder"))
explorer.file_signal.file_selected.connect(selected_paths.append)
explorer.exec()
if selected_paths:
self._start_egs_operation(app_id, [
"import", app_id, selected_paths[0], "--with-dlcs",
"--platform", "Windows", "-y",
], _("Import"))
def _start_egs_operation(
self, app_id: str, arguments: list[str], action: str
) -> None:
if getattr(self, "egs_process", None) is not None:
QMessageBox.warning(self, _("Error"), _("Another Epic operation is running"))
return
try:
command = self.egs_api.build_command(arguments)
except OSError as error:
QMessageBox.warning(self, _("Error"), str(error))
return
game = next(
(item for item in self.egs_api.load_library()
if str(item.get("app_id", "")) == app_id),
{"app_id": app_id, "title": app_id, "cover": ""},
)
self._start_egs_visible_operation(game, command, action)
def _egs_process_environment(self) -> QProcessEnvironment:
environment = QProcessEnvironment.systemEnvironment()
environment.insert("LEGENDARY_CONFIG_PATH", str(self.egs_api.config_dir))
return environment
def _on_egs_operation_finished(
self, app_id: str, action: str, exit_code: int
) -> None:
self._finish_egs_visible_operation(action, exit_code == 0)
if exit_code != 0:
logger.error(
"Epic %s failed for %s with code %s", action, app_id, exit_code
)
return
if action == "eos-install":
self._enable_egs_overlay(app_id)
return
self.loadGames(force_load=True)
def _enable_egs_overlay(self, app_id: str) -> None:
overlay_path = self.egs_api.data_dir / "eos_overlay"
install_record = self.egs_api.config_dir / "overlay_install.json"
if not install_record.is_file():
self._start_egs_operation(app_id, [
"eos-overlay", "install", "--path", str(overlay_path), "-y",
], "eos-install")
return
target = self.egs_api.get_launch_target(app_id)
prefix_name = get_prefix_name(target)
if not self.portproton_location:
QMessageBox.warning(self, _("Error"), _("PortProton directory not found"))
return
prefix_path = Path(self.portproton_location) / "data/prefixes" / prefix_name
if not (prefix_path / "user.reg").is_file():
QMessageBox.warning(self, _("Error"), _("Wine prefix not found"))
return
action = (
"disable"
if self.egs_api.is_eos_overlay_enabled(prefix_path)
else "enable"
)
self._start_egs_operation(app_id, [
"eos-overlay", action, "--prefix", str(prefix_path),
], f"eos-{action}")
def _is_egs_overlay_enabled(self, app_id: str) -> bool:
target = self.egs_api.get_launch_target(app_id)
if not target or not self.portproton_location:
return False
prefix_name = get_prefix_name(target)
prefix_path = Path(self.portproton_location) / "data/prefixes" / prefix_name
return self.egs_api.is_eos_overlay_enabled(prefix_path)
def _launch_egs_game(self, app_id: str, button=None) -> None:
target = self.egs_api.get_launch_target(app_id)
if not target:
QMessageBox.warning(self, _("Error"), _("Game not found."))
return
if not self.start_sh:
QMessageBox.warning(self, _("Error"), _("PortProton start script not found"))
return
target_name = os.path.basename(target)
if self.game_processes and self.target_exe == target_name:
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.egs_api.build_command([
"launch", app_id, "--wine", self.start_sh[0],
])
process = subprocess.Popen(
command, env=self.egs_api.get_environment(), shell=False,
preexec_fn=os.setsid, stdout=subprocess.PIPE,
stderr=subprocess.STDOUT, text=True, errors="replace",
)
except OSError as error:
logger.error("Failed to launch Epic 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_name
self.current_running_button = button
launch_time = datetime.now()
self.game_start_time = launch_time
self.game_start_exe = target
self.game_start_exact_path = True
save_last_launch(f"egs-{app_id}", launch_time)
self._update_last_launch_after_start(target, launch_time)
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 _launch_gog_game(
self, app_id: str, button=None, play_sound: bool = True
) -> None:
@@ -2528,6 +2787,10 @@ class MainWindow(
self._handle_gog_game(exec_line, button)
return
if exec_line.startswith("egs://"):
self._handle_egs_game(exec_line, button)
return
# Handle PortProton games
entry_exec_split = shlex.split(exec_line)
if not entry_exec_split:
+2 -1
View File
@@ -13,6 +13,7 @@ install_data(
'game_card.py',
'game_library_manager.py',
'gog_api.py',
'egs_api.py',
'howlongtobeat_api.py',
'image_utils.py',
'icon_extractor.py',
@@ -152,7 +153,7 @@ install_data(
['tabs/__init__.py',
'tabs/autoinstall_tab.py',
'tabs/control_hints.py',
'tabs/gog_tab.py',
'tabs/download_tab.py',
'tabs/library_tab.py',
'tabs/settings_tab.py',
'tabs/system_tab.py',
+5 -5
View File
@@ -625,7 +625,7 @@ def get_steam_game_info_async(
from portprotonqt.steam_api.utils import filter_candidates, remove_duplicates
is_autoinstall = exec_line.startswith("autoinstall:")
is_gog_uri = exec_line.startswith("gog://")
is_store_uri = exec_line.startswith(("gog://", "egs://"))
if game_exe.lower().endswith(('.bat', '.cmd')):
if os.path.exists(game_exe):
@@ -647,7 +647,7 @@ def get_steam_game_info_async(
else:
logger.error("Command file not found: %s", game_exe)
if is_gog_uri:
if is_store_uri:
meta_data = {}
elif not game_exe.lower().endswith('.exe'):
logger.error("Invalid executable path: %s. Expected .exe", game_exe)
@@ -671,15 +671,15 @@ def get_steam_game_info_async(
candidates.append(file_description)
if desktop_name:
candidates.append(desktop_name)
if exe_name and not is_gog_uri:
if exe_name and not is_store_uri:
candidates.append(exe_name)
if folder_name and not is_gog_uri:
if folder_name and not is_store_uri:
candidates.append(folder_name)
candidates = filter_candidates(candidates)
candidates = remove_duplicates(candidates)
candidates_ordered = sorted(candidates, key=lambda s: len(s.split()), reverse=True)
has_custom_data = (
is_autoinstall or is_gog_uri or _has_custom_data_for_exe(exe_name)
is_autoinstall or is_store_uri or _has_custom_data_for_exe(exe_name)
)
def on_steam_apps_and_index(
@@ -1,4 +1,4 @@
"""Shared download manager with GOG download support."""
"""Store account actions and shared game download manager."""
import os
import re
@@ -27,6 +27,7 @@ from PySide6.QtWidgets import (
)
from portprotonqt.dialogs.file_explorer import FileExplorer
from portprotonqt.egs_api import EGSAPI, EGS_LOGIN_URL
from portprotonqt.gog_api import GOGAPI, GOG_LOGIN_URL
from portprotonqt.image_utils import load_pixmap_async
from portprotonqt.localization import _
@@ -36,6 +37,7 @@ from portprotonqt.sound_manager import SoundManager
logger = get_logger(__name__)
GOG_CANCEL_KILL_TIMEOUT_MS = 3000
GOG_LOGIN_TIMEOUT_MS = 300000
EGS_LOGIN_TIMEOUT_MS = 300000
if TYPE_CHECKING:
from PySide6.QtWidgets import QMainWindow
@@ -64,6 +66,42 @@ class GOGLibraryWorker(QThread):
self.failed.emit(str(error))
class EGSLibraryWorker(QThread):
"""Refresh the Epic library outside the UI thread."""
loaded = Signal(list)
failed = Signal(str)
def __init__(self, api: EGSAPI) -> None:
super().__init__()
self.api = api
def run(self) -> None:
try:
self.loaded.emit(self.api.refresh_library())
except Exception as error:
logger.exception("Failed to refresh Epic library")
self.failed.emit(str(error))
class EGSAuthWorker(QThread):
"""Authenticate an Epic account outside the UI thread."""
authenticated = Signal(bool, str)
def __init__(self, api: EGSAPI, code: str) -> None:
super().__init__()
self.api = api
self.code = code
def run(self) -> None:
try:
self.authenticated.emit(*self.api.authenticate(self.code))
except Exception as error:
logger.exception("Failed to authenticate with Epic")
self.authenticated.emit(False, str(error))
class GOGAuthWorker(QThread):
"""Exchange the GOG OAuth code outside the UI thread."""
@@ -126,7 +164,7 @@ class GOGSupportWorker(QThread):
self.failed.emit(self.error)
class MainWindowGOGTabMixin(_MainWindowTypingBase):
class MainWindowDownloadTabMixin(_MainWindowTypingBase):
"""Add account actions and the shared downloads page."""
if TYPE_CHECKING:
@@ -134,8 +172,11 @@ class MainWindowGOGTabMixin(_MainWindowTypingBase):
def createGOGDownloadsTab(self) -> None:
self.gog_process = None
self.egs_process = None
self.gog_download_queue = []
self.gog_download_output = ""
self.egs_download_output = ""
self.egs_download_total = 0.0
self.gog_support_workers = []
self.downloadTableHeadings = {}
page = QWidget()
@@ -402,6 +443,124 @@ class MainWindowGOGTabMixin(_MainWindowTypingBase):
def _on_gog_library_worker_finished(self) -> None:
self.gog_library_worker = None
def _handle_egs_account_action(self) -> None:
if not self.egs_api.user_path.is_file():
self._start_egs_login()
return
if not self.egs_api.logout():
self.egsAccountStatus.setText(_("Failed to log out of Epic account"))
return
self._update_egs_account_state()
self.loadGames(force_load=True)
def _start_egs_login(self) -> None:
if getattr(self, "egs_auth_worker", None) is not None:
return
clipboard = QApplication.clipboard()
if not getattr(self, "egs_auth_clipboard_connected", False):
clipboard.dataChanged.connect(self._check_egs_login_clipboard)
self.egs_auth_clipboard_connected = True
self.egs_auth_timer = QTimer(self)
self.egs_auth_timer.setSingleShot(True)
self.egs_auth_timer.timeout.connect(self._cancel_egs_login)
self.egs_auth_timer.start(EGS_LOGIN_TIMEOUT_MS)
self.egsLoginButton.setEnabled(False)
self.egsAccountStatus.setText(_("Copy the authorization code to clipboard"))
if not QDesktopServices.openUrl(QUrl(EGS_LOGIN_URL)):
self._cancel_egs_login()
def _check_egs_login_clipboard(self) -> None:
code = self.egs_api.extract_auth_code(QApplication.clipboard().text())
if len(code) < 16:
return
self._stop_egs_login_clipboard()
self.egsAccountStatus.setText(_("Refreshing…"))
worker = EGSAuthWorker(self.egs_api, code)
worker.authenticated.connect(self._on_egs_authenticated)
worker.finished.connect(self._on_egs_auth_worker_finished)
self.egs_auth_worker = worker
worker.start()
def _stop_egs_login_clipboard(self) -> None:
timer = getattr(self, "egs_auth_timer", None)
if timer is not None:
timer.stop()
timer.deleteLater()
self.egs_auth_timer = None
if getattr(self, "egs_auth_clipboard_connected", False):
QApplication.clipboard().dataChanged.disconnect(
self._check_egs_login_clipboard
)
self.egs_auth_clipboard_connected = False
def _cancel_egs_login(self) -> None:
self._stop_egs_login_clipboard()
self.egsLoginButton.setEnabled(True)
self._update_egs_account_state()
def _on_egs_authenticated(self, authenticated: bool, error: str) -> None:
if not authenticated:
self.egsAccountStatus.setText(
_("Epic login failed: {0}").format(error or _("Unknown error"))
)
return
self._update_egs_account_state()
self._refresh_egs_library()
def _on_egs_auth_worker_finished(self) -> None:
self.egsLoginButton.setEnabled(True)
self.egs_auth_worker = None
def _update_egs_account_state(self) -> None:
connected = self.egs_api.user_path.is_file()
account_name = self.egs_api.get_account_name() if connected else ""
self.egsAccountStatus.setText(
account_name or (_("Epic account connected") if connected
else _("Epic account not connected"))
)
self.egsLoginButton.setText(_("Log out") if connected else _("Open login page"))
self.egsLoginButton.setEnabled(True)
def _refresh_egs_library(self) -> None:
if not self.egs_api.user_path.is_file():
return
if getattr(self, "egs_library_worker", None) is not None:
return
self.egsAccountStatus.setText(_("Refreshing…"))
worker = EGSLibraryWorker(self.egs_api)
worker.loaded.connect(self._on_egs_library_loaded)
worker.failed.connect(self._on_egs_library_failed)
worker.finished.connect(self._on_egs_library_worker_finished)
self.egs_library_worker = worker
worker.start()
def _on_egs_library_loaded(self, games: list) -> None:
self._update_egs_account_state()
manager = self.detail_page_manager
source = manager._current_detail_source
if source and source[0] == "game":
source_data = source[1]
if str(source_data.get("game_source", "")).lower() == "egs":
app_id = str(source_data.get("appid", ""))
game = next(
(item for item in games if str(item.get("app_id", "")) == app_id),
None,
)
if game:
source_data["description"] = str(game.get("description", ""))
source_data["cover_path"] = str(game.get("cover", ""))
if manager._detail_page_active:
manager._reopen_current_detail_page()
self.loadGames(force_load=True)
def _on_egs_library_failed(self, error: str) -> None:
self.egsAccountStatus.setText(
_("Failed to refresh Epic library: {0}").format(error)
)
def _on_egs_library_worker_finished(self) -> None:
self.egs_library_worker = None
def _install_gog_game(self, game: dict) -> None:
if self.gog_process is not None:
self.gog_download_queue.append(game)
@@ -449,6 +608,36 @@ class MainWindowGOGTabMixin(_MainWindowTypingBase):
return
self._start_gog_download(game, install_path, command, _("Install"))
def _install_egs_download(self, app_id: str) -> None:
if self.gog_process is not None or self.egs_process is not None:
QMessageBox.warning(self, _("Error"), _("Another download is running"))
return
game = next(
(item for item in self.egs_api.load_library()
if str(item.get("app_id", "")) == app_id),
{"app_id": app_id, "title": app_id, "cover": ""},
)
selected_paths: list[str] = []
explorer = FileExplorer(
self, theme=self.theme, initial_path=str(self.egs_api.games_dir),
directory_only=True,
)
explorer.setWindowTitle(_("Select installation folder"))
explorer.file_signal.file_selected.connect(selected_paths.append)
explorer.exec()
if not selected_paths:
return
install_path = Path(selected_paths[0])
try:
command = self.egs_api.build_command([
"install", app_id, "--base-path", str(install_path),
"--platform", "Windows", "--skip-sdl", "--skip-dlcs", "-y",
])
except OSError as error:
self.egsAccountStatus.setText(str(error))
return
self._start_egs_download(game, install_path, command)
def _repair_gog_game(self, game: dict) -> None:
if self.gog_process is not None:
self.gogAccountStatus.setText(_("Another GOG operation is already running"))
@@ -567,10 +756,78 @@ class MainWindowGOGTabMixin(_MainWindowTypingBase):
self.switchTab(6)
process.start()
def _start_egs_download(
self, game: dict, install_path: Path, command: list[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)
self._update_downloads_tab_visibility()
process = QProcess(self)
process.setProgram(command[0])
process.setArguments(command[1:])
process.setProcessEnvironment(self._egs_process_environment())
process.setProcessChannelMode(QProcess.ProcessChannelMode.MergedChannels)
process.readyReadStandardOutput.connect(self._read_egs_download_output)
process.finished.connect(self._on_egs_download_finished)
self.egs_process = process
self.egs_active_game = game
self.egs_active_install_path = install_path
self.egs_download_output = ""
self.egs_download_total = 0.0
self.downloadOverallProgress.setValue(0)
self.downloadCancelButton.setEnabled(True)
self.switchTab(6)
process.start()
def _start_egs_visible_operation(
self, game: dict, command: list[str], action: str
) -> None:
app_id = str(game["app_id"])
display_action = "EOS Overlay" if action.startswith("eos-") else action
self.downloadActiveTitle.setText(str(game["title"]))
self.downloadActiveDetails.setText(display_action)
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)
self._update_downloads_tab_visibility()
process = QProcess(self)
process.setProgram(command[0])
process.setArguments(command[1:])
process.setProcessEnvironment(self._egs_process_environment())
process.setProcessChannelMode(QProcess.ProcessChannelMode.MergedChannels)
process.readyReadStandardOutput.connect(self._read_egs_download_output)
process.finished.connect(
lambda code, _status: self._on_egs_operation_finished(
app_id, action, code
)
)
self.egs_process = process
self.egs_active_game = game
self.egs_download_output = ""
self.egs_download_total = 0.0
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:
process = self.gog_process or self.egs_process
if process is None:
return
process = self.gog_process
self.downloadCancelButton.setEnabled(False)
self.downloadActiveDetails.setText(_("Cancel"))
process.terminate()
@@ -585,13 +842,13 @@ class MainWindowGOGTabMixin(_MainWindowTypingBase):
process.kill()
def _append_download_row(
self, table: QTableWidget, game: dict, action: str
self, table: QTableWidget, game: dict, action: str, store: str = "GOG"
) -> 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")
values = (datetime.now().strftime("%H:%M:%S"), action, store)
for column, value in enumerate(values, 1):
table.setItem(row, column, QTableWidgetItem(value))
table.setRowHeight(row, self.theme.downloadsTableRowHeight)
@@ -670,6 +927,57 @@ class MainWindowGOGTabMixin(_MainWindowTypingBase):
if details:
self._update_active_download_details(details)
def _read_egs_download_output(self) -> None:
if self.egs_process is None:
return
output = bytes(
self.egs_process.readAllStandardOutput().data()
).decode(errors="replace")
self.egs_download_output = (self.egs_download_output + output)[-4096:]
total = re.findall(r"Download size:\s+([\d.]+)\s+MiB", output)
if total:
self.egs_download_total = float(total[-1])
verification = re.findall(
r"Verification progress:\s*\d+/\d+\s+\(([\d.]+)%\)"
r"(?:\s+\[([\d.]+)\s+MiB/s\])?",
self.egs_download_output,
)
if verification:
percent, speed = verification[-1]
self.downloadOverallProgress.setValue(min(100, int(float(percent))))
details = [f"{percent}%"]
if speed:
details.append(f"{speed} MiB/s")
self._update_active_download_details(details)
downloaded = re.findall(
r"Downloaded:\s+([\d.]+)\s+MiB", self.egs_download_output
)
if downloaded and self.egs_download_total > 0:
progress = float(downloaded[-1]) / self.egs_download_total * 100
self.downloadOverallProgress.setValue(min(100, int(progress)))
download_speeds = re.findall(
r"Download\t-\s+(\S+)\s+MiB", self.egs_download_output
)
disk_speeds = re.findall(
r"Disk\t-\s+(\S+)\s+MiB", self.egs_download_output
)
eta = re.findall(r"ETA:\s+(\d{2}:\d{2}:\d{2})", self.egs_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 = [f"{downloaded[-1]} MiB"] if downloaded else []
if eta:
details.append(_("ETA: {0}").format(eta[-1]))
if details:
self._update_active_download_details(details)
elif output.strip():
self.downloadActiveDetails.setText(output.strip().splitlines()[-1])
def _update_active_download_details(self, details: list[str]) -> None:
self.downloadActiveDetails.setText(" · ".join(details))
@@ -713,6 +1021,49 @@ class MainWindowGOGTabMixin(_MainWindowTypingBase):
self._update_download_table_height(self.downloadQueuedTable)
self._install_gog_game(next_game)
def _on_egs_download_finished(
self, code: int, _status: QProcess.ExitStatus
) -> None:
game = self.egs_active_game
app_id = str(game["app_id"])
self.downloadActiveHeading.setVisible(False)
self.downloadActiveCard.setVisible(False)
completed_action = _("Installed") if code == 0 else _("Failed")
_row, completed_details = self._append_download_row(
self.downloadCompletedTable, game, completed_action, "Epic Games"
)
if code == 0:
self._update_installed_egs_detail(app_id)
self.loadGames(force_load=True)
else:
error = self._get_egs_download_error()
completed_details.setText(error)
logger.error("Epic download failed for %s: %s", app_id, error)
process = self.egs_process
self.egs_process = None
if process is not None:
process.deleteLater()
self.downloadOverallProgress.setValue(0)
self.downloadSpeedLabel.setText(_("Downloading: ") + "\u2014")
self.diskSpeedLabel.setText(_("Disk: ") + "\u2014")
def _finish_egs_visible_operation(self, action: str, success: bool) -> None:
game = self.egs_active_game
display_action = "EOS Overlay" if action.startswith("eos-") else action
self.downloadActiveHeading.setVisible(False)
self.downloadActiveCard.setVisible(False)
_row, details = self._append_download_row(
self.downloadCompletedTable, game, display_action, "Epic Games"
)
details.setText(_("Completed") if success else self._get_egs_download_error())
process = self.egs_process
self.egs_process = None
if process is not None:
process.deleteLater()
self.downloadOverallProgress.setValue(0)
self.downloadSpeedLabel.setText(_("Downloading: ") + "\u2014")
self.diskSpeedLabel.setText(_("Disk: ") + "\u2014")
def _start_gog_support_setup(self, app_id: str, button=None) -> None:
if getattr(self, "gog_support_workers", []):
return
@@ -792,6 +1143,14 @@ class MainWindowGOGTabMixin(_MainWindowTypingBase):
useful = [line for line in lines if line and "[PROGRESS]" not in line]
return useful[-1] if useful else _("Download process failed")
def _get_egs_download_error(self) -> str:
lines = [line.strip() for line in self.egs_download_output.splitlines()]
errors = [line for line in lines if "ERROR" in line.upper()]
if errors:
return errors[-1]
useful = [line for line in lines if 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:
@@ -804,3 +1163,16 @@ class MainWindowGOGTabMixin(_MainWindowTypingBase):
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()
def _update_installed_egs_detail(self, app_id: str) -> None:
install_uri = f"egs://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"egs://launch/{app_id}"
self.current_exec_line = source[1]["exec_line"]
self.detail_page_manager._reopen_current_detail_page()
+2 -1
View File
@@ -302,11 +302,12 @@ class MainWindowLibraryTabMixin(_MainWindowTypingBase):
display_filter = game_config.get_display_filter()
only_installed = game_config.get_only_installed()
self.filter_keys = ["all", "steam", "gog", "portproton", "favorites"]
self.filter_keys = ["all", "steam", "gog", "egs", "portproton", "favorites"]
self.filter_labels = [
_("All"),
"Steam",
"GOG",
"Epic Games",
"PortProton",
_("Favorites"),
]
+20 -1
View File
@@ -167,6 +167,23 @@ class MainWindowSettingsTabMixin(_MainWindowTypingBase):
gog_account_title.setStyleSheet(self.theme.SETTINGS_TITLE_STYLE)
gogForm.addRow(gog_account_title, gog_buttons)
self._update_gog_account_state()
self.egsAccountStatus = QLabel()
self.egsAccountStatus.setStyleSheet(self.theme.SETTINGS_TITLE_STYLE)
self.egsAccountStatus.setSizePolicy(
QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed
)
self.egsLoginButton = AutoSizeButton(_("Open login page"))
self.egsLoginButton.setProperty("theme_style_name", "ACTION_BUTTON_STYLE")
self.egsLoginButton.setStyleSheet(self.theme.ACTION_BUTTON_STYLE)
self.egsLoginButton.setFocusPolicy(Qt.FocusPolicy.StrongFocus)
self.egsLoginButton.clicked.connect(self._handle_egs_account_action)
egs_buttons = QHBoxLayout()
egs_buttons.addWidget(self.egsAccountStatus)
egs_buttons.addWidget(self.egsLoginButton)
egs_account_title = QLabel(_("Account") + " Epic Games:")
egs_account_title.setStyleSheet(self.theme.SETTINGS_TITLE_STYLE)
gogForm.addRow(egs_account_title, egs_buttons)
self._update_egs_account_state()
self.timeDetailCombo = CustomComboBox(theme=self.theme)
self.timeDetailCombo.view().window().setWindowFlags(
@@ -224,7 +241,8 @@ class MainWindowSettingsTabMixin(_MainWindowTypingBase):
uiForm.addRow(self.trayMenuModeTitle, self.trayMenuModeCombo)
self.setTabOrder(self.timeDetailCombo, self.steamAccountCombo)
self.setTabOrder(self.steamAccountCombo, self.gogLoginButton)
self.setTabOrder(self.gogLoginButton, self.trayMenuModeCombo)
self.setTabOrder(self.gogLoginButton, self.egsLoginButton)
self.setTabOrder(self.egsLoginButton, self.trayMenuModeCombo)
self.gamepad_type_keys = ["auto", "xbox", "playstation"]
self.gamepad_type_labels = [_("Auto"), "Xbox", "PlayStation"]
@@ -758,6 +776,7 @@ class MainWindowSettingsTabMixin(_MainWindowTypingBase):
if reply == QMessageBox.StandardButton.Yes:
cache_config.clear_cache()
self.gog_api.clear_library_cache()
self.egs_api.clear_library_cache()
def integrateAppImage(self) -> None:
"""Install the running AppImage in the user application menu."""
+30
View File
@@ -574,6 +574,36 @@ def test_non_steam_without_shortcut_shows_add_not_edit(monkeypatch) -> None:
assert add_count == 1
def test_egs_install_detail_shows_install_button(monkeypatch) -> None:
monkeypatch.setattr("portprotonqt.detail_pages._", lambda text: text)
manager = _make_manager_with_shortcut_mock(has_shortcut=False, monkeypatch=monkeypatch)
play_button = cast(Any, manager._create_play_button).return_value
manager._create_game_buttons_layout({
"game_source": "egs",
"exec_line": "egs://install/doom64",
"appid": "doom64",
"name": "DOOM 64",
})
play_button.setText.assert_called_once_with("Install")
def test_installed_egs_detail_shows_store_controls(monkeypatch) -> None:
monkeypatch.setattr("portprotonqt.detail_pages._", lambda text: text)
manager = _make_manager_with_shortcut_mock(has_shortcut=False, monkeypatch=monkeypatch)
manager.main_window.egs_api.get_launch_target.return_value = "/games/DOOM64.exe"
manager._create_game_buttons_layout({
"game_source": "egs",
"exec_line": "egs://launch/doom64",
"appid": "doom64",
"name": "DOOM 64",
})
assert _get_added_buttons(manager) == ["Settings", "Create Log", "Open Folder"]
def test_steam_with_appid_shows_edit_not_add(monkeypatch) -> None:
manager = _make_manager_with_shortcut_mock(has_shortcut=False, monkeypatch=monkeypatch)
manager._create_game_buttons_layout({
+226
View File
@@ -0,0 +1,226 @@
from pathlib import Path
from types import SimpleNamespace
from typing import Any
import orjson
from pytest import MonkeyPatch
from portprotonqt.egs_api import EGSAPI
from portprotonqt.localization import get_store_content_languages
def test_build_command_uses_isolated_legendary_config(
tmp_path: Path, monkeypatch
) -> None:
api = EGSAPI()
api.config_dir = tmp_path / "legendary"
monkeypatch.setattr(api, "get_legendary_path", lambda: "/bin/legendary")
assert api.build_command(["list", "--json"]) == [
"/bin/legendary", "list", "--json",
]
assert api.get_environment()["LEGENDARY_CONFIG_PATH"] == str(api.config_dir)
def test_build_command_places_confirmation_before_subcommand(
tmp_path: Path, monkeypatch
) -> None:
api = EGSAPI()
api.config_dir = tmp_path / "legendary"
monkeypatch.setattr(api, "get_legendary_path", lambda: "/bin/legendary")
assert api.build_command(["install", "Game", "-y"]) == [
"/bin/legendary", "-y", "install", "Game",
]
def test_update_legendary_skips_matching_release(tmp_path: Path, monkeypatch) -> None:
api = EGSAPI()
api.data_dir = tmp_path
api.version_path = tmp_path / "bin/legendary.version"
bundled = tmp_path / "bin/legendary"
bundled.parent.mkdir()
bundled.touch(mode=0o755)
api.version_path.write_text("v1", encoding="utf-8")
monkeypatch.setattr(api, "_get_latest_legendary_release", lambda: ({}, "v1"))
assert api.update_legendary() == str(bundled)
def test_normalize_game_uses_epic_metadata() -> None:
game = EGSAPI._normalize_game({
"app_name": "TestGame",
"app_title": "Test Game",
"metadata": {
"description": "Description",
"namespace": "test-namespace",
"keyImages": [{"type": "DieselGameBoxTall", "url": "cover.jpg"}],
},
})
assert game == {
"app_id": "TestGame",
"title": "Test Game",
"description": "Description",
"cover": "cover.jpg",
"namespace": "test-namespace",
"description_locale": "",
"description_preference": "",
"description_cache_version": 0,
}
def test_store_description_uses_epic_product_page(monkeypatch: MonkeyPatch) -> None:
api = EGSAPI()
graphql_response = SimpleNamespace(
raise_for_status=lambda: None,
json=lambda: {
"data": {"Catalog": {"catalogNs": {
"mappings": [{"pageSlug": "control"}],
}}},
},
)
store_response = SimpleNamespace(
status_code=200,
raise_for_status=lambda: None,
json=lambda: {"pages": [{
"type": "productHome",
"data": {"about": {"description": "Real description"}},
}]},
)
monkeypatch.setattr(
"portprotonqt.egs_api.requests.post",
lambda *args, **kwargs: graphql_response,
)
monkeypatch.setattr(
"portprotonqt.egs_api.requests.get",
lambda *args, **kwargs: store_response,
)
monkeypatch.setattr("portprotonqt.localization.get_metadata_language", lambda: "en")
description = api._get_store_description({
"app_id": "Calluna", "title": "Control", "namespace": "calluna",
})
assert description == ("Real description", "en")
def test_store_description_prefers_localized_short_text() -> None:
data = {"pages": [{
"type": "productHome",
"data": {"about": {
"shortDescription": "Краткое описание",
"description": "Первый абзац\n\nВторой абзац",
}},
}]}
assert EGSAPI._parse_store_description(data) == "Краткое описание"
def test_store_description_limits_fallback_to_first_paragraph() -> None:
data = {"pages": [{
"type": "productHome",
"data": {"about": {"description": "Первый абзац\n\nВторой абзац"}},
}]}
assert EGSAPI._parse_store_description(data) == "Первый абзац"
def test_store_description_handles_missing_product_pages(
monkeypatch: MonkeyPatch,
) -> None:
api = EGSAPI()
response = SimpleNamespace(
status_code=200,
raise_for_status=lambda: None,
json=lambda: {"pages": None},
)
monkeypatch.setattr(api, "_get_product_slug", lambda _namespace, _title: "game")
monkeypatch.setattr("portprotonqt.egs_api.requests.get", lambda *args, **kwargs: response)
description = api._get_store_description({
"app_id": "Game", "title": "Game", "namespace": "game",
})
assert description == ("", "")
def test_content_languages_follow_epic_locale_conventions(
monkeypatch: MonkeyPatch,
) -> None:
monkeypatch.setattr("portprotonqt.localization.get_metadata_language", lambda: "pt")
monkeypatch.setattr("portprotonqt.localization.get_system_locale", lambda: "pt_BR")
assert get_store_content_languages() == ("pt-BR", "pt", "en-US")
monkeypatch.setattr("portprotonqt.localization.get_metadata_language", lambda: "zh")
monkeypatch.setattr("portprotonqt.localization.get_system_locale", lambda: "zh_Hant_TW")
assert get_store_content_languages() == ("zh-TW", "zh", "en-US")
def test_extract_auth_code_accepts_legendary_response() -> None:
response = '{"authorizationCode":"epic-authorization-code"}'
assert EGSAPI.extract_auth_code(response) == "epic-authorization-code"
assert EGSAPI.extract_auth_code("direct-code") == "direct-code"
assert EGSAPI.extract_auth_code("https://example.com") == ""
def test_get_launch_target_uses_legendary_install_record(tmp_path: Path) -> None:
api = EGSAPI()
api.config_dir = tmp_path / "legendary"
game_dir = tmp_path / "game"
executable = game_dir / "Binaries/Game.exe"
executable.parent.mkdir(parents=True)
executable.touch()
api.config_dir.mkdir()
(api.config_dir / "installed.json").write_bytes(orjson.dumps({
"TestGame": {
"install_path": str(game_dir),
"executable": "Binaries\\Game.exe",
},
}))
assert api.get_launch_target("TestGame") == str(executable)
def test_parse_download_sizes_reads_legendary_info() -> None:
output = orjson.dumps({
"manifest": {"download_size": 1024},
"install": {"install_size": 2048},
})
assert EGSAPI.parse_download_sizes(output) == (1024, 2048)
def test_parse_download_sizes_uses_manifest_disk_size() -> None:
output = orjson.dumps({"manifest": {"download_size": 1024, "disk_size": 4096}})
assert EGSAPI.parse_download_sizes(output) == (1024, 4096)
def test_zero_cached_install_size_is_refetched(tmp_path: Path) -> None:
api = EGSAPI()
api.sizes_path = tmp_path / "sizes.json"
api.sizes_path.write_bytes(orjson.dumps({"Game": [1024, 0]}))
assert api.get_cached_download_sizes("Game") is None
def test_authenticate_passes_epic_code(tmp_path: Path, monkeypatch) -> None:
api = EGSAPI()
api.config_dir = tmp_path / "legendary"
api.user_path = api.config_dir / "user.json"
monkeypatch.setattr(api, "ensure_legendary", lambda: "/bin/legendary")
monkeypatch.setattr(api, "build_command", lambda args: ["legendary", *args])
def run(command: list[str], **_kwargs: Any) -> SimpleNamespace:
api.config_dir.mkdir(parents=True)
api.user_path.write_text("{}", encoding="utf-8")
assert command == ["legendary", "auth", "--code", "epic-code"]
return SimpleNamespace(returncode=0, stderr=b"")
monkeypatch.setattr("portprotonqt.egs_api.subprocess.run", run)
assert api.authenticate("epic-code") == (True, "")
+248 -15
View File
@@ -3,6 +3,7 @@
import shlex
from collections.abc import Callable
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timedelta
from pathlib import Path
from queue import Queue
import threading
@@ -10,7 +11,7 @@ from types import SimpleNamespace
from typing import Any, cast
from unittest.mock import MagicMock
from pytest import MonkeyPatch
from pytest import MonkeyPatch, mark
from PySide6.QtCore import QObject, Qt
from PySide6.QtWidgets import QApplication, QComboBox, QGridLayout, QWidget
@@ -18,16 +19,19 @@ from portprotonqt.animations.library_controls import _animation_duration
from portprotonqt.animations.game_card import GameCardAnimations
from portprotonqt.config import game_config
from portprotonqt.detail_pages import DetailPageManager
from portprotonqt.egs_api import EGSAPI
from portprotonqt.tabs.download_tab import MainWindowDownloadTabMixin
from portprotonqt.game_card import GameCard, SourceCorner
from portprotonqt.game_library_manager import GameLibraryManager
from portprotonqt.gog_api import GOGAPI
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.download_tab as download_tab_module
import portprotonqt.tabs.library_tab as library_tab_module
import portprotonqt.tabs.system_tab as system_tab_module
import portprotonqt.steam_api.api as steam_api_module
import portprotonqt.main_window as main_window_module
from portprotonqt.tabs import (
MainWindowAutoInstallTabMixin,
MainWindowLibraryTabMixin,
@@ -37,7 +41,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.download_tab import MainWindowDownloadTabMixin 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
@@ -562,7 +566,7 @@ def test_gog_account_state_detects_saved_auth(
gogAccountStatus=Control(),
gogLoginButton=Control(),
)
monkeypatch.setattr(gog_tab_module, "_", lambda text: text)
monkeypatch.setattr(download_tab_module, "_", lambda text: text)
GOGMixin._update_gog_account_state(cast(Any, window))
@@ -598,7 +602,7 @@ def test_gog_account_action_logs_out_and_reloads_library(
window._update_gog_account_state = lambda: GOGMixin._update_gog_account_state(
cast(Any, window)
)
monkeypatch.setattr(gog_tab_module, "_", lambda text: text)
monkeypatch.setattr(download_tab_module, "_", lambda text: text)
GOGMixin._handle_gog_account_action(cast(Any, window))
@@ -750,7 +754,7 @@ def test_gog_library_refresh_failure_reloads_cached_games(
gogAccountStatus=SimpleNamespace(setText=lambda text: calls.append(text)),
loadGames=lambda **kwargs: calls.append(kwargs),
)
monkeypatch.setattr(gog_tab_module, "_", lambda text: text)
monkeypatch.setattr(download_tab_module, "_", lambda text: text)
GOGMixin._on_gog_library_failed(cast(Any, window), "network error")
@@ -768,7 +772,7 @@ def test_gog_login_shows_cached_games_before_refresh(monkeypatch: MonkeyPatch) -
loadGames=lambda **kwargs: calls.append(("load", kwargs)),
_refresh_gog_library=lambda: calls.append("refresh"),
)
monkeypatch.setattr(gog_tab_module, "_", lambda text: text)
monkeypatch.setattr(download_tab_module, "_", lambda text: text)
GOGMixin._on_gog_authenticated(cast(Any, window), True, "")
@@ -815,7 +819,7 @@ def test_gog_support_uses_regular_launch_output_monitor(
_set_running_button_stop=lambda: None,
checkTargetExe=lambda: None,
)
monkeypatch.setattr(gog_tab_module, "QTimer", FakeTimer)
monkeypatch.setattr(download_tab_module, "QTimer", FakeTimer)
GOGMixin._track_gog_support_process(
cast(Any, window), "123", cast(Any, process)
@@ -909,7 +913,7 @@ def test_install_gog_game_uses_support_path(
explorer.file_signal.file_selected.connect.side_effect = (
lambda callback: callback(str(selected_path))
)
monkeypatch.setattr(gog_tab_module, "FileExplorer", lambda *_args, **_kwargs: explorer)
monkeypatch.setattr(download_tab_module, "FileExplorer", lambda *_args, **_kwargs: explorer)
api = SimpleNamespace(
config_dir=tmp_path / "gogdl",
get_install_path=lambda _app_id, _title: selected_path,
@@ -930,11 +934,43 @@ def test_install_gog_game_uses_support_path(
]
def test_install_egs_game_selects_path_and_opens_download(
tmp_path: Path, monkeypatch: MonkeyPatch
) -> None:
selected_path = tmp_path / "Epic"
started: list[tuple] = []
explorer = MagicMock()
explorer.file_signal.file_selected.connect.side_effect = (
lambda callback: callback(str(selected_path))
)
monkeypatch.setattr(download_tab_module, "FileExplorer", lambda *_args, **_kwargs: explorer)
game = {"app_id": "doom64", "title": "DOOM 64", "cover": "cover"}
api = SimpleNamespace(
games_dir=tmp_path / "Games", load_library=lambda: [game],
build_command=lambda arguments: ["legendary", *arguments],
)
window = SimpleNamespace(
gog_process=None, egs_process=None, theme=object(), egs_api=api,
_start_egs_download=lambda *arguments: started.append(arguments),
egsAccountStatus=SimpleNamespace(setText=MagicMock()),
)
GOGMixin._install_egs_download(cast(Any, window), "doom64")
assert started[0] == (
game, selected_path,
[
"legendary", "install", "doom64", "--base-path", str(selected_path),
"--platform", "Windows", "--skip-sdl", "--skip-dlcs", "-y",
],
)
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,
state=lambda: download_tab_module.QProcess.ProcessState.Running,
)
window = SimpleNamespace(
gog_process=process,
@@ -944,7 +980,7 @@ def test_cancel_gog_download_terminates_then_kills(monkeypatch: MonkeyPatch) ->
cast(Any, window), active
),
)
monkeypatch.setattr(gog_tab_module.QTimer, "singleShot", lambda _delay, callback: callback())
monkeypatch.setattr(download_tab_module.QTimer, "singleShot", lambda _delay, callback: callback())
GOGMixin._cancel_gog_download(cast(Any, window))
@@ -986,7 +1022,7 @@ def test_import_gog_game_saves_selected_installation(
gogAccountStatus=SimpleNamespace(setText=lambda _text: None),
loadGames=lambda **_kwargs: None,
)
monkeypatch.setattr(gog_tab_module, "FileExplorer", Explorer)
monkeypatch.setattr(download_tab_module, "FileExplorer", Explorer)
GOGMixin._import_gog_game(cast(Any, window), {"app_id": "123", "title": "Game"})
@@ -1887,7 +1923,16 @@ def test_installed_filter_disabled_includes_uninstalled_gog_games(
assert results[0][0][5] == "gog://install/uninstalled"
def test_gog_metadata_search_ignores_uri_components(monkeypatch: MonkeyPatch) -> None:
@mark.parametrize(
("game_name", "launch_uri"),
(
("Unknown GOG Game", "gog://launch/123"),
("Unknown Epic Game", "egs://launch/Fortnite"),
),
)
def test_store_metadata_search_ignores_uri_components(
monkeypatch: MonkeyPatch, game_name: str, launch_uri: str
) -> None:
from portprotonqt.steam_api import get_steam_game_info_async
searched_candidates = []
@@ -1916,13 +1961,201 @@ def test_gog_metadata_search_ignores_uri_components(monkeypatch: MonkeyPatch) ->
)
results = []
get_steam_game_info_async("Unknown GOG Game", "gog://launch/123", results.append)
get_steam_game_info_async(game_name, launch_uri, results.append)
assert searched_candidates == ["Unknown GOG Game"]
assert searched_candidates == [game_name]
assert len(results) == 1
fetch_sgdb_cover.assert_not_called()
def test_egs_refresh_reopens_current_detail_with_new_description() -> None:
source_data = {
"appid": "AmongUs",
"game_source": "egs",
"description": "Old description",
"cover_path": "old.jpg",
}
manager = SimpleNamespace(
_current_detail_source=("game", source_data),
_detail_page_active=True,
_reopen_current_detail_page=MagicMock(),
)
window = cast(MainWindowDownloadTabMixin, SimpleNamespace(
detail_page_manager=manager,
_update_egs_account_state=MagicMock(),
loadGames=MagicMock(),
))
MainWindowDownloadTabMixin._on_egs_library_loaded(window, [{
"app_id": "AmongUs",
"description": "Новое описание",
"cover": "new.jpg",
}])
assert source_data["description"] == "Новое описание"
assert source_data["cover_path"] == "new.jpg"
manager._reopen_current_detail_page.assert_called_once_with()
window.loadGames.assert_called_once_with(force_load=True)
def test_egs_library_uses_steam_description_when_epic_has_title_only(
monkeypatch: MonkeyPatch,
) -> None:
game = {"app_id": "WitchIt", "title": "Witch It", "description": "Witch It"}
api = SimpleNamespace(
load_library=lambda: [game],
is_game_installed=lambda _app_id: False,
)
results = []
window = cast(MainWindow, SimpleNamespace(
egs_api=api, games=[],
))
monkeypatch.setattr(game_config, "get_only_installed", lambda: False)
monkeypatch.setattr(
"portprotonqt.main_window.get_steam_game_info_async",
lambda _name, _uri, callback: callback({
"description": "Witch It — игра в прятки по сети.",
}),
)
MainWindow._load_egs_games_async(window, results.append)
assert results[0][0][1] == "Witch It — игра в прятки по сети."
def test_egs_maintenance_uses_legendary_commands() -> None:
start_operation = MagicMock()
window = cast(MainWindow, SimpleNamespace(_start_egs_operation=start_operation))
MainWindow._repair_egs_game(window, "Game")
MainWindow._update_egs_game(window, "Game")
MainWindow._delete_egs_game(window, "Game")
calls = start_operation.call_args_list
assert calls[0].args[1] == ["repair", "Game", "--skip-sdl", "-y"]
assert calls[1].args[1] == [
"update", "Game", "--platform", "Windows", "--skip-sdl", "-y",
]
assert calls[2].args[1] == ["uninstall", "Game", "-y"]
def test_egs_operation_is_sent_to_visible_downloads() -> None:
game = {"app_id": "Game", "title": "Epic Game", "cover": "cover"}
visible_operation = MagicMock()
api = SimpleNamespace(
build_command=lambda arguments: ["legendary", *arguments],
load_library=lambda: [game],
)
window = cast(MainWindow, SimpleNamespace(
egs_process=None, egs_api=api,
_start_egs_visible_operation=visible_operation,
))
MainWindow._start_egs_operation(window, "Game", ["repair", "Game"], "Repair")
visible_operation.assert_called_once_with(
game, ["legendary", "repair", "Game"], "Repair"
)
def test_detached_store_game_keeps_running_state(monkeypatch: MonkeyPatch) -> None:
dead_launcher = SimpleNamespace(poll=lambda: 0)
game_process = SimpleNamespace(info={"name": "DOOM64_x64.exe"})
monkeypatch.setattr(
main_window_module.psutil, "process_iter", lambda attrs: [game_process]
)
window = cast(MainWindow, SimpleNamespace(
game_processes=[dead_launcher], target_exe="DOOM64_x64.exe",
game_start_time=datetime.now() - timedelta(minutes=1),
))
assert MainWindow._has_running_game_process(window)
def test_store_launch_grace_prevents_early_button_reset(
monkeypatch: MonkeyPatch,
) -> None:
monkeypatch.setattr(main_window_module.psutil, "process_iter", lambda attrs: [])
window = cast(MainWindow, SimpleNamespace(
game_processes=[], target_exe="DOOM64_x64.exe",
game_start_time=datetime.now(),
))
assert MainWindow._has_running_game_process(window)
def test_egs_verification_uses_total_progress_format() -> None:
import re
output = "Verification progress: 37/142 (26.4%) [132.7 MiB/s]"
match = re.findall(
r"Verification progress:\s*\d+/\d+\s+\(([\d.]+)%\)"
r"(?:\s+\[([\d.]+)\s+MiB/s\])?",
output,
)
assert match == [("26.4", "132.7")]
def test_egs_overlay_enable_uses_game_prefix(tmp_path: Path) -> None:
prefix_path = tmp_path / "data/prefixes/DEFAULT"
prefix_path.mkdir(parents=True)
(prefix_path / "user.reg").touch()
config_dir = tmp_path / "legendary"
config_dir.mkdir()
(config_dir / "overlay_install.json").touch()
api = SimpleNamespace(
config_dir=config_dir,
data_dir=tmp_path / "egs",
get_launch_target=MagicMock(return_value="/games/Game.exe"),
is_eos_overlay_enabled=EGSAPI.is_eos_overlay_enabled,
)
start_operation = MagicMock()
window = cast(MainWindow, SimpleNamespace(
egs_api=api,
portproton_location=str(tmp_path),
_start_egs_operation=start_operation,
))
MainWindow._enable_egs_overlay(window, "Game")
arguments = start_operation.call_args.args[1]
assert arguments == [
"eos-overlay", "enable", "--prefix", str(prefix_path),
]
def test_egs_overlay_disable_uses_game_prefix(tmp_path: Path) -> None:
prefix_path = tmp_path / "data/prefixes/DEFAULT"
prefix_path.mkdir(parents=True)
(prefix_path / "user.reg").write_text(
'[Software\\\\Epic Games\\\\EOS]\n"OverlayPath"="Z:/overlay"\n',
encoding="utf-8",
)
config_dir = tmp_path / "legendary"
config_dir.mkdir()
(config_dir / "overlay_install.json").touch()
api = SimpleNamespace(
config_dir=config_dir,
data_dir=tmp_path / "egs",
get_launch_target=MagicMock(return_value="/games/Game.exe"),
is_eos_overlay_enabled=EGSAPI.is_eos_overlay_enabled,
)
start_operation = MagicMock()
window = cast(MainWindow, SimpleNamespace(
egs_api=api,
portproton_location=str(tmp_path),
_start_egs_operation=start_operation,
))
MainWindow._enable_egs_overlay(window, "Game")
arguments = start_operation.call_args.args[1]
assert arguments == [
"eos-overlay", "disable", "--prefix", str(prefix_path),
]
def test_remove_empty_custom_data_dirs_keeps_non_empty_dirs(tmp_config_dir: Path) -> None:
custom_data_path = tmp_config_dir.parent / "data" / "PortProtonQt" / "custom_data"
(custom_data_path / "praest").mkdir(parents=True)