hunjing/backend/app/api/v1/endpoints/auth.py

"""认证 API。"""

from fastapi import APIRouter, Depends
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.api.v1.deps import get_current_user
from app.core.config import settings
from app.db.session import get_session
from app.models.rbac import Role, UserRole
from app.schemas.auth import (
    ChangePasswordRequest,
    CurrentUser,
    LoginRequest,
    LoginResponse,
    RefreshRequest,
    RegisterRequest,
    UpdateProfileRequest,
    UserResponse,
)
from app.services import auth as auth_service
from app.services.auth import get_user_permissions

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


def _jwt_enabled() -> None:
    if not settings.jwt_secret:
        from fastapi import HTTPException

        raise HTTPException(
            status_code=503,
            detail={
                "error": {
                    "code": "service_unavailable",
                    "message": "认证服务未配置（JWT_SECRET 为空）。",
                }
            },
        )


@router.post("/register", response_model=UserResponse)
async def register(
    payload: RegisterRequest,
    db: AsyncSession = Depends(get_session),
) -> UserResponse:
    _jwt_enabled()
    user = await auth_service.create_user(
        db,
        username=payload.username,
        email=payload.email,
        password=payload.password,
    )
    # 为新用户创建默认角色和默认人物
    from app.services.characters import ensure_default_character_for_owner

    await ensure_default_character_for_owner(db, user.id)
    # 分配普通用户角色
    role_result = await db.execute(select(Role).where(Role.name == "user"))
    user_role = role_result.scalar_one_or_none()
    if user_role:
        db.add(UserRole(user_id=user.id, role_id=user_role.id))
    await db.commit()
    await db.refresh(user)
    return UserResponse(
        id=user.id,
        username=user.username,
        email=user.email,
        is_admin=user.is_admin,
        created_at=user.created_at.isoformat(),
    )


@router.post("/login", response_model=LoginResponse)
async def login(
    payload: LoginRequest,
    db: AsyncSession = Depends(get_session),
) -> LoginResponse:
    _jwt_enabled()
    user = await auth_service.authenticate_user(db, payload.username, payload.password)
    access_token = auth_service.create_access_token(
        subject=user.id,
        extra={"owner_scope": user.id},
    )
    refresh_token = auth_service.create_refresh_token(subject=user.id)
    perms = await get_user_permissions(db, user.id)
    return LoginResponse(
        access_token=access_token,
        refresh_token=refresh_token,
        token_type="bearer",
        expires_in=settings.access_token_ttl,
        user=UserResponse(
            id=user.id,
            username=user.username,
            email=user.email,
            is_admin=user.is_admin,
            permissions=perms,
            created_at=user.created_at.isoformat(),
        ),
    )


@router.post("/refresh", response_model=LoginResponse)
async def refresh(
    payload: RefreshRequest,
    db: AsyncSession = Depends(get_session),
) -> LoginResponse:
    _jwt_enabled()
    data = auth_service.decode_token(payload.refresh_token)
    if data.get("type") != "refresh":
        from fastapi import HTTPException

        raise HTTPException(
            status_code=401,
            detail={"error": {"code": "unauthorized", "message": "Token 类型不匹配。"}},
        )
    user = await auth_service.get_user_by_id(db, data["sub"])
    if user is None or not user.is_active:
        from fastapi import HTTPException

        raise HTTPException(
            status_code=401,
            detail={"error": {"code": "unauthorized", "message": "用户不存在或已禁用。"}},
        )
    new_access = auth_service.create_access_token(
        subject=user.id,
        extra={"owner_scope": user.id},
    )
    new_refresh = auth_service.create_refresh_token(subject=user.id)
    perms = await get_user_permissions(db, user.id)
    return LoginResponse(
        access_token=new_access,
        refresh_token=new_refresh,
        token_type="bearer",
        expires_in=settings.access_token_ttl,
        user=UserResponse(
            id=user.id,
            username=user.username,
            email=user.email,
            is_admin=user.is_admin,
            permissions=perms,
            created_at=user.created_at.isoformat(),
        ),
    )


@router.get("/me", response_model=UserResponse)
async def get_me(
    current_user: CurrentUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_session),
) -> UserResponse:
    perms = await get_user_permissions(db, current_user.user_id)
    return UserResponse(
        id=current_user.user_id,
        username=current_user.username,
        email=current_user.email,
        is_admin=current_user.is_admin,
        permissions=perms,
        created_at="",
    )


@router.patch("/me", response_model=UserResponse)
async def update_profile(
    payload: UpdateProfileRequest,
    current_user: CurrentUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_session),
) -> UserResponse:
    """更新个人资料（当前仅支持邮箱）。"""
    if payload.email is not None:
        from sqlalchemy import select as _select

        from app.models.user import User

        result = await db.execute(_select(User).where(User.email == payload.email))
        existing = result.scalar_one_or_none()
        if existing is not None and existing.id != current_user.user_id:
            from app.core.errors import AppError

            raise AppError("conflict", "该邮箱已被其他用户使用。")
        user = await db.get(User, current_user.user_id)
        if user is not None:
            user.email = payload.email
            await db.commit()
    return await get_me(current_user, db)


@router.post("/me/change-password", status_code=204)
async def change_password(
    payload: ChangePasswordRequest,
    current_user: CurrentUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_session),
) -> None:
    """修改当前用户密码。"""
    from app.models.user import User

    user = await db.get(User, current_user.user_id)
    if user is None:
        from app.core.errors import AppError

        raise AppError("not_found", "用户不存在。")
    if not auth_service.verify_password(payload.old_password, user.password_hash):
        from app.core.errors import AppError

        raise AppError("unauthorized", "原密码错误。")
    user.password_hash = auth_service.hash_password(payload.new_password)
    await db.commit()