定时任务增加redis连接

app/celery_tasks.py

@@ -34,10 +34,14 @@ 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:
-            with engine.connect() as conn:
-                 # 更新任务状态
-                self.update_state(
-                    state='PROGRESS',
-                    meta={'current': 0, 'total': 100, 'status': f'正在获取下一条回调请求日志...'}
-                )
 
-                # 获取一条未完成的回调请求（按创建时间取最小值）
+            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("❌ 查询未完成的回调请求失败，尝试再查一次")
+                    logger.error("❌ 再次查询未完成的回调请求失败，停止任务执行")
-                    query_success, callback_log_data = get_uncompleted_callback_log(conn)
+                    raise Exception("查询未完成的回调请求失败，任务停止执行")
-                    if not query_success:
+            logger.info("3")
-                        logger.error("❌ 再次查询未完成的回调请求失败，停止任务执行")
+            if not callback_log_data:
-                        raise Exception("查询未完成的回调请求失败，任务停止执行")
+                logger.info("📋 没有找到未完成的回调请求")
+                return {"status": "skipped", "message": "没有找到未完成的回调请求"}
             
-                if not callback_log_data:
+            callback_log_id, site_id, request_headers_json, request_body_json = callback_log_data
-                    logger.info("📋 没有找到未完成的回调请求")
+            logger.info(f"📋 获取到未完成的回调请求: ID={callback_log_id}, site_id={site_id}")
-                    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': 25, 'total': 100, 'status': f'获取回调请求成功(ID={callback_log_id}, site_id={site_id})，开始分析处理...'}
+                    meta={'current': 85, 'total': 100, 'status': f'开始推送数据给DTC(ID={callback_log_id}, site_id={site_id})...'}
                 )
                 
-                # 解析请求头和请求体
+                success, retry_count = call_external_api_with_retry(
-                request_headers = json.loads(request_headers_json)
+                    conn=conn,
-                request_body = json.loads(request_body_json)
+                    request_body=filtered_request_body,
-
+                    request_headers=request_headers,
-                # 保存callback_data.data中的数据
+                    max_retries=settings.external_api_retry_max,
-                data_list = request_body.get('data', [])
+                    callback_failure_log_id=callback_log_id
-                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去重
+                if success:
-                data_list = request_body.get('data', [])
+                    logger.info(f"✅ 推送数据给DTC成功，处理的数据项数量: {len(related_records)}, 重试次数: {retry_count} - callback_log_id: {callback_log_id}, site_id: {site_id}")
-                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.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}")
+            logger.info(f"🎉 推送数据给DTC处理完成 - callback_log_id: {callback_log_id}, site_id: {site_id}")
             
-                # 标记CallbackFailureLog为已完成
+            # 标记CallbackFailureLog为已完成
-                mark_callback_log_completed(conn, callback_log_id)
+            mark_callback_log_completed(conn, callback_log_id)
             
-                # 更新任务状态
+            # 更新任务状态
-                self.update_state(
+            self.update_state(
-                    state='PROGRESS',
+                state='PROGRESS',
-                    meta={'current': 100, 'total': 100, 'status': f'标记回调请求日志为已完成(ID={callback_log_id}, site_id={site_id})'}
+                meta={'current': 100, 'total': 100, 'status': f'标记回调请求日志为已完成(ID={callback_log_id}, site_id={site_id})'}
-                )
+            )
             
-                return {
+            return {
-                    "status": "completed",
+                "status": "completed",
-                    "message": "任务完成"
+                "message": "任务完成"
-                }
+            }
-                
-        except Exception as e:
-            logger.error(f"❌ 推送数据给DTC时发生错误: {e}", exc_info=True)
-            return {"status": "error", "message": str(e)}
-            
-        finally:
-            # 释放分布式锁
-            try:
-                lock.release()
-                logger.info(f"🔓 释放任务 {task_name} 的分布式锁")
-            except Exception as e:
-                logger.error(f"❌ 释放任务 {task_name} 的分布式锁失败: {e}")
     
     except Exception as e:
-        logger.error(f"❌ 获取分布式锁失败: {e}", exc_info=True)
+        logger.error(f"❌ 任务执行失败: {e}", exc_info=True)
-        return {"status": "error", "message": f"获取分布式锁失败: {str(e)}"}
+        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}")
 

app/redis_lock.py

@@ -1,5 +1,4 @@
-import redis.asyncio as redis
+import redis
-import asyncio
 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):
+    def __exit__(self, exc_type, exc_val, exc_tb):
-        """异步上下文管理器出口"""
+        """同步上下文管理器出口"""
-        await self.release()
+        self.release()
 
 
 class RedisManager:
     def __init__(self):
         self.redis_client: Optional[redis.Redis] = None
 
-    async def connect(self):
+    def connect(self) -> bool:
-        """连接Redis"""
+        """连接Redis
-        self.redis_client = redis.from_url(
-            settings.redis_url,
-            encoding="utf-8",
-            decode_responses=True
-        )
-        await self.redis_client.ping()
         
-    async def disconnect(self):
+        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
+
+    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:
         """创建分布式锁"""

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)