400 lines
14 KiB
Python
400 lines
14 KiB
Python
import gzip
|
|
import json
|
|
import logging
|
|
import re
|
|
from typing import Any, Optional
|
|
|
|
logger = logging.getLogger("douyin_im.protocol")
|
|
|
|
|
|
def _walk_strings(data: Any, depth: int = 0, max_depth: int = 10):
|
|
if depth > max_depth:
|
|
return
|
|
if isinstance(data, dict):
|
|
for v in data.values():
|
|
yield from _walk_strings(v, depth + 1, max_depth)
|
|
elif isinstance(data, list):
|
|
for item in data:
|
|
yield from _walk_strings(item, depth + 1, max_depth)
|
|
elif isinstance(data, str) and data.strip():
|
|
yield data.strip()
|
|
|
|
|
|
def extract_json_objects(raw: bytes | str) -> list[dict]:
|
|
"""从二进制帧中尽量提取 JSON 对象"""
|
|
if isinstance(raw, bytes):
|
|
for codec in ("utf-8", "latin-1"):
|
|
try:
|
|
text = raw.decode(codec, errors="ignore")
|
|
break
|
|
except Exception:
|
|
text = ""
|
|
else:
|
|
text = ""
|
|
else:
|
|
text = raw
|
|
|
|
results = []
|
|
for match in re.finditer(r"\{[^{}]{0,2000}\}", text):
|
|
chunk = match.group(0)
|
|
try:
|
|
obj = json.loads(chunk)
|
|
if isinstance(obj, dict):
|
|
results.append(obj)
|
|
except Exception:
|
|
continue
|
|
return results
|
|
|
|
|
|
def parse_ws_payload(raw: bytes | str) -> list[dict]:
|
|
"""解析 WebSocket 二进制帧，返回标准化消息 dict 列表"""
|
|
messages = []
|
|
|
|
# 尝试 Protobuf 解包
|
|
if isinstance(raw, bytes):
|
|
try:
|
|
from .static import Live_pb2, Response_pb2
|
|
frame = Live_pb2.PushFrame()
|
|
frame.ParseFromString(raw)
|
|
if frame.payloadType == 'pb':
|
|
response = Response_pb2.Response()
|
|
response.ParseFromString(frame.payload)
|
|
body = response.body
|
|
if body.HasField("new_message_notify"):
|
|
notify = body.new_message_notify
|
|
if notify.HasField("message"):
|
|
msg = notify.message
|
|
sender = str(msg.sender)
|
|
msg_type = msg.message_type
|
|
conversation_id = msg.conversation_id
|
|
content_str = msg.content
|
|
server_message_id = str(getattr(msg, "server_message_id", "") or "")
|
|
conv_short_id = str(getattr(msg, "conversation_short_id", "") or "")
|
|
index_in_conv = str(getattr(msg, "index_in_conversation", "") or "")
|
|
|
|
content_json: dict | None = None
|
|
try:
|
|
parsed = json.loads(content_str)
|
|
if isinstance(parsed, dict):
|
|
content_json = parsed
|
|
except Exception:
|
|
content_json = None
|
|
|
|
from .im_debug_log import log_im_message
|
|
|
|
log_im_message(
|
|
direction="in",
|
|
message_type=msg_type,
|
|
conversation_id=conversation_id,
|
|
content=content_json if content_json is not None else content_str,
|
|
fields={
|
|
"conversation_id": conversation_id,
|
|
"conversation_type": getattr(msg, "conversation_type", 1),
|
|
"server_message_id": server_message_id,
|
|
"index_in_conversation": index_in_conv,
|
|
"conversation_short_id": conv_short_id,
|
|
"message_type": msg_type,
|
|
"sender": sender,
|
|
},
|
|
)
|
|
|
|
text_content = ""
|
|
try:
|
|
if content_json is not None:
|
|
if msg_type == 7:
|
|
text_content = content_json.get("text") or ""
|
|
elif msg_type == 70:
|
|
li = content_json.get("link_info") or {}
|
|
text_content = (
|
|
content_json.get("title")
|
|
or li.get("title")
|
|
or content_json.get("text")
|
|
or "[网页链接]"
|
|
)
|
|
elif msg_type == 5:
|
|
text_content = "[表情包]"
|
|
elif msg_type == 17:
|
|
text_content = "[语音]"
|
|
elif msg_type == 27:
|
|
text_content = "[图片]"
|
|
elif msg_type == 8:
|
|
text_content = "[视频]"
|
|
except Exception:
|
|
text_content = content_str or ""
|
|
|
|
if text_content and conversation_id:
|
|
sender_uid = str(msg.sender)
|
|
messages.append({
|
|
"sender_name": sender_uid,
|
|
"sender_uid": sender_uid,
|
|
"content": text_content,
|
|
"conversation_id": conversation_id,
|
|
"unread_count": 1,
|
|
"server_message_id": server_message_id,
|
|
"message_type": msg_type,
|
|
})
|
|
logger.info(
|
|
f"Protobuf WS message parsed: sender={sender}, "
|
|
f"type={msg_type}, content={text_content!r}, conv={conversation_id}"
|
|
)
|
|
return messages
|
|
except Exception as e:
|
|
logger.debug(f"Protobuf WS parse failed: {e}")
|
|
|
|
if isinstance(raw, str):
|
|
payloads = [raw.encode("utf-8", errors="ignore")]
|
|
else:
|
|
payloads = [raw]
|
|
# 尝试 gzip 解压（frontier 常见）
|
|
try:
|
|
payloads.append(gzip.decompress(raw))
|
|
except Exception:
|
|
pass
|
|
|
|
for payload in payloads:
|
|
# 1) 直接 JSON
|
|
if isinstance(payload, bytes):
|
|
text = payload.decode("utf-8", errors="ignore").strip()
|
|
else:
|
|
text = str(payload).strip()
|
|
if text.startswith("{") or text.startswith("["):
|
|
try:
|
|
data = json.loads(text)
|
|
messages.extend(normalize_im_payload(data))
|
|
continue
|
|
except Exception:
|
|
pass
|
|
|
|
# 2) 嵌入 JSON
|
|
for obj in extract_json_objects(payload):
|
|
messages.extend(normalize_im_payload(obj))
|
|
|
|
# 3) 纯文本兜底
|
|
if isinstance(payload, bytes):
|
|
text = payload.decode("utf-8", errors="ignore")
|
|
plain = _extract_plain_text(text)
|
|
if plain:
|
|
messages.append({"content": plain, "sender_name": "", "raw": True})
|
|
|
|
return messages
|
|
|
|
|
|
def normalize_im_payload_from_bytes(raw: bytes) -> list[dict]:
|
|
"""Try to extract conversation/message payloads from binary IM API responses."""
|
|
results: list[dict] = []
|
|
for obj in extract_json_objects(raw):
|
|
results.extend(normalize_im_payload(obj))
|
|
if results:
|
|
return results
|
|
|
|
try:
|
|
from .static import Response_pb2
|
|
response = Response_pb2.Response()
|
|
response.ParseFromString(raw)
|
|
body = response.body
|
|
for field in (
|
|
"get_conversation_info_list_v2_response_body",
|
|
"create_conversation_v2_body",
|
|
):
|
|
if body.HasField(field):
|
|
conv_body = getattr(body, field)
|
|
for conv in conv_body.conversation_info_list:
|
|
conv_id = conv.conversation_id
|
|
peer_uid = ""
|
|
parts = conv_id.split(":")
|
|
if len(parts) >= 4:
|
|
peer_uid = parts[-1]
|
|
label = f"用户{peer_uid[-6:]}" if peer_uid else conv_id
|
|
results.append({
|
|
"conversation_id": conv_id,
|
|
"sender_name": label,
|
|
"content": "",
|
|
"unread_count": 0,
|
|
"peer_uid": peer_uid,
|
|
"conversation_short_id": str(conv.conversation_short_id),
|
|
"ticket": conv.ticket,
|
|
})
|
|
except Exception:
|
|
pass
|
|
return results
|
|
|
|
|
|
def normalize_im_payload(data: Any, depth: int = 0) -> list[dict]:
|
|
"""递归标准化 IM JSON 为 {sender_name, content, conversation_id, unread_count}"""
|
|
if depth > 12:
|
|
return []
|
|
results = []
|
|
|
|
if isinstance(data, list):
|
|
for item in data:
|
|
results.extend(normalize_im_payload(item, depth + 1))
|
|
return results
|
|
|
|
if not isinstance(data, dict):
|
|
return results
|
|
|
|
sender = (
|
|
_pick_str(data, "sender_name", "senderName", "nickname", "nick_name", "userName", "peerName")
|
|
or _pick_nested(data, ("core_info", "user_info", "peer_info"), "nick_name", "nickname", "name")
|
|
)
|
|
sender_avatar = _pick_avatar_url(data)
|
|
content = _pick_message_text(data)
|
|
conv_id = _pick_str(
|
|
data,
|
|
"conversation_id",
|
|
"conversationId",
|
|
"conv_id",
|
|
"cid",
|
|
)
|
|
unread = data.get("unread_count") or data.get("unreadCount") or data.get("unread_cnt") or 0
|
|
try:
|
|
unread = int(unread or 0)
|
|
except (TypeError, ValueError):
|
|
unread = 0
|
|
|
|
if content and len(content) < 500:
|
|
from_self = data.get("is_self") or data.get("isSelf") or data.get("fromSelf") or data.get("self")
|
|
if not from_self:
|
|
results.append({
|
|
"sender_name": sender or "未知用户",
|
|
"sender_avatar": sender_avatar or None,
|
|
"content": content,
|
|
"conversation_id": conv_id or "",
|
|
"unread_count": unread,
|
|
})
|
|
|
|
if sender and unread > 0 and not content:
|
|
results.append({
|
|
"sender_name": sender,
|
|
"sender_avatar": sender_avatar or None,
|
|
"content": "[未读消息]",
|
|
"conversation_id": conv_id or "",
|
|
"unread_count": unread,
|
|
})
|
|
|
|
for key in ("conversations", "conversation_list", "data", "messages", "messagesList", "body"):
|
|
nested = data.get(key)
|
|
if nested is not None:
|
|
results.extend(normalize_im_payload(nested, depth + 1))
|
|
|
|
for value in data.values():
|
|
if isinstance(value, (dict, list)):
|
|
results.extend(normalize_im_payload(value, depth + 1))
|
|
|
|
return results
|
|
|
|
|
|
def _pick_str(data: dict, *keys: str) -> str:
|
|
for key in keys:
|
|
val = data.get(key)
|
|
if isinstance(val, str) and val.strip():
|
|
return val.strip()
|
|
return ""
|
|
|
|
|
|
def _pick_nested(data: dict, parent_keys: tuple, *child_keys: str) -> str:
|
|
for pk in parent_keys:
|
|
nested = data.get(pk)
|
|
if isinstance(nested, dict):
|
|
val = _pick_str(nested, *child_keys)
|
|
if val:
|
|
return val
|
|
return ""
|
|
|
|
|
|
def _avatar_from_value(val: Any) -> str:
|
|
if isinstance(val, str) and val.strip().startswith("http"):
|
|
return val.strip()
|
|
if isinstance(val, dict):
|
|
direct = val.get("url")
|
|
if isinstance(direct, str) and direct.startswith("http"):
|
|
return direct.strip()
|
|
for list_key in ("url_list", "urls"):
|
|
urls = val.get(list_key)
|
|
if isinstance(urls, list):
|
|
for item in urls:
|
|
if isinstance(item, str) and item.startswith("http"):
|
|
return item.strip()
|
|
return ""
|
|
|
|
|
|
def _pick_avatar_url(data: dict) -> str:
|
|
for key in ("avatar_url", "avatarUrl", "head_url", "headUrl", "avatar"):
|
|
url = _avatar_from_value(data.get(key))
|
|
if url:
|
|
return url
|
|
for thumb_key in ("avatar_thumb", "avatar_medium", "avatar_larger", "avatarThumb"):
|
|
url = _avatar_from_value(data.get(thumb_key))
|
|
if url:
|
|
return url
|
|
for parent_key in ("core_info", "user_info", "peer_info", "target_user", "conversation_core_info"):
|
|
nested = data.get(parent_key)
|
|
if isinstance(nested, dict):
|
|
url = _pick_avatar_url(nested)
|
|
if url:
|
|
return url
|
|
return ""
|
|
|
|
|
|
def _pick_message_text(data: dict) -> str:
|
|
for key in (
|
|
"text",
|
|
"content",
|
|
"message",
|
|
"msg",
|
|
"lastMessage",
|
|
"last_msg",
|
|
"preview",
|
|
"brief",
|
|
):
|
|
val = data.get(key)
|
|
if isinstance(val, str) and val.strip():
|
|
return val.strip()
|
|
if isinstance(val, dict):
|
|
inner = _pick_str(val, "text", "content", "message")
|
|
if inner:
|
|
return inner
|
|
return ""
|
|
|
|
|
|
def _extract_plain_text(text: str) -> Optional[str]:
|
|
text = (text or "").strip()
|
|
if not text or len(text) > 200:
|
|
return None
|
|
if text.startswith("{") or text.startswith("["):
|
|
return None
|
|
# 过滤明显二进制垃圾
|
|
printable = sum(1 for c in text if c.isprintable() or c in "\n\r\t")
|
|
if printable / max(len(text), 1) < 0.8:
|
|
return None
|
|
return text
|
|
|
|
|
|
_NON_TEXT_MESSAGE_MARKERS = {
|
|
"[表情包]",
|
|
"[语音]",
|
|
"[图片]",
|
|
"[视频]",
|
|
"[未读消息]",
|
|
}
|
|
|
|
|
|
def should_skip_auto_reply(content: str) -> tuple[bool, str]:
|
|
"""判断收到的内容是否不适合触发自动回复（如纯点赞/表情互动）。"""
|
|
text = (content or "").strip()
|
|
if not text:
|
|
return True, "空消息"
|
|
if text.startswith("{") and '"aweType":11400' in text.replace(" ", ""):
|
|
return True, "社交互动消息(点赞比心)，不宜触发自动回复"
|
|
if "[互动]" in text or "比了个心" in text:
|
|
return True, "社交互动消息，不宜触发自动回复"
|
|
if text in _NON_TEXT_MESSAGE_MARKERS:
|
|
return True, f"非文本消息({text})"
|
|
if re.fullmatch(r"(\[赞\])+", text):
|
|
return True, "表情互动消息(点赞)，抖音通常不允许对此类消息自动回复"
|
|
if re.fullmatch(r"\[[^\]]+\](\[[^\]]+\])*", text) and "http" not in text:
|
|
inner = re.sub(r"[\[\]]", "", text)
|
|
if len(inner) <= 20 and not any(ch.isalnum() for ch in inner):
|
|
return True, f"非文本互动消息({text})"
|
|
return False, ""
|