diff --git a/legendary/core.py b/legendary/core.py index 29ce3f1..7b715bb 100644 --- a/legendary/core.py +++ b/legendary/core.py @@ -226,6 +226,7 @@ class LegendaryCore: try: self.egs.resume_session(lock.data) self.logged_in = True + self.update_launcher_content() return True except InvalidCredentialsError as e: self.log.warning(f'Resuming failed due to invalid credentials: {e!r}') @@ -247,6 +248,7 @@ class LegendaryCore: lock.data = userdata self.logged_in = True + self.update_launcher_content() return True def login(self, force_refresh=False) -> bool: @@ -285,22 +287,56 @@ class LegendaryCore: 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']) - self._egl_version = version_info['egl_config'].get('version', self._egl_version) - for data_key in version_info['egl_config'].get('data_keys', []): + if egl_config := version_info.get('egl_config'): + self.egs.update_egs_params(egl_config) + self._egl_version = egl_config.get('version', self._egl_version) + for data_key in egl_config.get('data_keys', []): if data_key not in self.egl.data_keys: self.egl.data_keys.append(data_key) if game_overrides := version_info.get('game_overrides'): 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(): - if app_name not in sdl_games: - sdl_games[app_name] = None if lgd_config := version_info.get('legendary_config'): self.webview_killswitch = lgd_config.get('webview_killswitch', False) + def update_launcher_content(self) -> None: + """Updates metadata we need from the EGL manifest""" + cached = self.lgd.get_cached_egl_content_version() + if cached['version'] and (datetime.now().timestamp() - cached['last_update']) < 24*3600: + self.log.debug('EGL content version cache is still fresh, not updating') + return + + assets = self.egs.get_launcher_manifests() + asset = next(( + asset for asset in assets['elements'] + if asset['appName'] == 'EpicGamesLauncherContent' + ), None) + + if not asset: + raise ValueError('"EpicGamesLauncherContent" asset not found in launcher manifests') + + if asset['buildVersion'] == cached['version']: + self.log.debug('EGL content is up-to-date') + self.lgd.set_cached_egl_content_version(asset['buildVersion']) + return + + manifest_urls, base_urls, manifest_hash = self._get_cdn_urls_for_asset(asset) + manifest_bytes = self._download_manifest(manifest_urls, manifest_hash) + manifest = self.load_manifest(manifest_bytes) + + dlm = DLManager(self.lgd.egl_content_path, base_urls[0]) + dlm.log.setLevel(logging.WARNING) + + suffixes = [ + # We need sdmeta files for selective downloads + 'sdmeta' + ] + + dlm.run_analysis(manifest=manifest, file_suffix_filter=suffixes) + dlm.start() + dlm.join() + self.log.info(f'EGL content updated from {cached["version"]} to {asset["buildVersion"]}') + self.lgd.set_cached_egl_content_version(asset['buildVersion']) + def get_egl_version(self): return self._egl_version @@ -314,31 +350,6 @@ class LegendaryCore: return update_info.get('game_wiki', {}).get(app_name, {}).get(sys_platform) - def get_sdl_data(self, app_name, platform='Windows'): - if platform not in ('Win32', 'Windows'): - app_name = f'{app_name}_{platform}' - - if app_name not in sdl_games: - return None - # load hardcoded data as fallback - sdl_data = sdl_games[app_name] - # get cached data - cached = self.lgd.get_cached_sdl_data(app_name) - # check if newer version is available and/or download if necessary - version_info = self.lgd.get_cached_version()['data'] - latest = version_info.get('game_overrides', {}).get('sdl_config', {}).get(app_name) - if (not cached and latest) or (cached and latest and latest > cached['version']): - try: - sdl_data = self.lgdapi.get_sdl_config(app_name) - self.log.debug(f'Downloaded SDL data for "{app_name}", version: {latest}') - self.lgd.set_cached_sdl_data(app_name, latest, sdl_data) - except Exception as e: - self.log.warning(f'Downloading SDL data failed with {e!r}') - elif cached: - sdl_data = cached['data'] - # return data if available - return sdl_data - def update_aliases(self, force=False): _aliases_enabled = not self.lgd.config.getboolean('Legendary', 'disable_auto_aliasing', fallback=False) if _aliases_enabled and (force or not self.lgd.aliases): @@ -1252,10 +1263,14 @@ class LegendaryCore: if len(m_api_r['elements']) > 1: raise ValueError('Manifest response has more than one element!') - manifest_hash = m_api_r['elements'][0]['hash'] + return self._get_cdn_urls_for_asset(m_api_r['elements'][0]) + + @staticmethod + def _get_cdn_urls_for_asset(asset): + manifest_hash = asset['hash'] base_urls = [] manifest_urls = [] - for manifest in m_api_r['elements'][0]['manifests']: + for manifest in asset['manifests']: base_url = manifest['uri'].rpartition('/')[0] if base_url not in base_urls: base_urls.append(base_url) @@ -1276,6 +1291,9 @@ class LegendaryCore: if disable_https: manifest_urls = [url.replace('https://', 'http://') for url in manifest_urls] + return self._download_manifest(manifest_urls, manifest_hash), base_urls + + def _download_manifest(self, manifest_urls: list[str], manifest_hash: str): for url in manifest_urls: self.log.debug(f'Trying to download manifest from "{url}"...') try: @@ -1297,7 +1315,7 @@ class LegendaryCore: if sha1(manifest_bytes).hexdigest() != manifest_hash: raise ValueError('Manifest sha hash mismatch!') - return manifest_bytes, base_urls + return manifest_bytes def get_uri_manifest(self, uri): if uri.startswith('http'): diff --git a/legendary/lfs/lgndry.py b/legendary/lfs/lgndry.py index 2e9b0bf..9432e4e 100644 --- a/legendary/lfs/lgndry.py +++ b/legendary/lfs/lgndry.py @@ -46,6 +46,8 @@ class LGDLFS: # EOS Overlay install/update check info self._overlay_update_info = None self._overlay_install_info = None + # EGL content update check info + self._egl_content_update_info = None # Config with game specific settings (e.g. start parameters, env variables) self.config = LGDConf(comment_prefixes='/', allow_no_value=True) @@ -401,19 +403,31 @@ class LGDLFS: 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 + @property + def egl_content_path(self) -> Path: + return Path(self.path) / 'egl_content' - 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) + @property + def _egl_content_version_path(self) -> Path: + return Path(self.path) / 'egl_content_version.json' + + def get_cached_egl_content_version(self): + if self._egl_content_update_info: + return self._egl_content_update_info + + try: + self._egl_content_update_info = json.loads(self._egl_content_version_path.read_text()) + except Exception as e: + self.log.debug(f'Failed to load EGL content version: {e!r}') + self._egl_content_update_info = dict(last_update=0, version=None) + + return self._egl_content_update_info + + def set_cached_egl_content_version(self, new_version: str): + self._egl_content_update_info = dict(version=new_version, last_update=time()) + self._egl_content_version_path.write_text( + json.dumps(self._egl_content_update_info, indent=2, sort_keys=True) + ) def get_cached_overlay_version(self): if self._overlay_update_info: