Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions .github/workflows/lint-ruff.yml
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
name: Lint with Ruff
on:
push:
branches: [ '*' ]
pull_request:
branches: [ '*' ]

jobs:
lint:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v6
- uses: astral-sh/ruff-action@v3
10 changes: 6 additions & 4 deletions legendary/api/egs.py
Original file line number Diff line number Diff line change
@@ -1,16 +1,18 @@
# !/usr/bin/env python
# coding: utf-8

import logging
import urllib.parse

import requests
import requests.adapters
import logging

from requests.auth import HTTPBasicAuth

from legendary.models.exceptions import InvalidCredentialsError
from legendary.models.gql import *
from legendary.models.gql import (
uplay_claim_query,
uplay_codes_query,
uplay_redeem_query,
)


class EPCAPI:
Expand Down
4 changes: 2 additions & 2 deletions legendary/api/lgd.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,10 @@
# !/usr/bin/env python
# coding: utf-8

import logging
from platform import system

import requests

from platform import system
from legendary import __version__


Expand Down
121 changes: 70 additions & 51 deletions legendary/cli.py

Large diffs are not rendered by default.

111 changes: 68 additions & 43 deletions legendary/core.py
Original file line number Diff line number Diff line change
@@ -1,46 +1,73 @@
# coding: utf-8

import contextlib
import json
import logging
import os
import shlex
import shutil

from base64 import b64decode
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor
from datetime import timezone
from datetime import datetime, timezone
from hashlib import sha1
from locale import getdefaultlocale
from multiprocessing import Queue
from platform import system
from requests import session
from requests.exceptions import HTTPError, ConnectionError
from sys import platform as sys_platform
from urllib.parse import parse_qsl, urlencode, urlparse
from uuid import uuid4
from urllib.parse import urlencode, parse_qsl, urlparse

from requests import session
from requests.exceptions import ConnectionError, HTTPError

from legendary import __version__
from legendary.api.egs import EPCAPI
from legendary.api.lgd import LGDAPI
from legendary.downloader.mp.manager import DLManager
from legendary.lfs.crossover import (
mac_find_crossover_apps,
mac_get_bottle_path,
mac_get_crossover_version,
mac_is_valid_bottle,
)
from legendary.lfs.egl import EPCLFS
from legendary.lfs.eos import EOSOverlayApp, query_registry_entries
from legendary.lfs.lgndry import LGDLFS
from legendary.lfs.utils import clean_filename, delete_folder, delete_filelist, get_dir_size
from legendary.lfs.utils import (
clean_filename,
delete_filelist,
delete_folder,
get_dir_size,
)
from legendary.lfs.wine_helpers import (
case_insensitive_path_search,
get_shell_folders,
read_registry,
)
from legendary.models.chunk import Chunk
from legendary.models.downloading import AnalysisResult, ConditionCheckResult
from legendary.models.egl import EGLManifest
from legendary.models.exceptions import *
from legendary.models.game import *
from legendary.models.exceptions import InvalidCredentialsError
from legendary.models.game import (
Game,
GameAsset,
InstalledGame,
LaunchParameters,
SaveGameFile,
SaveGameStatus,
Sidecar,
)
from legendary.models.json_manifest import JSONManifest
from legendary.models.manifest import Manifest, ManifestMeta
from legendary.models.chunk import Chunk
from legendary.lfs.crossover import *
from legendary.utils.egl_crypt import decrypt_epic_data
from legendary.utils.env import is_windows_mac_or_pyi
from legendary.lfs.eos import EOSOverlayApp, query_registry_entries
from legendary.utils.game_workarounds import is_opt_enabled, update_workarounds, get_exe_override
from legendary.utils.game_workarounds import (
get_exe_override,
is_opt_enabled,
update_workarounds,
)
from legendary.utils.savegame_helper import SaveGameHelper
from legendary.utils.selective_dl import games as sdl_games
from legendary.lfs.wine_helpers import read_registry, get_shell_folders, case_insensitive_path_search


# ToDo: instead of true/false return values for success/failure actually raise an exception that the CLI/GUI
# can handle to give the user more details. (Not required yet since there's no GUI so log output is fine)
Expand Down Expand Up @@ -295,7 +322,7 @@ def apply_lgd_config(self, version_info=None):
update_workarounds(game_overrides)
if sdl_config := game_overrides.get('sdl_config'):
# add placeholder for games to fetch from API that aren't hardcoded
for app_name in sdl_config.keys():
for app_name in sdl_config:
if app_name not in sdl_games:
sdl_games[app_name] = None
if lgd_config := version_info.get('legendary_config'):
Expand Down Expand Up @@ -344,7 +371,7 @@ def update_aliases(self, force=False):
if _aliases_enabled and (force or not self.lgd.aliases):
self.lgd.generate_aliases()

def get_assets(self, update_assets=False, platform='Windows') -> List[GameAsset]:
def get_assets(self, update_assets=False, platform='Windows') -> list[GameAsset]:
# do not save and always fetch list when platform is overridden
if not self.lgd.assets or update_assets or platform not in self.lgd.assets:
# if not logged in, return empty list
Expand Down Expand Up @@ -372,8 +399,8 @@ def get_asset(self, app_name, platform='Windows', update=False) -> GameAsset:

try:
return next(i for i in self.lgd.assets[platform] if i.app_name == app_name)
except StopIteration:
raise ValueError
except StopIteration as e:
raise ValueError from e

def asset_valid(self, app_name) -> bool:
# EGL sync is only supported for Windows titles so this is fine
Expand All @@ -395,11 +422,11 @@ def get_game(self, app_name, update_meta=False, platform='Windows') -> Game:
self.get_game_list(True, platform=platform)
return self.lgd.get_game_meta(app_name)

def get_game_list(self, update_assets=True, platform='Windows') -> List[Game]:
def get_game_list(self, update_assets=True, platform='Windows') -> list[Game]:
return self.get_game_and_dlc_list(update_assets=update_assets, platform=platform)[0]

def get_game_and_dlc_list(self, update_assets=True, platform='Windows',
force_refresh=False, skip_ue=True) -> (List[Game], Dict[str, List[Game]]):
force_refresh=False, skip_ue=True) -> tuple[list[Game], dict[str, list[Game]]]:
_ret = []
_dlc = defaultdict(list)
meta_updated = False
Expand Down Expand Up @@ -433,7 +460,7 @@ def get_game_and_dlc_list(self, update_assets=True, platform='Windows',
game = self.lgd.get_game_meta(app_name)
asset_updated = sidecar_updated = False
if game:
asset_updated = any(game.app_version(_p) != app_assets[_p].build_version for _p in app_assets.keys())
asset_updated = any(game.app_version(_p) != app_assets[_p].build_version for _p in app_assets)
# assuming sidecar data is the same for all platforms, just check the baseline (Windows) for updates.
sidecar_updated = (app_assets['Windows'].sidecar_rev > 0 and
(not game.sidecar or game.sidecar.rev != app_assets['Windows'].sidecar_rev))
Expand Down Expand Up @@ -467,10 +494,8 @@ def fetch_game_meta(args):
sidecar=sidecar)
self.lgd.set_game_meta(game.app_name, game)
games[app_name] = game
try:
with contextlib.suppress(KeyError):
still_needs_update.remove(app_name)
except KeyError:
pass

# setup and teardown of thread pool takes some time, so only do it when it makes sense.
still_needs_update = {e[0] for e in fetch_list}
Expand Down Expand Up @@ -525,7 +550,7 @@ def _prune_metadata(self):
self.lgd.delete_game_meta(app_name)

def get_non_asset_library_items(self, force_refresh=False,
skip_ue=True) -> (List[Game], Dict[str, List[Game]]):
skip_ue=True) -> tuple[list[Game], dict[str, list[Game]]]:
"""
Gets a list of Games without assets for installation, for instance Games delivered via
third-party stores that do not have assets for installation
Expand Down Expand Up @@ -581,20 +606,20 @@ def get_dlc_for_game(self, app_name, platform='Windows'):
def get_installed_platforms(self):
return {i.platform for i in self._get_installed_list(False)}

def get_installed_list(self, include_dlc=False) -> List[InstalledGame]:
def get_installed_list(self, include_dlc=False) -> list[InstalledGame]:
if self.egl_sync_enabled:
self.log.debug('Running EGL sync...')
self.egl_sync()

return self._get_installed_list(include_dlc)

def _get_installed_list(self, include_dlc=False) -> List[InstalledGame]:
def _get_installed_list(self, include_dlc=False) -> list[InstalledGame]:
if include_dlc:
return self.lgd.get_installed_list()
else:
return [g for g in self.lgd.get_installed_list() if not g.is_dlc]

def get_installed_dlc_list(self) -> List[InstalledGame]:
def get_installed_dlc_list(self) -> list[InstalledGame]:
return [g for g in self.lgd.get_installed_list() if g.is_dlc]

def get_installed_game(self, app_name, skip_sync=False) -> InstalledGame:
Expand Down Expand Up @@ -831,7 +856,7 @@ def get_origin_uri(self, app_name: str, offline: bool = False) -> str:
return f'link2ea://launchgame/{app_name}?{urlencode(parameters)}'

def get_save_games(self, app_name: str = ''):
savegames = self.egs.get_user_cloud_saves(app_name, manifests=not not app_name)
savegames = self.egs.get_user_cloud_saves(app_name, manifests=bool(app_name))
_saves = []
for fname, f in savegames['files'].items():
if '.manifest' not in fname:
Expand Down Expand Up @@ -966,7 +991,7 @@ def get_save_path(self, app_name, platform='Windows'):

return absolute_path

def check_savegame_state(self, path: str, save: SaveGameFile) -> (SaveGameStatus, (datetime, datetime)):
def check_savegame_state(self, path: str, save: SaveGameFile) -> tuple[SaveGameStatus, tuple[datetime | None, datetime | None]]:
latest = 0
for _dir, _, _files in os.walk(path):
for _file in _files:
Expand Down Expand Up @@ -1107,8 +1132,8 @@ def download_saves(self, app_name='', manifest_name='', save_dir='', clean_dir=F
f'"legendary clean-saves {app_name}" and try again.')
return
else:
self.log.error(f'No chunks were available, skipping. You can run "legendary clean-saves" '
f'to remove this broken save from your account.')
self.log.error('No chunks were available, skipping. You can run "legendary clean-saves" '
'to remove this broken save from your account.')
continue

for fm in m.file_manifest_list.elements:
Expand Down Expand Up @@ -1183,14 +1208,14 @@ def clean_saves(self, app_name='', delete_incomplete=False):
deletion_list.append(fname)
continue
elif 0 < missing_chunks < total_chunks:
self.log.error(f'Some chunk(s) missing, optionally run "legendary download-saves" to obtain a backup '
f'of the corrupted save, then re-run this command with "--delete-incomplete" to remove '
f'it from the cloud save service.')
self.log.error('Some chunk(s) missing, optionally run "legendary download-saves" to obtain a backup '
'of the corrupted save, then re-run this command with "--delete-incomplete" to remove '
'it from the cloud save service.')

used_chunks |= chunk_fnames

# check for orphaned chunks (not used in any manifests)
for fname, f in files.items():
for fname in files:
if fname in used_chunks or '.manifest' in fname:
continue
# skip chunks where orphan status could not be reliably determined
Expand Down Expand Up @@ -1353,7 +1378,7 @@ def prepare_download(self, game: Game, base_game: Game = None, base_path: str =
game.base_urls = _base_urls

if not old_bytes:
self.log.error(f'Could not load old manifest, patching will not work!')
self.log.error('Could not load old manifest, patching will not work!')
else:
old_manifest = self.load_manifest(old_bytes)

Expand Down Expand Up @@ -1408,7 +1433,7 @@ def prepare_download(self, game: Game, base_game: Game = None, base_path: str =
f'"{new_manifest.meta.build_id}"...')
new_manifest.apply_delta_manifest(delta_manifest)
else:
self.log.debug(f'No Delta manifest received from CDN.')
self.log.debug('No Delta manifest received from CDN.')

# reuse existing installation's directory
if igame := self.get_installed_game(base_game.app_name if base_game else game.app_name):
Expand Down Expand Up @@ -1734,7 +1759,7 @@ def uninstall_tag(self, installed_game: InstalledGame):
installed_game.install_path, filelist,
case_insensitive=installed_game.platform.startswith('Win')
):
self.log.warning(f'Deleting some deselected files failed, please check/remove manually.')
self.log.warning('Deleting some deselected files failed, please check/remove manually.')

def prereq_installed(self, app_name):
igame = self.lgd.get_installed_game(app_name)
Expand Down Expand Up @@ -1775,7 +1800,7 @@ def import_game(self, game: Game, app_path: str, egl_guid='', platform='Windows'
needs_verify = True

if not needs_verify:
self.log.debug(f'No in-progress installation found, assuming complete...')
self.log.debug('No in-progress installation found, assuming complete...')

manifest_secrets = dict()
if not manifest_data:
Expand Down Expand Up @@ -1922,8 +1947,8 @@ def egl_uninstall(self, igame: InstalledGame, delete_files=True):
except ValueError as e:
self.log.warning(f'Deleting EGL manifest failed: {e!r}')
except FileNotFoundError:
self.log.warning(f'EGL manifest was already deleted, in case you uninstalled the Epic Games Launcher'
f' please disable and unlink EGL Sync.')
self.log.warning('EGL manifest was already deleted, in case you uninstalled the Epic Games Launcher'
' please disable and unlink EGL Sync.')

if delete_files:
delete_folder(os.path.join(igame.install_path, '.egstore'))
Expand Down
Loading
Loading