From 4cf817eadf7a4e1e49ff100b3f2344c46fac4746 Mon Sep 17 00:00:00 2001 From: "mark.tian" Date: Tue, 2 Dec 2025 17:25:37 +0800 Subject: [PATCH] =?UTF-8?q?=E5=88=9D=E5=A7=8B=E9=A1=B9=E7=9B=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 17 +++++ app/__init__.py | 1 + app/config.py | 28 +++++++ app/database.py | 62 +++++++++++++++ app/models.py | 63 ++++++++++++++++ app/redis_lock.py | 102 +++++++++++++++++++++++++ app/routes.py | 187 ++++++++++++++++++++++++++++++++++++++++++++++ main.py | 57 ++++++++++++++ requirements.txt | 12 +++ setup.py | 167 +++++++++++++++++++++++++++++++++++++++++ 10 files changed, 696 insertions(+) create mode 100644 .env.example create mode 100644 app/__init__.py create mode 100644 app/config.py create mode 100644 app/database.py create mode 100644 app/models.py create mode 100644 app/redis_lock.py create mode 100644 app/routes.py create mode 100644 main.py create mode 100644 requirements.txt create mode 100644 setup.py diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..91cd2d4 --- /dev/null +++ b/.env.example @@ -0,0 +1,17 @@ +# 数据库配置 +DATABASE_URL=postgresql+asyncpg://user:password@localhost:5432/ai_talk_callback_db + +# Redis配置 +REDIS_URL=redis://localhost:6379/0 +REDIS_LOCK_TIMEOUT=30 +REDIS_LOCK_RETRY_DELAY=0.1 +REDIS_LOCK_MAX_RETRIES=100 + +# 业务配置 +COUNT_THRESHOLD=3 +EXTERNAL_API_RETRY_MAX=4 +EXTERNAL_API_URL=https://external.com/api/openapi/customerApi/aiTaskResultFail + +# 应用配置 +APP_NAME=AI Talk Callback API +DEBUG=false \ No newline at end of file diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 0000000..eea3fa3 --- /dev/null +++ b/app/__init__.py @@ -0,0 +1 @@ +# AI Talk Callback API Application \ No newline at end of file diff --git a/app/config.py b/app/config.py new file mode 100644 index 0000000..4ef7b29 --- /dev/null +++ b/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() \ No newline at end of file diff --git a/app/database.py b/app/database.py new file mode 100644 index 0000000..c958d2f --- /dev/null +++ b/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) \ No newline at end of file diff --git a/app/models.py b/app/models.py new file mode 100644 index 0000000..4e5f478 --- /dev/null +++ b/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到响应中 \ No newline at end of file diff --git a/app/redis_lock.py b/app/redis_lock.py new file mode 100644 index 0000000..6ccdfef --- /dev/null +++ b/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() \ No newline at end of file diff --git a/app/routes.py b/app/routes.py new file mode 100644 index 0000000..9b1e739 --- /dev/null +++ b/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)}" + ) \ No newline at end of file diff --git a/main.py b/main.py new file mode 100644 index 0000000..ad9dde8 --- /dev/null +++ b/main.py @@ -0,0 +1,57 @@ +from fastapi import FastAPI +from fastapi.middleware.cors import CORSMiddleware +from contextlib import asynccontextmanager + +from app.config import settings +from app.database import init_db +from app.redis_lock import redis_manager +from app.routes import router + + +@asynccontextmanager +async def lifespan(app: FastAPI): + # 启动时初始化 + await init_db() + await redis_manager.connect() + yield + # 关闭时清理 + await redis_manager.disconnect() + + +app = FastAPI( + title=settings.app_name, + version="1.0.0", + lifespan=lifespan +) + +# 添加CORS中间件 +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], +) + +# 注册路由 +app.include_router(router) + + +@app.get("/") +async def root(): + return {"message": f"Welcome to {settings.app_name}"} + + +@app.get("/health") +async def health_check(): + return {"status": "healthy"} + + +if __name__ == "__main__": + import uvicorn + uvicorn.run( + "main:app", + host="0.0.0.0", + port=8000, + reload=settings.debug + ) \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..43ebfbb --- /dev/null +++ b/requirements.txt @@ -0,0 +1,12 @@ +fastapi==0.104.1 +uvicorn[standard]==0.24.0 +sqlalchemy==2.0.23 +asyncpg==0.29.0 +alembic==1.12.1 +redis==5.0.1 +aioredis==2.0.1 +pydantic==2.5.0 +pydantic-settings==2.1.0 +python-multipart==0.0.6 +httpx==0.25.2 +python-dotenv==1.0.0 \ No newline at end of file diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..04d19a1 --- /dev/null +++ b/setup.py @@ -0,0 +1,167 @@ +#!/usr/bin/env python3 +""" +数据库初始化脚本 +用于创建数据库表结构 +""" + +import asyncio +import sys +import os + +# 添加项目根目录到Python路径 +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from app.database import init_db, engine +from app.config import settings +from sqlalchemy import text + + +async def create_database(): + """创建数据库（如果不存在）""" + try: + # 从DATABASE_URL中提取数据库名 + db_url = settings.database_url + if "postgresql+asyncpg://" in db_url: + # 解析数据库URL获取数据库名 + # 格式: postgresql+asyncpg://user:password@host:port/dbname + import re + match = re.match(r'.+/([^/]+)$', db_url) + if match: + db_name = match.group(1) + # 创建不包含数据库名的连接URL + base_url = db_url.rsplit('/', 1)[0] + + # 连接到postgres默认数据库创建新数据库 + from sqlalchemy.ext.asyncio import create_async_engine + temp_engine = create_async_engine(f"{base_url}/postgres") + + async with temp_engine.begin() as conn: + # 检查数据库是否已存在 + result = await conn.execute( + text("SELECT 1 FROM pg_database WHERE datname = :db_name"), + {"db_name": db_name} + ) + exists = result.scalar() + + if not exists: + await conn.execute(text(f"CREATE DATABASE {db_name}")) + print(f"✅ 数据库 '{db_name}' 创建成功") + else: + print(f"✅ 数据库 '{db_name}' 已存在") + + await temp_engine.dispose() + + except Exception as e: + print(f"❌ 创建数据库失败: {e}") + return False + + return True + + +async def init_tables(): + """初始化数据库表结构""" + try: + print("🔄 开始初始化数据库表结构...") + + # 初始化所有表 + await init_db() + + print("✅ 数据库表结构初始化成功") + + # 验证表是否创建成功 + async with engine.begin() as conn: + result = await conn.execute(text(""" + SELECT table_name + FROM information_schema.tables + WHERE table_schema = 'public' + ORDER BY table_name + """)) + tables = [row[0] for row in result.fetchall()] + + print(f"📋 已创建的表: {', '.join(tables)}") + + # 检查必要的表 + required_tables = ['callback_logs', 'external_api_logs'] + missing_tables = [table for table in required_tables if table not in tables] + + if missing_tables: + print(f"❌ 缺少必要的表: {', '.join(missing_tables)}") + return False + else: + print("✅ 所有必要的表都已创建") + + return True + + except Exception as e: + print(f"❌ 初始化数据库表结构失败: {e}") + return False + + +async def verify_connection(): + """验证数据库连接""" + try: + print("🔄 验证数据库连接...") + + async with engine.begin() as conn: + result = await conn.execute(text("SELECT version()")) + version = result.scalar() + print(f"✅ 数据库连接成功") + print(f"📊 PostgreSQL版本: {version}") + + return True + + except Exception as e: + print(f"❌ 数据库连接失败: {e}") + print("\n💡 请检查以下配置:") + print(f" - DATABASE_URL: {settings.database_url}") + print(" - PostgreSQL服务是否运行") + print(" - 用户名密码是否正确") + print(" - 网络连接是否正常") + return False + + +async def main(): + """主函数""" + print("🚀 开始数据库初始化...") + print(f"📋 配置信息:") + print(f" - 数据库URL: {settings.database_url}") + print(f" - Redis URL: {settings.redis_url}") + print(f" - 应用名称: {settings.app_name}") + print(f" - 外部API URL: {settings.external_api_url}") + print() + + # 验证连接 + if not await verify_connection(): + sys.exit(1) + + print() + + # 创建数据库（如果需要） + if not await create_database(): + sys.exit(1) + + print() + + # 初始化表结构 + if not await init_tables(): + sys.exit(1) + + print() + print("🎉 数据库初始化完成！") + print("\n📝 下一步:") + print(" 1. 配置 .env 文件中的数据库连接信息") + print(" 2. 确保 Redis 服务正在运行") + print(" 3. 运行应用: python main.py") + print(f" 4. 访问API文档: http://localhost:8000/docs") + print(f" 5. 接口地址: POST /ai-talk/callback/{{siteId}}") + + +if __name__ == "__main__": + try: + asyncio.run(main()) + except KeyboardInterrupt: + print("\n⚠️ 用户中断操作") + sys.exit(1) + except Exception as e: + print(f"\n❌ 初始化过程中发生错误: {e}") + sys.exit(1) \ No newline at end of file