add 插件内存统计API

app/api/endpoints/plugin.py

@@ -13,7 +13,7 @@ from app import schemas
 from app.command import Command
 from app.core.config import settings
 from app.core.plugin import PluginManager
-from app.core.security import verify_apikey, verify_token
+from app.core.security import verify_apikey, verify_token, verify_apitoken
 from app.db.models import User
 from app.db.systemconfig_oper import SystemConfigOper
 from app.db.user_oper import get_current_active_superuser, get_current_active_superuser_async
@@ -21,6 +21,7 @@ from app.factory import app
 from app.helper.plugin import PluginHelper
 from app.log import logger
 from app.scheduler import Scheduler
+from app.schemas.plugin import PluginMemoryInfo
 from app.schemas.types import SystemConfigKey
 
 PROTECTED_ROUTES = {"/api/v1/openapi.json", "/docs", "/docs/oauth2-redirect", "/redoc"}
@@ -463,6 +464,87 @@ async def update_folder_plugins(folder_name: str, plugin_ids: List[str],
     return schemas.Response(success=True, message=f"文件夹 '{folder_name}' 中的插件已更新")
 
 
+@router.post("/clone/{plugin_id}", summary="创建插件分身", response_model=schemas.Response)
+def clone_plugin(plugin_id: str,
+                 clone_data: dict,
+                 _: User = Depends(get_current_active_superuser)) -> Any:
+    """
+    创建插件分身
+    """
+    try:
+        success, message = PluginManager().clone_plugin(
+            plugin_id=plugin_id,
+            suffix=clone_data.get("suffix", ""),
+            name=clone_data.get("name", ""),
+            description=clone_data.get("description", ""),
+            version=clone_data.get("version", ""),
+            icon=clone_data.get("icon", "")
+        )
+
+        if success:
+            # 注册插件服务
+            reload_plugin(message)
+            # 将分身插件添加到原插件所在的文件夹中
+            _add_clone_to_plugin_folder(plugin_id, message)
+            return schemas.Response(success=True, message="插件分身创建成功")
+        else:
+            return schemas.Response(success=False, message=message)
+    except Exception as e:
+        logger.error(f"创建插件分身失败：{str(e)}")
+        return schemas.Response(success=False, message=f"创建插件分身失败：{str(e)}")
+
+
+@router.get("/memory", summary="插件内存使用统计", response_model=List[PluginMemoryInfo])
+def plugin_memory_stats(_: Annotated[str, Depends(verify_apitoken)]) -> Any:
+    """
+    获取所有插件的内存使用统计信息
+    """
+    try:
+        plugin_manager = PluginManager()
+        memory_stats = plugin_manager.get_plugin_memory_stats()
+        return memory_stats
+    except Exception as e:
+        logger.error(f"获取插件内存统计失败：{str(e)}")
+        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                            detail=f"获取插件内存统计失败：{str(e)}")
+
+
+@router.get("/memory/{plugin_id}", summary="单个插件内存使用统计", response_model=PluginMemoryInfo)
+def plugin_memory_stat(plugin_id: str, _: Annotated[str, Depends(verify_apitoken)]) -> Any:
+    """
+    获取指定插件的内存使用统计信息
+    """
+    try:
+        plugin_manager = PluginManager()
+        memory_stats = plugin_manager.get_plugin_memory_stats(plugin_id)
+        if not memory_stats:
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
+                                detail=f"插件 {plugin_id} 不存在或未运行")
+        return memory_stats[0]
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"获取插件 {plugin_id} 内存统计失败：{str(e)}")
+        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                            detail=f"获取插件内存统计失败：{str(e)}")
+
+
+@router.delete("/memory/cache", summary="清除插件内存统计缓存")
+def clear_plugin_memory_cache(_: Annotated[str, Depends(verify_apitoken)],
+                              plugin_id: Optional[str] = None) -> Any:
+    """
+    清除插件内存统计缓存
+    """
+    try:
+        plugin_manager = PluginManager()
+        plugin_manager.clear_plugin_memory_cache(plugin_id)
+        message = f"已清除插件 {plugin_id} 的内存统计缓存" if plugin_id else "已清除所有插件的内存统计缓存"
+        return schemas.Response(success=True, message=message)
+    except Exception as e:
+        logger.error(f"清除插件内存统计缓存失败：{str(e)}")
+        return schemas.Response(success=False, message=f"清除缓存失败：{str(e)}")
+
+
 @router.get("/{plugin_id}", summary="获取插件配置")
 async def plugin_config(plugin_id: str,
                         _: User = Depends(get_current_active_superuser_async)) -> dict:
@@ -528,36 +610,6 @@ def uninstall_plugin(plugin_id: str,
     return schemas.Response(success=True)
 
 
-@router.post("/clone/{plugin_id}", summary="创建插件分身", response_model=schemas.Response)
-def clone_plugin(plugin_id: str,
-                 clone_data: dict,
-                 _: User = Depends(get_current_active_superuser)) -> Any:
-    """
-    创建插件分身
-    """
-    try:
-        success, message = PluginManager().clone_plugin(
-            plugin_id=plugin_id,
-            suffix=clone_data.get("suffix", ""),
-            name=clone_data.get("name", ""),
-            description=clone_data.get("description", ""),
-            version=clone_data.get("version", ""),
-            icon=clone_data.get("icon", "")
-        )
-
-        if success:
-            # 注册插件服务
-            reload_plugin(message)
-            # 将分身插件添加到原插件所在的文件夹中
-            _add_clone_to_plugin_folder(plugin_id, message)
-            return schemas.Response(success=True, message="插件分身创建成功")
-        else:
-            return schemas.Response(success=False, message=message)
-    except Exception as e:
-        logger.error(f"创建插件分身失败：{str(e)}")
-        return schemas.Response(success=False, message=f"创建插件分身失败：{str(e)}")
-
-
 def _add_clone_to_plugin_folder(original_plugin_id: str, clone_plugin_id: str):
     """
     将分身插件添加到原插件所在的文件夹中

app/core/plugin.py

@@ -21,7 +21,7 @@ from app.core.config import settings
 from app.core.event import eventmanager, Event
 from app.db.plugindata_oper import PluginDataOper
 from app.db.systemconfig_oper import SystemConfigOper
-from app.helper.plugin import PluginHelper
+from app.helper.plugin import PluginHelper, PluginMemoryMonitor
 from app.helper.sites import SitesHelper  # noqa
 from app.log import logger
 from app.schemas.types import EventType, SystemConfigKey
@@ -98,6 +98,8 @@ class PluginManager(metaclass=Singleton):
         self._config_key: str = "plugin.%s"
         # 监听器
         self._observer: Observer = None
+        # 内存监控器
+        self._memory_monitor = PluginMemoryMonitor()
         # 开发者模式监测插件修改
         if settings.DEV or settings.PLUGIN_AUTO_RELOAD:
             self.__start_monitor()
@@ -863,6 +865,28 @@ class PluginManager(metaclass=Singleton):
         """
         return list(self._running_plugins.keys())
 
+    def get_plugin_memory_stats(self, pid: Optional[str] = None) -> List[Dict[str, Any]]:
+        """
+        获取插件内存统计信息
+        :param pid: 插件ID，为空则获取所有插件
+        :return: 内存统计信息列表
+        """
+        if pid:
+            plugin_instance = self._running_plugins.get(pid)
+            if plugin_instance:
+                return [self._memory_monitor.get_plugin_memory_usage(pid, plugin_instance)]
+            else:
+                return []
+        else:
+            return self._memory_monitor.get_all_plugins_memory_usage(self._running_plugins)
+
+    def clear_plugin_memory_cache(self, pid: Optional[str] = None):
+        """
+        清除插件内存统计缓存
+        :param pid: 插件ID，为空则清除所有缓存
+        """
+        self._memory_monitor.clear_cache(pid)
+
     def get_online_plugins(self, force: bool = False) -> List[schemas.Plugin]:
         """
         获取所有在线插件信息

app/core/security.py

@@ -252,19 +252,19 @@ def __verify_key(key: str, expected_key: str, key_type: str) -> str:
 def verify_apitoken(token: Annotated[str, Security(__get_api_token)]) -> str:
     """
     使用 API Token 进行身份认证
-    :param token: API Token，从 URL 查询参数中获取
+    :param token: API Token，从 URL 查询参数中获取 token=xxx
     :return: 返回校验通过的 API Token
     """
-    return __verify_key(token, settings.API_TOKEN, "API_TOKEN")
+    return __verify_key(token, settings.API_TOKEN, "token")
 
 
 def verify_apikey(apikey: Annotated[str, Security(__get_api_key)]) -> str:
     """
     使用 API Key 进行身份认证
-    :param apikey: API Key，从 URL 查询参数或请求头中获取
+    :param apikey: API Key，从 URL 查询参数中获取 apikey=xxx
     :return: 返回校验通过的 API Key
     """
-    return __verify_key(apikey, settings.API_TOKEN, "API_KEY")
+    return __verify_key(apikey, settings.API_TOKEN, "apikey")
 
 
 def verify_password(plain_password: str, hashed_password: str) -> bool:

app/helper/plugin.py

@@ -4,10 +4,11 @@ import json
 import shutil
 import site
 import sys
+import time
 import traceback
 import zipfile
 from pathlib import Path
-from typing import Dict, List, Optional, Tuple, Set, Callable, Awaitable
+from typing import Dict, List, Optional, Tuple, Set, Callable, Awaitable, Any
 
 import aiofiles
 import aioshutil
@@ -24,6 +25,7 @@ from app.db.systemconfig_oper import SystemConfigOper
 from app.log import logger
 from app.schemas.types import SystemConfigKey
 from app.utils.http import RequestUtils, AsyncRequestUtils
+from app.utils.memory import MemoryCalculator
 from app.utils.singleton import WeakSingleton
 from app.utils.system import SystemUtils
 from app.utils.url import UrlUtils
@@ -1569,3 +1571,87 @@ class PluginHelper(metaclass=WeakSingleton):
         except Exception as e:
             logger.error(f"解压 Release 压缩包失败：{e}")
             return False, f"解压 Release 压缩包失败：{e}"
+
+
+class PluginMemoryMonitor:
+    """
+    插件内存监控器
+    """
+
+    def __init__(self):
+        self._calculator = MemoryCalculator()
+        self._cache = {}
+        self._cache_ttl = 300  # 缓存5分钟
+
+    def get_plugin_memory_usage(self, plugin_id: str, plugin_instance: Any) -> Dict[str, Any]:
+        """
+        获取插件内存使用情况
+        :param plugin_id: 插件ID
+        :param plugin_instance: 插件实例
+        :return: 内存使用信息
+        """
+        # 检查缓存
+        if self._is_cache_valid(plugin_id):
+            return self._cache[plugin_id]
+
+        # 计算内存使用
+        memory_info = self._calculator.calculate_object_memory(plugin_instance)
+
+        # 添加插件信息
+        result = {
+            'plugin_id': plugin_id,
+            'plugin_name': getattr(plugin_instance, 'plugin_name', 'Unknown'),
+            'plugin_version': getattr(plugin_instance, 'plugin_version', 'Unknown'),
+            'timestamp': time.time(),
+            **memory_info
+        }
+
+        # 更新缓存
+        self._cache[plugin_id] = result
+        return result
+
+    def get_all_plugins_memory_usage(self, plugins: Dict[str, Any]) -> List[Dict[str, Any]]:
+        """
+        获取所有插件的内存使用情况
+        :param plugins: 插件实例字典
+        :return: 内存使用信息列表
+        """
+        results = []
+        for plugin_id, plugin_instance in plugins.items():
+            if plugin_instance:
+                try:
+                    memory_info = self.get_plugin_memory_usage(plugin_id, plugin_instance)
+                    results.append(memory_info)
+                except Exception as e:
+                    logger.error(f"获取插件 {plugin_id} 内存使用情况失败：{str(e)}")
+                    results.append({
+                        'plugin_id': plugin_id,
+                        'plugin_name': getattr(plugin_instance, 'plugin_name', 'Unknown'),
+                        'error': str(e),
+                        'total_memory_bytes': 0,
+                        'total_memory_mb': 0,
+                        'object_count': 0,
+                        'calculation_time_ms': 0
+                    })
+
+        # 按内存使用量排序
+        results.sort(key=lambda x: x.get('total_memory_bytes', 0), reverse=True)
+        return results
+
+    def _is_cache_valid(self, plugin_id: str) -> bool:
+        """
+        检查缓存是否有效
+        """
+        if plugin_id not in self._cache:
+            return False
+        return time.time() - self._cache[plugin_id]['timestamp'] < self._cache_ttl
+
+    def clear_cache(self, plugin_id: Optional[str] = None):
+        """
+        清除缓存
+        :param plugin_id: 插件ID，为空则清除所有缓存
+        """
+        if plugin_id:
+            self._cache.pop(plugin_id, None)
+        else:
+            self._cache.clear()

app/schemas/plugin.py

@@ -1,4 +1,4 @@
-from typing import Optional, List
+from typing import Optional, List, Dict, Any
 
 from pydantic import BaseModel, Field
 
@@ -67,3 +67,17 @@ class PluginDashboard(Plugin):
     cols: Optional[dict] = Field(default_factory=dict)
     # 页面元素
     elements: Optional[List[dict]] = Field(default_factory=list)
+
+
+class PluginMemoryInfo(BaseModel):
+    """插件内存信息"""
+    plugin_id: str = Field(description="插件ID")
+    plugin_name: str = Field(description="插件名称")
+    plugin_version: str = Field(description="插件版本")
+    total_memory_bytes: int = Field(description="总内存使用量(字节)")
+    total_memory_mb: float = Field(description="总内存使用量(MB)")
+    object_count: int = Field(description="对象数量")
+    calculation_time_ms: float = Field(description="计算耗时(毫秒)")
+    timestamp: float = Field(description="统计时间戳")
+    error: Optional[str] = Field(default=None, description="错误信息")
+    object_details: Optional[List[Dict[str, Any]]] = Field(default=None, description="大对象详情")

app/utils/memory.py

@@ -0,0 +1,178 @@
+import sys
+import time
+from collections import deque
+from typing import Any, Dict, Set
+
+from app.log import logger
+
+
+class MemoryCalculator:
+    """
+    内存计算器，用于递归计算对象的内存占用
+    """
+
+    def __init__(self):
+        # 缓存已计算的对象ID，避免重复计算
+        self._calculated_ids: Set[int] = set()
+        # 最大递归深度，防止无限递归
+        self._max_depth = 10
+        # 最大对象数量，防止计算过多对象
+        self._max_objects = 10000
+
+    def calculate_object_memory(self, obj: Any, max_depth: int = None, max_objects: int = None) -> Dict[str, Any]:
+        """
+        计算对象的内存占用
+        :param obj: 要计算的对象
+        :param max_depth: 最大递归深度
+        :param max_objects: 最大对象数量
+        :return: 内存统计信息
+        """
+        if max_depth is None:
+            max_depth = self._max_depth
+        if max_objects is None:
+            max_objects = self._max_objects
+
+        # 重置缓存
+        self._calculated_ids.clear()
+
+        start_time = time.time()
+        object_details = []
+
+        try:
+            # 递归计算内存
+            memory_info = self._calculate_recursive(obj, depth=0, max_depth=max_depth,
+                                                    max_objects=max_objects, object_count=0)
+            total_memory = memory_info['total_memory']
+            object_count = memory_info['object_count']
+            object_details = memory_info['object_details']
+
+        except Exception as e:
+            logger.error(f"计算对象内存时出错：{str(e)}")
+            total_memory = 0
+            object_count = 0
+
+        calculation_time = time.time() - start_time
+
+        return {
+            'total_memory_bytes': total_memory,
+            'total_memory_mb': round(total_memory / (1024 * 1024), 2),
+            'object_count': object_count,
+            'calculation_time_ms': round(calculation_time * 1000, 2),
+            'object_details': object_details[:10]  # 只返回前10个最大的对象
+        }
+
+    def _calculate_recursive(self, obj: Any, depth: int, max_depth: int,
+                             max_objects: int, object_count: int) -> Dict[str, Any]:
+        """
+        递归计算对象内存
+        """
+        if depth > max_depth or object_count > max_objects:
+            return {
+                'total_memory': 0,
+                'object_count': object_count,
+                'object_details': []
+            }
+
+        total_memory = 0
+        object_details = []
+
+        # 获取对象ID，避免重复计算
+        obj_id = id(obj)
+        if obj_id in self._calculated_ids:
+            return {
+                'total_memory': 0,
+                'object_count': object_count,
+                'object_details': []
+            }
+
+        self._calculated_ids.add(obj_id)
+        object_count += 1
+
+        try:
+            # 计算对象本身的内存
+            obj_memory = sys.getsizeof(obj)
+            total_memory += obj_memory
+
+            # 记录大对象
+            if obj_memory > 1024:  # 大于1KB的对象
+                object_details.append({
+                    'type': type(obj).__name__,
+                    'memory_bytes': obj_memory,
+                    'memory_mb': round(obj_memory / (1024 * 1024), 2),
+                    'depth': depth
+                })
+
+            # 递归计算容器对象的内容
+            if depth < max_depth:
+                container_memory = self._calculate_container_memory(
+                    obj, depth + 1, max_depth, max_objects, object_count
+                )
+                total_memory += container_memory['total_memory']
+                object_count = container_memory['object_count']
+                object_details.extend(container_memory['object_details'])
+
+        except Exception as e:
+            logger.debug(f"计算对象 {type(obj).__name__} 内存时出错：{str(e)}")
+
+        return {
+            'total_memory': total_memory,
+            'object_count': object_count,
+            'object_details': object_details
+        }
+
+    def _calculate_container_memory(self, obj: Any, depth: int, max_depth: int,
+                                    max_objects: int, object_count: int) -> Dict[str, Any]:
+        """
+        计算容器对象的内存
+        """
+        total_memory = 0
+        object_details = []
+
+        try:
+            # 处理不同类型的容器
+            if isinstance(obj, (list, tuple, deque)):
+                for item in obj:
+                    if object_count > max_objects:
+                        break
+                    item_memory = self._calculate_recursive(item, depth, max_depth, max_objects, object_count)
+                    total_memory += item_memory['total_memory']
+                    object_count = item_memory['object_count']
+                    object_details.extend(item_memory['object_details'])
+
+            elif isinstance(obj, dict):
+                for key, value in obj.items():
+                    if object_count > max_objects:
+                        break
+                    # 计算key的内存
+                    key_memory = self._calculate_recursive(key, depth, max_depth, max_objects, object_count)
+                    total_memory += key_memory['total_memory']
+                    object_count = key_memory['object_count']
+                    object_details.extend(key_memory['object_details'])
+
+                    # 计算value的内存
+                    value_memory = self._calculate_recursive(value, depth, max_depth, max_objects, object_count)
+                    total_memory += value_memory['total_memory']
+                    object_count = value_memory['object_count']
+                    object_details.extend(value_memory['object_details'])
+
+            elif hasattr(obj, '__dict__'):
+                # 处理有__dict__属性的对象
+                for attr_name, attr_value in obj.__dict__.items():
+                    if object_count > max_objects:
+                        break
+                    # 跳过一些特殊属性
+                    if attr_name.startswith('_') and attr_name not in ['_calculated_ids']:
+                        continue
+                    attr_memory = self._calculate_recursive(attr_value, depth, max_depth, max_objects, object_count)
+                    total_memory += attr_memory['total_memory']
+                    object_count = attr_memory['object_count']
+                    object_details.extend(attr_memory['object_details'])
+
+        except Exception as e:
+            logger.debug(f"计算容器对象 {type(obj).__name__} 内存时出错：{str(e)}")
+
+        return {
+            'total_memory': total_memory,
+            'object_count': object_count,
+            'object_details': object_details
+        }