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 from app.logger import get_logger logger = get_logger("routes") router = APIRouter() async def log_callback_request( db: AsyncSession, request: Request, site_id: str, callback_data: CallbackRequest ) -> Optional[int]: """记录回调请求到数据库，返回记录ID""" import json # 获取客户端IP地址 client_ip = request.client.host if request.client else None client_port = request.client.port if request.client else None # 获取服务器IP地址 server_ip = None server_port = None if hasattr(request, 'scope') and 'server' in request.scope: server_host, server_port_info = request.scope['server'] server_ip = server_host server_port = server_port_info # 准备请求头信息（过滤敏感信息） request_headers = dict(request.headers) safe_headers = {} sensitive_headers = {'authorization', 'token', 'api-key', 'x-api-key', 'cookie'} for key, value in request_headers.items(): if key.lower() in sensitive_headers: safe_headers[key] = "***REDACTED***" else: safe_headers[key] = value # 准备请求体信息 request_body = callback_data.model_dump() # 记录请求URL（JSON格式） logger.info(f"🌐 请求URL: {json.dumps(str(request.url), ensure_ascii=False)}") # 记录site_id（JSON格式） logger.info(f"📝 site_id: {json.dumps(site_id, ensure_ascii=False)}") # 记录请求头（JSON格式） logger.info(f"📋 请求头: {json.dumps(safe_headers, ensure_ascii=False, indent=2)}") # 记录请求体（JSON格式） logger.info(f"📄 请求体: {json.dumps(request_body, ensure_ascii=False, indent=2)}") # 记录server_ip（JSON格式） server_info = { "ip": server_ip, "port": server_port } logger.info(f"🏠 server_ip: {json.dumps(server_info, ensure_ascii=False)}") # 记录client_ip（JSON格式） client_info = { "ip": client_ip, "port": client_port } logger.info(f"🖥️ client_ip: {json.dumps(client_info, ensure_ascii=False)}") # 记录callback_data（JSON格式） callback_data_json = { "count": callback_data.count, "data_count": len(callback_data.data) if callback_data.data else 0, "data_sample": callback_data.data[0].model_dump() if callback_data.data else None } logger.info(f"📦 callback_data: {json.dumps(callback_data_json, ensure_ascii=False, indent=2)}") try: # 保存到数据库 callback_log = CallbackLog( site_id=site_id, remote_address=f"{client_ip}:{client_port}" if client_ip and client_port else client_ip, server_ip=f"{server_ip}:{server_port}" if server_ip and server_port else server_ip, request_url=str(request.url), request_headers=safe_headers, # 保存过滤后的请求头 request_body=request_body ) db.add(callback_log) await db.commit() logger.info(f"✅ 回调请求记录成功保存到数据库，ID: {callback_log.id}") # 返回记录ID return callback_log.id except Exception as e: logger.error(f"❌ 保存回调请求到数据库失败: {e}", exc_info=True) # 不重新抛出异常，避免影响主业务流程 return None async def log_external_api_request( db: AsyncSession, callback_logs_id: int, 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( callback_logs_id=callback_logs_id, 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, callback_logs_id: 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 logger.info(f"🌐 开始调用外部API: {settings.external_api_url}, 最大重试次数: {max_retries}") headers = { "Content-Type": "application/json", "User-Agent": "AITalkCallbackService/1.0" } for attempt in range(1, max_retries+1): try: logger.debug(f"📤 第{attempt + 1}次尝试调用外部API") async with httpx.AsyncClient(timeout=30.0) as client: response = await client.post( settings.external_api_url, headers=headers, json=request_body ) logger.debug(f"📥 外部API响应: status={response.status_code}") # 记录每次尝试的结果 await log_external_api_request( db=db, callback_logs_id=callback_logs_id, 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: logger.info(f"✅ 外部API调用成功，状态码: {response.status_code}") return True, attempt else: logger.warning(f"⚠️ 外部API返回错误状态码: {response.status_code}") # 如果是最后一次尝试，直接返回失败 if attempt == max_retries: logger.error(f"❌ 外部API调用最终失败，状态码: {response.status_code}") return False, attempt # 否则等待一段时间后重试 wait_time = 1 * (attempt + 1) # 递增延迟 logger.info(f"⏳ 等待 {wait_time}秒 后重试...") await asyncio.sleep(wait_time) except httpx.RequestError as e: logger.warning(f"⚠️ 外部API网络错误 (第{attempt + 1}次尝试): {e}") # 记录网络错误 await log_external_api_request( db=db, callback_logs_id=callback_logs_id, 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: logger.error(f"❌ 外部API调用最终失败: {e}") return False, attempt # 否则等待一段时间后重试 wait_time = 1 * (attempt + 1) # 递增延迟 logger.info(f"⏳ 等待 {wait_time}秒 后重试...") await asyncio.sleep(wait_time) 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回调接口处理 """ logger.info(f"🔥 收到AI Talk回调请求: siteId={siteId}, count={callback_data.count}, data_count={len(callback_data.data)}") try: # 记录回调请求（包含siteId） callback_log_id = await log_callback_request(db, request, siteId, callback_data) # 判断count是否大于等于阈值，如果是直接返回 if callback_data.count >= settings.count_threshold: logger.info(f"✅ count={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 ) # 检查是否启用外部API调用 if not settings.external_api_enabled: logger.info(f"🔌 外部API调用已禁用，直接返回成功") return CallbackResponse( success=True, message="外部API调用已禁用，直接返回", processed=False, retry_count=0, site_id=siteId ) logger.info(f"📞 count={callback_data.count} < {settings.count_threshold}，调用外部API") # count < 3，调用外部API，使用分布式锁防止并发调用 async with redis_manager.create_lock(f"external_api_call_{siteId}_{callback_data.count}"): logger.debug(f"🔒 获取Redis锁成功: 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, callback_logs_id=callback_log_id ) if success: logger.info(f"✅ 外部API调用成功，重试次数: {retry_count}") return CallbackResponse( success=True, message="外部API调用成功", processed=True, retry_count=retry_count, site_id=siteId ) else: logger.error(f"❌ 外部API调用失败，已重试{retry_count}次") raise HTTPException( status_code=500, detail=f"外部API调用失败，已重试{retry_count}次" ) except HTTPException: raise except Exception as e: logger.error(f"💥 服务器内部错误: {e}", exc_info=True) raise HTTPException( status_code=500, detail=f"服务器内部错误: {str(e)}" )