From f06be6ed2156fe02ef2520c0d28ebfc54a10af28 Mon Sep 17 00:00:00 2001 From: Raven95676 Date: Tue, 6 May 2025 00:53:00 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20=E6=8B=86=E5=88=86cli=E4=BB=A5?= =?UTF-8?q?=E4=BE=BF=E5=90=8E=E7=BB=AD=E6=8B=93=E5=B1=95=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- astrbot/cli/__main__.py | 212 ++----------------------------- astrbot/cli/commands/cmd_init.py | 44 +++++++ astrbot/cli/commands/cmd_run.py | 32 +++++ astrbot/cli/utils/__init__.py | 3 + astrbot/cli/utils/basic.py | 103 +++++++++++++++ 5 files changed, 196 insertions(+), 198 deletions(-) create mode 100644 astrbot/cli/commands/cmd_init.py create mode 100644 astrbot/cli/commands/cmd_run.py create mode 100644 astrbot/cli/utils/__init__.py create mode 100644 astrbot/cli/utils/basic.py diff --git a/astrbot/cli/__main__.py b/astrbot/cli/__main__.py index 580ad2e1c..05a0493d0 100644 --- a/astrbot/cli/__main__.py +++ b/astrbot/cli/__main__.py @@ -1,11 +1,8 @@ -import asyncio -import os -import shutil -import sys import click -from pathlib import Path +import sys from astrbot.core.config.default import VERSION - +from .commands.cmd_init import init +from .commands.cmd_run import run logo_tmpl = r""" ___ _______.___________..______ .______ ______ .___________. @@ -14,115 +11,11 @@ logo_tmpl = r""" / /_\ \ \ \ | | | / | _ < | | | | | | / _____ \ .----) | | | | |\ \----.| |_) | | `--' | | | /__/ \__\ |_______/ |__| | _| `._____||______/ \______/ |__| - """ -# utils -def _get_astrbot_root(path: str | None) -> Path: - """获取astrbot根目录""" - match path: - case None: - match ASTRBOT_ROOT := os.getenv("ASTRBOT_ROOT"): - case None: - astrbot_root = Path.cwd() / "data" - case _: - astrbot_root = Path(ASTRBOT_ROOT).resolve() - case str(): - astrbot_root = Path(path).resolve() - - dot_astrbot = astrbot_root / ".astrbot" - if not dot_astrbot.exists(): - if click.confirm( - f"运行前必须先执行初始化！请检查当前目录是否正确，回车以继续: {astrbot_root}", - default=True, - abort=True, - ): - dot_astrbot.touch() - astrbot_root.mkdir(parents=True, exist_ok=True) - click.echo(f"Created {dot_astrbot}") - - return astrbot_root - - -# 通过类型来验证先后，必须先获取 Path 对象才能对该目录进行检查 -def _check_astrbot_root(astrbot_root: Path) -> None: - """验证""" - dot_astrbot = astrbot_root / ".astrbot" - if not astrbot_root.exists(): - click.echo(f"AstrBot root directory does not exist: {astrbot_root}") - click.echo("Please run 'astrbot init' to create the directory.") - sys.exit(1) - else: - click.echo(f"AstrBot root directory exists: {astrbot_root}") - if not dot_astrbot.exists(): - click.echo( - "如果你确认这是 Astrbot root directory, 你需要在当前目录下创建一个 .astrbot 文件标记该目录为 AstrBot 的数据目录。" - ) - if click.confirm( - f"请检查当前目录是否正确，确认正确请回车: {astrbot_root}", - default=True, - abort=True, - ): - dot_astrbot.touch() - click.echo(f"Created {dot_astrbot}") - else: - click.echo(f"Welcome back! AstrBot root directory: {astrbot_root}") - - -async def _check_dashboard(astrbot_root: Path) -> None: - """检查是否安装了dashboard""" - try: - from ..core.utils.io import get_dashboard_version, download_dashboard - except ImportError: - from astrbot.core.utils.io import get_dashboard_version, download_dashboard - - try: - # 添加 create=True 参数以确保在初始化时不会抛出异常 - dashboard_version = await get_dashboard_version() - match dashboard_version: - case None: - click.echo("未安装管理面板") - if click.confirm( - "是否安装管理面板？", - default=True, - abort=True, - ): - click.echo("正在安装管理面板...") - # 确保使用 create=True 参数 - await download_dashboard( - path="data/dashboard.zip", extract_path=str(astrbot_root) - ) - click.echo("管理面板安装完成") - - case str(): - if dashboard_version == f"v{VERSION}": - click.echo("无需更新") - else: - try: - version = dashboard_version.split("v")[1] - click.echo(f"管理面板版本: {version}") - # 确保使用 create=True 参数 - await download_dashboard( - path="data/dashboard.zip", extract_path=str(astrbot_root) - ) - except Exception as e: - click.echo(f"下载管理面板失败: {e}") - return - except FileNotFoundError: - click.echo("初始化管理面板目录...") - # 初始化模式下，下载到指定位置 - try: - await download_dashboard( - path=str(astrbot_root / "dashboard.zip"), extract_path=str(astrbot_root) - ) - click.echo("管理面板初始化完成") - except Exception as e: - click.echo(f"下载管理面板失败: {e}") - return - - -@click.group(name="astrbot") +@click.group() +@click.version_option(VERSION, prog_name="AstrBot") def cli() -> None: """The AstrBot CLI""" click.echo(logo_tmpl) @@ -130,94 +23,13 @@ def cli() -> None: click.echo(f"AstrBot version: {VERSION}") -# region init -@cli.command() -@click.option("--path", "-p", help="AstrBot 数据目录") -@click.option("--force", "-f", is_flag=True, help="强制初始化") -def init(path: str | None, force: bool) -> None: - """Initialize AstrBot""" - click.echo("Initializing AstrBot...") - astrbot_root = _get_astrbot_root(path) - if force: - if click.confirm( - "强制初始化会删除当前目录下的所有文件，是否继续？", - default=False, - abort=True, - ): - click.echo("正在删除当前目录下的所有文件...") - shutil.rmtree(astrbot_root, ignore_errors=True) - - _check_astrbot_root(astrbot_root) - - click.echo(f"AstrBot root directory: {astrbot_root}") - - if not astrbot_root.exists(): - # 创建目录 - astrbot_root.mkdir(parents=True, exist_ok=True) - click.echo(f"Created directory: {astrbot_root}") - else: - click.echo(f"Directory already exists: {astrbot_root}") - - config_path: Path = astrbot_root / "config" - plugins_path: Path = astrbot_root / "plugins" - temp_path: Path = astrbot_root / "temp" - config_path.mkdir(parents=True, exist_ok=True) - plugins_path.mkdir(parents=True, exist_ok=True) - temp_path.mkdir(parents=True, exist_ok=True) - - click.echo(f"Created directories: {config_path}, {plugins_path}, {temp_path}") - - # 检查是否安装了dashboard - asyncio.run(_check_dashboard(astrbot_root)) - - -# region run -@cli.command() -@click.option("--path", "-p", help="AstrBot 数据目录") -def run(path: str | None = None) -> None: - """Run AstrBot""" - # 解析为绝对路径 - try: - from ..core.log import LogBroker - from ..core import db_helper - from ..core.initial_loader import InitialLoader - except ImportError: - from astrbot.core.log import LogBroker - from astrbot.core import db_helper - from astrbot.core.initial_loader import InitialLoader - - astrbot_root = _get_astrbot_root(path) - - _check_astrbot_root(astrbot_root) - - asyncio.run(_check_dashboard(astrbot_root)) - - log_broker = LogBroker() - db = db_helper - - core_lifecycle = InitialLoader(db, log_broker) - try: - asyncio.run(core_lifecycle.start()) - except KeyboardInterrupt: - click.echo("接收到退出信号，正在关闭 AstrBot...") - except Exception as e: - click.echo(f"运行时出现错误: {e}") - - -# region Basic -@cli.command(name="version") -def version() -> None: - """Show the version of AstrBot""" - click.echo(f"AstrBot version: {VERSION}") - - -@cli.command() +@click.command() @click.argument("command_name", required=False, type=str) def help(command_name: str | None) -> None: - """Show help information for commands + """显示命令的帮助信息 - If COMMAND_NAME is provided, show detailed help for that command. - Otherwise, show general help information. + 如果提供了 COMMAND_NAME，则显示该命令的详细帮助信息。 + 否则，显示通用帮助信息。 """ ctx = click.get_current_context() if command_name: @@ -234,5 +46,9 @@ def help(command_name: str | None) -> None: click.echo(cli.get_help(ctx)) +cli.add_command(init) +cli.add_command(run) +cli.add_command(help) + if __name__ == "__main__": - cli() + cli() \ No newline at end of file diff --git a/astrbot/cli/commands/cmd_init.py b/astrbot/cli/commands/cmd_init.py new file mode 100644 index 000000000..58d5e8c18 --- /dev/null +++ b/astrbot/cli/commands/cmd_init.py @@ -0,0 +1,44 @@ +import shutil + +import click +import asyncio +from pathlib import Path +from ..utils import get_astrbot_root, check_astrbot_root, check_dashboard + + +@click.command() +@click.option("--path", "-p", help="AstrBot 数据目录") +@click.option("--force", "-f", is_flag=True, help="强制初始化") +def init(path: str | None, force: bool) -> None: + """初始化 AstrBot""" + click.echo("Initializing AstrBot...") + astrbot_root = get_astrbot_root(path) + if force: + if click.confirm( + "强制初始化会删除当前目录下的所有文件，是否继续？", + default=False, + abort=True, + ): + click.echo("正在删除当前目录下的所有文件...") + shutil.rmtree(astrbot_root, ignore_errors=True) + + check_astrbot_root(astrbot_root) + + click.echo(f"AstrBot root directory: {astrbot_root}") + + if not astrbot_root.exists(): + astrbot_root.mkdir(parents=True, exist_ok=True) + click.echo(f"Created directory: {astrbot_root}") + else: + click.echo(f"Directory already exists: {astrbot_root}") + + config_path: Path = astrbot_root / "config" + plugins_path: Path = astrbot_root / "plugins" + temp_path: Path = astrbot_root / "temp" + config_path.mkdir(parents=True, exist_ok=True) + plugins_path.mkdir(parents=True, exist_ok=True) + temp_path.mkdir(parents=True, exist_ok=True) + + click.echo(f"Created directories: {config_path}, {plugins_path}, {temp_path}") + + asyncio.run(_check_dashboard(astrbot_root)) \ No newline at end of file diff --git a/astrbot/cli/commands/cmd_run.py b/astrbot/cli/commands/cmd_run.py new file mode 100644 index 000000000..98f6e7ba2 --- /dev/null +++ b/astrbot/cli/commands/cmd_run.py @@ -0,0 +1,32 @@ +import click +import asyncio +from ..utils import get_astrbot_root, check_astrbot_root, check_dashboard + + +@click.command() +@click.option("--path", "-p", help="AstrBot 数据目录") +def run(path: str | None = None) -> None: + """运行 AstrBot""" + try: + from ..core.log import LogBroker + from ..core import db_helper + from ..core.initial_loader import InitialLoader + except ImportError: + from astrbot.core.log import LogBroker + from astrbot.core import db_helper + from astrbot.core.initial_loader import InitialLoader + + astrbot_root = get_astrbot_root(path) + check_astrbot_root(astrbot_root) + asyncio.run(check_dashboard(astrbot_root)) + + log_broker = LogBroker() + db = db_helper + + core_lifecycle = InitialLoader(db, log_broker) + try: + asyncio.run(core_lifecycle.start()) + except KeyboardInterrupt: + click.echo("接收到退出信号，正在关闭 AstrBot...") + except Exception as e: + click.echo(f"运行时出现错误: {e}") \ No newline at end of file diff --git a/astrbot/cli/utils/__init__.py b/astrbot/cli/utils/__init__.py new file mode 100644 index 000000000..4af69bf94 --- /dev/null +++ b/astrbot/cli/utils/__init__.py @@ -0,0 +1,3 @@ +from basic import get_astrbot_root,check_astrbot_root,check_dashboard + +__all__ = ["get_astrbot_root", "check_astrbot_root", "check_dashboard"] diff --git a/astrbot/cli/utils/basic.py b/astrbot/cli/utils/basic.py new file mode 100644 index 000000000..a243cdd7a --- /dev/null +++ b/astrbot/cli/utils/basic.py @@ -0,0 +1,103 @@ +import os +import sys +from pathlib import Path +import click +from astrbot.core.config.default import VERSION + + +def get_astrbot_root(path: str | None) -> Path: + """获取astrbot根目录""" + match path: + case None: + match ASTRBOT_ROOT := os.getenv("ASTRBOT_ROOT"): + case None: + astrbot_root = Path.cwd() / "data" + case _: + astrbot_root = Path(ASTRBOT_ROOT).resolve() + case str(): + astrbot_root = Path(path).resolve() + + dot_astrbot = astrbot_root / ".astrbot" + if not dot_astrbot.exists(): + if click.confirm( + f"运行前必须先执行初始化！请检查当前目录是否正确，回车以继续: {astrbot_root}", + default=True, + abort=True, + ): + dot_astrbot.touch() + astrbot_root.mkdir(parents=True, exist_ok=True) + click.echo(f"Created {dot_astrbot}") + + return astrbot_root + + +def check_astrbot_root(astrbot_root: Path) -> None: + """验证""" + dot_astrbot = astrbot_root / ".astrbot" + if not astrbot_root.exists(): + click.echo(f"AstrBot root directory does not exist: {astrbot_root}") + click.echo("Please run 'astrbot init' to create the directory.") + sys.exit(1) + else: + click.echo(f"AstrBot root directory exists: {astrbot_root}") + if not dot_astrbot.exists(): + click.echo( + "如果你确认这是 Astrbot root directory, 你需要在当前目录下创建一个 .astrbot 文件标记该目录为 AstrBot 的数据目录。" + ) + if click.confirm( + f"请检查当前目录是否正确，确认正确请回车: {astrbot_root}", + default=True, + abort=True, + ): + dot_astrbot.touch() + click.echo(f"Created {dot_astrbot}") + else: + click.echo(f"Welcome back! AstrBot root directory: {astrbot_root}") + + +async def check_dashboard(astrbot_root: Path) -> None: + """检查是否安装了dashboard""" + try: + from ..core.utils.io import get_dashboard_version, download_dashboard + except ImportError: + from astrbot.core.utils.io import get_dashboard_version, download_dashboard + + try: + dashboard_version = await get_dashboard_version() + match dashboard_version: + case None: + click.echo("未安装管理面板") + if click.confirm( + "是否安装管理面板？", + default=True, + abort=True, + ): + click.echo("正在安装管理面板...") + await download_dashboard( + path="data/dashboard.zip", extract_path=str(astrbot_root) + ) + click.echo("管理面板安装完成") + + case str(): + if dashboard_version == f"v{VERSION}": + click.echo("无需更新") + else: + try: + version = dashboard_version.split("v")[1] + click.echo(f"管理面板版本: {version}") + await download_dashboard( + path="data/dashboard.zip", extract_path=str(astrbot_root) + ) + except Exception as e: + click.echo(f"下载管理面板失败: {e}") + return + except FileNotFoundError: + click.echo("初始化管理面板目录...") + try: + await download_dashboard( + path=str(astrbot_root / "dashboard.zip"), extract_path=str(astrbot_root) + ) + click.echo("管理面板初始化完成") + except Exception as e: + click.echo(f"下载管理面板失败: {e}") + return \ No newline at end of file