fix(platform):streamline-dedup-architecture-and-logging

Author: 001VIsir

astrbot/core/event_bus.py

@@ -16,8 +16,8 @@
 from astrbot.core import logger
 from astrbot.core.astrbot_config_mgr import AstrBotConfigManager
 from astrbot.core.message.utils import (
-    build_event_content_dedup_key,
-    build_event_message_id_dedup_key,
+    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
@@ -26,90 +26,86 @@
 from .platform import AstrMessageEvent
 
 
-class EventDeduplicator:
-    """Event deduplicator using TTL-based registry.
-
-    This class handles deduplication of events based on content fingerprint
-    and message ID, with configurable TTL window.
-    """
+class EventBus:
+    """用于处理事件的分发和处理"""
 
-    def __init__(self, ttl_seconds: float = 0.5) -> None:
-        self._registry = TTLKeyRegistry(ttl_seconds)
+    def __init__(
+        self,
+        event_queue: Queue,
+        pipeline_scheduler_mapping: dict[str, PipelineScheduler],
+        astrbot_config_mgr: AstrBotConfigManager,
+    ) -> None:
+        self.event_queue = event_queue  # 事件队列
+        # abconf uuid -> scheduler
+        self.pipeline_scheduler_mapping = pipeline_scheduler_mapping
+        self.astrbot_config_mgr = astrbot_config_mgr
+        dedup_ttl_seconds = safe_positive_float(
+            self.astrbot_config_mgr.g(
+                None,
+                "event_bus_dedup_ttl_seconds",
+                0.5,
+            ),
+            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(),
+        )
 
-    def is_duplicate(self, event: AstrMessageEvent) -> bool:
-        """Check if the event is a duplicate.
+    @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 ""),
+        )
 
-        Returns False immediately if TTL is 0 (deduplication disabled).
-        Short-circuits on message_id key to avoid expensive attachment signature computation.
-        """
-        # TTL of 0 means deduplication is disabled
-        if self._registry.ttl_seconds == 0:
+    def _is_duplicate(self, event: AstrMessageEvent) -> bool:
+        if self._dedup_registry.ttl_seconds == 0:
             return False
 
-        # Short-circuit: check message_id first (cheap) before computing full content key (expensive)
-        message_id_key = build_event_message_id_dedup_key(event)
+        message_id_key = self._build_event_message_id_key(event)
         if message_id_key is not None:
-            if self._registry.contains(message_id_key):
+            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
-            # Register message_id key since we'll process the event
-            self._registry.add(message_id_key)
+            self._dedup_registry.add(message_id_key)
 
-        # Only compute full content key if we get past message_id check
-        content_key = build_event_content_dedup_key(event)
-        if self._registry.contains(content_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 content duplicate, also remove message_id to preserve existing behavior
             if message_id_key is not None:
-                self._registry.discard(message_id_key)
+                self._dedup_registry.discard(message_id_key)
             return True
 
-        # Register content key
-        self._registry.add(content_key)
+        self._dedup_registry.add(content_key)
         return False
 
-
-class EventBus:
-    """用于处理事件的分发和处理"""
-
-    def __init__(
-        self,
-        event_queue: Queue,
-        pipeline_scheduler_mapping: dict[str, PipelineScheduler],
-        astrbot_config_mgr: AstrBotConfigManager,
-    ) -> None:
-        self.event_queue = event_queue  # 事件队列
-        # abconf uuid -> scheduler
-        self.pipeline_scheduler_mapping = pipeline_scheduler_mapping
-        self.astrbot_config_mgr = astrbot_config_mgr
-        dedup_ttl_seconds = safe_positive_float(
-            self.astrbot_config_mgr.g(
-                None,
-                "event_bus_dedup_ttl_seconds",
-                0.5,
-            ),
-            default=0.5,
-        )
-        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._deduplicator.is_duplicate(event):
-                logger.debug(
-                    "Skip duplicate event in event_bus, umo=%s, sender=%s",
-                    event.unified_msg_origin,
-                    event.get_sender_id(),
-                )
+            if self._is_duplicate(event):
                 continue
             conf_info = self.astrbot_config_mgr.get_conf_info(event.unified_msg_origin)
             conf_id = conf_info["id"]

astrbot/core/message/utils.py

@@ -2,14 +2,9 @@
 
 import hashlib
 from collections.abc import Iterable
-from typing import TYPE_CHECKING
 
 from astrbot.core.message.components import BaseMessageComponent, File, Image
 
-if TYPE_CHECKING:
-    from astrbot.core.platform import AstrMessageEvent
-
-
 _MAX_RAW_TEXT_FINGERPRINT_LEN = 256
 
 
@@ -57,46 +52,50 @@ def build_sender_content_dedup_key(content: str, sender_id: str) -> str | None:
     return f"{sender_id}:{content_hash}"
 
 
-def build_event_content_dedup_key(event: "AstrMessageEvent") -> str:
-    """Build a content fingerprint key for EventBus deduplication."""
-    msg_text = str(event.get_message_str() or "").strip()
+def build_content_dedup_key(
+    *,
+    platform_id: str,
+    unified_msg_origin: str,
+    sender_id: str,
+    text: str,
+    components: Iterable[BaseMessageComponent],
+) -> str:
+    """Build a content fingerprint key for event deduplication."""
+    msg_text = str(text or "").strip()
     if len(msg_text) <= _MAX_RAW_TEXT_FINGERPRINT_LEN:
         msg_sig = msg_text
     else:
         msg_hash = hashlib.sha1(msg_text.encode("utf-8")).hexdigest()[:16]
         msg_sig = f"h:{len(msg_text)}:{msg_hash}"
 
-    attach_sig = build_component_dedup_signature(event.get_messages())
-    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 "")
+    attach_sig = build_component_dedup_signature(components)
     return "|".join(
         [
             "content",
-            platform_id,
-            unified_msg_origin,
-            sender_id,
+            str(platform_id or ""),
+            str(unified_msg_origin or ""),
+            str(sender_id or ""),
             msg_sig,
             attach_sig,
         ]
     )
 
 
-def build_event_message_id_dedup_key(event: "AstrMessageEvent") -> str | None:
-    """Build a message_id fingerprint key for EventBus deduplication."""
-    message_id = str(getattr(event.message_obj, "message_id", "") or "")
-    if not message_id:
-        message_id = str(getattr(event.message_obj, "id", "") or "")
-    if not message_id:
+def build_message_id_dedup_key(
+    *,
+    platform_id: str,
+    unified_msg_origin: str,
+    message_id: str,
+) -> str | None:
+    """Build a message_id fingerprint key for event deduplication."""
+    normalized_message_id = str(message_id or "")
+    if not normalized_message_id:
         return None
-
-    platform_id = str(event.get_platform_id() or "")
-    unified_msg_origin = str(event.unified_msg_origin or "")
     return "|".join(
         [
             "message_id",
-            platform_id,
-            unified_msg_origin,
-            message_id,
+            str(platform_id or ""),
+            str(unified_msg_origin or ""),
+            normalized_message_id,
         ]
     )

astrbot/core/platform/sources/qqofficial/qqofficial_platform_adapter.py

@@ -150,16 +150,11 @@ class botClient(Client):
     def set_platform(self, platform: QQOfficialPlatformAdapter) -> None:
         self.platform = platform
 
-    async def _should_drop_message(self, message) -> bool:
-        sender_id = _extract_sender_id(message)
-        content = getattr(message, "content", "") or ""
-        return await self.platform._is_duplicate_message(message.id, content, sender_id)
-
     # 收到群消息
     async def on_group_at_message_create(
         self, message: botpy.message.GroupMessage
     ) -> None:
-        if await self._should_drop_message(message):
+        if not await self.platform.should_handle_raw_message(message):
             return
         abm = QQOfficialPlatformAdapter._parse_from_qqofficial(
             message,
@@ -172,7 +167,7 @@ async def on_group_at_message_create(
 
     # 收到频道消息
     async def on_at_message_create(self, message: botpy.message.Message) -> None:
-        if await self._should_drop_message(message):
+        if not await self.platform.should_handle_raw_message(message):
             return
         abm = QQOfficialPlatformAdapter._parse_from_qqofficial(
             message,
@@ -187,7 +182,7 @@ async def on_at_message_create(self, message: botpy.message.Message) -> None:
     async def on_direct_message_create(
         self, message: botpy.message.DirectMessage
     ) -> None:
-        if await self._should_drop_message(message):
+        if not await self.platform.should_handle_raw_message(message):
             return
         abm = QQOfficialPlatformAdapter._parse_from_qqofficial(
             message,
@@ -199,7 +194,7 @@ async def on_direct_message_create(
 
     # 收到 C2C 消息
     async def on_c2c_message_create(self, message: botpy.message.C2CMessage) -> None:
-        if await self._should_drop_message(message):
+        if not await self.platform.should_handle_raw_message(message):
             return
         abm = QQOfficialPlatformAdapter._parse_from_qqofficial(
             message,
@@ -280,10 +275,16 @@ def __init__(
             cleanup_interval_seconds=cleanup_interval_seconds,
         )
 
-    async def _is_duplicate_message(
-        self, message_id: str, content: str = "", sender_id: str = ""
-    ) -> bool:
-        return await self._deduplicator.is_duplicate(message_id, content, sender_id)
+    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 ""
+        is_duplicate = await self._deduplicator.is_duplicate(
+            message.id,
+            content,
+            sender_id,
+        )
+        return not is_duplicate
 
     async def send_by_session(
         self,

astrbot/core/utils/ttl_registry.py

@@ -5,7 +5,7 @@
 """
 
 import time
-from typing import Hashable, Sequence
+from collections.abc import Hashable, Sequence
 
 
 class TTLKeyRegistry:
@@ -15,6 +15,12 @@ class TTLKeyRegistry:
     deduplication scenarios where old entries should be automatically cleaned up.
     Supports optional cleanup interval throttling to avoid per-access full scans.
 
+    Concurrency note:
+        This class is not thread-safe and does not provide internal locking.
+        It is designed for single-consumer/single-thread usage patterns.
+        If shared across concurrent tasks/threads, callers must provide
+        external synchronization.
+
     Example:
         registry = TTLKeyRegistry(ttl_seconds=0.5)
         if registry.seen("some_key"):