import functools import json import logging import os from multiprocessing import Queue from sys import platform as sys_platform from uuid import uuid4 # On Windows the monkeypatching of `run_real` below doesn't work like on Linux # This has the side effect of emitting the UIUpdate in DownloadThread complaining with a TypeError # So import `legendary.core` and monkeypatch its imported DLManager import legendary.core from legendary.core import LegendaryCore as LegendaryCoreReal from legendary.lfs.utils import delete_folder from legendary.models.downloading import AnalysisResult from legendary.models.egl import EGLManifest from legendary.models.game import Game, InstalledGame from legendary.models.manifest import ManifestMeta from rare.lgndr.downloader.mp.manager import DLManager from rare.lgndr.glue.exception import LgndrException, LgndrLogHandler from rare.lgndr.lfs.lgndry import LGDLFS legendary.core.DLManager = DLManager legendary.core.LGDLFS = LGDLFS # fmt: off class LegendaryCore(LegendaryCoreReal): def __init__(self, *args, **kwargs): super(LegendaryCore, self).__init__(*args, **kwargs) self.log.info("Using Rare's LegendaryCore monkey") self.log.info("Using config in %s", self.lgd.path) self.handler = LgndrLogHandler(logging.CRITICAL) self.log.addHandler(self.handler) @staticmethod def unlock_installed(func): @functools.wraps(func) def unlock(self, *args, **kwargs): self.log.debug("%s: Using unlock decorator", func.__name__) if not self.lgd.lock_installed(): self.log.info("Data is locked, trying to forcefully release it") # self.lgd._installed_lock.release(force=True) try: ret = func(self, *args, **kwargs) except Exception as e: raise e finally: self.lgd.unlock_installed() return ret return unlock @property def default_platform(self) -> str: os_default = "Mac" if sys_platform == "darwin" else "Windows" usr_platform = self.lgd.config.get("Legendary", "default_platform", fallback=os_default) return usr_platform if usr_platform in ("Windows", "Win32", "Mac") else os_default def update_check_enabled(self): return True def update_notice_enabled(self): return False # skip_sync defaults to false but since Rare is persistent, skip by default # def get_installed_game(self, app_name, skip_sync=True) -> InstalledGame: # return super(LegendaryCore, self).get_installed_game(app_name, skip_sync) def prepare_download(self, game: Game, base_game: Game = None, 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, disable_https: bool = False, bind_ip: str = None) -> (DLManager, AnalysisResult, ManifestMeta): 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, game_folder=game_folder, 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, egl_guid=egl_guid, preferred_cdn=preferred_cdn, disable_https=disable_https, bind_ip=bind_ip, ) # lk: monkeypatch run_real (the method that emits the stats) into DLManager # pylint: disable=E1111 dlm.run_real = DLManager.run_real.__get__(dlm, DLManager) # lk: set the queue for reporting statistics back the UI dlm.status_queue = Queue() # lk: set the queue to send control signals to the DLManager # lk: this doesn't exist in the original class, but it is monkeypatched in dlm.signals_queue = Queue() return dlm, analysis, igame def uninstall_game(self, installed_game: InstalledGame, delete_files=True, delete_root_directory=False): try: super(LegendaryCore, self).uninstall_game(installed_game, delete_files, delete_root_directory) except Exception as e: raise e finally: pass @unlock_installed.__func__ def egl_import(self, app_name): try: super(LegendaryCore, self).egl_import(app_name) except LgndrException as ret: raise ret finally: pass def egstore_write(self, app_name): self.log.debug(f'Exporting ".egstore" for "{app_name}"') # load igame/game lgd_game = self.get_game(app_name) lgd_igame = self._get_installed_game(app_name) manifest_data, _ = self.get_installed_manifest(app_name) if not manifest_data: self.log.error(f'Game Manifest for "{app_name}" not found, cannot export!') return # create guid if it's not set already if not lgd_igame.egl_guid: lgd_igame.egl_guid = str(uuid4()).replace('-', '').upper() _ = self._install_game(lgd_igame) # convert to egl manifest egl_game = EGLManifest.from_lgd_game(lgd_game, lgd_igame) # make sure .egstore folder exists egstore_folder = os.path.join(lgd_igame.install_path, '.egstore') if not os.path.exists(egstore_folder): os.makedirs(egstore_folder) # 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) mancpn = dict(FormatVersion=0, AppName=app_name, CatalogItemId=lgd_game.catalog_item_id, CatalogNamespace=lgd_game.namespace) 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) def egstore_delete(self, igame: InstalledGame, delete_files=True): self.log.debug(f'Removing ".egstore" for "{igame.app_name}"') if delete_files: delete_folder(os.path.join(igame.install_path, '.egstore')) @unlock_installed.__func__ def egl_export(self, app_name): try: super(LegendaryCore, self).egl_export(app_name) except LgndrException as ret: raise ret finally: pass @unlock_installed.__func__ def egl_sync(self, app_name=''): try: super(LegendaryCore, self).egl_sync(app_name) except LgndrException as ret: raise ret finally: pass def prepare_overlay_install(self, path=None): dlm, analysis_result, igame = super(LegendaryCore, self).prepare_overlay_install(path) # lk: monkeypatch status_q (the queue for download stats) # pylint: disable=E1111 dlm.run_real = DLManager.run_real.__get__(dlm, DLManager) # lk: set the queue for reporting statistics back the UI dlm.status_queue = Queue() # lk: set the queue to send control signals to the DLManager # lk: this doesn't exist in the original class, but it is monkeypatched in dlm.signals_queue = Queue() return dlm, analysis_result, igame # fmt: on