至少让插件跑起来了

plugins/napcat_adapter_plugin/src/config/__init__.py

@@ -0,0 +1,5 @@
+from .config import global_config
+
+__all__ = [
+    "global_config",
+]

plugins/napcat_adapter_plugin/src/config/config.py

@@ -0,0 +1,148 @@
+import os
+from dataclasses import dataclass
+from datetime import datetime
+
+import tomlkit
+import shutil
+
+from tomlkit import TOMLDocument
+from tomlkit.items import Table
+from src.common.logger import get_logger
+logger = get_logger("napcat_adapter")
+from rich.traceback import install
+
+from .config_base import ConfigBase
+from .official_configs import (
+    DebugConfig,
+    MaiBotServerConfig,
+    NapcatServerConfig,
+    NicknameConfig,
+    VoiceConfig,
+)
+
+install(extra_lines=3)
+
+TEMPLATE_DIR = "plugins/napcat_adapter_plugin/template"
+CONFIG_DIR = "plugins/napcat_adapter_plugin/config"
+OLD_CONFIG_DIR = "plugins/napcat_adapter_plugin/config/old"
+
+
+def ensure_config_directories():
+    """确保配置目录存在"""
+    os.makedirs(CONFIG_DIR, exist_ok=True)
+    os.makedirs(OLD_CONFIG_DIR, exist_ok=True)
+
+
+def update_config():
+    """更新配置文件，统一使用 config/old 目录进行备份"""
+    # 确保目录存在
+    ensure_config_directories()
+    
+    # 定义文件路径
+    template_path = f"{TEMPLATE_DIR}/template_config.toml"
+    config_path = f"{CONFIG_DIR}/config.toml"
+
+    # 检查配置文件是否存在
+    if not os.path.exists(config_path):
+        logger.info("主配置文件不存在，从模板创建新配置")
+        shutil.copy2(template_path, config_path)
+        logger.info(f"已创建新配置文件: {config_path}")
+        logger.info("程序将退出，请检查配置文件后重启")
+
+    # 读取配置文件和模板文件
+    with open(config_path, "r", encoding="utf-8") as f:
+        old_config = tomlkit.load(f)
+    with open(template_path, "r", encoding="utf-8") as f:
+        new_config = tomlkit.load(f)
+
+    # 检查version是否相同
+    if old_config and "inner" in old_config and "inner" in new_config:
+        old_version = old_config["inner"].get("version")
+        new_version = new_config["inner"].get("version")
+        if old_version and new_version and old_version == new_version:
+            logger.info(f"检测到配置文件版本号相同 (v{old_version})，跳过更新")
+            return
+        else:
+            logger.info(f"检测到版本号不同: 旧版本 v{old_version} -> 新版本 v{new_version}")
+    else:
+        logger.info("已有配置文件未检测到版本号，可能是旧版本。将进行更新")
+
+    # 创建备份文件
+    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+    backup_path = os.path.join(OLD_CONFIG_DIR, f"config.toml.bak.{timestamp}")
+
+    # 备份旧配置文件
+    shutil.copy2(config_path, backup_path)
+    logger.info(f"已备份旧配置文件到: {backup_path}")
+
+    # 复制模板文件到配置目录
+    shutil.copy2(template_path, config_path)
+    logger.info(f"已创建新配置文件: {config_path}")
+
+    def update_dict(target: TOMLDocument | dict, source: TOMLDocument | dict):
+        """将source字典的值更新到target字典中（如果target中存在相同的键）"""
+        for key, value in source.items():
+            # 跳过version字段的更新
+            if key == "version":
+                continue
+            if key in target:
+                if isinstance(value, dict) and isinstance(target[key], (dict, Table)):
+                    update_dict(target[key], value)
+                else:
+                    try:
+                        # 对数组类型进行特殊处理
+                        if isinstance(value, list):
+                            # 如果是空数组，确保它保持为空数组
+                            target[key] = tomlkit.array(str(value)) if value else tomlkit.array()
+                        else:
+                            # 其他类型使用item方法创建新值
+                            target[key] = tomlkit.item(value)
+                    except (TypeError, ValueError):
+                        # 如果转换失败，直接赋值
+                        target[key] = value
+
+    # 将旧配置的值更新到新配置中
+    logger.info("开始合并新旧配置...")
+    update_dict(new_config, old_config)
+
+    # 保存更新后的配置（保留注释和格式）
+    with open(config_path, "w", encoding="utf-8") as f:
+        f.write(tomlkit.dumps(new_config))
+    logger.info("配置文件更新完成，建议检查新配置文件中的内容，以免丢失重要信息")
+
+
+@dataclass
+class Config(ConfigBase):
+    """总配置类"""
+
+    nickname: NicknameConfig
+    napcat_server: NapcatServerConfig
+    maibot_server: MaiBotServerConfig
+    voice: VoiceConfig
+    debug: DebugConfig
+
+
+def load_config(config_path: str) -> Config:
+    """
+    加载配置文件
+    :param config_path: 配置文件路径
+    :return: Config对象
+    """
+    # 读取配置文件
+    with open(config_path, "r", encoding="utf-8") as f:
+        config_data = tomlkit.load(f)
+
+    # 创建Config对象
+    try:
+        return Config.from_dict(config_data)
+    except Exception as e:
+        logger.critical("配置文件解析失败")
+        raise e
+
+
+# 更新配置
+update_config()
+
+logger.info("正在品鉴配置文件...")
+global_config = load_config(config_path=f"{CONFIG_DIR}/config.toml")
+logger.info("非常的新鲜，非常的美味！")

plugins/napcat_adapter_plugin/src/config/config_base.py

@@ -0,0 +1,136 @@
+from dataclasses import dataclass, fields, MISSING
+from typing import TypeVar, Type, Any, get_origin, get_args, Literal, Dict, Union
+
+T = TypeVar("T", bound="ConfigBase")
+
+TOML_DICT_TYPE = {
+    int,
+    float,
+    str,
+    bool,
+    list,
+    dict,
+}
+
+
+@dataclass
+class ConfigBase:
+    """配置类的基类"""
+
+    @classmethod
+    def from_dict(cls: Type[T], data: Dict[str, Any]) -> T:
+        """从字典加载配置字段"""
+        if not isinstance(data, dict):
+            raise TypeError(f"Expected a dictionary, got {type(data).__name__}")
+
+        init_args: Dict[str, Any] = {}
+
+        for f in fields(cls):
+            field_name = f.name
+            field_type = f.type
+            if field_name.startswith("_"):
+                # 跳过以 _ 开头的字段
+                continue
+
+            if field_name not in data:
+                if f.default is not MISSING or f.default_factory is not MISSING:
+                    # 跳过未提供且有默认值/默认构造方法的字段
+                    continue
+                else:
+                    raise ValueError(f"Missing required field: '{field_name}'")
+
+            value = data[field_name]
+            try:
+                init_args[field_name] = cls._convert_field(value, field_type)
+            except TypeError as e:
+                raise TypeError(f"字段 '{field_name}' 出现类型错误: {e}") from e
+            except Exception as e:
+                raise RuntimeError(f"无法将字段 '{field_name}' 转换为目标类型，出现错误: {e}") from e
+
+        return cls(**init_args)
+
+    @classmethod
+    def _convert_field(cls, value: Any, field_type: Type[Any]) -> Any:
+        """
+        转换字段值为指定类型
+
+        1. 对于嵌套的 dataclass，递归调用相应的 from_dict 方法
+        2. 对于泛型集合类型（list, set, tuple），递归转换每个元素
+        3. 对于基础类型（int, str, float, bool），直接转换
+        4. 对于其他类型，尝试直接转换，如果失败则抛出异常
+        """
+        # 如果是嵌套的 dataclass，递归调用 from_dict 方法
+        if isinstance(field_type, type) and issubclass(field_type, ConfigBase):
+            return field_type.from_dict(value)
+
+        field_origin_type = get_origin(field_type)
+        field_args_type = get_args(field_type)
+
+        # 处理泛型集合类型（list, set, tuple）
+        if field_origin_type in {list, set, tuple}:
+            # 检查提供的value是否为list
+            if not isinstance(value, list):
+                raise TypeError(f"Expected an list for {field_type.__name__}, got {type(value).__name__}")
+
+            if field_origin_type is list:
+                return [cls._convert_field(item, field_args_type[0]) for item in value]
+            if field_origin_type is set:
+                return {cls._convert_field(item, field_args_type[0]) for item in value}
+            if field_origin_type is tuple:
+                # 检查提供的value长度是否与类型参数一致
+                if len(value) != len(field_args_type):
+                    raise TypeError(
+                        f"Expected {len(field_args_type)} items for {field_type.__name__}, got {len(value)}"
+                    )
+                return tuple(cls._convert_field(item, arg_type) for item, arg_type in zip(value, field_args_type))
+
+        if field_origin_type is dict:
+            # 检查提供的value是否为dict
+            if not isinstance(value, dict):
+                raise TypeError(f"Expected a dictionary for {field_type.__name__}, got {type(value).__name__}")
+
+            # 检查字典的键值类型
+            if len(field_args_type) != 2:
+                raise TypeError(f"Expected a dictionary with two type arguments for {field_type.__name__}")
+            key_type, value_type = field_args_type
+
+            return {cls._convert_field(k, key_type): cls._convert_field(v, value_type) for k, v in value.items()}
+
+        # 处理Optional类型
+        if field_origin_type is Union:  # assert get_origin(Optional[Any]) is Union
+            if value is None:
+                return None
+            # 如果有数据，检查实际类型
+            if type(value) not in field_args_type:
+                raise TypeError(f"Expected {field_args_type} for {field_type.__name__}, got {type(value).__name__}")
+            return cls._convert_field(value, field_args_type[0])
+
+        # 处理int, str, float, bool等基础类型
+        if field_origin_type is None:
+            if isinstance(value, field_type):
+                return field_type(value)
+            else:
+                raise TypeError(f"Expected {field_type.__name__}, got {type(value).__name__}")
+
+        # 处理Literal类型
+        if field_origin_type is Literal:
+            # 获取Literal的允许值
+            allowed_values = get_args(field_type)
+            if value in allowed_values:
+                return value
+            else:
+                raise TypeError(f"Value '{value}' is not in allowed values {allowed_values} for Literal type")
+
+        # 处理其他类型
+        if field_type is Any:
+            return value
+
+        # 其他类型直接转换
+        try:
+            return field_type(value)
+        except (ValueError, TypeError) as e:
+            raise TypeError(f"无法将 {type(value).__name__} 转换为 {field_type.__name__}") from e
+
+    def __str__(self):
+        """返回配置类的字符串表示"""
+        return f"{self.__class__.__name__}({', '.join(f'{f.name}={getattr(self, f.name)}' for f in fields(self))})"

plugins/napcat_adapter_plugin/src/config/config_utils.py

@@ -0,0 +1,146 @@
+"""
+配置文件工具模块
+提供统一的配置文件生成和管理功能
+"""
+import os
+import shutil
+from pathlib import Path
+from datetime import datetime
+from typing import Optional
+
+from src.common.logger import get_logger
+logger = get_logger("napcat_adapter")
+
+
+def ensure_config_directories():
+    """确保配置目录存在"""
+    os.makedirs("config", exist_ok=True)
+    os.makedirs("config/old", exist_ok=True)
+
+
+def create_config_from_template(
+    config_path: str,
+    template_path: str,
+    config_name: str = "配置文件",
+    should_exit: bool = True
+) -> bool:
+    """
+    从模板创建配置文件的统一函数
+    
+    Args:
+        config_path: 配置文件路径
+        template_path: 模板文件路径
+        config_name: 配置文件名称（用于日志显示）
+        should_exit: 创建后是否退出程序
+        
+    Returns:
+        bool: 是否成功创建配置文件
+    """
+    try:
+        # 确保配置目录存在
+        ensure_config_directories()
+        
+        config_path_obj = Path(config_path)
+        template_path_obj = Path(template_path)
+        
+        # 检查配置文件是否存在
+        if config_path_obj.exists():
+            return False  # 配置文件已存在，无需创建
+        
+        logger.info(f"{config_name}不存在，从模板创建新配置")
+        
+        # 检查模板文件是否存在
+        if not template_path_obj.exists():
+            logger.error(f"模板文件不存在: {template_path}")
+            if should_exit:
+                logger.critical("无法创建配置文件，程序退出")
+                quit(1)
+            return False
+        
+        # 确保配置文件目录存在
+        config_path_obj.parent.mkdir(parents=True, exist_ok=True)
+        
+        # 复制模板文件到配置目录
+        shutil.copy2(template_path_obj, config_path_obj)
+        logger.info(f"已创建新{config_name}: {config_path}")
+        
+        if should_exit:
+            logger.info("程序将退出，请检查配置文件后重启")
+            quit(0)
+        
+        return True
+        
+    except Exception as e:
+        logger.error(f"创建{config_name}失败: {e}")
+        if should_exit:
+            logger.critical("无法创建配置文件，程序退出")
+            quit(1)
+        return False
+
+
+def create_default_config_dict(default_values: dict, config_path: str, config_name: str = "配置文件") -> bool:
+    """
+    创建默认配置文件（使用字典数据）
+    
+    Args:
+        default_values: 默认配置值字典
+        config_path: 配置文件路径
+        config_name: 配置文件名称（用于日志显示）
+        
+    Returns:
+        bool: 是否成功创建配置文件
+    """
+    try:
+        import tomlkit
+        
+        config_path_obj = Path(config_path)
+        
+        # 确保配置文件目录存在
+        config_path_obj.parent.mkdir(parents=True, exist_ok=True)
+        
+        # 写入默认配置
+        with open(config_path_obj, "w", encoding="utf-8") as f:
+            tomlkit.dump(default_values, f)
+        
+        logger.info(f"已创建默认{config_name}: {config_path}")
+        return True
+        
+    except Exception as e:
+        logger.error(f"创建默认{config_name}失败: {e}")
+        return False
+
+
+def backup_config_file(config_path: str, backup_dir: str = "config/old") -> Optional[str]:
+    """
+    备份配置文件
+    
+    Args:
+        config_path: 要备份的配置文件路径
+        backup_dir: 备份目录
+        
+    Returns:
+        Optional[str]: 备份文件路径，失败时返回None
+    """
+    try:
+        config_path_obj = Path(config_path)
+        if not config_path_obj.exists():
+            return None
+        
+        # 确保备份目录存在
+        backup_dir_obj = Path(backup_dir)
+        backup_dir_obj.mkdir(parents=True, exist_ok=True)
+        
+        # 创建备份文件名
+        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+        backup_filename = f"{config_path_obj.stem}.toml.bak.{timestamp}"
+        backup_path = backup_dir_obj / backup_filename
+        
+        # 备份文件
+        shutil.copy2(config_path_obj, backup_path)
+        logger.info(f"已备份配置文件到: {backup_path}")
+        
+        return str(backup_path)
+        
+    except Exception as e:
+        logger.error(f"备份配置文件失败: {e}")
+        return None

plugins/napcat_adapter_plugin/src/config/features_config.py

@@ -0,0 +1,359 @@
+import asyncio
+from dataclasses import dataclass, field
+from typing import Literal, Optional
+from pathlib import Path
+import tomlkit
+from src.common.logger import get_logger
+logger = get_logger("napcat_adapter")
+from .config_base import ConfigBase
+from .config_utils import create_config_from_template, create_default_config_dict
+
+
+@dataclass
+class FeaturesConfig(ConfigBase):
+    """功能配置类"""
+    
+    group_list_type: Literal["whitelist", "blacklist"] = "whitelist"
+    """群聊列表类型 白名单/黑名单"""
+    
+    group_list: list[int] = field(default_factory=list)
+    """群聊列表"""
+    
+    private_list_type: Literal["whitelist", "blacklist"] = "whitelist"
+    """私聊列表类型 白名单/黑名单"""
+    
+    private_list: list[int] = field(default_factory=list)
+    """私聊列表"""
+    
+    ban_user_id: list[int] = field(default_factory=list)
+    """被封禁的用户ID列表，封禁后将无法与其进行交互"""
+    
+    ban_qq_bot: bool = False
+    """是否屏蔽QQ官方机器人，若为True，则所有QQ官方机器人将无法与MaiMCore进行交互"""
+    
+    enable_poke: bool = True
+    """是否启用戳一戳功能"""
+    
+    ignore_non_self_poke: bool = False
+    """是否无视不是针对自己的戳一戳"""
+    
+    poke_debounce_seconds: int = 3
+    """戳一戳防抖时间（秒），在指定时间内第二次针对机器人的戳一戳将被忽略"""
+
+    enable_reply_at: bool = True
+    """是否启用引用回复时艾特用户的功能"""
+
+    reply_at_rate: float = 0.5
+    """引用回复时艾特用户的几率 (0.0 ~ 1.0)"""
+    
+    enable_video_analysis: bool = True
+    """是否启用视频识别功能"""
+    
+    max_video_size_mb: int = 100
+    """视频文件最大大小限制（MB）"""
+    
+    download_timeout: int = 60
+    """视频下载超时时间（秒）"""
+    
+    supported_formats: list[str] = field(default_factory=lambda: ["mp4", "avi", "mov", "mkv", "flv", "wmv", "webm"])
+    """支持的视频格式"""
+    
+    # 消息缓冲配置
+    enable_message_buffer: bool = True
+    """是否启用消息缓冲合并功能"""
+    
+    message_buffer_enable_group: bool = True
+    """是否启用群消息缓冲合并"""
+    
+    message_buffer_enable_private: bool = True
+    """是否启用私聊消息缓冲合并"""
+    
+    message_buffer_interval: float = 3.0
+    """消息合并间隔时间（秒），在此时间内的连续消息将被合并"""
+    
+    message_buffer_initial_delay: float = 0.5
+    """消息缓冲初始延迟（秒），收到第一条消息后等待此时间开始合并"""
+    
+    message_buffer_max_components: int = 50
+    """单个会话最大缓冲消息组件数量，超过此数量将强制合并"""
+    
+    message_buffer_block_prefixes: list[str] = field(default_factory=lambda: ["/", "!", "！", ".", "。", "#", "%"])
+    """消息缓冲屏蔽前缀，以这些前缀开头的消息不会被缓冲"""
+
+
+class FeaturesManager:
+    """功能管理器，支持热重载"""
+    
+    def __init__(self, config_path: str = "plugins/napcat_adapter_plugin/config/features.toml"):
+        self.config_path = Path(config_path)
+        self.config: Optional[FeaturesConfig] = None
+        self._file_watcher_task: Optional[asyncio.Task] = None
+        self._last_modified: Optional[float] = None
+        self._callbacks: list = []
+        
+    def add_reload_callback(self, callback):
+        """添加配置重载回调函数"""
+        self._callbacks.append(callback)
+        
+    def remove_reload_callback(self, callback):
+        """移除配置重载回调函数"""
+        if callback in self._callbacks:
+            self._callbacks.remove(callback)
+    
+    async def _notify_callbacks(self):
+        """通知所有回调函数配置已重载"""
+        for callback in self._callbacks:
+            try:
+                if asyncio.iscoroutinefunction(callback):
+                    await callback(self.config)
+                else:
+                    callback(self.config)
+            except Exception as e:
+                logger.error(f"配置重载回调执行失败: {e}")
+    
+    def load_config(self) -> FeaturesConfig:
+        """加载功能配置文件"""
+        try:
+            # 检查配置文件是否存在，如果不存在则创建并退出程序
+            if not self.config_path.exists():
+                logger.info(f"功能配置文件不存在: {self.config_path}")
+                self._create_default_config()
+                # 配置文件创建后程序应该退出，让用户检查配置
+                logger.info("程序将退出，请检查功能配置文件后重启")
+                quit(0)
+            
+            with open(self.config_path, "r", encoding="utf-8") as f:
+                config_data = tomlkit.load(f)
+            
+            self.config = FeaturesConfig.from_dict(config_data)
+            self._last_modified = self.config_path.stat().st_mtime
+            logger.info(f"功能配置加载成功: {self.config_path}")
+            return self.config
+            
+        except Exception as e:
+            logger.error(f"功能配置加载失败: {e}")
+            logger.critical("无法加载功能配置文件，程序退出")
+            quit(1)
+    
+    def _create_default_config(self):
+        """创建默认功能配置文件"""
+        template_path = "template/features_template.toml"
+        
+        # 尝试从模板创建配置文件
+        if create_config_from_template(
+            str(self.config_path),
+            template_path,
+            "功能配置文件",
+            should_exit=False  # 不在这里退出，由调用方决定
+        ):
+            return
+        
+        # 如果模板文件不存在，创建基本配置
+        logger.info("模板文件不存在，创建基本功能配置")
+        default_config = {
+            "group_list_type": "whitelist",
+            "group_list": [],
+            "private_list_type": "whitelist",
+            "private_list": [],
+            "ban_user_id": [],
+            "ban_qq_bot": False,
+            "enable_poke": True,
+            "ignore_non_self_poke": False,
+            "poke_debounce_seconds": 3,
+            "enable_reply_at": True,
+            "reply_at_rate": 0.5,
+            "enable_video_analysis": True,
+            "max_video_size_mb": 100,
+            "download_timeout": 60,
+            "supported_formats": ["mp4", "avi", "mov", "mkv", "flv", "wmv", "webm"],
+            # 消息缓冲配置
+            "enable_message_buffer": True,
+            "message_buffer_enable_group": True,
+            "message_buffer_enable_private": True,
+            "message_buffer_interval": 3.0,
+            "message_buffer_initial_delay": 0.5,
+            "message_buffer_max_components": 50,
+            "message_buffer_block_prefixes": ["/", "!", "！", ".", "。", "#", "%"]
+        }
+        
+        if not create_default_config_dict(default_config, str(self.config_path), "功能配置文件"):
+            logger.critical("无法创建功能配置文件")
+            quit(1)
+    
+    async def reload_config(self) -> bool:
+        """重新加载配置文件"""
+        try:
+            if not self.config_path.exists():
+                logger.warning(f"功能配置文件不存在，无法重载: {self.config_path}")
+                return False
+            
+            current_modified = self.config_path.stat().st_mtime
+            if self._last_modified and current_modified <= self._last_modified:
+                return False  # 文件未修改
+            
+            old_config = self.config
+            new_config = self.load_config()
+            
+            # 检查配置是否真的发生了变化
+            if old_config and self._configs_equal(old_config, new_config):
+                return False
+            
+            logger.info("功能配置已重载")
+            await self._notify_callbacks()
+            return True
+            
+        except Exception as e:
+            logger.error(f"功能配置重载失败: {e}")
+            return False
+    
+    def _configs_equal(self, config1: FeaturesConfig, config2: FeaturesConfig) -> bool:
+        """比较两个配置是否相等"""
+        return (
+            config1.group_list_type == config2.group_list_type and
+            set(config1.group_list) == set(config2.group_list) and
+            config1.private_list_type == config2.private_list_type and
+            set(config1.private_list) == set(config2.private_list) and
+            set(config1.ban_user_id) == set(config2.ban_user_id) and
+            config1.ban_qq_bot == config2.ban_qq_bot and
+            config1.enable_poke == config2.enable_poke and
+            config1.ignore_non_self_poke == config2.ignore_non_self_poke and
+            config1.poke_debounce_seconds == config2.poke_debounce_seconds and
+            config1.enable_reply_at == config2.enable_reply_at and
+            config1.reply_at_rate == config2.reply_at_rate and
+            config1.enable_video_analysis == config2.enable_video_analysis and
+            config1.max_video_size_mb == config2.max_video_size_mb and
+            config1.download_timeout == config2.download_timeout and
+            set(config1.supported_formats) == set(config2.supported_formats) and
+            # 消息缓冲配置比较
+            config1.enable_message_buffer == config2.enable_message_buffer and
+            config1.message_buffer_enable_group == config2.message_buffer_enable_group and
+            config1.message_buffer_enable_private == config2.message_buffer_enable_private and
+            config1.message_buffer_interval == config2.message_buffer_interval and
+            config1.message_buffer_initial_delay == config2.message_buffer_initial_delay and
+            config1.message_buffer_max_components == config2.message_buffer_max_components and
+            set(config1.message_buffer_block_prefixes) == set(config2.message_buffer_block_prefixes)
+        )
+    
+    async def start_file_watcher(self, check_interval: float = 1.0):
+        """启动文件监控，定期检查配置文件变化"""
+        if self._file_watcher_task and not self._file_watcher_task.done():
+            logger.warning("文件监控已在运行")
+            return
+        
+        self._file_watcher_task = asyncio.create_task(
+            self._file_watcher_loop(check_interval)
+        )
+        logger.info(f"功能配置文件监控已启动，检查间隔: {check_interval}秒")
+    
+    async def stop_file_watcher(self):
+        """停止文件监控"""
+        if self._file_watcher_task and not self._file_watcher_task.done():
+            self._file_watcher_task.cancel()
+            try:
+                await self._file_watcher_task
+            except asyncio.CancelledError:
+                pass
+            logger.info("功能配置文件监控已停止")
+    
+    async def _file_watcher_loop(self, check_interval: float):
+        """文件监控循环"""
+        while True:
+            try:
+                await asyncio.sleep(check_interval)
+                await self.reload_config()
+            except asyncio.CancelledError:
+                break
+            except Exception as e:
+                logger.error(f"文件监控循环出错: {e}")
+                await asyncio.sleep(check_interval)
+    
+    def get_config(self) -> FeaturesConfig:
+        """获取当前功能配置"""
+        if self.config is None:
+            return self.load_config()
+        return self.config
+    
+    def is_group_allowed(self, group_id: int) -> bool:
+        """检查群聊是否被允许"""
+        config = self.get_config()
+        if config.group_list_type == "whitelist":
+            return group_id in config.group_list
+        else:  # blacklist
+            return group_id not in config.group_list
+    
+    def is_private_allowed(self, user_id: int) -> bool:
+        """检查私聊是否被允许"""
+        config = self.get_config()
+        if config.private_list_type == "whitelist":
+            return user_id in config.private_list
+        else:  # blacklist
+            return user_id not in config.private_list
+    
+    def is_user_banned(self, user_id: int) -> bool:
+        """检查用户是否被全局禁止"""
+        config = self.get_config()
+        return user_id in config.ban_user_id
+    
+    def is_qq_bot_banned(self) -> bool:
+        """检查是否禁止QQ官方机器人"""
+        config = self.get_config()
+        return config.ban_qq_bot
+    
+    def is_poke_enabled(self) -> bool:
+        """检查戳一戳功能是否启用"""
+        config = self.get_config()
+        return config.enable_poke
+    
+    def is_non_self_poke_ignored(self) -> bool:
+        """检查是否忽略非自己戳一戳"""
+        config = self.get_config()
+        return config.ignore_non_self_poke
+    
+    def is_message_buffer_enabled(self) -> bool:
+        """检查消息缓冲功能是否启用"""
+        config = self.get_config()
+        return config.enable_message_buffer
+    
+    def is_message_buffer_group_enabled(self) -> bool:
+        """检查群消息缓冲是否启用"""
+        config = self.get_config()
+        return config.message_buffer_enable_group
+    
+    def is_message_buffer_private_enabled(self) -> bool:
+        """检查私聊消息缓冲是否启用"""
+        config = self.get_config()
+        return config.message_buffer_enable_private
+    
+    def get_message_buffer_interval(self) -> float:
+        """获取消息缓冲间隔时间"""
+        config = self.get_config()
+        return config.message_buffer_interval
+    
+    def get_message_buffer_initial_delay(self) -> float:
+        """获取消息缓冲初始延迟"""
+        config = self.get_config()
+        return config.message_buffer_initial_delay
+    
+    def get_message_buffer_max_components(self) -> int:
+        """获取消息缓冲最大组件数量"""
+        config = self.get_config()
+        return config.message_buffer_max_components
+    
+    def is_message_buffer_group_enabled(self) -> bool:
+        """检查是否启用群聊消息缓冲"""
+        config = self.get_config()
+        return config.message_buffer_enable_group
+    
+    def is_message_buffer_private_enabled(self) -> bool:
+        """检查是否启用私聊消息缓冲"""
+        config = self.get_config()
+        return config.message_buffer_enable_private
+    
+    def get_message_buffer_block_prefixes(self) -> list[str]:
+        """获取消息缓冲屏蔽前缀列表"""
+        config = self.get_config()
+        return config.message_buffer_block_prefixes
+
+
+# 全局功能管理器实例
+features_manager = FeaturesManager()

plugins/napcat_adapter_plugin/src/config/migrate_features.py

@@ -0,0 +1,194 @@
+"""
+功能配置迁移脚本
+用于将旧的配置文件中的聊天、权限、视频处理等设置迁移到新的独立功能配置文件
+"""
+
+import os
+import shutil
+from pathlib import Path
+import tomlkit
+from src.common.logger import get_logger
+logger = get_logger("napcat_adapter")
+
+
+def migrate_features_from_config(old_config_path: str = "plugins/napcat_adapter_plugin/config/config.toml",
+                                new_features_path: str = "plugins/napcat_adapter_plugin/config/features.toml",
+                                template_path: str = "plugins/napcat_adapter_plugin/template/features_template.toml"):
+    """
+    从旧配置文件迁移功能设置到新的功能配置文件
+    
+    Args:
+        old_config_path: 旧配置文件路径
+        new_features_path: 新功能配置文件路径
+        template_path: 功能配置模板路径
+    """
+    try:
+        # 检查旧配置文件是否存在
+        if not os.path.exists(old_config_path):
+            logger.warning(f"旧配置文件不存在: {old_config_path}")
+            return False
+        
+        # 读取旧配置文件
+        with open(old_config_path, "r", encoding="utf-8") as f:
+            old_config = tomlkit.load(f)
+        
+        # 检查是否有chat配置段和video配置段
+        chat_config = old_config.get("chat", {})
+        video_config = old_config.get("video", {})
+        
+        # 检查是否有权限相关配置
+        permission_keys = ["group_list_type", "group_list", "private_list_type", 
+                          "private_list", "ban_user_id", "ban_qq_bot", 
+                          "enable_poke", "ignore_non_self_poke", "poke_debounce_seconds"]
+        video_keys = ["enable_video_analysis", "max_video_size_mb", "download_timeout", "supported_formats"]
+        
+        has_permission_config = any(key in chat_config for key in permission_keys)
+        has_video_config = any(key in video_config for key in video_keys)
+        
+        if not has_permission_config and not has_video_config:
+            logger.info("旧配置文件中没有找到功能相关配置，无需迁移")
+            return False
+        
+        # 确保新功能配置目录存在
+        new_features_dir = Path(new_features_path).parent
+        new_features_dir.mkdir(parents=True, exist_ok=True)
+        
+        # 如果新功能配置文件已存在，先备份
+        if os.path.exists(new_features_path):
+            backup_path = f"{new_features_path}.backup"
+            shutil.copy2(new_features_path, backup_path)
+            logger.info(f"已备份现有功能配置文件到: {backup_path}")
+        
+        # 创建新的功能配置
+        new_features_config = {
+            "group_list_type": chat_config.get("group_list_type", "whitelist"),
+            "group_list": chat_config.get("group_list", []),
+            "private_list_type": chat_config.get("private_list_type", "whitelist"),
+            "private_list": chat_config.get("private_list", []),
+            "ban_user_id": chat_config.get("ban_user_id", []),
+            "ban_qq_bot": chat_config.get("ban_qq_bot", False),
+            "enable_poke": chat_config.get("enable_poke", True),
+            "ignore_non_self_poke": chat_config.get("ignore_non_self_poke", False),
+            "poke_debounce_seconds": chat_config.get("poke_debounce_seconds", 3),
+            "enable_video_analysis": video_config.get("enable_video_analysis", True),
+            "max_video_size_mb": video_config.get("max_video_size_mb", 100),
+            "download_timeout": video_config.get("download_timeout", 60),
+            "supported_formats": video_config.get("supported_formats", ["mp4", "avi", "mov", "mkv", "flv", "wmv", "webm"])
+        }
+        
+        # 写入新的功能配置文件
+        with open(new_features_path, "w", encoding="utf-8") as f:
+            tomlkit.dump(new_features_config, f)
+        
+        logger.info(f"功能配置已成功迁移到: {new_features_path}")
+        
+        # 显示迁移的配置内容
+        logger.info("迁移的配置内容:")
+        for key, value in new_features_config.items():
+            logger.info(f"  {key}: {value}")
+        
+        return True
+        
+    except Exception as e:
+        logger.error(f"功能配置迁移失败: {e}")
+        return False
+
+
+def remove_features_from_old_config(config_path: str = "plugins/napcat_adapter_plugin/config/config.toml"):
+    """
+    从旧配置文件中移除功能相关配置，并将旧配置移动到 config/old/ 目录
+    
+    Args:
+        config_path: 配置文件路径
+    """
+    try:
+        if not os.path.exists(config_path):
+            logger.warning(f"配置文件不存在: {config_path}")
+            return False
+        
+        # 确保 config/old 目录存在
+        old_config_dir = "plugins/napcat_adapter_plugin/config/old"
+        os.makedirs(old_config_dir, exist_ok=True)
+        
+        # 备份原配置文件到 config/old 目录
+        old_config_path = os.path.join(old_config_dir, "config_with_features.toml")
+        shutil.copy2(config_path, old_config_path)
+        logger.info(f"已备份包含功能配置的原文件到: {old_config_path}")
+        
+        # 读取配置文件
+        with open(config_path, "r", encoding="utf-8") as f:
+            config = tomlkit.load(f)
+        
+        # 移除chat段中的功能相关配置
+        removed_keys = []
+        if "chat" in config:
+            chat_config = config["chat"]
+            permission_keys = ["group_list_type", "group_list", "private_list_type",
+                              "private_list", "ban_user_id", "ban_qq_bot",
+                              "enable_poke", "ignore_non_self_poke", "poke_debounce_seconds"]
+            
+            for key in permission_keys:
+                if key in chat_config:
+                    del chat_config[key]
+                    removed_keys.append(key)
+            
+            if removed_keys:
+                logger.info(f"已从chat配置段中移除功能相关配置: {removed_keys}")
+        
+        # 移除video段中的配置
+        if "video" in config:
+            video_config = config["video"]
+            video_keys = ["enable_video_analysis", "max_video_size_mb", "download_timeout", "supported_formats"]
+            
+            video_removed_keys = []
+            for key in video_keys:
+                if key in video_config:
+                    del video_config[key]
+                    video_removed_keys.append(key)
+            
+            if video_removed_keys:
+                logger.info(f"已从video配置段中移除配置: {video_removed_keys}")
+                removed_keys.extend(video_removed_keys)
+            
+            # 如果video段为空，则删除整个段
+            if not video_config:
+                del config["video"]
+                logger.info("已删除空的video配置段")
+        
+        if removed_keys:
+            logger.info(f"总共移除的配置项: {removed_keys}")
+        
+        # 写回配置文件
+        with open(config_path, "w", encoding="utf-8") as f:
+            f.write(tomlkit.dumps(config))
+        
+        logger.info(f"已更新配置文件: {config_path}")
+        return True
+        
+    except Exception as e:
+        logger.error(f"移除功能配置失败: {e}")
+        return False
+
+
+def auto_migrate_features():
+    """
+    自动执行功能配置迁移
+    """
+    logger.info("开始自动功能配置迁移...")
+    
+    # 执行迁移
+    if migrate_features_from_config():
+        logger.info("功能配置迁移成功")
+        
+        # 询问是否要从旧配置文件中移除功能配置
+        logger.info("功能配置已迁移到独立文件，建议从主配置文件中移除相关配置")
+        # 在实际使用中，这里可以添加用户确认逻辑
+        # 为了自动化，这里直接执行移除
+        remove_features_from_old_config()
+        
+    else:
+        logger.info("功能配置迁移跳过或失败")
+
+
+if __name__ == "__main__":
+    auto_migrate_features()

plugins/napcat_adapter_plugin/src/config/official_configs.py

@@ -0,0 +1,67 @@
+from dataclasses import dataclass, field
+from typing import Literal
+
+from .config_base import ConfigBase
+
+"""
+须知：
+1. 本文件中记录了所有的配置项
+2. 所有新增的class都需要继承自ConfigBase
+3. 所有新增的class都应在config.py中的Config类中添加字段
+4. 对于新增的字段，若为可选项，则应在其后添加field()并设置default_factory或default
+"""
+
+ADAPTER_PLATFORM = "qq"
+
+
+@dataclass
+class NicknameConfig(ConfigBase):
+    nickname: str
+    """机器人昵称"""
+
+
+@dataclass
+class NapcatServerConfig(ConfigBase):
+    mode: Literal["reverse", "forward"] = "reverse"
+    """连接模式：reverse=反向连接(作为服务器), forward=正向连接(作为客户端)"""
+
+    host: str = "localhost"
+    """主机地址"""
+
+    port: int = 8095
+    """端口号"""
+
+    url: str = ""
+    """正向连接时的完整WebSocket URL，如 ws://localhost:8080/ws"""
+
+    access_token: str = ""
+    """WebSocket 连接的访问令牌，用于身份验证"""
+
+    heartbeat_interval: int = 30
+    """心跳间隔时间，单位为秒"""
+
+
+@dataclass
+class MaiBotServerConfig(ConfigBase):
+    platform_name: str = field(default=ADAPTER_PLATFORM, init=False)
+    """平台名称，“qq”"""
+
+    host: str = "localhost"
+    """MaiMCore的主机地址"""
+
+    port: int = 8000
+    """MaiMCore的端口号"""
+
+
+
+
+@dataclass
+class VoiceConfig(ConfigBase):
+    use_tts: bool = False
+    """是否启用TTS功能"""
+
+
+@dataclass
+class DebugConfig(ConfigBase):
+    level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO"
+    """日志级别，默认为INFO"""