feat(message): 添加 normalize_message_type 函数以标准化消息类型

Author: whatevertogo

src/astrbot_sdk/_message_types.py

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+from typing import Any
+
+_GROUP_MESSAGE_TYPES = {"group", "groupmessage", "group_message"}
+_PRIVATE_MESSAGE_TYPES = {
+    "private",
+    "privatemessage",
+    "private_message",
+    "friend",
+    "friendmessage",
+    "friend_message",
+}
+_OTHER_MESSAGE_TYPES = {"other", "othermessage", "other_message"}
+
+
+def normalize_message_type(
+    value: Any,
+    *,
+    group_id: str | None = None,
+    user_id: str | None = None,
+    empty_default: str = "",
+) -> str:
+    """Collapse SDK-visible message types to canonical values."""
+
+    normalized = str(getattr(value, "value", value) or "").strip().lower()
+    if normalized in _GROUP_MESSAGE_TYPES:
+        return "group"
+    if normalized in _PRIVATE_MESSAGE_TYPES:
+        return "private"
+    if normalized in _OTHER_MESSAGE_TYPES:
+        return "other"
+    if group_id:
+        return "group"
+    if user_id:
+        return "private"
+    if not normalized:
+        return empty_default
+    return "other"

src/astrbot_sdk/context.py

@@ -41,6 +41,7 @@
 
 from ._internal.plugin_logger import PluginLogger
 from ._internal.star_runtime import current_star_instance
+from ._message_types import normalize_message_type
 from .clients import (
     DBClient,
     HTTPClient,
@@ -441,18 +442,7 @@ async def send_message_by_id(
 
     @staticmethod
     def _normalize_compat_message_type(value: str) -> str:
-        normalized = str(value).strip().lower()
-        if normalized in {"groupmessage", "group_message", "group"}:
-            return "group"
-        if normalized in {
-            "privatemessage",
-            "private_message",
-            "private",
-            "friendmessage",
-            "friend_message",
-            "friend",
-        }:
-            return "private"
+        normalized = normalize_message_type(value)
         if not normalized:
             raise AstrBotError.invalid_input("send_message_by_id requires type")
         return normalized

src/astrbot_sdk/events.py

@@ -17,6 +17,7 @@
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any
 
+from ._message_types import normalize_message_type
 from .message.components import (
     At,
     BaseMessageComponent,
@@ -177,7 +178,11 @@ def __init__(
         )
         self.self_id = _coerce_str(self_id)
         self.platform_id = _coerce_str(platform_id) or normalized_platform
-        self.message_type = _coerce_str(message_type).lower()
+        self.message_type = normalize_message_type(
+            message_type,
+            group_id=normalized_group_id,
+            user_id=normalized_user_id,
+        )
         self.sender_name = _coerce_str(sender_name)
         self._is_admin = bool(is_admin)
         self.raw = raw or {}

src/astrbot_sdk/message/session.py

@@ -15,6 +15,8 @@
 
 from dataclasses import dataclass
 
+from .._message_types import normalize_message_type
+
 
 @dataclass(slots=True)
 class MessageSession:
@@ -30,7 +32,7 @@ class MessageSession:
 
     def __post_init__(self) -> None:
         self.platform_id = str(self.platform_id)
-        self.message_type = str(self.message_type).lower()
+        self.message_type = normalize_message_type(self.message_type)
         self.session_id = str(self.session_id)
 
     def __str__(self) -> str: