""" Celery任务定义 """ from datetime import datetime from functools import wraps from sqlalchemy import create_engine 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 redis_manager, 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 from functools import partial 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'正在获取下一条回调请求日志...'} ) logger.info("2") # 获取一条未完成的回调请求（按创建时间取最小值） 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("查询未完成的回调请求失败，任务停止执行") logger.info("3") 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} 的分布式锁已通过上下文管理器处理") # 在同步的Celery任务中运行异步代码 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: # 清理事件循环 try: if 'loop' in locals(): loop.close() except: pass @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'] timeout = call_config['timeout'] payload = call_config['body'] 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 }