LQAdminPlatform/src/views/auth_view.py

"""
认证API端点
"""
import sys
import os

# 添加src目录到Python路径
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '../..'))
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '../../..'))

from fastapi import APIRouter, Depends, HTTPException, Request, Response
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from sqlalchemy.ext.asyncio import AsyncSession
from typing import Optional
from app.base import get_db
from app.schemas.auth import (
    LoginRequest, 
    TokenResponse, 
    RefreshTokenRequest,
    LogoutRequest,
    UserInfoResponse,
    CaptchaResponse
)
from app.services.auth_service import AuthService
from app.core.exceptions import AuthenticationError, ValidationError
from app.schemas.base import ResponseSchema
import base64
import io
from PIL import Image, ImageDraw, ImageFont
import random
import string
import logging

# 配置日志记录器
logger = logging.getLogger(__name__)

router = APIRouter(prefix="/auth", tags=["登录认证"])
security = HTTPBearer()


def get_client_ip(request: Request) -> str:
    """获取客户端IP地址"""
    forwarded = request.headers.get("X-Forwarded-For")
    if forwarded:
        return forwarded.split(",")[0].strip()
    return request.client.host


def get_user_agent(request: Request) -> str:
    """获取用户代理"""
    return request.headers.get("User-Agent", "")


@router.post("/login", response_model=ResponseSchema)
async def login(
    request: Request,
    login_data: LoginRequest,
    db: AsyncSession = Depends(get_db)
):
    """用户登录"""
    try:
        logger.info(f"收到登录请求: username={login_data.username}")
        logger.debug(f"登录详情: username={login_data.username}, ip={get_client_ip(request)}")
        
        auth_service = AuthService(db)
        
        user, token_response = await auth_service.authenticate_user(
            username=login_data.username,
            password=login_data.password,
            ip_address=get_client_ip(request),
            user_agent=get_user_agent(request)
        )
        
        logger.info(f"登录成功: username={login_data.username}, user_id={user.id}")
        return ResponseSchema(
        code="000000",
            message="登录成功",
            data=token_response.dict()
        )
        
    except AuthenticationError as e:
        logger.warning(f"认证失败: username={login_data.username}, reason={e.message}")
        return ResponseSchema(
            code=e.code,
            message=e.message,
            data=None
        )
    except Exception as e:
        logger.error(f"登录错误: {type(e).__name__}: {str(e)}", exc_info=True)
        return ResponseSchema(
        code="500001",
            message=f"服务器内部错误: {str(e)}",
            data=None
        )


@router.post("/refresh", response_model=ResponseSchema)
async def refresh_token(
    refresh_data: RefreshTokenRequest,
    db: AsyncSession = Depends(get_db)
):
    """刷新访问令牌"""
    try:
        auth_service = AuthService(db)
        
        token_response = await auth_service.refresh_access_token(
            refresh_token=refresh_data.refresh_token
        )
        
        return ResponseSchema(
        code="000000",
            message="令牌刷新成功",
            data=token_response.dict()
        )
        
    except AuthenticationError as e:
        return ResponseSchema(
            code=e.code,
            message=e.message,
            data=None
        )
    except Exception as e:
        return ResponseSchema(
        code="500001",
            message="服务器内部错误",
            data=None
        )


@router.post("/logout", response_model=ResponseSchema)
async def logout(
    credentials: HTTPAuthorizationCredentials = Depends(security),
    db: AsyncSession = Depends(get_db),
    logout_data: Optional[LogoutRequest] = None
):
    """用户登出"""
    try:
        auth_service = AuthService(db)
        
        # 优先使用Authorization头中的token
        token = credentials.credentials if credentials else None
        refresh_token = None
        
        # 如果提供了logout_data，使用其中的token和refresh_token
        if logout_data:
            if not token and logout_data.token:
                token = logout_data.token
            if logout_data.refresh_token:
                refresh_token = logout_data.refresh_token
        
        await auth_service.logout(
            token=token,
            refresh_token=refresh_token
        )
        
        return ResponseSchema(
        code="000000",
            message="登出成功",
            data=None
        )
        
    except Exception as e:
        logger.exception("登出内部错误")
        return ResponseSchema(
        code="500001",
            message="服务器内部错误",
            data=None
        )


@router.get("/userinfo", response_model=ResponseSchema)
async def get_user_info(
    credentials: HTTPAuthorizationCredentials = Depends(security),
    db: AsyncSession = Depends(get_db)
):
    """获取用户信息"""
    try:
        auth_service = AuthService(db)
        
        # get_current_user 现在返回 (user, new_token) 元组
        user, new_token = await auth_service.get_current_user(credentials.credentials)
        logger.info(f"user={user.username}, new_token={'存在' if new_token else '无'}")
        logger.info(f"user.id={user.id}")
        
        user_info = await auth_service.get_user_info(user)
        
        response_data = ResponseSchema(
        code="000000",
            message="获取用户信息成功",
            data=user_info.dict()
        )
        
        # 如果token被刷新，添加到响应中
        if new_token:
            response_dict = response_data.dict()
            response_dict["token_refreshed"] = True
            response_dict["new_token"] = new_token
            logger.info(f"用户信息API - Token已刷新，用户: {user.username}")
            return response_dict
        
        return response_data
        
    except AuthenticationError as e:
        return ResponseSchema(
            code=e.code,
            message=e.message,
            data=None
        )
    except Exception as e:
        logger.exception("获取用户信息内部错误")
        return ResponseSchema(
        code="500001",
            message="服务器内部错误",
            data=None
        )

@router.get("/captcha", response_model=ResponseSchema)
async def get_captcha():
    """获取验证码"""
    try:
        # 生成随机验证码
        captcha_text = ''.join(random.choices(string.ascii_uppercase + string.digits, k=4))
        
        # 创建图片
        width, height = 120, 40
        image = Image.new('RGB', (width, height), color='white')
        draw = ImageDraw.Draw(image)
        
        # 绘制验证码文字
        try:
            # 尝试使用系统字体
            font = ImageFont.truetype("arial.ttf", 20)
        except:
            # 使用默认字体
            font = ImageFont.load_default()
        
        # 计算文字位置
        text_width = draw.textlength(captcha_text, font=font)
        text_height = 20
        x = (width - text_width) // 2
        y = (height - text_height) // 2
        
        # 绘制文字
        draw.text((x, y), captcha_text, fill='black', font=font)
        
        # 添加干扰线
        for _ in range(5):
            x1 = random.randint(0, width)
            y1 = random.randint(0, height)
            x2 = random.randint(0, width)
            y2 = random.randint(0, height)
            draw.line([(x1, y1), (x2, y2)], fill='gray', width=1)
        
        # 转换为base64
        buffer = io.BytesIO()
        image.save(buffer, format='PNG')
        image_base64 = base64.b64encode(buffer.getvalue()).decode()
        
        # 生成验证码ID（实际应用中应该存储到Redis等缓存中）
        captcha_id = ''.join(random.choices(string.ascii_letters + string.digits, k=32))
        
        captcha_response = CaptchaResponse(
            captcha_id=captcha_id,
            captcha_image=f"data:image/png;base64,{image_base64}"
        )
        
        return ResponseSchema(
        code="000000",
            message="获取验证码成功",
            data=captcha_response.dict()
        )
        
    except Exception as e:
        return ResponseSchema(
        code="500001",
            message="生成验证码失败",
            data=None
        )


@router.get("/me", response_model=ResponseSchema)
async def get_current_user_info(
    credentials: HTTPAuthorizationCredentials = Depends(security),
    db: AsyncSession = Depends(get_db)
):
    """获取当前用户信息"""
    return await get_user_info(credentials, db)