"""admin 账号管理(仅 super_admin):列表 / 创建 / 改角色启停重置密码。均写审计。""" from __future__ import annotations from fastapi import APIRouter, Depends, HTTPException, Request from app.admin.audit import write_audit from app.admin.deps import AdminDb, CurrentAdmin, get_client_ip, require_role from app.admin.permissions import CUSTOM_ROLE, SUPER_ADMIN_ROLE, sanitize_pages from app.admin.repositories import admin_role as role_repo from app.admin.repositories import admin_user as admin_repo from app.admin.schemas.admin import AdminCreateRequest, AdminUpdateRequest from app.admin.schemas.auth import AdminOut from app.core.security import hash_password router = APIRouter( prefix="/admin/api/admins", tags=["admin-accounts"], dependencies=[Depends(require_role())], # require_role() 无参 = 仅 super_admin 通过 ) def _active_super_count(db: AdminDb) -> int: return sum( 1 for a in admin_repo.list_admins(db) if a.role == "super_admin" and a.status == "active" ) def _validate_role(db: AdminDb, role: str) -> None: """角色必须是 super_admin / custom(自定义) / admin_role 表里已存在的角色,否则 400。""" if role in (SUPER_ADMIN_ROLE, CUSTOM_ROLE): return role_repo.ensure_builtin_roles(db) # 空表(测试/全新库)兜底播种,再校验 if role_repo.get_role(db, role) is None: raise HTTPException(status_code=400, detail=f"角色不存在: {role}") def _clean_override(role: str, pages_override: list[str] | None) -> list[str] | None: """按最终角色算出该存的 pages_override:custom → 勾选集(过滤非法 key,可空列表); 非 custom → None(切回普通角色即清空自定义页)。""" if role == CUSTOM_ROLE: return sanitize_pages(pages_override) return None @router.get("", response_model=list[AdminOut], summary="管理员列表(含明文密码,super_admin 专属)") def list_admins(db: AdminDb) -> list[AdminOut]: out: list[AdminOut] = [] for a in admin_repo.list_admins(db): item = AdminOut.model_validate(a) item.password = a.plain_password # 明文:仅本 super_admin 专属路由下发,供权限管理页复看 out.append(item) return out @router.post("", response_model=AdminOut, summary="创建管理员") def create_admin( body: AdminCreateRequest, request: Request, admin: CurrentAdmin, db: AdminDb ) -> AdminOut: if admin_repo.get_by_username(db, body.username) is not None: raise HTTPException(status_code=409, detail="用户名已存在") _validate_role(db, body.role) override = _clean_override(body.role, body.pages_override) new = admin_repo.create_admin( db, username=body.username, password=body.password, role=body.role, plain_password=body.password, # UI 建的账号留存明文,供权限管理页复看 pages_override=override, ) write_audit( db, admin, action="admin.create", target_type="admin", target_id=new.id, detail={"username": new.username, "role": new.role, "pages_override": override}, ip=get_client_ip(request), commit=True, ) return AdminOut.model_validate(new) @router.patch("/{admin_id}", response_model=AdminOut, summary="改角色/启停/重置密码") def update_admin( admin_id: int, body: AdminUpdateRequest, request: Request, admin: CurrentAdmin, db: AdminDb ) -> AdminOut: target = admin_repo.get_by_id(db, admin_id) if target is None: raise HTTPException(status_code=404, detail="管理员不存在") if admin_id == admin.id and body.status == "disabled": raise HTTPException(status_code=400, detail="不能禁用自己") # 防自锁:降级 / 禁用某个 super_admin 前,确认操作后仍至少剩 1 个 active super_admin, # 否则会进入「零可用超管」死局——本路由仅 super 可进,只能改库恢复。 demotes_super = ( target.role == "super_admin" and target.status == "active" and ( (body.role is not None and body.role != "super_admin") or body.status == "disabled" ) ) if demotes_super and _active_super_count(db) <= 1: raise HTTPException(status_code=400, detail="不能降级/禁用最后一个超级管理员") if body.role is not None: _validate_role(db, body.role) changes: dict = {} role_changed = body.role is not None and body.role != target.role if role_changed: changes["role"] = {"before": target.role, "after": body.role} target.role = body.role if body.status is not None and body.status != target.status: changes["status"] = {"before": target.status, "after": body.status} target.status = body.status if body.password is not None: changes["password"] = "reset" target.password_hash = hash_password(body.password) target.plain_password = body.password # 同步留存明文,权限管理页复看保持一致 # 自定义可见页:按「最终角色」(target.role,已应用完角色变更)决定该存什么。 # - 切到/维持 custom 且传了 pages_override → 用勾选集;切到 custom 没传 → 清成空列表。 # - 切回普通角色 → 清空 override(_clean_override 返回 None)。 # - 角色没变、只传 pages_override(编辑现有 custom 用户的勾选)→ 也更新。 if role_changed or body.pages_override is not None: new_override = _clean_override(target.role, body.pages_override) if new_override != target.pages_override: changes["pages_override"] = {"before": target.pages_override, "after": new_override} target.pages_override = new_override if not changes: raise HTTPException(status_code=400, detail="无任何变更字段") db.commit() db.refresh(target) write_audit( db, admin, action="admin.update", target_type="admin", target_id=admin_id, detail=changes, ip=get_client_ip(request), commit=True, ) return AdminOut.model_validate(target) @router.delete("/{admin_id}", summary="删除管理员(带审计)") def delete_admin(admin_id: int, request: Request, admin: CurrentAdmin, db: AdminDb) -> dict: target = admin_repo.get_by_id(db, admin_id) if target is None: raise HTTPException(status_code=404, detail="管理员不存在") if admin_id == admin.id: raise HTTPException(status_code=400, detail="不能删除自己") # 防自锁:删掉某个 active super_admin 前,确认后仍至少剩 1 个,否则进「零可用超管」死局。 if ( target.role == "super_admin" and target.status == "active" and _active_super_count(db) <= 1 ): raise HTTPException(status_code=400, detail="不能删除最后一个超级管理员") username = target.username db.delete(target) db.commit() write_audit( db, admin, action="admin.delete", target_type="admin", target_id=admin_id, detail={"username": username, "role": target.role}, ip=get_client_ip(request), commit=True, ) return {"deleted": True}