yu-zhi-ran/platform/backend/api/admin.py

# 宇之然内容创作平台 - 管理员API

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List

from ..core.security import get_current_admin_user
from ..database import get_db
from ..models import User, AuditLog

router = APIRouter()

@router.get("/users", response_model=List[dict])
async def get_users(
    current_user: User = Depends(get_current_admin_user),
    db: Session = Depends(get_db)
):
    """获取用户列表（管理员功能）"""
    
    users = db.query(User).all()
    result = []
    for user in users:
        result.append({
            "id": user.id,
            "username": user.username,
            "role": user.role,
            "created_at": user.created_at.isoformat() if user.created_at else None,
            "last_login": user.last_login.isoformat() if user.last_login else None
        })
    
    return result

@router.post("/users", response_model=dict)
async def create_user(
    username: str,
    password: str,
    role: str = "user",
    current_user: User = Depends(get_current_admin_user),
    db: Session = Depends(get_db)
):
    """创建新用户（管理员功能）"""
    
    # 检查用户名是否已存在
    existing_user = db.query(User).filter(User.username == username).first()
    if existing_user:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="用户名已存在"
        )
    
    # 验证角色
    if role not in ["admin", "user"]:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="角色必须是 admin 或 user"
        )
    
    # 导入密码哈希函数
    from ..core.security import get_password_hash
    
    # 创建新用户
    new_user = User(
        username=username,
        password_hash=get_password_hash(password),
        role=role
    )
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    
    # 记录审计日志
    from ..core.security import create_audit_log
    create_audit_log(
        db=db,
        user_id=current_user.id,
        action="create_user",
        resource_type="user",
        resource_id=new_user.id,
        details=f"角色: {role}"
    )
    
    return {
        "id": new_user.id,
        "username": new_user.username,
        "role": new_user.role,
        "created_at": new_user.created_at.isoformat() if new_user.created_at else None
    }

@router.put("/users/{user_id}", response_model=dict)
async def update_user(
    user_id: int,
    username: str = None,
    role: str = None,
    current_user: User = Depends(get_current_admin_user),
    db: Session = Depends(get_db)
):
    """更新用户信息（管理员功能）"""
    
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    
    updates = {}
    
    if username is not None:
        # 检查新用户名是否已被使用（除了当前用户）
        existing = db.query(User).filter(
            User.username == username,
            User.id != user_id
        ).first()
        if existing:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="用户名已被使用"
            )
        user.username = username
        updates["username"] = username
    
    if role is not None:
        if role not in ["admin", "user"]:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="角色必须是 admin 或 user"
            )
        user.role = role
        updates["role"] = role
    
    if updates:
        db.commit()
        
        # 记录审计日志
        from ..core.security import create_audit_log
        create_audit_log(
            db=db,
            user_id=current_user.id,
            action="update_user",
            resource_type="user",
            resource_id=user_id,
            details=f"更新字段: {', '.join(updates.keys())}"
        )
    
    return {
        "id": user.id,
        "username": user.username,
        "role": user.role,
        "updated_at": datetime.utcnow().isoformat()
    }

@router.delete("/users/{user_id}")
async def delete_user(
    user_id: int,
    current_user: User = Depends(get_current_admin_user),
    db: Session = Depends(get_db)
):
    """删除用户（管理员功能）"""
    
    # 不能删除自己
    if user_id == current_user.id:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="不能删除自己的账户"
        )
    
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    
    db.delete(user)
    db.commit()
    
    # 记录审计日志
    from ..core.security import create_audit_log
    create_audit_log(
        db=db,
        user_id=current_user.id,
        action="delete_user",
        resource_type="user",
        resource_id=user_id,
        details="用户账户已删除"
    )
    
    return {"message": "用户已成功删除"}