From 6ac998ebdf6721a4a12714dce1f2041312d89ca4 Mon Sep 17 00:00:00 2001 From: "mark.tian" Date: Thu, 4 Dec 2025 08:14:21 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=9D=E5=AD=98=E8=AF=B7=E6=B1=82=E5=A4=B1?= =?UTF-8?q?=E8=B4=A5=E8=BF=94=E5=9B=9E500=20=E8=B0=83=E6=95=B4=E8=A1=A8?= =?UTF-8?q?=E7=BB=93=E6=9E=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/database.py | 60 ++++++++++++++------------- app/external_api_processor.py | 76 +++++++++++++++++++++++++++-------- app/routes.py | 73 ++++++++++++++++++++------------- setup.py | 16 +++++++- test_callback.py | 12 +++--- 5 files changed, 158 insertions(+), 79 deletions(-) diff --git a/app/database.py b/app/database.py index 773f1ae..1f161b5 100644 --- a/app/database.py +++ b/app/database.py @@ -1,9 +1,9 @@ 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, Index -from datetime import datetime +from sqlalchemy import Column, String, Integer, DateTime, Text, JSON, Index, text from app.config import settings from app.logger import get_logger +from sqlalchemy.sql.expression import func logger = get_logger("database") @@ -12,49 +12,46 @@ class Base(DeclarativeBase): pass -class CallbackLog(Base): - __tablename__ = "callback_logs" +class CallbackFailureLog(Base): + __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字段 remote_address = Column(String(45), nullable=True, comment="客户端IP地址") server_ip = Column(String(45), nullable=True, comment="服务器IP地址") - 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.now) + request_url = Column(String(500), nullable=False, comment="请求URL") + request_headers = Column(JSON, nullable=False, comment="请求头") + request_body = Column(JSON, nullable=False, comment="请求体") + created_at = Column(DateTime, server_default=func.now(), comment="创建时间") -class CallbackData(Base): - __tablename__ = "callback_data" +class CallbackFailureData(Base): + __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="手机号") task_id = Column(String(100), nullable=False, comment="任务ID") status = Column(Integer, nullable=False, comment="状态") status_description = Column(String(200), nullable=False, comment="状态描述") raw_data = Column(Text, nullable=False, comment="原始回调数据JSON字符串") - created_at = Column(DateTime, default=datetime.now) - - # 为手机号字段添加索引，提高查询性能 - __table_args__ = ( - Index('idx_phone_number', 'phone_number'), - ) + calldate = Column(DateTime, nullable=True, comment="通话日期") + created_at = Column(DateTime, server_default=func.now(), comment="创建时间") class ExternalApiLog(Base): __tablename__ = "external_api_logs" # 重命名表，更通用 - id = Column(Integer, primary_key=True, autoincrement=True) - callback_logs_id = Column(Integer, nullable=False) - 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.now) + id = Column(Integer, primary_key=True, autoincrement=True, comment="日志ID") + callback_failure_log_id = Column(Integer, nullable=False, comment="回调失败日志ID") + request_url = Column(String(500), nullable=False, comment="外部接口请求URL") + request_headers = Column(JSON, nullable=False, comment="外部接口请求头") + request_body = Column(JSON, nullable=False, comment="外部接口请求体") + response_status = Column(Integer, nullable=False, comment="外部接口响应状态码") + response_headers = Column(JSON, nullable=False, comment="外部接口响应头") + response_body = Column(Text, nullable=False, comment="外部接口响应体") + retry_count = Column(Integer, default=0, comment="重试次数") + created_at = Column(DateTime, server_default=func.now(), comment="创建时间") # 创建数据库引擎 @@ -85,6 +82,13 @@ async def init_db(): try: async with engine.begin() as conn: 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("✅ 数据库表结构初始化完成") except Exception as e: logger.error(f"❌ 数据库初始化失败: {e}") diff --git a/app/external_api_processor.py b/app/external_api_processor.py index be137f8..0ca0ab9 100644 --- a/app/external_api_processor.py +++ b/app/external_api_processor.py @@ -8,15 +8,49 @@ from fastapi import HTTPException import httpx import asyncio from typing import Dict, Any, Optional +import json - -from app.database import CallbackData, ExternalApiLog +from app.database import CallbackFailureData, ExternalApiLog, CallbackFailureLog from app.config import settings from app.logger import get_logger 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( db: AsyncSession, phone_number: str @@ -33,8 +67,8 @@ async def check_phone_number_threshold( """ try: # 查询手机号在数据库中出现的次数 - count_query = select(func.count(CallbackData.id)).where( - CallbackData.phone_number == phone_number + count_query = select(func.count(CallbackFailureData.id)).where( + CallbackFailureData.phone_number == phone_number ) result = await db.execute(count_query) phone_count = result.scalar() or 0 @@ -57,7 +91,7 @@ async def check_phone_number_threshold( async def log_external_api_request( db: AsyncSession, - callback_logs_id: int, + callback_failure_log_id: int, request_url: str, request_headers: Dict[str, Any], request_body: Dict[str, Any], @@ -69,7 +103,7 @@ async def log_external_api_request( """记录外部API请求日志""" try: api_log = ExternalApiLog( - callback_logs_id=callback_logs_id, + callback_failure_log_id=callback_failure_log_id, request_url=request_url, request_headers=request_headers, request_body=request_body, @@ -92,7 +126,7 @@ async def call_external_api_with_retry( db: AsyncSession, request_body: Dict[str, Any], max_retries: int = None, - callback_logs_id: int = None + callback_failure_log_id: int = None ) -> tuple[bool, int]: """ 调用外部API并支持重试机制 @@ -101,7 +135,7 @@ async def call_external_api_with_retry( db: 数据库会话 request_body: 请求体 max_retries: 最大重试次数 - callback_logs_id: 回调日志ID + callback_failure_log_id: 回调失败日志ID Returns: tuple[是否成功, 实际重试次数] @@ -132,7 +166,7 @@ async def call_external_api_with_retry( # 记录每次尝试的结果 await log_external_api_request( db=db, - callback_logs_id=callback_logs_id, + callback_failure_log_id=callback_failure_log_id, request_url=settings.external_api_url, request_headers=headers, request_body=request_body, @@ -172,26 +206,37 @@ async def call_external_api_with_retry( async def process_external_api_call( db: AsyncSession, - callback_data: dict, + callback_log_id: int, siteId: str, - callback_log_id: int ): """ 异步调用外部接口 Args: db: 数据库会话 - callback_data: 回调数据 - siteId: 站点ID callback_log_id: 回调日志ID + siteId: 站点ID Returns: 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() - data_list = callback_data.get('data', []) + data_list = request_body.get('data', []) for item in data_list: number_data = item.get('number_data', {}) phone_number = number_data.get('number') @@ -223,12 +268,11 @@ async def process_external_api_call( # 手机号出现的次数少于settings.count_threshold，调用外部API # 调用外部API并支持重试 - request_body = callback_data 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 + callback_failure_log_id=callback_log_id ) if success: diff --git a/app/routes.py b/app/routes.py index 4177dd8..ce68919 100644 --- a/app/routes.py +++ b/app/routes.py @@ -2,7 +2,7 @@ from fastapi import APIRouter, Request, Body, HTTPException, Depends, Path from sqlalchemy.ext.asyncio import AsyncSession 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.config import settings from app.logger import get_logger @@ -18,8 +18,8 @@ async def log_callback_request( request: Request, site_id: str, callback_data: dict -) -> Optional[int]: - """记录回调请求到数据库，返回记录ID""" +) -> tuple[bool, Optional[int]]: + """记录回调请求到数据库，返回操作是否成功和记录ID""" import json # 获取客户端IP地址 @@ -68,7 +68,7 @@ async def log_callback_request( try: # 保存到数据库 - callback_log = CallbackLog( + callback_log = CallbackFailureLog( site_id=site_id, 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, @@ -82,21 +82,23 @@ async def log_callback_request( logger.info(f"✅ 回调请求记录成功保存到数据库，ID: {callback_log.id}") - # 返回记录ID - return callback_log.id + # 返回成功标识和记录ID + return True, callback_log.id except Exception as e: logger.error(f"❌ 保存回调请求到数据库失败: {e}", exc_info=True) - # 不重新抛出异常，避免影响主业务流程 - return None + # 返回失败标识 + return False, None async def save_callback_data_items( db: AsyncSession, - callback_data_items: list + callback_data_items: list, + callback_failure_log_id: int ): """保存callback_data.data中的数据到数据库""" import json + from datetime import datetime try: callback_data_records = [] @@ -116,14 +118,30 @@ async def save_callback_data_items( status = item.get('status', 0) 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字符串保存 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, task_id=task_id, status=status, 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) @@ -159,27 +177,26 @@ async def ai_talk_callback( raise HTTPException(status_code=400, detail="data 参数必须是数组") # 记录回调请求（包含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中的数据 - data_list = callback_data.get('data', []) - if data_list and len(data_list) > 0: - await save_callback_data_items(db, data_list) + if not success: + logger.error("❌ 回调请求日志保存失败") + raise HTTPException(status_code=500, detail="回调请求日志保存失败") + + 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调用 - 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调用已禁用，直接返回成功") - 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( diff --git a/setup.py b/setup.py index b02d575..354e4c2 100644 --- a/setup.py +++ b/setup.py @@ -81,7 +81,7 @@ async def init_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] if missing_tables: @@ -89,6 +89,20 @@ async def init_tables(): return False else: 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 diff --git a/test_callback.py b/test_callback.py index 44dce0d..ffc9632 100644 --- a/test_callback.py +++ b/test_callback.py @@ -306,7 +306,7 @@ class TestAITalkCallback: sample_callback_request ): """测试remote_address和server_ip字段是否正确保存""" - from app.database import CallbackLog + from app.database import CallbackFailureLog # 设置模拟对象 mock_db = AsyncMock() @@ -333,9 +333,9 @@ class TestAITalkCallback: assert mock_db.add.called assert mock_db.commit.called - # 获取传递给add的CallbackLog对象 + # 获取传递给add的CallbackFailureLog对象 call_args = mock_db.add.call_args[0][0] - assert isinstance(call_args, CallbackLog) + assert isinstance(call_args, CallbackFailureLog) # 验证新字段存在（可能为None，因为测试环境） assert hasattr(call_args, 'remote_address') @@ -356,7 +356,7 @@ class TestAITalkCallback: sample_callback_request ): """测试JSON格式日志记录功能""" - from app.database import CallbackLog + from app.database import CallbackFailureLog # 设置模拟对象 mock_db = AsyncMock() @@ -382,9 +382,9 @@ class TestAITalkCallback: # 验证log_callback_request被调用 assert mock_db.add.called - # 获取传递给add的CallbackLog对象 + # 获取传递给add的CallbackFailureLog对象 call_args = mock_db.add.call_args[0][0] - assert isinstance(call_args, CallbackLog) + assert isinstance(call_args, CallbackFailureLog) # 验证数据结构适合JSON序列化 import json