🎨 改为Model类型

zhenxun/builtin_plugins/web_ui/__init__.py

@@ -92,6 +92,8 @@ WsApiRouter.include_router(chat_routes)
 @driver.on_startup
 async def _():
     try:
+        # 存储任务引用的列表，防止任务被垃圾回收
+        _tasks = []
 
         async def log_sink(message: str):
             loop = None
@@ -102,7 +104,8 @@ async def _():
                     logger.warning("Web Ui log_sink", e=e)
             if not loop:
                 loop = asyncio.new_event_loop()
-            loop.create_task(LOG_STORAGE.add(message.rstrip("\n")))
+            # 存储任务引用到外部列表中
+            _tasks.append(loop.create_task(LOG_STORAGE.add(message.rstrip("\n"))))
 
         logger_.add(
             log_sink, colorize=True, filter=default_filter, format=default_format

zhenxun/builtin_plugins/web_ui/api/tabs/plugin_manage/__init__.py

@@ -9,13 +9,14 @@ from ....base_model import Result
 from ....utils import authentication
 from .data_source import ApiDataSource
 from .model import (
+    BatchUpdatePlugins,
+    BatchUpdateResult,
     PluginCount,
     PluginDetail,
     PluginInfo,
     PluginSwitch,
-    UpdatePlugin,
-    BatchUpdatePlugins,
     RenameMenuTypePayload,
+    UpdatePlugin,
 )
 
 router = APIRouter(prefix="/plugin")
@@ -32,9 +33,8 @@ async def _(
     plugin_type: list[PluginType] = Query(None), menu_type: str | None = None
 ) -> Result[list[PluginInfo]]:
     try:
-        return Result.ok(
+        result = await ApiDataSource.get_plugin_list(plugin_type, menu_type)
-            await ApiDataSource.get_plugin_list(plugin_type, menu_type), "拿到信息啦!"
+        return Result.ok(result, "拿到信息啦!")
-        )
     except Exception as e:
         logger.error(f"{router.prefix}/get_plugin_list 调用错误", "WebUi", e=e)
         return Result.fail(f"发生了一点错误捏 {type(e)}: {e}")
@@ -146,9 +146,8 @@ async def _() -> Result[list[str]]:
 )
 async def _(module: str) -> Result[PluginDetail]:
     try:
-        return Result.ok(
+        detail = await ApiDataSource.get_plugin_detail(module)
-            await ApiDataSource.get_plugin_detail(module), "已经帮你写好啦!"
+        return Result.ok(detail, "已经帮你写好啦!")
-        )
     except (ValueError, KeyError):
         return Result.fail("插件数据不存在...")
     except Exception as e:
@@ -156,34 +155,51 @@ async def _(module: str) -> Result[PluginDetail]:
         return Result.fail(f"{type(e)}: {e}")
 
 
-@router.put("/plugins/batch_update", summary="批量更新插件配置")
+@router.put(
-async def batch_update_plugin_config_api(params: BatchUpdatePlugins):
+    "/plugins/batch_update",
+    dependencies=[authentication()],
+    response_model=Result[BatchUpdateResult],
+    response_class=JSONResponse,
+    summary="批量更新插件配置",
+)
+async def batch_update_plugin_config_api(
+    params: BatchUpdatePlugins,
+) -> Result[BatchUpdateResult]:
     """批量更新插件配置，如开关、类型等"""
-    result = await ApiDataSource.batch_update_plugins(params=params)
+    try:
-    if result["errors"]:
+        result_dict = await ApiDataSource.batch_update_plugins(params=params)
-        # 可以根据需要返回更详细的错误信息，或者只返回一个笼统的失败信息
+        result_model = BatchUpdateResult(
-        # 这里我们返回包含错误详情的 200 OK，让前端处理
+            success=result_dict["success"],
-        # 或者可以抛出 HTTPException
+            updated_count=result_dict["updated_count"],
-        # from fastapi import HTTPException
+            errors=result_dict["errors"],
-        # raise HTTPException(status_code=400, detail={"message": "部分插件更新失败", "errors": result["errors"]})
+        )
-        pass # 暂时只返回结果字典
+        return Result.ok(result_model, "插件配置更新完成")
-    return result
+    except Exception as e:
+        logger.error(f"{router.prefix}/plugins/batch_update 调用错误", "WebUi", e=e)
+        return Result.fail(f"发生了一点错误捏 {type(e)}: {e}")
 
 
 # 新增：重命名菜单类型路由
 @router.put(
-    "/menu_type/rename", 
+    "/menu_type/rename",
-    dependencies=[authentication()], 
+    dependencies=[authentication()],
     response_model=Result,
-    summary="重命名菜单类型"
+    summary="重命名菜单类型",
 )
 async def rename_menu_type_api(payload: RenameMenuTypePayload) -> Result:
     try:
-        result = await ApiDataSource.rename_menu_type(old_name=payload.old_name, new_name=payload.new_name)
+        result = await ApiDataSource.rename_menu_type(
+            old_name=payload.old_name, new_name=payload.new_name
+        )
         if result.get("success"):
-            return Result.ok(info=result.get("info", f"成功将 {result.get('updated_count', 0)} 个插件的菜单类型从 '{payload.old_name}' 修改为 '{payload.new_name}'"))
+            return Result.ok(
+                info=result.get(
+                    "info",
+                    f"成功将 {result.get('updated_count', 0)} 个插件的菜单类型从 "
+                    f"'{payload.old_name}' 修改为 '{payload.new_name}'",
+                )
+            )
         else:
-             # 这种情况理论上不会发生，因为 rename_menu_type 失败会抛异常
             return Result.fail(info=result.get("info", "重命名失败"))
     except ValueError as ve:
         return Result.fail(info=str(ve))

zhenxun/builtin_plugins/web_ui/api/tabs/plugin_manage/data_source.py

@@ -2,12 +2,12 @@ import re
 
 import cattrs
 from fastapi import Query
+from tortoise.exceptions import DoesNotExist
 
 from zhenxun.configs.config import Config
 from zhenxun.configs.utils import ConfigGroup
 from zhenxun.models.plugin_info import PluginInfo as DbPluginInfo
 from zhenxun.utils.enum import BlockType, PluginType
-from tortoise.exceptions import DoesNotExist
 
 from .model import (
     BatchUpdatePlugins,
@@ -134,7 +134,7 @@ class ApiDataSource:
                         errors.append(
                             {
                                 "module": item.module,
-                                "error": f"Save block_type failed: {str(e_save)}",
+                                "error": f"Save block_type failed: {e_save!s}",
                             }
                         )
                         plugin_changed_other = False
@@ -158,7 +158,7 @@ class ApiDataSource:
                 errors.append(
                     {
                         "module": "batch_update_other",
-                        "error": f"Bulk update failed: {str(e_bulk)}",
+                        "error": f"Bulk update failed: {e_bulk!s}",
                     }
                 )
 
@@ -217,20 +217,26 @@ class ApiDataSource:
         if not old_name or not new_name:
             raise ValueError("旧名称和新名称都不能为空")
         if old_name == new_name:
-            return {"success": True, "updated_count": 0, "info": "新旧名称相同，无需更新"}
+            return {
-        
+                "success": True,
+                "updated_count": 0,
+                "info": "新旧名称相同，无需更新",
+            }
+
         # 检查新名称是否已存在（理论上前端会校验，后端再保险一次）
         exists = await DbPluginInfo.filter(menu_type=new_name).exists()
         if exists:
-             raise ValueError(f"新的菜单类型名称 '{new_name}' 已被其他插件使用")
+            raise ValueError(f"新的菜单类型名称 '{new_name}' 已被其他插件使用")
 
         try:
             # 使用 filter().update() 进行批量更新
-            updated_count = await DbPluginInfo.filter(menu_type=old_name).update(menu_type=new_name)
+            updated_count = await DbPluginInfo.filter(menu_type=old_name).update(
+                menu_type=new_name
+            )
             return {"success": True, "updated_count": updated_count}
         except Exception as e:
             # 可以添加更详细的日志记录
-            raise RuntimeError(f"数据库更新菜单类型失败: {str(e)}")
+            raise RuntimeError(f"数据库更新菜单类型失败: {e!s}")
 
     @classmethod
     async def get_plugin_detail(cls, module: str) -> PluginDetail:

zhenxun/builtin_plugins/web_ui/api/tabs/plugin_manage/model.py

@@ -113,11 +113,15 @@ class BatchUpdatePluginItem(BaseModel):
     module: str = Field(..., description="插件模块名")
     default_status: bool | None = Field(None, description="默认状态(开关)")
     menu_type: str | None = Field(None, description="菜单类型")
-    block_type: BlockType | None = Field(None, description="插件禁用状态 (None: 启用, ALL: 禁用)")
+    block_type: BlockType | None = Field(
+        None, description="插件禁用状态 (None: 启用, ALL: 禁用)"
+    )
 
 
 class BatchUpdatePlugins(BaseModel):
-    updates: list[BatchUpdatePluginItem] = Field(..., description="要批量更新的插件列表")
+    updates: list[BatchUpdatePluginItem] = Field(
+        ..., description="要批量更新的插件列表"
+    )
 
 
 class PluginDetail(PluginInfo):
@@ -136,3 +140,18 @@ class RenameMenuTypePayload(BaseModel):
 class PluginIr(BaseModel):
     id: int
     """插件id"""
+
+
+class BatchUpdateResult(BaseModel):
+    """
+    批量更新插件结果
+    """
+
+    success: bool = Field(..., description="是否全部成功")
+    """是否全部成功"""
+    updated_count: int = Field(..., description="更新成功的数量")
+    """更新成功的数量"""
+    errors: list[dict[str, str]] = Field(
+        default_factory=list, description="错误信息列表"
+    )
+    """错误信息列表"""