Merge branch 'feat/optional-backend' into master

Author: LIghtJUNction

.python-version

@@ -1 +1 @@
-3.12
+3.12

astrbot/cli/commands/cmd_init.py

@@ -34,8 +34,13 @@ async def initialize_astrbot(astrbot_root: Path) -> None:
     for name, path in paths.items():
         path.mkdir(parents=True, exist_ok=True)
         click.echo(f"{'Created' if not path.exists() else 'Directory exists'}: {path}")
-
-    await check_dashboard(astrbot_root / "data")
+    if click.confirm(
+        "是否需要集成式 WebUI？（个人电脑推荐，服务器不推荐）",
+        default=True,
+    ):
+        await check_dashboard(astrbot_root / "data")
+    else:
+        click.echo("你可以使用在线面版（v4.14.4+），填写后端地址的方式来控制。")
 
 
 @click.command()

astrbot/cli/commands/cmd_run.py

@@ -15,7 +15,8 @@ async def run_astrbot(astrbot_root: Path) -> None:
     from astrbot.core import LogBroker, LogManager, db_helper, logger
     from astrbot.core.initial_loader import InitialLoader
 
-    await check_dashboard(astrbot_root / "data")
+    if os.environ.get("DASHBOARD_ENABLE") == "True":
+        await check_dashboard(astrbot_root / "data")
 
     log_broker = LogBroker()
     LogManager.set_queue_handler(logger, log_broker)
@@ -27,9 +28,17 @@ async def run_astrbot(astrbot_root: Path) -> None:
 
 
 @click.option("--reload", "-r", is_flag=True, help="插件自动重载")
-@click.option("--port", "-p", help="Astrbot Dashboard端口", required=False, type=str)
+@click.option(
+    "--host", "-H", help="Astrbot Dashboard Host,默认::", required=False, type=str
+)
+@click.option(
+    "--port", "-p", help="Astrbot Dashboard端口,默认6185", required=False, type=str
+)
+@click.option(
+    "--backend-only", is_flag=True, default=False, help="禁用WEBUI,仅启动后端"
+)
 @click.command()
-def run(reload: bool, port: str) -> None:
+def run(reload: bool, host: str, port: str, backend_only: bool) -> None:
     """运行 AstrBot"""
     try:
         os.environ["ASTRBOT_CLI"] = "1"
@@ -43,8 +52,11 @@ def run(reload: bool, port: str) -> None:
         os.environ["ASTRBOT_ROOT"] = str(astrbot_root)
         sys.path.insert(0, str(astrbot_root))
 
-        if port:
+        if port is not None:
             os.environ["DASHBOARD_PORT"] = port
+        if host is not None:
+            os.environ["DASHBOARD_HOST"] = host
+        os.environ["DASHBOARD_ENABLE"] = str(not backend_only)
 
         if reload:
             click.echo("启用插件自动重载")

astrbot/core/config/default.py

@@ -195,7 +195,7 @@
         "username": "astrbot",
         "password": "77b90590a8945a7d36c963981a307dc9",
         "jwt_secret": "",
-        "host": "0.0.0.0",
+        "host": "::",
         "port": 6185,
         "disable_access_log": True,
         "ssl": {

astrbot/core/platform/sources/aiocqhttp/aiocqhttp_platform_adapter.py

@@ -419,7 +419,7 @@ async def _convert_handle_message_event(
     def run(self) -> Awaitable[Any]:
         if not self.host or not self.port:
             logger.warning(
-                "aiocqhttp: 未配置 ws_reverse_host 或 ws_reverse_port，将使用默认值：http://0.0.0.0:6199",
+                "aiocqhttp: 未配置 ws_reverse_host 或 ws_reverse_port，将使用默认值：http://[::]:6199",
             )
             self.host = "0.0.0.0"
             self.port = 6199

astrbot/core/platform/sources/qqofficial_webhook/qo_webhook_server.py

@@ -21,7 +21,7 @@ def __init__(
         self.secret = config["secret"]
         self.port = config.get("port", 6196)
         self.is_sandbox = config.get("is_sandbox", False)
-        self.callback_server_host = config.get("callback_server_host", "0.0.0.0")
+        self.callback_server_host = config.get("callback_server_host", "::")
 
         if isinstance(self.port, str):
             self.port = int(self.port)

astrbot/core/platform/sources/wecom/wecom_adapter.py

@@ -43,7 +43,7 @@ class WecomServer:
     def __init__(self, event_queue: asyncio.Queue, config: dict) -> None:
         self.server = quart.Quart(__name__)
         self.port = int(cast(str, config.get("port")))
-        self.callback_server_host = config.get("callback_server_host", "0.0.0.0")
+        self.callback_server_host = config.get("callback_server_host", "::")
         self.server.add_url_rule(
             "/callback/command",
             view_func=self.verify,
@@ -407,7 +407,7 @@ async def convert_wechat_kf_message(self, msg: dict) -> AstrBotMessage | None:
             abm.message = [Image(file=path, url=path)]
         elif msgtype == "voice":
             media_id = msg.get("voice", {}).get("media_id", "")
-            resp: Response = await asyncio.get_event_loop().run_in_executor(
+            resp = await asyncio.get_event_loop().run_in_executor(
                 None,
                 self.client.media.download,
                 media_id,

astrbot/core/utils/io.py

@@ -1,3 +1,4 @@
+import asyncio
 import base64
 import logging
 import os
@@ -7,6 +8,7 @@
 import time
 import uuid
 import zipfile
+from ipaddress import IPv4Address, IPv6Address, ip_address
 from pathlib import Path
 
 import aiohttp
@@ -206,18 +208,53 @@ def file_to_base64(file_path: str) -> str:
     return "base64://" + base64_str
 
 
-def get_local_ip_addresses():
+def get_local_ip_addresses() -> list[IPv4Address | IPv6Address]:
     net_interfaces = psutil.net_if_addrs()
-    network_ips = []
+    network_ips: list[IPv4Address | IPv6Address] = []
 
-    for interface, addrs in net_interfaces.items():
+    for _, addrs in net_interfaces.items():
         for addr in addrs:
-            if addr.family == socket.AF_INET:  # 使用 socket.AF_INET 代替 psutil.AF_INET
-                network_ips.append(addr.address)
+            if addr.family == socket.AF_INET:
+                network_ips.append(ip_address(addr.address))
+            elif addr.family == socket.AF_INET6:
+                # 过滤掉 IPv6 的 link-local 地址（fe80:...）
+                ip = ip_address(addr.address.split("%")[0])  # 处理带 zone index 的情况
+                if not ip.is_link_local:
+                    network_ips.append(ip)
 
     return network_ips
 
 
+async def get_public_ip_address() -> list[IPv4Address | IPv6Address]:
+    urls = [
+        "https://api64.ipify.org",
+        "https://ident.me",
+        "https://ifconfig.me",
+        "https://icanhazip.com",
+    ]
+    found_ips: dict[int, IPv4Address | IPv6Address] = {}
+
+    async def fetch(session: aiohttp.ClientSession, url: str):
+        try:
+            async with session.get(url, timeout=3) as resp:
+                if resp.status == 200:
+                    raw_ip = (await resp.text()).strip()
+                    ip = ip_address(raw_ip)
+                    if ip.version not in found_ips:
+                        found_ips[ip.version] = ip
+        except Exception as e:
+            # Ignore errors from individual services so that a single failing
+            # endpoint does not prevent discovering the public IP from others.
+            logger.debug("Failed to fetch public IP from %s: %s", url, e)
+
+    async with aiohttp.ClientSession() as session:
+        tasks = [fetch(session, url) for url in urls]
+        await asyncio.gather(*tasks)
+
+    # 返回找到的所有 IP 对象列表
+    return list(found_ips.values())
+
+
 async def get_dashboard_version():
     dist_dir = os.path.join(get_astrbot_data_path(), "dist")
     if os.path.exists(dist_dir):

astrbot/dashboard/routes/__init__.py

@@ -9,16 +9,20 @@
 from .cron import CronRoute
 from .file import FileRoute
 from .knowledge_base import KnowledgeBaseRoute
+from .live_chat import LiveChatRoute
 from .log import LogRoute
 from .open_api import OpenApiRoute
 from .persona import PersonaRoute
 from .platform import PlatformRoute
 from .plugin import PluginRoute
+from .response import Response
+from .route import RouteContext
 from .session_management import SessionManagementRoute
 from .skills import SkillsRoute
 from .stat import StatRoute
 from .static_file import StaticFileRoute
 from .subagent import SubAgentRoute
+from .t2i import T2iRoute
 from .tools import ToolsRoute
 from .update import UpdateRoute
 
@@ -46,4 +50,8 @@
     "ToolsRoute",
     "SkillsRoute",
     "UpdateRoute",
+    "T2iRoute",
+    "LiveChatRoute",
+    "Response",
+    "RouteContext",
 ]

astrbot/dashboard/routes/route.py

@@ -1,4 +1,4 @@
-from dataclasses import dataclass
+from dataclasses import asdict, dataclass
 
 from quart import Quart
 
@@ -57,3 +57,7 @@ def ok(self, data: dict | list | None = None, message: str | None = None):
         self.data = data
         self.message = message
         return self
+
+    def to_json(self):
+        # Return a plain dict so callers can safely wrap with jsonify()
+        return asdict(self)