Merge branch 'dev' of https://github.com/MoFox-Studio/MoFox_Bot into dev

bot.py

@@ -76,7 +76,7 @@ async def request_shutdown() -> bool:
     try:
         if loop and not loop.is_closed():
             try:
-                loop.run_until_complete(graceful_shutdown())
+                loop.run_until_complete(graceful_shutdown(maibot.main_system))
             except Exception as ge:  # 捕捉优雅关闭时可能发生的错误
                 logger.error(f"优雅关闭时发生错误: {ge}")
                 return False
@@ -97,18 +97,15 @@ def easter_egg():
     logger.info(rainbow_text)
 
 
-async def graceful_shutdown():
+async def graceful_shutdown(main_system_instance):
+    """优雅地关闭所有系统组件"""
     try:
         logger.info("正在优雅关闭麦麦...")
 
-        # 首先停止服务器组件，避免网络连接被强制关闭
-        try:
-            global server
-            if server and hasattr(server, 'shutdown'):
-                logger.info("正在关闭服务器...")
-                await server.shutdown()
-        except Exception as e:
-            logger.warning(f"关闭服务器时出错: {e}")
+        # 停止MainSystem中的组件，它会处理服务器等
+        if main_system_instance and hasattr(main_system_instance, 'shutdown'):
+            logger.info("正在关闭MainSystem...")
+            await main_system_instance.shutdown()
 
         # 停止聊天管理器
         try:
@@ -138,14 +135,6 @@ async def graceful_shutdown():
         except Exception as e:
             logger.warning(f"停止记忆系统时出错: {e}")
 
-        # 停止MainSystem
-        try:
-            global main_system
-            if main_system and hasattr(main_system, 'shutdown'):
-                logger.info("正在停止MainSystem...")
-                await main_system.shutdown()
-        except Exception as e:
-            logger.warning(f"停止MainSystem时出错: {e}")
 
         # 停止所有异步任务
         try:
@@ -178,6 +167,15 @@ async def graceful_shutdown():
         # 关闭日志系统，释放文件句柄
         shutdown_logging()
 
+        # 尝试停止事件循环
+        try:
+            loop = asyncio.get_running_loop()
+            if loop.is_running():
+                loop.stop()
+                logger.info("事件循环已请求停止")
+        except RuntimeError:
+            pass  # 没有正在运行的事件循环
+
     except Exception as e:
         logger.error(f"麦麦关闭失败: {e}", exc_info=True)
 
@@ -305,18 +303,13 @@ if __name__ == "__main__":
         if "loop" in locals() and loop and not loop.is_closed():
             logger.info("开始执行最终关闭流程...")
             try:
-                loop.run_until_complete(graceful_shutdown())
+                # 传递main_system实例
+                loop.run_until_complete(graceful_shutdown(maibot.main_system))
             except Exception as ge:
                 logger.error(f"优雅关闭时发生错误: {ge}")
             loop.close()
             logger.info("事件循环已关闭")
 
-        # 关闭日志系统，释放文件句柄
-        try:
-            shutdown_logging()
-        except Exception as e:
-            print(f"关闭日志系统时出错: {e}")
-
         # 在程序退出前暂停，让你有机会看到输出
         # input("按 Enter 键退出...")  # <--- 添加这行
         sys.exit(exit_code)  # <--- 使用记录的退出码

plugins/bilibli/__init__.py

@@ -1,7 +1,10 @@
-#!/usr/bin/env python3
-"""
-Bilibili 插件包
-提供B站视频观看体验功能，像真实用户一样浏览和评价视频
-"""
+from src.plugin_system.base.plugin_metadata import PluginMetadata
 
-# 插件会通过 @register_plugin 装饰器自动注册，这里不需要额外的导入
+__plugin_meta__ = PluginMetadata(
+    name="Bilibili Plugin",
+    description="A plugin for interacting with Bilibili.",
+    usage="Usage details for Bilibili plugin.",
+    version="1.0.0",
+    author="Your Name",
+    license="MIT",
+)

plugins/echo_example/__init__.py

@@ -0,0 +1,10 @@
+from src.plugin_system.base.plugin_metadata import PluginMetadata
+
+__plugin_meta__ = PluginMetadata(
+    name="Echo Example Plugin",
+    description="An example plugin that echoes messages.",
+    usage="!echo [message]",
+    version="1.0.0",
+    author="Your Name",
+    license="MIT",
+)

plugins/hello_world_plugin/__init__.py

@@ -0,0 +1,10 @@
+from src.plugin_system.base.plugin_metadata import PluginMetadata
+
+__plugin_meta__ = PluginMetadata(
+    name="Hello World Plugin",
+    description="A simple hello world plugin.",
+    usage="!hello",
+    version="1.0.0",
+    author="Your Name",
+    license="MIT",
+)

src/api/message_router.py

@@ -4,7 +4,11 @@ from typing import Literal
 from fastapi import APIRouter, HTTPException, Query
 
 from src.config.config import global_config
-from src.plugin_system.apis import message_api
+from src.plugin_system.apis import message_api, chat_api, person_api
+from src.chat.message_receive.chat_stream import get_chat_manager
+from src.common.logger import get_logger
+
+logger = get_logger("HTTP消息API")
 
 router = APIRouter()
 
@@ -46,3 +50,128 @@ async def get_message_stats(
 
     except Exception as e:
         raise HTTPException(status_code=500, detail=str(e))
+
+
+@router.get("/messages/stats_by_chat")
+async def get_message_stats_by_chat(
+    days: int = Query(1, ge=1, description="指定查询过去多少天的数据"),
+    group_by_user: bool = Query(False, description="是否按用户进行分组统计"),
+    format: bool = Query(False, description="是否格式化输出，包含群聊和用户信息"),
+):
+    """
+    获取BOT在指定天数内按聊天流或按用户统计的消息数据。
+    """
+    try:
+        end_time = time.time()
+        start_time = end_time - (days * 24 * 3600)
+        messages = await message_api.get_messages_by_time(start_time, end_time)
+        bot_qq = str(global_config.bot.qq_account)
+
+        messages = [msg for msg in messages if msg.get("user_id") != bot_qq]
+
+        stats = {}
+
+        for msg in messages:
+            chat_id = msg.get("chat_id", "unknown")
+            user_id = msg.get("user_id")
+
+            if chat_id not in stats:
+                stats[chat_id] = {
+                    "total_stats": {"total": 0},
+                    "user_stats": {}
+                }
+
+            stats[chat_id]["total_stats"]["total"] += 1
+
+            if group_by_user:
+                if user_id not in stats[chat_id]["user_stats"]:
+                    stats[chat_id]["user_stats"][user_id] = 0
+                
+                stats[chat_id]["user_stats"][user_id] += 1
+
+        if not group_by_user:
+            stats = {chat_id: data["total_stats"] for chat_id, data in stats.items()}
+
+        if format:
+            chat_manager = get_chat_manager()
+            formatted_stats = {}
+            for chat_id, data in stats.items():
+                stream = chat_manager.streams.get(chat_id)
+                chat_name = "未知会话"
+                if stream:
+                    if stream.group_info and stream.group_info.group_name:
+                        chat_name = stream.group_info.group_name
+                    elif stream.user_info and stream.user_info.user_nickname:
+                        chat_name = stream.user_info.user_nickname
+                else:
+                    chat_name = f"未知会话 ({chat_id})"
+
+                formatted_data = {
+                    "chat_name": chat_name,
+                    "total_stats": data if not group_by_user else data["total_stats"],
+                }
+
+                if group_by_user and "user_stats" in data:
+                    formatted_data["user_stats"] = {}
+                    for user_id, count in data["user_stats"].items():
+                        person_id = person_api.get_person_id("qq", user_id)
+                        nickname = await person_api.get_person_value(person_id, "nickname", "未知用户")
+                        formatted_data["user_stats"][user_id] = {
+                            "nickname": nickname,
+                            "count": count
+                        }
+                
+                formatted_stats[chat_id] = formatted_data
+            return formatted_stats
+
+        return stats
+
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+@router.get("/messages/bot_stats_by_chat")
+async def get_bot_message_stats_by_chat(
+    days: int = Query(1, ge=1, description="指定查询过去多少天的数据"),
+    format: bool = Query(False, description="是否格式化输出，包含群聊和用户信息"),
+):
+    """
+    获取BOT在指定天数内按聊天流统计的已发送消息数据。
+    """
+    try:
+        end_time = time.time()
+        start_time = end_time - (days * 24 * 3600)
+        messages = await message_api.get_messages_by_time(start_time, end_time)
+        bot_qq = str(global_config.bot.qq_account)
+
+        # 筛选出机器人发送的消息
+        bot_messages = [msg for msg in messages if msg.get("user_id") == bot_qq]
+
+        stats = {}
+        for msg in bot_messages:
+            chat_id = msg.get("chat_id", "unknown")
+            if chat_id not in stats:
+                stats[chat_id] = 0
+            stats[chat_id] += 1
+
+        if format:
+            chat_manager = get_chat_manager()
+            formatted_stats = {}
+            for chat_id, count in stats.items():
+                stream = chat_manager.streams.get(chat_id)
+                chat_name = f"未知会话 ({chat_id})"
+                if stream:
+                    if stream.group_info and stream.group_info.group_name:
+                        chat_name = stream.group_info.group_name
+                    elif stream.user_info and stream.user_info.user_nickname:
+                        chat_name = stream.user_info.user_nickname
+                
+                formatted_stats[chat_id] = {
+                    "chat_name": chat_name,
+                    "count": count
+                }
+            return formatted_stats
+
+        return stats
+
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))

src/chat/emoji_system/emoji_manager.py

@@ -402,6 +402,12 @@ class EmojiManager:
 
         logger.info("启动表情包管理器")
 
+    def shutdown(self) -> None:
+        """关闭EmojiManager，取消正在运行的任务"""
+        if self._scan_task and not self._scan_task.done():
+            self._scan_task.cancel()
+            logger.info("表情包扫描任务已取消")
+
     def initialize(self) -> None:
         """初始化数据库连接和表情目录"""
 

src/chat/memory_system/memory_builder.py

@@ -32,7 +32,10 @@ import time
 from dataclasses import dataclass
 from datetime import datetime
 from enum import Enum
-from typing import Any
+from typing import Any, Type, TypeVar
+
+E = TypeVar("E", bound=Enum)
+
 
 import orjson
 
@@ -49,6 +52,21 @@ from src.llm_models.utils_model import LLMRequest
 logger = get_logger(__name__)
 
 
+CHINESE_TO_MEMORY_TYPE: dict[str, MemoryType] = {
+    "个人事实": MemoryType.PERSONAL_FACT,
+    "事件": MemoryType.EVENT,
+    "偏好": MemoryType.PREFERENCE,
+    "观点": MemoryType.OPINION,
+    "关系": MemoryType.RELATIONSHIP,
+    "情感": MemoryType.EMOTION,
+    "知识": MemoryType.KNOWLEDGE,
+    "技能": MemoryType.SKILL,
+    "目标": MemoryType.GOAL,
+    "经验": MemoryType.EXPERIENCE,
+    "上下文": MemoryType.CONTEXTUAL,
+}
+
+
 class ExtractionStrategy(Enum):
     """提取策略"""
 
@@ -428,7 +446,7 @@ class MemoryBuilder:
                     subject=normalized_subject,
                     predicate=predicate_value,
                     obj=object_value,
-                    memory_type=MemoryType(mem_data.get("type", "contextual")),
+                    memory_type=self._resolve_memory_type(mem_data.get("type")),
                     chat_id=context.get("chat_id"),
                     source_context=mem_data.get("reasoning", ""),
                     importance=importance_level,
@@ -459,7 +477,33 @@ class MemoryBuilder:
 
         return memories
 
-    def _parse_enum_value(self, enum_cls: type[Enum], raw_value: Any, default: Enum, field_name: str) -> Enum:
+    def _resolve_memory_type(self, type_str: Any) -> MemoryType:
+        """健壮地解析记忆类型，兼容中文和英文"""
+        if not isinstance(type_str, str) or not type_str.strip():
+            return MemoryType.CONTEXTUAL
+
+        cleaned_type = type_str.strip()
+
+        # 尝试中文映射
+        if cleaned_type in CHINESE_TO_MEMORY_TYPE:
+            return CHINESE_TO_MEMORY_TYPE[cleaned_type]
+
+        # 尝试直接作为枚举值解析
+        try:
+            return MemoryType(cleaned_type.lower().replace(" ", "_"))
+        except ValueError:
+            pass
+
+        # 尝试作为枚举名解析
+        try:
+            return MemoryType[cleaned_type.upper()]
+        except KeyError:
+            pass
+
+        logger.warning(f"无法解析未知的记忆类型 '{type_str}'，回退到上下文类型")
+        return MemoryType.CONTEXTUAL
+
+    def _parse_enum_value(self, enum_cls: Type[E], raw_value: Any, default: E, field_name: str) -> E:
         """解析枚举值，兼容数字/字符串表示"""
         if isinstance(raw_value, enum_cls):
             return raw_value

src/chat/memory_system/memory_system.py

@@ -162,7 +162,6 @@ class MemorySystem:
     async def initialize(self):
         """异步初始化记忆系统"""
         try:
-            logger.info("正在初始化记忆系统...")
 
             # 初始化LLM模型
             fallback_task = getattr(self.llm_model, "model_for_task", None) if self.llm_model else None
@@ -268,11 +267,8 @@ class MemorySystem:
                     logger.warning(f"海马体采样器初始化失败: {e}")
                     self.hippocampus_sampler = None
 
-            # 统一存储已经自动加载数据，无需额外加载
-            logger.info("✅ 简化版记忆系统初始化完成")
 
             self.status = MemorySystemStatus.READY
-            logger.info("✅ 记忆系统初始化完成")
 
         except Exception as e:
             self.status = MemorySystemStatus.ERROR
@@ -1425,16 +1421,6 @@ class MemorySystem:
     def _fingerprint_key(user_id: str, fingerprint: str) -> str:
         return f"{user_id!s}:{fingerprint}"
 
-    def get_system_stats(self) -> dict[str, Any]:
-        """获取系统统计信息"""
-        return {
-            "status": self.status.value,
-            "total_memories": self.total_memories,
-            "last_build_time": self.last_build_time,
-            "last_retrieval_time": self.last_retrieval_time,
-            "config": asdict(self.config),
-        }
-
     def _compute_memory_score(self, query_text: str, memory: MemoryChunk, context: dict[str, Any]) -> float:
         """根据查询和上下文为记忆计算匹配分数"""
         tokens_query = self._tokenize_text(query_text)
@@ -1542,7 +1528,7 @@ class MemorySystem:
 
             # 保存统一存储数据
             if self.unified_storage:
-                await self.unified_storage.cleanup()
+                self.unified_storage.cleanup()
 
             logger.info("✅ 简化记忆系统已关闭")
 

src/chat/memory_system/vector_memory_storage_v2.py

@@ -964,6 +964,11 @@ class VectorMemoryStorage:
 
         logger.info("Vector记忆存储系统已停止")
 
+    def cleanup(self):
+        """清理资源，兼容旧接口"""
+        logger.info("正在清理VectorMemoryStorage资源...")
+        self.stop()
+
 
 # 全局实例（可选）
 _global_vector_storage = None

src/chat/message_manager/adaptive_stream_manager.py

@@ -110,7 +110,6 @@ class AdaptiveStreamManager:
         self.is_running = True
         self.monitor_task = asyncio.create_task(self._system_monitor_loop(), name="system_monitor")
         self.adjustment_task = asyncio.create_task(self._adjustment_loop(), name="limit_adjustment")
-        logger.info("自适应流管理器已启动")
 
     async def stop(self):
         """停止自适应管理器"""

src/chat/message_manager/batch_database_writer.py

@@ -72,7 +72,6 @@ class BatchDatabaseWriter:
 
         self.is_running = True
         self.writer_task = asyncio.create_task(self._batch_writer_loop(), name="batch_database_writer")
-        logger.info("批量数据库写入器已启动")
 
     async def stop(self):
         """停止批量写入器"""

src/chat/message_manager/distribution_manager.py

@@ -59,7 +59,6 @@ class StreamLoopManager:
             return
 
         self.is_running = True
-        logger.info("流循环管理器已启动")
 
     async def stop(self) -> None:
         """停止流循环管理器"""

src/chat/message_manager/message_manager.py

@@ -60,7 +60,6 @@ class MessageManager:
         try:
             from src.chat.message_manager.batch_database_writer import init_batch_writer
             await init_batch_writer()
-            logger.info("📦 批量数据库写入器已启动")
         except Exception as e:
             logger.error(f"启动批量数据库写入器失败: {e}")
 
@@ -68,7 +67,6 @@ class MessageManager:
         try:
             from src.chat.message_manager.stream_cache_manager import init_stream_cache_manager
             await init_stream_cache_manager()
-            logger.info("🗄️ 流缓存管理器已启动")
         except Exception as e:
             logger.error(f"启动流缓存管理器失败: {e}")
 

src/chat/message_manager/stream_cache_manager.py

@@ -72,7 +72,6 @@ class TieredStreamCache:
 
         self.is_running = True
         self.cleanup_task = asyncio.create_task(self._cleanup_loop(), name="stream_cache_cleanup")
-        logger.info("分层流缓存管理器已启动")
 
     async def stop(self):
         """停止缓存管理器"""

src/common/logger.py

@@ -3,9 +3,11 @@
 import logging
 import threading
 import time
+import tarfile
 from collections.abc import Callable
 from datetime import datetime, timedelta
 from pathlib import Path
+from typing import Any, Optional, Dict
 
 import orjson
 import structlog
@@ -15,14 +17,69 @@ import tomlkit
 LOG_DIR = Path("logs")
 LOG_DIR.mkdir(exist_ok=True)
 
-# 全局handler实例，避免重复创建
-_file_handler = None
-_console_handler = None
+# 全局handler实例，避免重复创建（可能为None表示禁用文件日志）
+_file_handler: Optional[logging.Handler] = None
+_console_handler: Optional[logging.Handler] = None
+
+# 动态 logger 元数据注册表 (name -> {alias:str|None, color:str|None})
+_LOGGER_META_LOCK = threading.Lock()
+_LOGGER_META: Dict[str, Dict[str, Optional[str]]] = {}
+
+
+def _normalize_color(color: Optional[str]) -> Optional[str]:
+    """接受 ANSI 码 / #RRGGBB / rgb(r,g,b) / 颜色名(直接返回) -> ANSI 码.
+    不做复杂解析，只支持 #RRGGBB 转 24bit ANSI。
+    """
+    if not color:
+        return None
+    color = color.strip()
+    if color.startswith("\033["):
+        return color  # 已经是ANSI
+    if color.startswith("#") and len(color) == 7:
+        try:
+            r = int(color[1:3], 16)
+            g = int(color[3:5], 16)
+            b = int(color[5:7], 16)
+            return f"\033[38;2;{r};{g};{b}m"
+        except ValueError:
+            return None
+    # 简单 rgb(r,g,b)
+    if color.lower().startswith("rgb(") and color.endswith(")"):
+        try:
+            nums = color[color.find("(") + 1 : -1].split(",")
+            r, g, b = (int(x) for x in nums[:3])
+            return f"\033[38;2;{r};{g};{b}m"
+        except Exception:  # noqa: BLE001
+            return None
+    # 其他情况直接返回，假设是短ANSI或名称（控制台渲染器不做翻译，仅输出）
+    return color
+
+
+def _register_logger_meta(name: str, *, alias: Optional[str] = None, color: Optional[str] = None):
+    """注册/更新 logger 元数据。"""
+    if not name:
+        return
+    with _LOGGER_META_LOCK:
+        meta = _LOGGER_META.setdefault(name, {"alias": None, "color": None})
+        if alias is not None:
+            meta["alias"] = alias
+        if color is not None:
+            meta["color"] = _normalize_color(color)
+
+
+def get_logger_meta(name: str) -> Dict[str, Optional[str]]:
+    with _LOGGER_META_LOCK:
+        return _LOGGER_META.get(name, {"alias": None, "color": None}).copy()
 
 
 def get_file_handler():
-    """获取文件handler单例"""
+    """获取文件handler单例; 当 retention=0 时返回 None (禁用文件输出)。"""
     global _file_handler
+
+    retention_days = LOG_CONFIG.get("file_retention_days", 30)
+    if retention_days == 0:
+        return None
+
     if _file_handler is None:
         # 确保日志目录存在
         LOG_DIR.mkdir(exist_ok=True)
@@ -34,14 +91,12 @@ def get_file_handler():
                 _file_handler = handler
                 return _file_handler
 
-        # 使用基于时间戳的handler，简单的轮转份数限制
         _file_handler = TimestampedFileHandler(
             log_dir=LOG_DIR,
             max_bytes=5 * 1024 * 1024,  # 5MB
             backup_count=30,
             encoding="utf-8",
         )
-        # 设置文件handler的日志级别
         file_level = LOG_CONFIG.get("file_log_level", LOG_CONFIG.get("log_level", "INFO"))
         _file_handler.setLevel(getattr(logging, file_level.upper(), logging.INFO))
     return _file_handler
@@ -59,7 +114,16 @@ def get_console_handler():
 
 
 class TimestampedFileHandler(logging.Handler):
-    """基于时间戳的文件处理器，简单的轮转份数限制"""
+    """基于时间戳的文件处理器，带简单大小轮转 + 旧文件压缩/保留策略。
+
+    新策略:
+      - 日志文件命名 app_YYYYmmdd_HHMMSS.log.jsonl
+      - 轮转时会尝试压缩所有不再写入的 .log.jsonl -> .tar.gz
+      - retention:
+          file_retention_days = -1  永不删除
+          file_retention_days = 0   上层禁用文件日志(不会实例化此类)
+          file_retention_days = N>0 删除早于 N 天 (针对 .tar.gz 与遗留未压缩文件)
+    """
 
     def __init__(self, log_dir, max_bytes=5 * 1024 * 1024, backup_count=30, encoding="utf-8"):
         super().__init__()
@@ -77,8 +141,15 @@ class TimestampedFileHandler(logging.Handler):
 
     def _init_current_file(self):
         """初始化当前日志文件"""
-        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
-        self.current_file = self.log_dir / f"app_{timestamp}.log.jsonl"
+        # 使用微秒保证同一秒内多次轮转也获得不同文件名
+        while True:
+            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
+            candidate = self.log_dir / f"app_{timestamp}.log.jsonl"
+            if not candidate.exists():
+                self.current_file = candidate
+                break
+            # 极低概率碰撞，稍作等待
+            time.sleep(0.001)
         self.current_stream = open(self.current_file, "a", encoding=self.encoding)
 
     def _should_rollover(self):
@@ -88,34 +159,56 @@ class TimestampedFileHandler(logging.Handler):
         return False
 
     def _do_rollover(self):
-        """执行轮转：关闭当前文件，创建新文件"""
+        """执行轮转：关闭当前文件 -> 立即创建新文件 -> 压缩旧文件 -> 清理过期。
+
+        这样可以避免旧文件因为 self.current_file 仍指向它而被 _compress_stale_logs 跳过。
+        """
         if self.current_stream:
             self.current_stream.close()
+        # 记录旧文件引用，方便调试（暂不使用变量）
+        self._init_current_file()  # 先创建新文件，确保后续压缩不会跳过刚关闭的旧文件
+        try:
+            self._compress_stale_logs()
+            self._cleanup_old_files()
+        except Exception as e:  # noqa: BLE001
+            print(f"[日志轮转] 轮转过程出错: {e}")
 
-        # 清理旧文件
-        self._cleanup_old_files()
-
-        # 创建新文件
-        self._init_current_file()
+    def _compress_stale_logs(self):  # sourcery skip: extract-method
+        """将不再写入且未压缩的 .log.jsonl 文件压缩成 .tar.gz。"""
+        try:
+            for f in self.log_dir.glob("app_*.log.jsonl"):
+                if f == self.current_file:
+                    continue
+                tar_path = f.with_suffix(f.suffix + ".tar.gz")  # .log.jsonl.tar.gz
+                if tar_path.exists():
+                    continue
+                # 压缩
+                try:
+                    with tarfile.open(tar_path, "w:gz") as tf:  # noqa: SIM117
+                        tf.add(f, arcname=f.name)
+                    f.unlink(missing_ok=True)
+                except Exception as e:  # noqa: BLE001
+                    print(f"[日志压缩] 压缩 {f.name} 失败: {e}")
+        except Exception as e:  # noqa: BLE001
+            print(f"[日志压缩] 过程出错: {e}")
 
     def _cleanup_old_files(self):
-        """清理旧的日志文件，保留指定数量"""
+        """按 retention 天数删除压缩包/遗留文件。"""
+        retention_days = LOG_CONFIG.get("file_retention_days", 30)
+        if retention_days in (-1, 0):
+            return  # -1 永不删除；0 在外层已禁用
+        cutoff = datetime.now() - timedelta(days=retention_days)
         try:
-            # 获取所有日志文件
-            log_files = list(self.log_dir.glob("app_*.log.jsonl"))
-
-            # 按修改时间排序
-            log_files.sort(key=lambda f: f.stat().st_mtime, reverse=True)
-
-            # 删除超出数量限制的文件
-            for old_file in log_files[self.backup_count :]:
+            for f in self.log_dir.glob("app_*.log.jsonl*"):
+                if f == self.current_file:
+                    continue
                 try:
-                    old_file.unlink()
-                    print(f"[日志清理] 删除旧文件: {old_file.name}")
-                except Exception as e:
-                    print(f"[日志清理] 删除失败 {old_file}: {e}")
-
-        except Exception as e:
+                    mtime = datetime.fromtimestamp(f.stat().st_mtime)
+                    if mtime < cutoff:
+                        f.unlink(missing_ok=True)
+                except Exception as e:  # noqa: BLE001
+                    print(f"[日志清理] 删除 {f} 失败: {e}")
+        except Exception as e:  # noqa: BLE001
             print(f"[日志清理] 清理过程出错: {e}")
 
     def emit(self, record):
@@ -194,6 +287,7 @@ def load_log_config():  # sourcery skip: use-contextlib-suppress
         "log_level": "INFO",  # 全局日志级别（向下兼容）
         "console_log_level": "INFO",  # 控制台日志级别
         "file_log_level": "DEBUG",  # 文件日志级别
+        "file_retention_days": 30,  # 文件日志保留天数，0=禁用文件日志，-1=永不删除
         "suppress_libraries": [
             "faiss",
             "httpx",
@@ -210,6 +304,9 @@ def load_log_config():  # sourcery skip: use-contextlib-suppress
         "library_log_levels": {"aiohttp": "WARNING"},
     }
 
+
+        # 误加的即刻线程启动已移除；真正的线程在 start_log_cleanup_task 中按午夜调度
+
     try:
         if config_path.exists():
             with open(config_path, encoding="utf-8") as f:
@@ -329,8 +426,11 @@ def reconfigure_existing_loggers():
                     logger_obj.addHandler(handler)
 
 
-# 定义模块颜色映射
-MODULE_COLORS = {
+###########################
+# 默认颜色 / 别名 (仍然保留但可被动态覆盖)
+###########################
+
+DEFAULT_MODULE_COLORS = {
     # 核心模块
     "main": "\033[1;97m",  # 亮白色+粗体 (主程序)
     "api": "\033[92m",  # 亮绿色
@@ -528,8 +628,7 @@ MODULE_COLORS = {
     "event_manager": "\033[38;5;79m",  # 柔和的蓝绿色，稍微醒目但不刺眼
 }
 
-# 定义模块别名映射 - 将真实的logger名称映射到显示的别名
-MODULE_ALIASES = {
+DEFAULT_MODULE_ALIASES = {
     # 核心模块
     "individuality": "人格特质",
     "emoji": "表情包",
@@ -743,12 +842,17 @@ class ModuleColoredConsoleRenderer:
         # 获取模块颜色，用于full模式下的整体着色
         module_color = ""
         if self._colors and self._enable_module_colors and logger_name:
-            module_color = MODULE_COLORS.get(logger_name, "")
+            # 动态优先，其次默认表
+            meta = get_logger_meta(logger_name)
+            module_color = meta.get("color") or DEFAULT_MODULE_COLORS.get(logger_name, "")
 
         # 模块名称（带颜色和别名支持）
         if logger_name:
             # 获取别名，如果没有别名则使用原名称
-            display_name = MODULE_ALIASES.get(logger_name, logger_name)
+            # 若上面条件不成立需要再次获取 meta
+            if 'meta' not in locals():
+                meta = get_logger_meta(logger_name)
+            display_name = meta.get("alias") or DEFAULT_MODULE_ALIASES.get(logger_name, logger_name)
 
             if self._colors and self._enable_module_colors:
                 if module_color:
@@ -809,7 +913,7 @@ class ModuleColoredConsoleRenderer:
         # 处理其他字段
         extras = []
         for key, value in event_dict.items():
-            if key not in ("timestamp", "level", "logger_name", "event"):
+            if key not in ("timestamp", "level", "logger_name", "event") and key not in ("color", "alias"):
                 # 确保值也转换为字符串
                 if isinstance(value, (dict, list)):
                     try:
@@ -837,15 +941,34 @@ class ModuleColoredConsoleRenderer:
 file_handler = get_file_handler()
 console_handler = get_console_handler()
 
+handlers = [h for h in (file_handler, console_handler) if h is not None]
 logging.basicConfig(
     level=logging.INFO,
     format="%(message)s",
-    handlers=[file_handler, console_handler],
+    handlers=handlers,
 )
 
 
+def add_logger_metadata(logger: Any, method_name: str, event_dict: dict):  # type: ignore[override]
+    """structlog 自定义处理器: 注入 color / alias 字段 (用于 JSON 输出)。"""
+    name = event_dict.get("logger_name")
+    if name:
+        meta = get_logger_meta(name)
+        # 默认 fallback
+        if meta.get("color") is None and name in DEFAULT_MODULE_COLORS:
+            meta["color"] = DEFAULT_MODULE_COLORS[name]
+        if meta.get("alias") is None and name in DEFAULT_MODULE_ALIASES:
+            meta["alias"] = DEFAULT_MODULE_ALIASES[name]
+        # 注入
+        if meta.get("color"):
+            event_dict["color"] = meta["color"]
+        if meta.get("alias"):
+            event_dict["alias"] = meta["alias"]
+    return event_dict
+
+
 def configure_structlog():
-    """配置structlog"""
+    """配置structlog，加入自定义 metadata 处理器。"""
     structlog.configure(
         processors=[
             structlog.contextvars.merge_contextvars,
@@ -853,7 +976,7 @@ def configure_structlog():
             structlog.processors.StackInfoRenderer(),
             structlog.dev.set_exc_info,
             structlog.processors.TimeStamper(fmt=get_timestamp_format(), utc=False),
-            # 根据输出类型选择不同的渲染器
+            add_logger_metadata,  # 注入 color/alias
             structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
         ],
         wrapper_class=structlog.stdlib.BoundLogger,
@@ -913,16 +1036,17 @@ def _immediate_setup():
         root_logger.removeHandler(handler)
 
     # 使用单例handler避免重复创建
-    file_handler = get_file_handler()
-    console_handler = get_console_handler()
+    file_handler_local = get_file_handler()
+    console_handler_local = get_console_handler()
 
-    # 重新添加配置好的handler
-    root_logger.addHandler(file_handler)
-    root_logger.addHandler(console_handler)
+    for h in (file_handler_local, console_handler_local):
+        if h is not None:
+            root_logger.addHandler(h)
 
     # 设置格式化器
-    file_handler.setFormatter(file_formatter)
-    console_handler.setFormatter(console_formatter)
+    if file_handler_local is not None:
+        file_handler_local.setFormatter(file_formatter)
+    console_handler_local.setFormatter(console_formatter)
 
     # 清理重复的handler
     remove_duplicate_handlers()
@@ -942,15 +1066,23 @@ raw_logger: structlog.stdlib.BoundLogger = structlog.get_logger()
 binds: dict[str, Callable] = {}
 
 
-def get_logger(name: str | None) -> structlog.stdlib.BoundLogger:
-    """获取logger实例，支持按名称绑定"""
+def get_logger(name: str | None, *, color: Optional[str] = None, alias: Optional[str] = None) -> structlog.stdlib.BoundLogger:
+    """获取/创建 structlog logger。
+
+    新增:
+      - color: 传入 ANSI / #RRGGBB / rgb(r,g,b) 以注册显示颜色
+      - alias: 别名, 控制台模块显示 & JSON 中 alias 字段
+    多次调用可更新元数据 (后调用覆盖之前的 color/alias, 仅覆盖给定的)
+    """
     if name is None:
         return raw_logger
+    if color is not None or alias is not None:
+        _register_logger_meta(name, alias=alias, color=color)
     logger = binds.get(name)  # type: ignore
     if logger is None:
-        logger: structlog.stdlib.BoundLogger = structlog.get_logger(name).bind(logger_name=name)
+        logger = structlog.get_logger(name).bind(logger_name=name)  # type: ignore[assignment]
         binds[name] = logger
-    return logger
+    return logger  # type: ignore[return-value]
 
 
 def initialize_logging():
@@ -975,52 +1107,93 @@ def initialize_logging():
     logger.info("日志系统已初始化:")
     logger.info(f"  - 控制台级别: {console_level}")
     logger.info(f"  - 文件级别: {file_level}")
-    logger.info("  - 轮转份数: 30个文件|自动清理: 30天前的日志")
+    retention_days = LOG_CONFIG.get("file_retention_days", 30)
+    if retention_days == 0:
+        retention_desc = "文件日志已禁用"
+    elif retention_days == -1:
+        retention_desc = "永不删除 (仅压缩旧文件)"
+    else:
+        retention_desc = f"保留 {retention_days} 天"
+    logger.info(f"  - 文件保留策略: {retention_desc}")
 
 
 def cleanup_old_logs():
-    """清理过期的日志文件"""
+    """压缩遗留未压缩的日志并按 retention 策略删除。"""
+    retention_days = LOG_CONFIG.get("file_retention_days", 30)
+    if retention_days == 0:
+        return  # 已禁用
     try:
-        cleanup_days = 30  # 硬编码30天
-        cutoff_date = datetime.now() - timedelta(days=cleanup_days)
+        # 先压缩(复用 handler 的逻辑, 但 handler 可能未创建——手动调用)
+        try:
+            for f in LOG_DIR.glob("app_*.log.jsonl"):
+                # 当前写入文件无法可靠识别(仅 handler 知道); 粗略策略: 如果修改时间>5分钟也压缩
+                if time.time() - f.stat().st_mtime < 300:
+                    continue
+                tar_path = f.with_suffix(f.suffix + ".tar.gz")
+                if tar_path.exists():
+                    continue
+                with tarfile.open(tar_path, "w:gz") as tf:  # noqa: SIM117
+                    tf.add(f, arcname=f.name)
+                f.unlink(missing_ok=True)
+        except Exception as e:  # noqa: BLE001
+            logger = get_logger("logger")
+            logger.warning(f"周期压缩日志时出错: {e}")
+
+        if retention_days == -1:
+            return  # 永不删除
+        cutoff_date = datetime.now() - timedelta(days=retention_days)
         deleted_count = 0
         deleted_size = 0
-
-        # 遍历日志目录
-        for log_file in LOG_DIR.glob("*.log*"):
+        for log_file in LOG_DIR.glob("app_*.log.jsonl*"):
             try:
                 file_time = datetime.fromtimestamp(log_file.stat().st_mtime)
                 if file_time < cutoff_date:
-                    file_size = log_file.stat().st_size
-                    log_file.unlink()
+                    size = log_file.stat().st_size
+                    log_file.unlink(missing_ok=True)
                     deleted_count += 1
-                    deleted_size += file_size
-            except Exception as e:
+                    deleted_size += size
+            except Exception as e:  # noqa: BLE001
                 logger = get_logger("logger")
                 logger.warning(f"清理日志文件 {log_file} 时出错: {e}")
-
-        if deleted_count > 0:
+        if deleted_count:
             logger = get_logger("logger")
-            logger.info(f"清理了 {deleted_count} 个过期日志文件，释放空间 {deleted_size / 1024 / 1024:.2f} MB")
-
-    except Exception as e:
+            logger.info(
+                f"清理 {deleted_count} 个过期日志 (≈{deleted_size / 1024 / 1024:.2f}MB), 保留策略={retention_days}天"
+            )
+    except Exception as e:  # noqa: BLE001
         logger = get_logger("logger")
         logger.error(f"清理旧日志文件时出错: {e}")
 
 
 def start_log_cleanup_task():
-    """启动日志清理任务"""
+    """启动日志压缩/清理任务：每天本地时间 00:00 运行一次。"""
+    retention_days = LOG_CONFIG.get("file_retention_days", 30)
+    if retention_days == 0:
+        return  # 文件日志禁用无需周期任务
+
+    def seconds_until_next_midnight() -> float:
+        now = datetime.now()
+        tomorrow = now + timedelta(days=1)
+        midnight = datetime(year=tomorrow.year, month=tomorrow.month, day=tomorrow.day)
+        return (midnight - now).total_seconds()
 
     def cleanup_task():
+        # 首次等待到下一个本地午夜
+        time.sleep(max(1, seconds_until_next_midnight()))
         while True:
-            time.sleep(24 * 60 * 60)  # 每24小时执行一次
-            cleanup_old_logs()
-
-    cleanup_thread = threading.Thread(target=cleanup_task, daemon=True)
-    cleanup_thread.start()
+            try:
+                cleanup_old_logs()
+            except Exception as e:  # noqa: BLE001
+                print(f"[日志任务] 执行清理出错: {e}")
+            # 再次等待到下一个午夜
+            time.sleep(max(1, seconds_until_next_midnight()))
 
+    threading.Thread(target=cleanup_task, daemon=True, name="log-cleanup").start()
     logger = get_logger("logger")
-    logger.info("已启动日志清理任务，将自动清理30天前的日志文件（轮转份数限制: 30个文件）")
+    if retention_days == -1:
+        logger.info("已启动日志任务: 每天 00:00 压缩旧日志(不删除)")
+    else:
+        logger.info(f"已启动日志任务: 每天 00:00 压缩并删除早于 {retention_days} 天的日志")
 
 
 def shutdown_logging():

src/plugin_system/__init__.py

@@ -50,13 +50,6 @@ from .base import (
     create_plus_command_adapter,
 )
 
-# 导入工具模块
-from .utils import (
-    ManifestValidator,
-    # ManifestGenerator,
-    # validate_plugin_manifest,
-    # generate_plugin_manifest,
-)
 from .utils.dependency_config import configure_dependency_settings, get_dependency_config
 
 # 导入依赖管理模块

src/plugin_system/base/plugin_base.py

@@ -5,7 +5,6 @@ from abc import ABC, abstractmethod
 from pathlib import Path
 from typing import Any
 
-import orjson
 import toml
 
 from src.common.logger import get_logger
@@ -15,7 +14,7 @@ from src.plugin_system.base.component_types import (
     PythonDependency,
 )
 from src.plugin_system.base.config_types import ConfigField
-from src.plugin_system.utils.manifest_utils import ManifestValidator
+from src.plugin_system.base.plugin_metadata import PluginMetadata
 
 logger = get_logger("plugin_base")
 
@@ -33,39 +32,34 @@ class PluginBase(ABC):
     dependencies: list[str] = []
     python_dependencies: list[str | PythonDependency] = []
 
-    # manifest文件相关
-    manifest_file_name: str = "_manifest.json"  # manifest文件名
-    manifest_data: dict[str, Any] = {}  # manifest数据
-
     config_schema: dict[str, dict[str, ConfigField] | str] = {}
 
     config_section_descriptions: dict[str, str] = {}
 
-    def __init__(self, plugin_dir: str):
+    def __init__(self, plugin_dir: str, metadata: PluginMetadata):
         """初始化插件
 
         Args:
             plugin_dir: 插件目录路径，由插件管理器传递
+            metadata: 插件元数据对象
         """
         self.config: dict[str, Any] = {}  # 插件配置
         self.plugin_dir = plugin_dir  # 插件目录路径
+        self.plugin_meta = metadata  # 插件元数据
         self.log_prefix = f"[Plugin:{self.plugin_name}]"
         self._is_enabled = self.enable_plugin  # 从插件定义中获取默认启用状态
 
-        # 加载manifest文件
-        self._load_manifest()
-
         # 验证插件信息
         self._validate_plugin_info()
 
         # 加载插件配置
         self._load_plugin_config()
 
-        # 从manifest获取显示信息
-        self.display_name = self.get_manifest_info("name", self.plugin_name)
-        self.plugin_version = self.get_manifest_info("version", "1.0.0")
-        self.plugin_description = self.get_manifest_info("description", "")
-        self.plugin_author = self._get_author_name()
+        # 从元数据获取显示信息
+        self.display_name = self.plugin_meta.name
+        self.plugin_version = self.plugin_meta.version
+        self.plugin_description = self.plugin_meta.description
+        self.plugin_author = self.plugin_meta.author
 
         # 标准化Python依赖为PythonDependency对象
         normalized_python_deps = self._normalize_python_dependencies(self.python_dependencies)
@@ -85,15 +79,6 @@ class PluginBase(ABC):
             config_file=self.config_file_name or "",
             dependencies=self.dependencies.copy(),
             python_dependencies=normalized_python_deps,
-            # manifest相关信息
-            manifest_data=self.manifest_data.copy(),
-            license=self.get_manifest_info("license", ""),
-            homepage_url=self.get_manifest_info("homepage_url", ""),
-            repository_url=self.get_manifest_info("repository_url", ""),
-            keywords=self.get_manifest_info("keywords", []).copy() if self.get_manifest_info("keywords") else [],
-            categories=self.get_manifest_info("categories", []).copy() if self.get_manifest_info("categories") else [],
-            min_host_version=self.get_manifest_info("host_application.min_version", ""),
-            max_host_version=self.get_manifest_info("host_application.max_version", ""),
         )
 
         logger.debug(f"{self.log_prefix} 插件基类初始化完成")
@@ -103,93 +88,10 @@ class PluginBase(ABC):
         if not self.plugin_name:
             raise ValueError(f"插件类 {self.__class__.__name__} 必须定义 plugin_name")
 
-        # 验证manifest中的必需信息
-        if not self.get_manifest_info("name"):
-            raise ValueError(f"插件 {self.plugin_name} 的manifest中缺少name字段")
-        if not self.get_manifest_info("description"):
-            raise ValueError(f"插件 {self.plugin_name} 的manifest中缺少description字段")
-
-    def _load_manifest(self):  # sourcery skip: raise-from-previous-error
-        """加载manifest文件（强制要求）"""
-        if not self.plugin_dir:
-            raise ValueError(f"{self.log_prefix} 没有插件目录路径，无法加载manifest")
-
-        manifest_path = os.path.join(self.plugin_dir, self.manifest_file_name)
-
-        if not os.path.exists(manifest_path):
-            error_msg = f"{self.log_prefix} 缺少必需的manifest文件: {manifest_path}"
-            logger.error(error_msg)
-            raise FileNotFoundError(error_msg)
-
-        try:
-            with open(manifest_path, encoding="utf-8") as f:
-                self.manifest_data = orjson.loads(f.read())
-
-            logger.debug(f"{self.log_prefix} 成功加载manifest文件: {manifest_path}")
-
-            # 验证manifest格式
-            self._validate_manifest()
-
-        except orjson.JSONDecodeError as e:
-            error_msg = f"{self.log_prefix} manifest文件格式错误: {e}"
-            logger.error(error_msg)
-            raise ValueError(error_msg)
-        except OSError as e:
-            error_msg = f"{self.log_prefix} 读取manifest文件失败: {e}"
-            logger.error(error_msg)
-            raise IOError(error_msg)  # noqa
-
-    def _get_author_name(self) -> str:
-        """从manifest获取作者名称"""
-        author_info = self.get_manifest_info("author", {})
-        if isinstance(author_info, dict):
-            return author_info.get("name", "")
-        else:
-            return str(author_info) if author_info else ""
-
-    def _validate_manifest(self):
-        """验证manifest文件格式（使用强化的验证器）"""
-        if not self.manifest_data:
-            raise ValueError(f"{self.log_prefix} manifest数据为空，验证失败")
-
-        validator = ManifestValidator()
-        is_valid = validator.validate_manifest(self.manifest_data)
-
-        # 记录验证结果
-        if validator.validation_errors or validator.validation_warnings:
-            report = validator.get_validation_report()
-            logger.info(f"{self.log_prefix} Manifest验证结果:\n{report}")
-
-        # 如果有验证错误，抛出异常
-        if not is_valid:
-            error_msg = f"{self.log_prefix} Manifest文件验证失败"
-            if validator.validation_errors:
-                error_msg += f": {'; '.join(validator.validation_errors)}"
-            raise ValueError(error_msg)
-
-    def get_manifest_info(self, key: str, default: Any = None) -> Any:
-        """获取manifest信息
-
-        Args:
-            key: 信息键，支持点分割的嵌套键（如 "author.name"）
-            default: 默认值
-
-        Returns:
-            Any: 对应的值
-        """
-        if not self.manifest_data:
-            return default
-
-        keys = key.split(".")
-        value = self.manifest_data
-
-        for k in keys:
-            if isinstance(value, dict) and k in value:
-                value = value[k]
-            else:
-                return default
-
-        return value
+        if not self.plugin_meta.name:
+            raise ValueError(f"插件 {self.plugin_name} 的元数据中缺少 name 字段")
+        if not self.plugin_meta.description:
+            raise ValueError(f"插件 {self.plugin_name} 的元数据中缺少 description 字段")
 
     def _generate_and_save_default_config(self, config_file_path: str):
         """根据插件的Schema生成并保存默认配置文件"""
@@ -198,7 +100,7 @@ class PluginBase(ABC):
             return
 
         toml_str = f"# {self.plugin_name} - 自动生成的配置文件\n"
-        plugin_description = self.get_manifest_info("description", "插件配置文件")
+        plugin_description = self.plugin_meta.description or "插件配置文件"
         toml_str += f"# {plugin_description}\n\n"
 
         # 遍历每个配置节
@@ -333,7 +235,7 @@ class PluginBase(ABC):
             return
 
         toml_str = f"# {self.plugin_name} - 配置文件\n"
-        plugin_description = self.get_manifest_info("description", "插件配置文件")
+        plugin_description = self.plugin_meta.description or "插件配置文件"
         toml_str += f"# {plugin_description}\n\n"
 
         # 遍历每个配置节
@@ -564,3 +466,11 @@ class PluginBase(ABC):
             bool: 是否成功注册插件
         """
         raise NotImplementedError("Subclasses must implement this method")
+
+    async def on_plugin_loaded(self):
+        """插件加载完成后的钩子函数"""
+        pass
+
+    def on_unload(self):
+        """插件卸载时的钩子函数"""
+        pass

src/plugin_system/base/plugin_metadata.py

@@ -0,0 +1,25 @@
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional, Set
+
+@dataclass
+class PluginMetadata:
+    """
+    插件元数据，用于存储插件的开发者信息和用户帮助信息。
+    """
+    name: str  # 插件名称 (供用户查看)
+    description: str  # 插件功能描述
+    usage: str  # 插件使用方法
+
+    # 以下为可选字段，参考自 _manifest.json 和 NoneBot 设计
+    type: Optional[str] = None  # 插件类别: "library", "application"
+
+    # 从原 _manifest.json 迁移的字段
+    version: str = "1.0.0"  # 插件版本
+    author: str = ""  # 作者名称
+    license: Optional[str] = None  # 开源协议
+    repository_url: Optional[str] = None  # 仓库地址
+    keywords: List[str] = field(default_factory=list)  # 关键词
+    categories: List[str] = field(default_factory=list)  # 分类
+
+    # 扩展字段
+    extra: Dict[str, Any] = field(default_factory=dict)  # 其他任意信息

src/plugin_system/core/plugin_manager.py

@@ -9,7 +9,7 @@ from typing import Any, Optional
 from src.common.logger import get_logger
 from src.plugin_system.base.component_types import ComponentType
 from src.plugin_system.base.plugin_base import PluginBase
-from src.plugin_system.utils.manifest_utils import VersionComparator
+from src.plugin_system.base.plugin_metadata import PluginMetadata
 
 from .component_registry import component_registry
 
@@ -27,6 +27,7 @@ class PluginManager:
         self.plugin_directories: list[str] = []  # 插件根目录列表
         self.plugin_classes: dict[str, type[PluginBase]] = {}  # 全局插件类注册表，插件名 -> 插件类
         self.plugin_paths: dict[str, str] = {}  # 记录插件名到目录路径的映射，插件名 -> 目录路径
+        self.plugin_modules: dict[str, Any] = {}  # 记录插件名到模块的映射
 
         self.loaded_plugins: dict[str, PluginBase] = {}  # 已加载的插件类实例注册表，插件名 -> 插件类实例
         self.failed_plugins: dict[str, str] = {}  # 记录加载失败的插件文件及其错误信息，插件名 -> 错误信息
@@ -102,23 +103,25 @@ class PluginManager:
             if not plugin_dir:
                 return False, 1
 
-            plugin_instance = plugin_class(plugin_dir=plugin_dir)  # 实例化插件（可能因为缺少manifest而失败）
+            module = self.plugin_modules.get(plugin_name)
+            
+            if not module or not hasattr(module, "__plugin_meta__"):
+                self.failed_plugins[plugin_name] = "插件模块中缺少 __plugin_meta__"
+                logger.error(f"❌ 插件加载失败: {plugin_name} - 缺少 __plugin_meta__")
+                return False, 1
+
+            metadata: PluginMetadata = getattr(module, "__plugin_meta__")
+
+            plugin_instance = plugin_class(plugin_dir=plugin_dir, metadata=metadata)
             if not plugin_instance:
                 logger.error(f"插件 {plugin_name} 实例化失败")
                 return False, 1
+
             # 检查插件是否启用
             if not plugin_instance.enable_plugin:
                 logger.info(f"插件 {plugin_name} 已禁用，跳过加载")
                 return False, 0
 
-            # 检查版本兼容性
-            is_compatible, compatibility_error = self._check_plugin_version_compatibility(
-                plugin_name, plugin_instance.manifest_data
-            )
-            if not is_compatible:
-                self.failed_plugins[plugin_name] = compatibility_error
-                logger.error(f"❌ 插件加载失败: {plugin_name} - {compatibility_error}")
-                return False, 1
             if plugin_instance.register_plugin():
                 self.loaded_plugins[plugin_name] = plugin_instance
                 self._show_plugin_components(plugin_name)
@@ -138,21 +141,6 @@ class PluginManager:
                 logger.error(f"❌ 插件注册失败: {plugin_name}")
                 return False, 1
 
-        except FileNotFoundError as e:
-            # manifest文件缺失
-            error_msg = f"缺少manifest文件: {e!s}"
-            self.failed_plugins[plugin_name] = error_msg
-            logger.error(f"❌ 插件加载失败: {plugin_name} - {error_msg}")
-            return False, 1
-
-        except ValueError as e:
-            # manifest文件格式错误或验证失败
-            traceback.print_exc()
-            error_msg = f"manifest验证失败: {e!s}"
-            self.failed_plugins[plugin_name] = error_msg
-            logger.error(f"❌ 插件加载失败: {plugin_name} - {error_msg}")
-            return False, 1
-
         except Exception as e:
             # 其他错误
             error_msg = f"未知错误: {e!s}"
@@ -284,14 +272,23 @@ class PluginManager:
             if os.path.isdir(item_path) and not item.startswith(".") and not item.startswith("__"):
                 plugin_file = os.path.join(item_path, "plugin.py")
                 if os.path.exists(plugin_file):
-                    if self._load_plugin_module_file(plugin_file):
+                    module = self._load_plugin_module_file(plugin_file)
+                    if module:
+                        # 动态查找插件类并获取真实的 plugin_name
+                        for attr_name in dir(module):
+                            attr = getattr(module, attr_name)
+                            if isinstance(attr, type) and issubclass(attr, PluginBase) and attr is not PluginBase:
+                                plugin_name = getattr(attr, "plugin_name", None)
+                                if plugin_name:
+                                    self.plugin_modules[plugin_name] = module
+                                    break
                         loaded_count += 1
                     else:
                         failed_count += 1
 
         return loaded_count, failed_count
 
-    def _load_plugin_module_file(self, plugin_file: str) -> bool:
+    def _load_plugin_module_file(self, plugin_file: str) -> Optional[Any]:
         # sourcery skip: extract-method
         """加载单个插件模块文件
 
@@ -305,62 +302,36 @@ class PluginManager:
         module_name = ".".join(plugin_path.parent.parts)
 
         try:
-            # 动态导入插件模块
+            # 首先加载 __init__.py 来获取元数据
+            init_file = os.path.join(plugin_dir, "__init__.py")
+            if os.path.exists(init_file):
+                init_spec = spec_from_file_location(f"{module_name}.__init__", init_file)
+                if init_spec and init_spec.loader:
+                    init_module = module_from_spec(init_spec)
+                    init_spec.loader.exec_module(init_module)
+
+            # 然后加载 plugin.py
             spec = spec_from_file_location(module_name, plugin_file)
             if spec is None or spec.loader is None:
                 logger.error(f"无法创建模块规范: {plugin_file}")
-                return False
+                return None
 
             module = module_from_spec(spec)
-            module.__package__ = module_name  # 设置模块包名
+            module.__package__ = module_name
             spec.loader.exec_module(module)
 
+            # 将 __plugin_meta__ 从 init_module 附加到主模块
+            if init_module and hasattr(init_module, "__plugin_meta__"):
+                setattr(module, "__plugin_meta__", getattr(init_module, "__plugin_meta__"))
+
             logger.debug(f"插件模块加载成功: {plugin_file} -> {plugin_name} ({plugin_dir})")
-            return True
+            return module
 
         except Exception as e:
             error_msg = f"加载插件模块 {plugin_file} 失败: {e}"
             logger.error(error_msg)
             self.failed_plugins[plugin_name if "plugin_name" in locals() else module_name] = error_msg
-            return False
-
-    # == 兼容性检查 ==
-
-    @staticmethod
-    def _check_plugin_version_compatibility(plugin_name: str, manifest_data: dict[str, Any]) -> tuple[bool, str]:
-        """检查插件版本兼容性
-
-        Args:
-            plugin_name: 插件名称
-            manifest_data: manifest数据
-
-        Returns:
-            Tuple[bool, str]: (是否兼容, 错误信息)
-        """
-        if "host_application" not in manifest_data:
-            return True, ""  # 没有版本要求，默认兼容
-
-        host_app = manifest_data["host_application"]
-        if not isinstance(host_app, dict):
-            return True, ""
-
-        min_version = host_app.get("min_version", "")
-        max_version = host_app.get("max_version", "")
-
-        if not min_version and not max_version:
-            return True, ""  # 没有版本要求，默认兼容
-
-        try:
-            current_version = VersionComparator.get_current_host_version()
-            is_compatible, error_msg = VersionComparator.is_version_in_range(current_version, min_version, max_version)
-            if not is_compatible:
-                return False, f"版本不兼容: {error_msg}"
-            logger.debug(f"插件 {plugin_name} 版本兼容性检查通过")
-            return True, ""
-
-        except Exception as e:
-            logger.warning(f"插件 {plugin_name} 版本兼容性检查失败: {e}")
-            return False, f"插件 {plugin_name} 版本兼容性检查失败: {e}"  # 检查失败时默认不允许加载
+            return None
 
     # == 显示统计与插件信息 ==
 
@@ -396,17 +367,6 @@ class PluginManager:
 
                     logger.info(f"  📦 {plugin_info.display_name}{extra_info}")
 
-                    # Manifest信息
-                    if plugin_info.manifest_data:
-                        """
-                        if plugin_info.keywords:
-                            logger.info(f"    🏷️ 关键词: {', '.join(plugin_info.keywords)}")
-                        if plugin_info.categories:
-                            logger.info(f"    📁 分类: {', '.join(plugin_info.categories)}")
-                        """
-                        if plugin_info.homepage_url:
-                            logger.info(f"    🌐 主页: {plugin_info.homepage_url}")
-
                     # 组件列表
                     if plugin_info.components:
                         action_components = [

src/plugin_system/utils/__init__.py

@@ -2,18 +2,4 @@
 插件系统工具模块
 
 提供插件开发和管理的实用工具
-"""
-
-from .manifest_utils import (
-    ManifestValidator,
-    # ManifestGenerator,
-    # validate_plugin_manifest,
-    # generate_plugin_manifest,
-)
-
-__all__ = [
-    "ManifestValidator",
-    # "ManifestGenerator",
-    # "validate_plugin_manifest",
-    # "generate_plugin_manifest",
-]
+"""

src/plugin_system/utils/manifest_utils.py

@@ -1,517 +0,0 @@
-"""
-插件Manifest工具模块
-
-提供manifest文件的验证、生成和管理功能
-"""
-
-import re
-from typing import Any
-
-from src.common.logger import get_logger
-from src.config.config import MMC_VERSION
-
-# if TYPE_CHECKING:
-#     from src.plugin_system.base.base_plugin import BasePlugin
-
-logger = get_logger("manifest_utils")
-
-
-class VersionComparator:
-    """版本号比较器
-
-    支持语义化版本号比较，自动处理snapshot版本，并支持向前兼容性检查
-    """
-
-    # 版本兼容性映射表（硬编码）
-    # 格式: {插件最大支持版本: [实际兼容的版本列表]}
-    COMPATIBILITY_MAP = {
-        # 0.8.x 系列向前兼容规则
-        "0.8.0": ["0.8.1", "0.8.2", "0.8.3", "0.8.4", "0.8.5", "0.8.6", "0.8.7", "0.8.8", "0.8.9", "0.8.10"],
-        "0.8.1": ["0.8.2", "0.8.3", "0.8.4", "0.8.5", "0.8.6", "0.8.7", "0.8.8", "0.8.9", "0.8.10"],
-        "0.8.2": ["0.8.3", "0.8.4", "0.8.5", "0.8.6", "0.8.7", "0.8.8", "0.8.9", "0.8.10"],
-        "0.8.3": ["0.8.4", "0.8.5", "0.8.6", "0.8.7", "0.8.8", "0.8.9", "0.8.10"],
-        "0.8.4": ["0.8.5", "0.8.6", "0.8.7", "0.8.8", "0.8.9", "0.8.10"],
-        "0.8.5": ["0.8.6", "0.8.7", "0.8.8", "0.8.9", "0.8.10"],
-        "0.8.6": ["0.8.7", "0.8.8", "0.8.9", "0.8.10"],
-        "0.8.7": ["0.8.8", "0.8.9", "0.8.10"],
-        "0.8.8": ["0.8.9", "0.8.10"],
-        "0.8.9": ["0.8.10"],
-        # 可以根据需要添加更多兼容映射
-        # "0.9.0": ["0.9.1", "0.9.2", "0.9.3"],  # 示例：0.9.x系列兼容
-    }
-
-    @staticmethod
-    def normalize_version(version: str) -> str:
-        """标准化版本号，移除snapshot标识
-
-        Args:
-            version: 原始版本号，如 "0.8.0-snapshot.1"
-
-        Returns:
-            str: 标准化后的版本号，如 "0.8.0"
-        """
-        if not version:
-            return "0.0.0"
-
-        # 移除snapshot部分
-        normalized = re.sub(r"-snapshot\.\d+", "", version.strip())
-        normalized = re.sub(r"-alpha\-\d+", "", version.strip())
-
-        # 确保版本号格式正确
-        if not re.match(r"^\d+(\.\d+){0,2}$", normalized):
-            # 如果不是有效的版本号格式，返回默认版本
-            return "0.0.0"
-
-        # 尝试补全版本号
-        parts = normalized.split(".")
-        while len(parts) < 3:
-            parts.append("0")
-        normalized = ".".join(parts[:3])
-
-        return normalized
-
-    @staticmethod
-    def parse_version(version: str) -> tuple[int, int, int]:
-        """解析版本号为元组
-
-        Args:
-            version: 版本号字符串
-
-        Returns:
-            Tuple[int, int, int]: (major, minor, patch)
-        """
-        normalized = VersionComparator.normalize_version(version)
-        try:
-            parts = normalized.split(".")
-            return int(parts[0]), int(parts[1]), int(parts[2])
-        except (ValueError, IndexError):
-            logger.warning(f"无法解析版本号: {version}，使用默认版本 0.0.0")
-            return 0, 0, 0
-
-    @staticmethod
-    def compare_versions(version1: str, version2: str) -> int:
-        """比较两个版本号
-
-        Args:
-            version1: 第一个版本号
-            version2: 第二个版本号
-
-        Returns:
-            int: -1 if version1 < version2, 0 if equal, 1 if version1 > version2
-        """
-        v1_tuple = VersionComparator.parse_version(version1)
-        v2_tuple = VersionComparator.parse_version(version2)
-
-        if v1_tuple < v2_tuple:
-            return -1
-        elif v1_tuple > v2_tuple:
-            return 1
-        else:
-            return 0
-
-    @staticmethod
-    def check_forward_compatibility(current_version: str, max_version: str) -> tuple[bool, str]:
-        """检查向前兼容性（仅使用兼容性映射表）
-
-        Args:
-            current_version: 当前版本
-            max_version: 插件声明的最大支持版本
-
-        Returns:
-            Tuple[bool, str]: (是否兼容, 兼容信息)
-        """
-        current_normalized = VersionComparator.normalize_version(current_version)
-        max_normalized = VersionComparator.normalize_version(max_version)
-
-        # 检查兼容性映射表
-        if max_normalized in VersionComparator.COMPATIBILITY_MAP:
-            compatible_versions = VersionComparator.COMPATIBILITY_MAP[max_normalized]
-            if current_normalized in compatible_versions:
-                return True, f"根据兼容性映射表，版本 {current_normalized} 与 {max_normalized} 兼容"
-
-        return False, ""
-
-    @staticmethod
-    def is_version_in_range(version: str, min_version: str = "", max_version: str = "") -> tuple[bool, str]:
-        """检查版本是否在指定范围内，支持兼容性检查
-
-        Args:
-            version: 要检查的版本号
-            min_version: 最小版本号（可选）
-            max_version: 最大版本号（可选）
-
-        Returns:
-            Tuple[bool, str]: (是否兼容, 错误信息或兼容信息)
-        """
-        if not min_version and not max_version:
-            return True, ""
-
-        version_normalized = VersionComparator.normalize_version(version)
-
-        # 检查最小版本
-        if min_version:
-            min_normalized = VersionComparator.normalize_version(min_version)
-            if VersionComparator.compare_versions(version_normalized, min_normalized) < 0:
-                return False, f"版本 {version_normalized} 低于最小要求版本 {min_normalized}"
-
-        # 检查最大版本
-        if max_version:
-            max_normalized = VersionComparator.normalize_version(max_version)
-            comparison = VersionComparator.compare_versions(version_normalized, max_normalized)
-
-            if comparison > 0:
-                # 严格版本检查失败，尝试兼容性检查
-                is_compatible, compat_msg = VersionComparator.check_forward_compatibility(
-                    version_normalized, max_normalized
-                )
-
-                if not is_compatible:
-                    return False, f"版本 {version_normalized} 高于最大支持版本 {max_normalized}，且无兼容性映射"
-
-                logger.info(f"版本兼容性检查：{compat_msg}")
-                return True, compat_msg
-        return True, ""
-
-    @staticmethod
-    def get_current_host_version() -> str:
-        """获取当前主机应用版本
-
-        Returns:
-            str: 当前版本号
-        """
-        return VersionComparator.normalize_version(MMC_VERSION)
-
-    @staticmethod
-    def add_compatibility_mapping(base_version: str, compatible_versions: list) -> None:
-        """动态添加兼容性映射
-
-        Args:
-            base_version: 基础版本（插件声明的最大支持版本）
-            compatible_versions: 兼容的版本列表
-        """
-        base_normalized = VersionComparator.normalize_version(base_version)
-        VersionComparator.COMPATIBILITY_MAP[base_normalized] = [
-            VersionComparator.normalize_version(v) for v in compatible_versions
-        ]
-        logger.info(f"添加兼容性映射：{base_normalized} -> {compatible_versions}")
-
-    @staticmethod
-    def get_compatibility_info() -> dict[str, list]:
-        """获取当前的兼容性映射表
-
-        Returns:
-            Dict[str, list]: 兼容性映射表的副本
-        """
-        return VersionComparator.COMPATIBILITY_MAP.copy()
-
-
-class ManifestValidator:
-    """Manifest文件验证器"""
-
-    # 必需字段（必须存在且不能为空）
-    REQUIRED_FIELDS = ["manifest_version", "name", "version", "description", "author"]
-
-    # 可选字段（可以不存在或为空）
-    OPTIONAL_FIELDS = [
-        "license",
-        "host_application",
-        "homepage_url",
-        "repository_url",
-        "keywords",
-        "categories",
-        "default_locale",
-        "locales_path",
-        "plugin_info",
-    ]
-
-    # 建议填写的字段（会给出警告但不会导致验证失败）
-    RECOMMENDED_FIELDS = ["license", "keywords", "categories"]
-
-    SUPPORTED_MANIFEST_VERSIONS = [1]
-
-    def __init__(self):
-        self.validation_errors = []
-        self.validation_warnings = []
-
-    def validate_manifest(self, manifest_data: dict[str, Any]) -> bool:
-        """验证manifest数据
-
-        Args:
-            manifest_data: manifest数据字典
-
-        Returns:
-            bool: 是否验证通过（只有错误会导致验证失败，警告不会）
-        """
-        self.validation_errors.clear()
-        self.validation_warnings.clear()
-
-        # 检查必需字段
-        for field in self.REQUIRED_FIELDS:
-            if field not in manifest_data:
-                self.validation_errors.append(f"缺少必需字段: {field}")
-            elif not manifest_data[field]:
-                self.validation_errors.append(f"必需字段不能为空: {field}")
-
-        # 检查manifest版本
-        if "manifest_version" in manifest_data:
-            version = manifest_data["manifest_version"]
-            if version not in self.SUPPORTED_MANIFEST_VERSIONS:
-                self.validation_errors.append(
-                    f"不支持的manifest版本: {version}，支持的版本: {self.SUPPORTED_MANIFEST_VERSIONS}"
-                )
-
-        # 检查作者信息格式
-        if "author" in manifest_data:
-            author = manifest_data["author"]
-            if isinstance(author, dict):
-                if "name" not in author or not author["name"]:
-                    self.validation_errors.append("作者信息缺少name字段或为空")
-                # url字段是可选的
-                if author.get("url"):
-                    url = author["url"]
-                    if not (url.startswith("http://") or url.startswith("https://")):
-                        self.validation_warnings.append("作者URL建议使用完整的URL格式")
-            elif isinstance(author, str):
-                if not author.strip():
-                    self.validation_errors.append("作者信息不能为空")
-            else:
-                self.validation_errors.append("作者信息格式错误，应为字符串或包含name字段的对象")
-        # 检查主机应用版本要求（可选）
-        if "host_application" in manifest_data:
-            host_app = manifest_data["host_application"]
-            if isinstance(host_app, dict):
-                min_version = host_app.get("min_version", "")
-                max_version = host_app.get("max_version", "")
-
-                # 验证版本字段格式
-                for version_field in ["min_version", "max_version"]:
-                    if version_field in host_app and not host_app[version_field]:
-                        self.validation_warnings.append(f"host_application.{version_field}为空")
-
-                # 检查当前主机版本兼容性
-                if min_version or max_version:
-                    current_version = VersionComparator.get_current_host_version()
-                    is_compatible, error_msg = VersionComparator.is_version_in_range(
-                        current_version, min_version, max_version
-                    )
-
-                    if not is_compatible:
-                        self.validation_errors.append(f"版本兼容性检查失败: {error_msg} (当前版本: {current_version})")
-                    else:
-                        logger.debug(
-                            f"版本兼容性检查通过: 当前版本 {current_version} 符合要求 [{min_version}, {max_version}]"
-                        )
-            else:
-                self.validation_errors.append("host_application格式错误，应为对象")
-
-        # 检查URL格式（可选字段）
-        for url_field in ["homepage_url", "repository_url"]:
-            if manifest_data.get(url_field):
-                url: str = manifest_data[url_field]
-                if not (url.startswith("http://") or url.startswith("https://")):
-                    self.validation_warnings.append(f"{url_field}建议使用完整的URL格式")
-
-        # 检查数组字段格式（可选字段）
-        for list_field in ["keywords", "categories"]:
-            if list_field in manifest_data:
-                field_value = manifest_data[list_field]
-                if field_value is not None and not isinstance(field_value, list):
-                    self.validation_errors.append(f"{list_field}应为数组格式")
-                elif isinstance(field_value, list):
-                    # 检查数组元素是否为字符串
-                    for i, item in enumerate(field_value):
-                        if not isinstance(item, str):
-                            self.validation_warnings.append(f"{list_field}[{i}]应为字符串")
-
-        # 检查建议字段（给出警告）
-        for field in self.RECOMMENDED_FIELDS:
-            if field not in manifest_data or not manifest_data[field]:
-                self.validation_warnings.append(f"建议填写字段: {field}")
-
-        # 检查plugin_info结构（可选）
-        if "plugin_info" in manifest_data:
-            plugin_info = manifest_data["plugin_info"]
-            if isinstance(plugin_info, dict):
-                # 检查components数组
-                if "components" in plugin_info:
-                    components = plugin_info["components"]
-                    if not isinstance(components, list):
-                        self.validation_errors.append("plugin_info.components应为数组格式")
-                    else:
-                        for i, component in enumerate(components):
-                            if not isinstance(component, dict):
-                                self.validation_errors.append(f"plugin_info.components[{i}]应为对象")
-                            else:
-                                # 检查组件必需字段
-                                for comp_field in ["type", "name", "description"]:
-                                    if comp_field not in component or not component[comp_field]:
-                                        self.validation_errors.append(
-                                            f"plugin_info.components[{i}]缺少必需字段: {comp_field}"
-                                        )
-            else:
-                self.validation_errors.append("plugin_info应为对象格式")
-
-        return len(self.validation_errors) == 0
-
-    def get_validation_report(self) -> str:
-        """获取验证报告"""
-        report = []
-
-        if self.validation_errors:
-            report.append("❌ 验证错误:")
-            report.extend(f"  - {error}" for error in self.validation_errors)
-        if self.validation_warnings:
-            report.append("⚠️ 验证警告:")
-            report.extend(f"  - {warning}" for warning in self.validation_warnings)
-        if not self.validation_errors and not self.validation_warnings:
-            report.append("✅ Manifest文件验证通过")
-
-        return "\n".join(report)
-
-
-# class ManifestGenerator:
-#     """Manifest文件生成器"""
-
-#     def __init__(self):
-#         self.template = {
-#             "manifest_version": 1,
-#             "name": "",
-#             "version": "1.0.0",
-#             "description": "",
-#             "author": {"name": "", "url": ""},
-#             "license": "MIT",
-#             "host_application": {"min_version": "1.0.0", "max_version": "4.0.0"},
-#             "homepage_url": "",
-#             "repository_url": "",
-#             "keywords": [],
-#             "categories": [],
-#             "default_locale": "zh-CN",
-#             "locales_path": "_locales",
-#         }
-
-#     def generate_from_plugin(self, plugin_instance: BasePlugin) -> Dict[str, Any]:
-#         """从插件实例生成manifest
-
-#         Args:
-#             plugin_instance: BasePlugin实例
-
-#         Returns:
-#             Dict[str, Any]: 生成的manifest数据
-#         """
-#         manifest = self.template.copy()
-
-#         # 基本信息
-#         manifest["name"] = plugin_instance.plugin_name
-#         manifest["version"] = plugin_instance.plugin_version
-#         manifest["description"] = plugin_instance.plugin_description
-
-#         # 作者信息
-#         if plugin_instance.plugin_author:
-#             manifest["author"]["name"] = plugin_instance.plugin_author
-
-#         # 组件信息
-#         components = []
-#         plugin_components = plugin_instance.get_plugin_components()
-
-#         for component_info, component_class in plugin_components:
-#             component_data: Dict[str, Any] = {
-#                 "type": component_info.component_type.value,
-#                 "name": component_info.name,
-#                 "description": component_info.description,
-#             }
-
-#             # 添加激活模式信息（对于Action组件）
-#             if hasattr(component_class, "focus_activation_type"):
-#                 activation_modes = []
-#                 if hasattr(component_class, "focus_activation_type"):
-#                     activation_modes.append(component_class.focus_activation_type.value)
-#                 if hasattr(component_class, "normal_activation_type"):
-#                     activation_modes.append(component_class.normal_activation_type.value)
-#                 component_data["activation_modes"] = list(set(activation_modes))
-
-#             # 添加关键词信息
-#             if hasattr(component_class, "activation_keywords"):
-#                 keywords = getattr(component_class, "activation_keywords", [])
-#                 if keywords:
-#                     component_data["keywords"] = keywords
-
-#             components.append(component_data)
-
-#         manifest["plugin_info"] = {"is_built_in": True, "plugin_type": "general", "components": components}
-
-#         return manifest
-
-#     def save_manifest(self, manifest_data: Dict[str, Any], plugin_dir: str) -> bool:
-#         """保存manifest文件
-
-#         Args:
-#             manifest_data: manifest数据
-#             plugin_dir: 插件目录
-
-#         Returns:
-#             bool: 是否保存成功
-#         """
-#         try:
-#             manifest_path = os.path.join(plugin_dir, "_manifest.json")
-#             with open(manifest_path, "w", encoding="utf-8") as f:
-#                 orjson.dumps(manifest_data, f, ensure_ascii=False, indent=2)
-#             logger.info(f"Manifest文件已保存: {manifest_path}")
-#             return True
-#         except Exception as e:
-#             logger.error(f"保存manifest文件失败: {e}")
-#             return False
-
-
-# def validate_plugin_manifest(plugin_dir: str) -> bool:
-#     """验证插件目录中的manifest文件
-
-#     Args:
-#         plugin_dir: 插件目录路径
-
-#     Returns:
-#         bool: 是否验证通过
-#     """
-#     manifest_path = os.path.join(plugin_dir, "_manifest.json")
-
-#     if not os.path.exists(manifest_path):
-#         logger.warning(f"未找到manifest文件: {manifest_path}")
-#         return False
-
-#     try:
-#         with open(manifest_path, "r", encoding="utf-8") as f:
-#             manifest_data = orjson.loads(f.read())
-
-#         validator = ManifestValidator()
-#         is_valid = validator.validate_manifest(manifest_data)
-
-#         logger.info(f"Manifest验证结果:\n{validator.get_validation_report()}")
-
-#         return is_valid
-
-#     except Exception as e:
-#         logger.error(f"读取或验证manifest文件失败: {e}")
-#         return False
-
-
-# def generate_plugin_manifest(plugin_instance: BasePlugin, save_to_file: bool = True) -> Optional[Dict[str, Any]]:
-#     """为插件生成manifest文件
-
-#     Args:
-#         plugin_instance: BasePlugin实例
-#         save_to_file: 是否保存到文件
-
-#     Returns:
-#         Optional[Dict[str, Any]]: 生成的manifest数据
-#     """
-#     try:
-#         generator = ManifestGenerator()
-#         manifest_data = generator.generate_from_plugin(plugin_instance)
-
-#         if save_to_file and plugin_instance.plugin_dir:
-#             generator.save_manifest(manifest_data, plugin_instance.plugin_dir)
-
-#         return manifest_data
-
-#     except Exception as e:
-#         logger.error(f"生成manifest文件失败: {e}")
-#         return None

src/plugins/built_in/affinity_flow_chatter/__init__.py

@@ -0,0 +1,15 @@
+from src.plugin_system.base.plugin_metadata import PluginMetadata
+
+__plugin_meta__ = PluginMetadata(
+    name="Affinity Flow Chatter",
+    description="Built-in chatter plugin for affinity flow with interest scoring and relationship building",
+    usage="This plugin is automatically triggered by the system.",
+    version="1.0.0",
+    author="MoFox",
+    keywords=["chatter", "affinity", "conversation"],
+    categories=["Chat", "AI"],
+    extra={
+        "is_built_in": True
+    }
+)
+

src/plugins/built_in/affinity_flow_chatter/_manifest.json

@@ -1,23 +0,0 @@
-{
-  "manifest_version": 1,
-  "name": "affinity_chatter",
-  "display_name": "Affinity Flow Chatter",
-  "description": "Built-in chatter plugin for affinity flow with interest scoring and relationship building",
-  "version": "1.0.0",
-  "author": "MoFox",
-  "plugin_class": "AffinityChatterPlugin",
-  "enabled": true,
-  "is_built_in": true,
-  "components": [
-    {
-      "name": "affinity_chatter",
-      "type": "chatter",
-      "description": "Affinity flow chatter with intelligent interest scoring and relationship building",
-      "enabled": true,
-      "chat_type_allow": ["all"]
-    }
-  ],
-  "host_application": { "min_version": "0.8.0" },
-  "keywords": ["chatter", "affinity", "conversation"],
-  "categories": ["Chat", "AI"]
-}

src/plugins/built_in/core_actions/__init__.py

@@ -0,0 +1,17 @@
+from src.plugin_system.base.plugin_metadata import PluginMetadata
+
+__plugin_meta__ = PluginMetadata(
+    name="Emoji插件 (Emoji Actions)",
+    description="可以发送和管理Emoji",
+    usage="该插件提供 `emoji` action。",
+    version="1.0.0",
+    author="SengokuCola",
+    license="GPL-v3.0-or-later",
+    repository_url="https://github.com/MaiM-with-u/maibot",
+    keywords=["emoji", "action", "built-in"],
+    categories=["Emoji"],
+    extra={
+        "is_built_in": True,
+        "plugin_type": "action_provider",
+    }
+)

src/plugins/built_in/core_actions/_manifest.json

@@ -1,34 +0,0 @@
-{
-  "manifest_version": 1,
-  "name": "Emoji插件 (Emoji Actions)",
-  "version": "1.0.0",
-  "description": "可以发送和管理Emoji",
-  "author": {
-    "name": "SengokuCola",
-    "url": "https://github.com/MaiM-with-u"
-  },
-  "license": "GPL-v3.0-or-later",
-  
-  "host_application": {
-    "min_version": "0.10.0"
-  },
-  "homepage_url": "https://github.com/MaiM-with-u/maibot",
-  "repository_url": "https://github.com/MaiM-with-u/maibot",
-  "keywords": ["emoji", "action", "built-in"],
-  "categories": ["Emoji"],
-  
-  "default_locale": "zh-CN",
-  "locales_path": "_locales",
-  
-  "plugin_info": {
-    "is_built_in": true,
-    "plugin_type": "action_provider",
-    "components": [
-      {
-        "type": "action",
-        "name": "emoji",
-        "description": "作为一条全新的消息，发送一个符合当前情景的表情包来生动地表达情绪。"
-      }
-    ]
-  }
-}

src/plugins/built_in/maizone_refactored/__init__.py

@@ -1,8 +1,17 @@
-"""
-让框架能够发现并加载子目录中的组件。
-"""
+from src.plugin_system.base.plugin_metadata import PluginMetadata
 
-from .actions.read_feed_action import ReadFeedAction as ReadFeedAction
-from .actions.send_feed_action import SendFeedAction as SendFeedAction
-from .commands.send_feed_command import SendFeedCommand as SendFeedCommand
-from .plugin import MaiZoneRefactoredPlugin as MaiZoneRefactoredPlugin
+__plugin_meta__ = PluginMetadata(
+    name="MaiZone（麦麦空间）- 重构版",
+    description="（重构版）让你的麦麦发QQ空间说说、评论、点赞，支持AI配图、定时发送和自动监控功能",
+    usage="该插件提供 `send_feed` 和 `read_feed` action，以及 `send_feed` command。",
+    version="3.0.0",
+    author="MoFox-Studio",
+    license="GPL-v3.0",
+    repository_url="https://github.com/MoFox-Studio",
+    keywords=["QQ空间", "说说", "动态", "评论", "点赞", "自动化", "AI配图"],
+    categories=["社交", "自动化", "QQ空间"],
+    extra={
+        "is_built_in": False,
+        "plugin_type": "social",
+    }
+)

src/plugins/built_in/maizone_refactored/_manifest.json

@@ -1,47 +0,0 @@
-{
-  "manifest_version": 1,
-  "name": "MaiZone（麦麦空间）- 重构版",
-  "version": "3.0.0",
-  "description": "（重构版）让你的麦麦发QQ空间说说、评论、点赞，支持AI配图、定时发送和自动监控功能",
-  "author": {
-    "name": "MoFox-Studio",
-    "url": "https://github.com/MoFox-Studio"
-  },
-  "license": "GPL-v3.0",
-  
-  "host_application": {
-    "min_version": "0.10.0"
-  },
-  "keywords": ["QQ空间", "说说", "动态", "评论", "点赞", "自动化", "AI配图"],
-  "categories": ["社交", "自动化", "QQ空间"],
-  
-  "plugin_info": {
-    "is_built_in": false,
-    "plugin_type": "social",
-    "components": [
-      {
-        "type": "action",
-        "name": "send_feed",
-        "description": "根据指定主题发送一条QQ空间说说"
-      },
-      {
-        "type": "action", 
-        "name": "read_feed",
-        "description": "读取指定好友最近的说说，并评论点赞"
-      },
-      {
-        "type": "command",
-        "name": "send_feed", 
-        "description": "通过命令发送QQ空间说说"
-      }
-    ],
-    "features": [
-      "智能生成说说内容",
-      "AI自动配图（硅基流动）",
-      "自动点赞评论好友说说",
-      "定时发送说说",
-      "权限管理系统",
-      "历史记录避重"
-    ]
-  }
-}

src/plugins/built_in/napcat_adapter_plugin/__init__.py

@@ -0,0 +1,16 @@
+from src.plugin_system.base.plugin_metadata import PluginMetadata
+
+__plugin_meta__ = PluginMetadata(
+    name="napcat_plugin",
+    description="基于OneBot 11协议的NapCat QQ协议插件，提供完整的QQ机器人API接口，使用现有adapter连接",
+    usage="该插件提供 `napcat_tool` tool。",
+    version="1.0.0",
+    author="Windpicker_owo",
+    license="GPL-v3.0-or-later",
+    repository_url="https://github.com/Windpicker-owo/InternetSearchPlugin",
+    keywords=["qq", "bot", "napcat", "onebot", "api", "websocket"],
+    categories=["protocol"],
+    extra={
+        "is_built_in": False,
+    }
+)

src/plugins/built_in/napcat_adapter_plugin/_manifest.json

@@ -1,42 +0,0 @@
-{
-  "manifest_version": 1,
-  "name": "napcat_plugin",
-  "version": "1.0.0",
-  "description": "基于OneBot 11协议的NapCat QQ协议插件，提供完整的QQ机器人API接口，使用现有adapter连接",
-  "author": {
-    "name": "Windpicker_owo",
-    "url": "https://github.com/Windpicker-owo"
-  },
-  "license": "GPL-v3.0-or-later",
-  
-  "host_application": {
-    "min_version": "0.10.0",
-    "max_version": "0.11.0"
-  },
-  "homepage_url": "https://github.com/Windpicker-owo/InternetSearchPlugin",
-  "repository_url": "https://github.com/Windpicker-owo/InternetSearchPlugin",
-  "keywords": ["qq", "bot", "napcat", "onebot", "api", "websocket"],
-  "categories": ["protocol"],
-  "default_locale": "zh-CN",
-  "locales_path": "_locales",
-  
-  "plugin_info": {
-    "is_built_in": false,
-    "components": [
-      {
-        "type": "tool",
-        "name": "napcat_tool",
-        "description": "NapCat QQ协议综合工具，提供消息发送、群管理、好友管理、文件操作等完整功能"
-      }
-    ],
-    "features": [
-    "消息发送与接收",
-    "群管理功能",
-    "好友管理功能",
-    "文件上传下载",
-    "AI语音功能",
-    "群签到与戳一戳",
-    "现有adapter连接"
-  ]
-  }
-}

src/plugins/built_in/permission_management/__init__.py

@@ -0,0 +1,16 @@
+from src.plugin_system.base.plugin_metadata import PluginMetadata
+
+__plugin_meta__ = PluginMetadata(
+    name="权限管理插件（Permission Management）",
+    description="通过系统API管理权限",
+    usage="该插件提供 `permission_management` command。",
+    version="1.0.0",
+    author="MoFox-Studio",
+    license="GPL-v3.0-or-later",
+    repository_url="https://github.com/MoFox-Studio",
+    keywords=["plugins", "permission", "management", "built-in"],
+    extra={
+        "is_built_in": True,
+        "plugin_type": "permission",
+    }
+)

src/plugins/built_in/permission_management/_manifest.json

@@ -1,33 +0,0 @@
-{
-  "manifest_version": 1,
-  "name": "权限管理插件（Permission Management）",
-  "version": "1.0.0",
-  "description": "通过系统API管理权限",
-  "author": {
-    "name": "MoFox-Studio",
-    "url": "https://github.com/MoFox-Studio"
-  },
-  "license": "GPL-v3.0-or-later",
-  "host_application": {
-    "min_version": "0.10.0"
-  },
-  "keywords": [
-    "plugins",
-    "permission",
-    "management",
-    "built-in"
-  ],
-  "default_locale": "zh-CN",
-  "locales_path": "_locales",
-  "plugin_info": {
-    "is_built_in": true,
-    "plugin_type": "permission",
-    "components": [
-      {
-        "type": "command",
-        "name": "permission_management",
-        "description": "管理用户权限，包括添加、删除和修改权限等操作。"
-      }
-    ]
-  }
-}

src/plugins/built_in/plugin_management/__init__.py

@@ -0,0 +1,17 @@
+from src.plugin_system.base.plugin_metadata import PluginMetadata
+
+__plugin_meta__ = PluginMetadata(
+    name="插件和组件管理 (Plugin and Component Management)",
+    description="通过系统API管理插件和组件的生命周期，包括加载、卸载、启用和禁用等操作。",
+    usage="该插件提供 `plugin_management` command。",
+    version="1.0.0",
+    author="MaiBot团队",
+    license="GPL-v3.0-or-later",
+    repository_url="https://github.com/MaiM-with-u/maibot",
+    keywords=["plugins", "components", "management", "built-in"],
+    categories=["Core System", "Plugin Management"],
+    extra={
+        "is_built_in": True,
+        "plugin_type": "plugin_management",
+    }
+)

src/plugins/built_in/plugin_management/_manifest.json

@@ -1,39 +0,0 @@
-{
-  "manifest_version": 1,
-  "name": "插件和组件管理 (Plugin and Component Management)",
-  "version": "1.0.0",
-  "description": "通过系统API管理插件和组件的生命周期，包括加载、卸载、启用和禁用等操作。",
-  "author": {
-    "name": "MaiBot团队",
-    "url": "https://github.com/MaiM-with-u"
-  },
-  "license": "GPL-v3.0-or-later",
-  "host_application": {
-    "min_version": "0.9.1"
-  },
-  "homepage_url": "https://github.com/MaiM-with-u/maibot",
-  "repository_url": "https://github.com/MaiM-with-u/maibot",
-  "keywords": [
-    "plugins",
-    "components",
-    "management",
-    "built-in"
-  ],
-  "categories": [
-    "Core System",
-    "Plugin Management"
-  ],
-  "default_locale": "zh-CN",
-  "locales_path": "_locales",
-  "plugin_info": {
-    "is_built_in": true,
-    "plugin_type": "plugin_management",
-    "components": [
-      {
-        "type": "command",
-        "name": "plugin_management",
-        "description": "管理插件和组件的生命周期，包括加载、卸载、启用和禁用等操作。"
-      }
-    ]
-  }
-}

src/plugins/built_in/proactive_thinker/__init__.py

@@ -0,0 +1,17 @@
+from src.plugin_system.base.plugin_metadata import PluginMetadata
+
+__plugin_meta__ = PluginMetadata(
+    name="MoFox-Bot主动思考",
+    description="主动思考插件",
+    usage="该插件由系统自动触发。",
+    version="1.0.0",
+    author="MoFox-Studio",
+    license="GPL-v3.0-or-later",
+    repository_url="https://github.com/MoFox-Studio",
+    keywords=["主动思考","自己发消息"],
+    categories=["Chat", "Integration"],
+    extra={
+        "is_built_in": True,
+        "plugin_type": "functional"
+    }
+)

src/plugins/built_in/proactive_thinker/_manifest.json

@@ -1,25 +0,0 @@
-{
-  "manifest_version": 1,
-  "name": "MoFox-Bot主动思考",
-  "version": "1.0.0",
-  "description": "主动思考插件",
-  "author": {
-    "name": "MoFox-Studio",
-    "url": "https://github.com/MoFox-Studio"
-  },
-  "license": "GPL-v3.0-or-later",
-  
-  "host_application": {
-    "min_version": "0.10.0"
-  },
-  "keywords": ["主动思考","自己发消息"],
-  "categories": ["Chat", "Integration"],
-  
-  "default_locale": "zh-CN",
-  "locales_path": "_locales",
-  
-  "plugin_info": {
-    "is_built_in": "true",
-    "plugin_type": "functional"
-  }
-}

src/plugins/built_in/social_toolkit_plugin/__init__.py

@@ -0,0 +1,17 @@
+from src.plugin_system.base.plugin_metadata import PluginMetadata
+
+__plugin_meta__ = PluginMetadata(
+    name="MoFox-Bot工具箱",
+    description="一个集合多种实用功能的插件，旨在提升聊天体验和效率。",
+    usage="该插件提供多种命令，详情请查阅文档。",
+    version="1.0.0",
+    author="MoFox-Studio",
+    license="GPL-v3.0-or-later",
+    repository_url="https://github.com/MoFox-Studio",
+    keywords=["emoji", "reaction", "like", "表情", "回应", "点赞"],
+    categories=["Chat", "Integration"],
+    extra={
+        "is_built_in": "true",
+        "plugin_type": "functional"
+    }
+)

src/plugins/built_in/social_toolkit_plugin/_manifest.json

@@ -1,25 +0,0 @@
-{
-  "manifest_version": 1,
-  "name": "MoFox-Bot工具箱",
-  "version": "1.0.0",
-  "description": "一个集合多种实用功能的插件，旨在提升聊天体验和效率。",
-  "author": {
-    "name": "MoFox-Studio",
-    "url": "https://github.com/MoFox-Studio"
-  },
-  "license": "GPL-v3.0-or-later",
-  
-  "host_application": {
-    "min_version": "0.10.0"
-  },
-  "keywords": ["emoji", "reaction", "like", "表情", "回应", "点赞"],
-  "categories": ["Chat", "Integration"],
-  
-  "default_locale": "zh-CN",
-  "locales_path": "_locales",
-  
-  "plugin_info": {
-    "is_built_in": "true",
-    "plugin_type": "functional"
-  }
-}

src/plugins/built_in/tts_plugin/__init__.py

@@ -0,0 +1,17 @@
+from src.plugin_system.base.plugin_metadata import PluginMetadata
+
+__plugin_meta__ = PluginMetadata(
+    name="文本转语音插件 (Text-to-Speech)",
+    description="将文本转换为语音进行播放的插件，支持多种语音模式和智能语音输出场景判断。",
+    usage="该插件提供 `tts_action` action。",
+    version="0.1.0",
+    author="MaiBot团队",
+    license="GPL-v3.0-or-later",
+    repository_url="https://github.com/MaiM-with-u/maibot",
+    keywords=["tts", "voice", "audio", "speech", "accessibility"],
+    categories=["Audio Tools", "Accessibility", "Voice Assistant"],
+    extra={
+        "is_built_in": True,
+        "plugin_type": "audio_processor",
+    }
+)

src/plugins/built_in/tts_plugin/_manifest.json

@@ -1,42 +0,0 @@
-{
-  "manifest_version": 1,
-  "name": "文本转语音插件 (Text-to-Speech)",
-  "version": "0.1.0",
-  "description": "将文本转换为语音进行播放的插件，支持多种语音模式和智能语音输出场景判断。",
-  "author": {
-    "name": "MaiBot团队", 
-    "url": "https://github.com/MaiM-with-u"
-  },
-  "license": "GPL-v3.0-or-later",
-  
-  "host_application": {
-    "min_version": "0.8.0"
-  },
-  "homepage_url": "https://github.com/MaiM-with-u/maibot",
-  "repository_url": "https://github.com/MaiM-with-u/maibot",
-  "keywords": ["tts", "voice", "audio", "speech", "accessibility"],
-  "categories": ["Audio Tools", "Accessibility", "Voice Assistant"],
-  
-  "default_locale": "zh-CN",
-  "locales_path": "_locales",
-  
-  "plugin_info": {
-    "is_built_in": true,
-    "plugin_type": "audio_processor",
-    "components": [
-      {
-        "type": "action",
-        "name": "tts_action", 
-        "description": "将文本转换为语音进行播放",
-        "activation_modes": ["llm_judge", "keyword"],
-        "keywords": ["语音", "tts", "播报", "读出来", "语音播放", "听", "朗读"]
-      }
-    ],
-    "features": [
-      "文本转语音播放",
-      "智能场景判断", 
-      "关键词触发",
-      "支持多种语音模式"
-    ]
-  }
-}

src/plugins/built_in/web_search_tool/__init__.py

@@ -0,0 +1,16 @@
+from src.plugin_system.base.plugin_metadata import PluginMetadata
+
+__plugin_meta__ = PluginMetadata(
+    name="Web Search Tool",
+    description="A tool for searching the web.",
+    usage="This plugin provides a `web_search` tool.",
+    version="1.0.0",
+    author="MoFox-Studio",
+    license="GPL-v3.0-or-later",
+    repository_url="https://github.com/MoFox-Studio",
+    keywords=["web", "search", "tool"],
+    categories=["Tools"],
+    extra={
+        "is_built_in": True,
+    }
+)

template/bot_config_template.toml

@@ -1,5 +1,5 @@
 [inner]
-version = "7.1.6"
+version = "7.1.8"
 
 #----以下是给开发人员阅读的，如果你只是部署了MoFox-Bot，不需要阅读----
 #如果你想要修改配置文件，请递增version的值
@@ -209,7 +209,7 @@ enable_memory = true # 是否启用记忆系统
 memory_build_interval = 600 # 记忆构建间隔（秒）。间隔越低，学习越频繁，但可能产生更多冗余信息
 
 # === 记忆采样系统配置 ===
-memory_sampling_mode = "immediate" # 记忆采样模式：hippocampus(海马体定时采样)，immediate(即时采样)，all(所有模式)
+memory_sampling_mode = "immediate" # 记忆采样模式：'immediate'(即时采样), 'hippocampus'(海马体定时采样) or 'all'(双模式)
 
 # 海马体双峰采样配置
 enable_hippocampus_sampling = true # 启用海马体双峰采样策略
@@ -370,6 +370,7 @@ date_style = "m-d H:i:s" # 日期格式
 log_level_style = "lite" # 日志级别样式,可选FULL，compact，lite
 color_text = "full" # 日志文本颜色，可选none，title，full
 log_level = "INFO" # 全局日志级别（向下兼容，优先级低于下面的分别设置）
+file_retention_days = 30  # 文件日志保留天数，0=禁用文件日志，-1=永不删除
 console_log_level = "INFO" # 控制台日志级别，可选: DEBUG, INFO, WARNING, ERROR, CRITICAL
 file_log_level = "DEBUG" # 文件日志级别，可选: DEBUG, INFO, WARNING, ERROR, CRITICAL