"""业务分组 API —— 全员可见，操作按角色分层 权限模型（2026-09 调整：从「仅超管」放开为分层，但**操作边界必须收窄**）： | 谁 | 看 | 改 | |---------------------|------------------------|--------------------------------------| | SUPER_ADMIN | 所有组 | 全部（建组/改组/删组/配范围/管成员） | | 主管(SUPERVISOR) | 自己所属的组 | **仅本组成员**（加人/移人/设组长） | | 组长(is_leader) | 自己所属的组 | **仅本组成员** | | 普通成员 | 自己所属的组 | 无（只读） | ⚠️ **为什么「主管」不能改可见范围、不能建删组**： 按数据范围规则，被分进组的 SUPERVISOR 会从「全厂」降级为只看本组。 若允许他改范围，他只要把自己那组改成「生产 + 售后」就恢复全厂视野； 若能建组，他新建一个组再把自己塞进去，同样绕过。**能给自己加组 = 提权**。 所以「管成员」与「配范围」必须分开：前者安全（组长给自己加组会被唯一约束 挡住，移出自己只是失去权限），后者是提权入口。 写操作会被 audit_middleware 自动采集 —— 分组变更是高权限动作，追责必须有据。 """ from __future__ import annotations from fastapi import APIRouter, Depends, HTTPException, Query, status from sqlalchemy import delete, func, select from sqlalchemy.ext.asyncio import AsyncSession from app.core.config import settings from app.core.database import get_db from app.core.lifecycle import PHASE_LABELS from app.core.mom_database import MomSessionLocal from app.core.roles import ADMIN_ROLES, SUPER_ADMIN from app.services.auth_service import get_current_user from app.models.business_group import ( BusinessGroup, BusinessGroupMember, BusinessGroupPhase, ) from app.schemas.group import ( GroupCreate, GroupDetailOut, GroupMemberAdd, GroupMemberOut, GroupMemberUpdate, GroupOut, GroupUpdate, MemberCandidate, PhaseOption, ) router = APIRouter(prefix="/groups", tags=["业务分组"]) # ============================================================ # 权限判定 # ============================================================ def _is_super(user: dict) -> bool: return (user or {}).get("role") == SUPER_ADMIN async def _my_group_ids(db: AsyncSession, user: dict) -> set[int]: """当前用户所属（且启用中）的组 id""" username = (user or {}).get("username") or "" if not username: return set() rows = await db.execute( select(BusinessGroupMember.group_id) .join(BusinessGroup, BusinessGroup.id == BusinessGroupMember.group_id) .where( BusinessGroupMember.user_id == username, BusinessGroup.is_active.is_(True), ) ) return {r[0] for r in rows.all()} async def _leader_group_ids(db: AsyncSession, user: dict) -> set[int]: """当前用户当组长的组 id""" username = (user or {}).get("username") or "" if not username: return set() rows = await db.execute( select(BusinessGroupMember.group_id) .join(BusinessGroup, BusinessGroup.id == BusinessGroupMember.group_id) .where( BusinessGroupMember.user_id == username, BusinessGroupMember.is_leader.is_(True), BusinessGroup.is_active.is_(True), ) ) return {r[0] for r in rows.all()} async def _can_manage_members(db: AsyncSession, user: dict, group_id: int) -> bool: """能否管理**这个组**的成员。 超管、部门主管(SUPERVISOR)、本组组长 —— 三者之一即可。 ⚠️ 组长只能管自己当组长的组；主管虽然能管所有组的成员，但**改不了范围** （范围相关端点单独限超管），所以不构成提权。 """ if _is_super(user): return True if (user or {}).get("role") in ADMIN_ROLES: return True return group_id in await _leader_group_ids(db, user) async def _is_super_admin_account(username: str) -> bool: """这个 MOM 账号是不是超管。 要在 MOM 侧查（Track 不存用户表）。查不到（账号不存在）时返回 False —— 交给后续流程处理，不在这里假装是超管。 """ if not username: return False db = MomSessionLocal() try: from sqlalchemy import text row = db.execute( text("SELECT role FROM sys_user WHERE username LIKE :p LIMIT 1"), {"p": f"%/{username}"}, ).fetchone() return bool(row) and (row[0] or "") == SUPER_ADMIN except Exception: # MOM 不可用时**不放行**为「不是超管」的反面 —— 这里失败意味着查不到身份， # 保守做法是当作超管拦下（误拦只是加不进去，误放会留下脏数据） return True finally: db.close() def _require_super(user: dict) -> None: """建组 / 改组 / 删组 / 配范围 —— 这些是提权入口，只给超管""" if not _is_super(user): raise HTTPException( status.HTTP_403_FORBIDDEN, "只有超级管理员可以新建、修改或删除分组，以及调整分组的可见范围", ) # ============================================================ # 内部工具 # ============================================================ def _valid_phases() -> dict[str, str]: """合法的 phase 取值 → 中文标签（单一事实来源是 core/lifecycle.py）""" return dict(PHASE_LABELS) async def _load_phases(db: AsyncSession, group_ids: list[int]) -> dict[int, list[str]]: """批量取这些组**自己配置**的可见范围（不含继承）""" if not group_ids: return {} rows = await db.execute( select(BusinessGroupPhase.group_id, BusinessGroupPhase.phase) .where(BusinessGroupPhase.group_id.in_(group_ids)) ) out: dict[int, list[str]] = {} for gid, ph in rows.all(): out.setdefault(gid, []).append(ph) return out async def _effective_phases( db: AsyncSession, group: BusinessGroup, own: dict[int, list[str]], ) -> list[str]: """实际生效的可见范围：自己配了就用，没配则向上取父组的。 继承让「生产大组配一次 PRODUCTION，下面的生产/测试小组都不用再配」成立。 """ mine = own.get(group.id) if mine: return sorted(mine) if group.parent_id: return sorted(own.get(group.parent_id, [])) return [] async def _member_counts(db: AsyncSession) -> dict[int, int]: rows = await db.execute( select(BusinessGroupMember.group_id, func.count()) .group_by(BusinessGroupMember.group_id) ) return {gid: cnt for gid, cnt in rows.all()} def _to_out(group: BusinessGroup, phases: list[str], count: int, parent_name: str | None, own_phases: list[str], *, can_manage_members: bool = False, can_manage_group: bool = False, is_my_leader: bool = False) -> GroupOut: """把 ORM 对象转成响应。 能力标记（能否管成员 / 能否改组）由服务端算好下发，前端不要自己按 role 推导 —— 组长身份是按组算的，前端凭 role 推不出来。 """ return GroupOut( id=group.id, name=group.name, parent_id=group.parent_id, parent_name=parent_name, description=group.description, sort_order=group.sort_order, is_active=group.is_active, phases=sorted(own_phases), effective_phases=phases, phase_labels=[PHASE_LABELS.get(p, p) for p in phases], member_count=count, can_manage_members=can_manage_members, can_manage_group=can_manage_group, is_my_leader=is_my_leader, ) async def _get_group_or_404(db: AsyncSession, group_id: int) -> BusinessGroup: group = await db.get(BusinessGroup, group_id) if not group: raise HTTPException(status.HTTP_404_NOT_FOUND, f"分组 {group_id} 不存在") return group # ============================================================ # 元数据：可选阶段 # 放在 /{group_id} 之前注册，否则 "phases" 会被当成 group_id 解析 # ============================================================ @router.get("/phase-options", response_model=list[PhaseOption]) async def list_phase_options(current_user: dict = Depends(get_current_user)): """可选的生命周期阶段 —— 供前端渲染勾选框，避免前端写死这两个值""" return [PhaseOption(value=v, label=l) for v, l in _valid_phases().items()] @router.get("/member-candidates", response_model=list[MemberCandidate]) def list_member_candidates( keyword: str = Query("", description="按姓名/账号模糊搜索"), limit: int = Query(500, ge=1, le=1000), current_user: dict = Depends(get_current_user), ): """候选人下拉 —— 复用与 users.py 一致的 MOM 查询口径（部门已钉死为 ORG_DEPARTMENT）。 ⚠️ **排除超级管理员**：超管的数据范围是硬编码的全厂（见 resolve_data_scope 规则 1），分组对他根本不生效。把超管放进分组只会造成两处困惑： · 界面上「他在这个组里」暗示他受这个组约束，但实际不受； · 若再给他打组长标记，会产生「组长却不受组范围限制」的矛盾状态。 所以既不显示在候选里，也不允许加入（add_member 会再挡一道）。 注意这里**不复用 users.py 的端点函数**：那个函数与 FastAPI 的 Query 默认值 耦合，直接调用拿到的是 Query 对象而非值。所以照抄同一条 SQL 的写法， 但部门条件取自同一处 settings.ORG_DEPARTMENT，口径不会漂移。 """ db = MomSessionLocal() try: base_sql = """ SELECT username, SPLIT_PART(username, '/', 1) AS full_name FROM sys_user WHERE department = :dept AND COALESCE(role, '') <> :super_admin """ params = {"dept": settings.ORG_DEPARTMENT, "lim": limit, "super_admin": SUPER_ADMIN} if keyword.strip(): sql_text = base_sql + " AND username ILIKE :kw ORDER BY username LIMIT :lim" params["kw"] = f"%{keyword.strip()}%" else: sql_text = base_sql + " ORDER BY username LIMIT :lim" from sqlalchemy import text rows = db.execute(text(sql_text), params).fetchall() return [ MemberCandidate( username=row.username.split("/")[-1] if "/" in row.username else row.username, full_name=row.full_name or row.username, ) for row in rows ] except Exception as e: raise HTTPException( status.HTTP_502_BAD_GATEWAY, f"MOM 用户查询失败: {str(e)}" ) finally: db.close() # ============================================================ # 组的 CRUD # ============================================================ @router.get("", response_model=list[GroupOut]) async def list_groups( db: AsyncSession = Depends(get_db), current_user: dict = Depends(get_current_user), ): """列出业务分组。 - 超管：全部（含停用的） - 其他人：**只返回自己所属的组** —— 普通成员不该看到别人的组有哪些人。 连"这个组存在但你没份"都不暴露，避免被拿来推测组织架构。 """ stmt = select(BusinessGroup).order_by(BusinessGroup.sort_order, BusinessGroup.id) if not _is_super(current_user): mine = await _my_group_ids(db, current_user) if not mine: return [] stmt = stmt.where(BusinessGroup.id.in_(tuple(mine))) groups = (await db.execute(stmt)).scalars().all() own = await _load_phases(db, [g.id for g in groups]) counts = await _member_counts(db) names = {g.id: g.name for g in groups} # 能力标记一次算好，避免在循环里反复查库 is_super = _is_super(current_user) is_admin_role = (current_user or {}).get("role") in ADMIN_ROLES leader_ids = set() if is_super else await _leader_group_ids(db, current_user) return [ _to_out( g, await _effective_phases(db, g, own), counts.get(g.id, 0), names.get(g.parent_id) if g.parent_id else None, own.get(g.id, []), can_manage_members=is_super or is_admin_role or g.id in leader_ids, can_manage_group=is_super, is_my_leader=g.id in leader_ids, ) for g in groups ] @router.get("/{group_id}", response_model=GroupDetailOut) async def get_group( group_id: int, db: AsyncSession = Depends(get_db), current_user: dict = Depends(get_current_user), ): """分组详情 + 成员列表（非超管只能看自己所属的组）""" group = await _get_group_or_404(db, group_id) if not _is_super(current_user): mine = await _my_group_ids(db, current_user) if group_id not in mine: raise HTTPException( status.HTTP_403_FORBIDDEN, "你不在该分组中，无法查看" ) own = await _load_phases(db, [group.id] + ([group.parent_id] if group.parent_id else [])) members = (await db.execute( select(BusinessGroupMember) .where(BusinessGroupMember.group_id == group.id) .order_by(BusinessGroupMember.is_leader.desc(), BusinessGroupMember.user_id) )).scalars().all() # 成员姓名走 MOM（带 2h TTL 缓存，见 mom_cache） from app.services.mom_cache import get_display_names name_map = get_display_names([m.user_id for m in members]) if members else {} parent_name = None if group.parent_id: parent = await db.get(BusinessGroup, group.parent_id) parent_name = parent.name if parent else None is_super = _is_super(current_user) is_admin_role = (current_user or {}).get("role") in ADMIN_ROLES leader_ids = set() if is_super else await _leader_group_ids(db, current_user) base = _to_out( group, await _effective_phases(db, group, own), len(members), parent_name, own.get(group.id, []), can_manage_members=is_super or is_admin_role or group.id in leader_ids, can_manage_group=is_super, is_my_leader=group.id in leader_ids, ) return GroupDetailOut( **base.model_dump(), members=[ GroupMemberOut( user_id=m.user_id, display_name=name_map.get(m.user_id) or m.user_id, is_leader=m.is_leader, ) for m in members ], ) @router.post("", response_model=GroupOut, status_code=status.HTTP_201_CREATED) async def create_group( payload: GroupCreate, db: AsyncSession = Depends(get_db), current_user: dict = Depends(get_current_user), ): """新建分组。parent_id 为空即建大组，否则是挂在某个大组下的小组。 ⚠️ 仅超管 —— 「能建组」意味着「能新建一个组再把自己塞进去」，是提权入口。 """ _require_super(current_user) valid = _valid_phases() bad = [p for p in payload.phases if p not in valid] if bad: raise HTTPException(status.HTTP_400_BAD_REQUEST, f"非法的阶段取值: {bad}") dup = await db.scalar(select(BusinessGroup.id).where(BusinessGroup.name == payload.name)) if dup: raise HTTPException(status.HTTP_409_CONFLICT, f"分组名「{payload.name}」已存在") if payload.parent_id is not None: parent = await db.get(BusinessGroup, payload.parent_id) if not parent: raise HTTPException(status.HTTP_400_BAD_REQUEST, "上级分组不存在") if parent.parent_id is not None: raise HTTPException(status.HTTP_400_BAD_REQUEST, "只支持两级：不能挂在子组下") group = BusinessGroup( name=payload.name, parent_id=payload.parent_id, description=payload.description, sort_order=payload.sort_order, ) db.add(group) await db.flush() for p in payload.phases: db.add(BusinessGroupPhase(group_id=group.id, phase=p)) await db.commit() await db.refresh(group) # 生效范围要算上继承 —— 否则新建子组时返回的 effective_phases 是空的， # 与紧接着的列表查询结果对不上（前端直接拿返回值渲染会闪一下「无范围」） own = await _load_phases(db, [group.id] + ([group.parent_id] if group.parent_id else [])) parent_name = None if group.parent_id: parent = await db.get(BusinessGroup, group.parent_id) parent_name = parent.name if parent else None return _to_out( group, await _effective_phases(db, group, own), 0, parent_name, own.get(group.id, []), ) @router.patch("/{group_id}", response_model=GroupOut) async def update_group( group_id: int, payload: GroupUpdate, db: AsyncSession = Depends(get_db), current_user: dict = Depends(get_current_user), ): """修改分组（含可见范围）。 ⚠️ **仅超管** —— 这是最关键的一条：`phases` 决定能看到哪些数据，谁能改它 谁就能给自己扩权（比如把本组改成「生产 + 售后」= 全厂）。 主管能管成员但不能碰范围，就是这个原因。 ⚠️ `is_active=false` 的语义是「该组所有成员**立即**退回未分组状态」—— 这是一次批量权限变更，前端必须二次确认后再调。 """ _require_super(current_user) group = await _get_group_or_404(db, group_id) if payload.name is not None and payload.name != group.name: dup = await db.scalar( select(BusinessGroup.id).where( BusinessGroup.name == payload.name, BusinessGroup.id != group_id, ) ) if dup: raise HTTPException(status.HTTP_409_CONFLICT, f"分组名「{payload.name}」已存在") group.name = payload.name if payload.parent_id is not None: if payload.parent_id == group_id: raise HTTPException(status.HTTP_400_BAD_REQUEST, "不能把自己设为自己的上级") parent = await db.get(BusinessGroup, payload.parent_id) if not parent: raise HTTPException(status.HTTP_400_BAD_REQUEST, "上级分组不存在") if parent.parent_id is not None: raise HTTPException(status.HTTP_400_BAD_REQUEST, "只支持两级：不能挂在子组下") group.parent_id = payload.parent_id if payload.description is not None: group.description = payload.description if payload.sort_order is not None: group.sort_order = payload.sort_order if payload.is_active is not None: group.is_active = payload.is_active # 可见范围：整组覆盖式更新 if payload.phases is not None: valid = _valid_phases() bad = [p for p in payload.phases if p not in valid] if bad: raise HTTPException(status.HTTP_400_BAD_REQUEST, f"非法的阶段取值: {bad}") await db.execute( delete(BusinessGroupPhase).where(BusinessGroupPhase.group_id == group_id) ) for p in payload.phases: db.add(BusinessGroupPhase(group_id=group_id, phase=p)) await db.commit() await db.refresh(group) own = await _load_phases(db, [group.id] + ([group.parent_id] if group.parent_id else [])) counts = await _member_counts(db) parent_name = None if group.parent_id: parent = await db.get(BusinessGroup, group.parent_id) parent_name = parent.name if parent else None return _to_out( group, await _effective_phases(db, group, own), counts.get(group.id, 0), parent_name, own.get(group.id, []), ) @router.delete("/{group_id}", status_code=status.HTTP_204_NO_CONTENT) async def delete_group( group_id: int, db: AsyncSession = Depends(get_db), current_user: dict = Depends(get_current_user), ): """删除分组（仅超管）。 ⚠️ **仅允许删空组**。有成员时返回 409，要求先移除成员或改为停用。 级联删除是一次**静默的批量权限变更** —— 误点一下，一批人就突然看不到 数据了。强制多走一步，出错时是可见的。 """ _require_super(current_user) group = await _get_group_or_404(db, group_id) member_count = await db.scalar( select(func.count()).select_from(BusinessGroupMember) .where(BusinessGroupMember.group_id == group_id) ) or 0 if member_count: raise HTTPException( status.HTTP_409_CONFLICT, f"该分组下还有 {member_count} 名成员。请先移除成员，或改为「停用」而不是删除。", ) children = await db.scalar( select(func.count()).select_from(BusinessGroup) .where(BusinessGroup.parent_id == group_id) ) or 0 if children: raise HTTPException( status.HTTP_409_CONFLICT, f"该分组下还有 {children} 个子组，请先处理子组。", ) await db.delete(group) await db.commit() # ============================================================ # 成员管理 # ============================================================ @router.post("/{group_id}/members", response_model=GroupMemberOut, status_code=status.HTTP_201_CREATED) async def add_member( group_id: int, payload: GroupMemberAdd, db: AsyncSession = Depends(get_db), current_user: dict = Depends(get_current_user), ): """把一个人加进分组。一个人可以在多个组 —— 多组 = 多看一个范围。 权限：超管 / 部门主管 / **本组组长**。组长给自己加组会被唯一约束挡住 （他本来就在组里），所以这里不构成提权。 """ await _get_group_or_404(db, group_id) if not await _can_manage_members(db, current_user, group_id): raise HTTPException( status.HTTP_403_FORBIDDEN, "只有超级管理员、主管或本组组长可以添加成员" ) # 超管不进分组：他的范围是硬编码全厂，分组对他不生效。 # 放进来的话界面会暗示「他受这个组约束」，与事实不符。 if await _is_super_admin_account(payload.user_id): raise HTTPException( status.HTTP_400_BAD_REQUEST, "超级管理员不受业务分组约束（数据范围恒为全厂），无需加入分组", ) exists = await db.scalar( select(BusinessGroupMember.id).where( BusinessGroupMember.group_id == group_id, BusinessGroupMember.user_id == payload.user_id, ) ) if exists: raise HTTPException(status.HTTP_409_CONFLICT, "该成员已在此分组中") member = BusinessGroupMember( group_id=group_id, user_id=payload.user_id, is_leader=payload.is_leader, ) db.add(member) await db.commit() from app.services.mom_cache import get_display_names name_map = get_display_names([payload.user_id]) return GroupMemberOut( user_id=payload.user_id, display_name=name_map.get(payload.user_id) or payload.user_id, is_leader=payload.is_leader, ) @router.patch("/{group_id}/members/{user_id}", response_model=GroupMemberOut) async def update_member( group_id: int, user_id: str, payload: GroupMemberUpdate, db: AsyncSession = Depends(get_db), current_user: dict = Depends(get_current_user), ): """设置/取消组长。组长数据范围与组员相同，额外能管理本组成员。""" if not await _can_manage_members(db, current_user, group_id): raise HTTPException( status.HTTP_403_FORBIDDEN, "只有超级管理员、主管或本组组长可以调整组长" ) member = (await db.execute( select(BusinessGroupMember).where( BusinessGroupMember.group_id == group_id, BusinessGroupMember.user_id == user_id, ) )).scalar_one_or_none() if not member: raise HTTPException(status.HTTP_404_NOT_FOUND, "该成员不在此分组中") member.is_leader = payload.is_leader await db.commit() from app.services.mom_cache import get_display_names name_map = get_display_names([user_id]) return GroupMemberOut( user_id=user_id, display_name=name_map.get(user_id) or user_id, is_leader=payload.is_leader, ) @router.delete("/{group_id}/members/{user_id}", status_code=status.HTTP_204_NO_CONTENT) async def remove_member( group_id: int, user_id: str, db: AsyncSession = Depends(get_db), current_user: dict = Depends(get_current_user), ): """把成员移出分组。若此人不再属于任何组，将退回「未分组」状态。""" if not await _can_manage_members(db, current_user, group_id): raise HTTPException( status.HTTP_403_FORBIDDEN, "只有超级管理员、主管或本组组长可以移除成员" ) result = await db.execute( delete(BusinessGroupMember).where( BusinessGroupMember.group_id == group_id, BusinessGroupMember.user_id == user_id, ) ) if result.rowcount == 0: raise HTTPException(status.HTTP_404_NOT_FOUND, "该成员不在此分组中") await db.commit()