feat(cli): enhance run/init/help cmds & service config support

AGENTS.md

@@ -36,6 +36,7 @@ Runs on `http://localhost:3000` by default.
    - Update `astrbot/cli/commands/cmd_run.py`:
      - Add to the module docstring under "Environment Variables Used in Project".
      - Add to the `keys_to_print` list in the `run` function for debug output.
+9. To check all available CLI commands and their usage recursively, run `astrbot help --all`.
 
 
 ## PR instructions

astrbot/cli/__main__.py

@@ -1,8 +1,10 @@
 """AstrBot CLI entry point"""
 
+import os
 import sys
 
 import click
+from click.shell_completion import get_completion_class
 
 from . import __version__
 from .commands import bk, conf, init, plug, run, uninstall
@@ -28,19 +30,43 @@ def cli() -> None:
 
 @click.command()
 @click.argument("command_name", required=False, type=str)
-def help(command_name: str | None) -> None:
+@click.option(
+    "--all", "-a", is_flag=True, help="Show help for all commands recursively."
+)
+def help(command_name: str | None, all: bool) -> None:
     """Display help information for commands
 
     If COMMAND_NAME is provided, display detailed help for that command.
     Otherwise, display general help information.
     """
     ctx = click.get_current_context()
+
+    if all:
+
+        def print_recursive_help(command, parent_ctx):
+            name = command.name
+            if parent_ctx is None:
+                name = "astrbot"
+
+            cmd_ctx = click.Context(command, info_name=name, parent=parent_ctx)
+            click.echo(command.get_help(cmd_ctx))
+            click.echo("\n" + "-" * 50 + "\n")
+
+            if isinstance(command, click.Group):
+                for subcommand in command.commands.values():
+                    print_recursive_help(subcommand, cmd_ctx)
+
+        print_recursive_help(cli, None)
+        return
+
     if command_name:
         # Find the specified command
         command = cli.get_command(ctx, command_name)
         if command:
             # Display help for the specific command
-            click.echo(command.get_help(ctx))
+            parent = ctx.parent if ctx.parent else ctx
+            cmd_ctx = click.Context(command, info_name=command.name, parent=parent)
+            click.echo(command.get_help(cmd_ctx))
         else:
             click.echo(f"Unknown command: {command_name}")
             sys.exit(1)
@@ -57,5 +83,34 @@ cli.add_command(conf)
 cli.add_command(uninstall)
 cli.add_command(bk)
 
+
+@click.command()
+@click.argument("shell", required=False, type=click.Choice(["bash", "zsh", "fish"]))
+def completion(shell: str | None) -> None:
+    """Generate shell completion script"""
+    if shell is None:
+        shell_path = os.environ.get("SHELL", "")
+        if "zsh" in shell_path:
+            shell = "zsh"
+        elif "bash" in shell_path:
+            shell = "bash"
+        elif "fish" in shell_path:
+            shell = "fish"
+        else:
+            click.echo(
+                "Could not detect shell. Please specify one of: bash, zsh, fish",
+                err=True,
+            )
+            sys.exit(1)
+
+    comp_cls = get_completion_class(shell)
+    comp = comp_cls(
+        cli, ctx_args={}, prog_name="astrbot", complete_var="_ASTRBOT_COMPLETE"
+    )
+    click.echo(comp.source())
+
+
+cli.add_command(completion)
+
 if __name__ == "__main__":
     cli()

astrbot/cli/commands/cmd_run.py

@@ -75,14 +75,23 @@ async def run_astrbot(astrbot_root: Path) -> None:
 @click.option("--host", "-H", help="AstrBot Dashboard Host", required=False, type=str)
 @click.option("--port", "-p", help="AstrBot Dashboard port", required=False, type=str)
 @click.option("--root", help="AstrBot root directory", required=False, type=str)
+@click.option(
+    "--service-config",
+    "-c",
+    help="Service configuration file path",
+    required=False,
+    type=str,
+)
 @click.option(
     "--backend-only",
+    "-b",
     is_flag=True,
     default=False,
     help="Disable WebUI, run backend only",
 )
 @click.option(
     "--log-level",
+    "-l",
     help="Log level",
     required=False,
     type=str,
@@ -95,6 +104,7 @@ def run(
     host: str,
     port: str,
     root: str,
+    service_config: str,
     backend_only: bool,
     log_level: str,
     debug: bool,
@@ -104,6 +114,31 @@ def run(
         if debug:
             log_level = "DEBUG"
 
+        if service_config:
+            svc_path = Path(service_config)
+            if svc_path.exists():
+                content = svc_path.read_text(encoding="utf-8")
+                for line in content.splitlines():
+                    line = line.strip()
+                    if not line or line.startswith("#"):
+                        continue
+                    if "=" in line:
+                        key, value = line.split("=", 1)
+                        key = key.strip()
+                        value = value.strip()
+                        # Remove quotes
+                        if (value.startswith('"') and value.endswith('"')) or (
+                            value.startswith("'") and value.endswith("'")
+                        ):
+                            value = value[1:-1]
+
+                        if key == "HOST" and not host:
+                            host = value
+                        elif key == "PORT" and not port:
+                            port = value
+                        elif key == "ASTRBOT_ROOT" and not root:
+                            root = value
+
         # Normalize environment variables for backward compatibility
         # If the legacy env var is set but the new one isn't, copy it over.
         env_map = {

astrbot/core/provider/sources/edge_tts_source.py

@@ -3,6 +3,7 @@ import os
 import subprocess
 import uuid
 
+import anyio
 import edge_tts
 
 from astrbot.core import logger
@@ -99,8 +100,9 @@ class ProviderEdgeTTS(TTSProvider):
                 logger.debug(f"FFmpeg错误输出: {stderr.decode().strip()}")
                 logger.info(f"[EdgeTTS] 返回值(0代表成功): {p.returncode}")
 
-            os.remove(mp3_path)
-            if os.path.exists(wav_path) and os.path.getsize(wav_path) > 0:
+            await anyio.Path(mp3_path).unlink()
+            wav_path_obj = anyio.Path(wav_path)
+            if await wav_path_obj.exists() and (await wav_path_obj.stat()).st_size > 0:
                 return wav_path
             logger.error("生成的WAV文件不存在或为空")
             raise RuntimeError("生成的WAV文件不存在或为空")

astrbot/core/utils/io.py

@@ -12,6 +12,7 @@ from ipaddress import IPv4Address, IPv6Address, ip_address
 from pathlib import Path
 
 import aiohttp
+import anyio
 import certifi
 import psutil
 from PIL import Image
@@ -85,15 +86,15 @@ async def download_image_by_url(
                 async with session.post(url, json=post_data) as resp:
                     if not path:
                         return save_temp_img(await resp.read())
-                    with open(path, "wb") as f:
-                        f.write(await resp.read())
+                    async with await anyio.open_file(path, "wb") as f:
+                        await f.write(await resp.read())
                     return path
             else:
                 async with session.get(url) as resp:
                     if not path:
                         return save_temp_img(await resp.read())
-                    with open(path, "wb") as f:
-                        f.write(await resp.read())
+                    async with await anyio.open_file(path, "wb") as f:
+                        await f.write(await resp.read())
                     return path
     except (aiohttp.ClientConnectorSSLError, aiohttp.ClientConnectorCertificateError):
         # 关闭SSL验证（仅在证书验证失败时作为fallback）
@@ -111,15 +112,15 @@ async def download_image_by_url(
                 async with session.post(url, json=post_data, ssl=ssl_context) as resp:
                     if not path:
                         return save_temp_img(await resp.read())
-                    with open(path, "wb") as f:
-                        f.write(await resp.read())
+                    async with await anyio.open_file(path, "wb") as f:
+                        await f.write(await resp.read())
                     return path
             else:
                 async with session.get(url, ssl=ssl_context) as resp:
                     if not path:
                         return save_temp_img(await resp.read())
-                    with open(path, "wb") as f:
-                        f.write(await resp.read())
+                    async with await anyio.open_file(path, "wb") as f:
+                        await f.write(await resp.read())
                     return path
     except Exception as e:
         raise e
@@ -144,12 +145,12 @@ async def download_file(url: str, path: str, show_progress: bool = False) -> Non
                 start_time = time.time()
                 if show_progress:
                     print(f"文件大小: {total_size / 1024:.2f} KB | 文件地址: {url}")
-                with open(path, "wb") as f:
+                async with await anyio.open_file(path, "wb") as f:
                     while True:
                         chunk = await resp.content.read(8192)
                         if not chunk:
                             break
-                        f.write(chunk)
+                        await f.write(chunk)
                         downloaded_size += len(chunk)
                         if show_progress:
                             elapsed_time = (
@@ -183,12 +184,12 @@ async def download_file(url: str, path: str, show_progress: bool = False) -> Non
                 start_time = time.time()
                 if show_progress:
                     print(f"文件大小: {total_size / 1024:.2f} KB | 文件地址: {url}")
-                with open(path, "wb") as f:
+                async with await anyio.open_file(path, "wb") as f:
                     while True:
                         chunk = await resp.content.read(8192)
                         if not chunk:
                             break
-                        f.write(chunk)
+                        await f.write(chunk)
                         downloaded_size += len(chunk)
                         if show_progress:
                             elapsed_time = time.time() - start_time
@@ -258,16 +259,16 @@ async def get_public_ip_address() -> list[IPv4Address | IPv6Address]:
 async def get_dashboard_version():
     # First check user data directory (manually updated / downloaded dashboard).
     dist_dir = os.path.join(get_astrbot_data_path(), "dist")
-    if not os.path.exists(dist_dir):
+    if not await anyio.Path(dist_dir).exists():
         # Fall back to the dist bundled inside the installed wheel.
         _bundled = Path(get_astrbot_path()) / "astrbot" / "dashboard" / "dist"
         if _bundled.exists():
             dist_dir = str(_bundled)
-    if os.path.exists(dist_dir):
+    if await anyio.Path(dist_dir).exists():
         version_file = os.path.join(dist_dir, "assets", "version")
-        if os.path.exists(version_file):
-            with open(version_file, encoding="utf-8") as f:
-                v = f.read().strip()
+        if await anyio.Path(version_file).exists():
+            async with await anyio.open_file(version_file, encoding="utf-8") as f:
+                v = (await f.read()).strip()
                 return v
     return None
 
@@ -281,14 +282,14 @@ async def download_dashboard(
 ) -> None:
     """下载管理面板文件"""
     if path is None:
-        zip_path = Path(get_astrbot_data_path()).absolute() / "dashboard.zip"
+        zip_path = anyio.Path(get_astrbot_data_path()) / "dashboard.zip"
     else:
-        zip_path = Path(path).absolute()
+        zip_path = anyio.Path(path)
 
     # 缓存机制
-    cache_dir = Path(get_astrbot_data_path()).absolute() / "cache"
-    if not cache_dir.exists():
-        cache_dir.mkdir(parents=True, exist_ok=True)
+    cache_dir = anyio.Path(get_astrbot_data_path()) / "cache"
+    if not await cache_dir.exists():
+        await cache_dir.mkdir(parents=True, exist_ok=True)
 
     use_cache = False
 
@@ -297,69 +298,68 @@ async def download_dashboard(
         cache_name = f"dashboard_{version}.zip"
         cache_path = cache_dir / cache_name
 
-        if cache_path.exists():
+        if await cache_path.exists():
             logger.info(f"发现本地缓存的管理面板文件: {cache_path}")
             try:
-                with zipfile.ZipFile(cache_path, "r") as z:
+                with zipfile.ZipFile(str(cache_path), "r") as z:
                     if z.testzip() is None:
                         logger.info("缓存文件校验通过，将直接使用缓存。")
                         if str(cache_path) != str(zip_path):
-                            shutil.copy(cache_path, zip_path)
+                            shutil.copy(str(cache_path), str(zip_path))
                         use_cache = True
                     else:
                         logger.warning("缓存文件损坏，将重新下载。")
-                        os.remove(cache_path)
+                        await cache_path.unlink()
             except zipfile.BadZipFile:
                 logger.warning("缓存文件损坏 (BadZipFile)，将重新下载。")
-                os.remove(cache_path)
-
-    if not use_cache:
-        if latest or len(str(version)) != 40:
-            ver_name = "latest" if latest else version
-            dashboard_release_url = f"https://astrbot-registry.soulter.top/download/astrbot-dashboard/{ver_name}/dist.zip"
-            logger.info(
-                f"准备下载指定发行版本的 AstrBot WebUI 文件: {dashboard_release_url}",
-            )
-            try:
-                await download_file(
-                    dashboard_release_url,
-                    str(zip_path),
-                    show_progress=True,
+                await cache_path.unlink()
+        if not use_cache:
+            if latest or len(str(version)) != 40:
+                ver_name = "latest" if latest else version
+                dashboard_release_url = f"https://astrbot-registry.soulter.top/download/astrbot-dashboard/{ver_name}/dist.zip"
+                logger.info(
+                    f"准备下载指定发行版本的 AstrBot WebUI 文件: {dashboard_release_url}",
                 )
-            except BaseException as _:
                 try:
-                    if latest:
-                        dashboard_release_url = "https://github.com/AstrBotDevs/AstrBot/releases/latest/download/dist.zip"
-                    else:
-                        dashboard_release_url = f"https://github.com/AstrBotDevs/AstrBot/releases/download/{version}/dist.zip"
-                    if proxy:
-                        dashboard_release_url = f"{proxy}/{dashboard_release_url}"
                     await download_file(
                         dashboard_release_url,
                         str(zip_path),
                         show_progress=True,
                     )
-                except Exception as e:
-                    if not latest:
-                        logger.warning(
-                            f"下载指定版本({version})失败: {e}，尝试下载最新版本。"
+                except BaseException as _:
+                    try:
+                        if latest:
+                            dashboard_release_url = "https://github.com/AstrBotDevs/AstrBot/releases/latest/download/dist.zip"
+                        else:
+                            dashboard_release_url = f"https://github.com/AstrBotDevs/AstrBot/releases/download/{version}/dist.zip"
+                        if proxy:
+                            dashboard_release_url = f"{proxy}/{dashboard_release_url}"
+                        await download_file(
+                            dashboard_release_url,
+                            str(zip_path),
+                            show_progress=True,
                         )
-                        await download_dashboard(
-                            path=path,
-                            extract_path=extract_path,
-                            latest=True,
-                            proxy=proxy,
-                        )
-                        return
-                    raise e
-        else:
-            url = f"https://github.com/AstrBotDevs/astrbot-release-harbour/releases/download/release-{version}/dist.zip"
-            logger.info(f"准备下载指定版本的 AstrBot WebUI: {url}")
-            if proxy:
-                url = f"{proxy}/{url}"
-            await download_file(url, str(zip_path), show_progress=True)
+                    except Exception as e:
+                        if not latest:
+                            logger.warning(
+                                f"下载指定版本({version})失败: {e}，尝试下载最新版本。"
+                            )
+                            await download_dashboard(
+                                path=path,
+                                extract_path=extract_path,
+                                latest=True,
+                                proxy=proxy,
+                            )
+                            return
+                        raise e
+            else:
+                url = f"https://github.com/AstrBotDevs/astrbot-release-harbour/releases/download/release-{version}/dist.zip"
+                logger.info(f"准备下载指定版本的 AstrBot WebUI: {url}")
+                if proxy:
+                    url = f"{proxy}/{url}"
+                await download_file(url, str(zip_path), show_progress=True)
 
-        # 下载完成后存入缓存
+            # 下载完成后存入缓存
         try:
             save_cache_name = None
             if not latest and version:

main.py

@@ -6,13 +6,10 @@ import sys
 from pathlib import Path
 
 import runtime_bootstrap
-
-runtime_bootstrap.initialize_runtime_bootstrap()
-
-from astrbot.core import LogBroker, LogManager, db_helper, logger  # noqa: E402
-from astrbot.core.config.default import VERSION  # noqa: E402
-from astrbot.core.initial_loader import InitialLoader  # noqa: E402
-from astrbot.core.utils.astrbot_path import (  # noqa: E402
+from astrbot.core import LogBroker, LogManager, db_helper, logger
+from astrbot.core.config.default import VERSION
+from astrbot.core.initial_loader import InitialLoader
+from astrbot.core.utils.astrbot_path import (
     get_astrbot_config_path,
     get_astrbot_data_path,
     get_astrbot_knowledge_base_path,
@@ -21,11 +18,14 @@ from astrbot.core.utils.astrbot_path import (  # noqa: E402
     get_astrbot_site_packages_path,
     get_astrbot_temp_path,
 )
-from astrbot.core.utils.io import (  # noqa: E402
+from astrbot.core.utils.io import (
     download_dashboard,
     get_dashboard_version,
 )
 
+runtime_bootstrap.initialize_runtime_bootstrap()
+
+
 # 将父目录添加到 sys.path
 sys.path.append(Path(__file__).parent.as_posix())