You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
70 lines
2.1 KiB
70 lines
2.1 KiB
"""Redis 消息缓存（可选）"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import time
|
|
|
|
from ..models.message import Message
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class RedisStorage:
|
|
"""基于 Redis 的消息缓存，用于实时消息分发与去重"""
|
|
|
|
def __init__(self, redis_url: str = "redis://localhost:6379/0") -> None:
|
|
self._url = redis_url
|
|
self._client = None
|
|
|
|
def connect(self) -> bool:
|
|
try:
|
|
import redis
|
|
self._client = redis.from_url(self._url, decode_responses=True)
|
|
self._client.ping()
|
|
logger.info("Redis 连接成功: %s", self._url)
|
|
return True
|
|
except ImportError:
|
|
logger.warning("redis-py 未安装，Redis 存储不可用")
|
|
return False
|
|
except Exception:
|
|
logger.exception("Redis 连接失败")
|
|
return False
|
|
|
|
def close(self) -> None:
|
|
if self._client:
|
|
self._client.close()
|
|
|
|
def publish_message(self, msg: Message, channel: str = "wechat:messages") -> bool:
|
|
"""将消息发布到 Redis 频道"""
|
|
if self._client is None:
|
|
return False
|
|
try:
|
|
data = json.dumps(msg.to_dict(), ensure_ascii=False)
|
|
self._client.publish(channel, data)
|
|
return True
|
|
except Exception:
|
|
logger.exception("Redis 发布失败")
|
|
return False
|
|
|
|
def cache_message(self, msg: Message, ttl: int = 3600) -> bool:
|
|
"""将消息缓存到 Redis（用于去重）"""
|
|
if self._client is None:
|
|
return False
|
|
try:
|
|
key = f"wechat:msg:{msg.msg_id}"
|
|
self._client.setex(key, ttl, json.dumps(msg.to_dict(), ensure_ascii=False))
|
|
return True
|
|
except Exception:
|
|
logger.exception("Redis 缓存失败")
|
|
return False
|
|
|
|
def is_message_seen(self, msg_id: str) -> bool:
|
|
"""检查消息是否已处理"""
|
|
if self._client is None:
|
|
return False
|
|
try:
|
|
return self._client.exists(f"wechat:msg:{msg_id}") > 0
|
|
except Exception:
|
|
return False
|
|
|