shanxia2.0/backend/app/core/data_scope.py

"""数据权限过滤（V0.3：基于角色 data_scope 的查询级行级过滤）

data_scope 值：
  1 = 全部数据
  2 = 本部门数据
  3 = 本部门及以下数据
  4 = 仅本人数据
  5 = 自定义（sys_role_data_scope_detail）
"""
from __future__ import annotations

from typing import List, Set, Tuple

from sqlalchemy import or_
from sqlalchemy.orm import Session

from app.core.config import settings
from app.models.models import (
    SysDept,
    SysRole,
    SysRoleDataScopeDetail,
    SysUser,
    SysUserRole,
)


def get_user_dept_id(db: Session, user_id: str) -> str | None:
    """从 SysUser 读取用户的 dept_id。"""
    row = (
        db.query(SysUser.dept_id)
        .filter(SysUser.id == user_id, SysUser.del_flag == 0)
        .first()
    )
    return row[0] if row and row[0] else None


def _get_child_dept_ids(db: Session, dept_id: str, tenant_no: str) -> List[str]:
    """递归获取 dept_id 及其所有子部门 ID（含自身）。"""
    result: List[str] = [dept_id]
    # 用循环逐层扩展，避免深度递归栈溢出
    queue: List[str] = [dept_id]
    while queue:
        parents = queue
        queue = []
        children = (
            db.query(SysDept.id)
            .filter(
                SysDept.parent_id.in_(parents),
                SysDept.del_flag == 0,
                SysDept.tenant_no == tenant_no,
            )
            .all()
        )
        for child in children:
            cid = child[0]
            if cid not in result:
                result.append(cid)
                queue.append(cid)
    return result


def _resolve_user_role_scopes(
    db: Session, user_id: str, tenant_no: str
) -> List[dict]:
    """获取用户所有角色的数据权限配置。

    返回: [{"data_scope": str, "details": [SysRoleDataScopeDetail, ...]}]
    """
    roles = (
        db.query(SysRole)
        .join(SysUserRole, SysUserRole.role_id == SysRole.id)
        .filter(
            SysUserRole.user_id == user_id,
            SysUserRole.del_flag == 0,
            SysUserRole.tenant_no == tenant_no,
            SysRole.del_flag == 0,
            SysRole.status == "1",
            SysRole.tenant_no == tenant_no,
        )
        .all()
    )
    if not roles:
        return []

    result: List[dict] = []
    role_ids = [r.id for r in roles]

    # 批量查所有自定义权限明细
    all_details = (
        db.query(SysRoleDataScopeDetail)
        .filter(
            SysRoleDataScopeDetail.role_id.in_(role_ids),
            SysRoleDataScopeDetail.del_flag == 0,
            SysRoleDataScopeDetail.tenant_no == tenant_no,
        )
        .all()
    )
    details_by_role: dict[str, list] = {}
    for d in all_details:
        details_by_role.setdefault(d.role_id, []).append(d)

    for role in roles:
        result.append(
            {
                "data_scope": role.data_scope or "1",
                "details": details_by_role.get(role.id, []),
            }
        )
    return result


def apply_data_scope(query, model, user_info: dict, db: Session, tenant_no: str = None):
    """根据用户角色的数据权限过滤查询。

    Args:
        query: SQLAlchemy query 对象
        model: 查询的主模型类（需要有 create_dept_id / create_by 字段）
        user_info: 当前用户信息字典，需含 role / user_id / dept_id 等
        db: 数据库 Session
        tenant_no: 租户编号，默认使用 settings.DEFAULT_TENANT_NO

    Returns:
        应用数据权限过滤后的 query
    """
    # 1. 无用户信息 → 不过滤
    if not user_info:
        return query

    # 2. super_admin 直接放行
    if user_info.get("role") == "super_admin":
        return query

    # 3. 获取 user_id
    user_id = user_info.get("user_id")
    if not user_id:
        return query

    # 4. 确定 tenant_no
    effective_tenant = tenant_no or settings.DEFAULT_TENANT_NO

    # 5. 检查 model 是否有可过滤字段
    has_dept = hasattr(model, "create_dept_id")
    has_user = hasattr(model, "create_by")
    if not has_dept and not has_user:
        return query

    # 6. 获取用户所有角色的数据权限
    role_scopes = _resolve_user_role_scopes(db, user_id, effective_tenant)
    if not role_scopes:
        return query.filter(False)

    # 7. 任一角色拥有全部数据权限 → 放行（union 语义）
    if any(rs["data_scope"] == "1" for rs in role_scopes):
        return query

    # 8. 收集所有角色的 dept_ids 和 user_ids（union 合并）
    dept_ids: Set[str] = set()
    user_ids: Set[str] = set()

    # 获取用户部门
    user_dept_id = user_info.get("dept_id") or user_info.get("region_id") or get_user_dept_id(db, user_id)

    for rs in role_scopes:
        scope = rs["data_scope"]

        if scope == "2":
            # 本部门
            if user_dept_id:
                dept_ids.add(user_dept_id)

        elif scope == "3":
            # 本部门及以下
            if user_dept_id:
                dept_ids.update(_get_child_dept_ids(db, user_dept_id, effective_tenant))

        elif scope == "4":
            # 仅本人
            user_ids.add(user_id)

        elif scope == "5":
            # 自定义
            for detail in rs["details"]:
                val = (detail.scope_value or "").strip()
                if not val:
                    continue
                if detail.scope_type == "dept":
                    dept_ids.add(val)
                elif detail.scope_type == "user":
                    user_ids.add(val)

    # 9. 构建 OR 条件并应用
    conditions = []
    if dept_ids and has_dept:
        conditions.append(model.create_dept_id.in_(dept_ids))
    if user_ids and has_user:
        conditions.append(model.create_by.in_(user_ids))

    if not conditions:
        # 没有任何可匹配条件 → 返回空结果
        return query.filter(False)

    return query.filter(or_(*conditions))