refactor(platform):harden-dedup-and-extract-event-deduplicator

001VIsir · 001VIsir · commit 7bd3ca412b27 · 2026-03-18T00:37:25.000+08:00
diff --git a/astrbot/core/event_bus.py b/astrbot/core/event_bus.py
@@ -15,14 +15,10 @@
 
 from astrbot.core import logger
 from astrbot.core.astrbot_config_mgr import AstrBotConfigManager
-from astrbot.core.message.utils import (
-    build_content_dedup_key,
-    build_message_id_dedup_key,
-)
 from astrbot.core.pipeline.scheduler import PipelineScheduler
 from astrbot.core.utils.number_utils import safe_positive_float
-from astrbot.core.utils.ttl_registry import TTLKeyRegistry
 
+from .event_dedup import EventDeduplicator
 from .platform import AstrMessageEvent
 
 
@@ -47,65 +43,13 @@ def __init__(
             ),
             default=0.5,
         )
-        self._dedup_registry = TTLKeyRegistry(ttl_seconds=dedup_ttl_seconds)
-
-    @staticmethod
-    def _build_event_content_key(event: AstrMessageEvent) -> str:
-        return build_content_dedup_key(
-            platform_id=str(event.get_platform_id() or ""),
-            unified_msg_origin=str(event.unified_msg_origin or ""),
-            sender_id=str(event.get_sender_id() or ""),
-            text=str(event.get_message_str() or ""),
-            components=event.get_messages(),
-        )
-
-    @staticmethod
-    def _build_event_message_id_key(event: AstrMessageEvent) -> str | None:
-        message_id = getattr(event.message_obj, "message_id", "") or getattr(
-            event.message_obj,
-            "id",
-            "",
-        )
-        return build_message_id_dedup_key(
-            platform_id=str(event.get_platform_id() or ""),
-            unified_msg_origin=str(event.unified_msg_origin or ""),
-            message_id=str(message_id or ""),
-        )
-
-    def _is_duplicate(self, event: AstrMessageEvent) -> bool:
-        if self._dedup_registry.ttl_seconds == 0:
-            return False
-
-        message_id_key = self._build_event_message_id_key(event)
-        if message_id_key is not None:
-            if self._dedup_registry.contains(message_id_key):
-                logger.debug(
-                    "Skip duplicate event in event_bus (by message_id): umo=%s, sender=%s",
-                    event.unified_msg_origin,
-                    event.get_sender_id(),
-                )
-                return True
-            self._dedup_registry.add(message_id_key)
-
-        content_key = self._build_event_content_key(event)
-        if self._dedup_registry.contains(content_key):
-            logger.debug(
-                "Skip duplicate event in event_bus (by content): umo=%s, sender=%s",
-                event.unified_msg_origin,
-                event.get_sender_id(),
-            )
-            if message_id_key is not None:
-                self._dedup_registry.discard(message_id_key)
-            return True
-
-        self._dedup_registry.add(content_key)
-        return False
+        self._deduplicator = EventDeduplicator(ttl_seconds=dedup_ttl_seconds)
 
     async def dispatch(self) -> None:
         # event_queue 由单一消费者处理；去重结构不是线程安全的，按设计仅在此循环中使用。
         while True:
             event: AstrMessageEvent = await self.event_queue.get()
-            if self._is_duplicate(event):
+            if self._deduplicator.is_duplicate(event):
                 continue
             conf_info = self.astrbot_config_mgr.get_conf_info(event.unified_msg_origin)
             conf_id = conf_info["id"]
diff --git a/astrbot/core/event_dedup.py b/astrbot/core/event_dedup.py
@@ -0,0 +1,65 @@
+from astrbot.core import logger
+from astrbot.core.message.utils import (
+    build_content_dedup_key,
+    build_message_id_dedup_key,
+)
+from astrbot.core.utils.ttl_registry import TTLKeyRegistry
+
+from .platform import AstrMessageEvent
+
+
+class EventDeduplicator:
+    def __init__(self, ttl_seconds: float = 0.5) -> None:
+        self._registry = TTLKeyRegistry(ttl_seconds=ttl_seconds)
+
+    def is_duplicate(self, event: AstrMessageEvent) -> bool:
+        if self._registry.ttl_seconds == 0:
+            return False
+
+        message_id_key = self._build_message_id_key(event)
+        if message_id_key is not None:
+            if self._registry.contains(message_id_key):
+                logger.debug(
+                    "Skip duplicate event in event_bus (by message_id): umo=%s, sender=%s",
+                    event.unified_msg_origin,
+                    event.get_sender_id(),
+                )
+                return True
+            self._registry.add(message_id_key)
+
+        content_key = self._build_content_key(event)
+        if self._registry.contains(content_key):
+            logger.debug(
+                "Skip duplicate event in event_bus (by content): umo=%s, sender=%s",
+                event.unified_msg_origin,
+                event.get_sender_id(),
+            )
+            if message_id_key is not None:
+                self._registry.discard(message_id_key)
+            return True
+
+        self._registry.add(content_key)
+        return False
+
+    @staticmethod
+    def _build_content_key(event: AstrMessageEvent) -> str:
+        return build_content_dedup_key(
+            platform_id=str(event.get_platform_id() or ""),
+            unified_msg_origin=str(event.unified_msg_origin or ""),
+            sender_id=str(event.get_sender_id() or ""),
+            text=str(event.get_message_str() or ""),
+            components=event.get_messages(),
+        )
+
+    @staticmethod
+    def _build_message_id_key(event: AstrMessageEvent) -> str | None:
+        message_id = getattr(event.message_obj, "message_id", "") or getattr(
+            event.message_obj,
+            "id",
+            "",
+        )
+        return build_message_id_dedup_key(
+            platform_id=str(event.get_platform_id() or ""),
+            unified_msg_origin=str(event.unified_msg_origin or ""),
+            message_id=str(message_id or ""),
+        )
diff --git a/astrbot/core/platform/sources/qqofficial/qqofficial_platform_adapter.py b/astrbot/core/platform/sources/qqofficial/qqofficial_platform_adapter.py
@@ -5,6 +5,7 @@
 import os
 import random
 import time
+from types import SimpleNamespace
 from typing import Any, cast
 
 import botpy
@@ -90,8 +91,50 @@ def __init__(
         )
         self._lock = asyncio.Lock()
 
-    def _build_content_key(self, content: str, sender_id: str) -> str | None:
-        return build_sender_content_dedup_key(content, sender_id)
+    def _id_dedup_enabled(self, message_id: str) -> bool:
+        return self._message_ids.ttl_seconds > 0 and bool(message_id)
+
+    def _content_dedup_enabled(self) -> bool:
+        return self._content_keys.ttl_seconds > 0
+
+    def _register_message_id(self, message_id: str) -> bool:
+        """Return True if duplicate by ID, False otherwise (and register)."""
+        if self._message_ids.contains(message_id):
+            logger.debug(
+                "[QQOfficial] Duplicate message detected (by ID): %s...",
+                message_id[:50],
+            )
+            return True
+
+        self._message_ids.add(message_id)
+        return False
+
+    def _register_content(
+        self,
+        message_id: str,
+        content: str,
+        sender_id: str,
+        id_dedup_enabled: bool,
+    ) -> bool:
+        """Return True if duplicate by content, False otherwise (and register)."""
+        content_key = build_sender_content_dedup_key(content, sender_id)
+        if content_key is None:
+            logger.debug("[QQOfficial] New message registered: %s...", message_id[:50])
+            return False
+
+        if self._content_keys.contains(content_key):
+            logger.debug(
+                "[QQOfficial] Duplicate message detected (by content): %s",
+                content_key,
+            )
+            # Preserve existing behavior: do not keep message_id on content duplicates
+            if id_dedup_enabled:
+                self._message_ids.discard(message_id)
+            return True
+
+        self._content_keys.add(content_key)
+        logger.debug("[QQOfficial] New message registered: %s...", message_id[:50])
+        return False
 
     async def is_duplicate(
         self,
@@ -100,22 +143,14 @@ async def is_duplicate(
         sender_id: str = "",
     ) -> bool:
         async with self._lock:
-            id_dedup_enabled = self._message_ids.ttl_seconds > 0 and bool(message_id)
-            content_dedup_enabled = self._content_keys.ttl_seconds > 0
+            id_dedup_enabled = self._id_dedup_enabled(message_id)
+            content_dedup_enabled = self._content_dedup_enabled()
 
             if not id_dedup_enabled and not content_dedup_enabled:
                 return False
 
-            # 1) ID-based dedup
-            if id_dedup_enabled:
-                if self._message_ids.contains(message_id):
-                    logger.debug(
-                        "[QQOfficial] Duplicate message detected (by ID): %s...",
-                        message_id[:50],
-                    )
-                    return True
-
-                self._message_ids.add(message_id)
+            if id_dedup_enabled and self._register_message_id(message_id):
+                return True
 
             # 2) Content-based dedup
             if not content_dedup_enabled:
@@ -124,26 +159,12 @@ async def is_duplicate(
                 )
                 return False
 
-            content_key = self._build_content_key(content, sender_id)
-            if content_key is None:
-                logger.debug(
-                    "[QQOfficial] New message registered: %s...", message_id[:50]
-                )
-                return False
-
-            if self._content_keys.contains(content_key):
-                logger.debug(
-                    "[QQOfficial] Duplicate message detected (by content): %s",
-                    content_key,
-                )
-                # Preserve existing behavior: do not keep message_id on content duplicates
-                if id_dedup_enabled:
-                    self._message_ids.discard(message_id)
-                return True
-
-            self._content_keys.add(content_key)
-            logger.debug("[QQOfficial] New message registered: %s...", message_id[:50])
-            return False
+            return self._register_content(
+                message_id=message_id,
+                content=content,
+                sender_id=sender_id,
+                id_dedup_enabled=id_dedup_enabled,
+            )
 
 
 class botClient(Client):
@@ -278,9 +299,11 @@ def __init__(
     async def should_handle_raw_message(self, message: Any) -> bool:
         """Return False if the raw incoming message should be dropped."""
         sender_id = _extract_sender_id(message)
-        content = getattr(message, "content", "") or ""
+        message_id = str(getattr(message, "id", "") or "")
+        raw_content = getattr(message, "content", "")
+        content = str(raw_content or "")
         is_duplicate = await self._deduplicator.is_duplicate(
-            message.id,
+            message_id,
             content,
             sender_id,
         )
@@ -327,7 +350,7 @@ async def _send_by_session_common(
 
         payload: dict[str, Any] = {"content": plain_text, "msg_id": msg_id}
         ret: Any = None
-        send_helper = self.client
+        send_helper = SimpleNamespace(bot=self.client)
 
         if session.message_type == MessageType.GROUP_MESSAGE:
             scene = self._session_scene.get(session.session_id)
