shaguabijia-app-server/app/admin/routers/roles.py

"""admin RBAC 角色管理(仅 super_admin):列角色 / 权限目录 / 增删改角色。均写审计。

角色 = 一组「可见页面」(见 app/admin/permissions.py)。角色登录后台后左侧只展示其 pages 对应导航项。
只有 super_admin(内建全权角色)能进本组端点(dependencies=require_role() 无参 = 仅 super_admin)。
"""
from __future__ import annotations

from fastapi import APIRouter, Depends, HTTPException, Request
from sqlalchemy import func, select

from app.admin.audit import write_audit
from app.admin.deps import AdminDb, CurrentAdmin, get_client_ip, require_role
from app.admin.permissions import (
    PERMISSION_CATALOG,
    SUPER_ADMIN_ROLE,
    effective_pages,
    sanitize_pages,
)
from app.admin.repositories import admin_role as role_repo
from app.admin.schemas.role import (
    PermissionGroup,
    RoleCreateRequest,
    RoleOut,
    RoleUpdateRequest,
)
from app.models.admin import AdminUser
from app.models.admin_role import AdminRole

router = APIRouter(
    prefix="/admin/api/roles",
    tags=["admin-roles"],
    dependencies=[Depends(require_role())],  # 无参 = 仅 super_admin
)


def _usage_counts(db: AdminDb) -> dict[str, int]:
    rows = db.execute(
        select(AdminUser.role, func.count(AdminUser.id)).group_by(AdminUser.role)
    ).all()
    return {r: c for r, c in rows}


def _to_out(role: AdminRole, usage: dict[str, int]) -> RoleOut:
    return RoleOut(
        id=role.id,
        name=role.name,
        label=role.label or role.name,
        pages=effective_pages(role.name, role.pages),
        is_builtin=role.is_builtin,
        in_use=usage.get(role.name, 0),
    )


@router.get("", response_model=list[RoleOut], summary="角色列表(含可见页 + 使用数)")
def list_roles(db: AdminDb) -> list[RoleOut]:
    usage = _usage_counts(db)
    return [_to_out(r, usage) for r in role_repo.list_roles(db)]


@router.get("/catalog", response_model=list[PermissionGroup], summary="页面权限目录(分组)")
def get_catalog() -> list[PermissionGroup]:
    return [PermissionGroup(**g) for g in PERMISSION_CATALOG]


@router.post("", response_model=RoleOut, summary="新增角色(带审计)")
def create_role(
    body: RoleCreateRequest, request: Request, admin: CurrentAdmin, db: AdminDb
) -> RoleOut:
    name = body.name.strip()
    if not name:
        raise HTTPException(status_code=400, detail="角色名称不能为空")
    if name == SUPER_ADMIN_ROLE:
        raise HTTPException(status_code=400, detail="super_admin 为内建角色,不能新建")
    if role_repo.get_role(db, name) is not None:
        raise HTTPException(status_code=409, detail="角色名称已存在")
    # 自定义角色:name(key)= label = 输入名称
    role = role_repo.create_role(db, name=name, label=name, pages=body.pages)
    write_audit(
        db, admin, action="role.create", target_type="role", target_id=str(role.id),
        detail={"name": name, "pages": role.pages}, ip=get_client_ip(request), commit=True,
    )
    return _to_out(role, _usage_counts(db))


@router.patch("/{role_id}", response_model=RoleOut, summary="改角色(展示名/可见页,带审计)")
def update_role(
    role_id: int, body: RoleUpdateRequest, request: Request, admin: CurrentAdmin, db: AdminDb
) -> RoleOut:
    role = db.get(AdminRole, role_id)
    if role is None:
        raise HTTPException(status_code=404, detail="角色不存在")
    if role.is_builtin:
        raise HTTPException(status_code=400, detail="内建角色「管理员」不可编辑")

    new_label = body.label.strip() if body.label is not None else None
    changes: dict = {}
    if new_label and new_label != role.label:
        changes["label"] = {"before": role.label, "after": new_label}
    if body.pages is not None:
        changes["pages"] = {"before": role.pages, "after": sanitize_pages(body.pages)}
    if not changes:
        raise HTTPException(status_code=400, detail="无任何变更字段")

    # 只改展示名 + 可见页;key(name)不可变,故无需级联 admin_user.role
    role_repo.update_role(db, role, label=new_label, pages=body.pages)
    write_audit(
        db, admin, action="role.update", target_type="role", target_id=str(role_id),
        detail=changes, ip=get_client_ip(request), commit=True,
    )
    return _to_out(role, _usage_counts(db))


@router.delete("/{role_id}", summary="删角色(带审计;内建/在用不可删)")
def delete_role(role_id: int, request: Request, admin: CurrentAdmin, db: AdminDb) -> dict:
    role = db.get(AdminRole, role_id)
    if role is None:
        raise HTTPException(status_code=404, detail="角色不存在")
    if role.is_builtin:
        raise HTTPException(status_code=400, detail="内建角色不可删除")
    used = _usage_counts(db).get(role.name, 0)
    if used > 0:
        raise HTTPException(status_code=400, detail=f"该角色仍有 {used} 名管理员在用,请先改派再删")
    name = role.name
    role_repo.delete_role(db, role)
    write_audit(
        db, admin, action="role.delete", target_type="role", target_id=str(role_id),
        detail={"name": name}, ip=get_client_ip(request), commit=True,
    )
    return {"deleted": True}