geMoldInsight/src/api/auth_routes.py

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.ext.asyncio import AsyncSession
from pydantic import BaseModel, EmailStr
from typing import Optional
from datetime import timedelta

from database.database import get_db_session
from services.auth_service import (
    authenticate_user,
    create_access_token,
    create_user,
    get_user_by_username,
    get_user_by_email,
    get_current_active_user,
    get_current_admin_user
)
from models.database import User
from config.settings import settings

router = APIRouter(prefix="/api/auth", tags=["认证"])


class UserCreate(BaseModel):
    username: str
    email: EmailStr
    password: str
    full_name: Optional[str] = None


class UserResponse(BaseModel):
    id: int
    username: str
    email: str
    full_name: Optional[str]
    is_active: bool
    is_superuser: bool

    class Config:
        from_attributes = True


class Token(BaseModel):
    access_token: str
    token_type: str
    user: UserResponse


class LoginRequest(BaseModel):
    username: str
    password: str


@router.post("/login", response_model=Token)
async def login(
    form_data: OAuth2PasswordRequestForm = Depends(),
    db_session: AsyncSession = Depends(get_db_session)
):
    user = await authenticate_user(db_session, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误",
            headers={"WWW-Authenticate": "Bearer"},
        )
    
    access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    
    return Token(
        access_token=access_token,
        token_type="bearer",
        user=UserResponse.from_orm(user)
    )


@router.post("/login/json", response_model=Token)
async def login_json(
    login_data: LoginRequest,
    db_session: AsyncSession = Depends(get_db_session)
):
    user = await authenticate_user(db_session, login_data.username, login_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误",
        )
    
    access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    
    return Token(
        access_token=access_token,
        token_type="bearer",
        user=UserResponse.from_orm(user)
    )


@router.post("/register", response_model=UserResponse, status_code=201)
async def register(
    user_data: UserCreate,
    db_session: AsyncSession = Depends(get_db_session)
):
    existing_user = await get_user_by_username(db_session, user_data.username)
    if existing_user:
        raise HTTPException(status_code=400, detail="用户名已存在")
    
    existing_email = await get_user_by_email(db_session, user_data.email)
    if existing_email:
        raise HTTPException(status_code=400, detail="邮箱已被注册")
    
    user = await create_user(
        db_session=db_session,
        username=user_data.username,
        email=user_data.email,
        password=user_data.password,
        full_name=user_data.full_name
    )
    
    return UserResponse.from_orm(user)


@router.get("/me", response_model=UserResponse)
async def get_current_user_info(
    current_user: User = Depends(get_current_active_user)
):
    return UserResponse.from_orm(current_user)


@router.post("/logout")
async def logout():
    return {"message": "已登出"}


@router.get("/users", response_model=list[UserResponse])
async def list_users(
    db_session: AsyncSession = Depends(get_db_session),
    admin_user: User = Depends(get_current_admin_user)
):
    from sqlalchemy import select
    result = await db_session.execute(select(User))
    users = result.scalars().all()
    return [UserResponse.from_orm(u) for u in users]


@router.put("/users/{user_id}/toggle-active", response_model=UserResponse)
async def toggle_user_active(
    user_id: int,
    db_session: AsyncSession = Depends(get_db_session),
    admin_user: User = Depends(get_current_admin_user)
):
    from sqlalchemy import select
    result = await db_session.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()
    
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    
    if user.id == admin_user.id:
        raise HTTPException(status_code=400, detail="不能禁用自己的账户")
    
    user.is_active = not user.is_active
    await db_session.commit()
    await db_session.refresh(user)
    
    return UserResponse.from_orm(user)


@router.put("/users/{user_id}/toggle-admin", response_model=UserResponse)
async def toggle_user_admin(
    user_id: int,
    db_session: AsyncSession = Depends(get_db_session),
    admin_user: User = Depends(get_current_admin_user)
):
    from sqlalchemy import select
    result = await db_session.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()
    
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    
    if user.id == admin_user.id:
        raise HTTPException(status_code=400, detail="不能修改自己的管理员权限")
    
    user.is_superuser = not user.is_superuser
    await db_session.commit()
    await db_session.refresh(user)
    
    return UserResponse.from_orm(user)