[cli/core/models/utils] Add basic cloud save syncing support

This commit is contained in:
derrod 2020-05-14 14:52:33 +02:00
parent 0df80773c0
commit 98df2a0a38
4 changed files with 252 additions and 21 deletions

View file

@ -17,6 +17,7 @@ from sys import exit, stdout
from legendary import __version__, __codename__
from legendary.core import LegendaryCore
from legendary.models.exceptions import InvalidCredentialsError
from legendary.models.game import SaveGameStatus
from legendary.utils.custom_parser import AliasedSubParsersAction
# todo custom formatter for cli logger (clean info, highlighted error/warning)
@ -223,10 +224,115 @@ class LegendaryCLI:
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()
self.core.download_saves(args.app_name)
def sync_saves(self, args):
if not self.core.login():
logger.error('Login failed! Cannot continue with download process.')
exit(1)
igames = self.core.get_installed_list()
if args.app_name:
igame = self.core.get_installed_game(args.app_name)
if not igame:
logger.fatal(f'Game not installed: {args.app_name}')
exit(1)
igames = [igame]
# check available saves
saves = self.core.get_save_games()
latest_save = dict()
for save in sorted(saves, key=lambda a: a.datetime):
latest_save[save.app_name] = save
logger.info(f'Got {len(latest_save)} remote save game(s)')
# evaluate current save state for each game.
for igame in igames:
if igame.app_name not in latest_save:
continue
game = self.core.get_game(igame.app_name)
if 'CloudSaveFolder' not in game.metadata['customAttributes']:
# this should never happen unless cloud save support was removed from a game
logger.warning(f'{igame.app_name} has remote save(s) but does not support cloud saves?!')
continue
# override save path only if app name is specified
if args.app_name and args.save_path:
logger.info(f'Overriding save path with "{args.save_path}"...')
igame.save_path = args.save_path
self.core.lgd.set_installed_game(igame.app_name, igame)
# if there is no saved save path, try to get one
if not igame.save_path:
save_path = self.core.get_save_path(igame.app_name)
# ask user if path is correct if computing for the first time
logger.info(f'Computed save path: "{save_path}"')
if '%' in save_path or '{' in save_path:
logger.warning('Path contains unprocessed variables, please enter the correct path manually.')
yn = 'n'
else:
yn = input('Is this correct? [Y/n] ')
if yn and yn.lower()[0] != 'y':
save_path = input('Please enter the correct path (leave empty to skip): ')
if not save_path:
logger.info('Empty input, skipping...')
continue
if not os.path.exists(save_path):
os.makedirs(save_path)
igame.save_path = save_path
self.core.lgd.set_installed_game(igame.app_name, igame)
# check if *any* file in the save game directory is newer than the latest uploaded save
res, (dt_l, dt_r) = self.core.check_savegame_state(igame.save_path, latest_save[igame.app_name])
if res == SaveGameStatus.SAME_AGE and not (args.force_upload or args.force_download):
logger.info(f'Save game for "{igame.title}" is up to date, skipping...')
continue
if (res == SaveGameStatus.REMOTE_NEWER and not args.force_upload) or args.force_download:
if res == SaveGameStatus.REMOTE_NEWER: # only print this info if not forced
logger.info(f'Cloud save for "{igame.title}" is newer:')
logger.info(f'- Cloud save date: {dt_l.strftime("%Y-%m-%d %H:%M:%S")}')
logger.info(f'- Local save date: {dt_r.strftime("%Y-%m-%d %H:%M:%S")}')
if args.upload_only:
logger.info('Save game downloading is disabled, skipping...')
continue
if not args.yes and not args.force_download:
choice = input(f'Download cloud save? [Y/n]: ')
if choice and choice.lower()[0] != 'y':
logger.info('Not downloading...')
continue
logger.info('Downloading remote savegame...')
self.core.download_saves(igame.app_name, save_dir=igame.save_path, clean_dir=True,
manifest_name=latest_save[igame.app_name].manifest_name)
elif res == SaveGameStatus.LOCAL_NEWER or args.force_upload:
if res == SaveGameStatus.LOCAL_NEWER:
logger.info(f'Local save for "{igame.title}" is newer')
logger.info(f'- Cloud save date: {dt_l.strftime("%Y-%m-%d %H:%M:%S")}')
logger.info(f'- Local save date: {dt_r.strftime("%Y-%m-%d %H:%M:%S")}')
if args.download_only:
logger.info('Save game uploading is disabled, skipping...')
continue
if not args.yes and not args.force_upload:
choice = input(f'Upload local save? [Y/n]: ')
if choice and choice.lower()[0] != 'y':
logger.info('Not uploading...')
continue
logger.info('Uploading local savegame...')
self.core.upload_save(igame.app_name, igame.save_path, dt_l)
def launch_game(self, args, extra):
app_name = args.app_name
@ -472,6 +578,7 @@ def main():
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')
sync_saves_parser = subparsers.add_parser('sync-saves', help='Sync cloud saves')
install_parser.add_argument('app_name', help='Name of the app', metavar='<App Name>')
uninstall_parser.add_argument('app_name', help='Name of the app', metavar='<App Name>')
@ -480,6 +587,10 @@ def main():
help='Name of the app (optional)')
list_saves_parser.add_argument('app_name', nargs='?', metavar='<App Name>', default='',
help='Name of the app (optional)')
download_saves_parser.add_argument('app_name', nargs='?', metavar='<App Name>', default='',
help='Name of the app (optional)')
sync_saves_parser.add_argument('app_name', nargs='?', metavar='<App Name>', default='',
help='Name of the app (optional)')
# importing only works on Windows right now
if os.name == 'nt':
@ -570,6 +681,17 @@ def main():
list_files_parser.add_argument('--install-tag', dest='install_tag', action='store', metavar='<tag>',
type=str, help='Show only files with specified install tag')
sync_saves_parser.add_argument('--skip-upload', dest='download_only', action='store_true',
help='Only download new saves from cloud, don\'t upload')
sync_saves_parser.add_argument('--skip-download', dest='upload_only', action='store_true',
help='Only upload new saves from cloud, don\'t download')
sync_saves_parser.add_argument('--force-upload', dest='force_upload', action='store_true',
help='Force upload even if local saves are older')
sync_saves_parser.add_argument('--force-download', dest='force_download', action='store_true',
help='Force download even if local saves are newer')
sync_saves_parser.add_argument('--save-path', dest='save_path', action='store',
help='Override savegame path (only if app name is specified)')
args, extra = parser.parse_known_args()
if args.version:
@ -578,7 +700,7 @@ def main():
if args.subparser_name not in ('auth', 'list-games', 'list-installed', 'list-files',
'launch', 'download', 'uninstall', 'install', 'update',
'list-saves', 'download-saves'):
'list-saves', 'download-saves', 'sync-saves'):
print(parser.format_help())
# Print the main help *and* the help for all of the subcommands. Thanks stackoverflow!
@ -622,6 +744,8 @@ def main():
cli.list_saves(args)
elif args.subparser_name == 'download-saves':
cli.download_saves(args)
elif args.subparser_name == 'sync-saves':
cli.sync_saves(args)
except KeyboardInterrupt:
logger.info('Command was aborted via KeyboardInterrupt, cleaning up...')

View file

@ -8,8 +8,8 @@ import shlex
import shutil
from base64 import b64decode
from collections import defaultdict, namedtuple
from datetime import datetime
from collections import defaultdict
from datetime import datetime, timezone
from multiprocessing import Queue
from random import choice as randchoice
from requests.exceptions import HTTPError
@ -27,6 +27,7 @@ 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
from legendary.utils.savegame_helper import SaveGameHelper
# ToDo: instead of true/false return values for success/failure actually raise an exception that the CLI/GUI
@ -45,6 +46,8 @@ class LegendaryCore:
self.egs = EPCAPI()
self.lgd = LGDLFS()
self.local_timezone = datetime.now().astimezone().tzinfo
# epic lfs only works on Windows right now
if os.name == 'nt':
self.egl = EPCLFS()
@ -279,34 +282,108 @@ 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)
savegames = self.egs.get_user_cloud_saves(app_name, manifests=not not 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']))
_saves.append(SaveGameFile(app_name=f_parts[2], filename=fname, manifest=f_parts[4],
datetime=datetime.fromisoformat(f['lastModified'][:-1])))
return _saves
def download_saves(self):
def get_save_path(self, app_name, wine_prefix='~/.wine'):
game = self.lgd.get_game_meta(app_name)
save_path = game.metadata['customAttributes'].get('CloudSaveFolder', {}).get('value')
if not save_path:
raise ValueError('Game does not support cloud saves')
igame = self.lgd.get_installed_game(app_name)
if not igame:
raise ValueError('Game is not installed!')
# the following variables are known:
path_vars = {
'{appdata}': os.path.expandvars('%APPDATA%'),
'{installdir}': igame.install_path,
'{userdir}': os.path.expandvars('%userprofile%/documents'),
'{epicid}': self.lgd.userdata['account_id']
}
# the following variables are in the EGL binary but are not used by any of
# my games and I'm not sure where they actually point at:
# {UserProfile} (Probably %USERPROFILE%)
# {UserSavedGames}
# these paths should always use a forward slash
new_save_path = [path_vars.get(p.lower(), p) for p in save_path.split('/')]
return os.path.join(*new_save_path)
def check_savegame_state(self, path: str, save: SaveGameFile) -> (SaveGameStatus, (datetime, datetime)):
latest = 0
for _dir, _, _files in os.walk(path):
for _file in _files:
s = os.stat(os.path.join(_dir, _file))
latest = max(latest, s.st_mtime)
# timezones are fun!
dt_local = datetime.fromtimestamp(latest).replace(tzinfo=self.local_timezone).astimezone(timezone.utc)
dt_remote = datetime.strptime(save.manifest_name, '%Y.%m.%d-%H.%M.%S.manifest').replace(tzinfo=timezone.utc)
self.log.debug(f'Local save date: {str(dt_local)}, Remote save date: {str(dt_remote)}')
# Ideally we check the files themselves based on manifest,
# this is mostly a guess but should be accurate enough.
if abs((dt_local - dt_remote).total_seconds()) < 60:
return SaveGameStatus.SAME_AGE, (dt_local, dt_remote)
elif dt_local > dt_remote:
return SaveGameStatus.LOCAL_NEWER, (dt_local, dt_remote)
else:
return SaveGameStatus.REMOTE_NEWER, (dt_local, dt_remote)
def upload_save(self, app_name, save_dir, local_dt: datetime = None):
game = self.lgd.get_game_meta(app_name)
save_path = game.metadata['customAttributes'].get('CloudSaveFolder', {}).get('value')
if not save_path:
raise ValueError('Game does not support cloud saves')
sgh = SaveGameHelper()
files = sgh.package_savegame(save_dir, app_name, self.egs.user.get('account_id'),
save_path, local_dt)
self.log.debug(f'Packed files: {str(files)}, creating cloud files...')
resp = self.egs.create_game_cloud_saves(app_name, list(files.keys()))
self.log.info('Starting upload...')
for remote_path, file_info in resp['files'].items():
self.log.debug(f'Uploading "{remote_path}"')
f = files.get(remote_path)
self.egs.unauth_session.put(file_info['writeLink'], data=f.read())
self.log.info('Finished uploading savegame.')
def download_saves(self, app_name='', manifest_name='', save_dir='', clean_dir=False):
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()
savegames = self.egs.get_user_cloud_saves(app_name=app_name)
files = savegames['files']
for fname, f in files.items():
if '.manifest' not in fname:
continue
f_parts = fname.split('/')
if manifest_name and f_parts[4] != manifest_name:
continue
if not save_dir:
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)
if clean_dir:
self.log.info('Deleting old save files...')
delete_folder(save_dir)
self.log.info(f'Downloading "{fname.split("/", 2)[2]}"...')
# download manifest
r = self.egs.unauth_session.get(f['readLink'])
@ -315,7 +392,7 @@ class LegendaryCore:
continue
m = self.load_manfiest(r.content)
# download chunks requierd for extraction
# download chunks required for extraction
chunks = dict()
for chunk in m.chunk_data_list.elements:
cpath_p = fname.split('/', 3)[:3]
@ -341,6 +418,13 @@ class LegendaryCore:
for cp in fm.chunk_parts:
fh.write(chunks[cp.guid_num][cp.offset:cp.offset+cp.size])
# set modified time to savegame creation timestamp
m_date = datetime.strptime(f_parts[4], '%Y.%m.%d-%H.%M.%S.manifest')
m_date = m_date.replace(tzinfo=timezone.utc).astimezone(self.local_timezone)
os.utime(fpath, (m_date.timestamp(), m_date.timestamp()))
self.log.info('Successfully completed savegame download.')
def is_offline_game(self, app_name: str) -> bool:
return self.lgd.config.getboolean(app_name, 'offline', fallback=False)

View file

@ -1,6 +1,8 @@
#!/usr/bin/env python
# coding: utf-8
from enum import Enum
class GameAsset:
def __init__(self):
@ -73,7 +75,7 @@ class Game:
class InstalledGame:
def __init__(self, app_name='', title='', version='', manifest_path='', base_urls=None,
install_path='', executable='', launch_parameters='', prereq_info=None,
can_run_offline=False, requires_ot=False, is_dlc=False):
can_run_offline=False, requires_ot=False, is_dlc=False, save_path=None):
self.app_name = app_name
self.title = title
self.version = version
@ -87,6 +89,7 @@ class InstalledGame:
self.can_run_offline = can_run_offline
self.requires_ot = requires_ot
self.is_dlc = is_dlc
self.save_path = save_path
@classmethod
def from_json(cls, json):
@ -105,4 +108,19 @@ class InstalledGame:
tmp.can_run_offline = json.get('can_run_offline', False)
tmp.requires_ot = json.get('requires_ot', False)
tmp.is_dlc = json.get('is_dlc', False)
tmp.save_path = json.get('save_path', None)
return tmp
class SaveGameFile:
def __init__(self, app_name='', filename='', manifest='', datetime=None):
self.app_name = app_name
self.filename = filename
self.manifest_name = manifest
self.datetime = datetime
class SaveGameStatus(Enum):
LOCAL_NEWER = 1
REMOTE_NEWER = -1
SAME_AGE = 0

View file

@ -1,6 +1,7 @@
import logging
import os
import time
from datetime import datetime
from hashlib import sha1
from io import BytesIO
from tempfile import TemporaryFile
@ -30,12 +31,14 @@ class SaveGameHelper:
return ci
def package_savegame(self, input_folder: str, app_name: str = '',
epic_id: str = '', cloud_folder: str = ''):
epic_id: str = '', cloud_folder: str = '',
manifest_dt: datetime = None):
"""
:param input_folder: Folder to be packaged into chunks/manifest
:param app_name: App name for savegame being stored
:param epic_id: Epic account ID
:param cloud_folder: Folder the savegame resides in (based on game metadata)
:param manifest_dt: datetime for the manifest name (optional)
:return:
"""
m = Manifest()
@ -45,7 +48,9 @@ class SaveGameHelper:
m.custom_fields = CustomFields()
# create metadata for savegame
m.meta.app_name = f'{app_name}{epic_id}'
m.meta.build_version = time.strftime('%Y.%m.%d-%H.%M.%S')
if not manifest_dt:
manifest_dt = datetime.utcnow()
m.meta.build_version = manifest_dt.strftime('%Y.%m.%d-%H.%M.%S')
m.custom_fields['CloudSaveFolder'] = cloud_folder
self.log.info(f'Packing savegame for "{app_name}", input folder: {input_folder}')