1
0
Fork 0
mirror of synced 2024-06-02 18:54:41 +12:00

Lgndr: Add logger monkey class

The class acts as an intermediate between the logger and the function call
It keeps the last message that was sent to the logger. The instance of the
class can be returned as a return value from the LegendaryCLI methods to
provide return status and the message related to it.

The level at which it considers the logged message as an error is configurable.
By default it considers logging.ERROR and above as faulty return values
This commit is contained in:
loathingKernel 2022-07-20 23:33:44 +03:00
parent bbaff5f42c
commit 28531eec38
6 changed files with 215 additions and 75 deletions

View file

@ -87,10 +87,10 @@ class ImportWorker(QRunnable):
if app_name or (app_name := find_app_name(str(path), self.core)):
result.app_name = app_name
app_title = self.core.get_game(app_name).app_title
err = self.__import_game(path, app_name, app_title)
if err:
success, message = self.__import_game(path, app_name, app_title)
if not success:
result.result = ImportResult.FAILED
result.message = f"{app_title} - {err}"
result.message = f"{app_title} - {message}"
else:
result.result = ImportResult.SUCCESS
result.message = self.tr("{} - Imported successfully").format(app_title)
@ -105,11 +105,7 @@ class ImportWorker(QRunnable):
app_name=app_name,
get_boolean_choice=lambda prompt, default=True: self.import_dlcs
)
try:
cli.import_game(args)
return ""
except LgndrException as ret:
return ret.message
return cli.import_game(args)
class AppNameCompleter(QCompleter):

View file

@ -22,3 +22,57 @@ class UILogHandler(logging.Handler):
def emit(self, record: logging.LogRecord) -> None:
self.widget.setText(record.getMessage())
class LgndrReturnLogger:
def __init__(self, logger: logging.Logger, level: int = logging.ERROR):
self.logger = logger
self.level = level
self.value = False
self.message = ""
def _log(self, level: int, msg: str):
self.value = True if level < self.level else False
self.message = msg
self.logger.log(level, msg)
def debug(self, msg: str):
self._log(logging.DEBUG, msg)
def info(self, msg: str):
self._log(logging.INFO, msg)
def warning(self, msg: str):
self._log(logging.WARNING, msg)
def error(self, msg: str):
self._log(logging.ERROR, msg)
def critical(self, msg: str):
self._log(logging.CRITICAL, msg)
def fatal(self, msg: str):
self.critical(msg)
def __len__(self):
if self.message:
return 2
else:
return 0
def __bool__(self):
return self.value
def __getitem__(self, item):
if item == 0:
return self.value
elif item == 1:
return self.message
else:
raise IndexError
def __iter__(self):
return iter((self.value, self.message))
def __str__(self):
return self.message

View file

@ -4,8 +4,7 @@ import logging
import time
from typing import Optional, Union
import legendary.cli
from legendary.cli import logger
from legendary.cli import LegendaryCLI as LegendaryCLIReal
from legendary.models.downloading import AnalysisResult, ConditionCheckResult
from legendary.models.game import Game, InstalledGame, VerifyResult
from legendary.utils.lfs import validate_files
@ -14,14 +13,12 @@ from legendary.utils.selective_dl import get_sdl_appname
from .core import LegendaryCore
from .manager import DLManager
from .api_arguments import LgndrInstallGameArgs, LgndrImportGameArgs, LgndrVerifyGameArgs, LgndrUninstallGameArgs
from .api_exception import LgndrException, LgndrCLILogHandler
from .api_monkeys import return_exit as exit, get_boolean_choice
legendary.cli.exit = exit
from .api_monkeys import return_exit, get_boolean_choice, LgndrReturnLogger
class LegendaryCLI(legendary.cli.LegendaryCLI):
class LegendaryCLI(LegendaryCLIReal):
"""
@staticmethod
def apply_wrap(func):
@ -30,33 +27,35 @@ class LegendaryCLI(legendary.cli.LegendaryCLI):
old_exit = legendary.cli.exit
legendary.cli.exit = exit
old_choice = legendary.cli.get_boolean_choice
if hasattr(args, 'get_boolean_choice') and args.get_boolean_choice is not None:
# g['get_boolean_choice'] = args.get_boolean_choice
legendary.cli.get_boolean_choice = args.get_boolean_choice
# old_choice = legendary.cli.get_boolean_choice
# if hasattr(args, 'get_boolean_choice') and args.get_boolean_choice is not None:
# legendary.cli.get_boolean_choice = args.get_boolean_choice
try:
return func(self, args, *oargs, **kwargs)
except LgndrException as ret:
raise ret
finally:
# g['get_boolean_choice'] = old_choice
legendary.cli.get_boolean_choice = old_choice
# legendary.cli.get_boolean_choice = old_choice
legendary.cli.exit = old_exit
return inner
"""
def __init__(self, override_config=None, api_timeout=None):
self.core = LegendaryCore(override_config, timeout=api_timeout)
self.logger = logging.getLogger('cli')
self.logging_queue = None
self.handler = LgndrCLILogHandler()
self.logger.addHandler(self.handler)
# self.handler = LgndrCLILogHandler()
# self.logger.addHandler(self.handler)
def resolve_aliases(self, name):
return super(LegendaryCLI, self)._resolve_aliases(name)
def install_game(self, args: LgndrInstallGameArgs) -> (DLManager, AnalysisResult, InstalledGame, Game, bool, Optional[str], ConditionCheckResult):
# Override logger for the local context to use message as part of the return value
logger = LgndrReturnLogger(self.logger)
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)
@ -80,6 +79,7 @@ class LegendaryCLI(legendary.cli.LegendaryCLI):
if not game:
logger.error(f'Could not find "{args.app_name}" in list of available games, '
f'did you type the name correctly?')
return logger
if args.platform not in game.asset_infos:
if not args.no_install:
@ -90,7 +90,7 @@ class LegendaryCLI(legendary.cli.LegendaryCLI):
logger.error(f'No app asset found for platform "{args.platform}", run '
f'"legendary info --platform {args.platform}" and make '
f'sure the app is available for the specified platform.')
return
return logger
else:
logger.warning(f'No asset found for platform "{args.platform}", '
f'trying anyway since --no-install is set.')
@ -115,14 +115,14 @@ class LegendaryCLI(legendary.cli.LegendaryCLI):
logger.info('Game has not been verified yet.')
if not args.yes:
if not args.get_boolean_choice(f'Verify "{game.app_name}" now ("no" will abort repair)?'):
return
return logger
try:
self.verify_game(args, print_command=False, repair_mode=True, repair_online=args.repair_and_update)
except ValueError:
logger.error('To repair a game with a missing manifest you must run the command with '
'"--repair-and-update". However this will redownload any file that does '
'not match the current hash in its entirety.')
return
return logger
else:
logger.info(f'Using existing repair file: {repair_file}')
@ -210,6 +210,9 @@ class LegendaryCLI(legendary.cli.LegendaryCLI):
return dlm, analysis, igame, game, args.repair_mode, repair_file, res
def clean_post_install(self, game: Game, igame: InstalledGame, repair: bool = False, repair_file: str = ''):
# Override logger for the local context to use message as part of the return value
logger = LgndrReturnLogger(self.logger)
old_igame = self.core.get_installed_game(game.app_name)
if old_igame and repair and os.path.exists(repair_file):
if old_igame.needs_verification:
@ -226,20 +229,24 @@ class LegendaryCLI(legendary.cli.LegendaryCLI):
self.core.uninstall_tag(old_igame)
self.core.install_game(old_igame)
@apply_wrap
return logger
def handle_postinstall(self, postinstall, igame, yes=False):
super(LegendaryCLI, self)._handle_postinstall(postinstall, igame, yes)
def uninstall_game(self, args: LgndrUninstallGameArgs):
# Override logger for the local context to use message as part of the return value
logger = LgndrReturnLogger(self.logger, level=logging.WARNING)
args.app_name = self._resolve_aliases(args.app_name)
igame = self.core.get_installed_game(args.app_name)
if not igame:
logger.error(f'Game {args.app_name} not installed, cannot uninstall!')
exit(0)
return logger
if not args.yes:
if not get_boolean_choice(f'Do you wish to uninstall "{igame.title}"?', default=False):
return False
if not args.get_boolean_choice(f'Do you wish to uninstall "{igame.title}"?', default=False):
return logger
try:
if not igame.is_dlc:
@ -254,16 +261,19 @@ class LegendaryCLI(legendary.cli.LegendaryCLI):
self.core.uninstall_game(igame, delete_files=not args.keep_files,
delete_root_directory=not igame.is_dlc)
logger.info('Game has been uninstalled.')
return True
return logger
except Exception as e:
logger.warning(f'Removing game failed: {e!r}, please remove {igame.install_path} manually.')
return False
return logger
def verify_game(self, args: Union[LgndrVerifyGameArgs, LgndrInstallGameArgs], print_command=True, repair_mode=False, repair_online=False):
# Override logger for the local context to use message as part of the return value
logger = LgndrReturnLogger(self.logger)
args.app_name = self._resolve_aliases(args.app_name)
if not self.core.is_installed(args.app_name):
logger.error(f'Game "{args.app_name}" is not installed')
return
return logger
logger.info(f'Loading installed manifest for "{args.app_name}"')
igame = self.core.get_installed_game(args.app_name)
@ -271,7 +281,7 @@ class LegendaryCLI(legendary.cli.LegendaryCLI):
logger.error(f'Install path "{igame.install_path}" does not exist, make sure all necessary mounts '
f'are available. If you previously deleted the game folder without uninstalling, run '
f'"legendary uninstall -y {igame.app_name}" and reinstall from scratch.')
return
return logger
manifest_data, _ = self.core.get_installed_manifest(args.app_name)
if manifest_data is None:
@ -291,7 +301,7 @@ class LegendaryCLI(legendary.cli.LegendaryCLI):
logger.critical(f'Manifest appears to be missing! To repair, run "legendary repair '
f'{args.app_name} --repair-and-update", this will however redownload all files '
f'that do not match the latest manifest in their entirety.')
return
return logger
manifest = self.core.load_manifest(manifest_data)
@ -339,14 +349,14 @@ class LegendaryCLI(legendary.cli.LegendaryCLI):
repair_file.append(f'{result_hash}:{path}')
continue
elif result == VerifyResult.HASH_MISMATCH:
logger.info(f'File does not match hash: "{path}"')
logger.error(f'File does not match hash: "{path}"')
repair_file.append(f'{result_hash}:{path}')
failed.append(path)
elif result == VerifyResult.FILE_MISSING:
logger.info(f'File is missing: "{path}"')
logger.error(f'File is missing: "{path}"')
missing.append(path)
else:
logger.info(f'Other failure (see log), treating file as missing: "{path}"')
logger.error(f'Other failure (see log), treating file as missing: "{path}"')
missing.append(path)
if args.verify_stdout:
@ -361,13 +371,106 @@ class LegendaryCLI(legendary.cli.LegendaryCLI):
if not missing and not failed:
logger.info('Verification finished successfully.')
return True, 0, 0
return logger, 0, 0
else:
logger.warning(f'Verification failed, {len(failed)} file(s) corrupted, {len(missing)} file(s) are missing.')
if print_command:
logger.info(f'Run "legendary repair {args.app_name}" to repair your game installation.')
return False, len(failed), len(missing)
return logger, len(failed), len(missing)
@apply_wrap
def import_game(self, args: LgndrImportGameArgs):
super(LegendaryCLI, self).import_game(args)
# Override logger for the local context to use message as part of the return value
logger = LgndrReturnLogger(self.logger)
# make sure path is absolute
args.app_path = os.path.abspath(args.app_path)
args.app_name = self._resolve_aliases(args.app_name)
if not os.path.exists(args.app_path):
logger.error(f'Specified path "{args.app_path}" does not exist!')
return logger
if self.core.is_installed(args.app_name):
logger.error('Game is already installed!')
return logger
if not self.core.login():
logger.error('Log in failed!')
return logger
# do some basic checks
game = self.core.get_game(args.app_name, update_meta=True, platform=args.platform)
if not game:
logger.fatal(f'Did not find game "{args.app_name}" on account.')
return logger
if game.is_dlc:
release_info = game.metadata.get('mainGameItem', {}).get('releaseInfo')
if release_info:
main_game_appname = release_info[0]['appId']
main_game_title = game.metadata['mainGameItem']['title']
if not self.core.is_installed(main_game_appname):
logger.error(f'Import candidate is DLC but base game "{main_game_title}" '
f'(App name: "{main_game_appname}") is not installed!')
return logger
else:
logger.fatal(f'Unable to get base game information for DLC, cannot continue.')
return logger
# get everything needed for import from core, then run additional checks.
manifest, igame = self.core.import_game(game, args.app_path, platform=args.platform)
exe_path = os.path.join(args.app_path, manifest.meta.launch_exe.lstrip('/'))
# check if most files at least exist or if user might have specified the wrong directory
total = len(manifest.file_manifest_list.elements)
found = sum(os.path.exists(os.path.join(args.app_path, f.filename))
for f in manifest.file_manifest_list.elements)
ratio = found / total
if not found:
logger.error(f'No files belonging to {"DLC" if game.is_dlc else "Game"} "{game.app_title}" '
f'({game.app_name}) found in the specified location, please verify that the path is correct.')
if not game.is_dlc:
# check if game folder is in path, suggest alternative
folder = game.metadata.get('customAttributes', {}).get('FolderName', {}).get('value', game.app_name)
if folder and folder not in args.app_path:
new_path = os.path.join(args.app_path, folder)
logger.info(f'Did you mean "{new_path}"?')
return logger
if not game.is_dlc and not os.path.exists(exe_path) and not args.disable_check:
logger.error(f'Game executable could not be found at "{exe_path}", '
f'please verify that the specified path is correct.')
return logger
if ratio < 0.95:
logger.warning('Some files are missing from the game installation, install may not '
'match latest Epic Games Store version or might be corrupted.')
else:
logger.info(f'{"DLC" if game.is_dlc else "Game"} install appears to be complete.')
self.core.install_game(igame)
if igame.needs_verification:
logger.info(f'NOTE: The {"DLC" if game.is_dlc else "Game"} installation will have to be '
f'verified before it can be updated with legendary.')
logger.info(f'Run "legendary repair {args.app_name}" to do so.')
else:
logger.info(f'Installation had Epic Games Launcher metadata for version "{igame.version}", '
f'verification will not be required.')
# check for importable DLC
if not args.skip_dlcs:
dlcs = self.core.get_dlc_for_game(game.app_name)
if dlcs:
logger.info(f'Found {len(dlcs)} items of DLC that could be imported.')
import_dlc = True
if not args.yes and not args.with_dlcs:
if not args.get_boolean_choice(f'Do you wish to automatically attempt to import all DLCs?'):
import_dlc = False
if import_dlc:
for dlc in dlcs:
args.app_name = dlc.app_name
self.import_game(args)
logger.info(f'{"DLC" if game.is_dlc else "Game"} "{game.app_title}" has been imported.')
return logger

View file

@ -1,6 +1,6 @@
from multiprocessing import Queue
import legendary.core
from legendary.core import LegendaryCore as LegendaryCoreReal
from legendary.models.downloading import AnalysisResult
from legendary.models.game import Game, InstalledGame
from legendary.models.manifest import ManifestMeta
@ -8,11 +8,8 @@ from legendary.models.manifest import ManifestMeta
from .api_exception import LgndrException, LgndrCoreLogHandler
from .manager import DLManager
# lk: Monkeypatch the modified DLManager into LegendaryCore
legendary.core.DLManager = DLManager
class LegendaryCore(legendary.core.LegendaryCore):
class LegendaryCore(LegendaryCoreReal):
def __init__(self, override_config=None, timeout=10.0):
super(LegendaryCore, self).__init__(override_config=override_config, timeout=timeout)
@ -31,7 +28,7 @@ class LegendaryCore(legendary.core.LegendaryCore):
disable_delta: bool = False, override_delta_manifest: str = '',
egl_guid: str = '', preferred_cdn: str = None,
disable_https: bool = False) -> (DLManager, AnalysisResult, ManifestMeta):
return super(LegendaryCore, self).prepare_download(
dlm, analysis, igame = super(LegendaryCore, self).prepare_download(
game=game, base_game=base_game, base_path=base_path,
status_q=status_q, max_shm=max_shm, max_workers=max_workers,
force=force, disable_patching=disable_patching,
@ -45,6 +42,9 @@ class LegendaryCore(legendary.core.LegendaryCore):
egl_guid=egl_guid, preferred_cdn=preferred_cdn,
disable_https=disable_https
)
# lk: monkeypatch run_real (the method that emits the stats) into DLManager
dlm.run_real = DLManager.run_real.__get__(dlm, DLManager)
return dlm, analysis, igame
def uninstall_game(self, installed_game: InstalledGame, delete_files=True, delete_root_directory=False):
try:
@ -73,6 +73,7 @@ class LegendaryCore(legendary.core.LegendaryCore):
def prepare_overlay_install(self, path=None, status_q: Queue = None):
dlm, analysis_result, igame = super(LegendaryCore, self).prepare_overlay_install(path)
# lk: monkeypatch status_q (the queue for download stats)
dlm.run_real = DLManager.run_real.__get__(dlm, DLManager)
dlm.status_queue = status_q
return dlm, analysis_result, igame

View file

@ -6,14 +6,14 @@ from multiprocessing.shared_memory import SharedMemory
from sys import exit
from threading import Condition, Thread
import legendary.downloader.mp.manager
from legendary.downloader.mp.manager import DLManager as DLManagerReal
from legendary.downloader.mp.workers import DLWorker, FileWorker
from legendary.models.downloading import ChunkTask, SharedMemorySegment, TerminateWorkerTask
from .downloading import UIUpdate
class DLManager(legendary.downloader.mp.manager.DLManager):
class DLManager(DLManagerReal):
# fmt: off
# @staticmethod
def run_real(self):

View file

@ -95,28 +95,14 @@ class VerifyWorker(QRunnable):
def run(self):
args = LgndrVerifyGameArgs(app_name=self.app_name,
verify_stdout=self.status_callback)
try:
# TODO: offer this as an alternative when manifest doesn't exist
# TODO: requires the client to be online. To do it this way, we need to
# TODO: somehow detect the error and offer a dialog in which case `verify_games` is
# TODO: re-run with `repair_mode` and `repair_online`
success, failed, missing = self.cli.verify_game(
args, print_command=False, repair_mode=True, repair_online=True)
# success, failed, missing = self.cli.verify_game(args, print_command=False)
self.signals.result.emit(self.app_name, success, failed, missing)
except LgndrException as ret:
self.signals.error.emit(self.app_name, ret.message)
# FIXME: lk: ah ef me sideways, we can't even import this thing properly
# FIXME: lk: so copy it here
def resolve_aliases(core: LegendaryCore, name):
# make sure aliases exist if not yet created
core.update_aliases(force=False)
name = name.strip()
# resolve alias (if any) to real app name
return core.lgd.config.get(
section='Legendary.aliases', option=name,
fallback=core.lgd.aliases.get(name.lower(), name)
)
# TODO: offer this as an alternative when manifest doesn't exist
# TODO: requires the client to be online. To do it this way, we need to
# TODO: somehow detect the error and offer a dialog in which case `verify_games` is
# TODO: re-run with `repair_mode` and `repair_online`
result, failed, missing = self.cli.verify_game(
args, print_command=False, repair_mode=True, repair_online=True)
# success, failed, missing = self.cli.verify_game(args, print_command=False)
if result:
self.signals.result.emit(self.app_name, not failed and not missing, failed, missing)
else:
self.signals.error.emit(self.app_name, result.message)