__package__ = "archivebox.cli" __command__ = "archivebox" import os import sys from importlib import import_module import rich_click as click from rich.console import Console from archivebox.config.version import VERSION STDERR = Console(stderr=True) if "--debug" in sys.argv: os.environ["DEBUG"] = "True" sys.argv.remove("--debug") # Universal `--init` flag: when passed to ANY subcommand (e.g. `archivebox server --init`, # `archivebox add --init`, `archivebox shell --init`), run a `quick` archivebox init before # the subcommand executes. `--quick-init` is the old server spelling. Strip either from # argv here so each subcommand's own click parser never sees it. Ignored for `help` and # `init` themselves. if "--init" in sys.argv or "--quick-init" in sys.argv: sys.argv = [arg for arg in sys.argv if arg not in ("--init", "--quick-init")] os.environ["ARCHIVEBOX_WANTS_INIT"] = "1" class ArchiveBoxGroup(click.Group): """lazy loading click group for archivebox commands""" meta_commands = { "help": "archivebox.cli.archivebox_help.main", "version": "archivebox.cli.archivebox_version.main", "mcp": "archivebox.cli.archivebox_mcp.main", "oneshot": "archivebox.cli.archivebox_oneshot.main", } setup_commands = { "init": "archivebox.cli.archivebox_init.main", "install": "archivebox.cli.archivebox_install.main", } # Model commands (CRUD operations via subcommands) model_commands = { "crawl": "archivebox.cli.archivebox_crawl.main", "snapshot": "archivebox.cli.archivebox_snapshot.main", "archiveresult": "archivebox.cli.archivebox_archiveresult.main", "tag": "archivebox.cli.archivebox_tag.main", "binary": "archivebox.cli.archivebox_binary.main", "process": "archivebox.cli.archivebox_process.main", "machine": "archivebox.cli.archivebox_machine.main", "persona": "archivebox.cli.archivebox_persona.main", } archive_commands = { # High-level commands "add": "archivebox.cli.archivebox_add.main", "extract": "archivebox.cli.archivebox_extract.main", "list": "archivebox.cli.archivebox_list.main", "remove": "archivebox.cli.archivebox_remove.main", "run": "archivebox.cli.archivebox_run.main", "update": "archivebox.cli.archivebox_update.main", "status": "archivebox.cli.archivebox_status.main", "search": "archivebox.cli.archivebox_search.main", "config": "archivebox.cli.archivebox_config.main", "schedule": "archivebox.cli.archivebox_schedule.main", "server": "archivebox.cli.archivebox_server.main", "shell": "archivebox.cli.archivebox_shell.main", "manage": "archivebox.cli.archivebox_manage.main", # Introspection commands "pluginmap": "archivebox.cli.archivebox_pluginmap.main", } all_subcommands = { **meta_commands, **setup_commands, **model_commands, **archive_commands, } renamed_commands = { "setup": "install", "import": "add", "archive": "add", } @classmethod def get_canonical_name(cls, cmd_name): return cls.renamed_commands.get(cmd_name, cmd_name) @classmethod def _needs_django_for_lazy_import(cls, cmd_name: str) -> bool: wants_help = any(arg in ("-h", "--help", "--version") for arg in sys.argv[1:]) return not wants_help and (cmd_name in cls.archive_commands or cmd_name in cls.model_commands) @classmethod def _setup_django_for_lazy_import(cls, cmd_name: str) -> None: if not cls._needs_django_for_lazy_import(cmd_name): return from django.apps import apps if apps.ready: return from archivebox.config.django import setup_django setup_django() def get_command(self, ctx, cmd_name): # handle renamed commands if cmd_name in self.renamed_commands: new_name = self.renamed_commands[cmd_name] STDERR.print( f" [violet]Hint:[/violet] `archivebox {cmd_name}` has been renamed to `archivebox {new_name}`", ) cmd_name = new_name ctx.invoked_subcommand = cmd_name # handle lazy loading of commands if cmd_name in self.all_subcommands: self._setup_django_for_lazy_import(cmd_name) return self._lazy_load(cmd_name) # fall-back to using click's default command lookup return super().get_command(ctx, cmd_name) @classmethod def _lazy_load(cls, cmd_name_or_path): import_path = cls.all_subcommands.get(cmd_name_or_path) if import_path is None: import_path = cmd_name_or_path modname, funcname = import_path.rsplit(".", 1) mod = import_module(modname) func = vars(mod)[funcname] if func.__doc__ is None: raise ValueError(f"lazy loading of {import_path} failed - no docstring found on method") return func @click.group(cls=ArchiveBoxGroup, invoke_without_command=True) @click.option("--help", "-h", is_flag=True, help="Show help") @click.version_option(VERSION, "-v", "--version", package_name="archivebox", message="%(version)s") @click.pass_context def cli(ctx, help=False): """ArchiveBox: The self-hosted internet archive""" subcommand = ArchiveBoxGroup.get_canonical_name(ctx.invoked_subcommand) # if --help is passed or no subcommand is given, show custom help message if help or ctx.invoked_subcommand is None: ctx.invoke(ctx.command.get_command(ctx, "help")) # if the subcommand is in archive_commands or model_commands, # then we need to set up the django environment and check that we're in a valid data folder wants_help = any(arg in ("-h", "--help", "--version") for arg in sys.argv[1:]) if not wants_help and (subcommand in ArchiveBoxGroup.archive_commands or subcommand in ArchiveBoxGroup.model_commands): try: if subcommand == "server": run_in_debug = "--reload" in sys.argv or os.environ.get("DEBUG") in ("1", "true", "True", "TRUE", "yes") if run_in_debug: os.environ["ARCHIVEBOX_RUNSERVER"] = "1" if "--reload" in sys.argv: os.environ["ARCHIVEBOX_AUTORELOAD"] = "1" from archivebox.config.django import setup_django from archivebox.misc.checks import check_data_folder, check_migrations setup_django() if os.environ.get("ARCHIVEBOX_WANTS_INIT") == "1" and subcommand not in ("init", "help"): # Universal `--init` was passed: build/upgrade the data folder before # the regular preflight runs, so it succeeds on a fresh dir and an # out-of-date schema both. Drop the env var afterwards so spawned # subprocesses (supervisord workers, daphne, runner, etc.) inherit # a clean env and don't re-trigger init in every child. from archivebox.cli.archivebox_init import init as archivebox_init archivebox_init(quick=True) os.environ.pop("ARCHIVEBOX_WANTS_INIT", None) check_data_folder() if subcommand != "update": check_migrations(auto_apply=True) except Exception as e: STDERR.print(f"[red][X] Error setting up Django or checking data folder: {e}[/red]") if subcommand not in ("manage", "shell"): # not all management commands need django to be setup beforehand raise def main(args=None, prog_name=None): from archivebox.config.constants import CONSTANTS os.environ.setdefault("ABX_PLUGINS_DIR", str(CONSTANTS.USER_PLUGINS_DIR)) # show `docker run archivebox xyz` in help messages if running in docker IN_DOCKER = os.environ.get("IN_DOCKER", False) in ("1", "true", "True", "TRUE", "yes") IS_TTY = sys.stdin.isatty() prog_name = prog_name or (f"docker compose run{'' if IS_TTY else ' -T'} archivebox" if IN_DOCKER else "archivebox") previous_unraisablehook = sys.unraisablehook def ignore_shutdown_unraisable(unraisable): if isinstance(unraisable.exc_value, (KeyboardInterrupt, SystemExit)): return previous_unraisablehook(unraisable) sys.unraisablehook = ignore_shutdown_unraisable try: cli(args=args, prog_name=prog_name, standalone_mode=False) except click.Abort: STDERR.print("\n[red][X] Got CTRL+C. Exiting...[/red]") raise SystemExit(130) from None except click.ClickException as err: err.show() raise SystemExit(err.exit_code) from None except click.exceptions.Exit as err: raise SystemExit(err.exit_code) from None except KeyboardInterrupt: STDERR.print("\n[red][X] Got CTRL+C. Exiting...[/red]") raise SystemExit(130) from None finally: sys.unraisablehook = previous_unraisablehook if __name__ == "__main__": main()