diff --git a/legendary/api/egs.py b/legendary/api/egs.py index 8384ee1..38366e7 100644 --- a/legendary/api/egs.py +++ b/legendary/api/egs.py @@ -51,10 +51,7 @@ class EPCAPI: self.language_code = lc self.country_code = cc - if timeout > 0: - self.request_timeout = timeout - else: - self.request_timeout = None + self.request_timeout = timeout if timeout > 0 else None def get_auth_url(self): login_url = 'https://www.epicgames.com/id/login?redirectUrl=' @@ -236,10 +233,8 @@ class EPCAPI: return records def get_user_cloud_saves(self, app_name='', manifests=False, filenames=None): - if app_name and manifests: - app_name += '/manifests/' - elif app_name: - app_name += '/' + if app_name: + app_name += '/manifests/' if manifests else '/' user_id = self.user.get('account_id') diff --git a/legendary/cli.py b/legendary/cli.py index 0730bb4..149a0af 100644 --- a/legendary/cli.py +++ b/legendary/cli.py @@ -247,7 +247,7 @@ class LegendaryCLI: elif _store: print(f' ! This game has to be installed through a third-party store ({_store}, not supported)') else: - print(f' ! No version information (unknown cause)') + print(' ! No version information (unknown cause)') # Games that have assets, but only require a one-time activation before they can be independently installed # via a third-party platform (e.g. Uplay) if game.partner_link_type: @@ -380,15 +380,16 @@ class LegendaryCLI: writer.writerow(['path', 'hash', 'size', 'install_tags']) writer.writerows((fm.filename, fm.hash.hex(), fm.file_size, '|'.join(fm.install_tags)) for fm in files) elif args.json: - _files = [] - for fm in files: - _files.append(dict( + _files = [ + dict( filename=fm.filename, sha_hash=fm.hash.hex(), install_tags=fm.install_tags, file_size=fm.file_size, - flags=fm.flags, - )) + flags=fm.flags + ) + for fm in files + ] return self._print_json(_files, args.pretty_json) else: install_tags = set() @@ -430,7 +431,7 @@ class LegendaryCLI: if not self.core.login(): logger.error('Login failed! Cannot continue with download process.') exit(1) - logger.info(f'Cleaning saves...') + logger.info('Cleaning saves...') self.core.clean_saves(self._resolve_aliases(args.app_name), args.delete_incomplete) def sync_saves(self, args): @@ -449,10 +450,9 @@ class LegendaryCLI: # check available saves saves = self.core.get_save_games() - latest_save = dict() - - for save in sorted(saves, key=lambda a: a.datetime): - latest_save[save.app_name] = save + latest_save = { + save.app_name: save for save in sorted(saves, key=lambda a: a.datetime) + } logger.info(f'Got {len(latest_save)} remote save game(s)') @@ -475,7 +475,7 @@ class LegendaryCLI: # if there is no saved save path, try to get one if not igame.save_path: if args.yes: - logger.info(f'Save path for this title has not been set, skipping due to --yes') + logger.info('Save path for this title has not been set, skipping due to --yes') continue save_path = self.core.get_save_path(igame.app_name, platform=igame.platform) diff --git a/legendary/core.py b/legendary/core.py index c3404b6..6b0290f 100644 --- a/legendary/core.py +++ b/legendary/core.py @@ -85,7 +85,7 @@ class LegendaryCore: except Exception as e: self.log.warning(f'Getting locale failed: {e!r}, falling back to using en-US.') elif system() != 'Darwin': # macOS doesn't have a default locale we can query - self.log.warning(f'Could not determine locale, falling back to en-US') + self.log.warning('Could not determine locale, falling back to en-US') self.update_available = False self.force_show_update = False @@ -123,9 +123,9 @@ class LegendaryCore: if r.status_code == 200: return r.json()['code'] - else: - self.log.error(f'Getting exchange code failed: {r.json()}') - return '' + + self.log.error(f'Getting exchange code failed: {r.json()}') + return '' def auth_code(self, code) -> bool: """ @@ -274,10 +274,10 @@ class LegendaryCore: """Applies configuration options returned by update API""" if not version_info: version_info = self.lgd.get_cached_version()['data'] - # if cached data is invalid - if not version_info: - self.log.debug('No cached legendary config to apply.') - return + # if cached data is invalid + if not version_info: + self.log.debug('No cached legendary config to apply.') + return if 'egl_config' in version_info: self.egs.update_egs_params(version_info['egl_config']) @@ -342,10 +342,7 @@ class LegendaryCore: if not self.egs.user: return [] - if self.lgd.assets: - assets = self.lgd.assets.copy() - else: - assets = dict() + assets = self.lgd.assets.copy() if self.lgd.assets else dict() assets.update({ platform: [ @@ -579,9 +576,17 @@ class LegendaryCore: # get environment overrides from config env = dict() if 'default.env' in self.lgd.config: - env.update({k: v for k, v in self.lgd.config[f'default.env'].items() if v and not k.startswith(';')}) + env |= { + k: v + for k, v in self.lgd.config['default.env'].items() + if v and not k.startswith(';') + } if f'{app_name}.env' in self.lgd.config: - env.update({k: v for k, v in self.lgd.config[f'{app_name}.env'].items() if v and not k.startswith(';')}) + env |= { + k: v + for k, v in self.lgd.config[f'{app_name}.env'].items() + if v and not k.startswith(';') + } if disable_wine: return env @@ -719,10 +724,8 @@ class LegendaryCore: elif not install.can_run_offline: self.log.warning('Game is not approved for offline use and may not work correctly.') - user_name = self.lgd.userdata['displayName'] account_id = self.lgd.userdata['account_id'] - if user: - user_name = user + user_name = user or self.lgd.userdata['displayName'] params.egl_parameters.extend([ '-AUTH_LOGIN=unused', @@ -760,10 +763,7 @@ class LegendaryCore: return params def get_origin_uri(self, app_name: str, offline: bool = False) -> str: - if offline: - token = '0' - else: - token = self.egs.get_game_token()['code'] + token = '0' if offline else self.egs.get_game_token()['code'] user_name = self.lgd.userdata['displayName'] account_id = self.lgd.userdata['account_id'] @@ -819,18 +819,18 @@ class LegendaryCore: } if sys_platform == 'win32': - path_vars.update({ + path_vars |= { '{appdata}': os.path.expandvars('%LOCALAPPDATA%'), '{userdir}': os.path.expandvars('%userprofile%/documents'), '{userprofile}': os.path.expandvars('%userprofile%'), - '{usersavedgames}': os.path.expandvars('%userprofile%/Saved Games') - }) + '{usersavedgames}': os.path.expandvars('%userprofile%/Saved Games'), + } elif sys_platform == 'darwin' and platform == 'Mac': - path_vars.update({ + path_vars |= { '{appdata}': os.path.expanduser('~/Library/Application Support'), '{userdir}': os.path.expanduser('~/Documents'), - '{userlibrary}': os.path.expanduser('~/Library') - }) + '{userlibrary}': os.path.expanduser('~/Library'), + } else: wine_pfx = None # on mac CrossOver takes precedence so check for a bottle first @@ -868,10 +868,10 @@ class LegendaryCore: wine_pfx = mac_get_bottle_path(cx_bottle) if not wine_pfx: - proton_pfx = os.getenv('STEAM_COMPAT_DATA_PATH') - if proton_pfx: + if proton_pfx := os.getenv('STEAM_COMPAT_DATA_PATH'): wine_pfx = f'{proton_pfx}/pfx' - wine_pfx = os.getenv('WINEPREFIX', wine_pfx) + else: + wine_pfx = os.getenv('WINEPREFIX', wine_pfx) # if all else fails, use the WINE default if not wine_pfx: @@ -1111,7 +1111,7 @@ class LegendaryCore: missing_chunks += 1 if (0 < missing_chunks < total_chunks and delete_incomplete) or missing_chunks == total_chunks: - self.log.error(f'Chunk(s) missing, marking manifest for deletion.') + self.log.error('Chunk(s) missing, marking manifest for deletion.') deletion_list.append(fname) continue elif 0 < missing_chunks < total_chunks: @@ -1153,10 +1153,7 @@ class LegendaryCore: for ass in self.get_assets(True): if ass.app_name == app_name: - if ass.build_version != installed.version: - return False - else: - return True + return ass.build_version == installed.version # if we get here something is very wrong raise ValueError(f'Could not find {app_name} in asset list!') @@ -1167,10 +1164,10 @@ class LegendaryCore: return self._get_installed_game(app_name) is not None def is_dlc(self, app_name: str) -> bool: - meta = self.lgd.get_game_meta(app_name) - if not meta: + if meta := self.lgd.get_game_meta(app_name): + return meta.is_dlc + else: raise ValueError('Game unknown!') - return meta.is_dlc @staticmethod def load_manifest(data: bytes) -> Manifest: @@ -1252,10 +1249,7 @@ class LegendaryCore: return None r = self.egs.unauth_session.get(f'{base_url}/Deltas/{new_build_id}/{old_build_id}.delta') - if r.status_code == 200: - return r.content - else: - return None + return r.content if r.status_code == 200 else None def prepare_download(self, game: Game, base_game: Game = None, base_path: str = '', status_q: Queue = None, max_shm: int = 0, max_workers: int = 0, diff --git a/legendary/downloader/mp/manager.py b/legendary/downloader/mp/manager.py index 1a1ab2f..38b4449 100644 --- a/legendary/downloader/mp/manager.py +++ b/legendary/downloader/mp/manager.py @@ -29,7 +29,7 @@ class DLManager(Process): self.base_url = base_url self.dl_dir = download_dir - self.cache_dir = cache_dir if cache_dir else os.path.join(download_dir, '.cache') + self.cache_dir = cache_dir or os.path.join(download_dir, '.cache') # All the queues! self.logging_queue = None @@ -37,7 +37,7 @@ class DLManager(Process): self.writer_queue = None self.dl_result_q = None self.writer_result_q = None - self.max_workers = max_workers if max_workers else min(cpu_count() * 2, 16) + self.max_workers = max_workers or min(cpu_count() * 2, 16) self.dl_timeout = dl_timeout # Analysis stuff diff --git a/legendary/downloader/mp/workers.py b/legendary/downloader/mp/workers.py index 0b647c9..e16cd96 100644 --- a/legendary/downloader/mp/workers.py +++ b/legendary/downloader/mp/workers.py @@ -51,12 +51,12 @@ class DLWorker(Process): empty = False except Empty: if not empty: - logger.debug(f'Queue Empty, waiting for more...') + logger.debug('Queue Empty, waiting for more...') empty = True continue if isinstance(job, TerminateWorkerTask): # let worker die - logger.debug(f'Worker received termination signal, shutting down...') + logger.debug('Worker received termination signal, shutting down...') break tries = 0 @@ -99,7 +99,7 @@ class DLWorker(Process): break if not chunk: - logger.warning(f'Chunk somehow None?') + logger.warning('Chunk somehow None?') self.o_q.put(DownloaderTaskResult(success=False, **job.__dict__)) continue @@ -107,7 +107,7 @@ class DLWorker(Process): try: size = len(chunk.data) if size > job.shm.size: - logger.fatal(f'Downloaded chunk is longer than SharedMemorySegment!') + logger.fatal('Downloaded chunk is longer than SharedMemorySegment!') self.shm.buf[job.shm.offset:job.shm.offset + size] = bytes(chunk.data) del chunk @@ -130,7 +130,7 @@ class FileWorker(Process): self.q = queue self.o_q = out_queue self.base_path = base_path - self.cache_path = cache_path if cache_path else os.path.join(base_path, '.cache') + self.cache_path = cache_path or os.path.join(base_path, '.cache') self.shm = SharedMemory(name=shm) self.log_level = logging.getLogger().level self.logging_queue = logging_queue @@ -143,7 +143,7 @@ class FileWorker(Process): logger = logging.getLogger(self.name) logger.setLevel(self.log_level) - logger.debug(f'Download worker reporting for duty!') + logger.debug('Download worker reporting for duty!') last_filename = '' current_file = None @@ -159,7 +159,7 @@ class FileWorker(Process): if isinstance(j, TerminateWorkerTask): if current_file: current_file.close() - logger.debug(f'Worker received termination signal, shutting down...') + logger.debug('Worker received termination signal, shutting down...') # send termination task to results halnder as well self.o_q.put(TerminateWorkerTask()) break diff --git a/legendary/lfs/lgndry.py b/legendary/lfs/lgndry.py index a0d273e..bc75263 100644 --- a/legendary/lfs/lgndry.py +++ b/legendary/lfs/lgndry.py @@ -90,7 +90,7 @@ class LGDLFS: except Exception as e: self.log.error(f'Unable to read configuration file, please ensure that file is valid! ' f'(Error: {repr(e)})') - self.log.warning(f'Continuing with blank config in safe-mode...') + self.log.warning('Continuing with blank config in safe-mode...') self.config.read_only = True # make sure "Legendary" section exists @@ -221,8 +221,7 @@ class LGDLFS: f.write(manifest_data) def get_game_meta(self, app_name): - _meta = self._game_metadata.get(app_name, None) - if _meta: + if _meta := self._game_metadata.get(app_name, None): return Game.from_json(_meta) return None @@ -233,14 +232,14 @@ class LGDLFS: json.dump(json_meta, open(meta_file, 'w'), indent=2, sort_keys=True) def delete_game_meta(self, app_name): - if app_name in self._game_metadata: - del self._game_metadata[app_name] - meta_file = os.path.join(self.path, 'metadata', f'{app_name}.json') - if os.path.exists(meta_file): - os.remove(meta_file) - else: + if app_name not in self._game_metadata: raise ValueError(f'Game {app_name} does not exist in metadata DB!') + del self._game_metadata[app_name] + meta_file = os.path.join(self.path, 'metadata', f'{app_name}.json') + if os.path.exists(meta_file): + os.remove(meta_file) + def get_game_app_names(self): return sorted(self._game_metadata.keys()) @@ -264,9 +263,16 @@ class LGDLFS: self.log.warning(f'Failed to delete file "{f}": {e!r}') def clean_manifests(self, in_use): - in_use_files = set(f'{clean_filename(f"{app_name}_{version}")}.manifest' for app_name, version, _ in in_use) - in_use_files |= set(f'{clean_filename(f"{app_name}_{platform}_{version}")}.manifest' - for app_name, version, platform in in_use) + in_use_files = { + f'{clean_filename(f"{app_name}_{version}")}.manifest' + for app_name, version, _ in in_use + } + + in_use_files |= { + f'{clean_filename(f"{app_name}_{platform}_{version}")}.manifest' + for app_name, version, platform in in_use + } + for f in os.listdir(os.path.join(self.path, 'manifests')): if f not in in_use_files: try: @@ -282,8 +288,7 @@ class LGDLFS: self.log.debug(f'Failed to load installed game data: {e!r}') return None - game_json = self._installed.get(app_name, None) - if game_json: + if game_json := self._installed.get(app_name, None): return InstalledGame.from_json(game_json) return None @@ -392,7 +397,7 @@ class LGDLFS: def get_overlay_install_info(self): if not self._overlay_install_info: try: - data = json.load(open(os.path.join(self.path, f'overlay_install.json'))) + data = json.load(open(os.path.join(self.path, 'overlay_install.json'))) self._overlay_install_info = InstalledGame.from_json(data) except Exception as e: self.log.debug(f'Failed to load overlay install data: {e!r}') @@ -440,9 +445,7 @@ class LGDLFS: def serialise_sets(obj): """Turn sets into sorted lists for storage""" - if isinstance(obj, set): - return sorted(obj) - return obj + return sorted(obj) if isinstance(obj, set) else obj json.dump(alias_map, open(os.path.join(self.path, 'aliases.json'), 'w', newline='\n'), indent=2, sort_keys=True, default=serialise_sets) diff --git a/legendary/lfs/utils.py b/legendary/lfs/utils.py index 74b4cbc..95043b9 100644 --- a/legendary/lfs/utils.py +++ b/legendary/lfs/utils.py @@ -40,7 +40,7 @@ def delete_filelist(path: str, filenames: List[str], _dir, _fn = os.path.split(filename) if _dir: dirs.add(_dir) - + try: os.remove(os.path.join(path, _dir, _fn)) except Exception as e: @@ -66,14 +66,14 @@ def delete_filelist(path: str, filenames: List[str], if not silent: logger.error(f'Failed removing directory "{_dir}" with {e!r}') no_error = False - + if delete_root_directory: try: os.rmdir(path) except Exception as e: if not silent: logger.error(f'Removing game directory failed with {e!r}') - + return no_error diff --git a/legendary/models/chunk.py b/legendary/models/chunk.py index c32a2b5..6510912 100644 --- a/legendary/models/chunk.py +++ b/legendary/models/chunk.py @@ -113,10 +113,7 @@ class Chunk: return _chunk def write(self, fp=None, compress=True): - if not fp: - bio = BytesIO() - else: - bio = fp + bio = fp or BytesIO() self.uncompressed_size = self.compressed_size = len(self.data) if compress or self.compressed: @@ -143,7 +140,4 @@ class Chunk: # finally, add the data bio.write(self._data) - if not fp: - return bio.getvalue() - else: - return bio.tell() + return bio.tell() if fp else bio.getvalue() diff --git a/legendary/models/egl.py b/legendary/models/egl.py index 94ff3ac..164939d 100644 --- a/legendary/models/egl.py +++ b/legendary/models/egl.py @@ -145,9 +145,9 @@ class EGLManifest: tmp.executable = igame.executable tmp.main_game_appname = game.app_name # todo for DLC support this needs to be the base game tmp.app_folder_name = game.metadata.get('customAttributes', {}).get('FolderName', {}).get('value', '') - tmp.manifest_location = igame.install_path + '/.egstore' + tmp.manifest_location = f'{igame.install_path}/.egstore' tmp.ownership_token = igame.requires_ot - tmp.staging_location = igame.install_path + '/.egstore/bps' + tmp.staging_location = f'{igame.install_path}/.egstore/bps' tmp.can_run_offline = igame.can_run_offline tmp.is_incomplete_install = False tmp.needs_validation = igame.needs_verification diff --git a/legendary/models/manifest.py b/legendary/models/manifest.py index 174726a..0cbefb4 100644 --- a/legendary/models/manifest.py +++ b/legendary/models/manifest.py @@ -92,8 +92,7 @@ class Manifest: _m.file_manifest_list = FML.read(_tmp) _m.custom_fields = CustomFields.read(_tmp) - unhandled_data = _tmp.read() - if unhandled_data: + if unhandled_data := _tmp.read(): logger.warning(f'Did not read {len(unhandled_data)} remaining bytes in manifest! ' f'This may not be a problem.') @@ -152,10 +151,7 @@ class Manifest: self.data = zlib.compress(self.data) self.size_compressed = len(self.data) - if not fp: - bio = BytesIO() - else: - bio = fp + bio = fp or BytesIO() bio.write(struct.pack('