Install/update EGL content after login

This commit is contained in:
Katharina Dröge 2025-11-04 00:19:40 +01:00
parent 16123f7259
commit e9b9b60874
No known key found for this signature in database
GPG key ID: 3D586C5E334E5A26
2 changed files with 81 additions and 49 deletions

View file

@ -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'):

View file

@ -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: