Merge pull request #574 from Soulter/perf-port-check

🎈 perf: 启动时检查端口占用

astrbot/dashboard/server.py

@@ -2,6 +2,9 @@
 import jwt
 import asyncio
 import os
+import socket
+import sys
+import psutil
 from astrbot.core.config.default import VERSION
 from quart import Quart, request, jsonify, g
 from quart.logging import default_handler
@@ -67,6 +70,47 @@ async def shutdown_trigger_placeholder(self):
             await asyncio.sleep(1)
         logger.info("管理面板已关闭。")
 
+    def check_port_in_use(self, port: int) -> bool:
+        """
+        跨平台检测端口是否被占用
+        """
+        try:
+            # 创建 IPv4 TCP Socket
+            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+            # 设置超时时间
+            sock.settimeout(2)
+            result = sock.connect_ex(('127.0.0.1', port))
+            sock.close()
+            # result 为 0 表示端口被占用
+            return result == 0
+        except Exception as e:
+            logger.warning(f"检查端口 {port} 时发生错误: {str(e)}")
+            # 如果出现异常，保守起见认为端口可能被占用
+            return True
+
+    def get_process_using_port(self, port: int) -> str:
+        """获取占用端口的进程详细信息"""
+        try:
+            for conn in psutil.net_connections(kind='inet'):
+                if conn.laddr.port == port:
+                    try:
+                        process = psutil.Process(conn.pid)
+                        # 获取详细信息
+                        proc_info = [
+                            f"进程名: {process.name()}",
+                            f"PID: {process.pid}",
+                            f"执行路径: {process.exe()}",
+                            f"工作目录: {process.cwd()}",
+                            f"启动命令: {' '.join(process.cmdline())}"
+                        ]
+                        return "\n           ".join(proc_info)
+                    except (psutil.NoSuchProcess, psutil.AccessDenied) as e:
+                        return f"无法获取进程详细信息(可能需要管理员权限): {str(e)}"
+            return "未找到占用进程"
+        except Exception as e:
+            return f"获取进程信息失败: {str(e)}"
+
     def run(self):
         try:
             ip_addr = get_local_ip_addresses()
@@ -76,6 +120,17 @@ def run(self):
         port = self.core_lifecycle.astrbot_config['dashboard'].get("port", 6185)
         if isinstance(port, str):
             port = int(port)
+
+        if self.check_port_in_use(port):
+            process_info = self.get_process_using_port(port)
+            logger.error(f"错误：端口 {port} 已被占用\n"
+                        f"占用信息: \n           {process_info}\n"
+                        f"请确保：\n"
+                        f"1. 没有其他 AstrBot 实例正在运行\n"
+                        f"2. 端口 {port} 没有被其他程序占用\n"
+                        f"3. 如需使用其他端口，请修改配置文件")
+
+            raise Exception(f"端口 {port} 已被占用")
 
         display = f"\n ✨✨✨\n  AstrBot v{VERSION} 管理面板已启动，可访问\n\n"
         display += f"   ➜  本地: http://localhost:{port}\n"

requirements.txt

@@ -18,6 +18,7 @@ apscheduler
 docstring_parser
 aiodocker
 silk-python
+psutil>=5.8.0
 
 lark-oapi
 ormsgpack