85 lines
2.3 KiB
Python
85 lines
2.3 KiB
Python
|
"""fmessage 系统消息 XML 解析器。
|
|||
|
|
|
|||
|
|
解析微信 4.x Linux 的 fmessage sysmsg XML，提取好友申请信息。
|
|||
|
|
无状态函数，便于单测。
|
|||
|
|
|
|||
|
|
XML 格式（推断，需环境验证）：
|
|||
|
|
<sysmsg type="verifyUser" ...>
|
|||
|
|
<Link ...>
|
|||
|
|
<UserName>xxx@stranger</UserName>
|
|||
|
|
<NickName>申请人昵称</NickName>
|
|||
|
|
<Content>验证消息内容</Content>
|
|||
|
|
<Scene>...</Scene>
|
|||
|
|
</Link>
|
|||
|
|
</sysmsg>
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import logging
|
|||
|
|
import xml.etree.ElementTree as ET
|
|||
|
|
from typing import TYPE_CHECKING, Optional
|
|||
|
|
|
|||
|
|
if TYPE_CHECKING:
|
|||
|
|
# 仅类型注解用，避免与 friend_watcher.py 形成运行时循环 import
|
|||
|
|
from woc_bridge.messaging.friend_watcher import FriendRequestInfo
|
|||
|
|
|
|||
|
|
logger = logging.getLogger("woc-bridge")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def parse_friend_request(
|
|||
|
|
content: str, create_time: int, msg_local_id: int
|
|||
|
|
) -> Optional[FriendRequestInfo]:
|
|||
|
|
"""解析 fmessage 系统消息 XML，提取好友申请信息。
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
content: message_content 原始文本（XML）
|
|||
|
|
create_time: 消息时间戳
|
|||
|
|
msg_local_id: 消息 local_id
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
FriendRequestInfo 或 None（解析失败 / 非 verifyUser 类型）
|
|||
|
|
"""
|
|||
|
|
if not content or not content.strip():
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
root = ET.fromstring(content)
|
|||
|
|
except ET.ParseError:
|
|||
|
|
# 部分消息可能不是合法 XML（如纯文本通知），跳过
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
# 检查是否为 verifyUser 类型 sysmsg
|
|||
|
|
msg_type = root.get("type", "")
|
|||
|
|
if msg_type != "verifyUser":
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
# 提取 Link 节点中的申请人信息
|
|||
|
|
link = root.find(".//Link")
|
|||
|
|
if link is None:
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
stranger_wxid = _text(link, "UserName")
|
|||
|
|
nickname = _text(link, "NickName")
|
|||
|
|
verify_message = _text(link, "Content")
|
|||
|
|
scene = _text(link, "Scene")
|
|||
|
|
|
|||
|
|
if not stranger_wxid:
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
return FriendRequestInfo(
|
|||
|
|
stranger_wxid=stranger_wxid,
|
|||
|
|
nickname=nickname,
|
|||
|
|
verify_message=verify_message,
|
|||
|
|
scene=scene,
|
|||
|
|
raw_xml=content,
|
|||
|
|
create_time=create_time,
|
|||
|
|
msg_local_id=msg_local_id,
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _text(parent: ET.Element, tag: str) -> str:
|
|||
|
|
"""安全提取子元素文本（None 安全）。"""
|
|||
|
|
el = parent.find(tag)
|
|||
|
|
return el.text.strip() if el is not None and el.text else ""
|