"""数据权限过滤（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, ): """根据用户角色的数据权限过滤查询。 适用于业务单据列表/详情（报销单、客户、运单等）。 以下场景 SHALL NOT 调用本函数（仅租户隔离即可）： - 下拉/选项接口（/options） - 租户级主数据树与码表（部门、字典、费用项目、产品分类、民族等） - 已有专用归属过滤的接口（如发票夹按 owner_user_id、待办按审批人） 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="无权访问该数据")