初始项目

app/__init__.py

@@ -0,0 +1 @@
+# AI Talk Callback API Application

app/config.py

@@ -0,0 +1,28 @@
+from pydantic_settings import BaseSettings
+
+
+class Settings(BaseSettings):
+    # 数据库配置
+    database_url: str = "postgresql+asyncpg://user:password@localhost:5432/ai_talk_callback_db"
+    
+    # Redis配置
+    redis_url: str = "redis://localhost:6379/0"
+    redis_lock_timeout: int = 30  # 锁超时时间（秒）
+    redis_lock_retry_delay: float = 0.1  # 锁重试延迟（秒）
+    redis_lock_max_retries: int = 100  # 最大重试次数
+    
+    # 业务配置
+    count_threshold: int = 3  # count阈值，大于等于此值直接返回
+    external_api_retry_max: int = 4  # 外部API最大重试次数
+    external_api_url: str = "https://external.com/api/openapi/customerApi/aiTaskResultFail"
+    
+    # 应用配置
+    app_name: str = "AI Talk Callback API"
+    debug: bool = False
+    
+    class Config:
+        env_file = ".env"
+        case_sensitive = False
+
+
+settings = Settings()

app/database.py

@@ -0,0 +1,62 @@
+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 datetime import datetime
+from app.config import settings
+
+
+class Base(DeclarativeBase):
+    pass
+
+
+class CallbackLog(Base):
+    __tablename__ = "callback_logs"
+    
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    site_id = Column(String(100), nullable=False, comment="站点ID")  # 新增siteId字段
+    request_url = Column(String(500), nullable=False)
+    request_headers = Column(JSON, nullable=False)
+    request_body = Column(JSON, nullable=False)
+    created_at = Column(DateTime, default=datetime.utcnow)
+
+
+class ExternalApiLog(Base):
+    __tablename__ = "external_api_logs"  # 重命名表，更通用
+    
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    request_url = Column(String(500), nullable=False)
+    request_headers = Column(JSON, nullable=False)
+    request_body = Column(JSON, nullable=False)
+    response_status = Column(Integer)
+    response_headers = Column(JSON)
+    response_body = Column(Text)
+    retry_count = Column(Integer, default=0)  # 重试次数
+    created_at = Column(DateTime, default=datetime.utcnow)
+
+
+# 创建数据库引擎
+engine = create_async_engine(
+    settings.database_url,
+    echo=settings.debug,
+    future=True
+)
+
+# 创建会话工厂
+AsyncSessionLocal = async_sessionmaker(
+    engine,
+    class_=AsyncSession,
+    expire_on_commit=False
+)
+
+
+async def get_db():
+    async with AsyncSessionLocal() as session:
+        try:
+            yield session
+        finally:
+            await session.close()
+
+
+async def init_db():
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)

app/models.py

@@ -0,0 +1,63 @@
+from pydantic import BaseModel, Field
+from typing import List, Optional
+
+
+class NumberData(BaseModel):
+    number: str
+    province: str
+    city: str
+    operator: str
+
+
+class Group(BaseModel):
+    id: int
+    name: str
+
+
+class Task(BaseModel):
+    id: str
+    name: str
+
+
+class User(BaseModel):
+    id: str
+    name: str
+
+
+class CustomerData(BaseModel):
+    name: str
+    email: str
+    company: Optional[str] = None
+    extra: Optional[str] = None
+
+
+class CallbackItem(BaseModel):
+    bill: int
+    duration: int
+    callid: str
+    calldate: str
+    number: str
+    numberid: str
+    customer_id: str
+    status: int
+    status_str: str
+    user_id: str
+    type: int
+    number_data: NumberData
+    group: Group
+    task: Task
+    user: User
+    customer_data: CustomerData
+
+
+class CallbackRequest(BaseModel):
+    count: int = Field(..., ge=0, description="通话失败个数")
+    data: List[CallbackItem]
+
+
+class CallbackResponse(BaseModel):
+    success: bool
+    message: str
+    processed: bool = False
+    retry_count: int = 0
+    site_id: str  # 添加siteId到响应中

app/redis_lock.py

@@ -0,0 +1,102 @@
+import redis.asyncio as redis
+import asyncio
+import uuid
+from typing import Optional
+from app.config import settings
+
+
+class RedisLock:
+    def __init__(self, redis_client: redis.Redis, key: str, timeout: int = None):
+        self.redis_client = redis_client
+        self.key = f"lock:{key}"
+        self.timeout = timeout or settings.redis_lock_timeout
+        self.identifier = str(uuid.uuid4())
+        self.acquired = False
+
+    async def acquire(self) -> bool:
+        """获取分布式锁"""
+        lua_script = """
+        if redis.call("GET", KEYS[1]) == false then
+            return redis.call("SETEX", KEYS[1], ARGV[1], ARGV[2])
+        else
+            return false
+        end
+        """
+        
+        result = await self.redis_client.eval(
+            lua_script,
+            1,
+            self.key,
+            self.timeout,
+            self.identifier
+        )
+        
+        self.acquired = bool(result)
+        return self.acquired
+
+    async def release(self) -> bool:
+        """释放分布式锁"""
+        if not self.acquired:
+            return False
+            
+        lua_script = """
+        if redis.call("GET", KEYS[1]) == ARGV[1] then
+            return redis.call("DEL", KEYS[1])
+        else
+            return 0
+        end
+        """
+        
+        result = await self.redis_client.eval(
+            lua_script,
+            1,
+            self.key,
+            self.identifier
+        )
+        
+        self.acquired = False
+        return bool(result)
+
+    async def __aenter__(self):
+        """异步上下文管理器入口"""
+        retries = 0
+        while retries < settings.redis_lock_max_retries:
+            if await self.acquire():
+                return self
+            await asyncio.sleep(settings.redis_lock_retry_delay)
+            retries += 1
+        
+        raise TimeoutError(f"Failed to acquire lock {self.key} after {retries} retries")
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        """异步上下文管理器出口"""
+        await self.release()
+
+
+class RedisManager:
+    def __init__(self):
+        self.redis_client: Optional[redis.Redis] = None
+
+    async def connect(self):
+        """连接Redis"""
+        self.redis_client = redis.from_url(
+            settings.redis_url,
+            encoding="utf-8",
+            decode_responses=True
+        )
+        await self.redis_client.ping()
+
+    async def disconnect(self):
+        """断开Redis连接"""
+        if self.redis_client:
+            await self.redis_client.close()
+
+    def create_lock(self, key: str, timeout: int = None) -> RedisLock:
+        """创建分布式锁"""
+        if not self.redis_client:
+            raise RuntimeError("Redis client not connected")
+        return RedisLock(self.redis_client, key, timeout)
+
+
+# 全局Redis管理器实例
+redis_manager = RedisManager()

app/routes.py

@@ -0,0 +1,187 @@
+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
+):
+    """记录回调请求到数据库"""
+    callback_log = CallbackLog(
+        site_id=site_id,  # 记录siteId
+        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}", 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)}"
+        )