[cli] Rework CLI

- Class instead of giant main() function
- Uses subparsers for commands (cleaner)
- Will make future enhancements easier

legendary/cli.py

@@ -22,92 +22,18 @@ logging.basicConfig(
 logger = logging.getLogger('cli')
 
 
-# todo refactor this
+# todo logger with QueueHandler/QueueListener
+# todo custom formatter for cli logger (clean info, highlighted error/warning)
 
-def main():
+class LegendaryCLI:
-    parser = argparse.ArgumentParser(description='Legendary (Game Launcher)')
+    def __init__(self):
+        self.core = LegendaryCore()
+        self.logger = logging.getLogger('cli')
 
-    group = parser.add_mutually_exclusive_group()
+    def auth(self, args):
-    group.required = True
-    group.title = 'Commands'
-    group.add_argument('--auth', dest='auth', action='store_true',
-                       help='Authenticate Legendary with your account')
-    group.add_argument('--download', dest='download', action='store',
-                       help='Download a game\'s files', metavar='<name>')
-    group.add_argument('--install', dest='install', action='store',
-                       help='Download and install a game', metavar='<name>')
-    group.add_argument('--update', dest='update', action='store',
-                       help='Update a game (alias for --install)', metavar='<name>')
-    group.add_argument('--uninstall', dest='uninstall', action='store',
-                       help='Remove a game', metavar='<name>')
-    group.add_argument('--launch', dest='launch', action='store',
-                       help='Launch game', metavar='<name>')
-    group.add_argument('--list-games', dest='list_games', action='store_true',
-                       help='List available games')
-    group.add_argument('--list-installed', dest='list_installed', action='store_true',
-                       help='List installed games')
-
-    # general arguments
-    parser.add_argument('-v', dest='debug', action='store_true', help='Set loglevel to debug')
-
-    # arguments for the different commands
-    if os.name == 'nt':
-        auth_group = parser.add_argument_group('Authentication options')
-        # auth options
-        auth_group.add_argument('--import', dest='import_egs_auth', action='store_true',
-                                help='Import EGS authentication data')
-
-    download_group = parser.add_argument_group('Downloading options')
-    download_group.add_argument('--base-path', dest='base_path', action='store', metavar='<path>',
-                                help='Path for game installations (defaults to ~/legendary)')
-    download_group.add_argument('--game-folder', dest='game_folder', action='store', metavar='<path>',
-                                help='Folder for game installation (defaults to folder in metadata)')
-    download_group.add_argument('--max-shared-memory', dest='shared_memory', action='store', metavar='<size>',
-                                type=int, help='Maximum amount of shared memory to use (in MiB), default: 1 GiB')
-    download_group.add_argument('--max-workers', dest='max_workers', action='store', metavar='<num>',
-                                type=int, help='Maximum amount of download workers, default: 2 * logical CPU')
-    download_group.add_argument('--manifest', dest='override_manifest', action='store', metavar='<uri>',
-                                help='Manifest URL or path to use instead of the CDN one (e.g. for downgrading)')
-    download_group.add_argument('--old-manifest', dest='override_old_manifest', action='store', metavar='<uri>',
-                                help='Manifest URL or path to use as the old one (e.g. for testing patching)')
-    download_group.add_argument('--base-url', dest='override_base_url', action='store', metavar='<url>',
-                                help='Base URL to download from (e.g. to test or switch to a different CDNs)')
-    download_group.add_argument('--force', dest='force', action='store_true',
-                                help='Ignore existing files (overwrite)')
-
-    install_group = parser.add_argument_group('Installation options')
-    install_group.add_argument('--disable-patching', dest='disable_patching', action='store_true',
-                               help='Do not attempt to patch existing installations (download full game)')
-
-    launch_group = parser.add_argument_group('Game launch options',
-                                             description='Note: any additional arguments will be passed to the game.')
-    launch_group.add_argument('--offline', dest='offline', action='store_true',
-                              default=False, help='Skip login and launch game without online authentication')
-    launch_group.add_argument('--skip-version-check', dest='skip_version_check', action='store_true',
-                              default=False, help='Skip version check when launching game in online mode')
-    launch_group.add_argument('--override-username', dest='user_name_override', action='store', metavar='<username>',
-                              help='Override username used when launching the game (only works with some titles)')
-    launch_group.add_argument('--dry-run', dest='dry_run', action='store_true',
-                              help='Print the command line that would have been used to launch the game and exit')
-
-    list_group = parser.add_argument_group('Listing options')
-    list_group.add_argument('--check-updates', dest='check_updates', action='store_true',
-                            help='Check for updates when listing installed games')
-
-    args, extra = parser.parse_known_args()
-    core = LegendaryCore()
-
-    config_ll = core.lgd.config.get('Legendary', 'log_level', fallback='info')
-    if config_ll == 'debug' or args.debug:
-        logging.getLogger().setLevel(level=logging.DEBUG)
-        # keep requests quiet
-        logging.getLogger('requests').setLevel(logging.WARNING)
-        logging.getLogger('urllib3').setLevel(logging.WARNING)
-
-    if args.auth:
         try:
             logger.info('Testing existing login data if present...')
-            if core.login():
+            if self.core.login():
                 logger.info('Stored credentials are still valid, if you wish to switch to a different'
                             'account, delete ~/.config/legendary/user.json and try again.')
                 exit(0)
@@ -115,14 +41,14 @@ def main():
             pass
         except InvalidCredentialsError:
             logger.error('Stored credentials were found but were no longer valid. Continuing with login...')
-            core.lgd.invalidate_userdata()
+            self.core.lgd.invalidate_userdata()
 
         if os.name == 'nt' and args.import_egs_auth:
             logger.info('Importing login session from the Epic Launcher...')
             try:
-                if core.auth_import():
+                if self.core.auth_import():
                     logger.info('Successfully imported login session from EGS!')
-                    logger.info(f'Now logged in as user "{core.lgd.userdata["displayName"]}"')
+                    logger.info(f'Now logged in as user "{self.core.lgd.userdata["displayName"]}"')
                     exit(0)
                 else:
                     logger.warning('Login session from EGS seems to no longer be valid.')
@@ -133,26 +59,28 @@ def main():
 
         # unfortunately the captcha stuff makes a complete CLI login flow kinda impossible right now...
         print('Please login via the epic web login!')
-        webbrowser.open('https://www.epicgames.com/id/login?redirectUrl=https%3A%2F%2Fwww.epicgames.com%2Fid%2Fapi%2Fexchange')
+        webbrowser.open(
+            'https://www.epicgames.com/id/login?redirectUrl=https%3A%2F%2Fwww.epicgames.com%2Fid%2Fapi%2Fexchange'
+        )
         print('If web page did not open automatically, please navigate '
               'to https://www.epicgames.com/id/login in your web browser')
-        print('- In case you opened the link manually; please now navigate to '
+        print('- In case you opened the link manually; please open https://www.epicgames.com/id/api/exchange'
-              'https://www.epicgames.com/id/api/exchange in your web browser.')
+              'in your web browser after you have finished logging in.')
         exchange_code = input('Please enter code from JSON response: ')
         exchange_token = exchange_code.strip().strip('"')
 
-        if core.auth_code(exchange_token):
+        if self.core.auth_code(exchange_token):
-            logger.info(f'Successfully logged in as "{core.lgd.userdata["displayName"]}"')
+            logger.info(f'Successfully logged in as "{self.core.lgd.userdata["displayName"]}"')
         else:
             logger.error('Login attempt failed, please see log for details.')
 
-    elif args.list_games:
+    def list_games(self):
         logger.info('Logging in...')
-        if not core.login():
+        if not self.core.login():
             logger.error('Login failed, cannot continue!')
             exit(1)
         logger.info('Getting game list... (this may take a while)')
-        games, dlc_list = core.get_game_and_dlc_list()
+        games, dlc_list = self.core.get_game_and_dlc_list()
 
         print('\nAvailable games:')
         for game in sorted(games, key=lambda x: x.app_title):
@@ -162,51 +90,51 @@ def main():
 
         print(f'\nTotal: {len(games)}')
 
-    elif args.list_installed:
+    def list_installed(self, args):
-        games = core.get_installed_list()
+        games = self.core.get_installed_list()
 
         if args.check_updates:
             logger.info('Logging in to check for updates...')
-            if not core.login():
+            if not self.core.login():
                 logger.error('Login failed! Not checking for updates.')
             else:
-                core.get_assets(True)
+                self.core.get_assets(True)
 
         print('\nInstalled games:')
         for game in sorted(games, key=lambda x: x.title):
             print(f'  * {game.title} (App name: {game.app_name}, version: {game.version})')
-            game_asset = core.get_asset(game.app_name)
+            game_asset = self.core.get_asset(game.app_name)
             if game_asset.build_version != game.version:
                 print(f'    -> Update available! Installed: {game.version}, Latest: {game_asset.build_version}')
 
         print(f'\nTotal: {len(games)}')
 
-    elif args.launch:
+    def launch_game(self, args, extra):
-        app_name = args.launch.strip()
+        app_name = args.app_name
-        if not core.is_installed(app_name):
+        if not self.core.is_installed(app_name):
             logger.error(f'Game {app_name} is not currently installed!')
             exit(1)
 
-        if core.is_dlc(app_name):
+        if self.core.is_dlc(app_name):
             logger.error(f'{app_name} is DLC; please launch the base game instead!')
             exit(1)
 
-        if not args.offline and not core.is_offline_game(app_name):
+        if not args.offline and not self.core.is_offline_game(app_name):
             logger.info('Logging in...')
-            if not core.login():
+            if not self.core.login():
                 logger.error('Login failed, cannot continue!')
                 exit(1)
 
-            if not args.skip_version_check and not core.is_noupdate_game(app_name):
+            if not args.skip_version_check and not self.core.is_noupdate_game(app_name):
                 logger.info('Checking for updates...')
-                installed = core.lgd.get_installed_game(app_name)
+                installed = self.core.lgd.get_installed_game(app_name)
-                latest = core.get_asset(app_name, update=True)
+                latest = self.core.get_asset(app_name, update=True)
                 if latest.build_version != installed.version:
                     logger.error('Game is out of date, please update or launch with update check skipping!')
                     exit(1)
 
-        params, cwd, env = core.get_launch_parameters(app_name=app_name, offline=args.offline,
+        params, cwd, env = self.core.get_launch_parameters(app_name=app_name, offline=args.offline,
-                                                      extra_args=extra, user=args.user_name_override)
+                                                           extra_args=extra, user=args.user_name_override)
 
         logger.info(f'Launching {app_name}...')
         if args.dry_run:
@@ -222,51 +150,48 @@ def main():
 
             subprocess.Popen(params, cwd=cwd, env=env)
 
-    elif args.download or args.install or args.update:
+    def install_game(self, args):
-        if not core.login():
+        if not self.core.login():
             logger.error('Login failed! Cannot continue with download process.')
             exit(1)
 
-        target_app = next(i for i in (args.install, args.update, args.download) if i)
+        if args.update_only:
-        if args.update:
+            if not self.core.is_installed(args.app_name):
-            if not core.is_installed(target_app):
+                logger.error(f'Update requested for "{args.app_name}", but app not installed!')
-                logger.error(f'Update requested for "{target_app}", but app not installed!')
                 exit(1)
 
-        game = core.get_game(target_app, update_meta=True)
+        game = self.core.get_game(args.app_name, update_meta=True)
 
         if not game:
-            logger.fatal(f'Could not find "{target_app}" in list of available games, did you type the name correctly?')
+            logger.error(f'Could not find "{args.app_name}" in list of available games,'
+                         f'did you type the name correctly?')
             exit(1)
 
         if game.is_dlc:
             logger.info('Install candidate is DLC')
             app_name = game.metadata['mainGameItem']['releaseInfo'][0]['appId']
-            base_game = core.get_game(app_name)
+            base_game = self.core.get_game(app_name)
             # check if base_game is actually installed
-            if not core.is_installed(app_name):
+            if not self.core.is_installed(app_name):
                 # download mode doesn't care about whether or not something's installed
-                if args.install or args.update:
+                if not args.no_install:
                     logger.fatal(f'Base game "{app_name}" is not installed!')
                     exit(1)
         else:
             base_game = None
 
         # todo use status queue to print progress from CLI
-        dlm, analysis, igame = core.prepare_download(game=game, base_game=base_game, base_path=args.base_path,
+        dlm, analysis, igame = self.core.prepare_download(game=game, base_game=base_game, base_path=args.base_path,
-                                                     force=args.force, max_shm=args.shared_memory,
+                                                          force=args.force, max_shm=args.shared_memory,
-                                                     max_workers=args.max_workers, game_folder=args.game_folder,
+                                                          max_workers=args.max_workers, game_folder=args.game_folder,
-                                                     disable_patching=args.disable_patching,
+                                                          disable_patching=args.disable_patching,
-                                                     override_manifest=args.override_manifest,
+                                                          override_manifest=args.override_manifest,
-                                                     override_old_manifest=args.override_old_manifest,
+                                                          override_old_manifest=args.override_old_manifest,
-                                                     override_base_url=args.override_base_url)
+                                                          override_base_url=args.override_base_url)
 
         # game is either up to date or hasn't changed, so we have nothing to do
         if not analysis.dl_size:
             logger.info('Download size is 0, the game is either already up to date or has not changed. Exiting...')
-            # if game is downloaded but not "installed", "install" it now (todo handle postinstall as well)
-            if args.install:
-                core.install_game(igame)
             exit(0)
 
         logger.info(f'Install size: {analysis.install_size / 1024 / 1024:.02f} MiB')
@@ -276,7 +201,7 @@ def main():
         logger.info(f'Reusable size: {analysis.reuse_size / 1024 / 1024:.02f} MiB (chunks) / '
                     f'{analysis.unchanged / 1024 / 1024:.02f} MiB (unchanged)')
 
-        res = core.check_installation_conditions(analysis=analysis, install=igame)
+        res = self.core.check_installation_conditions(analysis=analysis, install=igame)
 
         if res.failures:
             logger.fatal('Download cannot proceed, the following errors occured:')
@@ -289,10 +214,11 @@ def main():
             for warn in sorted(res.warnings):
                 logger.warning(warn)
 
-        choice = input(f'Do you wish to install "{igame.title}"? [Y/n]: ')
+        if not args.yes:
-        if choice and choice.lower()[0] != 'y':
+            choice = input(f'Do you wish to install "{igame.title}"? [Y/n]: ')
-            print('Aborting...')
+            if choice and choice.lower()[0] != 'y':
-            exit(0)
+                print('Aborting...')
+                exit(0)
 
         start_t = time.time()
 
@@ -307,66 +233,178 @@ def main():
                            f'If it continues to fail please open an issue on GitHub.')
         else:
             end_t = time.time()
-            if args.install or args.update:
+            if not args.no_install:
-                postinstall = core.install_game(igame)
+                postinstall = self.core.install_game(igame)
 
-                dlcs = core.get_dlc_for_game(game.app_name)
+                dlcs = self.core.get_dlc_for_game(game.app_name)
                 if dlcs:
                     print('The following DLCs are available for this game:')
                     for dlc in dlcs:
                         print(f' - {dlc.app_title} (App name: {dlc.app_name}, version: {dlc.app_version})')
+                    # todo recursively call install with modified args to install DLC automatically (after confirm)
                     print('Installing DLCs works the same as the main game, just use the DLC app name instead.')
                     print('Automatic installation of DLC is currently not supported.')
 
                 if postinstall:
-                    logger.info('This game lists the following prequisites to be installed:')
+                    self._handle_postinstall(postinstall, igame, yes=args.yes)
-                    logger.info(f'{postinstall["name"]}: {" ".join((postinstall["path"], postinstall["args"]))}')
-                    if os.name == 'nt':
-                        choice = input('Do you wish to install the prerequisites? ([y]es, [n]o, [i]gnore): ')
-                        c = choice.lower()[0]
-                        if c == 'i':
-                            core.prereq_installed(igame.app_name)
-                        elif c == 'y':
-                            req_path, req_exec = os.path.split(postinstall['path'])
-                            work_dir = os.path.join(igame.install_path, req_path)
-                            fullpath = os.path.join(work_dir, req_exec)
-                            subprocess.Popen([fullpath, postinstall['args']], cwd=work_dir)
-                    else:
-                        logger.info('Automatic installation not available on Linux.')
 
             logger.info(f'Finished installation process in {end_t - start_t:.02f} seconds.')
 
-    elif args.uninstall:
+    def _handle_postinstall(self, postinstall, igame, yes=False):
-        target_app = args.uninstall
+        print('This game lists the following prequisites to be installed:')
-        igame = core.get_installed_game(target_app)
+        print(f'- {postinstall["name"]}: {" ".join((postinstall["path"], postinstall["args"]))}')
+        if os.name == 'nt':
+            if yes:
+                c = 'n'  # we don't want to launch anything, just silent install.
+            else:
+                choice = input('Do you wish to install the prerequisites? ([y]es, [n]o, [i]gnore): ')
+                c = choice.lower()[0]
+
+            if c == 'i':  # just set it to installed
+                print('Marking prerequisites as installed...')
+                self.core.prereq_installed(igame.app_name)
+            elif c == 'y':  # set to installed and launch installation
+                print('Launching prerequisite executable..')
+                self.core.prereq_installed(igame.app_name)
+                req_path, req_exec = os.path.split(postinstall['path'])
+                work_dir = os.path.join(igame.install_path, req_path)
+                fullpath = os.path.join(work_dir, req_exec)
+                subprocess.Popen([fullpath, postinstall['args']], cwd=work_dir)
+        else:
+            logger.info('Automatic installation not available on Linux.')
+
+    def uninstall_game(self, args):
+        igame = self.core.get_installed_game(args.app_name)
         if not igame:
-            logger.error(f'Game {target_app} not installed, cannot uninstall!')
+            logger.error(f'Game {args.app_name} not installed, cannot uninstall!')
             exit(0)
         if igame.is_dlc:
             logger.error('Uninstalling DLC is not supported.')
             exit(1)
 
-        choice = input(f'Do you wish to uninstall "{igame.title}"? [y/N]: ')
+        if not args.yes:
-        if not choice or choice.lower()[0] != 'y':
+            choice = input(f'Do you wish to uninstall "{igame.title}"? [y/N]: ')
-            print('Aborting...')
+            if not choice or choice.lower()[0] != 'y':
-            exit(0)
+                print('Aborting...')
+                exit(0)
 
         try:
             logger.info(f'Removing "{igame.title}" from "{igame.install_path}"...')
-            core.uninstall_game(igame)
+            self.core.uninstall_game(igame)
 
-            dlcs = core.get_dlc_for_game(igame.app_name)
+            # DLCs are already removed once we delete the main game, so this just removes them from the list
+            dlcs = self.core.get_dlc_for_game(igame.app_name)
             for dlc in dlcs:
-                idlc = core.get_installed_game(dlc.app_name)
+                idlc = self.core.get_installed_game(dlc.app_name)
-                if core.is_installed(dlc.app_name):
+                if self.core.is_installed(dlc.app_name):
                     logger.info(f'Uninstalling DLC "{dlc.app_name}"...')
-                    core.uninstall_game(idlc, delete_files=False)
+                    self.core.uninstall_game(idlc, delete_files=False)
 
             logger.info('Game has been uninstalled.')
         except Exception as e:
             logger.warning(f'Removing game failed: {e!r}, please remove {igame.install_path} manually.')
 
-    core.exit()
+
+def main():
+    parser = argparse.ArgumentParser(description='Legendary (Game Launcher)')
+
+    # general arguments
+    parser.add_argument('-v', dest='debug', action='store_true', help='Set loglevel to debug')
+    parser.add_argument('-y', dest='yes', action='store_true', help='Default to yes for all prompts.')
+
+    # all the commands
+    subparsers = parser.add_subparsers(title='Commands', dest='subparser_name')
+    auth_parser = subparsers.add_parser('auth', help='Authenticate with EPIC')
+    install_parser = subparsers.add_parser('install', help='Download a game',
+                                           usage='%(prog)s <App Name> [options]')
+    uninstall_parser = subparsers.add_parser('uninstall', help='Uninstall (delete) a game')
+    launch_parser = subparsers.add_parser('launch', help='Launch a game', usage='%(prog)s <App Name> [options]',
+                                          description='Note: additional arguments are passed to the game.')
+    _ = subparsers.add_parser('list-games', help='List available (installable) games.')
+    listi_parser = subparsers.add_parser('list-installed', help='List installed games')
+
+    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>')
+    launch_parser.add_argument('app_name', help='Name of the app', metavar='<App Name>')
+
+    # importing only works on Windows right now
+    if os.name == 'nt':
+        auth_parser.add_argument('--import', dest='import_egs_auth', action='store_true',
+                                 help='Import EGS authentication data')
+
+    install_parser.add_argument('--base-path', dest='base_path', action='store', metavar='<path>',
+                                help='Path for game installations (defaults to ~/legendary)')
+    install_parser.add_argument('--game-folder', dest='game_folder', action='store', metavar='<path>',
+                                help='Folder for game installation (defaults to folder in metadata)')
+    install_parser.add_argument('--max-shared-memory', dest='shared_memory', action='store', metavar='<size>',
+                                type=int, help='Maximum amount of shared memory to use (in MiB), default: 1 GiB')
+    install_parser.add_argument('--max-workers', dest='max_workers', action='store', metavar='<num>',
+                                type=int, help='Maximum amount of download workers, default: 2 * logical CPU')
+    install_parser.add_argument('--manifest', dest='override_manifest', action='store', metavar='<uri>',
+                                help='Manifest URL or path to use instead of the CDN one (e.g. for downgrading)')
+    install_parser.add_argument('--old-manifest', dest='override_old_manifest', action='store', metavar='<uri>',
+                                help='Manifest URL or path to use as the old one (e.g. for testing patching)')
+    install_parser.add_argument('--base-url', dest='override_base_url', action='store', metavar='<url>',
+                                help='Base URL to download from (e.g. to test or switch to a different CDNs)')
+    install_parser.add_argument('--force', dest='force', action='store_true',
+                                help='Ignore existing files (overwrite)')
+    install_parser.add_argument('--disable-patching', dest='disable_patching', action='store_true',
+                                help='Do not attempt to patch existing installations (download entire changed file)')
+    install_parser.add_argument('--download-only', dest='no_install', action='store_true',
+                                help='Do not mark game as intalled and do not run prereq installers after download.')
+    install_parser.add_argument('--update-only', dest='update_pnly', action='store_true',
+                                help='Abort if game is not already installed (for automation)')
+
+    launch_parser.add_argument('--offline', dest='offline', action='store_true',
+                               default=False, help='Skip login and launch game without online authentication')
+    launch_parser.add_argument('--skip-version-check', dest='skip_version_check', action='store_true',
+                               default=False, help='Skip version check when launching game in online mode')
+    launch_parser.add_argument('--override-username', dest='user_name_override', action='store', metavar='<username>',
+                               help='Override username used when launching the game (only works with some titles)')
+    launch_parser.add_argument('--dry-run', dest='dry_run', action='store_true',
+                               help='Print the command line that would have been used to launch the game and exit')
+
+    listi_parser.add_argument('--check-updates', dest='check_updates', action='store_true',
+                              help='Check for updates when listing installed games')
+
+    args, extra = parser.parse_known_args()
+
+    if not args.subparser_name:
+        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():
+            print(f'\nCommand: {choice}')
+            print(subparser.format_help())
+        return
+
+    cli = LegendaryCLI()
+
+    config_ll = cli.core.lgd.config.get('Legendary', 'log_level', fallback='info')
+    if config_ll == 'debug' or args.debug:
+        logging.getLogger().setLevel(level=logging.DEBUG)
+        # keep requests quiet
+        logging.getLogger('requests').setLevel(logging.WARNING)
+        logging.getLogger('urllib3').setLevel(logging.WARNING)
+
+    # technically args.func() with setdefaults could work (see docs on subparsers)
+    # but that would require all funcs to accept args and extra...
+    if args.subparser_name == 'auth':
+        cli.auth(args)
+    elif args.subparser_name == 'list-games':
+        cli.list_games()
+    elif args.subparser_name == 'list-installed':
+        cli.list_installed(args)
+    elif args.subparser_name == 'launch':
+        cli.launch_game(args, extra)
+    elif args.subparser_name == 'download':
+        cli.install_game(args)
+    elif args.subparser_name == 'uninstall':
+        cli.uninstall_game(args)
+
+    cli.core.exit()
     exit(0)