check_queue.py

#!/usr/bin/env python3
"""
检查队列状态
"""

import sys
import os
import json
sys.path.append(os.path.dirname(__file__))

import redis.asyncio as redis
from src.utils.config import config

async def check_queue():
    """检查队列状态"""
    print("🔍 检查队列状态...")
    
    # 连接Redis
    try:
        redis_url = config.redis.url
        if "redis:" in redis_url and not os.path.exists("/.dockerenv"):
            redis_url = redis_url.replace("redis://redis:", "redis://localhost:")
        
        r = redis.from_url(redis_url, decode_responses=True)
        await r.ping()
        print("✅ Redis连接成功")
    except Exception as e:
        print(f"❌ Redis连接失败: {e}")
        return
    
    try:
        # 检查email_queue
        email_queue_len = await r.llen("email_queue")
        print(f"\n📧 email_queue长度: {email_queue_len}")
        
        if email_queue_len > 0:
            print("email_queue内容:")
            for i in range(email_queue_len):
                item = await r.lindex("email_queue", i)
                if item:
                    try:
                        data = json.loads(item)
                        print(f"  [{i}]: {data['type']} - {data['ticker']}")
                    except:
                        print(f"  [{i}]: {item[:100]}...")
        else:
            print("email_queue为空")
        
        # 检查event_queue
        event_queue_len = await r.llen("event_queue")
        print(f"\n📨 event_queue长度: {event_queue_len}")
        
        if event_queue_len > 0:
            print("event_queue内容:")
            for i in range(event_queue_len):
                item = await r.lindex("event_queue", i)
                if item:
                    try:
                        data = json.loads(item)
                        print(f"  [{i}]: {data['type']} - {data['ticker']}")
                    except:
                        print(f"  [{i}]: {item[:100]}...")
        else:
            print("event_queue为空")
        
        # 检查是否有其他相关键
        print(f"\n🔍 检查其他相关键...")
        keys = await r.keys("*")
        email_related = [k for k in keys if "email" in k.lower()]
        event_related = [k for k in keys if "event" in k.lower()]
        
        if email_related:
            print(f"邮件相关键: {email_related}")
        if event_related:
            print(f"事件相关键: {event_related}")
        
    except Exception as e:
        print(f"❌ 检查队列失败: {e}")
        import traceback
        traceback.print_exc()
    
    finally:
        await r.aclose()

if __name__ == "__main__":
    import asyncio
    asyncio.run(check_queue())