389 lines
19 KiB
Python
389 lines
19 KiB
Python
"""
|
||
Celery任务定义
|
||
"""
|
||
|
||
from datetime import datetime
|
||
from functools import wraps
|
||
import json
|
||
from app.celery_app import celery_app
|
||
from app.config import settings
|
||
from app.logger import get_celery_tasks_logger
|
||
from app.callback_service import (
|
||
save_callback_data_items,
|
||
get_uncompleted_callback_log,
|
||
call_external_api_with_retry,
|
||
mark_callback_log_completed,
|
||
get_related_records_by_unique_data_list
|
||
)
|
||
from app.redis_lock import async_redis_manager, distributed_lock
|
||
import requests
|
||
from app.api_config import API_CONFIG, RETRY_CONFIG
|
||
|
||
logger = get_celery_tasks_logger()
|
||
|
||
def conditional_task(enabled=True):
|
||
def decorator(task_func):
|
||
@wraps(task_func)
|
||
def wrapper(*args, **kwargs):
|
||
if not enabled:
|
||
logger.info(f"Task {task_func.__name__} is disabled")
|
||
return None
|
||
return task_func(*args, **kwargs)
|
||
return wrapper
|
||
return decorator
|
||
|
||
# 创建异步数据库连接用于Celery任务
|
||
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
|
||
|
||
# 使用异步数据库连接
|
||
async_engine = create_async_engine(
|
||
settings.database_url,
|
||
echo=settings.debug,
|
||
future=True,
|
||
)
|
||
|
||
AsyncSessionLocal = async_sessionmaker(
|
||
async_engine, class_=AsyncSession, expire_on_commit=False
|
||
)
|
||
|
||
@celery_app.task(bind=True, name='push_data_to_dtc')
|
||
@conditional_task(enabled=settings.enabled_push_data_to_dtc_task)
|
||
def push_data_to_dtc_task(self):
|
||
"""
|
||
推送数据给DTC的Celery任务
|
||
自动获取一条未完成的回调请求进行处理
|
||
"""
|
||
import asyncio
|
||
|
||
async def async_task():
|
||
task_name = 'push_data_to_dtc'
|
||
logger.info(f"🌿 开始推送数据给DTC任务")
|
||
|
||
try:
|
||
# 连接异步Redis
|
||
connection_success = await async_redis_manager.connect()
|
||
if not connection_success:
|
||
logger.error(f"❌ Redis管理器连接失败，任务停止执行")
|
||
raise Exception("Redis连接失败")
|
||
|
||
redis_pool = await async_redis_manager.get_redis_pool()
|
||
|
||
# 使用异步分布式锁
|
||
async with distributed_lock(redis_pool, f"celery_task:{task_name}", timeout=300) as lock_acquired:
|
||
if not lock_acquired:
|
||
logger.warning(f"⚠️ 任务 {task_name} 正在执行中，跳过本次执行")
|
||
return {"status": "skipped", "message": f"任务 {task_name} 正在执行中，跳过本次执行"}
|
||
|
||
logger.info(f"🔒 成功获取任务 {task_name} 的分布式锁")
|
||
|
||
async with AsyncSessionLocal() as db:
|
||
# 更新任务状态
|
||
self.update_state(
|
||
state='PROGRESS',
|
||
meta={'current': 0, 'total': 100, 'status': f'正在获取下一条回调请求日志...'}
|
||
)
|
||
|
||
# 获取一条未完成的回调请求（按创建时间取最小值）
|
||
query_success, callback_log_data = await get_uncompleted_callback_log(db)
|
||
if not query_success:
|
||
logger.error("❌ 查询未完成的回调请求失败，尝试再查一次")
|
||
query_success, callback_log_data = await get_uncompleted_callback_log(db)
|
||
if not query_success:
|
||
logger.error("❌ 再次查询未完成的回调请求失败，停止任务执行")
|
||
raise Exception("查询未完成的回调请求失败，任务停止执行")
|
||
|
||
if not callback_log_data:
|
||
logger.info("📋 没有找到未完成的回调请求")
|
||
return {"status": "skipped", "message": "没有找到未完成的回调请求"}
|
||
|
||
callback_log_id, site_id, request_headers_json, request_body_json = callback_log_data
|
||
logger.info(f"📋 获取到未完成的回调请求: ID={callback_log_id}, site_id={site_id}")
|
||
|
||
# 更新任务状态
|
||
self.update_state(
|
||
state='PROGRESS',
|
||
meta={'current': 25, 'total': 100, 'status': f'获取回调请求成功(ID={callback_log_id}, site_id={site_id})，开始分析处理...'}
|
||
)
|
||
|
||
# 解析请求头和请求体
|
||
request_headers = json.loads(request_headers_json)
|
||
request_body = json.loads(request_body_json)
|
||
|
||
# 保存callback_data.data中的数据
|
||
data_list = request_body.get('data', [])
|
||
if data_list and len(data_list) > 0:
|
||
save_success = await save_callback_data_items(db, data_list, callback_log_id)
|
||
if not save_success:
|
||
logger.error(f"❌ 保存callback_data:{callback_log_id}失败，停止任务执行")
|
||
raise Exception(f"保存callback_data:{callback_log_id}失败，任务停止执行")
|
||
|
||
# 更新任务状态
|
||
self.update_state(
|
||
state='PROGRESS',
|
||
meta={'current': 50, 'total': 100, 'status': f'请求体中通话明细处理成功(ID={callback_log_id}, site_id={site_id})，开始过滤需要转发的通话记录...'}
|
||
)
|
||
|
||
# 根据手机号、task_id、user_id给data_list去重
|
||
data_list = request_body.get('data', [])
|
||
unique_data_list = []
|
||
seen_records = set()
|
||
original_count = len(data_list)
|
||
|
||
for item in data_list:
|
||
# 提取手机号
|
||
number_data = item.get('number_data', {})
|
||
phone_number = number_data.get('number', '')
|
||
|
||
# 提取task_id
|
||
task = item.get('task', {})
|
||
task_id = task.get('id', '')
|
||
|
||
# 提取user_id
|
||
user_id = item.get('user_id', '')
|
||
|
||
# 创建唯一标识
|
||
unique_key = (phone_number, task_id, user_id)
|
||
|
||
# 如果这个组合没见过，则添加到去重列表中
|
||
if unique_key not in seen_records:
|
||
seen_records.add(unique_key)
|
||
unique_data_list.append(item)
|
||
|
||
logger.info(f"🔄 数据去重完成: 原始数据 {original_count} 条，去重后 {len(unique_data_list)} 条 - callback_log_id: {callback_log_id}, site_id: {site_id}")
|
||
|
||
if len(unique_data_list) < original_count:
|
||
logger.info(f"🗑️ 移除了 {original_count - len(unique_data_list)} 条重复数据 - callback_log_id: {callback_log_id}, site_id: {site_id}")
|
||
|
||
# 查询相关数据：根据手机号、task_id、user_id作为条件，按创建时间排序获取前三条记录
|
||
query_success, related_records = await get_related_records_by_unique_data_list(
|
||
db, unique_data_list, callback_log_id, settings.count_threshold
|
||
)
|
||
|
||
if not query_success:
|
||
logger.error("❌ 查询相关记录失败，停止任务执行")
|
||
raise Exception("查询相关记录失败，任务停止执行")
|
||
|
||
# 没有有效通过记录的情况
|
||
if not related_records:
|
||
logger.info(f"📋 没有有效的通过记录需要处理，直接返回 - callback_log_id: {callback_log_id}, site_id={site_id}")
|
||
return {"status": "completed", "message": "没有有效的通过记录需要处理"}
|
||
|
||
# 更新任务状态
|
||
self.update_state(
|
||
state='PROGRESS',
|
||
meta={'current': 70, 'total': 100, 'status': f'需要推送的通过记录已获取成功(ID={callback_log_id}, site_id={site_id})，准备转发...'}
|
||
)
|
||
|
||
# 由于已经确认有记录，直接进入推送逻辑
|
||
# 创建只包含有效数据项的请求体
|
||
filtered_request_body = request_body.copy()
|
||
filtered_request_body['data'] = related_records
|
||
|
||
# 更新任务状态
|
||
self.update_state(
|
||
state='PROGRESS',
|
||
meta={'current': 85, 'total': 100, 'status': f'开始推送数据给DTC(ID={callback_log_id}, site_id={site_id})...'}
|
||
)
|
||
|
||
success, retry_count = await call_external_api_with_retry(
|
||
db=db,
|
||
request_body=filtered_request_body,
|
||
request_headers=request_headers,
|
||
max_retries=settings.external_api_retry_max,
|
||
callback_failure_log_id=callback_log_id
|
||
)
|
||
|
||
if success:
|
||
logger.info(f"✅ 推送数据给DTC成功，处理的数据项数量: {len(related_records)}, 重试次数: {retry_count} - callback_log_id: {callback_log_id}, site_id: {site_id}")
|
||
logger.info(f"🎉 推送数据给DTC处理完成 - callback_log_id: {callback_log_id}, site_id: {site_id}")
|
||
else:
|
||
logger.error(f"❌ 推送数据给DTC失败，重试次数: {retry_count} - callback_log_id: {callback_log_id}, site_id: {site_id}")
|
||
|
||
# 标记CallbackFailureLog为已完成
|
||
mark_success = await mark_callback_log_completed(db, callback_log_id)
|
||
|
||
if not mark_success:
|
||
logger.error(f"❌ 标记回调请求日志状态失败，任务终止执行")
|
||
raise Exception("标记回调请求日志状态失败，任务终止执行")
|
||
|
||
# 更新任务状态
|
||
self.update_state(
|
||
state='PROGRESS',
|
||
meta={'current': 100, 'total': 100, 'status': f'标记回调请求日志状态(ID={callback_log_id}, site_id={site_id})'}
|
||
)
|
||
|
||
logger.info(f"✅ 回调日志 {callback_log_id} 已成功标记为完成")
|
||
return {
|
||
"status": "completed",
|
||
"message": "任务完成，回调日志已标记为完成"
|
||
}
|
||
|
||
except Exception as e:
|
||
logger.error(f"❌ 任务执行失败: {e}", exc_info=True)
|
||
return {"status": "error", "message": str(e)}
|
||
|
||
finally:
|
||
# 分布式锁会通过上下文管理器自动释放
|
||
logger.debug(f"🔓 任务 {task_name} 的分布式锁已通过上下文管理器处理")
|
||
|
||
# 清理Redis连接
|
||
try:
|
||
await async_redis_manager.close()
|
||
logger.debug(f"🔌 Redis连接已清理")
|
||
except Exception as e:
|
||
logger.warning(f"⚠️ 清理Redis连接时出现警告: {e}")
|
||
|
||
# 在同步的Celery任务中运行异步代码
|
||
loop = None
|
||
try:
|
||
# 创建新的事件循环
|
||
loop = asyncio.new_event_loop()
|
||
asyncio.set_event_loop(loop)
|
||
return loop.run_until_complete(async_task())
|
||
except Exception as e:
|
||
logger.error(f"❌ 异步任务执行失败: {e}", exc_info=True)
|
||
return {"status": "error", "message": str(e)}
|
||
finally:
|
||
# 确保所有异步任务完成后再关闭事件循环
|
||
if loop and not loop.is_closed():
|
||
try:
|
||
# 等待所有待处理的任务完成
|
||
pending = asyncio.all_tasks(loop)
|
||
if pending:
|
||
logger.debug(f"⏳ 等待 {len(pending)} 个异步任务完成...")
|
||
loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True))
|
||
|
||
# 关闭事件循环
|
||
loop.close()
|
||
logger.debug(f"🔌 事件循环已正确关闭")
|
||
except Exception as e:
|
||
logger.warning(f"⚠️ 关闭事件循环时出现警告: {e}")
|
||
else:
|
||
logger.debug(f"🔌 事件循环已关闭或未创建")
|
||
|
||
|
||
@celery_app.task(bind=True, name='call_api', max_retries=None)
|
||
@conditional_task(enabled=settings.enabled_call_api_task)
|
||
def execute_call_api_task(self):
|
||
"""遍历API配置文件中的所有API信息并调用接口"""
|
||
current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
||
logger.info(f"开始遍历API配置文件执行任务 - 当前时间: {current_time}")
|
||
|
||
# 获取重试配置
|
||
max_retries = RETRY_CONFIG['max_retries']
|
||
retry_delay = RETRY_CONFIG['retry_delay']
|
||
retry_on_status = RETRY_CONFIG['retry_on_status']
|
||
|
||
# 遍历所有API信息
|
||
results = []
|
||
success_count = 0
|
||
|
||
for api_key, call_config in API_CONFIG.items():
|
||
logger.info(f"开始处理API配置: {api_key}")
|
||
|
||
call_url = call_config['url']
|
||
method = call_config['method'].upper()
|
||
headers = call_config['headers'].copy() # 复制headers避免修改原配置
|
||
timeout = call_config['timeout']
|
||
payload = call_config['body']
|
||
|
||
# 从环境变量获取Authorization并添加到headers
|
||
if settings.api_authorization_token:
|
||
headers['Authorization'] = settings.api_authorization_token
|
||
|
||
try:
|
||
logger.info(f"正在调用任务API: {method} {call_url}")
|
||
|
||
# 根据method选择请求方式
|
||
if method == 'POST':
|
||
response = requests.post(call_url, headers=headers, json=payload, timeout=timeout)
|
||
elif method == 'GET':
|
||
response = requests.get(call_url, headers=headers, params=payload, timeout=timeout)
|
||
elif method == 'PUT':
|
||
response = requests.put(call_url, headers=headers, json=payload, timeout=timeout)
|
||
elif method == 'DELETE':
|
||
response = requests.delete(call_url, headers=headers, timeout=timeout)
|
||
else:
|
||
raise ValueError(f"不支持的HTTP方法: {method}")
|
||
|
||
# 检查响应状态
|
||
if response.status_code in [200, 201, 204]: # 成功状态码
|
||
try:
|
||
result_data = response.json()
|
||
except:
|
||
result_data = {'response': response.text}
|
||
|
||
logger.info(f"任务API调用成功，响应数据: {result_data}")
|
||
print(f"[{current_time}] ✅ 任务API成功 ({api_key}) - {method} {call_url}")
|
||
success_count += 1
|
||
|
||
results.append({
|
||
'call_api_key': api_key,
|
||
'status': 'success',
|
||
'message': f'任务API调用成功 ({method})',
|
||
'response': result_data,
|
||
'timestamp': current_time,
|
||
'retry_count': getattr(self.request, 'retries', 0)
|
||
})
|
||
elif response.status_code in retry_on_status:
|
||
# 需要重试的状态码
|
||
current_retry = getattr(self.request, 'retries', 0)
|
||
if current_retry < max_retries:
|
||
logger.warning(f"任务API需要重试，当前重试次数: {current_retry + 1}/{max_retries}")
|
||
print(f"[{current_time}] 🔄 任务API失败，正在重试 ({current_retry + 1}/{max_retries}) - {api_key} - 状态码: {response.status_code}")
|
||
raise self.retry(countdown=retry_delay, exc=Exception(f"任务API失败，状态码: {response.status_code}"))
|
||
else:
|
||
logger.error(f"任务API失败，已达到最大重试次数: {max_retries}")
|
||
print(f"[{current_time}] ❌ 任务API失败，已达到最大重试次数 - {api_key} - 状态码: {response.status_code}")
|
||
results.append({
|
||
'call_api_key': api_key,
|
||
'status': 'failed',
|
||
'message': f'任务API失败，已达到最大重试次数: {max_retries}',
|
||
'response': response.text,
|
||
'timestamp': current_time,
|
||
'retry_count': current_retry + 1
|
||
})
|
||
else:
|
||
# 其他失败状态码，不重试
|
||
logger.error(f"任务API失败，状态码: {response.status_code}, 响应: {response.text}")
|
||
print(f"[{current_time}] ❌ 任务API失败 - {api_key} - 状态码: {response.status_code}")
|
||
results.append({
|
||
'call_api_key': api_key,
|
||
'status': 'failed',
|
||
'message': f'任务API失败，状态码: {response.status_code}',
|
||
'response': response.text,
|
||
'timestamp': current_time,
|
||
'retry_count': getattr(self.request, 'retries', 0)
|
||
})
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
# 网络异常，需要重试
|
||
current_retry = getattr(self.request, 'retries', 0)
|
||
if current_retry < max_retries:
|
||
logger.warning(f"任务API异常，需要重试，当前重试次数: {current_retry + 1}/{max_retries}")
|
||
print(f"[{current_time}] 🔄 任务API异常，正在重试 ({current_retry + 1}/{max_retries}) - {api_key} - 异常: {str(e)}")
|
||
raise self.retry(countdown=retry_delay, exc=e)
|
||
else:
|
||
logger.error(f"任务API异常，已达到最大重试次数: {max_retries}")
|
||
print(f"[{current_time}] ❌ 任务API异常，已达到最大重试次数 - {api_key} - 异常: {str(e)}")
|
||
results.append({
|
||
'call_api_key': api_key,
|
||
'status': 'error',
|
||
'message': f'任务API异常，已达到最大重试次数: {max_retries}',
|
||
'error': str(e),
|
||
'timestamp': current_time,
|
||
'retry_count': current_retry + 1
|
||
})
|
||
|
||
# 汇总结果
|
||
total_count = len(results)
|
||
print(f"[{current_time}] 📊 任务汇总 - 成功: {success_count}/{total_count}")
|
||
|
||
return {
|
||
'status': 'completed',
|
||
'total_apis': total_count,
|
||
'success_count': success_count,
|
||
'failed_count': total_count - success_count,
|
||
'results': results,
|
||
'timestamp': current_time
|
||
} |