diff --git a/app/database.py b/app/database.py index be92ef3..f7307b6 100644 --- a/app/database.py +++ b/app/database.py @@ -1,6 +1,6 @@ from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker from sqlalchemy.orm import DeclarativeBase -from sqlalchemy import Column, String, Integer, DateTime, Text, JSON +from sqlalchemy import Column, String, Integer, DateTime, Text, JSON, Index from datetime import datetime from app.config import settings from app.logger import get_logger @@ -25,6 +25,22 @@ class CallbackLog(Base): created_at = Column(DateTime, default=datetime.now) +class CallbackData(Base): + __tablename__ = "callback_data" + + id = Column(Integer, primary_key=True, autoincrement=True) + phone_number = Column(String(20), nullable=False, comment="手机号") + task_id = Column(String(100), nullable=False, comment="任务ID") + status = Column(Integer, nullable=False, comment="状态") + status_description = Column(String(200), nullable=False, comment="状态描述") + created_at = Column(DateTime, default=datetime.now) + + # 为手机号字段添加索引，提高查询性能 + __table_args__ = ( + Index('idx_phone_number', 'phone_number'), + ) + + class ExternalApiLog(Base): __tablename__ = "external_api_logs" # 重命名表，更通用 diff --git a/app/routes.py b/app/routes.py index 308e47e..27fe442 100644 --- a/app/routes.py +++ b/app/routes.py @@ -1,10 +1,11 @@ from fastapi import APIRouter, Request, HTTPException, Depends, Path from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy import func, select import httpx import asyncio from typing import Dict, Any, Optional -from app.database import get_db, CallbackLog, ExternalApiLog +from app.database import get_db, CallbackLog, ExternalApiLog, CallbackData from app.models import CallbackRequest, CallbackResponse from app.redis_lock import redis_manager from app.config import settings @@ -18,8 +19,7 @@ router = APIRouter() async def log_callback_request( db: AsyncSession, request: Request, - site_id: str, - callback_data: CallbackRequest + site_id: str ) -> Optional[int]: """记录回调请求到数据库，返回记录ID""" import json @@ -53,7 +53,7 @@ async def log_callback_request( except Exception as e: logger.warning(f"⚠️ 读取请求体失败: {e}") request_body = "Unable to read request body" - + # 记录请求URL（JSON格式） logger.info(f"🌐 请求URL: {request.url}") @@ -105,6 +105,77 @@ async def log_callback_request( return None +async def check_phone_number_threshold( + db: AsyncSession, + phone_number: str +) -> tuple[bool, bool]: + """ + 检查手机号在数据库中的出现次数是否超过阈值 + + Args: + db: 数据库会话 + phone_number: 要检查的手机号 + + Returns: + tuple[是否超过阈值, 是否查询失败] + """ + try: + # 查询手机号在数据库中出现的次数 + count_query = select(func.count(CallbackData.id)).where( + CallbackData.phone_number == phone_number + ) + result = await db.execute(count_query) + phone_count = result.scalar() or 0 + + logger.info(f"📊 手机号 {phone_number} 在数据库中出现次数: {phone_count}") + + # 判断是否超过阈值 + exceeds_threshold = phone_count >= settings.count_threshold + + if exceeds_threshold: + logger.info(f"✅ 手机号 {phone_number} 出现次数 {phone_count} >= {settings.count_threshold}，超过阈值") + + return exceeds_threshold, False + + except Exception as e: + logger.error(f"❌ 查询手机号 {phone_number} 失败: {e}", exc_info=True) + # 查询失败时返回False，不影响主流程 + return False, True + + +async def save_callback_data_items( + db: AsyncSession, + callback_data_items: list +): + """保存callback_data.data中的数据到数据库""" + try: + callback_data_records = [] + for item in callback_data_items: + if not hasattr(item, 'number_data') or not item.number_data: + continue + + if not hasattr(item.number_data, 'number') or not item.number_data.number: + continue + + callback_data_record = CallbackData( + phone_number=item.number_data.number, + task_id=item.task.id, + status=item.status, + status_description=item.status_str + ) + callback_data_records.append(callback_data_record) + + if len(callback_data_records) > 0: + db.add_all(callback_data_records) + await db.commit() + + logger.info(f"✅ 成功保存 {len(callback_data_records)} 条callback_data记录到数据库") + + except Exception as e: + logger.error(f"❌ 保存callback_data到数据库失败: {e}", exc_info=True) + # 不重新抛出异常，避免影响主业务流程 + + async def log_external_api_request( db: AsyncSession, callback_logs_id: int, @@ -238,65 +309,87 @@ async def ai_talk_callback( """ AI Talk回调接口处理 """ - logger.info(f"🔥 收到AI Talk回调请求: siteId={siteId}, count={callback_data.count}, data_count={len(callback_data.data)}") - try: + # 参数校验 + if not callback_data: + logger.error("❌ callback_data 参数为空") + raise HTTPException(status_code=400, detail="callback_data 参数不能为空") + + if not isinstance(callback_data.data, list): + logger.error(f"❌ data 参数类型无效: {type(callback_data.data)}") + raise HTTPException(status_code=400, detail="data 参数必须是数组") + # 记录回调请求（包含siteId） - callback_log_id = await log_callback_request(db, request, siteId, callback_data) + callback_log_id = await log_callback_request(db, request, siteId) - # 判断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 - ) + # 使用分布式锁防止并发调用，使用callback_log_id作为唯一值 + async with redis_manager.create_lock(f"ai_talk_callback_{callback_log_id}"): + logger.debug(f"🔒 获取Redis锁成功: ai_talk_callback_{callback_log_id}") - # 检查是否启用外部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}") + # 保存callback_data.data中的数据 + if callback_data.data and len(callback_data.data) > 0: + await save_callback_data_items(db, callback_data.data) - # 调用外部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}") + # 检查是否启用外部API调用 + if not settings.external_api_enabled: + logger.info(f"🔌 外部API调用已禁用，直接返回成功") return CallbackResponse( success=True, - message="外部API调用成功", - processed=True, - retry_count=retry_count, + message="外部API调用已禁用，直接返回", + processed=False, + retry_count=0, site_id=siteId ) - else: - logger.error(f"❌ 外部API调用失败，已重试{retry_count}次") - raise HTTPException( - status_code=500, - detail=f"外部API调用失败，已重试{retry_count}次" + + # 提取所有手机号并去重 + phone_numbers_set = set() + for item in callback_data.data: + phone_numbers_set.add(item.number_data.number) + + phone_numbers = list(phone_numbers_set) + logger.info(f"📱 提取到的手机号列表（去重后）: {phone_numbers}") + + # 检查去重后的手机号是否超过 settings.count_threshold，超过直接返回 + for phone_number in phone_numbers: + + # 检查手机号是否超过阈值 + exceeds_threshold, query_failed = await check_phone_number_threshold(db, phone_number) + + # 如果查询出错，跳过此手机号 + if query_failed: + logger.warning(f"⚠️ 查询手机号 {phone_number} 失败，跳过处理") + continue + + if exceeds_threshold: + logger.info(f"✅ 手机号 {phone_number} 出现次数超过阈值，跳过处理") + continue + + # 手机号出现的次数少于settings.count_threshold，调用外部API + # 调用外部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: