from abc import ABC, abstractmethod from typing import Optional from dataclasses import dataclass import shutil import zipfile from pathlib import Path from src.modules import user_modules as usermod import src.file_store_api as file_M class MessageContext: """封装消息处理的上下文数据""" def __init__(self, uid: str, gid: Optional[str], raw_message: str,id:str): self.raw_message = raw_message self._processed = False self.response: Optional[str] = None # 核心服务实例化 self.chat_manager = file_M.ChatManager() self.user = usermod.User(user_id=uid) self.rebot_id = id # 动态加载数据 if gid: self.group = usermod.Group(group_id=gid) self.group.current_user = self.user else: self.group = None @dataclass class PluginPermission: access_private: bool = False # 允许处理私聊消息 access_group: bool = True # 允许处理群消息 read_history: bool = False # 允许读取历史记录 from pathlib import Path import toml import os class BasePlugin(ABC): def __init__(self, ctx: MessageContext): self.ctx = ctx self._config_dir = self._get_plugin_config_path() self._config_manager = file_M.ConfigManager(self._config_dir) def _get_plugin_resource(self, resource_path: str) -> bytes: plugin_name = self.__class__.__name__.lower() zip_path = Path("plugins") / f"{plugin_name}.zip" if not zip_path.exists(): raise FileNotFoundError(f"插件ZIP包不存在: {zip_path}") # 遍历可能的ZIP内路径 possible_paths = ( resource_path, # 直接路径(config.toml) f"{plugin_name}/{resource_path}" # 插件子目录路径(myplugin/config.toml) ) with zipfile.ZipFile(zip_path, 'r') as zf: for path in possible_paths: if path in zf.namelist(): return zf.read(path) raise FileNotFoundError(f"文件 '{resource_path}' 不在ZIP包中") def _ensure_config_exists(self): """确保配置文件存在(不存在时从ZIP复制默认配置)""" config_file = Path(self._config_dir) / "config.toml" # 外部配置已存在则直接返回 if config_file.exists(): return # 尝试从ZIP复制默认配置 try: for ext in ['.toml', '.json', '.yaml']: try: config_data = self._get_plugin_resource(f"config{ext}") # 确保配置目录存在 config_file.parent.mkdir(parents=True, exist_ok=True) # 统一保存为.toml格式(或保持原格式) config_file.write_bytes(config_data) print(f"✅ 已将默认配置复制到: {config_file}") break except FileNotFoundError: continue else: print(f"⚠️ 插件包内未找到默认配置文件") except Exception as e: print(f"❌ 初始化配置失败: {str(e)}") @property def config(self) -> dict: """始终读取外部配置文件(确保已通过_ensure_config_exists初始化)""" try: config_data = self._config_manager.load_config("config") except: self._ensure_config_exists() config_data = self._config_manager.load_config("config") return config_data or {} def _get_plugin_config_path(self) -> str: """获取插件配置目录路径(保持原有逻辑)""" plugin_name = self.__class__.__name__.lower() return str(Path("config") / plugin_name) def save_config(self, config_dict: dict) -> bool: """保存配置到外部目录(与之前逻辑一致)""" self._config_manager.update_config({"config": config_dict}) return self._config_manager.save_config()