# coding: utf-8 import json import os import logging from contextlib import contextmanager from collections import defaultdict from pathlib import Path from time import time from filelock import FileLock from .utils import clean_filename, LockedJSONData from legendary.models.game import * from legendary.utils.aliasing import generate_aliases from legendary.models.config import LGDConf from legendary.utils.env import is_windows_mac_or_pyi FILELOCK_DEBUG = False class LGDLFS: def __init__(self, config_file=None): self.log = logging.getLogger('LGDLFS') if config_path := os.environ.get('LEGENDARY_CONFIG_PATH'): self.path = config_path elif config_path := os.environ.get('XDG_CONFIG_HOME'): self.path = os.path.join(config_path, 'legendary') else: self.path = os.path.expanduser('~/.config/legendary') # EGS user info self._user_data = None # EGS entitlements self._entitlements = None # EGS asset data self._assets = None # EGS metadata self._game_metadata = dict() # Legendary update check info self._update_info = None # EOS Overlay install/update check info self._overlay_update_info = None self._overlay_install_info = None # Config with game specific settings (e.g. start parameters, env variables) self.config = LGDConf(comment_prefixes='/', allow_no_value=True) if config_file: # if user specified a valid relative/absolute path use that, # otherwise create file in legendary config directory if os.path.exists(config_file): self.config_path = os.path.abspath(config_file) else: self.config_path = os.path.join(self.path, clean_filename(config_file)) self.log.info(f'Using non-default config file "{self.config_path}"') else: self.config_path = os.path.join(self.path, 'config.ini') # ensure folders exist. for f in ['', 'manifests', 'metadata', 'tmp']: if not os.path.exists(os.path.join(self.path, f)): os.makedirs(os.path.join(self.path, f)) # if "old" folder exists migrate files and remove it if os.path.exists(os.path.join(self.path, 'manifests', 'old')): self.log.info('Migrating manifest files from old folders to new, please wait...') # remove unversioned manifest files for _f in os.listdir(os.path.join(self.path, 'manifests')): if '.manifest' not in _f: continue if '_' not in _f or (_f.startswith('UE_') and _f.count('_') < 2): self.log.debug(f'Deleting "{_f}" ...') os.remove(os.path.join(self.path, 'manifests', _f)) # move files from "old" to the base folder for _f in os.listdir(os.path.join(self.path, 'manifests', 'old')): try: self.log.debug(f'Renaming "{_f}"') os.rename(os.path.join(self.path, 'manifests', 'old', _f), os.path.join(self.path, 'manifests', _f)) except Exception as e: self.log.warning(f'Renaming manifest file "{_f}" failed: {e!r}') # remove "old" folder try: os.removedirs(os.path.join(self.path, 'manifests', 'old')) except Exception as e: self.log.warning(f'Removing "{os.path.join(self.path, "manifests", "old")}" folder failed: ' f'{e!r}, please remove manually') if not FILELOCK_DEBUG: # Prevent filelock logger from spamming Legendary debug output filelock_logger = logging.getLogger('filelock') filelock_logger.setLevel(logging.INFO) # try loading config try: self.config.read(self.config_path) 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('Continuing with blank config in safe-mode...') self.config.read_only = True # make sure "Legendary" section exists if 'Legendary' not in self.config: self.config.add_section('Legendary') # Add opt-out options with explainers if not self.config.has_option('Legendary', 'disable_update_check'): self.config.set('Legendary', '; Disables the automatic update check') self.config.set('Legendary', 'disable_update_check', 'false') if not self.config.has_option('Legendary', 'disable_update_notice'): self.config.set('Legendary', '; Disables the notice about an available update on exit') self.config.set('Legendary', 'disable_update_notice', 'false' if is_windows_mac_or_pyi() else 'true') self._installed_lock = FileLock(os.path.join(self.path, 'installed.json') + '.lock') try: self._installed = json.load(open(os.path.join(self.path, 'installed.json'))) except Exception as e: self.log.debug(f'Loading installed games failed: {e!r}') self._installed = None # load existing app metadata for gm_file in os.listdir(os.path.join(self.path, 'metadata')): try: _meta = json.load(open(os.path.join(self.path, 'metadata', gm_file))) self._game_metadata[_meta['app_name']] = _meta except Exception as e: self.log.debug(f'Loading game meta file "{gm_file}" failed: {e!r}') # load auto-aliases if enabled self.aliases = dict() if not self.config.getboolean('Legendary', 'disable_auto_aliasing', fallback=False): try: _j = json.load(open(os.path.join(self.path, 'aliases.json'))) for app_name, aliases in _j.items(): for alias in aliases: self.aliases[alias] = app_name except Exception as e: self.log.debug(f'Loading aliases failed with {e!r}') @property @contextmanager def userdata_lock(self) -> LockedJSONData: """Wrapper around the lock to automatically update user data when it is released""" with LockedJSONData(os.path.join(self.path, 'user.json')) as lock: try: yield lock finally: self._user_data = lock.data @property def userdata(self): if self._user_data is not None: return self._user_data try: with self.userdata_lock as locked: return locked.data except Exception as e: self.log.debug(f'Failed to load user data: {e!r}') return None @userdata.setter def userdata(self, userdata): raise NotImplementedError('The setter has been removed, use the locked userdata instead.') def invalidate_userdata(self): with self.userdata_lock as lock: lock.clear() @property def entitlements(self): if self._entitlements is not None: return self._entitlements try: self._entitlements = json.load(open(os.path.join(self.path, 'entitlements.json'))) return self._entitlements except Exception as e: self.log.debug(f'Failed to load entitlements data: {e!r}') return None @entitlements.setter def entitlements(self, entitlements): if entitlements is None: raise ValueError('Entitlements is none!') self._entitlements = entitlements json.dump(entitlements, open(os.path.join(self.path, 'entitlements.json'), 'w'), indent=2, sort_keys=True) @property def assets(self): if self._assets is None: try: tmp = json.load(open(os.path.join(self.path, 'assets.json'))) self._assets = {k: [GameAsset.from_json(j) for j in v] for k, v in tmp.items()} except Exception as e: self.log.debug(f'Failed to load assets data: {e!r}') return None return self._assets @assets.setter def assets(self, assets): if assets is None: raise ValueError('Assets is none!') self._assets = assets json.dump({platform: [a.__dict__ for a in assets] for platform, assets in self._assets.items()}, open(os.path.join(self.path, 'assets.json'), 'w'), indent=2, sort_keys=True) def _get_manifest_filename(self, app_name, version, platform=None): if platform: fname = clean_filename(f'{app_name}_{platform}_{version}') else: fname = clean_filename(f'{app_name}_{version}') return os.path.join(self.path, 'manifests', f'{fname}.manifest') def load_manifest(self, app_name, version, platform='Windows'): try: return open(self._get_manifest_filename(app_name, version, platform), 'rb').read() except FileNotFoundError: # all other errors should propagate self.log.debug(f'Loading manifest failed, retrying without platform in filename...') try: return open(self._get_manifest_filename(app_name, version), 'rb').read() except FileNotFoundError: # all other errors should propagate return None def save_manifest(self, app_name, manifest_data, version, platform='Windows'): with open(self._get_manifest_filename(app_name, version, platform), 'wb') as f: f.write(manifest_data) def get_game_meta(self, app_name): if _meta := self._game_metadata.get(app_name, None): return Game.from_json(_meta) return None def set_game_meta(self, app_name, meta): json_meta = meta.__dict__ self._game_metadata[app_name] = json_meta meta_file = os.path.join(self.path, 'metadata', f'{app_name}.json') json.dump(json_meta, open(meta_file, 'w'), indent=2, sort_keys=True) def delete_game_meta(self, app_name): 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()) def get_tmp_path(self): return os.path.join(self.path, 'tmp') def clean_tmp_data(self): for f in os.listdir(os.path.join(self.path, 'tmp')): try: os.remove(os.path.join(self.path, 'tmp', f)) except Exception as e: self.log.warning(f'Failed to delete file "{f}": {e!r}') def clean_metadata(self, app_names): for f in os.listdir(os.path.join(self.path, 'metadata')): app_name = f.rpartition('.')[0] if app_name not in app_names: try: os.remove(os.path.join(self.path, 'metadata', f)) except Exception as e: self.log.warning(f'Failed to delete file "{f}": {e!r}') def clean_manifests(self, 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: os.remove(os.path.join(self.path, 'manifests', f)) except Exception as e: self.log.warning(f'Failed to delete file "{f}": {e!r}') def lock_installed(self) -> bool: """ Locks the install data. We do not care about releasing this lock. If it is acquired by a Legendary instance it should own the lock until it exits. Some operations such as egl sync may be simply skipped if a lock cannot be acquired """ if self._installed_lock.is_locked: return True try: self._installed_lock.acquire(blocking=False) # reload data in case it has been updated elsewhere try: self._installed = json.load(open(os.path.join(self.path, 'installed.json'))) except Exception as e: self.log.debug(f'Failed to load installed game data: {e!r}') return True except TimeoutError: return False def get_installed_game(self, app_name): if self._installed is None: try: self._installed = json.load(open(os.path.join(self.path, 'installed.json'))) except Exception as e: self.log.debug(f'Failed to load installed game data: {e!r}') return None if game_json := self._installed.get(app_name, None): return InstalledGame.from_json(game_json) return None def set_installed_game(self, app_name, install_info): if self._installed is None: self._installed = dict() if app_name in self._installed: self._installed[app_name].update(install_info.__dict__) else: self._installed[app_name] = install_info.__dict__ json.dump(self._installed, open(os.path.join(self.path, 'installed.json'), 'w'), indent=2, sort_keys=True) def remove_installed_game(self, app_name): if self._installed is None: self.log.warning('Trying to remove a game, but no installed games?!') return if app_name in self._installed: del self._installed[app_name] else: self.log.warning('Trying to remove non-installed game:', app_name) return json.dump(self._installed, open(os.path.join(self.path, 'installed.json'), 'w'), indent=2, sort_keys=True) def get_installed_list(self): if not self._installed: return [] return [InstalledGame.from_json(i) for i in self._installed.values()] def save_config(self): # do not save if in read-only mode or file hasn't changed if self.config.read_only or not self.config.modified: return # if config file has been modified externally, back-up the user-modified version before writing if os.path.exists(self.config_path): if (modtime := int(os.stat(self.config_path).st_mtime)) != self.config.modtime: new_filename = f'config.{modtime}.ini' self.log.warning(f'Configuration file has been modified while legendary was running, ' f'user-modified config will be renamed to "{new_filename}"...') os.rename(self.config_path, os.path.join(os.path.dirname(self.config_path), new_filename)) with open(self.config_path, 'w') as cf: self.config.write(cf) def get_dir_size(self): return sum(f.stat().st_size for f in Path(self.path).glob('**/*') if f.is_file()) def get_cached_version(self): if self._update_info: return self._update_info try: self._update_info = json.load(open(os.path.join(self.path, 'version.json'))) except Exception as e: self.log.debug(f'Failed to load cached update data: {e!r}') self._update_info = dict(last_update=0, data=None) return self._update_info def set_cached_version(self, version_data): if not version_data: return self._update_info = dict(last_update=time(), data=version_data) json.dump(self._update_info, open(os.path.join(self.path, 'version.json'), 'w'), indent=2, sort_keys=True) def get_cached_sdl_data(self, app_name): try: return json.load(open(os.path.join(self.path, 'tmp', f'{app_name}.json'))) except Exception as e: self.log.debug(f'Failed to load cached SDL data: {e!r}') return None def set_cached_sdl_data(self, app_name, sdl_version, sdl_data): if not app_name or not sdl_data: return json.dump(dict(version=sdl_version, data=sdl_data), open(os.path.join(self.path, 'tmp', f'{app_name}.json'), 'w'), indent=2, sort_keys=True) def get_cached_overlay_version(self): if self._overlay_update_info: return self._overlay_update_info try: self._overlay_update_info = json.load(open( os.path.join(self.path, 'overlay_version.json'))) except Exception as e: self.log.debug(f'Failed to load cached Overlay update data: {e!r}') self._overlay_update_info = dict(last_update=0, data=None) return self._overlay_update_info def set_cached_overlay_version(self, version_data): self._overlay_update_info = dict(last_update=time(), data=version_data) json.dump(self._overlay_update_info, open(os.path.join(self.path, 'overlay_version.json'), 'w'), indent=2, sort_keys=True) def get_overlay_install_info(self): if not self._overlay_install_info: try: 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}') return self._overlay_install_info def set_overlay_install_info(self, igame: InstalledGame): self._overlay_install_info = igame json.dump(vars(igame), open(os.path.join(self.path, 'overlay_install.json'), 'w'), indent=2, sort_keys=True) def remove_overlay_install_info(self): try: self._overlay_install_info = None os.remove(os.path.join(self.path, 'overlay_install.json')) except Exception as e: self.log.debug(f'Failed to delete overlay install data: {e!r}') def generate_aliases(self): self.log.debug('Generating list of aliases...') self.aliases = dict() aliases = set() collisions = set() alias_map = defaultdict(set) for app_name in self._game_metadata.keys(): game = self.get_game_meta(app_name) if game.is_dlc: continue game_folder = game.metadata.get('customAttributes', {}).get('FolderName', {}).get('value', None) _aliases = generate_aliases(game.app_title, game_folder=game_folder, app_name=game.app_name) for alias in _aliases: if alias not in aliases: aliases.add(alias) alias_map[game.app_name].add(alias) else: collisions.add(alias) # remove colliding aliases from map and add aliases to lookup table for app_name, aliases in alias_map.items(): alias_map[app_name] -= collisions for alias in alias_map[app_name]: self.aliases[alias] = app_name def serialise_sets(obj): """Turn sets into sorted lists for storage""" 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)