From 1f4f4093bdf194820d56b9c2c4ae59b0f505f606 Mon Sep 17 00:00:00 2001 From: Boris Yumankulov Date: Sun, 22 Jun 2025 10:51:06 +0500 Subject: [PATCH] feat(egs-api): Implement add_egs_to_steam to add EGS games to Steam via shortcuts.vdf Signed-off-by: Boris Yumankulov --- portprotonqt/egs_api.py | 242 +++++++++++++++++++++++++++++++++++++++- 1 file changed, 240 insertions(+), 2 deletions(-) diff --git a/portprotonqt/egs_api.py b/portprotonqt/egs_api.py index 488e695..c9960b9 100644 --- a/portprotonqt/egs_api.py +++ b/portprotonqt/egs_api.py @@ -14,11 +14,18 @@ from portprotonqt.logger import get_logger from portprotonqt.image_utils import load_pixmap_async from portprotonqt.time_utils import parse_playtime_file, format_playtime, get_last_launch, get_last_launch_timestamp from portprotonqt.config_utils import get_portproton_location -from portprotonqt.steam_api import get_weanticheatyet_status_async, get_steam_apps_and_index_async, get_protondb_tier_async, search_app - +from portprotonqt.steam_api import ( + get_weanticheatyet_status_async, get_steam_apps_and_index_async, get_protondb_tier_async, + search_app, get_steam_home, get_last_steam_user, convert_steam_id, generate_thumbnail +) +import vdf +import shutil +import zlib +from portprotonqt.downloader import Downloader from PySide6.QtGui import QPixmap logger = get_logger(__name__) +downloader = Downloader() def get_egs_executable(app_name: str, legendary_config_path: str) -> str | None: """Получает путь к исполняемому файлу EGS-игры из installed.json с использованием orjson.""" @@ -52,6 +59,237 @@ def get_cache_dir() -> Path: cache_dir.mkdir(parents=True, exist_ok=True) return cache_dir +def add_egs_to_steam(app_name: str, game_title: str, legendary_path: str, callback: Callable[[tuple[bool, str]], None]) -> None: + """ + Asynchronously adds an EGS game to Steam via shortcuts.vdf with PortProton tag. + Creates a launch script using legendary CLI with --no-wine and PortProton wrapper. + Wrapper is flatpak run if portproton_location is None or contains .var, otherwise start.sh. + Downloads Steam Grid covers if the game exists in Steam, and generates a thumbnail. + Calls the callback with (success, message). + """ + if not app_name or not app_name.strip() or not game_title or not game_title.strip(): + logger.error("Invalid app_name or game_title: empty or whitespace") + callback((False, "Game name or app name is empty or invalid")) + return + + if not os.path.exists(legendary_path): + logger.error(f"Legendary executable not found: {legendary_path}") + callback((False, f"Legendary executable not found: {legendary_path}")) + return + + portproton_dir = get_portproton_location() + if not portproton_dir: + logger.error("PortProton directory not found") + callback((False, "PortProton directory not found")) + return + + # Determine wrapper + wrapper = "flatpak run ru.linux_gaming.PortProton" + start_sh_path = os.path.join(portproton_dir, "data", "scripts", "start.sh") + if portproton_dir is not None and ".var" not in portproton_dir: + wrapper = start_sh_path + if not os.path.exists(start_sh_path): + logger.error(f"start.sh not found at {start_sh_path}") + callback((False, f"start.sh not found at {start_sh_path}")) + return + + # Create launch script + steam_scripts_dir = os.path.join(portproton_dir, "steam_scripts") + os.makedirs(steam_scripts_dir, exist_ok=True) + safe_game_name = re.sub(r'[<>:"/\\|?*]', '_', game_title.strip()) + script_path = os.path.join(steam_scripts_dir, f"{safe_game_name}_egs.sh") + legendary_config_path = os.path.dirname(legendary_path) + + script_content = f"""#!/usr/bin/env bash +export LD_PRELOAD= +export LEGENDARY_CONFIG_PATH="{legendary_config_path}" +"{legendary_path}" launch {app_name} --no-wine --wrapper "env START_FROM_STEAM=1 {wrapper}" "$@" +""" + try: + with open(script_path, "w", encoding="utf-8") as f: + f.write(script_content) + os.chmod(script_path, 0o755) + logger.info(f"Created launch script for EGS game: {script_path}") + except Exception as e: + logger.error(f"Failed to create launch script {script_path}: {e}") + callback((False, f"Failed to create launch script: {e}")) + return + + # Generate thumbnail + generated_icon_path = os.path.join(portproton_dir, "data", "img", f"{safe_game_name}_egs.png") + try: + img_dir = os.path.join(portproton_dir, "data", "img") + os.makedirs(img_dir, exist_ok=True) + game_exe = get_egs_executable(app_name, legendary_config_path) + if not game_exe or not os.path.exists(game_exe): + logger.warning(f"Executable not found for {app_name}, skipping thumbnail generation") + icon_path = "" + elif os.path.exists(generated_icon_path): + logger.info(f"Reusing existing thumbnail: {generated_icon_path}") + icon_path = generated_icon_path + else: + success = generate_thumbnail(game_exe, generated_icon_path, size=128, force_resize=True) + if not success or not os.path.exists(generated_icon_path): + logger.warning(f"generate_thumbnail failed for {game_exe}") + icon_path = "" + else: + logger.info(f"Generated thumbnail: {generated_icon_path}") + icon_path = generated_icon_path + except Exception as e: + logger.error(f"Error generating thumbnail for {app_name}: {e}") + icon_path = "" + + # Get Steam directories + steam_home = get_steam_home() + if not steam_home: + logger.error("Steam home directory not found") + callback((False, "Steam directory not found")) + return + + last_user = get_last_steam_user(steam_home) + if not last_user or 'SteamID' not in last_user: + logger.error("Failed to retrieve Steam user ID") + callback((False, "Failed to get Steam user ID")) + return + + userdata_dir = steam_home / "userdata" + user_id = last_user['SteamID'] + unsigned_id = convert_steam_id(user_id) + user_dir = userdata_dir / str(unsigned_id) + steam_shortcuts_path = user_dir / "config" / "shortcuts.vdf" + grid_dir = user_dir / "config" / "grid" + os.makedirs(grid_dir, exist_ok=True) + + # Backup shortcuts.vdf + backup_path = f"{steam_shortcuts_path}.backup" + if os.path.exists(steam_shortcuts_path): + try: + shutil.copy2(steam_shortcuts_path, backup_path) + logger.info(f"Created backup of shortcuts.vdf at {backup_path}") + except Exception as e: + logger.error(f"Failed to create backup of shortcuts.vdf: {e}") + callback((False, f"Failed to create backup of shortcuts.vdf: {e}")) + return + + # Generate unique appid + unique_string = f"{script_path}{game_title}" + baseid = zlib.crc32(unique_string.encode('utf-8')) & 0xffffffff + appid = baseid | 0x80000000 + if appid > 0x7FFFFFFF: + aidvdf = appid - 0x100000000 + else: + aidvdf = appid + + steam_appid = None + downloaded_count = 0 + total_covers = 4 + download_lock = threading.Lock() + + def on_cover_download(cover_file: str, cover_type: str): + nonlocal downloaded_count + try: + if cover_file and os.path.exists(cover_file): + logger.info(f"Downloaded cover {cover_type} to {cover_file}") + else: + logger.warning(f"Failed to download cover {cover_type} for appid {steam_appid}") + except Exception as e: + logger.error(f"Error processing cover {cover_type} for appid {steam_appid}: {e}") + with download_lock: + downloaded_count += 1 + if downloaded_count == total_covers: + finalize_shortcut() + + def finalize_shortcut(): + tags_dict = {'0': 'PortProton'} + shortcut = { + "appid": aidvdf, + "AppName": game_title, + "Exe": f'"{script_path}"', + "StartDir": f'"{os.path.dirname(script_path)}"', + "icon": icon_path, + "LaunchOptions": "", + "IsHidden": 0, + "AllowDesktopConfig": 1, + "AllowOverlay": 1, + "openvr": 0, + "Devkit": 0, + "DevkitGameID": "", + "LastPlayTime": 0, + "tags": tags_dict + } + logger.info(f"Shortcut entry for EGS game: {shortcut}") + + try: + if not os.path.exists(steam_shortcuts_path): + os.makedirs(os.path.dirname(steam_shortcuts_path), exist_ok=True) + open(steam_shortcuts_path, 'wb').close() + + try: + if os.path.getsize(steam_shortcuts_path) > 0: + with open(steam_shortcuts_path, 'rb') as f: + shortcuts_data = vdf.binary_load(f) + else: + shortcuts_data = {"shortcuts": {}} + except Exception as load_err: + logger.warning(f"Failed to load shortcuts.vdf, starting fresh: {load_err}") + shortcuts_data = {"shortcuts": {}} + + shortcuts = shortcuts_data.get("shortcuts", {}) + for _key, entry in shortcuts.items(): + if entry.get("AppName") == game_title and entry.get("Exe") == f'"{script_path}"': + logger.info(f"EGS game '{game_title}' already exists in Steam shortcuts") + callback((False, f"Game '{game_title}' already exists in Steam")) + return + + new_index = str(len(shortcuts)) + shortcuts[new_index] = shortcut + + with open(steam_shortcuts_path, 'wb') as f: + vdf.binary_dump({"shortcuts": shortcuts}, f) + except Exception as e: + logger.error(f"Failed to update shortcuts.vdf: {e}") + if os.path.exists(backup_path): + try: + shutil.copy2(backup_path, steam_shortcuts_path) + logger.info("Restored shortcuts.vdf from backup") + except Exception as restore_err: + logger.error(f"Failed to restore shortcuts.vdf: {restore_err}") + callback((False, f"Failed to update shortcuts.vdf: {e}")) + return + + logger.info(f"EGS game '{game_title}' added to Steam") + callback((True, f"Game '{game_title}' added to Steam with covers")) + + def on_steam_apps(steam_data: tuple[list, dict]): + nonlocal steam_appid + steam_apps, steam_apps_index = steam_data + matching_app = search_app(game_title, steam_apps_index) + steam_appid = matching_app.get("appid") if matching_app else None + + if not steam_appid: + logger.info(f"No Steam appid found for EGS game {game_title}, skipping cover download") + finalize_shortcut() + return + + cover_types = [ + (".jpg", "header.jpg"), + ("p.jpg", "library_600x900_2x.jpg"), + ("_hero.jpg", "library_hero.jpg"), + ("_logo.png", "logo.png") + ] + + for suffix, cover_type in cover_types: + cover_file = os.path.join(grid_dir, f"{appid}{suffix}") + cover_url = f"https://cdn.cloudflare.steamstatic.com/steam/apps/{steam_appid}/{cover_type}" + downloader.download_async( + cover_url, + cover_file, + timeout=5, + callback=lambda result, cfile=cover_file, ctype=cover_type: on_cover_download(cfile, ctype) + ) + + get_steam_apps_and_index_async(on_steam_apps) + def get_egs_game_description_async( app_name: str, callback: Callable[[str], None],