| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950 |
- """message_id 幂等去重：基于 OrderedDict 实现 LRU + 容量上限。
- 飞书在网络抖动时会重发事件推送，不能依赖 event_id，必须用 message_id 本地去重。
- """
- from __future__ import annotations
- import logging
- import threading
- from collections import OrderedDict
- from typing import Optional
- logger = logging.getLogger(__name__)
- class MessageDedup:
- """线程安全的 LRU message_id 去重器。
- 飞书事件推送在网络抖动时会重发，必须用 message_id 在本地去重，
- 不能依赖 event_id（官方文档明确说明 event_id 不保证幂等）。
- """
- def __init__(self, max_size: int = 2000) -> None:
- if max_size <= 0:
- raise ValueError("max_size 必须为正数")
- self._max_size = max_size
- self._seen: OrderedDict[str, None] = OrderedDict()
- self._lock = threading.Lock()
- def check_and_mark(self, message_id: str) -> bool:
- """若 message_id 首次出现则记录并返回 True；已存在则返回 False（重复）。"""
- if not message_id:
- return False
- with self._lock:
- if message_id in self._seen:
- # 命中：移到末尾（LRU）
- self._seen.move_to_end(message_id)
- return False
- self._seen[message_id] = None
- if len(self._seen) > self._max_size:
- evicted_k, _ = self._seen.popitem(last=False)
- logger.debug("去重缓存淘汰最旧 message_id: %s", evicted_k)
- return True
- def __len__(self) -> int:
- with self._lock:
- return len(self._seen)
- def stats(self) -> dict:
- with self._lock:
- return {"size": len(self._seen), "max_size": self._max_size}
|
