From 31530692ef8bda338f125881173535795d6efc11 Mon Sep 17 00:00:00 2001 From: derrod Date: Wed, 6 May 2020 15:40:04 +0200 Subject: [PATCH] [api/cli/core] Add extremely basic support for cloud saves Currently only supports downloading all saves to a folder, in the future it should support automatically extracting save files to the proper directory (at least on Windows). --- legendary/api/egs.py | 21 ++++++++++++++ legendary/cli.py | 44 +++++++++++++++++++++++++++-- legendary/core.py | 66 +++++++++++++++++++++++++++++++++++++++++++- 3 files changed, 127 insertions(+), 4 deletions(-) diff --git a/legendary/api/egs.py b/legendary/api/egs.py index 756ce6b..d876659 100644 --- a/legendary/api/egs.py +++ b/legendary/api/egs.py @@ -20,6 +20,7 @@ class EPCAPI: _entitlements_host = 'entitlement-public-service-prod08.ol.epicgames.com' _catalog_host = 'catalog-public-service-prod06.ol.epicgames.com' _ecommerce_host = 'ecommerceintegration-public-service-ecomprod02.ol.epicgames.com' + _datastorage_host = 'datastorage-public-service-liveegs.live.use1a.on.epicgames.com' def __init__(self): self.session = requests.session() @@ -117,3 +118,23 @@ class EPCAPI: country='US', locale='en')) r.raise_for_status() return r.json().get(catalog_item_id, None) + + 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 += '/' + + user_id = self.user.get('account_id') + + if filenames: + r = self.session.post(f'https://{self._datastorage_host}/api/v1/access/egstore/savesync/' + f'{user_id}/{app_name}', json=dict(files=filenames)) + else: + r = self.session.get(f'https://{self._datastorage_host}/api/v1/access/egstore/savesync/' + f'{user_id}/{app_name}') + r.raise_for_status() + return r.json() + + def create_game_cloud_saves(self, app_name, filenames): + return self.get_user_cloud_saves(app_name, filenames=filenames) diff --git a/legendary/cli.py b/legendary/cli.py index f7b61d6..b9f2a6e 100644 --- a/legendary/cli.py +++ b/legendary/cli.py @@ -194,6 +194,34 @@ class LegendaryCLI: # use the log output so this isn't included when piping file list into file logger.info(f'Install tags: {", ".join(sorted(install_tags))}') + def list_saves(self, args): + if not self.core.login(): + logger.error('Login failed! Cannot continue with download process.') + exit(1) + # update game metadata + logger.debug('Refreshing games list...') + _ = self.core.get_game_and_dlc_list(update_assets=True) + # then get the saves + logger.info('Getting list of saves...') + saves = self.core.get_save_games(args.app_name) + last_app = '' + print('Save games:') + for save in sorted(saves, key=lambda a: a.app_name): + if save.app_name != last_app: + game_title = self.core.get_game(save.app_name).app_title + last_app = save.app_name + print(f'- {game_title} ("{save.app_name}")') + print(' +', save.manifest_name) + + def download_saves(self, args): + if not self.core.login(): + logger.error('Login failed! Cannot continue with download process.') + exit(1) + # then get the saves + logger.info(f'Downloading saves to "{self.core.get_default_install_dir()}"') + # todo expand this to allow downloading single saves and extracting them to the correct directory + self.core.download_saves() + def launch_game(self, args, extra): app_name = args.app_name if not self.core.is_installed(app_name): @@ -435,11 +463,16 @@ def main(): list_parser = subparsers.add_parser('list-games', help='List available (installable) games') list_installed_parser = subparsers.add_parser('list-installed', help='List installed games') list_files_parser = subparsers.add_parser('list-files', help='List files in manifest') + list_saves_parser = subparsers.add_parser('list-saves', help='List available cloud saves') + download_saves_parser = subparsers.add_parser('download-saves', help='Download all cloud saves') install_parser.add_argument('app_name', help='Name of the app', metavar='') uninstall_parser.add_argument('app_name', help='Name of the app', metavar='') launch_parser.add_argument('app_name', help='Name of the app', metavar='') - list_files_parser.add_argument('app_name', nargs='?', help='Name of the app', metavar='') + list_files_parser.add_argument('app_name', nargs='?', metavar='', + help='Name of the app (optional)') + list_saves_parser.add_argument('app_name', nargs='?', metavar='', default='', + help='Name of the app (optional)') # importing only works on Windows right now if os.name == 'nt': @@ -526,14 +559,15 @@ def main(): exit(0) if args.subparser_name not in ('auth', 'list-games', 'list-installed', 'list-files', - 'launch', 'download', 'uninstall', 'install', 'update'): + 'launch', 'download', 'uninstall', 'install', 'update', + 'list-saves', 'download-saves'): print(parser.format_help()) # Print the main help *and* the help for all of the subcommands. Thanks stackoverflow! print('Individual command help:') subparsers = next(a for a in parser._actions if isinstance(a, argparse._SubParsersAction)) for choice, subparser in subparsers.choices.items(): - if choice in ('install', 'update'): + if choice in ('download', 'update'): continue print(f'\nCommand: {choice}') print(subparser.format_help()) @@ -566,6 +600,10 @@ def main(): cli.uninstall_game(args) elif args.subparser_name == 'list-files': cli.list_files(args) + elif args.subparser_name == 'list-saves': + cli.list_saves(args) + elif args.subparser_name == 'download-saves': + cli.download_saves(args) except KeyboardInterrupt: logger.info('Command was aborted via KeyboardInterrupt, cleaning up...') diff --git a/legendary/core.py b/legendary/core.py index 6a208aa..016aa12 100644 --- a/legendary/core.py +++ b/legendary/core.py @@ -8,7 +8,7 @@ import shlex import shutil from base64 import b64decode -from collections import defaultdict +from collections import defaultdict, namedtuple from datetime import datetime from multiprocessing import Queue from random import choice as randchoice @@ -25,6 +25,7 @@ from legendary.models.exceptions import * from legendary.models.game import * from legendary.models.json_manifest import JSONManifest from legendary.models.manifest import Manifest, ManifestMeta +from legendary.models.chunk import Chunk from legendary.utils.game_workarounds import is_opt_enabled @@ -272,6 +273,69 @@ class LegendaryCore: return params, working_dir, env + def get_save_games(self, app_name: str = ''): + # todo make this a proper class in legendary.models.egs or something + CloudSave = namedtuple('CloudSave', ['filename', 'app_name', 'manifest_name', 'iso_date']) + savegames = self.egs.get_user_cloud_saves(app_name) + _saves = [] + for fname, f in savegames['files'].items(): + if '.manifest' not in fname: + continue + f_parts = fname.split('/') + _saves.append(CloudSave(filename=fname, app_name=f_parts[2], + manifest_name=f_parts[4], iso_date=f['lastModified'])) + + return _saves + + def download_saves(self): + save_path = os.path.join(self.get_default_install_dir(), '.saves') + if not os.path.exists(save_path): + os.makedirs(save_path) + + savegames = self.egs.get_user_cloud_saves() + files = savegames['files'] + for fname, f in files.items(): + if '.manifest' not in fname: + continue + f_parts = fname.split('/') + save_dir = os.path.join(save_path, f'{f_parts[2]}/{f_parts[4].rpartition(".")[0]}') + if not os.path.exists(save_dir): + os.makedirs(save_dir) + + self.log.info(f'Downloading "{fname.split("/", 2)[2]}"...') + # download manifest + r = self.egs.unauth_session.get(f['readLink']) + if r.status_code != 200: + self.log.error(f'Download failed, status code: {r.status_code}') + continue + m = self.load_manfiest(r.content) + + # download chunks requierd for extraction + chunks = dict() + for chunk in m.chunk_data_list.elements: + cpath_p = fname.split('/', 3)[:3] + cpath_p.append(chunk.path) + cpath = '/'.join(cpath_p) + self.log.debug(f'Downloading chunk "{cpath}"') + r = self.egs.unauth_session.get(files[cpath]['readLink']) + if r.status_code != 200: + self.log.error(f'Download failed, status code: {r.status_code}') + break + c = Chunk.read_buffer(r.content) + chunks[c.guid_num] = c.data + + for fm in m.file_manifest_list.elements: + dirs, fname = os.path.split(fm.filename) + fdir = os.path.join(save_dir, dirs) + fpath = os.path.join(fdir, fname) + if not os.path.exists(fdir): + os.makedirs(fdir) + + self.log.debug(f'Writing "{fpath}"...') + with open(fpath, 'wb') as fh: + for cp in fm.chunk_parts: + fh.write(chunks[cp.guid_num][cp.offset:cp.offset+cp.size]) + def is_offline_game(self, app_name: str) -> bool: return self.lgd.config.getboolean(app_name, 'offline', fallback=False)