增强拍一拍事件识别并解析sysmsg pat结构
1. 新增对 sysmsg type=pat 的结构化解析,兼容 fromusername/chatusername/pattedusername/template 等字段。 2. 拍一拍事件优先走XML结构识别,不再仅依赖关键词匹配。 3. 将拍一拍元数据注入响应模板上下文,支持在文案中使用 pat_* 占位符。
This commit is contained in:
@@ -13,6 +13,7 @@ import threading
|
||||
from collections import OrderedDict
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
import xml.etree.ElementTree as ET
|
||||
|
||||
from loguru import logger
|
||||
|
||||
@@ -234,16 +235,70 @@ class FunCommandPlayPlugin(MessagePluginInterface):
|
||||
return "private", sender, sender
|
||||
|
||||
@staticmethod
|
||||
def _extract_event_key(message: Dict[str, Any]) -> str:
|
||||
"""提取事件触发键。
|
||||
def _parse_pat_event_meta(message: Dict[str, Any]) -> Dict[str, str]:
|
||||
"""解析拍一拍事件元数据。
|
||||
|
||||
当前内置:
|
||||
- PAT:拍一拍事件
|
||||
兼容格式示例(用户实测):
|
||||
<sysmsg type="pat">
|
||||
<pat>
|
||||
<fromusername>Jyunere</fromusername>
|
||||
<chatusername>56594698995@chatroom</chatusername>
|
||||
<pattedusername>wxid_xxx</pattedusername>
|
||||
<template><![CDATA["${Jyunere}" 拍了拍 "${wxid_xxx}"]]></template>
|
||||
</pat>
|
||||
</sysmsg>
|
||||
|
||||
检测策略:
|
||||
1. 系统消息文案包含“拍了拍”。
|
||||
2. XML 内容包含 patMsg 结构。
|
||||
返回字段:
|
||||
- event_key: PAT
|
||||
- pat_from_username / pat_chatusername / pat_pattedusername
|
||||
- pat_suffix / pat_suffix_version / pat_template
|
||||
"""
|
||||
content = str(message.get("content", "") or "")
|
||||
if not content:
|
||||
return {}
|
||||
|
||||
normalized = content.strip()
|
||||
if "<sysmsg" not in normalized:
|
||||
return {}
|
||||
|
||||
try:
|
||||
root = ET.fromstring(normalized)
|
||||
except ET.ParseError:
|
||||
return {}
|
||||
|
||||
if root.tag != "sysmsg":
|
||||
return {}
|
||||
if str(root.attrib.get("type", "") or "").strip().lower() != "pat":
|
||||
return {}
|
||||
|
||||
pat_node = root.find("pat")
|
||||
if pat_node is None:
|
||||
return {}
|
||||
|
||||
def _read_text(tag_name: str) -> str:
|
||||
node = pat_node.find(tag_name)
|
||||
if node is None or node.text is None:
|
||||
return ""
|
||||
return str(node.text or "").strip()
|
||||
|
||||
return {
|
||||
"event_key": "PAT",
|
||||
"pat_from_username": _read_text("fromusername"),
|
||||
"pat_chatusername": _read_text("chatusername"),
|
||||
"pat_pattedusername": _read_text("pattedusername"),
|
||||
"pat_suffix": _read_text("patsuffix"),
|
||||
"pat_suffix_version": _read_text("patsuffixversion"),
|
||||
"pat_template": _read_text("template"),
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def _extract_event_key(message: Dict[str, Any]) -> str:
|
||||
"""提取事件触发键。"""
|
||||
# 优先走结构化解析:sysmsg type="pat" 命中即判定为 PAT。
|
||||
pat_meta = FunCommandPlayPlugin._parse_pat_event_meta(message)
|
||||
if pat_meta.get("event_key"):
|
||||
return pat_meta["event_key"]
|
||||
|
||||
content = str(message.get("content", "") or "")
|
||||
full_msg = message.get("full_wx_msg")
|
||||
|
||||
@@ -261,15 +316,20 @@ class FunCommandPlayPlugin(MessagePluginInterface):
|
||||
return ""
|
||||
|
||||
@staticmethod
|
||||
def _build_message_context(message: Dict[str, Any], event_key: str) -> Dict[str, str]:
|
||||
def _build_message_context(message: Dict[str, Any], event_key: str, event_meta: Optional[Dict[str, str]] = None) -> Dict[str, str]:
|
||||
"""构建模板变量上下文。"""
|
||||
room_id = str(message.get("roomid", "") or "")
|
||||
sender = str(message.get("sender", "") or "")
|
||||
return {
|
||||
context = {
|
||||
"sender": sender,
|
||||
"roomid": room_id,
|
||||
"event": event_key or "",
|
||||
}
|
||||
# 拍一拍扩展变量,便于在规则文案中使用更精细的占位符。
|
||||
# 例如:{pat_from_username}、{pat_pattedusername}、{pat_template}
|
||||
for key, value in (event_meta or {}).items():
|
||||
context[str(key)] = str(value or "")
|
||||
return context
|
||||
|
||||
@staticmethod
|
||||
def _render_template(text: str, context: Dict[str, str]) -> str:
|
||||
@@ -331,8 +391,10 @@ class FunCommandPlayPlugin(MessagePluginInterface):
|
||||
# 先做一次匹配并塞入 message,process_message 阶段直接复用,减少重复计算。
|
||||
matched_rule = self._find_match_rule(message)
|
||||
if matched_rule:
|
||||
event_meta = self._parse_pat_event_meta(message)
|
||||
message["_fun_rule_match"] = matched_rule
|
||||
message["_fun_event_key"] = self._extract_event_key(message)
|
||||
message["_fun_event_key"] = str(event_meta.get("event_key") or self._extract_event_key(message))
|
||||
message["_fun_event_meta"] = event_meta
|
||||
return True
|
||||
return False
|
||||
|
||||
@@ -417,7 +479,10 @@ class FunCommandPlayPlugin(MessagePluginInterface):
|
||||
return False, "规则无响应动作"
|
||||
|
||||
event_key = str(message.get("_fun_event_key", "") or "")
|
||||
context = self._build_message_context(message, event_key=event_key)
|
||||
event_meta = message.get("_fun_event_meta")
|
||||
if not isinstance(event_meta, dict):
|
||||
event_meta = self._parse_pat_event_meta(message)
|
||||
context = self._build_message_context(message, event_key=event_key, event_meta=event_meta)
|
||||
|
||||
try:
|
||||
for action in responses:
|
||||
|
||||
Reference in New Issue
Block a user