ai-talk-callback/app/routes.py

from fastapi import APIRouter, Request, HTTPException, Depends, Path
from sqlalchemy.ext.asyncio import AsyncSession
import httpx
import asyncio
from typing import Dict, Any, Optional

from app.database import get_db, CallbackLog, ExternalApiLog
from app.models import CallbackRequest, CallbackResponse
from app.redis_lock import redis_manager
from app.config import settings

router = APIRouter()


async def log_callback_request(
    db: AsyncSession,
    request: Request,
    site_id: str,
    callback_data: CallbackRequest
):
    """记录回调请求到数据库"""
    # 获取客户端IP地址
    client_ip = request.client.host if request.client else None
    
    # 获取服务器IP地址
    server_ip = None
    if hasattr(request, 'scope') and 'server' in request.scope:
        server_host, server_port = request.scope['server']
        server_ip = server_host
    
    callback_log = CallbackLog(
        site_id=site_id,  # 记录siteId
        remote_address=client_ip,
        server_ip=server_ip,
        request_url=str(request.url),
        request_headers=dict(request.headers),
        request_body=callback_data.model_dump()
    )
    db.add(callback_log)
    await db.commit()


async def log_external_api_request(
    db: AsyncSession,
    request_url: str,
    request_headers: Dict[str, Any],
    request_body: Dict[str, Any],
    response_status: Optional[int],
    response_headers: Optional[Dict[str, Any]],
    response_body: Optional[str],
    retry_count: int = 0
):
    """记录外部API请求到数据库"""
    external_api_log = ExternalApiLog(
        request_url=request_url,
        request_headers=request_headers,
        request_body=request_body,
        response_status=response_status,
        response_headers=response_headers,
        response_body=response_body,
        retry_count=retry_count
    )
    db.add(external_api_log)
    await db.commit()


async def call_external_api_with_retry(
    db: AsyncSession,
    request_body: Dict[str, Any],
    max_retries: int = None
) -> tuple[bool, int]:
    """
    调用外部API并支持重试机制
    
    Args:
        db: 数据库会话
        request_body: 请求体
        max_retries: 最大重试次数
    
    Returns:
        tuple[是否成功, 实际重试次数]
    """
    if max_retries is None:
        max_retries = settings.external_api_retry_max
    
    headers = {
        "Content-Type": "application/json",
        "User-Agent": "AITalkCallbackService/1.0"
    }
    
    for attempt in range(max_retries + 1):  # +1 因为第一次不算重试
        try:
            async with httpx.AsyncClient(timeout=30.0) as client:
                response = await client.post(
                    settings.external_api_url,
                    headers=headers,
                    json=request_body
                )
                
                # 记录每次尝试的结果
                await log_external_api_request(
                    db=db,
                    request_url=settings.external_api_url,
                    request_headers=headers,
                    request_body=request_body,
                    response_status=response.status_code,
                    response_headers=dict(response.headers),
                    response_body=response.text,
                    retry_count=attempt
                )
                
                # 检查响应状态
                if response.status_code < 400:
                    return True, attempt
                else:
                    # 如果是最后一次尝试，直接返回失败
                    if attempt == max_retries:
                        return False, attempt
                    # 否则等待一段时间后重试
                    await asyncio.sleep(1 * (attempt + 1))  # 递增延迟
                    
        except httpx.RequestError as e:
            # 记录网络错误
            await log_external_api_request(
                db=db,
                request_url=settings.external_api_url,
                request_headers=headers,
                request_body=request_body,
                response_status=None,
                response_headers=None,
                response_body=f"RequestError: {str(e)}",
                retry_count=attempt
            )
            
            # 如果是最后一次尝试，直接返回失败
            if attempt == max_retries:
                return False, attempt
            # 否则等待一段时间后重试
            await asyncio.sleep(1 * (attempt + 1))  # 递增延迟
    
    return False, max_retries


@router.post("/ai-talk/callback/{siteId}/failure", response_model=CallbackResponse)
async def ai_talk_callback(
    callback_data: CallbackRequest,
    request: Request,
    siteId: str = Path(..., description="站点ID"),
    db: AsyncSession = Depends(get_db)
):
    """
    AI Talk回调接口处理
    """
    try:
        # 记录回调请求（包含siteId）
        await log_callback_request(db, request, siteId, callback_data)
        
        # 判断count是否大于等于阈值，如果是直接返回
        if callback_data.count >= settings.count_threshold:
            return CallbackResponse(
                success=True,
                message=f"count={callback_data.count} >= {settings.count_threshold}，直接返回",
                processed=False,
                retry_count=0,
                site_id=siteId
            )
        
        # count < 3，调用外部API，使用分布式锁防止并发调用
        async with redis_manager.create_lock(f"external_api_call_{siteId}_{callback_data.count}"):
            # 调用外部API并支持重试
            request_body = callback_data.model_dump()
            success, retry_count = await call_external_api_with_retry(
                db=db,
                request_body=request_body,
                max_retries=settings.external_api_retry_max
            )
            
            if success:
                return CallbackResponse(
                    success=True,
                    message="外部API调用成功",
                    processed=True,
                    retry_count=retry_count,
                    site_id=siteId
                )
            else:
                raise HTTPException(
                    status_code=500,
                    detail=f"外部API调用失败，已重试{retry_count}次"
                )
        
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(
            status_code=500,
            detail=f"服务器内部错误: {str(e)}"
        )