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, Optional, Set, Tuple

from fastapi import HTTPException
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 _collect_scope_ids(
    db: Session,
    user_info: dict,
    user_id: str,
    effective_tenant: str,
    role_scopes: List[dict],
) -> Tuple[Set[str], Set[str]]:
    """合并多角色数据范围为 dept_ids / user_ids 并集。"""
    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)
    return dept_ids, user_ids


def _append_user_scope_conditions(
    conditions: list,
    model,
    user_ids: Set[str],
    *,
    has_user: bool,
    owner_field: Optional[str],
) -> None:
    if not user_ids:
        return
    user_conds = []
    if has_user:
        user_conds.append(model.create_by.in_(user_ids))
    if owner_field and hasattr(model, owner_field):
        user_conds.append(getattr(model, owner_field).in_(user_ids))
    if user_conds:
        conditions.append(or_(*user_conds))


def apply_data_scope(
    query,
    model,
    user_info: dict,
    db: Session,
    tenant_no: str = None,
    owner_field: Optional[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
        owner_field: 可选负责人字段名（如 CRM 的 owner_user_id），与 create_by 取 OR

    Returns:
        应用数据权限过滤后的 query
    """
    if not user_info:
        return query

    if user_info.get("role") == "super_admin":
        return query

    user_id = user_info.get("user_id")
    if not user_id:
        return query

    effective_tenant = tenant_no or settings.DEFAULT_TENANT_NO

    has_dept = hasattr(model, "create_dept_id")
    has_user = hasattr(model, "create_by")
    if not has_dept and not has_user:
        return query

    role_scopes = _resolve_user_role_scopes(db, user_id, effective_tenant)
    if not role_scopes:
        return query.filter(False)

    if any(rs["data_scope"] == "1" for rs in role_scopes):
        return query

    dept_ids, user_ids = _collect_scope_ids(db, user_info, user_id, effective_tenant, role_scopes)

    conditions = []
    if dept_ids and has_dept:
        conditions.append(model.create_dept_id.in_(dept_ids))
    _append_user_scope_conditions(
        conditions, model, user_ids, has_user=has_user, owner_field=owner_field
    )

    if not conditions:
        return query.filter(False)

    return query.filter(or_(*conditions))


def row_in_data_scope(
    db: Session,
    row,
    user_info: dict,
    tenant_no: Optional[str] = None,
    owner_field: Optional[str] = None,
) -> bool:
    """判断单行数据是否在当前用户数据权限范围内。"""
    if not user_info or row is None:
        return False
    if user_info.get("role") == "super_admin":
        return True

    user_id = user_info.get("user_id")
    if not user_id:
        return False

    effective_tenant = tenant_no or settings.DEFAULT_TENANT_NO
    role_scopes = _resolve_user_role_scopes(db, user_id, effective_tenant)
    if not role_scopes:
        return False
    if any(rs["data_scope"] == "1" for rs in role_scopes):
        return True

    dept_ids, user_ids = _collect_scope_ids(db, user_info, user_id, effective_tenant, role_scopes)

    row_dept = (getattr(row, "create_dept_id", None) or "").strip()
    row_creator = (getattr(row, "create_by", None) or "").strip()
    row_owner = (
        (getattr(row, owner_field, None) or "").strip() if owner_field else ""
    )

    if dept_ids and row_dept and row_dept in dept_ids:
        return True
    if user_ids and row_creator and row_creator in user_ids:
        return True
    if user_ids and row_owner and row_owner in user_ids:
        return True
    return False


def assert_row_in_data_scope(
    db: Session,
    row,
    _model,
    user_info: dict,
    tenant_no: Optional[str] = None,
    owner_field: Optional[str] = None,
) -> None:
    """不在数据权限范围内时抛出 403。"""
    if not row_in_data_scope(db, row, user_info, tenant_no, owner_field=owner_field):
        raise HTTPException(status_code=403, detail="无权访问该数据")