228 lines
12 KiB
Python
228 lines
12 KiB
Python
import os
|
|
|
|
from multiprocessing import Queue
|
|
from typing import Callable
|
|
|
|
from legendary.utils.lfs import validate_files
|
|
from legendary.models.downloading import AnalysisResult
|
|
from legendary.models.game import *
|
|
from legendary.utils.selective_dl import get_sdl_appname
|
|
|
|
from legendary.core import LegendaryCore as LegendaryCoreReal
|
|
from .manager import DLManager
|
|
|
|
|
|
class LegendaryCore(LegendaryCoreReal):
|
|
|
|
def prepare_download(self, app_name: str, base_path: str = '',
|
|
status_q: Queue = None, max_shm: int = 0, max_workers: int = 0,
|
|
force: bool = False, disable_patching: bool = False,
|
|
game_folder: str = '', override_manifest: str = '',
|
|
override_old_manifest: str = '', override_base_url: str = '',
|
|
platform: str = 'Windows', file_prefix_filter: list = None,
|
|
file_exclude_filter: list = None, file_install_tag: list = None,
|
|
dl_optimizations: bool = False, dl_timeout: int = 10,
|
|
repair: bool = False, repair_use_latest: bool = False,
|
|
disable_delta: bool = False, override_delta_manifest: str = '',
|
|
egl_guid: str = '', preferred_cdn: str = None,
|
|
no_install: bool = False, ignore_space_req: bool = False,
|
|
disable_sdl: bool = False, reset_sdl: bool = False, skip_sdl: bool = False,
|
|
sdl_prompt: Callable[[str, str], List[str]] = list,
|
|
disable_https: bool = False) -> (DLManager, AnalysisResult, InstalledGame, Game):
|
|
if self.is_installed(app_name):
|
|
igame = self.get_installed_game(app_name)
|
|
platform = igame.platform
|
|
if igame.needs_verification and not repair:
|
|
self.log.info('Game needs to be verified before updating, switching to repair mode...')
|
|
repair = True
|
|
|
|
repair_file = None
|
|
if repair:
|
|
repair = True
|
|
no_install = repair_use_latest is False
|
|
repair_file = os.path.join(self.lgd.get_tmp_path(), f'{app_name}.repair')
|
|
|
|
if not self.login():
|
|
raise RuntimeError('Login failed! Cannot continue with download process.')
|
|
|
|
if file_prefix_filter or file_exclude_filter or file_install_tag:
|
|
no_install = True
|
|
|
|
game = self.get_game(app_name, update_meta=True)
|
|
|
|
if not game:
|
|
raise RuntimeError(f'Could not find "{app_name}" in list of available games,'
|
|
f'did you type the name correctly?')
|
|
|
|
if game.is_dlc:
|
|
self.log.info('Install candidate is DLC')
|
|
app_name = game.metadata['mainGameItem']['releaseInfo'][0]['appId']
|
|
base_game = self.get_game(app_name)
|
|
# check if base_game is actually installed
|
|
if not self.is_installed(app_name):
|
|
# download mode doesn't care about whether or not something's installed
|
|
if not no_install:
|
|
raise RuntimeError(f'Base game "{app_name}" is not installed!')
|
|
else:
|
|
base_game = None
|
|
|
|
if repair:
|
|
if not self.is_installed(game.app_name):
|
|
raise RuntimeError(f'Game "{game.app_title}" ({game.app_name}) is not installed!')
|
|
|
|
if not os.path.exists(repair_file):
|
|
self.log.info('Verifing game...')
|
|
self.verify_game(app_name)
|
|
else:
|
|
self.log.info(f'Using existing repair file: {repair_file}')
|
|
|
|
# check if SDL should be disabled
|
|
sdl_enabled = not file_install_tag and not game.is_dlc
|
|
config_tags = self.lgd.config.get(game.app_name, 'install_tags', fallback=None)
|
|
config_disable_sdl = self.lgd.config.getboolean(game.app_name, 'disable_sdl', fallback=False)
|
|
# remove config flag if SDL is reset
|
|
if config_disable_sdl and reset_sdl and not disable_sdl:
|
|
self.lgd.config.remove_option(game.app_name, 'disable_sdl')
|
|
# if config flag is not yet set, set it and remove previous install tags
|
|
elif not config_disable_sdl and disable_sdl:
|
|
self.log.info('Clearing install tags from config and disabling SDL for title.')
|
|
if config_tags:
|
|
self.lgd.config.remove_option(game.app_name, 'install_tags')
|
|
config_tags = None
|
|
self.lgd.config.set(game.app_name, 'disable_sdl', True)
|
|
sdl_enabled = False
|
|
# just disable SDL, but keep config tags that have been manually specified
|
|
elif config_disable_sdl or disable_sdl:
|
|
sdl_enabled = False
|
|
|
|
if sdl_enabled and ((sdl_name := get_sdl_appname(game.app_name)) is not None):
|
|
if not self.is_installed(game.app_name) or config_tags is None or reset_sdl:
|
|
sdl_data = self.get_sdl_data(sdl_name, platform=platform)
|
|
if sdl_data:
|
|
if skip_sdl:
|
|
file_install_tag = ['']
|
|
if '__required' in sdl_data:
|
|
file_install_tag.extend(sdl_data['__required']['tags'])
|
|
else:
|
|
file_install_tag = sdl_prompt(sdl_data, game.app_title)
|
|
self.lgd.config.set(game.app_name, 'install_tags', ','.join(file_install_tag))
|
|
else:
|
|
self.log.error(f'Unable to get SDL data for {sdl_name}')
|
|
else:
|
|
file_install_tag = config_tags.split(',')
|
|
elif file_install_tag and not game.is_dlc and not no_install:
|
|
config_tags = ','.join(file_install_tag)
|
|
self.log.info(f'Saving install tags for "{game.app_name}" to config: {config_tags}')
|
|
self.lgd.config.set(game.app_name, 'install_tags', config_tags)
|
|
elif not game.is_dlc:
|
|
if config_tags and reset_sdl:
|
|
self.log.info('Clearing install tags from config.')
|
|
self.lgd.config.remove_option(game.app_name, 'install_tags')
|
|
elif config_tags:
|
|
self.log.info(f'Using install tags from config: {config_tags}')
|
|
file_install_tag = config_tags.split(',')
|
|
|
|
dlm, analysis, igame = super(LegendaryCore, self).prepare_download(game=game, base_game=base_game, base_path=base_path,
|
|
force=force, max_shm=max_shm,
|
|
max_workers=max_workers, game_folder=game_folder,
|
|
disable_patching=disable_patching,
|
|
override_manifest=override_manifest,
|
|
override_old_manifest=override_old_manifest,
|
|
override_base_url=override_base_url,
|
|
platform=platform,
|
|
file_prefix_filter=file_prefix_filter,
|
|
file_exclude_filter=file_exclude_filter,
|
|
file_install_tag=file_install_tag,
|
|
dl_optimizations=dl_optimizations,
|
|
dl_timeout=dl_timeout,
|
|
repair=repair,
|
|
repair_use_latest=repair_use_latest,
|
|
disable_delta=disable_delta,
|
|
override_delta_manifest=override_delta_manifest,
|
|
preferred_cdn=preferred_cdn,
|
|
disable_https=disable_https)
|
|
|
|
# game is either up to date or hasn't changed, so we have nothing to do
|
|
if not analysis.dl_size:
|
|
self.log.info('Download size is 0, the game is either already up to date or has not changed. Exiting...')
|
|
self.clean_post_install(game, igame, repair, repair_file)
|
|
|
|
raise RuntimeError('Nothing to do.')
|
|
|
|
res = self.check_installation_conditions(analysis=analysis, install=igame, game=game,
|
|
updating=self.is_installed(app_name),
|
|
ignore_space_req=ignore_space_req)
|
|
|
|
return dlm, analysis, igame, game, repair, repair_file, res
|
|
|
|
def verify_game(self, app_name: str, callback: Callable[[int, int], None] = print):
|
|
if not self.is_installed(app_name):
|
|
self.log.error(f'Game "{app_name}" is not installed')
|
|
return
|
|
|
|
self.log.info(f'Loading installed manifest for "{app_name}"')
|
|
igame = self.get_installed_game(app_name)
|
|
manifest_data, _ = self.get_installed_manifest(app_name)
|
|
manifest = self.load_manifest(manifest_data)
|
|
|
|
files = sorted(manifest.file_manifest_list.elements,
|
|
key=lambda a: a.filename.lower())
|
|
|
|
# build list of hashes
|
|
file_list = [(f.filename, f.sha_hash.hex()) for f in files]
|
|
total = len(file_list)
|
|
num = 0
|
|
failed = []
|
|
missing = []
|
|
|
|
self.log.info(f'Verifying "{igame.title}" version "{manifest.meta.build_version}"')
|
|
repair_file = []
|
|
for result, path, result_hash in validate_files(igame.install_path, file_list):
|
|
if callback:
|
|
num += 1
|
|
callback(num, total)
|
|
|
|
if result == VerifyResult.HASH_MATCH:
|
|
repair_file.append(f'{result_hash}:{path}')
|
|
continue
|
|
elif result == VerifyResult.HASH_MISMATCH:
|
|
self.log.error(f'File does not match hash: "{path}"')
|
|
repair_file.append(f'{result_hash}:{path}')
|
|
failed.append(path)
|
|
elif result == VerifyResult.FILE_MISSING:
|
|
self.log.error(f'File is missing: "{path}"')
|
|
missing.append(path)
|
|
else:
|
|
self.log.error(f'Other failure (see log), treating file as missing: "{path}"')
|
|
missing.append(path)
|
|
|
|
# always write repair file, even if all match
|
|
if repair_file:
|
|
repair_filename = os.path.join(self.lgd.get_tmp_path(), f'{app_name}.repair')
|
|
with open(repair_filename, 'w') as f:
|
|
f.write('\n'.join(repair_file))
|
|
self.log.debug(f'Written repair file to "{repair_filename}"')
|
|
|
|
if not missing and not failed:
|
|
self.log.info('Verification finished successfully.')
|
|
else:
|
|
raise RuntimeError(
|
|
f'Verification failed, {len(failed)} file(s) corrupted, {len(missing)} file(s) are missing.')
|
|
|
|
def clean_post_install(self, game: Game, igame: InstalledGame, repair: bool = False, repair_file: str = ''):
|
|
old_igame = self.get_installed_game(game.app_name)
|
|
if old_igame and repair and os.path.exists(repair_file):
|
|
if old_igame.needs_verification:
|
|
old_igame.needs_verification = False
|
|
self.install_game(old_igame)
|
|
|
|
self.log.debug('Removing repair file.')
|
|
os.remove(repair_file)
|
|
|
|
# check if install tags have changed, if they did; try deleting files that are no longer required.
|
|
if old_igame and old_igame.install_tags != igame.install_tags:
|
|
old_igame.install_tags = igame.install_tags
|
|
self.log.info('Deleting now untagged files.')
|
|
self.uninstall_tag(old_igame)
|
|
self.install_game(old_igame)
|