[v4.11.1] - 🛠️ 修复 QQ 官方机器人昵称与提及清洗 (#207)

Fix QQ official nickname and mention handling 

* fix: restore QQ official nicknames and sanitize mentions

* fix: protect QQ official identities and whitespace

* docs: update QQ official nickname guidance

* chore: bump version to v4.11.1

* docs: remove obsolete PDF guidance

* chore: finalize v4.11.1 release prep

* fix(qqofficial): 修复 _sanitize_qq_official_mentions .strip() 吞换行,补充 allow_alphanumeric_user_ids 注释

- .strip() → .strip(' \t'):仅清除首尾空白/制表符,保留用户创作的 \n 换行
- allow_alphanumeric_user_ids:在参数定义处补充注释,说明该标记还控制 ID 正则化和回退显示名
- 响应 Sourcery 代码审查 #207 指出的问题

* chore:  删除不必要的描述

---------

Co-authored-by: SXP-Simon <sxp20061207@163.com>
This commit is contained in:
clown145
2026-07-21 14:53:31 +08:00
committed by GitHub
co-authored by SXP-Simon
parent 20019baa2c
commit f2bcd6801e
13 changed files with 442 additions and 41 deletions
@@ -7,6 +7,9 @@ from astrbot.api.star import Context
from ...infrastructure.persistence.platform_group_registry import PlatformGroupRegistry
from ...utils.logger import logger
_QQ_OFFICIAL_PLATFORM_NAMES = frozenset({"qq_official", "qq_official_webhook"})
_QQ_OFFICIAL_MENTION_PATTERN = re.compile(r"<@!?([A-Za-z0-9_-]+)>")
class MessageProcessingService:
"""
@@ -187,6 +190,12 @@ class MessageProcessingService:
"""从事件中提取消息内容"""
message_parts = []
message = event.message_obj
platform_name = str(event.get_platform_name() or "").strip().lower()
qq_mention_replacements = (
self._extract_qq_official_mention_replacements(event)
if platform_name in _QQ_OFFICIAL_PLATFORM_NAMES
else None
)
# 收集 @ 标记
pending_mentions: Counter[str] = Counter()
@@ -221,6 +230,10 @@ class MessageProcessingService:
text = seg.data.get("text")
if text:
text = self._strip_known_mentions(text, pending_mentions)
if qq_mention_replacements is not None:
text = self._sanitize_qq_official_mentions(
text, qq_mention_replacements
)
message_parts.append({"type": "plain", "text": text})
elif seg_type in ("Image", "image"):
@@ -262,7 +275,12 @@ class MessageProcessingService:
message_parts.append({"type": "video", "url": str(url or "")})
if not message_parts and event.message_str:
message_parts.append({"type": "plain", "text": event.message_str})
fallback_text = str(event.message_str)
if qq_mention_replacements is not None:
fallback_text = self._sanitize_qq_official_mentions(
fallback_text, qq_mention_replacements
)
message_parts.append({"type": "plain", "text": fallback_text})
# 清理空文本段
message_parts = [
@@ -275,6 +293,81 @@ class MessageProcessingService:
return message_parts
@classmethod
def _extract_qq_official_mention_replacements(
cls, event: AstrMessageEvent
) -> dict[str, str]:
message_obj = getattr(event, "message_obj", None)
raw_message = getattr(message_obj, "raw_message", None)
raw_candidates = [raw_message]
nested_message = cls._read_field(raw_message, "message")
if nested_message is not None and nested_message is not raw_message:
raw_candidates.insert(0, nested_message)
mentions = None
for candidate in raw_candidates:
mentions = cls._read_field(candidate, "mentions")
if mentions is not None:
break
replacements: dict[str, str] = {}
if not isinstance(mentions, (list, tuple)):
return replacements
for mention in mentions:
mention_id = str(
cls._read_field(
mention,
"id",
"member_openid",
"memberopenid",
"user_openid",
"useropenid",
)
or ""
).strip()
if not mention_id:
continue
if cls._read_field(mention, "is_you") is True:
replacements[mention_id] = ""
continue
display_name = str(
cls._read_field(mention, "username", "name", "nickname") or ""
).strip()
display_name = display_name.lstrip("@").strip()
if cls._is_placeholder_sender_name(display_name, mention_id):
display_name = "群友"
replacements[mention_id] = f"@{display_name}"
return replacements
@staticmethod
def _sanitize_qq_official_mentions(text: str, replacements: dict[str, str]) -> str:
def replace_mention(match: re.Match[str]) -> str:
mention_id = match.group(1)
if mention_id.lower() in {"all", "everyone"}:
return "@全体成员"
return replacements.get(mention_id, "@群友")
cleaned = _QQ_OFFICIAL_MENTION_PATTERN.sub(replace_mention, str(text))
return re.sub(r"[^\S\r\n]{2,}", " ", cleaned).strip(" \t")
@staticmethod
def _read_field(source: object, *names: str) -> object | None:
if isinstance(source, dict):
for name in names:
if name in source:
return source[name]
return None
for name in names:
value = getattr(source, name, None)
if value is not None:
return value
return None
@staticmethod
def _strip_known_mentions(text: str, pending_mentions: Counter[str]) -> str:
"""从文本中移除已识别的 @ 提及"""
@@ -299,7 +392,7 @@ class MessageProcessingService:
if pending_mentions[mention] <= 0:
pending_mentions.pop(mention, None)
return re.sub(r"\s{2,}", " ", cleaned).strip()
return re.sub(r"[^\S\r\n]{2,}", " ", cleaned).strip()
@staticmethod
def _is_placeholder_sender_name(name: str | None, sender_id: str) -> bool: