58 lines
1.9 KiB
Python
58 lines
1.9 KiB
Python
"""
|
||
消息解析模块
|
||
负责从企业微信 SDK 的消息帧中提取结构化数据。
|
||
|
||
frame 结构: {cmd, headers, body: {msgtype, text, from, ...}}
|
||
注意：msgtype 在 body 内部，不在 frame 顶层。
|
||
"""
|
||
|
||
|
||
def extract_text_content(frame: dict) -> str:
|
||
"""
|
||
从消息帧中提取文本内容。
|
||
根据不同的消息类型，提取可用于 Agent 处理的文本描述。
|
||
"""
|
||
body = frame.get("body", {})
|
||
msg_type = body.get("msgtype", "")
|
||
|
||
if msg_type == "text":
|
||
return body.get("text", {}).get("content", "")
|
||
|
||
elif msg_type == "image":
|
||
return "[用户发送了一张图片]"
|
||
|
||
elif msg_type == "voice":
|
||
return body.get("voice", {}).get("content", "[用户发送了一条语音消息]")
|
||
|
||
elif msg_type == "file":
|
||
filename = body.get("file", {}).get("filename", "未知文件")
|
||
return f"[用户发送了一个文件: {filename}]"
|
||
|
||
elif msg_type == "video":
|
||
return "[用户发送了一段视频]"
|
||
|
||
elif msg_type == "mixed":
|
||
mixed_items = body.get("mixed", {}).get("content", [])
|
||
text_parts = []
|
||
for item in mixed_items:
|
||
if isinstance(item, dict) and item.get("type") == "text":
|
||
text_parts.append(item.get("content", ""))
|
||
elif isinstance(item, str):
|
||
text_parts.append(item)
|
||
return "\n".join(text_parts) if text_parts else "[用户发送了图文混排消息]"
|
||
|
||
return f"[未知消息类型: {msg_type}]"
|
||
|
||
|
||
def extract_metadata(frame: dict) -> dict:
|
||
"""提取消息的元数据（msgtype 在 body 内部）"""
|
||
body = frame.get("body", {})
|
||
from_info = body.get("from", {})
|
||
return {
|
||
"user_id": from_info.get("userid", ""),
|
||
"chat_id": body.get("chatid", ""),
|
||
"chat_type": body.get("chattype", ""),
|
||
"msg_id": body.get("msgid", ""),
|
||
"msg_type": body.get("msgtype", ""),
|
||
}
|