Optimize code and comments with cursor

.gitignore

@@ -1,5 +1,6 @@
 .idea/
 .vscode/
+.DS_Store
 venv/
 .venv/
 .python-version

backend/app/admin/crud/crud_user.py

@@ -176,10 +176,10 @@ async def get_list(self, dept: int = None, username: str = None, phone: str = No
         """
         获取用户列表
 
-        :param dept:
-        :param username:
-        :param phone:
-        :param status:
+        :param dept: 部门 ID（可选）
+        :param username: 用户名（可选）
+        :param phone: 电话号码（可选）
+        :param status: 用户状态（可选）
         :return:
         """
         stmt = (
@@ -191,17 +191,22 @@ async def get_list(self, dept: int = None, username: str = None, phone: str = No
             )
             .order_by(desc(self.model.join_time))
         )
-        where_list = []
+
+        # 构建过滤条件
+        filters = []
         if dept:
-            where_list.append(self.model.dept_id == dept)
+            filters.append(self.model.dept_id == dept)
         if username:
-            where_list.append(self.model.username.like(f'%{username}%'))
+            filters.append(self.model.username.like(f'%{username}%'))
         if phone:
-            where_list.append(self.model.phone.like(f'%{phone}%'))
+            filters.append(self.model.phone.like(f'%{phone}%'))
         if status is not None:
-            where_list.append(self.model.status == status)
-        if where_list:
-            stmt = stmt.where(and_(*where_list))
+            filters.append(self.model.status == status)
+
+        # 应用过滤条件
+        if filters:
+            stmt = stmt.where(and_(*filters))
+
         return stmt
 
     async def get_super(self, db: AsyncSession, user_id: int) -> bool:

backend/plugin/tools.py

@@ -6,6 +6,8 @@
 import sys
 import warnings
 
+from typing import Any
+
 import rtoml
 
 from fastapi import APIRouter
@@ -17,156 +19,196 @@
 
 
 class PluginInjectError(Exception):
+    """插件注入错误"""
     pass
 
 
 def get_plugins() -> list[str]:
-    """获取插件"""
+    """获取插件列表"""
     plugin_packages = []
-
+
+    # 遍历插件目录
     for item in os.listdir(PLUGIN_DIR):
         item_path = os.path.join(PLUGIN_DIR, item)
-
-        if os.path.isdir(item_path):
-            if '__init__.py' in os.listdir(item_path):
-                plugin_packages.append(item)
-
+        
+        # 检查是否为目录且包含 __init__.py 文件
+        if os.path.isdir(item_path) and '__init__.py' in os.listdir(item_path):
+            plugin_packages.append(item)
+    
     return plugin_packages
 
 
-def get_plugin_models() -> list:
+def get_plugin_models() -> list[type]:
     """获取插件所有模型类"""
     classes = []
+
+    # 获取所有插件
     plugins = get_plugins()
+
+    # 遍历插件列表
     for plugin in plugins:
+        # 导入插件的模型模块
         module_path = f'backend.plugin.{plugin}.model'
         module = import_module_cached(module_path)
+
+        # 获取模块中的所有类
         for name, obj in inspect.getmembers(module):
             if inspect.isclass(obj):
                 classes.append(obj)
+
     return classes
 
 
-def plugin_router_inject() -> None:
+def _load_plugin_config(plugin: str) -> dict[str, Any]:
+    """
+    加载插件配置
+
+    :param plugin: 插件名称
+    :return:
     """
-    插件路由注入
+    toml_path = os.path.join(PLUGIN_DIR, plugin, 'plugin.toml')
+    if not os.path.exists(toml_path):
+        raise PluginInjectError(f'插件 {plugin} 缺少 plugin.toml 配置文件，请检查插件是否合法')
+
+    with open(toml_path, 'r', encoding='utf-8') as f:
+        return rtoml.load(f)
+
 
+def _inject_extra_router(plugin: str, data: dict[str, Any]) -> None:
+    """
+    扩展级插件路由注入
+
+    :param plugin: 插件名称
+    :param data: 插件配置数据
     :return:
     """
-    plugins = get_plugins()
-    for plugin in plugins:
-        toml_path = os.path.join(PLUGIN_DIR, plugin, 'plugin.toml')
-        if not os.path.exists(toml_path):
-            raise PluginInjectError(f'插件 {plugin} 缺少 plugin.toml 配置文件，请检查插件是否合法')
-
-        # 获取 plugin.toml 配置
-        with open(toml_path, 'r', encoding='utf-8') as f:
-            data = rtoml.load(f)
-        api = data.get('api', {})
-
-        # 非独立 app
-        if api:
-            app_include = data.get('app', {}).get('include', '')
-            if not app_include:
-                raise PluginInjectError(f'非独立 app 插件 {plugin} 配置文件存在错误，请检查')
-
-            # 插件中 API 路由文件的路径
-            plugin_api_path = os.path.join(PLUGIN_DIR, plugin, 'api')
-            if not os.path.exists(plugin_api_path):
-                raise PluginInjectError(f'插件 {plugin} 缺少 api 目录，请检查插件文件是否完整')
-
-            # 将插件路由注入到对应模块的路由中
-            for root, _, api_files in os.walk(plugin_api_path):
-                for file in api_files:
-                    if file.endswith('.py') and file != '__init__.py':
-                        # 解析插件路由配置
-                        prefix = data.get('api', {}).get(f'{file[:-3]}', {}).get('prefix', '')
-                        tags = data.get('api', {}).get(f'{file[:-3]}', {}).get('tags', [])
-
-                        # 获取插件路由模块
-                        file_path = os.path.join(root, file)
-                        path_to_module_str = os.path.relpath(file_path, PLUGIN_DIR).replace(os.sep, '.')[:-3]
-                        module_path = f'backend.plugin.{path_to_module_str}'
-                        try:
-                            module = import_module_cached(module_path)
-                        except PluginInjectError as e:
-                            raise PluginInjectError(f'导入非独立 app 插件 {plugin} 模块 {module_path} 失败：{e}') from e
-                        plugin_router = getattr(module, 'router', None)
-                        if not plugin_router:
-                            warnings.warn(
-                                f'非独立 app 插件 {plugin} 模块 {module_path} 中没有有效的 router，'
-                                '请检查插件文件是否完整',
-                                FutureWarning,
-                            )
-                            continue
-
-                        # 获取源程序路由模块
-                        relative_path = os.path.relpath(root, plugin_api_path)
-                        target_module_path = f'backend.app.{app_include}.api.{relative_path.replace(os.sep, ".")}'
-                        try:
-                            target_module = import_module_cached(target_module_path)
-                        except PluginInjectError as e:
-                            raise PluginInjectError(f'导入源程序模块 {target_module_path} 失败：{e}') from e
-                        target_router = getattr(target_module, 'router', None)
-                        if not target_router or not isinstance(target_router, APIRouter):
-                            raise PluginInjectError(
-                                f'非独立 app 插件 {plugin} 模块 {module_path} 中没有有效的 router，'
-                                '请检查插件文件是否完整'
-                            )
-
-                        # 将插件路由注入到目标 router 中
-                        target_router.include_router(
-                            router=plugin_router,
-                            prefix=prefix,
-                            tags=[tags] if tags else [],
-                        )
-        # 独立 app
-        else:
-            # 将插件中的路由直接注入到总路由中
-            module_path = f'backend.plugin.{plugin}.api.router'
+    app_include = data.get('app', {}).get('include', '')
+    if not app_include:
+        raise PluginInjectError(f'扩展级插件 {plugin} 配置文件存在错误，请检查')
+
+    plugin_api_path = os.path.join(PLUGIN_DIR, plugin, 'api')
+    if not os.path.exists(plugin_api_path):
+        raise PluginInjectError(f'插件 {plugin} 缺少 api 目录，请检查插件文件是否完整')
+
+    for root, _, api_files in os.walk(plugin_api_path):
+        for file in api_files:
+            if not (file.endswith('.py') and file != '__init__.py'):
+                continue
+
+            file_config = data.get('api', {}).get(f'{file[:-3]}', {})
+            prefix = file_config.get('prefix', '')
+            tags = file_config.get('tags', [])
+
+            file_path = os.path.join(root, file)
+            path_to_module_str = os.path.relpath(file_path, PLUGIN_DIR).replace(os.sep, '.')[:-3]
+            module_path = f'backend.plugin.{path_to_module_str}'
+
             try:
                 module = import_module_cached(module_path)
-            except PluginInjectError as e:
-                raise PluginInjectError(f'导入独立 app 插件 {plugin} 模块 {module_path} 失败：{e}') from e
-            routers = data.get('app', {}).get('router', [])
-            if not routers or not isinstance(routers, list):
-                raise PluginInjectError(f'独立 app 插件 {plugin} 配置文件存在错误，请检查')
-            for router in routers:
-                plugin_router = getattr(module, router, None)
-                if not plugin_router or not isinstance(plugin_router, APIRouter):
-                    raise PluginInjectError(
-                        f'独立 app 插件 {plugin} 模块 {module_path} 中没有有效的 router，请检查插件文件是否完整'
+                plugin_router = getattr(module, 'router', None)
+                if not plugin_router:
+                    warnings.warn(
+                        f'扩展级插件 {plugin} 模块 {module_path} 中没有有效的 router，'
+                        '请检查插件文件是否完整',
+                        FutureWarning,
                     )
-                target_module_path = 'backend.app.router'
+                    continue
+
+                relative_path = os.path.relpath(root, plugin_api_path)
+                target_module_path = f'backend.app.{app_include}.api.{relative_path.replace(os.sep, ".")}'
                 target_module = import_module_cached(target_module_path)
-                target_router = getattr(target_module, 'router')
+                target_router = getattr(target_module, 'router', None)
+
+                if not target_router or not isinstance(target_router, APIRouter):
+                    raise PluginInjectError(
+                        f'扩展级插件 {plugin} 模块 {module_path} 中没有有效的 router，'
+                        '请检查插件文件是否完整'
+                    )
+
+                target_router.include_router(
+                    router=plugin_router,
+                    prefix=prefix,
+                    tags=[tags] if tags else [],
+                )
+            except Exception as e:
+                raise PluginInjectError(f'注入扩展级插件 {plugin} 路由失败：{str(e)}') from e
 
-                # 将插件路由注入到目标 router 中
-                target_router.include_router(plugin_router)
+
+def _inject_app_router(plugin: str, data: dict[str, Any]) -> None:
+    """
+    应用级插件路由注入
+
+    :param plugin: 插件名称
+    :param data: 插件配置数据
+    :return:
+    """
+    module_path = f'backend.plugin.{plugin}.api.router'
+    try:
+        module = import_module_cached(module_path)
+        routers = data.get('app', {}).get('router', [])
+        if not routers or not isinstance(routers, list):
+            raise PluginInjectError(f'应用级插件 {plugin} 配置文件存在错误，请检查')
+
+        target_module = import_module_cached('backend.app.router')
+        target_router = getattr(target_module, 'router')
+
+        for router in routers:
+            plugin_router = getattr(module, router, None)
+            if not plugin_router or not isinstance(plugin_router, APIRouter):
+                raise PluginInjectError(
+                    f'应用级插件 {plugin} 模块 {module_path} 中没有有效的 router，请检查插件文件是否完整'
+                )
+            target_router.include_router(plugin_router)
+    except Exception as e:
+        raise PluginInjectError(f'注入应用级插件 {plugin} 路由失败：{str(e)}') from e
+
+
+def plugin_router_inject() -> None:
+    """插件路由注入"""
+    for plugin in get_plugins():
+        try:
+            data = _load_plugin_config(plugin)
+            # 基于插件 plugin.toml 配置文件，判断插件类型
+            if data.get('api'):
+                _inject_extra_router(plugin, data)
+            else:
+                _inject_app_router(plugin, data)
+        except Exception as e:
+            raise PluginInjectError(f'插件 {plugin} 路由注入失败：{str(e)}') from e
+
+
+def _install_plugin_requirements(plugin: str, requirements_file: str) -> None:
+    """
+    安装单个插件的依赖
+
+    :param plugin: 插件名称
+    :param requirements_file: 依赖文件路径
+    :return:
+    """
+    try:
+        ensurepip_install = [sys.executable, '-m', 'ensurepip', '--upgrade']
+        pip_install = [sys.executable, '-m', 'pip', 'install', '-r', requirements_file]
+        if settings.PLUGIN_PIP_CHINA:
+            pip_install.extend(['-i', settings.PLUGIN_PIP_INDEX_URL])
+        subprocess.check_call(ensurepip_install)
+        subprocess.check_call(pip_install)
+    except subprocess.CalledProcessError as e:
+        raise PluginInjectError(f'插件 {plugin} 依赖安装失败：{e.stderr}') from e
 
 
 def install_requirements() -> None:
     """安装插件依赖"""
-    plugins = get_plugins()
-    for plugin in plugins:
+    for plugin in get_plugins():
         requirements_file = os.path.join(PLUGIN_DIR, plugin, 'requirements.txt')
-        if not os.path.exists(requirements_file):
-            continue
-        else:
-            try:
-                ensurepip_install = [sys.executable, '-m', 'ensurepip', '--upgrade']
-                pip_install = [sys.executable, '-m', 'pip', 'install', '-r', requirements_file]
-                if settings.PLUGIN_PIP_CHINA:
-                    pip_install.extend(['-i', settings.PLUGIN_PIP_INDEX_URL])
-                subprocess.check_call(ensurepip_install)
-                subprocess.check_call(pip_install)
-            except subprocess.CalledProcessError as e:
-                raise PluginInjectError(f'插件 {plugin} 依赖安装失败：{e.stderr}') from e
+        if os.path.exists(requirements_file):
+            _install_plugin_requirements(plugin, requirements_file)
 
 
 async def install_requirements_async() -> None:
     """
-    异步安装插件依赖（由于 Windows 平台限制，无法实现完美的全异步方案），详情：
+    异步安装插件依赖
+
+    由于 Windows 平台限制，无法实现完美的全异步方案，详情：
     https://stackoverflow.com/questions/44633458/why-am-i-getting-notimplementederror-with-async-and-await-on-windows
     """
     await run_in_threadpool(install_requirements)