124 lines
3.7 KiB
Python
124 lines
3.7 KiB
Python
import logging
|
|
import threading
|
|
from typing import Dict, Any, Tuple, Optional, List
|
|
|
|
from plugin_common.plugin_interface import PluginInterface, PluginStatus
|
|
from .dashboard_server import DashboardServer
|
|
|
|
|
|
class StatsDashboardPlugin(PluginInterface):
|
|
"""统计看板插件"""
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return "统计看板"
|
|
|
|
@property
|
|
def version(self) -> str:
|
|
return "1.0.0"
|
|
|
|
@property
|
|
def description(self) -> str:
|
|
return "提供插件使用统计数据的可视化界面"
|
|
|
|
@property
|
|
def author(self) -> str:
|
|
return "Trae AI"
|
|
|
|
@property
|
|
def command_prefix(self) -> Optional[str]:
|
|
return "" # 不需要前缀，直接匹配命令
|
|
|
|
@property
|
|
def commands(self) -> List[str]:
|
|
return []
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.logger = logging.getLogger("StatsDashboard")
|
|
self.LOG.info(f"正在初始化 {self.name} 插件...")
|
|
self.server = None
|
|
self.server_thread = None
|
|
|
|
def initialize(self, config: Dict[str, Any]) -> bool:
|
|
|
|
if not self._config["enable"]:
|
|
self.logger.info("统计看板插件已禁用")
|
|
return False
|
|
|
|
# 创建看板服务器
|
|
self.server = DashboardServer(
|
|
host=self._config["host"],
|
|
port=self._config["port"],
|
|
username=self._config["username"],
|
|
password=self._config["password"]
|
|
)
|
|
|
|
# 如果配置为自动启动，则启动服务器
|
|
if self._config["auto_start"]:
|
|
self.start_server()
|
|
|
|
return True
|
|
|
|
def start_server(self) -> bool:
|
|
"""启动看板服务器"""
|
|
if self.server_thread and self.server_thread.is_alive():
|
|
self.logger.warning("服务器已经在运行中")
|
|
return False
|
|
|
|
try:
|
|
self.server_thread = threading.Thread(target=self.server.run, daemon=True)
|
|
self.server_thread.start()
|
|
self.logger.info(f"统计看板服务器已启动，访问地址: http://{self._config['host']}:{self._config['port']}")
|
|
return True
|
|
except Exception as e:
|
|
self.logger.error(f"启动统计看板服务器失败: {e}")
|
|
return False
|
|
|
|
def stop_server(self) -> bool:
|
|
"""停止看板服务器"""
|
|
if not self.server_thread or not self.server_thread.is_alive():
|
|
self.logger.warning("服务器未运行")
|
|
return False
|
|
|
|
try:
|
|
# 修改：添加超时处理和错误检查
|
|
self.server.stop()
|
|
# 等待线程结束，但设置超时
|
|
self.server_thread.join(timeout=5)
|
|
|
|
# 检查线程是否真的结束了
|
|
if self.server_thread.is_alive():
|
|
self.logger.warning("服务器停止超时，可能需要手动终止")
|
|
return False
|
|
|
|
self.logger.info("统计看板服务器已停止")
|
|
return True
|
|
except Exception as e:
|
|
self.logger.error(f"停止统计看板服务器失败: {e}")
|
|
return False
|
|
|
|
def match_command(self, content: str) -> bool:
|
|
"""匹配命令"""
|
|
return content.strip().startswith("/stats")
|
|
|
|
def process_message(self, message: Dict[str, Any]) -> Tuple[bool, str]:
|
|
return False, ""
|
|
|
|
def shutdown(self) -> None:
|
|
"""关闭插件"""
|
|
if self.server:
|
|
self.stop_server()
|
|
|
|
def start(self) -> bool:
|
|
"""启动插件"""
|
|
self.LOG.info(f"[{self.name}] 插件已启动")
|
|
self.status = PluginStatus.RUNNING
|
|
return True
|
|
|
|
def stop(self) -> bool:
|
|
"""停止插件"""
|
|
self.LOG.info(f"[{self.name}] 插件已停止")
|
|
self.status = PluginStatus.STOPPED
|
|
return True
|