Kris
15fc62d478
- 新增六层UI自动化架构：从Backend到Capabilities的完整分层实现 - 添加WeChat 4.0分辨率适配Profile与图像模板资源 - 实现幂等缓存、熔断器、重试策略、链路追踪与监控指标 - 新增头像下载安全校验、发布朋友圈路径白名单防护 - 优化密钥缓存、DB校验逻辑与初始化流程 - 补充完整错误码体系与启动清场机制
132 lines
3.9 KiB
Python
132 lines
3.9 KiB
Python
"""UI 自动化幂等缓存：防止短时间内重复发送相同消息。
|
||
|
||
key 由 (flow_name, to_wxid, content, client_request_id) 经 SHA256 完整哈希生成，
|
||
5 分钟 TTL，最多 1000 条。命中时返回缓存的 SendResult，避免重复发送。
|
||
|
||
LRU 淘汰 + TTL 过期双策略，保证内存占用有上限且旧数据自动失效。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import logging
|
||
import time
|
||
from collections import OrderedDict
|
||
from typing import Any, Optional
|
||
|
||
logger = logging.getLogger("woc-bridge")
|
||
|
||
|
||
class IdemCache:
|
||
"""幂等缓存：LRU + TTL 双策略。
|
||
|
||
Attributes:
|
||
ttl: 缓存存活秒数，默认 300（5 分钟）
|
||
max_size: 最大缓存条目数，默认 1000
|
||
_cache: OrderedDict[key_str, (value, expire_at)]
|
||
"""
|
||
|
||
def __init__(self, ttl: float = 300.0, max_size: int = 1000) -> None:
|
||
"""初始化幂等缓存。
|
||
|
||
Args:
|
||
ttl: 缓存存活秒数
|
||
max_size: 最大缓存条目数
|
||
"""
|
||
self.ttl = ttl
|
||
self.max_size = max_size
|
||
self._cache: OrderedDict[str, tuple[Any, float]] = OrderedDict()
|
||
|
||
@staticmethod
|
||
def _make_key(
|
||
flow_name: str,
|
||
to_wxid: str,
|
||
content: str,
|
||
client_request_id: str = "",
|
||
) -> str:
|
||
"""生成缓存 key：完整 SHA256（不截断）。
|
||
|
||
Args:
|
||
flow_name: 流程名（如 "send_text"）
|
||
to_wxid: 目标 wxid
|
||
content: 消息内容
|
||
client_request_id: 客户端请求 ID（可选，用于显式幂等）
|
||
|
||
Returns:
|
||
64 字符 hex SHA256 摘要
|
||
"""
|
||
raw = f"{flow_name}|{to_wxid}|{content}|{client_request_id}"
|
||
return hashlib.sha256(raw.encode("utf-8")).hexdigest()
|
||
|
||
def get(
|
||
self,
|
||
flow_name: str,
|
||
to_wxid: str,
|
||
content: str,
|
||
client_request_id: str = "",
|
||
) -> Optional[Any]:
|
||
"""查询缓存。
|
||
|
||
命中且未过期：move_to_end（LRU），返回 value。
|
||
命中但已过期：删除，返回 None。
|
||
未命中：返回 None。
|
||
|
||
Args:
|
||
flow_name: 流程名
|
||
to_wxid: 目标 wxid
|
||
content: 消息内容
|
||
client_request_id: 客户端请求 ID
|
||
|
||
Returns:
|
||
缓存的 value 或 None
|
||
"""
|
||
key = self._make_key(flow_name, to_wxid, content, client_request_id)
|
||
entry = self._cache.get(key)
|
||
if entry is None:
|
||
return None
|
||
value, expire_at = entry
|
||
if time.monotonic() >= expire_at:
|
||
# TTL 过期，删除
|
||
del self._cache[key]
|
||
logger.debug("[idem] key expired: %s", key[:12])
|
||
return None
|
||
# LRU：移到末尾
|
||
self._cache.move_to_end(key)
|
||
logger.debug("[idem] hit: %s", key[:12])
|
||
return value
|
||
|
||
def set(
|
||
self,
|
||
flow_name: str,
|
||
to_wxid: str,
|
||
content: str,
|
||
value: Any,
|
||
client_request_id: str = "",
|
||
) -> None:
|
||
"""写入缓存。
|
||
|
||
若 key 已存在则更新 value 并 move_to_end。
|
||
超限 max_size 时 popitem(last=False) 删最旧。
|
||
|
||
Args:
|
||
flow_name: 流程名
|
||
to_wxid: 目标 wxid
|
||
content: 消息内容
|
||
value: 缓存的值（通常为 FlowResult）
|
||
client_request_id: 客户端请求 ID
|
||
"""
|
||
key = self._make_key(flow_name, to_wxid, content, client_request_id)
|
||
expire_at = time.monotonic() + self.ttl
|
||
if key in self._cache:
|
||
self._cache.move_to_end(key)
|
||
self._cache[key] = (value, expire_at)
|
||
# LRU 淘汰
|
||
while len(self._cache) > self.max_size:
|
||
self._cache.popitem(last=False)
|
||
logger.debug("[idem] set: %s (size=%d)", key[:12], len(self._cache))
|
||
|
||
def clear(self) -> None:
|
||
"""清空缓存。"""
|
||
self._cache.clear()
|
||
logger.info("[idem] cache cleared")
|