refactor: 拆分cli以便后续拓展功能

2025-05-06 00:53:00 +08:00
parent bf3fbe3e96
commit f06be6ed21
5 changed files with 196 additions and 198 deletions
@@ -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()
@@ -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))
@@ -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}")
@@ -0,0 +1,3 @@
+from basic import get_astrbot_root,check_astrbot_root,check_dashboard
+
+__all__ = ["get_astrbot_root", "check_astrbot_root", "check_dashboard"]
@@ -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