111 lines
4.1 KiB
Python
111 lines
4.1 KiB
Python
"""抖音 IM 社交互动类消息（如 aweType=11400 点赞比心）构造。
|
|
|
|
此类消息通常由客户端在作品互动链路上产生；Web IM 发送时 content JSON 结构与分享视频类似，
|
|
message_type 实测多为 8（与 aweType=800 视频分享同族）。若发送失败可尝试 im_message_type=114。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import os
|
|
from typing import Any
|
|
|
|
# 与 aweType=800 视频分享消息同族的 protobuf message_type
|
|
MSG_TYPE_INTERACTION_DEFAULT = 8
|
|
|
|
# 用户抓包样本：「向对方的点赞比了个心」
|
|
LIKE_HEART_SAMPLE: dict[str, Any] = {
|
|
"aweType": 11400,
|
|
"comment": "",
|
|
"content_thumb": {
|
|
"uri": "100x100/aweme-avatar/tos-cn-i-0813c001_2e91b5a5a18345b196038a72330a917f",
|
|
"url_list": [
|
|
"https://p3.douyinpic.com/aweme/100x100/aweme-avatar/tos-cn-i-0813c001_2e91b5a5a18345b196038a72330a917f.jpeg?biz_tag=aweme_im&from=3782654143&s=im_102&sc=avatar",
|
|
],
|
|
},
|
|
"content_title": "",
|
|
"cover_height": 720.0,
|
|
"cover_url": {
|
|
"data_size": 0,
|
|
"height": 720,
|
|
"uri": "tos-cn-p-0015c000-ce/o8fy5FTAEBBiTUwv8VBonpIjfEFU0S9IKixAPC",
|
|
"url_list": [
|
|
"https://p3-sign.douyinpic.com/obj/tos-cn-p-0015c000-ce/o8fy5FTAEBBiTUwv8VBonpIjfEFU0S9IKixAPC?lk3s=9698d6d2&x-expires=1814252400&x-signature=oOR8Kz7HyoL1tcn%2B7xyD7ZEcnMU%3D&from=2526598866&s=im_102&se=false&sc=cover&biz_tag=aweme_im&l=021782716851230fdbddc0c0001054a000000000000005676aa83",
|
|
],
|
|
"width": 720,
|
|
},
|
|
"cover_width": 720.0,
|
|
"createdAt": 0,
|
|
"is_card": False,
|
|
"itemId": "7636216221181053883",
|
|
"msgHint": "",
|
|
"push_internal_object_title": "对你的点赞比了个心",
|
|
"push_internal_object_title_v2": "",
|
|
"ref_type": 2,
|
|
"scene_type": 0,
|
|
"subject_title": "向对方的点赞比了个心",
|
|
"subject_title_v2": "",
|
|
"uid": "101081217900",
|
|
"without_panel": 0,
|
|
}
|
|
|
|
|
|
def default_interaction_message_type() -> int:
|
|
raw = (os.getenv("KEFU_INTERACTION_MSG_TYPE") or "").strip()
|
|
if raw.isdigit():
|
|
return int(raw)
|
|
return MSG_TYPE_INTERACTION_DEFAULT
|
|
|
|
|
|
def _merge_media_object(base: dict[str, Any], override: Any) -> dict[str, Any]:
|
|
if isinstance(override, dict):
|
|
merged = copy.deepcopy(base)
|
|
merged.update(override)
|
|
return merged
|
|
return copy.deepcopy(base)
|
|
|
|
|
|
def build_like_heart_payload(spec: dict[str, Any], *, my_uid: str = "") -> dict[str, Any]:
|
|
"""构造「对点赞比了个心」互动消息体（aweType=11400）。"""
|
|
payload = copy.deepcopy(LIKE_HEART_SAMPLE)
|
|
|
|
if spec.get("payload") and isinstance(spec["payload"], dict):
|
|
payload.update(copy.deepcopy(spec["payload"]))
|
|
|
|
if spec.get("item_id") or spec.get("itemId"):
|
|
payload["itemId"] = str(spec.get("item_id") or spec.get("itemId"))
|
|
if spec.get("subject_title"):
|
|
payload["subject_title"] = str(spec["subject_title"])
|
|
if spec.get("push_internal_object_title"):
|
|
payload["push_internal_object_title"] = str(spec["push_internal_object_title"])
|
|
if spec.get("ref_type") is not None:
|
|
payload["ref_type"] = int(spec["ref_type"])
|
|
if spec.get("aweType") is not None:
|
|
payload["aweType"] = int(spec["aweType"])
|
|
|
|
uid = str(spec.get("uid") or my_uid or "").strip()
|
|
if uid:
|
|
payload["uid"] = uid
|
|
|
|
if spec.get("content_thumb") is not None:
|
|
payload["content_thumb"] = _merge_media_object(
|
|
payload.get("content_thumb") or {}, spec["content_thumb"]
|
|
)
|
|
if spec.get("cover_url") is not None:
|
|
payload["cover_url"] = _merge_media_object(
|
|
payload.get("cover_url") or {}, spec["cover_url"]
|
|
)
|
|
if spec.get("cover_width") is not None:
|
|
payload["cover_width"] = float(spec["cover_width"])
|
|
if spec.get("cover_height") is not None:
|
|
payload["cover_height"] = float(spec["cover_height"])
|
|
|
|
return payload
|
|
|
|
|
|
def resolve_interaction_message_type(spec: dict[str, Any]) -> int:
|
|
for key in ("message_type", "im_message_type"):
|
|
val = spec.get(key)
|
|
if val is not None and str(val).strip().isdigit():
|
|
return int(val)
|
|
return default_interaction_message_type()
|