diff --git a/app/celery_tasks.py b/app/celery_tasks.py index b45d14c..5767028 100644 --- a/app/celery_tasks.py +++ b/app/celery_tasks.py @@ -33,11 +33,15 @@ def push_data_to_dtc_task(self): """ task_name = 'push_data_to_dtc' logger.info(f"🌿 开始推送数据给DTC任务") - - # 获取分布式锁，使用任务名称作为锁标识 - lock = redis_manager.create_lock(f"celery_task:{task_name}", timeout=300) # 5分钟超时 try: + # 获取分布式锁，使用任务名称作为锁标识 + connection_success = redis_manager.connect() + if not connection_success: + logger.error(f"❌ Redis管理器连接失败，任务停止执行") + raise Exception("Redis连接失败") + + lock = redis_manager.create_lock(f"celery_task:{task_name}", timeout=300) # 5分钟超时 # 尝试获取锁 if not lock.acquire(blocking=False): logger.warning(f"⚠️ 任务 {task_name} 正在执行中，跳过本次执行") @@ -45,161 +49,158 @@ def push_data_to_dtc_task(self): logger.info(f"🔒 成功获取任务 {task_name} 的分布式锁") - try: - with engine.connect() as conn: - # 更新任务状态 - self.update_state( - state='PROGRESS', - meta={'current': 0, 'total': 100, 'status': f'正在获取下一条回调请求日志...'} - ) + with engine.connect() as conn: - # 获取一条未完成的回调请求（按创建时间取最小值） + logger.info("1") + # 更新任务状态 + self.update_state( + state='PROGRESS', + meta={'current': 0, 'total': 100, 'status': f'正在获取下一条回调请求日志...'} + ) + logger.info("2") + # 获取一条未完成的回调请求（按创建时间取最小值） + query_success, callback_log_data = get_uncompleted_callback_log(conn) + if not query_success: + logger.error("❌ 查询未完成的回调请求失败，尝试再查一次") query_success, callback_log_data = get_uncompleted_callback_log(conn) if not query_success: - logger.error("❌ 查询未完成的回调请求失败，尝试再查一次") - query_success, callback_log_data = get_uncompleted_callback_log(conn) - 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 = save_callback_data_items(conn, 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 = get_related_records_by_unique_data_list( - conn, 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})，准备转发...'} - ) - - if related_records: - # 创建只包含有效数据项的请求体 - 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 = call_external_api_with_retry( - conn=conn, - 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}") - else: - logger.error(f"❌ 推送数据给DTC失败，重试次数: {retry_count} - callback_log_id: {callback_log_id}, site_id: {site_id}") - else: - logger.info(f"📋 没有有效的通过记录需要处理 - callback_log_id: {callback_log_id}, site_id: {site_id}") - - logger.info(f"🎉 推送数据给DTC处理完成 - callback_log_id: {callback_log_id}, site_id: {site_id}") - - # 标记CallbackFailureLog为已完成 - mark_callback_log_completed(conn, callback_log_id) - - # 更新任务状态 - self.update_state( - state='PROGRESS', - meta={'current': 100, 'total': 100, 'status': f'标记回调请求日志为已完成(ID={callback_log_id}, site_id={site_id})'} - ) - - return { - "status": "completed", - "message": "任务完成" - } - - except Exception as e: - logger.error(f"❌ 推送数据给DTC时发生错误: {e}", exc_info=True) - return {"status": "error", "message": str(e)} + logger.error("❌ 再次查询未完成的回调请求失败，停止任务执行") + raise Exception("查询未完成的回调请求失败，任务停止执行") + logger.info("3") + if not callback_log_data: + logger.info("📋 没有找到未完成的回调请求") + return {"status": "skipped", "message": "没有找到未完成的回调请求"} - finally: - # 释放分布式锁 - try: - lock.release() - logger.info(f"🔓 释放任务 {task_name} 的分布式锁") - except Exception as e: - logger.error(f"❌ 释放任务 {task_name} 的分布式锁失败: {e}") + 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 = save_callback_data_items(conn, 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 = get_related_records_by_unique_data_list( + conn, 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})，准备转发...'} + ) + + if related_records: + # 创建只包含有效数据项的请求体 + 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 = call_external_api_with_retry( + conn=conn, + 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}") + else: + logger.error(f"❌ 推送数据给DTC失败，重试次数: {retry_count} - callback_log_id: {callback_log_id}, site_id: {site_id}") + else: + logger.info(f"📋 没有有效的通过记录需要处理 - callback_log_id: {callback_log_id}, site_id: {site_id}") + + logger.info(f"🎉 推送数据给DTC处理完成 - callback_log_id: {callback_log_id}, site_id: {site_id}") + + # 标记CallbackFailureLog为已完成 + mark_callback_log_completed(conn, callback_log_id) + + # 更新任务状态 + self.update_state( + state='PROGRESS', + meta={'current': 100, 'total': 100, 'status': f'标记回调请求日志为已完成(ID={callback_log_id}, site_id={site_id})'} + ) + + return { + "status": "completed", + "message": "任务完成" + } except Exception as e: - logger.error(f"❌ 获取分布式锁失败: {e}", exc_info=True) - return {"status": "error", "message": f"获取分布式锁失败: {str(e)}"} + logger.error(f"❌ 任务执行失败: {e}", exc_info=True) + return {"status": "error", "message": str(e)} + + finally: + # 释放分布式锁 + try: + if 'lock' in locals(): + lock.release() + logger.info(f"🔓 释放任务 {task_name} 的分布式锁") + except Exception as e: + logger.error(f"❌ 释放任务 {task_name} 的分布式锁失败: {e}") diff --git a/app/redis_lock.py b/app/redis_lock.py index 0deb1d2..7bff8e3 100644 --- a/app/redis_lock.py +++ b/app/redis_lock.py @@ -1,5 +1,4 @@ -import redis.asyncio as redis -import asyncio +import redis import uuid from typing import Optional from app.config import settings @@ -9,6 +8,7 @@ logger = get_redis_logger() class RedisLock: + """Redis分布式锁""" def __init__(self, redis_client: redis.Redis, key: str, timeout: int = None): self.redis_client = redis_client self.key = f"lock:{key}" @@ -16,7 +16,7 @@ class RedisLock: self.identifier = str(uuid.uuid4()) self.acquired = False - async def acquire(self) -> bool: + def acquire(self, blocking: bool = True, timeout: int = None) -> bool: """获取分布式锁""" logger.debug(f"🔒 尝试获取Redis锁: {self.key}") @@ -28,7 +28,7 @@ class RedisLock: end """ - result = await self.redis_client.eval( + result = self.redis_client.eval( lua_script, 1, self.key, @@ -44,7 +44,7 @@ class RedisLock: return self.acquired - async def release(self) -> bool: + def release(self) -> bool: """释放分布式锁""" if not self.acquired: return False @@ -57,7 +57,7 @@ class RedisLock: end """ - result = await self.redis_client.eval( + result = self.redis_client.eval( lua_script, 1, self.key, @@ -67,39 +67,58 @@ class RedisLock: self.acquired = False return bool(result) - async def __aenter__(self): - """异步上下文管理器入口""" + def __enter__(self): + """同步上下文管理器入口""" retries = 0 while retries < settings.redis_lock_max_retries: - if await self.acquire(): + if self.acquire(blocking=False): return self - await asyncio.sleep(settings.redis_lock_retry_delay) + import time + time.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() + def __exit__(self, exc_type, exc_val, exc_tb): + """同步上下文管理器出口""" + 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() + def connect(self) -> bool: + """连接Redis + + Returns: + bool: 连接是否成功 + """ + try: + logger.info(f"🔴 正在连接Redis: {settings.redis_url}") + self.redis_client = redis.from_url( + settings.redis_url, + encoding="utf-8", + decode_responses=True, + socket_connect_timeout=5, + socket_timeout=5, + retry_on_timeout=True + ) + result = self.redis_client.ping() + if result: + logger.info("✅ Redis连接成功") + return True + else: + logger.error("❌ Redis ping失败") + return False + except Exception as e: + logger.error(f"❌ Redis连接失败: {e}") + return False - async def disconnect(self): + def disconnect(self): """断开Redis连接""" if self.redis_client: - await self.redis_client.close() + self.redis_client.close() def create_lock(self, key: str, timeout: int = None) -> RedisLock: """创建分布式锁""" diff --git a/test_redis_manager_connection.py b/test_redis_manager_connection.py new file mode 100644 index 0000000..e9623d8 --- /dev/null +++ b/test_redis_manager_connection.py @@ -0,0 +1,126 @@ +#!/usr/bin/env python3 +""" +测试RedisManager连接功能 +验证connect方法的返回值 +""" + +from app.redis_lock import redis_manager +from app.logger import get_redis_logger + +logger = get_redis_logger() + +def test_connection(): + """测试连接""" + logger.info("🧪 测试Redis连接...") + + # 测试连接 + success = redis_manager.connect() + + if success: + logger.info("✅ 连接成功") + + # 测试创建锁 + try: + lock = redis_manager.create_lock("test:lock", timeout=10) + logger.info("✅ 锁创建成功") + + # 尝试获取锁 + acquired = lock.acquire(blocking=False) + if acquired: + logger.info("✅ 锁获取成功") + + # 释放锁 + released = lock.release() + if released: + logger.info("✅ 锁释放成功") + else: + logger.error("❌ 锁释放失败") + return False + else: + logger.error("❌ 锁获取失败") + return False + + except Exception as e: + logger.error(f"❌ 锁操作失败: {e}") + return False + else: + logger.error("❌ 连接失败") + return False + + return True + +def test_connection_failures(): + """测试连接失败场景""" + logger.info("🧪 测试连接失败处理...") + + # 直接测试连接一个不存在的主机 + from app.config import settings + original_url = settings.redis_url + + try: + # 创建一个新的RedisManager实例用于测试失败场景 + from app.redis_lock import RedisManager + test_manager = RedisManager() + + # 模拟连接失败 - 直接使用无效的URL + import redis + invalid_redis_client = redis.from_url( + "redis://invalid_host:6379/0", + encoding="utf-8", + decode_responses=True, + socket_connect_timeout=1, + socket_timeout=1 + ) + + test_manager.redis_client = invalid_redis_client + + # 尝试ping，这应该会失败 + try: + result = test_manager.redis_client.ping() + logger.error("❌ 连接应该失败但成功了") + return False + except Exception: + logger.info("✅ 连接失败处理正确") + return True + + except Exception as e: + logger.info("✅ 连接失败处理正确") + return True + +def main(): + """主函数""" + logger.info("🚀 开始RedisManager连接测试") + + # 测试1: 连接 + logger.info("\n" + "="*50) + logger.info("测试1: 连接") + logger.info("="*50) + + connection_success = test_connection() + + # 测试2: 连接失败处理 + logger.info("\n" + "="*50) + logger.info("测试2: 连接失败处理") + logger.info("="*50) + + failure_success = test_connection_failures() + + # 断开连接 + redis_manager.disconnect() + logger.info("🔌 Redis连接已断开") + + # 总结 + logger.info("\n" + "="*50) + logger.info("测试结果总结") + logger.info("="*50) + + if connection_success and failure_success: + logger.info("🎉 所有测试通过！RedisManager连接功能正常") + return True + else: + logger.error("❌ 部分测试失败，请检查实现") + return False + +if __name__ == "__main__": + success = main() + exit(0 if success else 1) \ No newline at end of file