671 lines
24 KiB
Python
671 lines
24 KiB
Python
"""自动回复内容解析与 IM 消息体构造（文本 / 网址 / 卡片）。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import os
|
||
import socket
|
||
import ssl
|
||
from typing import Any, Tuple
|
||
from urllib.parse import urlparse
|
||
|
||
logger = logging.getLogger("douyin_im.reply_payload")
|
||
|
||
_REPLY_SPEC_TYPES = ("text", "link", "hyperlink", "hyperlink_multiline", "card", "image", "raw_im", "interaction")
|
||
|
||
|
||
def _safe_int(value: Any, default: int = 0) -> int:
|
||
try:
|
||
if value is None or value == "":
|
||
return default
|
||
return int(value)
|
||
except (TypeError, ValueError):
|
||
return default
|
||
|
||
|
||
def _sanitize_im_content(value: Any) -> Any:
|
||
"""递归剔除 None，避免 JSON 出现 null 导致抖音客户端显示「null」。"""
|
||
if isinstance(value, dict):
|
||
return {k: _sanitize_im_content(v) for k, v in value.items() if v is not None}
|
||
if isinstance(value, list):
|
||
return [_sanitize_im_content(v) for v in value if v is not None]
|
||
return value
|
||
|
||
|
||
def _image_url_list(spec: dict[str, Any], uri: str) -> list[str]:
|
||
from .message_content import uri_to_cdn_urls
|
||
from .image_upload import is_douyin_cdn_url
|
||
|
||
raw_list = spec.get("url_list")
|
||
if isinstance(raw_list, list):
|
||
urls = [
|
||
str(u).strip()
|
||
for u in raw_list
|
||
if isinstance(u, str) and u.strip().startswith("http") and is_douyin_cdn_url(u)
|
||
]
|
||
if urls:
|
||
return urls
|
||
cover_url = str(spec.get("cover_url") or spec.get("douyin_url") or "").strip()
|
||
cdn_urls = uri_to_cdn_urls(uri)
|
||
if cdn_urls:
|
||
return cdn_urls
|
||
if cover_url.startswith("http") and is_douyin_cdn_url(cover_url):
|
||
return [cover_url]
|
||
return []
|
||
|
||
|
||
def _build_image_payload(spec: dict[str, Any]) -> tuple[dict[str, Any] | None, int | None]:
|
||
"""构造 IM 图片消息体（message_type=27），需已上传至抖音 CDN 的 uri。"""
|
||
from .message_content import MSG_TYPE_IMAGE
|
||
|
||
uri = str(spec.get("uri") or spec.get("cover_uri") or "").strip().lstrip("/")
|
||
if not uri:
|
||
return None, None
|
||
|
||
md5 = str(spec.get("md5") or spec.get("cover_md5") or "").strip()
|
||
url_list = _image_url_list(spec, uri)
|
||
|
||
resource_url: dict[str, Any] = {"uri": uri}
|
||
if url_list:
|
||
resource_url["url_list"] = url_list
|
||
if md5:
|
||
resource_url["md5"] = md5
|
||
|
||
msg_content: dict[str, Any] = {
|
||
"aweType": 2702,
|
||
"from_gallery": 1,
|
||
"create_type": 0,
|
||
"resource_url": resource_url,
|
||
"local_path": uri,
|
||
}
|
||
if md5:
|
||
msg_content["md5"] = md5
|
||
return msg_content, MSG_TYPE_IMAGE
|
||
|
||
|
||
def _build_card_cover_fields(spec: dict[str, Any]) -> dict[str, str | list[str]]:
|
||
"""链接卡片封面字段，对齐网页 IM 抓包（cover_url/image_url/image_list 等）。"""
|
||
from .image_upload import is_douyin_cdn_url
|
||
|
||
cover_uri = str(spec.get("cover_uri") or spec.get("uri") or "").strip().lstrip("/")
|
||
if not cover_uri:
|
||
return {}
|
||
|
||
image_list = _image_url_list(spec, cover_uri)
|
||
image_list = [u for u in image_list if is_douyin_cdn_url(u)]
|
||
if not image_list:
|
||
from .message_content import uri_to_cdn_urls
|
||
|
||
image_list = uri_to_cdn_urls(cover_uri)
|
||
if not image_list:
|
||
return {}
|
||
|
||
primary = image_list[0]
|
||
return {
|
||
"cover_url": primary,
|
||
"image_url": primary,
|
||
"image": primary,
|
||
"thumb_url": primary,
|
||
"image_list": image_list,
|
||
}
|
||
|
||
|
||
def _merge_link_card_cover_fields(
|
||
msg_content: dict[str, Any],
|
||
link_info: dict[str, Any],
|
||
cover_fields: dict[str, Any],
|
||
) -> None:
|
||
"""对齐成功抓包：封面详情在 link_info；顶层仅 cover_url + image_url。"""
|
||
if not cover_fields:
|
||
return
|
||
link_info.update(cover_fields)
|
||
msg_content["cover_url"] = cover_fields["cover_url"]
|
||
msg_content["image_url"] = cover_fields["image_url"]
|
||
|
||
|
||
def parse_reply_content(raw: str) -> dict[str, Any]:
|
||
"""解析规则中的 reply_content。纯字符串视为文本，JSON 为结构化回复。"""
|
||
raw = (raw or "").strip()
|
||
if not raw:
|
||
return {"type": "text", "text": ""}
|
||
if raw.startswith("{"):
|
||
try:
|
||
data = json.loads(raw)
|
||
if isinstance(data, dict) and data.get("type") in _REPLY_SPEC_TYPES:
|
||
return data
|
||
except json.JSONDecodeError:
|
||
pass
|
||
return {"type": "text", "text": raw}
|
||
|
||
|
||
def format_reply_display(raw: str) -> str:
|
||
"""将 reply_content 格式化为日志/列表中的可读摘要。"""
|
||
spec = parse_reply_content(raw)
|
||
reply_type = spec.get("type", "text")
|
||
if reply_type == "text":
|
||
return (spec.get("text") or "").strip()
|
||
if reply_type == "link":
|
||
text = (spec.get("text") or spec.get("title") or "").strip()
|
||
url = (spec.get("url") or "").strip()
|
||
if text and url:
|
||
return f"{text} → {url}"
|
||
return text or url
|
||
if reply_type == "hyperlink":
|
||
title = (spec.get("title") or "").strip()
|
||
url = (spec.get("url") or spec.get("target_url") or spec.get("link_url") or "").strip()
|
||
if title and url:
|
||
return f"[超链接] {title}"
|
||
return title or url or "[超链接]"
|
||
if reply_type == "card":
|
||
title = (spec.get("title") or "").strip()
|
||
url = (spec.get("url") or "").strip()
|
||
if title and url:
|
||
return f"[卡片] {title} → {url}"
|
||
return title or url or "[卡片]"
|
||
if reply_type == "interaction":
|
||
title = (
|
||
str(spec.get("subject_title") or spec.get("push_internal_object_title") or "").strip()
|
||
or "点赞比心互动"
|
||
)
|
||
return f"[互动] {title}"
|
||
if reply_type == "raw_im":
|
||
mt = spec.get("message_type") or spec.get("im_message_type") or "?"
|
||
return f"[原始IM] message_type={mt}"
|
||
if reply_type == "image":
|
||
return (spec.get("text") or "").strip() or "[图片]"
|
||
return raw
|
||
|
||
|
||
def serialize_reply_content(spec: dict[str, Any]) -> str:
|
||
return json.dumps(spec, ensure_ascii=False, separators=(",", ":"))
|
||
|
||
|
||
def build_msg_payload(spec: dict[str, Any]) -> Tuple[dict[str, Any], int]:
|
||
"""根据回复规格构造 IM msg_content 与 message_type。"""
|
||
reply_type = spec.get("type", "text")
|
||
|
||
if reply_type == "raw_im":
|
||
payload = spec.get("payload") if spec.get("payload") is not None else spec.get("content")
|
||
if isinstance(payload, str):
|
||
payload = json.loads(payload)
|
||
if not isinstance(payload, dict):
|
||
raise ValueError("raw_im 缺少 payload/content 对象")
|
||
mt = _safe_int(spec.get("message_type") or spec.get("im_message_type"), 7)
|
||
return payload, mt
|
||
|
||
if reply_type == "interaction":
|
||
from .interaction_messages import build_like_heart_payload, resolve_interaction_message_type
|
||
|
||
variant = str(spec.get("variant") or "like_heart").strip()
|
||
if variant != "like_heart":
|
||
raise ValueError(f"不支持的互动类型 variant={variant!r}")
|
||
my_uid = str(spec.get("my_uid") or spec.get("uid") or "").strip()
|
||
msg_content = build_like_heart_payload(spec, my_uid=my_uid)
|
||
return msg_content, resolve_interaction_message_type(spec)
|
||
|
||
if reply_type == "text":
|
||
text = (spec.get("text") or "").strip()
|
||
return (
|
||
{
|
||
"mention_users": [],
|
||
"aweType": 700,
|
||
"richTextInfos": [],
|
||
"text": text,
|
||
},
|
||
7,
|
||
)
|
||
|
||
if reply_type == "link":
|
||
display = (spec.get("text") or spec.get("title") or "").strip()
|
||
url = (spec.get("url") or "").strip()
|
||
if not display:
|
||
display = url
|
||
msg_content: dict[str, Any] = {
|
||
"mention_users": [],
|
||
"aweType": 700,
|
||
"richTextInfos": [],
|
||
"text": display,
|
||
}
|
||
if url and display:
|
||
msg_content["richTextInfos"] = [
|
||
{
|
||
"start": 0,
|
||
"end": len(display),
|
||
"type": 2,
|
||
"link": url,
|
||
"text": display,
|
||
}
|
||
]
|
||
return msg_content, 7
|
||
|
||
if reply_type == "hyperlink":
|
||
from .message_content import MSG_TYPE_LINK_CARD
|
||
|
||
link_url = str(
|
||
spec.get("url") or spec.get("target_url") or spec.get("link_url") or ""
|
||
).strip()
|
||
if is_internal_http_url(link_url):
|
||
raise ValueError("超链接 URL 无效")
|
||
cover_fields = _build_card_cover_fields(spec)
|
||
if not cover_fields:
|
||
raise ValueError("网页链接卡封面未上传到抖音 CDN")
|
||
|
||
title = (spec.get("title") or "").strip() or "网页链接"
|
||
desc = (spec.get("desc") or spec.get("description") or "").strip() or title
|
||
|
||
link_info: dict[str, Any] = {
|
||
"title": title,
|
||
"desc": desc,
|
||
"description": desc,
|
||
"url": link_url,
|
||
"link_url": link_url,
|
||
}
|
||
msg_content = {
|
||
"mention_users": [],
|
||
"aweType": 0,
|
||
"richTextInfos": [],
|
||
"text": link_url or title,
|
||
"link_info": link_info,
|
||
"link_url": link_url,
|
||
"title": title,
|
||
"desc": desc,
|
||
"description": desc,
|
||
}
|
||
_merge_link_card_cover_fields(msg_content, link_info, cover_fields)
|
||
return _sanitize_im_content(msg_content), MSG_TYPE_LINK_CARD
|
||
|
||
if reply_type == "hyperlink_multiline":
|
||
title = (spec.get("title") or "").strip()
|
||
desc = (spec.get("desc") or spec.get("description") or "").strip()
|
||
url = str(
|
||
spec.get("url") or spec.get("target_url") or spec.get("link_url") or ""
|
||
).strip()
|
||
lines = [x for x in (title, desc, url) if x]
|
||
text = "\n".join(lines) if lines else url
|
||
return (
|
||
{
|
||
"mention_users": [],
|
||
"aweType": 700,
|
||
"richTextInfos": [],
|
||
"text": text,
|
||
},
|
||
7,
|
||
)
|
||
|
||
if reply_type == "card":
|
||
from .message_content import MSG_TYPE_LINK_CARD
|
||
|
||
link_url = _resolve_card_link_url(spec)
|
||
cover_fields = _build_card_cover_fields(spec)
|
||
if not cover_fields:
|
||
raise ValueError("链接卡片封面未上传到抖音 CDN")
|
||
|
||
title = (spec.get("title") or "").strip() or "链接卡片"
|
||
desc = (spec.get("desc") or spec.get("description") or "").strip() or title
|
||
|
||
link_info: dict[str, Any] = {
|
||
"title": title,
|
||
"desc": desc,
|
||
"description": desc,
|
||
"url": link_url,
|
||
"link_url": link_url,
|
||
}
|
||
msg_content: dict[str, Any] = {
|
||
"mention_users": [],
|
||
"aweType": 0,
|
||
"richTextInfos": [],
|
||
"text": link_url or title,
|
||
"link_info": link_info,
|
||
"link_url": link_url,
|
||
"title": title,
|
||
"desc": desc,
|
||
"description": desc,
|
||
}
|
||
_merge_link_card_cover_fields(msg_content, link_info, cover_fields)
|
||
return _sanitize_im_content(msg_content), MSG_TYPE_LINK_CARD
|
||
|
||
if reply_type == "image":
|
||
msg_content, message_type = _build_image_payload(spec)
|
||
if msg_content is not None and message_type is not None:
|
||
return msg_content, message_type
|
||
text = (spec.get("text") or "").strip() or "[图片]"
|
||
return (
|
||
{
|
||
"mention_users": [],
|
||
"aweType": 700,
|
||
"richTextInfos": [],
|
||
"text": text,
|
||
},
|
||
7,
|
||
)
|
||
|
||
text = format_reply_display(serialize_reply_content(spec))
|
||
return (
|
||
{
|
||
"mention_users": [],
|
||
"aweType": 700,
|
||
"richTextInfos": [],
|
||
"text": text,
|
||
},
|
||
7,
|
||
)
|
||
|
||
|
||
def parse_reply_messages(raw: str) -> list:
|
||
"""解析多条回复（messages数组）或单条回复 spec。"""
|
||
raw = (raw or "").strip()
|
||
if not raw:
|
||
return [{"type": "text", "text": ""}]
|
||
if raw.startswith("{") or raw.startswith("["):
|
||
try:
|
||
data = json.loads(raw)
|
||
if isinstance(data, dict) and isinstance(data.get("messages"), list) and data["messages"]:
|
||
return [m for m in data["messages"] if isinstance(m, dict) and m.get("type") in _REPLY_SPEC_TYPES]
|
||
if isinstance(data, list):
|
||
return [m for m in data if isinstance(m, dict) and m.get("type") in _REPLY_SPEC_TYPES]
|
||
if isinstance(data, dict) and data.get("type") in _REPLY_SPEC_TYPES:
|
||
return [data]
|
||
except json.JSONDecodeError:
|
||
pass
|
||
return [{"type": "text", "text": raw}]
|
||
|
||
|
||
def _abs_public_url(path_or_url: str) -> str:
|
||
value = (path_or_url or "").strip()
|
||
if not value:
|
||
return ""
|
||
if value.startswith("/"):
|
||
try:
|
||
from .image_upload import _public_base_url
|
||
|
||
base = _public_base_url().rstrip("/")
|
||
if base:
|
||
return f"{base}{value}"
|
||
except Exception:
|
||
pass
|
||
if value and not value.startswith("http"):
|
||
return f"https://{value.lstrip('/')}"
|
||
return value
|
||
|
||
|
||
def is_internal_http_url(url: str) -> bool:
|
||
"""localhost / 127.0.0.1 / 非 http(s) 均视为不可用于抖音卡片外显。"""
|
||
raw = (url or "").strip()
|
||
if not raw:
|
||
return True
|
||
if raw.startswith("/"):
|
||
return True
|
||
lower = raw.lower()
|
||
if not lower.startswith("http://") and not lower.startswith("https://"):
|
||
return True
|
||
return "localhost" in lower or "127.0.0.1" in lower
|
||
|
||
|
||
_DISALLOWED_IM_LINK_HOSTS = (
|
||
"e.douyin.com",
|
||
"developer.open-douyin.com",
|
||
"partner.open-douyin.com",
|
||
"open.douyin.com",
|
||
"imapi.douyin.com",
|
||
)
|
||
|
||
_DISALLOWED_IM_LINK_PATH_PARTS = (
|
||
"im-manage/card",
|
||
"operation-center/im-manage",
|
||
"douyin-mp/operation-center",
|
||
)
|
||
|
||
|
||
def disallowed_im_link_reason(url: str) -> str:
|
||
"""抖音不允许把后台/开放平台地址当作私信网页链接跳转目标。"""
|
||
raw = (url or "").strip()
|
||
if not raw:
|
||
return ""
|
||
try:
|
||
parsed = urlparse(raw)
|
||
except Exception:
|
||
return ""
|
||
host = (parsed.hostname or "").lower()
|
||
path = (parsed.path or "").lower()
|
||
if host in _DISALLOWED_IM_LINK_HOSTS:
|
||
return (
|
||
f"链接指向抖音后台/开放平台（{host}），不能作为私信超链接跳转地址。"
|
||
"请填写您自己的业务页，例如企微获客链接、官网、表单页等。"
|
||
)
|
||
if host.endswith(".douyin.com") and any(p in path for p in _DISALLOWED_IM_LINK_PATH_PARTS):
|
||
return (
|
||
"链接指向抖音运营后台「IM 卡片管理」页面，不是给用户点击的业务 URL。"
|
||
"请改为 https://work.weixin.qq.com/... 或您自己的 HTTPS 业务页。"
|
||
)
|
||
if "example.com" in host:
|
||
return "请填写真实可访问的业务 HTTPS 链接，不要使用 example.com 示例域名。"
|
||
return ""
|
||
|
||
|
||
def public_https_url_ok(url: str, *, timeout: float = 8.0) -> tuple[bool, str]:
|
||
"""检测 HTTPS 链接是否证书有效且可连接（抖音发卡前会校验域名）。"""
|
||
raw = (url or "").strip()
|
||
if not raw:
|
||
return False, "URL 为空"
|
||
parsed = urlparse(raw)
|
||
if parsed.scheme != "https":
|
||
return False, "非 HTTPS 链接"
|
||
host = parsed.hostname
|
||
if not host:
|
||
return False, "无效 URL"
|
||
try:
|
||
ctx = ssl.create_default_context()
|
||
with socket.create_connection((host, 443), timeout=timeout) as sock:
|
||
with ctx.wrap_socket(sock, server_hostname=host) as ssock:
|
||
ssock.getpeercert()
|
||
return True, ""
|
||
except ssl.SSLError as exc:
|
||
return False, f"SSL 证书与域名不匹配或未信任（{exc})"
|
||
except OSError as exc:
|
||
return False, f"无法连接 {host}（{exc})"
|
||
|
||
|
||
def _card_im_link_mode() -> str:
|
||
"""page=落地页；target=业务 URL；auto=落地页证书有效时用落地页，否则用 target。"""
|
||
return (os.getenv("KEFU_CARD_IM_LINK") or "auto").strip().lower()
|
||
|
||
|
||
def _resolve_card_target_url(spec: dict[str, Any]) -> str:
|
||
target_url = str(spec.get("target_url") or "").strip()
|
||
if is_internal_http_url(target_url):
|
||
return ""
|
||
resolved = _abs_public_url(target_url)
|
||
return "" if is_internal_http_url(resolved) else resolved
|
||
|
||
|
||
def _resolve_card_page_url(spec: dict[str, Any]) -> str:
|
||
"""SEO 落地页地址（/p/slug）。"""
|
||
page_url = str(spec.get("url") or spec.get("page_url") or "").strip()
|
||
if is_internal_http_url(page_url):
|
||
return ""
|
||
resolved = _abs_public_url(page_url)
|
||
return "" if is_internal_http_url(resolved) else resolved
|
||
|
||
|
||
def _resolve_card_link_url(spec: dict[str, Any]) -> str:
|
||
"""IM 卡片点击跳转 URL。发送重试可通过 _im_link_url 临时覆盖。"""
|
||
override = str(spec.get("_im_link_url") or "").strip()
|
||
if override and not is_internal_http_url(override):
|
||
return override
|
||
|
||
page_url = _resolve_card_page_url(spec)
|
||
target_url = _resolve_card_target_url(spec)
|
||
mode = _card_im_link_mode()
|
||
|
||
if mode == "target":
|
||
return target_url or page_url
|
||
if mode == "page":
|
||
return page_url or target_url
|
||
|
||
# auto：落地页 HTTPS 证书有效则优先落地页，否则回退业务 URL
|
||
if page_url:
|
||
ok, reason = public_https_url_ok(page_url)
|
||
if ok:
|
||
return page_url
|
||
if target_url:
|
||
logger.warning(
|
||
"Card landing page HTTPS check failed (%s), using target_url for IM link: %s",
|
||
reason,
|
||
target_url[:120],
|
||
)
|
||
return target_url
|
||
logger.warning("Card landing page HTTPS check failed: %s", reason)
|
||
return page_url
|
||
return target_url
|
||
|
||
|
||
def validate_hyperlink_send_spec(spec: dict[str, Any], *, require_cover_uri: bool = False) -> str:
|
||
"""超链接（网页链接卡）：直连业务 URL，无需 SEO 落地页。"""
|
||
if spec.get("type") != "hyperlink":
|
||
return ""
|
||
if not (spec.get("title") or "").strip():
|
||
return "缺少超链接标题"
|
||
cover_ref = str(
|
||
spec.get("cover_url") or spec.get("cover") or spec.get("image_path") or ""
|
||
).strip()
|
||
if not cover_ref:
|
||
return "网页链接卡必须上传封面图，否则客户端只会显示为多行纯文字"
|
||
link_url = str(
|
||
spec.get("url") or spec.get("target_url") or spec.get("link_url") or ""
|
||
).strip()
|
||
if not link_url or is_internal_http_url(link_url):
|
||
return "超链接 URL 无效，请填写公网 HTTPS 地址"
|
||
if not link_url.lower().startswith("https://"):
|
||
return "超链接须为 HTTPS 地址"
|
||
blocked = disallowed_im_link_reason(link_url)
|
||
if blocked:
|
||
return blocked
|
||
ok, reason = public_https_url_ok(link_url)
|
||
if not ok:
|
||
return f"超链接 HTTPS 不可用（{reason}）"
|
||
if require_cover_uri:
|
||
uri = str(spec.get("cover_uri") or spec.get("uri") or "").strip().lstrip("/")
|
||
if not uri:
|
||
return "超链接封面尚未上传到抖音 CDN"
|
||
return ""
|
||
|
||
|
||
def validate_card_send_spec(spec: dict[str, Any], *, require_cover_uri: bool = False) -> str:
|
||
"""校验卡片回复是否满足发送条件，返回错误文案；空串表示通过。"""
|
||
if spec.get("type") != "card":
|
||
return ""
|
||
if not (spec.get("title") or "").strip():
|
||
return "缺少卡片标题"
|
||
cover_ref = str(
|
||
spec.get("cover_url") or spec.get("cover") or spec.get("image_path") or ""
|
||
).strip()
|
||
if not cover_ref:
|
||
return "缺少卡片封面图"
|
||
target = str(spec.get("target_url") or spec.get("link_url") or "").strip()
|
||
blocked = disallowed_im_link_reason(target)
|
||
if blocked:
|
||
return blocked
|
||
link_url = _resolve_card_link_url(spec)
|
||
if not link_url:
|
||
return (
|
||
"卡片跳转链接无效：落地页为 localhost 或未配置公网地址。"
|
||
"请设置环境变量 KEFU_PUBLIC_BASE_URL=https://你的公网域名 后重新保存规则"
|
||
)
|
||
page_url = _resolve_card_page_url(spec)
|
||
if page_url and _card_im_link_mode() != "target":
|
||
ok, reason = public_https_url_ok(page_url)
|
||
if not ok and not _resolve_card_target_url(spec):
|
||
return (
|
||
f"落地页 HTTPS 不可用（{reason}），且无有效 target_url 可回退。"
|
||
"请为 KEFU_PUBLIC_BASE_URL 配置有效 SSL 证书，或设置 KEFU_CARD_IM_LINK=target"
|
||
)
|
||
if require_cover_uri:
|
||
uri = str(spec.get("cover_uri") or spec.get("uri") or "").strip().lstrip("/")
|
||
if not uri:
|
||
return "卡片封面尚未上传到抖音 CDN"
|
||
cover_fields = _build_card_cover_fields(spec)
|
||
if not cover_fields:
|
||
return "卡片封面 CDN 地址无效，请重新上传封面或检查 IM 凭证"
|
||
return ""
|
||
|
||
|
||
def _extract_link_card_media_path(url: str) -> str:
|
||
value = (url or "").strip()
|
||
if not value:
|
||
return ""
|
||
idx = value.find("/api/media/link-cards/")
|
||
if idx >= 0:
|
||
return value[idx:]
|
||
idx = value.find("/api/media/messages/")
|
||
if idx >= 0:
|
||
return value[idx:]
|
||
return ""
|
||
|
||
|
||
def expand_card_spec(spec: dict) -> list:
|
||
"""已废弃：勿将卡片拆成「图片 + 多行文字」，对方看不到网页链接卡。
|
||
|
||
保留函数仅供排查旧数据；发送链路不再调用。
|
||
"""
|
||
title = (spec.get("title") or "").strip()
|
||
desc = (spec.get("desc") or spec.get("description") or "").strip()
|
||
target = (spec.get("target_url") or spec.get("url") or spec.get("link_url") or "").strip()
|
||
if "localhost" in target or "127.0.0.1" in target:
|
||
target = (spec.get("target_url") or spec.get("link_url") or "").strip() or target
|
||
cover = (spec.get("image_path") or "").strip()
|
||
if not cover:
|
||
cover = _extract_link_card_media_path(spec.get("cover_url") or "")
|
||
out = []
|
||
if cover:
|
||
out.append({"type": "image", "url": cover, "text": "[图片]"})
|
||
lines = [x for x in (title, desc, target) if x]
|
||
text = "\n".join(lines) if lines else target
|
||
if text:
|
||
out.append({"type": "text", "text": text})
|
||
if not out:
|
||
out.append({"type": "text", "text": target})
|
||
return out
|
||
|
||
|
||
def split_reply_payloads(raw: str) -> list:
|
||
"""将规则 reply_content 拆成可逐条发送的 payload。卡片以原生链接卡（type 70）单条发送。
|
||
|
||
多条消息时先发文/图再发卡片，降低陌生人会话首条即结构化消息被拒（8004）的概率。
|
||
"""
|
||
specs = parse_reply_messages(raw)
|
||
non_cards = [s for s in specs if s.get("type") not in ("card", "hyperlink")]
|
||
cards = [s for s in specs if s.get("type") in ("card", "hyperlink")]
|
||
ordered = non_cards + cards
|
||
has_lead = bool(non_cards)
|
||
out: list[str] = []
|
||
for spec in ordered:
|
||
item = dict(spec)
|
||
if has_lead and item.get("type") in ("card", "hyperlink"):
|
||
item["_skip_preface"] = True
|
||
out.append(serialize_reply_content(item))
|
||
return out
|
||
|
||
|
||
def serialize_reply_log(payloads: list) -> str:
|
||
"""把实际逐条发送的 payload 合并为结构化日志内容。"""
|
||
specs = []
|
||
for payload in payloads:
|
||
raw_s = (payload or "").strip()
|
||
if not raw_s:
|
||
continue
|
||
spec = None
|
||
if raw_s.startswith("{"):
|
||
try:
|
||
data = json.loads(raw_s)
|
||
if isinstance(data, dict) and data.get("type") in _REPLY_SPEC_TYPES:
|
||
spec = data
|
||
except json.JSONDecodeError:
|
||
pass
|
||
if spec is None:
|
||
spec = {"type": "text", "text": raw_s}
|
||
specs.append(spec)
|
||
if len(specs) == 1:
|
||
return serialize_reply_content(specs[0])
|
||
return json.dumps({"messages": specs}, ensure_ascii=False, separators=(",", ":"))
|