diff --git a/legendary/cli.py b/legendary/cli.py index 9d8430d..998b017 100644 --- a/legendary/cli.py +++ b/legendary/cli.py @@ -283,7 +283,7 @@ class LegendaryCLI: versions = dict() for game in games: try: - versions[game.app_name] = self.core.get_asset(game.app_name, platform=game.platform).build_version + versions[game.app_name] = self.core.get_asset(game.app_name, platform=game.platform, namespace=game.namespace).build_version except ValueError: logger.warning(f'Metadata for "{game.app_name}" is missing, the game may have been removed from ' f'your account or not be in legendary\'s database yet, try rerunning the command ' @@ -822,6 +822,8 @@ class LegendaryCLI: args.app_name = self._resolve_aliases(args.app_name) if self.core.is_installed(args.app_name): igame = self.core.get_installed_game(args.app_name) + if not args.namespace: + args.namespace = igame.namespace args.platform = igame.platform if igame.needs_verification and not args.repair_mode: logger.info('Game needs to be verified before updating, switching to repair mode...') @@ -875,6 +877,23 @@ class LegendaryCLI: else: logger.warning(f'No asset found for platform "{args.platform}", ' f'trying anyway since --no-install is set.') + elif not args.namespace and len(game.asset_infos[args.platform]) > 1: + asset_infos = [] + for asset in game.asset_infos[args.platform]: + namespace_info = game.namespaces.get(asset.namespace) + if namespace_info: + asset_infos.append((namespace_info.get('sandboxType'), namespace_info.get('sandboxName'), asset.namespace)) + type_name_str = '\n'.join([f'{_t}\t{_n}\t{_ns}' for _t,_n,_ns in asset_infos]) + logger.error('You have access to more than one asset for this game\n' + f'Type\tName\tNamespace\n' + +type_name_str + +'\nuse --namespace to pick one') + exit(1) + + if args.namespace and args.namespace not in game.namespaces: + available_namespaces = '\n'.join(list(game.namespaces.keys())) + logger.error("Unknown namespace\n" + available_namespaces) + exit(1) if game.is_dlc: logger.info('Install candidate is DLC') @@ -979,6 +998,7 @@ class LegendaryCLI: override_delta_manifest=args.override_delta_manifest, preferred_cdn=args.preferred_cdn, disable_https=args.disable_https, + namespace=args.namespace, bind_ip=args.bind_ip) # game is either up-to-date or hasn't changed, so we have nothing to do @@ -1644,6 +1664,7 @@ class LegendaryCLI: manifest_data = None entitlements = None + namespace = args.namespace or game.namespace # load installed manifest or URI if args.offline or manifest_uri: if app_name and self.core.is_installed(app_name): @@ -1659,20 +1680,21 @@ class LegendaryCLI: logger.info('Game not installed and offline mode enabled, cannot load manifest.') elif game: entitlements = self.core.egs.get_user_entitlements_full() - egl_meta = self.core.egs.get_game_info(game.namespace, game.catalog_item_id) + egl_meta = self.core.egs.get_game_info(namespace, game.catalog_item_id) game.metadata = egl_meta # Get manifest if asset exists for current platform if args.platform in game.asset_infos: - manifest_data, _ = self.core.get_cdn_manifest(game, args.platform) + manifest_data, _ = self.core.get_cdn_manifest(game, args.platform, namespace) if game: game_infos = info_items['game'] game_infos.append(InfoItem('App name', 'app_name', game.app_name, game.app_name)) game_infos.append(InfoItem('Title', 'title', game.app_title, game.app_title)) - game_infos.append(InfoItem('Latest version', 'version', game.app_version(args.platform), - game.app_version(args.platform))) - all_versions = {k: v.build_version for k, v in game.asset_infos.items()} - game_infos.append(InfoItem('All versions', 'platform_versions', all_versions, all_versions)) + _v = game.app_version(args.platform, namespace) + game_infos.append(InfoItem('Latest version', 'version', _v, _v)) + all_versions = {k: '; '.join([a.build_version for a in v]) for k, v in game.asset_infos.items()} + all_versions_json = {k: [a.__dict__ for a in v] for k,v in game.asset_infos.items()} + game_infos.append(InfoItem('All versions', 'platform_versions', all_versions, all_versions_json)) # Cloud save support for Mac and Windows game_infos.append(InfoItem('Cloud saves supported', 'cloud_saves_supported', game.supports_cloud_saves or game.supports_mac_cloud_saves, @@ -1747,6 +1769,13 @@ class LegendaryCLI: else: game_infos.append(InfoItem('Owned DLC', 'owned_dlc', None, [])) + if len(game.namespaces.keys()) > 0: + all_namespaces = {_n['sandboxName']: '({}) - {}'.format(_n['sandboxType'], _n['namespace']) for _n in game.namespaces.values()} + game_infos.append(InfoItem('Namespaces', 'namespaces', all_namespaces, list(game.namespaces.values()))) + else: + game_infos.append(InfoItem('Namespaces', 'namespaces', None, [])) + + igame = self.core.get_installed_game(app_name) if igame: installation_info = info_items['install'] @@ -2799,6 +2828,7 @@ def main(): help='Do not ask about installing DLCs.') install_parser.add_argument('--bind', dest='bind_ip', action='store', metavar='', type=str, help='Comma-separated list of IPs to bind to for downloading') + install_parser.add_argument('--namespace', dest='namespace', help='Specify namespace to pick sandbox from which to install') uninstall_parser.add_argument('--keep-files', dest='keep_files', action='store_true', help='Keep files but remove game from Legendary database') @@ -2962,6 +2992,7 @@ def main(): help='Output information in JSON format') info_parser.add_argument('--platform', dest='platform', action='store', metavar='', type=str, help='Platform to fetch info for (default: installed or Mac on macOS, Windows otherwise)') + info_parser.add_argument('--namespace', dest='namespace', type=str, help='Specify namespace to return primary data of') store_group = activate_parser.add_mutually_exclusive_group(required=True) store_group.add_argument('-U', '--uplay', dest='uplay', action='store_true', diff --git a/legendary/core.py b/legendary/core.py index 04344e3..38070a7 100644 --- a/legendary/core.py +++ b/legendary/core.py @@ -365,13 +365,20 @@ class LegendaryCore: self.lgd.assets = assets return self.lgd.assets[platform] + + def get_library_items(self, include_metadata=True, force_refresh=False): + if force_refresh or not self.lgd.library_items: + lib = self.egs.get_library_items(include_metadata) + if self.lgd.library_items != lib: + self.lgd.library_items = lib + return self.lgd.library_items - def get_asset(self, app_name, platform='Windows', update=False) -> GameAsset: + def get_asset(self, app_name, platform='Windows', update=False, namespace=None) -> GameAsset: if update or platform not in self.lgd.assets: self.get_assets(update_assets=True, platform=platform) try: - return next(i for i in self.lgd.assets[platform] if i.app_name == app_name) + return next(i for i in self.lgd.assets[platform] if i.app_name == app_name and (namespace is None or i.namespace == namespace)) except StopIteration: raise ValueError @@ -412,6 +419,8 @@ class LegendaryCore: for _platform in platforms: self.get_assets(update_assets=update_assets, platform=_platform) + library_items = self.get_library_items(force_refresh=update_assets) + if not self.lgd.assets: return _ret, _dlc @@ -419,31 +428,35 @@ class LegendaryCore: for _platform, _assets in self.lgd.assets.items(): for _asset in _assets: if _asset.app_name in assets: - assets[_asset.app_name][_platform] = _asset + assets[_asset.app_name][_platform].append(_asset) else: - assets[_asset.app_name] = {_platform: _asset} + assets[_asset.app_name] = {_platform: [_asset]} fetch_list = [] games = {} + sidecars = {} for app_name, app_assets in sorted(assets.items()): - if skip_ue and any(v.namespace == 'ue' for v in app_assets.values()): + if skip_ue and any(v.namespace == 'ue' for _assets in app_assets.values() for v in _assets): continue 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, _a.namespace) != _a.build_version for _p in app_assets.keys() for _a in app_assets[_p]) # 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)) + sidecar_updated = any(_a.sidecar_rev > 0 and + (not game.sidecars or _a.namespace not in game.sidecars + or game.sidecars[_a.namespace].rev != _a.sidecar_rev) + for _a in app_assets['Windows']) games[app_name] = game if update_assets and (not game or force_refresh or (game and (asset_updated or sidecar_updated))): self.log.debug(f'Scheduling metadata update for {app_name}') # namespace/catalog item are the same for all platforms, so we can just use the first one - _ga = next(iter(app_assets.values())) - fetch_list.append((app_name, _ga.namespace, _ga.catalog_item_id, sidecar_updated)) + _gas = next(iter(app_assets.values())) + for _ga in _gas: + fetch_list.append((app_name, _ga.namespace, _ga.catalog_item_id, sidecar_updated)) meta_updated = True def fetch_game_meta(args): @@ -453,7 +466,6 @@ class LegendaryCore: self.log.warning(f'App {app_name} does not have any metadata!') eg_meta = dict(title='Unknown') - sidecar = None if update_sidecar: self.log.debug(f'Updating sidecar information for {app_name}...') manifest_api_response = self.egs.get_game_manifest(namespace, catalog_item_id, app_name) @@ -462,9 +474,13 @@ class LegendaryCore: if 'sidecar' in manifest_info: sidecar_json = json.loads(manifest_info['sidecar']['config']) sidecar = Sidecar(config=sidecar_json, rev=manifest_info['sidecar']['rvn']) + if not app_name in sidecars: + sidecars[app_name] = {namespace: sidecar} + else: + sidecars[app_name].update({namespace:sidecar}) game = Game(app_name=app_name, app_title=eg_meta['title'], metadata=eg_meta, asset_infos=assets[app_name], - sidecar=sidecar) + sidecars=sidecars.get(app_name)) self.lgd.set_game_meta(game.app_name, game) games[app_name] = game try: @@ -482,7 +498,7 @@ class LegendaryCore: executor.map(fetch_game_meta, fetch_list, timeout=60.0) for app_name, app_assets in sorted(assets.items()): - if skip_ue and any(v.namespace == 'ue' for v in app_assets.values()): + if skip_ue and any(v.namespace == 'ue' for a in app_assets.values() for v in a): continue game = games.get(app_name) @@ -490,8 +506,9 @@ class LegendaryCore: if not game or app_name in still_needs_update: if use_threads: self.log.warning(f'Fetching metadata for {app_name} failed, retrying') - _ga = next(iter(app_assets.values())) - fetch_game_meta((app_name, _ga.namespace, _ga.catalog_item_id, True)) + _gas = next(iter(app_assets.values())) + for _ga in _gas: + fetch_game_meta((app_name, _ga.namespace, _ga.catalog_item_id, True)) game = games[app_name] if game.is_dlc and platform in app_assets: @@ -499,6 +516,15 @@ class LegendaryCore: elif not any(i['path'] == 'mods' for i in game.metadata.get('categories', [])) and platform in app_assets: _ret.append(game) + # append info about each library entry per namespace + for record in library_items: + record_type = record.get("recordType") + if record_type and (record_type.lower() != "application"): + continue + if game := games.get(record["appName"]): + game.namespaces.update({record["namespace"]: record}) + self.lgd.set_game_meta(game.app_name, game) + self.update_aliases(force=meta_updated) if meta_updated: self._prune_metadata() @@ -541,14 +567,14 @@ class LegendaryCore: # broken old app name that we should always ignore ignore |= {'1'} - for libitem in self.egs.get_library_items(): + for libitem in self.get_library_items(): if libitem['namespace'] == 'ue' and skip_ue: continue if 'appName' not in libitem: continue if libitem['appName'] in ignore: continue - if libitem['sandboxType'] == 'PRIVATE': + if libitem['sandboxType'].lower() == 'private': continue game = self.lgd.get_game_meta(libitem['appName']) @@ -777,11 +803,13 @@ class LegendaryCore: '-AUTH_TYPE=exchangecode', f'-epicapp={app_name}', '-epicenv=Prod']) + + namespace = install.namespace or game.namespace if install.requires_ot and not offline: self.log.info('Getting ownership token.') - ovt = self.egs.get_ownership_token(game.namespace, game.catalog_item_id) - ovt_path = os.path.join(self.lgd.get_tmp_path(), f'{game.namespace}{game.catalog_item_id}.ovt') + ovt = self.egs.get_ownership_token(namespace, game.catalog_item_id) + ovt_path = os.path.join(self.lgd.get_tmp_path(), f'{namespace}{game.catalog_item_id}.ovt') with open(ovt_path, 'wb') as f: f.write(ovt) params.egl_parameters.append(f'-epicovt={ovt_path}') @@ -795,10 +823,10 @@ class LegendaryCore: f'-epicusername={user_name}', f'-epicuserid={account_id}', f'-epiclocale={language_code}', - f'-epicsandboxid={game.namespace}' + f'-epicsandboxid={namespace}' ]) - if sidecar := game.sidecar: + if sidecar := game.sidecars and game.sidecars.get(namespace): if deployment_id := sidecar.config.get('deploymentId', None): params.egl_parameters.append(f'-epicdeploymentid={deployment_id}') @@ -1244,8 +1272,9 @@ class LegendaryCore: old_bytes = self.lgd.load_manifest(app_name, igame.version, igame.platform) return old_bytes, igame.base_urls - def get_cdn_urls(self, game, platform='Windows'): - m_api_r = self.egs.get_game_manifest(game.namespace, game.catalog_item_id, + def get_cdn_urls(self, game, platform='Windows', namespace=None): + ns = namespace or game.namespace + m_api_r = self.egs.get_game_manifest(ns, game.catalog_item_id, game.app_name, platform) # never seen this outside the launcher itself, but if it happens: PANIC! @@ -1268,8 +1297,8 @@ class LegendaryCore: return manifest_urls, base_urls, manifest_hash - def get_cdn_manifest(self, game, platform='Windows', disable_https=False): - manifest_urls, base_urls, manifest_hash = self.get_cdn_urls(game, platform) + def get_cdn_manifest(self, game, platform='Windows', namespace=None, disable_https=False): + manifest_urls, base_urls, manifest_hash = self.get_cdn_urls(game, platform, namespace) if not manifest_urls: raise ValueError('No manifest URLs returned by API') @@ -1331,7 +1360,8 @@ class LegendaryCore: repair: bool = False, repair_use_latest: bool = False, disable_delta: bool = False, override_delta_manifest: str = '', egl_guid: str = '', preferred_cdn: str = None, - disable_https: bool = False, bind_ip: str = None) -> (DLManager, AnalysisResult, ManifestMeta): + disable_https: bool = False, bind_ip: str = None, + namespace: str = None) -> (DLManager, AnalysisResult, ManifestMeta): # load old manifest old_manifest = None @@ -1363,7 +1393,7 @@ class LegendaryCore: if _base_urls: base_urls = _base_urls else: - new_manifest_data, base_urls = self.get_cdn_manifest(game, platform, disable_https=disable_https) + new_manifest_data, base_urls = self.get_cdn_manifest(game, platform, namespace=namespace, disable_https=disable_https) # overwrite base urls in metadata with current ones to avoid using old/dead CDNs game.base_urls = base_urls # save base urls to game metadata @@ -1539,7 +1569,8 @@ class LegendaryCore: can_run_offline=offline == 'true', requires_ot=ot == 'true', is_dlc=base_game is not None, install_size=anlres.install_size, egl_guid=egl_guid, install_tags=file_install_tag, - platform=platform, uninstaller=uninstaller) + platform=platform, uninstaller=uninstaller, + namespace=namespace) return dlm, anlres, igame @@ -1888,10 +1919,10 @@ class LegendaryCore: # copy manifest and create mancpn file in .egstore folder with open(os.path.join(egstore_folder, f'{egl_game.installation_guid}.manifest', ), 'wb') as mf: mf.write(manifest_data) - + ns = lgd_igame.namespace or lgd_game.namespace mancpn = dict(FormatVersion=0, AppName=app_name, CatalogItemId=lgd_game.catalog_item_id, - CatalogNamespace=lgd_game.namespace) + CatalogNamespace=ns) with open(os.path.join(egstore_folder, f'{egl_game.installation_guid}.mancpn', ), 'w') as mcpnf: json.dump(mancpn, mcpnf, indent=4, sort_keys=True) diff --git a/legendary/lfs/lgndry.py b/legendary/lfs/lgndry.py index 2e9b0bf..dc65c87 100644 --- a/legendary/lfs/lgndry.py +++ b/legendary/lfs/lgndry.py @@ -37,6 +37,8 @@ class LGDLFS: self._user_data = None # EGS entitlements self._entitlements = None + # EGS library items + self._library_items = None # EGS asset data self._assets = None # EGS metadata @@ -121,7 +123,8 @@ class LGDLFS: 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'))) + with open(os.path.join(self.path, 'installed.json')) as f: + self._installed = json.load(f) except Exception as e: self.log.debug(f'Loading installed games failed: {e!r}') self._installed = None @@ -129,7 +132,8 @@ class LGDLFS: # 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))) + with open(os.path.join(self.path, 'metadata', gm_file)) as f: + _meta = json.load(f) self._game_metadata[_meta['app_name']] = _meta except Exception as e: self.log.debug(f'Loading game meta file "{gm_file}" failed: {e!r}') @@ -138,7 +142,8 @@ class LGDLFS: 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'))) + with open(os.path.join(self.path, 'aliases.json')) as f: + _j = json.load(f) for app_name, aliases in _j.items(): for alias in aliases: self.aliases[alias] = app_name @@ -181,7 +186,8 @@ class LGDLFS: return self._entitlements try: - self._entitlements = json.load(open(os.path.join(self.path, 'entitlements.json'))) + with open(os.path.join(self.path, 'entitlements.json')) as f: + self._entitlements = json.load(f) return self._entitlements except Exception as e: self.log.debug(f'Failed to load entitlements data: {e!r}') @@ -193,14 +199,15 @@ class LGDLFS: 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) + with open(os.path.join(self.path, 'entitlements.json'), 'w') as f: + json.dump(entitlements, f, 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'))) + with open(os.path.join(self.path, 'assets.json')) as f: + tmp = json.load(f) 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}') @@ -214,9 +221,29 @@ class LGDLFS: 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) + with open(os.path.join(self.path, 'assets.json'), 'w') as f: + json.dump({platform: [a.__dict__ for a in assets] for platform, assets in self._assets.items()}, + f, indent=2, sort_keys=True) + + @property + def library_items(self): + if self._library_items is not None: + return self._library_items + try: + with open(os.path.join(self.path, 'library_items.json')) as f: + self._library_items = json.load(f) + return self._library_items + except Exception as e: + self.log.debug(f'Failed to load library items data: {e!r}') + return None + + @library_items.setter + def library_items(self, library_items): + if library_items is None: + raise ValueError("Library items is none!") + self._library_items = library_items + with open(os.path.join(self.path, 'library_items.json'), 'w') as f: + json.dump(library_items, f, indent=2, sort_keys=True) def _get_manifest_filename(self, app_name, version, platform=None): if platform: @@ -227,11 +254,13 @@ class LGDLFS: def load_manifest(self, app_name, version, platform='Windows'): try: - return open(self._get_manifest_filename(app_name, version, platform), 'rb').read() + with open(self._get_manifest_filename(app_name, version, platform), 'rb') as f: + return f.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() + with open(self._get_manifest_filename(app_name, version), 'rb') as f: + return f.read() except FileNotFoundError: # all other errors should propagate return None @@ -248,7 +277,8 @@ class LGDLFS: 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) + with open(meta_file, 'w') as f: + json.dump(json_meta, f, indent=2, sort_keys=True) def delete_game_meta(self, app_name): if app_name not in self._game_metadata: @@ -312,7 +342,8 @@ class LGDLFS: 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'))) + with open(os.path.join(self.path, 'installed.json')) as f: + self._installed = json.load(f) except Exception as e: self.log.debug(f'Failed to load installed game data: {e!r}') @@ -323,7 +354,8 @@ class LGDLFS: 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'))) + with open(os.path.join(self.path, 'installed.json')) as f: + self._installed = json.load(f) except Exception as e: self.log.debug(f'Failed to load installed game data: {e!r}') return None @@ -341,8 +373,8 @@ class LGDLFS: 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) + with open(os.path.join(self.path, 'installed.json'), 'w') as f: + json.dump(self._installed, f, indent=2, sort_keys=True) def remove_installed_game(self, app_name): if self._installed is None: @@ -355,8 +387,8 @@ class LGDLFS: 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) + with open(os.path.join(self.path, 'installed.json'), 'w') as f: + json.dump(self._installed, f, indent=2, sort_keys=True) def get_installed_list(self): if not self._installed: @@ -387,7 +419,8 @@ class LGDLFS: return self._update_info try: - self._update_info = json.load(open(os.path.join(self.path, 'version.json'))) + with open(os.path.join(self.path, 'version.json')) as f: + self._update_info = json.load(f) 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) @@ -398,12 +431,13 @@ class LGDLFS: 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) + with open(os.path.join(self.path, 'version.json'), 'w') as f: + json.dump(self._update_info, f, 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'))) + with open(os.path.join(self.path, 'tmp', f'{app_name}.json')) as f: + return json.load(f) except Exception as e: self.log.debug(f'Failed to load cached SDL data: {e!r}') return None @@ -411,17 +445,16 @@ class LGDLFS: 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) + with open(os.path.join(self.path, 'tmp', f'{app_name}.json'), 'w') as f: + json.dump(dict(version=sdl_version, data=sdl_data), f, 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'))) + with open(os.path.join(self.path, 'overlay_version.json')) as f: + self._overlay_update_info = json.load(f) 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) @@ -430,14 +463,14 @@ class LGDLFS: 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) + with open(os.path.join(self.path, 'overlay_version.json'), 'w') as f: + json.dump(self._overlay_update_info, f, 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'))) + with open(os.path.join(self.path, 'overlay_install.json')) as f: + data = json.load(f) self._overlay_install_info = InstalledGame.from_json(data) except Exception as e: self.log.debug(f'Failed to load overlay install data: {e!r}') @@ -446,8 +479,8 @@ class LGDLFS: 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) + with open(os.path.join(self.path, 'overlay_install.json'), 'w') as f: + json.dump(vars(igame), f, indent=2, sort_keys=True) def remove_overlay_install_info(self): try: @@ -487,5 +520,5 @@ class LGDLFS: """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) + with open(os.path.join(self.path, 'aliases.json'), 'w', newline='\n') as f: + json.dump(alias_map, f, indent=2, sort_keys=True, default=serialise_sets) diff --git a/legendary/models/game.py b/legendary/models/game.py index 5a7eec6..64749fd 100644 --- a/legendary/models/game.py +++ b/legendary/models/game.py @@ -71,15 +71,35 @@ class Game: app_name: str app_title: str - asset_infos: Dict[str, GameAsset] = field(default_factory=dict) + asset_infos: Dict[str, List[GameAsset]] = field(default_factory=dict) base_urls: List[str] = field(default_factory=list) metadata: Dict = field(default_factory=dict) - sidecar: Optional[Sidecar] = None + sidecars: Optional[Dict[str, Sidecar]] = None + namespaces: Dict = field(default_factory=dict) - def app_version(self, platform='Windows'): + def app_version(self, platform='Windows', namespace=None): if platform not in self.asset_infos: return None - return self.asset_infos[platform].build_version + if len(self.asset_infos[platform]) == 1: + return self.asset_infos[platform][0].build_version + + # Pick specified namespace, otherwise default to public one + if namespace is None: + return self._get_public_asset(platform).build_version + for asset in self.asset_infos[platform]: + if asset.namespace == namespace: + return asset.build_version + + return self.asset_infos[platform][0].build_version + + def _get_public_asset(self, platform='Windows'): + if len(self.asset_infos[platform]) == 1: + return self.asset_infos[platform][0] + for asset in self.asset_infos[platform]: + version_namespace = self.namespaces.get(asset.namespace) + if version_namespace and version_namespace["sandboxType"].lower() == "public": + return asset + return self.asset_infos[platform][0] @property def is_dlc(self): @@ -135,9 +155,10 @@ class Game: @property def namespace(self): - if not self.metadata: - return None - return self.metadata['namespace'] + """Namespace of public asset""" + if self.asset_infos: + return self._get_public_asset().namespace + return self.metadata.get('namespace') @classmethod def from_json(cls, json): @@ -146,14 +167,23 @@ class Game: app_title=json.get('app_title', ''), ) tmp.metadata = json.get('metadata', dict()) - if 'asset_infos' in json: - tmp.asset_infos = {k: GameAsset.from_json(v) for k, v in json['asset_infos'].items()} + if 'asset_infos_v2' in json: + tmp.asset_infos = {k: [GameAsset.from_json(a) for a in v] for k, v in json['asset_infos_v2'].items()} + elif 'asset_infos' in json: + # Migrate from one asset per platform to multiple + tmp.asset_infos = {k: [GameAsset.from_json(v)] for k, v in json['asset_infos'].items()} else: # Migrate old asset_info to new asset_infos - tmp.asset_infos['Windows'] = GameAsset.from_json(json.get('asset_info', dict())) + tmp.asset_infos['Windows'] = [GameAsset.from_json(json.get('asset_info', dict()))] + tmp.namespaces = json.get('namespaces', dict()) - if sidecar := json.get('sidecar', None): - tmp.sidecar = Sidecar.from_json(sidecar) + # We used to get json.get('sidecar', None) here, lets drop that field and instead + # use per namespace mapping + if sidecars := json.get('sidecars', None): + tmp.sidecars = {namespace: Sidecar.from_json(sidecar) for namespace, sidecar in sidecars.items()} + elif sidecar := json.get('sidecar', None): + ns = tmp._get_public_asset().namespace + tmp.sidecars = {ns: Sidecar.from_json(sidecar)} tmp.base_urls = json.get('base_urls', list()) return tmp @@ -161,10 +191,11 @@ class Game: @property def __dict__(self): """This is just here so asset_infos gets turned into a dict as well""" - assets_dictified = {k: v.__dict__ for k, v in self.asset_infos.items()} - sidecar_dictified = self.sidecar.__dict__ if self.sidecar else None - return dict(metadata=self.metadata, asset_infos=assets_dictified, app_name=self.app_name, - app_title=self.app_title, base_urls=self.base_urls, sidecar=sidecar_dictified) + assets_dictified = {k: [a.__dict__ for a in v] for k, v in self.asset_infos.items()} + sidecars_dictified = {k: s.__dict__ for k,s in self.sidecars.items()} if self.sidecars else None + return dict(metadata=self.metadata, asset_infos_v2=assets_dictified, app_name=self.app_name, + app_title=self.app_title, base_urls=self.base_urls, sidecars=sidecars_dictified, + namespaces=self.namespaces) @dataclass @@ -176,6 +207,7 @@ class InstalledGame: install_path: str title: str version: str + namespace: Optional[str] = None base_urls: List[str] = field(default_factory=list) can_run_offline: bool = False @@ -201,7 +233,7 @@ class InstalledGame: title=json.get('title', ''), version=json.get('version', ''), ) - + tmp.namespace = json.get('namespace', None) tmp.base_urls = json.get('base_urls', list()) tmp.executable = json.get('executable', '') tmp.launch_parameters = json.get('launch_parameters', '')