dibang-xiaoshou/backend/app/api/routes/auth.py

from __future__ import annotations

from datetime import datetime, timedelta

import jwt
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from sqlmodel import select
from sqlmodel.ext.asyncio.session import AsyncSession

from app.core.config import get_settings
from app.db.session import get_async_session
from app.models.users import User
from pydantic import BaseModel, Field

router = APIRouter(prefix="/auth", tags=["auth"])
security = HTTPBearer(auto_error=False)
settings = get_settings()


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


class LoginResponse(BaseModel):
    accessToken: str
    refreshToken: str | None = None


class UserInfo(BaseModel):
    id: int
    username: str
    role: str
    name: str | None = None
    avatar: str | None = None
    email: str | None = None

    model_config = {"from_attributes": True}


def create_access_token(user_id: int, username: str) -> str:
    """创建访问令牌"""
    expire = datetime.utcnow() + timedelta(minutes=settings.access_token_expires_minutes)
    payload = {
        "sub": str(user_id),
        "username": username,
        "exp": expire,
        "iat": datetime.utcnow(),
    }
    token = jwt.encode(payload, settings.jwt_secret, algorithm="HS256")
    return token


def verify_password(plain_password: str, stored_password: str) -> bool:
    """验证密码（明文比较）"""
    return plain_password == stored_password


@router.post("/login", response_model=LoginResponse)
async def login(
    payload: LoginRequest,
    session: AsyncSession = Depends(get_async_session),
) -> LoginResponse:
    """用户登录"""
    # 查找用户
    result = await session.exec(
        select(User).where(
            User.username == payload.username, User.is_active.is_(True)
        )
    )
    user = result.one_or_none()

    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误",
        )

    # 验证密码（明文比较）
    if not verify_password(payload.password, user.password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误",
        )

    # 生成 token
    access_token = create_access_token(user.id, user.username)

    return LoginResponse(accessToken=access_token)


async def get_current_user(
    credentials: HTTPAuthorizationCredentials | None = Depends(security),
    session: AsyncSession = Depends(get_async_session),
) -> User:
    """从 token 获取当前用户"""
    if not credentials:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="未提供认证令牌",
        )

    token = credentials.credentials

    try:
        payload = jwt.decode(token, settings.jwt_secret, algorithms=["HS256"])
        user_id = int(payload.get("sub"))
    except jwt.ExpiredSignatureError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="令牌已过期",
        )
    except jwt.InvalidTokenError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="无效的令牌",
        )

    # 查找用户
    result = await session.exec(
        select(User).where(User.id == user_id, User.is_active.is_(True))
    )
    user = result.one_or_none()

    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户不存在或已被禁用",
        )

    return user


@router.get("/userinfo", response_model=UserInfo)
async def get_userinfo(
    current_user: User = Depends(get_current_user),
) -> UserInfo:
    """获取当前用户信息"""
    return UserInfo(
        id=current_user.id,
        username=current_user.username,
        role=current_user.role,
        name=current_user.username,  # 如果没有 name 字段，使用 username
        avatar=None,
        email=None,
    )