mark.tian/ai-talk-callback
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

保存请求失败返回500

Browse Source 
调整表结构
This commit is contained in:
mark.tian
2025-12-04 08:14:21 +08:00
parent f1f694bd63
commit 6ac998ebdf
5 changed files with 158 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files

60
app/database.py
View File

@@ -1,9 +1,9 @@
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase from sqlalchemy.orm import DeclarativeBase
from sqlalchemy import Column, String, Integer, DateTime, Text, JSON, Index from sqlalchemy import Column, String, Integer, DateTime, Text, JSON, Index, text
from datetime import datetime
from app.config import settings from app.config import settings
from app.logger import get_logger from app.logger import get_logger
from sqlalchemy.sql.expression import func
logger = get_logger("database") logger = get_logger("database")
@@ -12,49 +12,46 @@ class Base(DeclarativeBase):
pass pass
class CallbackLog(Base): class CallbackFailureLog(Base):
__tablename__ = "callback_logs" __tablename__ = "callback_failure_logs"
id = Column(Integer, primary_key=True, autoincrement=True) id = Column(Integer, primary_key=True, autoincrement=True, comment="日志ID")
site_id = Column(String(100), nullable=False, comment="站点ID") # 新增siteId字段 site_id = Column(String(100), nullable=False, comment="站点ID") # 新增siteId字段
remote_address = Column(String(45), nullable=True, comment="客户端IP地址") remote_address = Column(String(45), nullable=True, comment="客户端IP地址")
server_ip = Column(String(45), nullable=True, comment="服务器IP地址") server_ip = Column(String(45), nullable=True, comment="服务器IP地址")
request_url = Column(String(500), nullable=False) request_url = Column(String(500), nullable=False, comment="请求URL")
request_headers = Column(JSON, nullable=False) request_headers = Column(JSON, nullable=False, comment="请求头")
request_body = Column(JSON, nullable=False) request_body = Column(JSON, nullable=False, comment="请求体")
created_at = Column(DateTime, default=datetime.now) created_at = Column(DateTime, server_default=func.now(), comment="创建时间")
class CallbackData(Base): class CallbackFailureData(Base):
__tablename__ = "callback_data" __tablename__ = "callback_failure_data"
id = Column(Integer, primary_key=True, autoincrement=True) id = Column(Integer, primary_key=True, autoincrement=True, comment="数据ID")
callback_failure_log_id = Column(Integer, nullable=False, comment="回调失败日志ID")
phone_number = Column(String(20), nullable=False, comment="手机号") phone_number = Column(String(20), nullable=False, comment="手机号")
task_id = Column(String(100), nullable=False, comment="任务ID") task_id = Column(String(100), nullable=False, comment="任务ID")
status = Column(Integer, nullable=False, comment="状态") status = Column(Integer, nullable=False, comment="状态")
status_description = Column(String(200), nullable=False, comment="状态描述") status_description = Column(String(200), nullable=False, comment="状态描述")
raw_data = Column(Text, nullable=False, comment="原始回调数据JSON字符串") raw_data = Column(Text, nullable=False, comment="原始回调数据JSON字符串")
created_at = Column(DateTime, default=datetime.now) calldate = Column(DateTime, nullable=True, comment="通话日期")
created_at = Column(DateTime, server_default=func.now(), comment="创建时间")
# 为手机号字段添加索引，提高查询性能
__table_args__ = (
Index('idx_phone_number', 'phone_number'),
)
class ExternalApiLog(Base): class ExternalApiLog(Base):
__tablename__ = "external_api_logs" # 重命名表，更通用 __tablename__ = "external_api_logs" # 重命名表，更通用
id = Column(Integer, primary_key=True, autoincrement=True) id = Column(Integer, primary_key=True, autoincrement=True, comment="日志ID")
callback_logs_id = Column(Integer, nullable=False) callback_failure_log_id = Column(Integer, nullable=False, comment="回调失败日志ID")
request_url = Column(String(500), nullable=False) request_url = Column(String(500), nullable=False, comment="外部接口请求URL")
request_headers = Column(JSON, nullable=False) request_headers = Column(JSON, nullable=False, comment="外部接口请求头")
request_body = Column(JSON, nullable=False) request_body = Column(JSON, nullable=False, comment="外部接口请求体")
response_status = Column(Integer) response_status = Column(Integer, nullable=False, comment="外部接口响应状态码")
response_headers = Column(JSON) response_headers = Column(JSON, nullable=False, comment="外部接口响应头")
response_body = Column(Text) response_body = Column(Text, nullable=False, comment="外部接口响应体")
retry_count = Column(Integer, default=0) # 重试次数 retry_count = Column(Integer, default=0, comment="重试次数")
created_at = Column(DateTime, default=datetime.now) created_at = Column(DateTime, server_default=func.now(), comment="创建时间")
# 创建数据库引擎 # 创建数据库引擎
@@ -85,6 +82,13 @@ async def init_db():
try: try:
async with engine.begin() as conn: async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all) await conn.run_sync(Base.metadata.create_all)
# 检查并创建手机号索引
await conn.execute(text("""
CREATE INDEX IF NOT EXISTS idx_phone_number
ON callback_failure_data (phone_number)
"""))
logger.info("✅ 数据库表结构初始化完成") logger.info("✅ 数据库表结构初始化完成")
except Exception as e: except Exception as e:
logger.error(f"❌ 数据库初始化失败: {e}") logger.error(f"❌ 数据库初始化失败: {e}")

76
app/external_api_processor.py
View File

@@ -8,15 +8,49 @@ from fastapi import HTTPException
import httpx import httpx
import asyncio import asyncio
from typing import Dict, Any, Optional from typing import Dict, Any, Optional
import json
from app.database import CallbackFailureData, ExternalApiLog, CallbackFailureLog
from app.database import CallbackData, ExternalApiLog
from app.config import settings from app.config import settings
from app.logger import get_logger from app.logger import get_logger
logger = get_logger("external_api_processor") logger = get_logger("external_api_processor")
async def get_callback_log_data(
db: AsyncSession,
callback_log_id: int
) -> tuple[bool, Optional[Dict[str, Any]], Optional[str]]:
"""
从回调日志中获取请求头和请求体
Args:
db: 数据库会话
callback_log_id: 回调日志ID
Returns:
tuple[查询是否成功, 请求头字典, 请求体字符串]
"""
try:
# 查询回调日志
query = select(CallbackFailureLog).where(CallbackFailureLog.id == callback_log_id)
result = await db.execute(query)
callback_log = result.scalar_one_or_none()
if not callback_log:
logger.warning(f"⚠️ 未找到回调日志记录，ID: {callback_log_id}")
return False, None, None
logger.info(f"✅ 成功获取回调日志，ID: {callback_log_id}")
# 返回查询成功标识、请求头和请求体
return True, callback_log.request_headers, callback_log.request_body
except Exception as e:
logger.error(f"❌ 获取回调日志失败，ID: {callback_log_id}, 错误: {e}", exc_info=True)
return False, None, None
async def check_phone_number_threshold( async def check_phone_number_threshold(
db: AsyncSession, db: AsyncSession,
phone_number: str phone_number: str
@@ -33,8 +67,8 @@ async def check_phone_number_threshold(
""" """
try: try:
# 查询手机号在数据库中出现的次数 # 查询手机号在数据库中出现的次数
count_query = select(func.count(CallbackData.id)).where( count_query = select(func.count(CallbackFailureData.id)).where(
CallbackData.phone_number == phone_number CallbackFailureData.phone_number == phone_number
) )
result = await db.execute(count_query) result = await db.execute(count_query)
phone_count = result.scalar() or 0 phone_count = result.scalar() or 0
@@ -57,7 +91,7 @@ async def check_phone_number_threshold(
async def log_external_api_request( async def log_external_api_request(
db: AsyncSession, db: AsyncSession,
callback_logs_id: int, callback_failure_log_id: int,
request_url: str, request_url: str,
request_headers: Dict[str, Any], request_headers: Dict[str, Any],
request_body: Dict[str, Any], request_body: Dict[str, Any],
@@ -69,7 +103,7 @@ async def log_external_api_request(
"""记录外部API请求日志""" """记录外部API请求日志"""
try: try:
api_log = ExternalApiLog( api_log = ExternalApiLog(
callback_logs_id=callback_logs_id, callback_failure_log_id=callback_failure_log_id,
request_url=request_url, request_url=request_url,
request_headers=request_headers, request_headers=request_headers,
request_body=request_body, request_body=request_body,
@@ -92,7 +126,7 @@ async def call_external_api_with_retry(
db: AsyncSession, db: AsyncSession,
request_body: Dict[str, Any], request_body: Dict[str, Any],
max_retries: int = None, max_retries: int = None,
callback_logs_id: int = None callback_failure_log_id: int = None
) -> tuple[bool, int]: ) -> tuple[bool, int]:
""" """
调用外部API并支持重试机制 调用外部API并支持重试机制
@@ -101,7 +135,7 @@ async def call_external_api_with_retry(
db: 数据库会话 db: 数据库会话
request_body: 请求体 request_body: 请求体
max_retries: 最大重试次数 max_retries: 最大重试次数
callback_logs_id: 回调日志ID callback_failure_log_id: 回调失败日志ID
Returns: Returns:
tuple[是否成功, 实际重试次数] tuple[是否成功, 实际重试次数]
@@ -132,7 +166,7 @@ async def call_external_api_with_retry(
# 记录每次尝试的结果 # 记录每次尝试的结果
await log_external_api_request( await log_external_api_request(
db=db, db=db,
callback_logs_id=callback_logs_id, callback_failure_log_id=callback_failure_log_id,
request_url=settings.external_api_url, request_url=settings.external_api_url,
request_headers=headers, request_headers=headers,
request_body=request_body, request_body=request_body,
@@ -172,26 +206,37 @@ async def call_external_api_with_retry(
async def process_external_api_call( async def process_external_api_call(
db: AsyncSession, db: AsyncSession,
callback_data: dict, callback_log_id: int,
siteId: str, siteId: str,
callback_log_id: int
): ):
""" """
异步调用外部接口 异步调用外部接口
Args: Args:
db: 数据库会话 db: 数据库会话
callback_data: 回调数据
siteId: 站点ID
callback_log_id: 回调日志ID callback_log_id: 回调日志ID
siteId: 站点ID
Returns: Returns:
None: 无返回值 None: 无返回值
""" """
success, request_headers_json, request_body_json = await get_callback_log_data(db, callback_log_id)
if not success:
# 查询失败，处理错误情况
logger.error(f"获取回调日志数据失败:{callback_log_id}")
return
# 查询成功，可以使用获取到的数据
logger.info(f"📋 请求头: {callback_log_id}|{request_headers_json}")
logger.info(f"📄 请求体: {callback_log_id}|{request_body_json}")
request_headers = json.loads(request_headers_json)
request_body = json.loads(request_body_json)
# 提取所有手机号并去重 # 提取所有手机号并去重
phone_numbers_set = set() phone_numbers_set = set()
data_list = callback_data.get('data', []) data_list = request_body.get('data', [])
for item in data_list: for item in data_list:
number_data = item.get('number_data', {}) number_data = item.get('number_data', {})
phone_number = number_data.get('number') phone_number = number_data.get('number')
@@ -223,12 +268,11 @@ async def process_external_api_call(
# 手机号出现的次数少于settings.count_threshold，调用外部API # 手机号出现的次数少于settings.count_threshold，调用外部API
# 调用外部API并支持重试 # 调用外部API并支持重试
request_body = callback_data
success, retry_count = await call_external_api_with_retry( success, retry_count = await call_external_api_with_retry(
db=db, db=db,
request_body=request_body, request_body=request_body,
max_retries=settings.external_api_retry_max, max_retries=settings.external_api_retry_max,
callback_logs_id=callback_log_id callback_failure_log_id=callback_log_id
) )
if success: if success:

73
app/routes.py
View File

@@ -2,7 +2,7 @@ from fastapi import APIRouter, Request, Body, HTTPException, Depends, Path
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from typing import Optional from typing import Optional
from app.database import get_db, CallbackLog, CallbackData from app.database import get_db, CallbackFailureLog, CallbackFailureData
from app.models import CallbackResponse from app.models import CallbackResponse
from app.config import settings from app.config import settings
from app.logger import get_logger from app.logger import get_logger
@@ -18,8 +18,8 @@ async def log_callback_request(
request: Request, request: Request,
site_id: str, site_id: str,
callback_data: dict callback_data: dict
) -> Optional[int]: ) -> tuple[bool, Optional[int]]:
"""记录回调请求到数据库，返回记录ID""" """记录回调请求到数据库，返回操作是否成功和记录ID"""
import json import json
# 获取客户端IP地址 # 获取客户端IP地址
@@ -68,7 +68,7 @@ async def log_callback_request(
try: try:
# 保存到数据库 # 保存到数据库
callback_log = CallbackLog( callback_log = CallbackFailureLog(
site_id=site_id, site_id=site_id,
remote_address=f"{client_ip}:{client_port}" if client_ip and client_port else client_ip, 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, server_ip=f"{server_ip}:{server_port}" if server_ip and server_port else server_ip,
@@ -82,21 +82,23 @@ async def log_callback_request(
logger.info(f"✅ 回调请求记录成功保存到数据库，ID: {callback_log.id}") logger.info(f"✅ 回调请求记录成功保存到数据库，ID: {callback_log.id}")
# 返回记录ID # 返回成功标识和记录ID
return callback_log.id return True, callback_log.id
except Exception as e: except Exception as e:
logger.error(f"❌ 保存回调请求到数据库失败: {e}", exc_info=True) logger.error(f"❌ 保存回调请求到数据库失败: {e}", exc_info=True)
# 不重新抛出异常，避免影响主业务流程 # 返回失败标识
return None return False, None
async def save_callback_data_items( async def save_callback_data_items(
db: AsyncSession, db: AsyncSession,
callback_data_items: list callback_data_items: list,
callback_failure_log_id: int
): ):
"""保存callback_data.data中的数据到数据库""" """保存callback_data.data中的数据到数据库"""
import json import json
from datetime import datetime
try: try:
callback_data_records = [] callback_data_records = []
@@ -116,14 +118,30 @@ async def save_callback_data_items(
status = item.get('status', 0) status = item.get('status', 0)
status_description = item.get('status_str', '') status_description = item.get('status_str', '')
# 获取通话日期，默认使用当前时间
calldate = datetime.now()
if 'calldate' in item:
try:
# 如果calldate是字符串，尝试解析为datetime
if isinstance(item['calldate'], str):
calldate = datetime.fromisoformat(item['calldate'].replace('Z', '+00:00'))
elif isinstance(item['calldate'], (int, float)):
# 如果是时间戳，转换为datetime
calldate = datetime.fromtimestamp(item['calldate'])
except (ValueError, TypeError) as e:
logger.warning(f"⚠️ 解析calldate失败: {item.get('calldate')}, 使用当前时间, 错误: {e}")
calldate = datetime.now()
# 将整个item转换为JSON字符串保存 # 将整个item转换为JSON字符串保存
raw_data_json = json.dumps(item, ensure_ascii=False) raw_data_json = json.dumps(item, ensure_ascii=False)
callback_data_record = CallbackData( callback_data_record = CallbackFailureData(
callback_failure_log_id=callback_failure_log_id,
phone_number=phone_number, phone_number=phone_number,
task_id=task_id, task_id=task_id,
status=status, status=status,
status_description=status_description, status_description=status_description,
raw_data=raw_data_json # 保存原始JSON字符串 raw_data=raw_data_json, # 保存原始JSON字符串
calldate=calldate # 保存通话日期
) )
callback_data_records.append(callback_data_record) callback_data_records.append(callback_data_record)
@@ -159,27 +177,26 @@ async def ai_talk_callback(
raise HTTPException(status_code=400, detail="data 参数必须是数组") raise HTTPException(status_code=400, detail="data 参数必须是数组")
# 记录回调请求（包含siteId） # 记录回调请求（包含siteId）
callback_log_id = await log_callback_request(db, request, siteId, callback_data) success, callback_log_id = await log_callback_request(db, request, siteId, callback_data)
# 保存callback_data.data中的数据 if not success:
data_list = callback_data.get('data', []) logger.error("❌ 回调请求日志保存失败")
if data_list and len(data_list) > 0: raise HTTPException(status_code=500, detail="回调请求日志保存失败")
await save_callback_data_items(db, data_list)
if callback_log_id:
# 保存callback_data.data中的数据
data_list = callback_data.get('data', [])
if data_list and len(data_list) > 0:
await save_callback_data_items(db, data_list, callback_log_id)
# 检查是否启用外部API调用 # 检查是否启用外部API调用
if not settings.external_api_enabled: if settings.external_api_enabled:
# 异步调用外部接口，不等待结果（只有在成功获取callback_log_id时才调用）
if success and callback_log_id:
import asyncio
asyncio.create_task(process_external_api_call(db, callback_data, siteId, callback_log_id))
else:
logger.info(f"🔌 外部API调用已禁用，直接返回成功") logger.info(f"🔌 外部API调用已禁用，直接返回成功")
return CallbackResponse(
success=True,
message="成功",
processed=False,
retry_count=0,
site_id=siteId
)
# 异步调用外部接口，不等待结果
import asyncio
asyncio.create_task(process_external_api_call(db, callback_data, siteId, callback_log_id))
# 立即返回成功响应 # 立即返回成功响应
return CallbackResponse( return CallbackResponse(

16
setup.py
View File

@@ -81,7 +81,7 @@ async def init_tables():
print(f"📋 已创建的表: {', '.join(tables)}") print(f"📋 已创建的表: {', '.join(tables)}")
# 检查必要的表 # 检查必要的表
required_tables = ['callback_logs', 'external_api_logs'] required_tables = ['callback_failure_logs', 'callback_failure_data', 'external_api_logs']
missing_tables = [table for table in required_tables if table not in tables] missing_tables = [table for table in required_tables if table not in tables]
if missing_tables: if missing_tables:
@@ -89,6 +89,20 @@ async def init_tables():
return False return False
else: else:
print("✅ 所有必要的表都已创建") print("✅ 所有必要的表都已创建")
# 检查手机号索引
result = await conn.execute(text("""
SELECT indexname
FROM pg_indexes
WHERE tablename = 'callback_failure_data'
AND indexname = 'idx_phone_number'
"""))
phone_index = result.fetchone()
if phone_index:
print("✅ 手机号索引 'idx_phone_number' 已创建")
else:
print("⚠️ 手机号索引 'idx_phone_number' 未找到")
return True return True

12
test_callback.py
View File

@@ -306,7 +306,7 @@ class TestAITalkCallback:
sample_callback_request sample_callback_request
): ):
"""测试remote_address和server_ip字段是否正确保存""" """测试remote_address和server_ip字段是否正确保存"""
from app.database import CallbackLog from app.database import CallbackFailureLog
# 设置模拟对象 # 设置模拟对象
mock_db = AsyncMock() mock_db = AsyncMock()
@@ -333,9 +333,9 @@ class TestAITalkCallback:
assert mock_db.add.called assert mock_db.add.called
assert mock_db.commit.called assert mock_db.commit.called
# 获取传递给add的CallbackLog对象 # 获取传递给add的CallbackFailureLog对象
call_args = mock_db.add.call_args[0][0] call_args = mock_db.add.call_args[0][0]
assert isinstance(call_args, CallbackLog) assert isinstance(call_args, CallbackFailureLog)
# 验证新字段存在（可能为None，因为测试环境） # 验证新字段存在（可能为None，因为测试环境）
assert hasattr(call_args, 'remote_address') assert hasattr(call_args, 'remote_address')
@@ -356,7 +356,7 @@ class TestAITalkCallback:
sample_callback_request sample_callback_request
): ):
"""测试JSON格式日志记录功能""" """测试JSON格式日志记录功能"""
from app.database import CallbackLog from app.database import CallbackFailureLog
# 设置模拟对象 # 设置模拟对象
mock_db = AsyncMock() mock_db = AsyncMock()
@@ -382,9 +382,9 @@ class TestAITalkCallback:
# 验证log_callback_request被调用 # 验证log_callback_request被调用
assert mock_db.add.called assert mock_db.add.called
# 获取传递给add的CallbackLog对象 # 获取传递给add的CallbackFailureLog对象
call_args = mock_db.add.call_args[0][0] call_args = mock_db.add.call_args[0][0]
assert isinstance(call_args, CallbackLog) assert isinstance(call_args, CallbackFailureLog)
# 验证数据结构适合JSON序列化 # 验证数据结构适合JSON序列化
import json import json