记录未打通通过记录的保存

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"  # 重命名表，更通用
     

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,
+    site_id: str
-    callback_data: CallbackRequest
 ) -> Optional[int]:
     """记录回调请求到数据库，返回记录ID"""
     import json
@@ -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是否大于等于阈值，如果是直接返回
+        # 使用分布式锁防止并发调用，使用callback_log_id作为唯一值
-        if callback_data.count >= settings.count_threshold:
+        async with redis_manager.create_lock(f"ai_talk_callback_{callback_log_id}"):
-            logger.info(f"✅ count={callback_data.count} >= {settings.count_threshold}，直接返回")
+            logger.debug(f"🔒 获取Redis锁成功: ai_talk_callback_{callback_log_id}")
-            return CallbackResponse(
-                success=True,
-                message=f"count={callback_data.count} >= {settings.count_threshold}，直接返回",
-                processed=False,
-                retry_count=0,
-                site_id=siteId
-            )
         
-        # 检查是否启用外部API调用
+            # 保存callback_data.data中的数据
-        if not settings.external_api_enabled:
+            if callback_data.data and len(callback_data.data) > 0:
-            logger.info(f"🔌 外部API调用已禁用，直接返回成功")
+                await save_callback_data_items(db, callback_data.data)
-            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")
+            # 检查是否启用外部API调用
-        
+            if not settings.external_api_enabled:
-        # count < 3，调用外部API，使用分布式锁防止并发调用
+                logger.info(f"🔌 外部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调用成功",
+                    message="外部API调用已禁用，直接返回",
-                    processed=True,
+                    processed=False,
-                    retry_count=retry_count,
+                    retry_count=0,
                     site_id=siteId
                 )
-            else:
+            
-                logger.error(f"❌ 外部API调用失败，已重试{retry_count}次")
+            # 提取所有手机号并去重
-                raise HTTPException(
+            phone_numbers_set = set()
-                    status_code=500,
+            for item in callback_data.data:
-                    detail=f"外部API调用失败，已重试{retry_count}次"
+                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: