chore: remove obsolete role management files

- Deleted unused role management schemas, CRUD operations, and service files to streamline the codebase.
- Removed related frontend components and API interactions to enhance maintainability and reduce complexity.

frontend/修改的文件/auth的schema.py

@@ -1,84 +0,0 @@
-from datetime import datetime
-
-from pydantic import BaseModel, ConfigDict, Field, model_validator
-from sqlalchemy.ext.asyncio import AsyncSession
-
-from app.api.v1.module_system.user.model import UserModel
-
-
-class AuthSchema(BaseModel):
-    """权限认证模型"""
-
-    model_config = ConfigDict(arbitrary_types_allowed=True)
-
-    user: UserModel | None = Field(default=None, description="用户信息")
-    check_data_scope: bool = Field(default=True, description="是否检查数据权限")
-    db: AsyncSession = Field(description="数据库会话")
-    current_permission: str | None = Field(default=None, description="当前请求的权限标识")
-
-
-class JWTPayloadSchema(BaseModel):
-    """JWT载荷模型"""
-
-    sub: str = Field(..., description="用户登录信息")
-    is_refresh: bool = Field(default=False, description="是否刷新token")
-    exp: datetime | int = Field(..., description="过期时间")
-
-    @model_validator(mode="after")
-    def validate_fields(self):
-        if not self.sub or len(self.sub.strip()) == 0:
-            raise ValueError("会话编号不能为空")
-        return self
-
-
-class JWTOutSchema(BaseModel):
-    """JWT响应模型"""
-
-    model_config = ConfigDict(from_attributes=True)
-
-    access_token: str = Field(..., min_length=1, description="访问token")
-    refresh_token: str = Field(..., min_length=1, description="刷新token")
-    token_type: str = Field(default="Bearer", description="token类型")
-    expires_in: int = Field(..., gt=0, description="过期时间(秒)")
-
-
-class RefreshTokenPayloadSchema(BaseModel):
-    """刷新Token载荷模型"""
-
-    refresh_token: str = Field(..., min_length=1, description="刷新token")
-
-
-class LogoutPayloadSchema(BaseModel):
-    """退出登录载荷模型"""
-
-    token: str = Field(..., min_length=1, description="token")
-
-
-class CaptchaOutSchema(BaseModel):
-    """验证码响应模型"""
-
-    model_config = ConfigDict(from_attributes=True)
-
-    enable: bool = Field(default=True, description="是否启用验证码")
-    key: str = Field(..., min_length=1, description="验证码唯一标识")
-    img_base: str = Field(..., min_length=1, description="Base64编码的验证码图片")
-
-
-class AutoLoginUserSchema(BaseModel):
-    """免登录用户信息模型"""
-
-    model_config = ConfigDict(from_attributes=True)
-
-    id: int = Field(..., description="用户ID")
-    username: str = Field(..., description="用户名")
-    name: str = Field(..., description="用户姓名")
-    avatar: str | None = Field(default=None, description="头像")
-
-
-class AutoLoginTokenSchema(BaseModel):
-    """免登录Token响应模型"""
-
-    model_config = ConfigDict(from_attributes=True)
-
-    token: str = Field(..., description="免登录Token")
-    user: AutoLoginUserSchema = Field(..., description="用户信息")

frontend/修改的文件/crud.py

@@ -1,199 +0,0 @@
-from collections.abc import Sequence
-
-from sqlalchemy import delete, select, update
-
-from app.api.v1.module_system.auth.schema import AuthSchema
-from app.api.v1.module_system.dept.crud import DeptCRUD
-from app.api.v1.module_system.menu.crud import MenuCRUD
-from app.core.base_crud import CRUDBase
-
-from .model import RoleMenuDeptsModel, RoleMenusModel, RoleModel
-from .schema import MenuDataScopeItem, RoleCreateSchema, RoleUpdateSchema
-
-
-class RoleCRUD(CRUDBase[RoleModel, RoleCreateSchema, RoleUpdateSchema]):
-    """角色模块数据层"""
-
-    def __init__(self, auth: AuthSchema) -> None:
-        """
-        初始化角色模块数据层
-
-        参数:
-        - auth (AuthSchema): 认证信息模型
-        """
-        self.auth = auth
-        super().__init__(model=RoleModel, auth=auth)
-
-    async def get_by_id_crud(self, id: int, preload: list | None = None) -> RoleModel | None:
-        """
-        根据id获取角色信息
-
-        参数:
-        - id (int): 角色ID
-        - preload (list | None): 预加载选项
-
-        返回:
-        - RoleModel | None: 角色模型对象
-        """
-        return await self.get(id=id, preload=preload)
-
-    async def get_list_crud(
-        self,
-        search: dict | None = None,
-        order_by: list | None = None,
-        preload: list | None = None,
-    ) -> Sequence[RoleModel]:
-        """
-        获取角色列表
-
-        参数:
-        - search (dict | None): 查询参数
-        - order_by (list | None): 排序参数
-        - preload (list | None): 预加载选项
-
-        返回:
-        - Sequence[RoleModel]: 角色模型对象列表
-        """
-        return await self.list(search=search, order_by=order_by, preload=preload)
-
-    async def set_role_menus_crud(self, role_ids: list[int], menu_ids: list[int]) -> None:
-        """
-        设置角色的菜单权限
-
-        参数:
-        - role_ids (List[int]): 角色ID列表
-        - menu_ids (List[int]): 菜单ID列表
-
-        返回:
-        - None
-        """
-        roles = await self.list(search={"id": ("in", role_ids)})
-        menus = await MenuCRUD(self.auth).get_list_crud(search={"id": ("in", menu_ids)})
-
-        for obj in roles:
-            relationship = obj.menus
-            relationship.clear()
-            relationship.extend(menus)
-        await self.auth.db.flush()
-
-    async def set_role_data_scope_crud(self, role_ids: list[int], data_scope: int) -> None:
-        """
-        设置角色的数据范围
-
-        参数:
-        - role_ids (list[int]): 角色ID列表
-        - data_scope (int): 数据范围
-
-        返回:
-        - None
-        """
-        await self.set(ids=role_ids, data_scope=data_scope)
-
-    async def set_role_depts_crud(self, role_ids: list[int], dept_ids: list[int]) -> None:
-        """
-        设置角色的部门权限
-
-        参数:
-        - role_ids (list[int]): 角色ID列表
-        - dept_ids (list[int]): 部门ID列表
-
-        返回:
-        - None
-        """
-        roles = await self.list(search={"id": ("in", role_ids)})
-        depts = await DeptCRUD(self.auth).get_list_crud(search={"id": ("in", dept_ids)})
-
-        for obj in roles:
-            relationship = obj.depts
-            relationship.clear()
-            relationship.extend(depts)
-        await self.auth.db.flush()
-
-    async def set_available_crud(self, ids: list[int], status: str) -> None:
-        """
-        设置角色的可用状态
-
-        参数:
-        - ids (list[int]): 角色ID列表
-        - status (str): 可用状态
-
-        返回:
-        - None
-        """
-        await self.set(ids=ids, status=status)
-
-    async def set_role_menu_data_scopes_crud(
-        self, role_id: int, menu_data_scopes: list[MenuDataScopeItem]
-    ) -> None:
-        """
-        设置角色的菜单级数据权限
-
-        参数:
-        - role_id (int): 角色ID
-        - menu_data_scopes (list[MenuDataScopeItem]): 菜单级数据权限配置列表
-
-        返回:
-        - None
-        """
-        for item in menu_data_scopes:
-            # 更新 sys_role_menus 中的 data_scope
-            await self.auth.db.execute(
-                update(RoleMenusModel)
-                .where(RoleMenusModel.role_id == role_id, RoleMenusModel.menu_id == item.menu_id)
-                .values(data_scope=item.data_scope)
-            )
-
-            # 先清除该（角色, 菜单）对的旧自定义部门记录
-            await self.auth.db.execute(
-                delete(RoleMenuDeptsModel).where(
-                    RoleMenuDeptsModel.role_id == role_id,
-                    RoleMenuDeptsModel.menu_id == item.menu_id,
-                )
-            )
-
-            # 如果 data_scope=5，写入新的自定义部门记录
-            if item.data_scope == 5 and item.dept_ids:
-                for dept_id in item.dept_ids:
-                    self.auth.db.add(
-                        RoleMenuDeptsModel(role_id=role_id, menu_id=item.menu_id, dept_id=dept_id)
-                    )
-
-        await self.auth.db.flush()
-
-    async def get_role_menu_data_scopes_crud(self, role_id: int) -> list[dict]:
-        """
-        获取角色的菜单级数据权限配置
-
-        参数:
-        - role_id (int): 角色ID
-
-        返回:
-        - list[dict]: 菜单级数据权限配置列表
-        """
-        # 查询所有设置了菜单级 data_scope 的记录
-        result = await self.auth.db.execute(
-            select(RoleMenusModel.menu_id, RoleMenusModel.data_scope).where(
-                RoleMenusModel.role_id == role_id,
-                RoleMenusModel.data_scope.isnot(None),
-            )
-        )
-        rows = result.all()
-
-        items = []
-        for row in rows:
-            dept_ids: list[int] = []
-            if row.data_scope == 5:
-                # 查询自定义部门
-                dept_result = await self.auth.db.execute(
-                    select(RoleMenuDeptsModel.dept_id).where(
-                        RoleMenuDeptsModel.role_id == role_id,
-                        RoleMenuDeptsModel.menu_id == row.menu_id,
-                    )
-                )
-                dept_ids = [r.dept_id for r in dept_result.all()]
-
-            items.append(
-                {"menu_id": row.menu_id, "data_scope": row.data_scope, "dept_ids": dept_ids}
-            )
-
-        return items

frontend/修改的文件/dependencies.py

@@ -1,301 +0,0 @@
-import json
-from collections.abc import AsyncGenerator
-
-from fastapi import Depends, Query, Request
-from redis.asyncio.client import Redis
-from sqlalchemy.ext.asyncio import AsyncSession
-from sqlalchemy.orm import selectinload
-
-from app.api.v1.module_system.auth.schema import AuthSchema
-from app.api.v1.module_system.user.crud import UserCRUD
-from app.api.v1.module_system.user.model import UserModel
-from app.common.enums import RedisInitKeyConfig
-from app.config.setting import settings
-from app.core.database import async_db_session
-from app.core.exceptions import CustomException
-from app.core.logger import log
-from app.core.redis_crud import RedisCURD
-from app.core.security import OAuth2Schema, decode_access_token
-
-
-async def db_getter() -> AsyncGenerator[AsyncSession, None]:
-    """获取数据库会话连接
-
-    返回:
-    - AsyncSession: 数据库会话连接
-    """
-    async with async_db_session() as session:
-        async with session.begin():
-            yield session
-
-
-async def redis_getter(request: Request) -> Redis:
-    """获取Redis连接
-
-    参数:
-    - request (Request): 请求对象
-
-    返回:
-    - Redis: Redis连接
-    """
-    return request.app.state.redis
-
-
-async def get_current_user(
-    request: Request,
-    db: AsyncSession = Depends(db_getter),
-    redis: Redis = Depends(redis_getter),
-    token: str = Depends(OAuth2Schema),
-) -> AuthSchema:
-    """获取当前用户
-
-    参数:
-    - request (Request): 请求对象
-    - db (AsyncSession): 数据库会话
-    - redis (Redis): Redis连接
-    - token (str): 访问令牌
-
-    返回:
-    - AuthSchema: 认证信息模型
-    """
-    if not token:
-        raise CustomException(msg="认证已失效", code=10401, status_code=401)
-
-    # 处理Bearer token
-    if token.startswith("Bearer"):
-        token = token.split(" ")[1]
-
-    payload = decode_access_token(token)
-    if not payload or not hasattr(payload, "is_refresh") or payload.is_refresh:
-        raise CustomException(msg="非法凭证", code=10401, status_code=401)
-
-    online_user_info = payload.sub
-    # 从Redis中获取用户信息
-    user_info = json.loads(online_user_info)  # 确保是字典类型
-
-    session_id = user_info.get("session_id")
-    if not session_id:
-        raise CustomException(msg="认证已失效", code=10401, status_code=401)
-
-    # 检查用户是否在线
-    online_ok = await RedisCURD(redis).exists(
-        key=f"{RedisInitKeyConfig.ACCESS_TOKEN.key}:{session_id}"
-    )
-    if not online_ok:
-        raise CustomException(msg="认证已失效", code=10401, status_code=401)
-
-    # 如果启用了滑动过期，自动续期token
-    if settings.TOKEN_SLIDING_EXPIRE:
-        await RedisCURD(redis).expire(
-            key=f"{RedisInitKeyConfig.ACCESS_TOKEN.key}:{session_id}",
-            expire=settings.ACCESS_TOKEN_EXPIRE_MINUTES,
-        )
-        await RedisCURD(redis).expire(
-            key=f"{RedisInitKeyConfig.REFRESH_TOKEN.key}:{session_id}",
-            expire=settings.REFRESH_TOKEN_EXPIRE_MINUTES,
-        )
-
-    # 关闭数据权限过滤，避免当前用户查询被拦截
-    auth = AuthSchema(db=db, check_data_scope=False)
-    username = user_info.get("user_name")
-    if not username:
-        raise CustomException(msg="认证已失效", code=10401, status_code=401)
-    # 获取用户信息，使用深层预加载确保RoleModel.creator被正确加载
-    user = await UserCRUD(auth).get_by_username_crud(
-        username=username,
-        preload=[
-            "dept",
-            selectinload(UserModel.roles),
-            "positions",
-            "created_by",
-        ],
-    )
-    if not user:
-        raise CustomException(msg="用户不存在", code=10401, status_code=401)
-    if user.status == "1":
-        raise CustomException(msg="用户已被停用", code=10401, status_code=401)
-
-    # 设置请求上下文
-    request.scope["user_id"] = user.id
-    request.scope["user_username"] = user.username
-
-    # 过滤可用的角色和职位
-    if hasattr(user, "roles"):
-        user.roles = [role for role in user.roles if role and role.status]
-    if hasattr(user, "positions"):
-        user.positions = [pos for pos in user.positions if pos and pos.status]
-
-    auth.user = user
-    return auth
-
-
-async def get_current_user_ws(
-    token: str = Query(..., description="认证token"),
-    db: AsyncSession = Depends(db_getter),
-    redis: Redis = Depends(redis_getter),
-) -> AuthSchema:
-    """获取当前用户（WebSocket专用，从查询参数获取token）
-
-    参数:
-    - token (str): 认证token
-    - db (AsyncSession): 数据库会话
-    - redis (Redis): Redis连接
-
-    返回:
-    - AuthSchema: 认证信息模型
-    """
-    return await _verify_token(token, db, redis)
-
-
-async def _verify_token(
-    token: str,
-    db: AsyncSession,
-    redis: Redis,
-) -> AuthSchema:
-    """验证token并返回用户信息
-
-    参数:
-    - token (str): 认证token
-    - db (AsyncSession): 数据库会话
-    - redis (Redis): Redis连接
-
-    返回:
-    - AuthSchema: 认证信息模型
-    """
-    if not token:
-        raise CustomException(msg="认证已失效", code=10401, status_code=401)
-
-    # 处理Bearer token（如果通过查询参数传递时包含Bearer前缀）
-    if token.startswith("Bearer"):
-        token = token.split(" ")[1]
-
-    payload = decode_access_token(token)
-    if not payload or not hasattr(payload, "is_refresh") or payload.is_refresh:
-        raise CustomException(msg="非法凭证", code=10401, status_code=401)
-
-    online_user_info = payload.sub
-    # 从Redis中获取用户信息
-    user_info = json.loads(online_user_info)  # 确保是字典类型
-
-    session_id = user_info.get("session_id")
-    if not session_id:
-        raise CustomException(msg="认证已失效", code=10401, status_code=401)
-
-    # 检查用户是否在线
-    online_ok = await RedisCURD(redis).exists(
-        key=f"{RedisInitKeyConfig.ACCESS_TOKEN.key}:{session_id}"
-    )
-    if not online_ok:
-        raise CustomException(msg="认证已失效", code=10401, status_code=401)
-
-    # 如果启用了滑动过期，自动续期token
-    if settings.TOKEN_SLIDING_EXPIRE:
-        await RedisCURD(redis).expire(
-            key=f"{RedisInitKeyConfig.ACCESS_TOKEN.key}:{session_id}",
-            expire=settings.ACCESS_TOKEN_EXPIRE_MINUTES,
-        )
-        await RedisCURD(redis).expire(
-            key=f"{RedisInitKeyConfig.REFRESH_TOKEN.key}:{session_id}",
-            expire=settings.REFRESH_TOKEN_EXPIRE_MINUTES,
-        )
-
-    # 关闭数据权限过滤，避免当前用户查询被拦截
-    auth = AuthSchema(db=db, check_data_scope=False)
-    username = user_info.get("user_name")
-    if not username:
-        raise CustomException(msg="认证已失效", code=10401, status_code=401)
-    # 获取用户信息，使用深层预加载确保RoleModel.creator被正确加载
-    user = await UserCRUD(auth).get_by_username_crud(
-        username=username,
-        preload=[
-            "dept",
-            selectinload(UserModel.roles),
-            "positions",
-            "created_by",
-        ],
-    )
-    if not user:
-        raise CustomException(msg="用户不存在", code=10401, status_code=401)
-    if user.status == "1":
-        raise CustomException(msg="用户已被停用", code=10401, status_code=401)
-
-    # 设置请求上下文
-    # request.scope["user_id"] = user.id
-    # request.scope["user_username"] = user.username
-
-    # 过滤可用的角色和职位
-    if hasattr(user, "roles"):
-        user.roles = [role for role in user.roles if role and role.status]
-    if hasattr(user, "positions"):
-        user.positions = [pos for pos in user.positions if pos and pos.status]
-
-    auth.user = user
-    return auth
-
-
-class AuthPermission:
-    """权限验证类"""
-
-    def __init__(
-        self,
-        permissions: list[str] | None = None,
-        check_data_scope: bool = True,
-    ) -> None:
-        """
-        初始化权限验证
-
-        参数:
-        - permissions (list[str] | None): 权限标识列表。
-        - check_data_scope (bool): 是否启用严格模式校验。
-        """
-        self.permissions = permissions or []
-        self.check_data_scope = check_data_scope
-
-    async def __call__(self, auth: AuthSchema = Depends(get_current_user)) -> AuthSchema:
-        """
-        调用权限验证
-
-        参数:
-        - auth (AuthSchema): 认证信息对象。
-
-        返回:
-        - AuthSchema: 认证信息对象。
-        """
-        auth.check_data_scope = self.check_data_scope
-
-        # 超级管理员直接通过
-        if auth.user and auth.user.is_superuser:
-            return auth
-
-        # 无需验证权限
-        if not self.permissions:
-            return auth
-
-        # 超级管理员权限标识
-        if "*" in self.permissions or "*:*:*" in self.permissions:
-            return auth
-
-        # 检查用户是否有角色
-        if not auth.user or not auth.user.roles:
-            raise CustomException(msg="无权限操作", code=10403, status_code=403)
-
-        # 获取用户权限集合
-        user_permissions = {
-            menu.permission
-            for role in auth.user.roles
-            for menu in role.menus
-            if role.status == "0" and menu.permission and menu.status == "0"
-        }
-
-        # 权限验证 - 满足任一权限即可
-        if not any(perm in user_permissions for perm in self.permissions):
-            log.error(f"用户缺少任何所需的权限: {self.permissions}")
-            raise CustomException(msg="无权限操作", code=10403, status_code=403)
-
-        # 记录匹配到的权限标识，供数据权限引擎使用
-        for perm in self.permissions:
-            if perm in user_permissions:
-                auth.current_permission = perm
-                break
-
-        return auth

frontend/修改的文件/model.py

@@ -1,137 +0,0 @@
-from typing import TYPE_CHECKING
-
-from sqlalchemy import ForeignKey, Integer, String
-from sqlalchemy.orm import Mapped, mapped_column, relationship
-
-from app.common.enums import PermissionFilterStrategy
-from app.core.base_model import MappedBase, ModelMixin
-
-if TYPE_CHECKING:
-    from app.api.v1.module_system.dept.model import DeptModel
-    from app.api.v1.module_system.menu.model import MenuModel
-    from app.api.v1.module_system.user.model import UserModel
-
-
-class RoleMenusModel(MappedBase):
-    """
-    角色菜单关联表
-
-    定义角色与菜单的多对多关系，用于权限控制
-    支持菜单级数据权限覆盖：data_scope 为 NULL 时继承角色级 data_scope
-    """
-
-    __tablename__: str = "sys_role_menus"
-    __table_args__: dict[str, str] = {"comment": "角色菜单关联表"}
-
-    role_id: Mapped[int] = mapped_column(
-        Integer,
-        ForeignKey("sys_role.id", ondelete="CASCADE", onupdate="CASCADE"),
-        primary_key=True,
-        comment="角色ID",
-    )
-    menu_id: Mapped[int] = mapped_column(
-        Integer,
-        ForeignKey("sys_menu.id", ondelete="CASCADE", onupdate="CASCADE"),
-        primary_key=True,
-        comment="菜单ID",
-    )
-    data_scope: Mapped[int | None] = mapped_column(
-        Integer,
-        nullable=True,
-        default=None,
-        comment="菜单级数据权限范围(NULL则继承角色级, 1:仅本人 2:本部门 3:本部门及以下 4:全部 5:自定义)",
-    )
-
-
-class RoleDeptsModel(MappedBase):
-    """
-    角色部门关联表
-
-    定义角色与部门的多对多关系，用于数据权限控制
-    仅当角色的data_scope=5(自定义数据权限)时使用此表
-    """
-
-    __tablename__: str = "sys_role_depts"
-    __table_args__: dict[str, str] = {"comment": "角色部门关联表"}
-
-    role_id: Mapped[int] = mapped_column(
-        Integer,
-        ForeignKey("sys_role.id", ondelete="CASCADE", onupdate="CASCADE"),
-        primary_key=True,
-        comment="角色ID",
-    )
-    dept_id: Mapped[int] = mapped_column(
-        Integer,
-        ForeignKey("sys_dept.id", ondelete="CASCADE", onupdate="CASCADE"),
-        primary_key=True,
-        comment="部门ID",
-    )
-
-
-class RoleMenuDeptsModel(MappedBase):
-    """
-    角色菜单部门关联表
-
-    定义角色菜单级别的自定义部门数据权限
-    仅当 sys_role_menus.data_scope=5(自定义数据权限)时使用此表
-    """
-
-    __tablename__: str = "sys_role_menu_depts"
-    __table_args__: dict[str, str] = {"comment": "角色菜单部门关联表(菜单级自定义数据权限)"}
-
-    role_id: Mapped[int] = mapped_column(
-        Integer,
-        ForeignKey("sys_role.id", ondelete="CASCADE", onupdate="CASCADE"),
-        primary_key=True,
-        comment="角色ID",
-    )
-    menu_id: Mapped[int] = mapped_column(
-        Integer,
-        ForeignKey("sys_menu.id", ondelete="CASCADE", onupdate="CASCADE"),
-        primary_key=True,
-        comment="菜单ID",
-    )
-    dept_id: Mapped[int] = mapped_column(
-        Integer,
-        ForeignKey("sys_dept.id", ondelete="CASCADE", onupdate="CASCADE"),
-        primary_key=True,
-        comment="部门ID",
-    )
-
-
-class RoleModel(ModelMixin):
-    """
-    角色模型
-
-    角色列表只显示当前用户绑定的角色
-    """
-
-    __tablename__: str = "sys_role"
-    __table_args__: dict[str, str] = {"comment": "角色表"}
-    __loader_options__: list[str] = ["menus", "depts"]
-    __permission_strategy__: PermissionFilterStrategy = PermissionFilterStrategy.USER_ROLE
-
-    name: Mapped[str] = mapped_column(String(64), nullable=False, comment="角色名称")
-    code: Mapped[str | None] = mapped_column(
-        String(16), nullable=True, index=True, comment="角色编码"
-    )
-    order: Mapped[int] = mapped_column(Integer, nullable=False, default=999, comment="显示排序")
-    data_scope: Mapped[int] = mapped_column(
-        Integer,
-        default=1,
-        nullable=False,
-        comment="数据权限范围(1:仅本人 2:本部门 3:本部门及以下 4:全部 5:自定义)",
-    )
-
-    menus: Mapped[list["MenuModel"]] = relationship(
-        secondary="sys_role_menus",
-        back_populates="roles",
-        lazy="selectin",
-        order_by="MenuModel.order",
-    )
-    depts: Mapped[list["DeptModel"]] = relationship(
-        secondary="sys_role_depts", back_populates="roles", lazy="selectin"
-    )
-    users: Mapped[list["UserModel"]] = relationship(
-        secondary="sys_user_roles", back_populates="roles", lazy="selectin"
-    )

frontend/修改的文件/permission.py

@@ -1,428 +0,0 @@
-from typing import Any
-
-from sqlalchemy import and_, or_, select
-from sqlalchemy.sql.elements import ColumnElement
-
-from app.api.v1.module_system.auth.schema import AuthSchema
-from app.api.v1.module_system.dept.model import DeptModel
-from app.api.v1.module_system.user.model import UserModel
-from app.common.enums import PermissionFilterStrategy
-from app.utils.common_util import get_child_id_map, get_child_recursion
-
-
-class Permission:
-    """
-    为业务模型提供数据权限过滤功能
-
-    使用策略模式，根据模型的 __permission_strategy__ 属性选择合适的过滤策略
-    """
-
-    # 数据权限常量定义，提高代码可读性
-    DATA_SCOPE_SELF = 1  # 仅本人数据
-    DATA_SCOPE_DEPT = 2  # 本部门数据
-    DATA_SCOPE_DEPT_AND_CHILD = 3  # 本部门及以下数据
-    DATA_SCOPE_ALL = 4  # 全部数据
-    DATA_SCOPE_CUSTOM = 5  # 自定义数据
-
-    def __init__(self, model: Any, auth: AuthSchema) -> None:
-        """
-        初始化权限过滤器实例
-
-        Args:
-            db: 数据库会话
-            model: 数据模型类
-            current_user: 当前用户对象
-            auth: 认证信息对象
-        """
-        self.model = model
-        self.auth = auth
-        self.conditions: list[ColumnElement] = []  # 权限条件列表
-
-    async def filter_query(self, query: Any) -> Any:
-        """
-        异步过滤查询对象
-
-        Args:
-            query: SQLAlchemy查询对象
-
-        Returns:
-            过滤后的查询对象
-        """
-        condition = await self.__permission_condition()
-        return query.where(condition) if condition is not None else query
-
-    async def __permission_condition(self) -> ColumnElement | None:
-        """
-        应用数据范围权限隔离
-
-        根据模型的权限过滤策略，选择合适的过滤方法
-        """
-        # 如果不需要检查数据权限,则不限制
-        if not self.auth.user:
-            return None
-
-        # 如果检查数据权限为False,则不限制
-        if not self.auth.check_data_scope:
-            return None
-
-        # 超级管理员可以查看所有数据
-        if self.auth.user.is_superuser:
-            return None
-
-        # 获取模型的权限过滤策略
-        strategy = getattr(self.model, "__permission_strategy__", PermissionFilterStrategy.DATA_SCOPE)
-
-        # 根据策略选择过滤方法
-        if strategy == PermissionFilterStrategy.ROLE_BASED:
-            return await self.__filter_by_role_based()
-        elif strategy == PermissionFilterStrategy.DEPT_BASED:
-            return await self.__filter_by_dept_based()
-        elif strategy == PermissionFilterStrategy.SELF_ONLY:
-            return await self.__filter_by_self_only()
-        elif strategy == PermissionFilterStrategy.USER_ROLE:
-            return await self.__filter_by_user_role()
-        else:
-            return await self.__filter_by_data_scope()
-
-    async def __filter_by_role_based(self) -> ColumnElement | None:
-        """
-        基于角色授权的权限过滤（适用于菜单等）
-
-        只显示用户角色授权的菜单
-        """
-        roles = getattr(self.auth.user, "roles", []) or []
-        if not roles:
-            id_attr = getattr(self.model, "id", None)
-            if id_attr is not None:
-                return id_attr == -1
-            return None
-
-        menu_ids = set()
-        for role in roles:
-            if hasattr(role, "menus") and role.menus:
-                menu_ids.update(menu.id for menu in role.menus if menu.status == "0")
-
-        if menu_ids:
-            id_attr = getattr(self.model, "id", None)
-            if id_attr is not None:
-                return id_attr.in_(list(menu_ids))
-
-        id_attr = getattr(self.model, "id", None)
-        if id_attr is not None:
-            return id_attr == -1
-        return None
-
-    async def __filter_by_user_role(self) -> ColumnElement | None:
-        """
-        基于当前用户绑定角色的权限过滤（适用于角色列表）
-
-        只显示当前用户绑定的角色
-        """
-        roles = getattr(self.auth.user, "roles", []) or []
-        if not roles:
-            id_attr = getattr(self.model, "id", None)
-            if id_attr is not None:
-                return id_attr == -1
-            return None
-
-        role_ids = [role.id for role in roles]
-        id_attr = getattr(self.model, "id", None)
-        if id_attr is not None:
-            return id_attr.in_(role_ids)
-        return None
-
-    async def __filter_by_dept_based(self) -> ColumnElement | None:
-        """
-        基于部门关联的权限过滤（适用于部门、角色等）
-
-        根据用户的部门权限范围过滤数据
-        """
-        # 如果用户没有角色,则只能查看自己部门的数据
-        roles = getattr(self.auth.user, "roles", []) or []
-        if not roles:
-            user_dept_id = getattr(self.auth.user, "dept_id", None)
-            if user_dept_id is not None and hasattr(self.model, "id"):
-                id_attr = getattr(self.model, "id", None)
-                if id_attr is not None:
-                    return id_attr == user_dept_id
-            return None
-
-        # 获取用户所有角色的权限范围（支持菜单/按钮级覆盖）
-        data_scopes, custom_dept_ids = await self.__resolve_menu_data_scope(
-            roles, self.auth.current_permission
-        )
-
-        # 全部数据权限最高优先级
-        if self.DATA_SCOPE_ALL in data_scopes:
-            return None
-
-        # 收集所有可访问的部门ID
-        accessible_dept_ids = await self.__get_accessible_dept_ids(data_scopes, custom_dept_ids)
-
-        # 根据模型类型过滤
-        if self.model.__name__ == "DeptModel":
-            return self.__filter_dept_model(accessible_dept_ids)
-        elif self.model.__name__ == "UserModel":
-            return self.__filter_user_model(accessible_dept_ids)
-        else:
-            return None
-
-    async def __filter_by_self_only(self) -> ColumnElement | None:
-        """
-        仅本人数据权限过滤
-        """
-        created_id_attr = getattr(self.model, "created_id", None)
-        if created_id_attr is not None and self.auth.user:
-            return created_id_attr == self.auth.user.id
-        return None
-
-    async def __filter_by_data_scope(self) -> ColumnElement | None:
-        """
-        基于数据范围权限的通用过滤（默认策略）
-
-        适用于大多数业务模型
-        """
-        # 如果模型没有创建人created_id字段,则不限制
-        if not hasattr(self.model, "created_id"):
-            return None
-
-        # 如果用户没有角色,则只能查看自己的数据
-        roles = getattr(self.auth.user, "roles", []) or []
-        if not roles:
-            created_id_attr = getattr(self.model, "created_id", None)
-            if created_id_attr is not None and self.auth.user:
-                return created_id_attr == self.auth.user.id
-            return None
-
-        # 获取用户所有角色的权限范围（支持菜单/按钮级覆盖）
-        data_scopes, custom_dept_ids = await self.__resolve_menu_data_scope(
-            roles, self.auth.current_permission
-        )
-
-        # 全部数据权限最高优先级
-        if self.DATA_SCOPE_ALL in data_scopes:
-            return None
-
-        # 收集所有可访问的部门ID
-        accessible_dept_ids = await self.__get_accessible_dept_ids(data_scopes, custom_dept_ids)
-
-        # 如果有部门权限，使用部门过滤
-        if accessible_dept_ids:
-            # 特殊处理：如果模型本身就是UserModel，直接过滤用户的dept_id
-            if self.model.__name__ == "UserModel" and hasattr(self.model, "dept_id"):
-                dept_id_attr = getattr(self.model, "dept_id", None)
-                if dept_id_attr is not None:
-                    return dept_id_attr.in_(list(accessible_dept_ids))
-
-            # 其他模型：通过created_by关系过滤创建人的部门
-            creator_rel = getattr(self.model, "created_by", None)
-            if creator_rel is not None and hasattr(UserModel, "dept_id"):
-                return creator_rel.has(UserModel.dept_id.in_(list(accessible_dept_ids)))
-
-            # 降级方案：只能查看自己的数据
-            created_id_attr = getattr(self.model, "created_id", None)
-            if created_id_attr is not None and self.auth.user:
-                return created_id_attr == self.auth.user.id
-            return None
-
-        # 处理仅本人数据权限
-        if self.DATA_SCOPE_SELF in data_scopes:
-            created_id_attr = getattr(self.model, "created_id", None)
-            if created_id_attr is not None and self.auth.user:
-                return created_id_attr == self.auth.user.id
-            return None
-
-        # 默认情况：只能查看自己的数据
-        created_id_attr = getattr(self.model, "created_id", None)
-        if created_id_attr is not None and self.auth.user:
-            return created_id_attr == self.auth.user.id
-        return None
-
-    async def __resolve_menu_data_scope(
-        self, roles: list, current_permission: str | None
-    ) -> tuple[set[int], set[int]]:
-        """
-        解析数据权限范围，支持菜单/按钮级覆盖
-
-        三级继承链路：按钮自身 data_scope → 父菜单 data_scope → 角色 data_scope
-        多角色场景采用并集策略（权限最大化）
-
-        Args:
-            roles: 用户的角色列表
-            current_permission: 当前请求的权限标识（如 "module_system:user:query"）
-
-        Returns:
-            (data_scopes, custom_dept_ids) 元组
-        """
-        data_scopes: set[int] = set()
-        custom_dept_ids: set[int] = set()
-
-        if not current_permission:
-            # 回退：使用角色级 data_scope（现有行为）
-            for role in roles:
-                data_scopes.add(role.data_scope)
-                if role.data_scope == self.DATA_SCOPE_CUSTOM and hasattr(role, "depts") and role.depts:
-                    custom_dept_ids.update(dept.id for dept in role.depts)
-            return data_scopes, custom_dept_ids
-
-        # 在所有角色中查找匹配 current_permission 的菜单，同时记录菜单对象以获取 type 和 parent_id
-        role_menu_pairs: list[tuple[int, int, Any, Any]] = []  # (role_id, menu_id, role, menu)
-        for role in roles:
-            for menu in (getattr(role, "menus", None) or []):
-                if (
-                    getattr(menu, "permission", None) == current_permission
-                    and getattr(menu, "status", None) == "0"
-                ):
-                    role_menu_pairs.append((role.id, menu.id, role, menu))
-                    break  # 每个角色只取一个匹配
-
-        if not role_menu_pairs:
-            # 没有角色包含此权限——回退到角色级
-            for role in roles:
-                data_scopes.add(role.data_scope)
-                if role.data_scope == self.DATA_SCOPE_CUSTOM and hasattr(role, "depts") and role.depts:
-                    custom_dept_ids.update(dept.id for dept in role.depts)
-            return data_scopes, custom_dept_ids
-
-        # 批量查询 RoleMenusModel 中的菜单级 data_scope
-        from app.api.v1.module_system.role.model import RoleMenuDeptsModel, RoleMenusModel
-
-        pairs_filter = or_(
-            *[
-                and_(RoleMenusModel.role_id == rid, RoleMenusModel.menu_id == mid)
-                for rid, mid, _, _ in role_menu_pairs
-            ]
-        )
-        result = await self.auth.db.execute(
-            select(RoleMenusModel.role_id, RoleMenusModel.menu_id, RoleMenusModel.data_scope).where(
-                pairs_filter
-            )
-        )
-        per_menu_scopes = {(row.role_id, row.menu_id): row.data_scope for row in result}
-
-        # 对 type=3 且自身 data_scope 为 NULL 的按钮，收集需要查询父菜单 data_scope 的对
-        parent_lookups: list[tuple[int, int]] = []  # (role_id, parent_menu_id)
-        for role_id, menu_id, _, menu in role_menu_pairs:
-            scope = per_menu_scopes.get((role_id, menu_id))
-            parent_id = getattr(menu, "parent_id", None)
-            if scope is None and getattr(menu, "type", None) == 3 and parent_id:
-                parent_lookups.append((role_id, parent_id))
-
-        # 批量查询父菜单 data_scope
-        parent_scopes: dict[tuple[int, int], int | None] = {}
-        if parent_lookups:
-            parent_filter = or_(
-                *[
-                    and_(RoleMenusModel.role_id == rid, RoleMenusModel.menu_id == pid)
-                    for rid, pid in parent_lookups
-                ]
-            )
-            parent_result = await self.auth.db.execute(
-                select(
-                    RoleMenusModel.role_id, RoleMenusModel.menu_id, RoleMenusModel.data_scope
-                ).where(parent_filter)
-            )
-            parent_scopes = {(row.role_id, row.menu_id): row.data_scope for row in parent_result}
-
-        # 解析每个角色在该权限下的实际 data_scope（三级继承）
-        for role_id, menu_id, role, menu in role_menu_pairs:
-            scope = per_menu_scopes.get((role_id, menu_id))
-            # 记录 scope 实际来源的 menu_id（用于查找自定义部门）
-            scope_menu_id: int | None = menu_id
-
-            # 三级继承：按钮自身 → 父菜单 → 角色
-            if scope is None and getattr(menu, "type", None) == 3 and getattr(menu, "parent_id", None):
-                parent_scope = parent_scopes.get((role_id, menu.parent_id))
-                if parent_scope is not None:
-                    scope = parent_scope
-                    scope_menu_id = menu.parent_id  # scope 来源于父菜单
-
-            if scope is None:
-                scope = role.data_scope  # 最终回退到角色级
-                scope_menu_id = None  # scope 来源于角色级
-
-            data_scopes.add(scope)
-
-            if scope == self.DATA_SCOPE_CUSTOM:
-                if scope_menu_id is not None:
-                    # 从 scope 来源的菜单级查找自定义部门
-                    dept_result = await self.auth.db.execute(
-                        select(RoleMenuDeptsModel.dept_id).where(
-                            RoleMenuDeptsModel.role_id == role_id,
-                            RoleMenuDeptsModel.menu_id == scope_menu_id,
-                        )
-                    )
-                    menu_dept_ids = {row.dept_id for row in dept_result}
-                    if menu_dept_ids:
-                        custom_dept_ids.update(menu_dept_ids)
-                    elif hasattr(role, "depts") and role.depts:
-                        # 菜单级没有配置部门，回退到角色级自定义部门
-                        custom_dept_ids.update(dept.id for dept in role.depts)
-                elif hasattr(role, "depts") and role.depts:
-                    # scope 来源于角色级，直接使用角色级自定义部门
-                    custom_dept_ids.update(dept.id for dept in role.depts)
-
-        return data_scopes, custom_dept_ids
-
-    async def __get_accessible_dept_ids(
-        self, data_scopes: set, custom_dept_ids: set
-    ) -> set[int]:
-        """
-        获取用户可访问的所有部门ID
-
-        Args:
-            data_scopes: 用户角色的数据权限范围集合
-            custom_dept_ids: 自定义权限关联的部门ID集合
-
-        Returns:
-            可访问的部门ID集合
-        """
-        accessible_dept_ids = set()
-        user_dept_id = getattr(self.auth.user, "dept_id", None)
-
-        # 处理自定义数据权限（5）
-        if self.DATA_SCOPE_CUSTOM in data_scopes:
-            accessible_dept_ids.update(custom_dept_ids)
-
-        # 处理本部门数据权限（2）
-        if self.DATA_SCOPE_DEPT in data_scopes and user_dept_id is not None:
-            accessible_dept_ids.add(user_dept_id)
-
-        # 处理本部门及以下数据权限（3）
-        if self.DATA_SCOPE_DEPT_AND_CHILD in data_scopes and user_dept_id is not None:
-            try:
-                dept_sql = select(DeptModel)
-                dept_result = await self.auth.db.execute(dept_sql)
-                dept_objs = dept_result.scalars().all()
-                id_map = get_child_id_map(dept_objs)
-                dept_with_children_ids = get_child_recursion(id=user_dept_id, id_map=id_map)
-                accessible_dept_ids.update(dept_with_children_ids)
-            except Exception:
-                accessible_dept_ids.add(user_dept_id)
-
-        return accessible_dept_ids
-
-    def __filter_dept_model(self, accessible_dept_ids: set[int]) -> ColumnElement | None:
-        """
-        过滤部门模型
-        """
-        if accessible_dept_ids:
-            id_attr = getattr(self.model, "id", None)
-            if id_attr is not None:
-                return id_attr.in_(list(accessible_dept_ids))
-        user_dept_id = getattr(self.auth.user, "dept_id", None)
-        if user_dept_id is not None:
-            id_attr = getattr(self.model, "id", None)
-            if id_attr is not None:
-                return id_attr == user_dept_id
-        return None
-
-    def __filter_user_model(self, accessible_dept_ids: set[int]) -> ColumnElement | None:
-        """
-        过滤用户模型
-        """
-        if accessible_dept_ids:
-            dept_id_attr = getattr(self.model, "dept_id", None)
-            if dept_id_attr is not None:
-                return dept_id_attr.in_(list(accessible_dept_ids))
-        return None

frontend/修改的文件/role.ts

@@ -1,127 +0,0 @@
-import request from "@/utils/request";
-
-const API_PATH = "/system/role";
-
-const RoleAPI = {
-  listRole(query?: TablePageQuery) {
-    return request<ApiResponse<PageResult<RoleTable[]>>>({
-      url: `${API_PATH}/list`,
-      method: "get",
-      params: query,
-    });
-  },
-
-  detailRole(query: number) {
-    return request<ApiResponse<RoleTable>>({
-      url: `${API_PATH}/detail/${query}`,
-      method: "get",
-    });
-  },
-
-  createRole(body: RoleForm) {
-    return request<ApiResponse>({
-      url: `${API_PATH}/create`,
-      method: "post",
-      data: body,
-    });
-  },
-
-  updateRole(id: number, body: RoleForm) {
-    return request<ApiResponse>({
-      url: `${API_PATH}/update/${id}`,
-      method: "put",
-      data: body,
-    });
-  },
-
-  deleteRole(body: number[]) {
-    return request<ApiResponse>({
-      url: `${API_PATH}/delete`,
-      method: "delete",
-      data: body,
-    });
-  },
-
-  batchRole(body: BatchType) {
-    return request<ApiResponse>({
-      url: `${API_PATH}/available/setting`,
-      method: "patch",
-      data: body,
-    });
-  },
-
-  setPermission(body: permissionDataType) {
-    return request<ApiResponse>({
-      url: `${API_PATH}/permission/setting`,
-      method: "patch",
-      data: body,
-    });
-  },
-
-  exportRole(body: TablePageQuery) {
-    return request<Blob>({
-      url: `${API_PATH}/export`,
-      method: "post",
-      data: body,
-      responseType: "blob",
-    });
-  },
-};
-
-export default RoleAPI;
-
-export interface TablePageQuery extends PageQuery {
-  name?: string;
-  status?: string;
-  created_time?: string[];
-  updated_time?: string[];
-}
-
-export interface RoleTable extends BaseType {
-  id: number;
-  name: string;
-  order?: number;
-  code?: string;
-  data_scope?: number;
-  menus?: permissionMenuType[];
-  depts?: permissionDeptType[];
-  menu_data_scopes?: MenuDataScopeItem[];
-}
-
-export interface RoleForm extends BaseFormType {
-  name?: string;
-  order?: number;
-  code?: string;
-}
-
-export interface MenuDataScopeItem {
-  menu_id: number;
-  data_scope: number | null; // null = 继承角色默认
-  dept_ids: number[];
-}
-
-export interface permissionDataType {
-  data_scope: number;
-  role_ids: RoleTable["id"][];
-  menu_ids: permissionMenuType["id"][];
-  dept_ids: permissionDeptType["id"][];
-  menu_data_scopes: MenuDataScopeItem[];
-}
-
-export interface permissionDeptType {
-  id: number;
-  name: string;
-  parent_id: number;
-  children: permissionDeptType[];
-}
-
-export interface permissionMenuType {
-  id: number;
-  name: string;
-  type: number;
-  permission: string;
-  parent_id?: number;
-  status: string;
-  description?: string;
-  children?: permissionMenuType[];
-}

frontend/修改的文件/schema.py

@@ -1,116 +0,0 @@
-from fastapi import Query
-from pydantic import (
-    BaseModel,
-    ConfigDict,
-    Field,
-    field_validator,
-    model_validator,
-)
-
-from app.api.v1.module_system.dept.schema import DeptOutSchema
-from app.api.v1.module_system.menu.schema import MenuOutSchema
-from app.common.enums import QueueEnum
-from app.core.base_schema import BaseSchema
-from app.core.validator import (
-    DateTimeStr,
-    code_validator,
-    role_permission_request_validator,
-)
-
-
-class RoleCreateSchema(BaseModel):
-    """角色创建模型"""
-
-    name: str = Field(..., max_length=64, description="角色名称")
-    code: str | None = Field(default=None, max_length=16, description="角色编码")
-    order: int | None = Field(default=1, ge=1, description="显示排序")
-    data_scope: int | None = Field(
-        default=1,
-        description="数据权限范围(1:仅本人 2:本部门 3:本部门及以下 4:全部 5:自定义)",
-    )
-    status: str = Field(default="0", description="是否启用")
-    description: str | None = Field(default=None, max_length=255, description="描述")
-
-    @field_validator("code")
-    @classmethod
-    def validate_code(cls, value: str | None):
-        return code_validator(value)
-
-
-class MenuDataScopeItem(BaseModel):
-    """菜单级数据权限配置项"""
-
-    menu_id: int = Field(..., description="菜单ID")
-    data_scope: int | None = Field(default=None, description="数据权限范围(NULL继承角色级)")
-    dept_ids: list[int] = Field(default_factory=list, description="自定义部门ID列表(data_scope=5时)")
-
-
-class RolePermissionSettingSchema(BaseModel):
-    """角色权限配置模型"""
-
-    data_scope: int = Field(
-        default=1,
-        description="数据权限范围(1:仅本人 2:本部门 3:本部门及以下 4:全部 5:自定义)",
-    )
-    role_ids: list[int] = Field(default_factory=list, description="角色ID列表")
-    menu_ids: list[int] = Field(default_factory=list, description="菜单ID列表")
-    dept_ids: list[int] = Field(default_factory=list, description="部门ID列表")
-    menu_data_scopes: list[MenuDataScopeItem] = Field(
-        default_factory=list, description="菜单级数据权限配置列表"
-    )
-
-    @model_validator(mode="after")
-    def validate_fields(self):
-        """验证权限配置字段"""
-        return role_permission_request_validator(self)
-
-
-class RoleUpdateSchema(RoleCreateSchema):
-    """角色更新模型"""
-
-
-class RoleOutSchema(RoleCreateSchema, BaseSchema):
-    """角色信息响应模型"""
-
-    model_config = ConfigDict(from_attributes=True)
-
-    menus: list[MenuOutSchema] = Field(default_factory=list, description="角色菜单列表")
-    depts: list[DeptOutSchema] = Field(default_factory=list, description="角色部门列表")
-    menu_data_scopes: list[MenuDataScopeItem] = Field(
-        default_factory=list, description="菜单级数据权限配置列表"
-    )
-
-
-class RoleQueryParam:
-    """角色管理查询参数"""
-
-    def __init__(
-        self,
-        name: str | None = Query(None, description="角色名称"),
-        description: str | None = Query(None, description="描述"),
-        status: str | None = Query(None, description="是否启用"),
-        created_time: list[DateTimeStr] | None = Query(
-            None,
-            description="创建时间范围",
-            examples=["2025-01-01 00:00:00", "2025-12-31 23:59:59"],
-        ),
-        updated_time: list[DateTimeStr] | None = Query(
-            None,
-            description="更新时间范围",
-            examples=["2025-01-01 00:00:00", "2025-12-31 23:59:59"],
-        ),
-    ) -> None:
-        # 模糊查询字段
-        self.name = (QueueEnum.like.value, name)
-        if description:
-            self.description = (QueueEnum.like.value, description)
-
-        # 精确查询字段
-        if status:
-            self.status = (QueueEnum.eq.value, status)
-
-        # 时间范围查询
-        if created_time and len(created_time) == 2:
-            self.created_time = (QueueEnum.between.value, (created_time[0], created_time[1]))
-        if updated_time and len(updated_time) == 2:
-            self.updated_time = (QueueEnum.between.value, (updated_time[0], updated_time[1]))

frontend/修改的文件/service.py

@@ -1,219 +0,0 @@
-from typing import Any
-
-from app.api.v1.module_system.auth.schema import AuthSchema
-from app.core.base_schema import BatchSetAvailable
-from app.core.exceptions import CustomException
-from app.utils.excel_util import ExcelUtil
-
-from .crud import RoleCRUD
-from .schema import (
-    RoleCreateSchema,
-    RoleOutSchema,
-    RolePermissionSettingSchema,
-    RoleQueryParam,
-    RoleUpdateSchema,
-)
-
-
-class RoleService:
-    """角色模块服务层"""
-
-    @classmethod
-    async def get_role_detail_service(cls, auth: AuthSchema, id: int) -> dict:
-        """
-        获取角色详情
-
-        参数:
-        - auth (AuthSchema): 认证信息模型
-        - id (int): 角色ID
-
-        返回:
-        - dict: 角色详情字典
-        """
-        role = await RoleCRUD(auth).get_by_id_crud(id=id)
-        role_dict = RoleOutSchema.model_validate(role).model_dump()
-        # 加载菜单级数据权限配置
-        role_dict["menu_data_scopes"] = await RoleCRUD(auth).get_role_menu_data_scopes_crud(
-            role_id=id
-        )
-        return role_dict
-
-    @classmethod
-    async def get_role_list_service(
-        cls,
-        auth: AuthSchema,
-        search: RoleQueryParam | None = None,
-        order_by: list[dict[str, str]] | None = None,
-    ) -> list[dict]:
-        """
-        获取角色列表
-
-        参数:
-        - auth (AuthSchema): 认证信息模型
-        - search (RoleQueryParam | None): 查询参数模型
-        - order_by (list[dict[str, str]] | None): 排序参数列表
-
-        返回:
-        - list[dict]: 角色详情字典列表
-        """
-        role_list = await RoleCRUD(auth).get_list_crud(search=search.__dict__, order_by=order_by)
-        return [RoleOutSchema.model_validate(role).model_dump() for role in role_list]
-
-    @classmethod
-    async def create_role_service(cls, auth: AuthSchema, data: RoleCreateSchema) -> dict:
-        """
-        创建角色
-
-        参数:
-        - auth (AuthSchema): 认证信息模型
-        - data (RoleCreateSchema): 创建角色模型
-
-        返回:
-        - dict: 新创建的角色详情字典
-        """
-        role = await RoleCRUD(auth).get(name=data.name)
-        if role:
-            raise CustomException(msg="创建失败，该角色已存在")
-        obj = await RoleCRUD(auth).get(code=data.code)
-        if obj:
-            raise CustomException(msg="创建失败，编码已存在")
-        new_role = await RoleCRUD(auth).create(data=data)
-        return RoleOutSchema.model_validate(new_role).model_dump()
-
-    @classmethod
-    async def update_role_service(cls, auth: AuthSchema, id: int, data: RoleUpdateSchema) -> dict:
-        """
-        更新角色
-
-        参数:
-        - auth (AuthSchema): 认证信息模型
-        - id (int): 角色ID
-        - data (RoleUpdateSchema): 更新角色模型
-
-        返回:
-        - dict: 更新后的角色详情字典
-        """
-        role = await RoleCRUD(auth).get_by_id_crud(id=id)
-        if not role:
-            raise CustomException(msg="更新失败，该角色不存在")
-        exist_role = await RoleCRUD(auth).get(name=data.name)
-        if exist_role and exist_role.id != id:
-            raise CustomException(msg="更新失败，角色名称重复")
-        updated_role = await RoleCRUD(auth).update(id=id, data=data)
-        return RoleOutSchema.model_validate(updated_role).model_dump()
-
-    @classmethod
-    async def delete_role_service(cls, auth: AuthSchema, ids: list[int]) -> None:
-        """
-        删除角色
-
-        参数:
-        - auth (AuthSchema): 认证信息模型
-        - ids (list[int]): 角色ID列表
-
-        返回:
-        - None
-        """
-        if len(ids) < 1:
-            raise CustomException(msg="删除失败，删除对象不能为空")
-        for id in ids:
-            role = await RoleCRUD(auth).get_by_id_crud(id=id)
-            if not role:
-                raise CustomException(msg="删除失败，该角色不存在")
-        await RoleCRUD(auth).delete(ids=ids)
-
-    @classmethod
-    async def set_role_permission_service(
-        cls, auth: AuthSchema, data: RolePermissionSettingSchema
-    ) -> None:
-        """
-        设置角色权限
-
-        参数:
-        - auth (AuthSchema): 认证信息模型
-        - data (RolePermissionSettingSchema): 角色权限设置模型
-
-        返回:
-        - None
-        """
-        # 设置角色菜单权限
-        await RoleCRUD(auth).set_role_menus_crud(role_ids=data.role_ids, menu_ids=data.menu_ids)
-
-        # 设置数据权限范围
-        await RoleCRUD(auth).set_role_data_scope_crud(
-            role_ids=data.role_ids, data_scope=data.data_scope
-        )
-
-        # 设置自定义数据权限部门
-        if data.data_scope == 5 and data.dept_ids:
-            await RoleCRUD(auth).set_role_depts_crud(role_ids=data.role_ids, dept_ids=data.dept_ids)
-        else:
-            await RoleCRUD(auth).set_role_depts_crud(role_ids=data.role_ids, dept_ids=[])
-
-        # 设置菜单级数据权限
-        if data.menu_data_scopes:
-            for role_id in data.role_ids:
-                await RoleCRUD(auth).set_role_menu_data_scopes_crud(
-                    role_id=role_id, menu_data_scopes=data.menu_data_scopes
-                )
-
-    @classmethod
-    async def set_role_available_service(cls, auth: AuthSchema, data: BatchSetAvailable) -> None:
-        """
-        设置角色可用状态
-
-        参数:
-        - auth (AuthSchema): 认证信息模型
-        - data (BatchSetAvailable): 批量设置可用状态模型
-
-        返回:
-        - None
-        """
-        await RoleCRUD(auth).set_available_crud(ids=data.ids, status=data.status)
-
-    @classmethod
-    async def export_role_list_service(cls, role_list: list[dict[str, Any]]) -> bytes:
-        """
-        导出角色列表
-
-        参数:
-        - role_list (list[dict[str, Any]]): 角色详情字典列表
-
-        返回:
-        - bytes: Excel文件字节流
-        """
-        # 字段映射配置
-        mapping_dict = {
-            "id": "角色编号",
-            "name": "角色名称",
-            "order": "显示顺序",
-            "data_scope": "数据权限",
-            "status": "状态",
-            "description": "备注",
-            "created_time": "创建时间",
-            "updated_time": "更新时间",
-            "created_id": "创建者ID",
-            "updated_id": "更新者ID",
-        }
-
-        # 数据权限映射
-        data_scope_map = {
-            1: "仅本人数据权限",
-            2: "本部门数据权限",
-            3: "本部门及以下数据权限",
-            4: "全部数据权限",
-            5: "自定义数据权限",
-        }
-
-        # 处理数据
-        data = role_list.copy()
-        for item in data:
-            item["status"] = "启用" if item.get("status") == "0" else "停用"
-            item["data_scope"] = data_scope_map.get(item.get("data_scope", 1), "")
-            item["creator"] = (
-                item.get("creator", {}).get("name", "未知")
-                if isinstance(item.get("creator"), dict)
-                else "未知"
-            )
-
-        return ExcelUtil.export_list2excel(list_data=data, mapping_dict=mapping_dict)

frontend/修改的文件/validator.py

@@ -1,285 +0,0 @@
-import re
-from datetime import date, datetime, time
-from typing import Annotated, Any
-
-from pydantic import AfterValidator, PlainSerializer, WithJsonSchema
-
-from app.common.constant import RET
-from app.core.exceptions import CustomException
-
-# 自定义日期时间字符串类型
-DateTimeStr = Annotated[
-    datetime,
-    AfterValidator(lambda x: datetime_validator(x)),
-    PlainSerializer(
-        lambda x: x.strftime("%Y-%m-%d %H:%M:%S") if isinstance(x, datetime) else str(x),
-        return_type=str,
-    ),
-    WithJsonSchema({"type": "string"}, mode="serialization"),
-]
-
-# 自定义日期字符串类型
-DateStr = Annotated[
-    date,
-    AfterValidator(lambda x: date_validator(x)),
-    PlainSerializer(
-        lambda x: x.strftime("%Y-%m-%d") if isinstance(x, date) else str(x),
-        return_type=str,
-    ),
-    WithJsonSchema({"type": "string"}, mode="serialization"),
-]
-
-# 自定义时间字符串类型
-TimeStr = Annotated[
-    time,
-    AfterValidator(lambda x: time_validator(x)),
-    PlainSerializer(
-        lambda x: x.strftime("%H:%M:%S") if isinstance(x, time) else str(x),
-        return_type=str,
-    ),
-    WithJsonSchema({"type": "string"}, mode="serialization"),
-]
-
-# 自定义手机号类型
-Telephone = Annotated[
-    str,
-    AfterValidator(lambda x: mobile_validator(x)),
-    PlainSerializer(lambda x: x, return_type=str),
-    WithJsonSchema({"type": "string"}, mode="serialization"),
-]
-
-# 自定义邮箱类型
-Email = Annotated[
-    str,
-    AfterValidator(lambda x: email_validator(x)),
-    PlainSerializer(lambda x: x, return_type=str),
-    WithJsonSchema({"type": "string"}, mode="serialization"),
-]
-
-
-def datetime_validator(value: str | datetime) -> datetime:
-    """
-    日期格式验证器。
-
-    参数:
-    - value (str | datetime): 日期值。
-
-    返回:
-    - datetime: 格式化后的日期。
-
-    异常:
-    - CustomException: 日期格式无效时抛出。
-    """
-    pattern = "%Y-%m-%d %H:%M:%S"
-    try:
-        if isinstance(value, str):
-            return datetime.strptime(value, pattern)
-        if isinstance(value, datetime):
-            return value
-    except Exception:
-        raise CustomException(code=RET.ERROR.code, msg="无效的日期格式")
-
-
-def date_validator(value: str | date) -> date:
-    """
-    日期格式验证器。
-
-    参数:
-    - value (str | date): 日期值。
-
-    返回:
-    - date: 格式化后的日期。
-
-    异常:
-    - CustomException: 日期格式无效时抛出。
-    """
-    try:
-        if isinstance(value, str):
-            return datetime.strptime(value, "%Y-%m-%d").date()
-        if isinstance(value, date):
-            return value
-    except Exception:
-        raise CustomException(code=RET.ERROR.code, msg="无效的日期格式")
-
-
-def time_validator(value: str | time) -> time:
-    """
-    时间格式验证器。
-
-    参数:
-    - value (str | time): 时间值。
-
-    返回:
-    - time: 格式化后的时间。
-
-    异常:
-    - CustomException: 时间格式无效时抛出。
-    """
-    try:
-        if isinstance(value, str):
-            return datetime.strptime(value, "%H:%M:%S").time()
-        if isinstance(value, time):
-            return value
-    except Exception:
-        raise CustomException(code=RET.ERROR.code, msg="无效的时间格式")
-
-
-def email_validator(value: str) -> str:
-    """
-    邮箱地址验证器。
-
-    参数:
-    - value (str): 邮箱地址。
-
-    返回:
-    - str: 验证后的邮箱地址。
-
-    异常:
-    - CustomException: 邮箱格式无效时抛出。
-    """
-    if not value:
-        raise CustomException(code=RET.ERROR.code, msg="邮箱地址不能为空")
-
-    regex = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
-
-    if not re.match(regex, value):
-        raise CustomException(code=RET.ERROR.code, msg="邮箱地址格式不正确")
-
-    return value
-
-
-def mobile_validator(value: str | None) -> str | None:
-    """
-    手机号验证器。
-
-    参数:
-    - value (str | None): 手机号。
-
-    返回:
-    - str | None: 验证后的手机号。
-
-    异常:
-    - CustomException: 手机号格式无效时抛出。
-    """
-    if not value:
-        return value
-
-    if len(value) != 11 or not value.isdigit():
-        raise CustomException(code=RET.ERROR.code, msg="手机号格式不正确")
-
-    regex = r"^1(3\d|4[4-9]|5[0-35-9]|6[67]|7[013-8]|8[0-9]|9[0-9])\d{8}$"
-
-    if not re.match(regex, value):
-        raise CustomException(code=RET.ERROR.code, msg="手机号格式不正确")
-
-    return value
-
-
-def code_validator(value: str | None) -> str | None:
-    """
-    编码验证器。
-
-    参数:
-    - value (str | None): 编码。
-
-    返回:
-    - str | None: 验证后的编码。
-
-    异常:
-    - CustomException: 编码格式无效时抛出。
-    """
-    if not value:
-        return value
-    v = value.strip()
-    if not re.match(r"^[A-Za-z][A-Za-z0-9_]{1,15}$", v):
-        raise CustomException(
-            code=RET.ERROR.code,
-            msg="编码需字母开头，允许字母/数字/下划线，长度2-16",
-        )
-    return v
-
-
-def menu_request_validator(data: Any) -> Any:
-    """
-    菜单请求数据验证器。
-
-    参数:
-    - data (Any): 请求数据。
-
-    返回:
-    - Any: 验证后的请求数据。
-
-    异常:
-    - CustomException: 请求数据无效时抛出。
-    """
-    menu_types = {1: "目录", 2: "功能", 3: "权限", 4: "外链"}
-
-    if data.type not in menu_types:
-        raise CustomException(
-            code=RET.ERROR.code,
-            msg=f"菜单类型必须为: {','.join(map(str, menu_types.keys()))}",
-        )
-
-    if data.type in [1, 2]:
-        if not data.route_name:
-            raise CustomException(code=RET.ERROR.code, msg="路由名称不能为空")
-        if not data.route_path:
-            raise CustomException(code=RET.ERROR.code, msg="路由路径不能为空")
-
-    if data.type == 2 and not data.component_path:
-        raise CustomException(code=RET.ERROR.code, msg="组件路径不能为空")
-
-    return data
-
-
-def role_permission_request_validator(data: Any) -> Any:
-    """
-    角色权限设置数据验证器。
-
-    参数:
-    - data (Any): 请求数据。
-
-    返回:
-    - Any: 验证后的请求数据。
-
-    异常:
-    - CustomException: 请求数据无效时抛出。
-    """
-    data_scopes = {
-        1: "仅本人数据权限",
-        2: "本部门数据权限",
-        3: "本部门及以下数据权限",
-        4: "全部数据权限",
-        5: "自定义数据权限",
-    }
-
-    if data.data_scope not in data_scopes:
-        raise CustomException(
-            code=RET.ERROR.code,
-            msg=f"数据权限范围必须为: {','.join(map(str, data_scopes.keys()))}",
-        )
-
-    if not data.role_ids:
-        raise CustomException(code=RET.ERROR.code, msg="角色不能为空")
-
-    # 验证菜单级数据权限配置
-    if hasattr(data, "menu_data_scopes") and data.menu_data_scopes:
-        menu_id_set = set(data.menu_ids) if data.menu_ids else set()
-        for item in data.menu_data_scopes:
-            if item.data_scope is not None and item.data_scope not in data_scopes:
-                raise CustomException(
-                    code=RET.ERROR.code,
-                    msg=f"菜单级数据权限范围必须为: {','.join(map(str, data_scopes.keys()))} 或 NULL",
-                )
-            if item.data_scope != 5 and item.dept_ids:
-                raise CustomException(
-                    code=RET.ERROR.code,
-                    msg="仅当数据权限范围为自定义(5)时才能配置部门列表",
-                )
-            if menu_id_set and item.menu_id not in menu_id_set:
-                raise CustomException(
-                    code=RET.ERROR.code,
-                    msg=f"菜单级数据权限中的菜单ID {item.menu_id} 不在已分配的菜单列表中",
-                )
-
-    return data