1331 lines
56 KiB
Python
1331 lines
56 KiB
Python
import asyncio
|
||
import json
|
||
import logging
|
||
from typing import Any, Optional
|
||
from urllib.parse import urlparse
|
||
|
||
import httpx
|
||
|
||
from utils import system_logger
|
||
from .conv_util import build_conversation_id, normalize_conversation_id, resolve_peer_uid
|
||
from .protocol import normalize_im_payload, normalize_im_payload_from_bytes, _pick_avatar_url
|
||
from .session import DouyinImSession
|
||
|
||
logger = logging.getLogger("douyin_im.http")
|
||
|
||
IMAPI_BASE = "https://imapi.douyin.com"
|
||
|
||
# 抖音 web IM 发送私信的内嵌 status_code 提示（非官方开放平台错误码，依据实测经验）
|
||
# 注意：7911 是风控/安全校验类错误，抖音返回“系统繁忙，重新登录后可以正常使用私信功能”，
|
||
# 真实含义是“本次发送请求未通过抖音的安全校验”——通常是签名/凭证失效，需要重新登录刷新。
|
||
_STATUS_CODE_HINTS = {
|
||
7911: "私信请求未通过抖音安全校验（风控）。抖音提示“重新登录后可正常使用”，"
|
||
"通常是 IM 签名/凭证失效或与浏览器环境不一致：请停止托管后用『浏览器模式』"
|
||
"重新登录该账号并打开一次私信页，刷新 cookie、web_protect、keys、msToken/a_bogus 后再试；"
|
||
"同时放慢自动回复频率，避免触发风控。",
|
||
7913: "私信被风控拦截（发送过于频繁/内容触发风控），建议降低自动回复频率、放慢节奏。",
|
||
7919: "对方设置了不接收陌生人私信。",
|
||
8003: "发送频率或会话状态受限（可能触发频控、陌生人私信窗口已过期，或对方未互关）。"
|
||
"建议放慢自动回复节奏，让对方先发一条文字消息后再回复。",
|
||
8004: "抖音内嵌 status_code=8004（raw_check_code 可能为 1）。对网页链接卡（type=70），"
|
||
"若响应 message=OK 且含 server_message_id，消息通常已写入会话，"
|
||
"不应与「未互关/陌生人窗口」混为一谈。真正未投递时 message 非 OK 或无 message_id。"
|
||
"其他常见原因：① 封面 tos uri 刚上传尚未可用（勿重复上传，用缓存 uri）；"
|
||
"② link_url 域名 HTTPS 无效；③ 发送过于频繁。",
|
||
8101: "抖音业务规则拒绝投递（raw_check_code=1，签名/凭证已正常）。这通常是"
|
||
"“关系/隐私/频率”限制，而非程序问题：① 你与对方非互关，陌生人主动私信有条数上限"
|
||
"（常为很少几条，发完即被拦）；② 对方隐私设置为“不接收陌生人私信”；"
|
||
"③ 短时间内对同一会话发送过多被临时限制。建议：用一个与该账号【互相关注】、"
|
||
"或【对方先主动发起会话】的真实用户来测试，不要用两个互不关注的小号互发。",
|
||
}
|
||
|
||
# 8xxx 段普遍是“业务/风控/关系”层面的拒绝（签名已通过），统一兜底文案
|
||
_BUSINESS_REJECT_FALLBACK = (
|
||
"抖音业务层拒绝投递（签名与凭证已通过安全校验，请求已到达抖音服务端）。"
|
||
"多为关系链/隐私设置/陌生人私信条数或频率限制所致，并非发送程序的 bug。"
|
||
"请改用与该账号互关、或对方先主动私信过的真实用户进行测试。"
|
||
)
|
||
|
||
# 这些 status_code 表示“签名凭证失效/安全校验未通过”，可通过重新采集 web_protect 后重试
|
||
_CREDENTIAL_EXPIRED_CODES = {7911}
|
||
|
||
|
||
def _fresh_cover_wait_seconds() -> float:
|
||
"""刚上传到抖音 CDN 的封面图片还未完成内容审核，过早发送 type=70 几乎必定拿到
|
||
status_code=8004 + raw_check_code=1（服务端已写入消息，但富媒体展示被剥离，
|
||
对方看到的就是一个空白卡片）。发送前多等一会儿，给审核流水线一个机会。
|
||
默认 6 秒，可用 KEFU_HYPERLINK_FRESH_COVER_WAIT 覆盖。
|
||
"""
|
||
import os as _os
|
||
|
||
try:
|
||
return max(0.0, float(_os.getenv("KEFU_HYPERLINK_FRESH_COVER_WAIT", "6") or "6"))
|
||
except ValueError:
|
||
return 6.0
|
||
|
||
|
||
def _mask_proxy(url: str) -> str:
|
||
"""隐藏代理 URL 中的用户名/密码，仅用于日志展示。"""
|
||
try:
|
||
parsed = urlparse(url)
|
||
if parsed.username or parsed.password:
|
||
host = parsed.hostname or ""
|
||
if parsed.port:
|
||
host += f":{parsed.port}"
|
||
return f"{parsed.scheme}://***@{host}"
|
||
except Exception:
|
||
pass
|
||
return url
|
||
|
||
|
||
def _mask_value(val: str, head: int = 12, tail: int = 0) -> str:
|
||
"""敏感值脱敏：保留前后若干字符 + 总长度。"""
|
||
s = str(val or "")
|
||
if not s:
|
||
return "(empty)"
|
||
if len(s) <= head + tail + 3:
|
||
return f"{s[:head]}…(len={len(s)})"
|
||
if tail:
|
||
return f"{s[:head]}…{s[-tail:]}(len={len(s)})"
|
||
return f"{s[:head]}…(len={len(s)})"
|
||
|
||
|
||
def _cookie_keys(cookie: dict | None) -> str:
|
||
if not cookie:
|
||
return "(none)"
|
||
keys = sorted(cookie.keys())
|
||
important = [k for k in keys if k.lower() in {
|
||
"sessionid", "sessionid_ss", "msToken", "s_v_web_id", "ttwid",
|
||
"uid_tt", "sid_guard", "sid_tt", "odin_tt",
|
||
}]
|
||
extra = len(keys) - len(important)
|
||
suffix = f" +{extra}others" if extra else ""
|
||
return ",".join(important) + suffix if important else ",".join(keys[:8]) + suffix
|
||
|
||
|
||
def _format_im_request_debug(
|
||
*,
|
||
label: str,
|
||
url: str,
|
||
params: dict | None,
|
||
headers: dict[str, str],
|
||
auth,
|
||
session: DouyinImSession,
|
||
payload_len: int = 0,
|
||
proto_hint: dict | None = None,
|
||
proxy: str = "",
|
||
) -> str:
|
||
"""格式化 IM 请求诊断信息（脱敏），便于用户贴日志排查 7911。"""
|
||
lines = [f"[IM请求/{label}] POST {url}"]
|
||
|
||
if params:
|
||
q_parts = []
|
||
for k, v in params.items():
|
||
if k in ("msToken", "a_bogus"):
|
||
q_parts.append(f"{k}={_mask_value(v, 16)}")
|
||
elif k in ("verifyFp", "fp"):
|
||
q_parts.append(f"{k}={_mask_value(v, 10)}")
|
||
else:
|
||
q_parts.append(f"{k}={v}")
|
||
lines.append(f" query: {'&'.join(q_parts)}")
|
||
else:
|
||
lines.append(" query: (unsigned)")
|
||
|
||
hdr = dict(headers or {})
|
||
ua = hdr.pop("User-Agent", hdr.pop("user-agent", ""))
|
||
cookie_hdr = hdr.pop("Cookie", hdr.pop("cookie", ""))
|
||
lines.append(f" User-Agent: {ua or '(missing)'}")
|
||
if cookie_hdr:
|
||
lines.append(f" CookieHeader: len={len(cookie_hdr)} keys≈{_cookie_keys(getattr(auth, 'cookie', None))}")
|
||
else:
|
||
lines.append(f" CookieHeader: (none) cookie_dict={_cookie_keys(getattr(auth, 'cookie', None))}")
|
||
|
||
bd_keys = [
|
||
"bd-ticket-guard-client-data",
|
||
"bd-ticket-guard-client-cert",
|
||
"bd-ticket-guard-ree-public-key",
|
||
"bd-ticket-guard-version",
|
||
"bd-ticket-guard-web-version",
|
||
"bd-ticket-guard-iteration-version",
|
||
]
|
||
for bk in bd_keys:
|
||
if bk in hdr:
|
||
lines.append(f" {bk}: {_mask_value(hdr[bk], 20)}")
|
||
for k, v in sorted(hdr.items()):
|
||
if k.lower().startswith("bd-ticket"):
|
||
continue
|
||
if k.lower() in ("content-type", "accept", "referer", "origin"):
|
||
lines.append(f" {k}: {v}")
|
||
|
||
lines.append(
|
||
" auth: "
|
||
f"ticket={_mask_value(getattr(auth, 'ticket', ''), 10)} "
|
||
f"ts_sign={_mask_value(getattr(auth, 'ts_sign', ''), 12)} "
|
||
f"client_cert={_mask_value(getattr(auth, 'client_cert', ''), 16)} "
|
||
f"private_key={'len'+str(len(auth.private_key)) if getattr(auth, 'private_key', None) else 'MISSING'} "
|
||
f"device_id={getattr(auth, 'device_id', '') or 'MISSING'} "
|
||
f"web_id={getattr(auth, 'web_id', '') or 'MISSING'}"
|
||
)
|
||
lines.append(
|
||
" session: "
|
||
f"my_uid={session.my_uid or 0} "
|
||
f"device_id={session.device_id or 'MISSING'} "
|
||
f"web_id={session.web_id or 'MISSING'} "
|
||
f"frontier_ws={'yes' if session.frontier_ws_url() else 'no'} "
|
||
f"sdk_cert={'len'+str(len(session.sdk_cert)) if session.sdk_cert else 'no'}"
|
||
)
|
||
if proxy:
|
||
lines.append(f" proxy: {_mask_proxy(proxy)}")
|
||
if payload_len:
|
||
lines.append(f" body: protobuf len={payload_len}")
|
||
if proto_hint:
|
||
ph = " ".join(f"{k}={v}" for k, v in proto_hint.items())
|
||
lines.append(f" proto: {ph}")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def format_session_credential_summary(session: DouyinImSession) -> str:
|
||
"""账号 IM 凭证摘要（脱敏），启动托管时打印便于对照本地/服务器差异。"""
|
||
from .auth import DouyinAuth
|
||
|
||
auth = DouyinAuth.from_im_session(session)
|
||
return (
|
||
f"[IM凭证] my_uid={session.my_uid or 0} "
|
||
f"device_id={session.device_id or 'MISSING'} "
|
||
f"web_id={session.web_id or 'MISSING'} "
|
||
f"sessionid={'yes' if session.cookies.get('sessionid') or session.cookies.get('sessionid_ss') else 'MISSING'} "
|
||
f"msToken={'yes' if session.cookies.get('msToken') else 'MISSING'} "
|
||
f"s_v_web_id={'yes' if session.cookies.get('s_v_web_id') else 'MISSING'} "
|
||
f"keys={'len'+str(len(session.keys_str)) if session.keys_str else 'MISSING'} "
|
||
f"web_protect={'len'+str(len(session.web_protect_str)) if session.web_protect_str else 'MISSING'} "
|
||
f"auth_ticket={_mask_value(auth.ticket, 10)} "
|
||
f"auth_ts_sign={_mask_value(auth.ts_sign, 12)} "
|
||
f"auth_cert={_mask_value(auth.client_cert, 16)} "
|
||
f"auth_device_id={auth.device_id or 'MISSING'} "
|
||
f"frontier_ws={'yes' if session.frontier_ws_url() else 'no'} "
|
||
f"ua={session.user_agent[:80] + '…' if len(session.user_agent or '') > 80 else session.user_agent}"
|
||
)
|
||
|
||
|
||
class DouyinImHttpClient:
|
||
"""抖音 IM HTTP API 客户端（基于 Cookie 鉴权）"""
|
||
|
||
def __init__(self, session: DouyinImSession, account_id: Optional[int] = None):
|
||
self.session = session
|
||
self.account_id = account_id
|
||
self._client: Optional[httpx.AsyncClient] = None
|
||
self.last_send_meta: dict[str, dict] = dict(session.conv_meta or {})
|
||
self.last_error: str = ""
|
||
# True 表示本次发送失败是“签名凭证失效(7911)”，上层应刷新 web_protect 后重试
|
||
self.last_send_needs_refresh: bool = False
|
||
self.last_request_debug: str = ""
|
||
# 非空表示“接口判定为已投递，但抖音内容风控可能已剥离富媒体展示”（如网页链接卡显示为空）
|
||
self.last_send_risk_notice: str = ""
|
||
self._proxy_url: str = ""
|
||
|
||
async def __aenter__(self):
|
||
from rpa_engine.runtime_config import httpx_proxy
|
||
|
||
headers = {
|
||
"User-Agent": self.session.user_agent,
|
||
"Cookie": self.session.cookie_header(),
|
||
"Referer": "https://www.douyin.com/",
|
||
"Origin": "https://www.douyin.com",
|
||
"Accept": "application/json, text/plain, */*",
|
||
"Content-Type": "application/json; charset=UTF-8",
|
||
}
|
||
client_kwargs: dict[str, Any] = {
|
||
"headers": headers,
|
||
"timeout": httpx.Timeout(20.0, connect=10.0),
|
||
"follow_redirects": True,
|
||
}
|
||
# 配置 KEFU_DOUYIN_PROXY 时让全部抖音 IM 请求走住宅代理，绕开机房 IP 风控(7911)
|
||
proxy = httpx_proxy()
|
||
if proxy:
|
||
client_kwargs["proxy"] = proxy
|
||
self._proxy_url = proxy
|
||
logger.info(f"IM HTTP client using proxy: {_mask_proxy(proxy)}")
|
||
self._client = httpx.AsyncClient(**client_kwargs)
|
||
return self
|
||
|
||
async def __aexit__(self, *args):
|
||
if self._client:
|
||
await self._client.aclose()
|
||
self._client = None
|
||
|
||
async def _request(self, method: str, path: str, json_body: Optional[dict] = None) -> Any:
|
||
if not self._client:
|
||
raise RuntimeError("HTTP client not started")
|
||
params = self.session.common_params()
|
||
url = f"{IMAPI_BASE}{path}"
|
||
try:
|
||
if method.upper() == "GET":
|
||
resp = await self._client.get(url, params={**params, **(json_body or {})})
|
||
else:
|
||
body = {**params, **(json_body or {})}
|
||
resp = await self._client.post(url, json=body)
|
||
resp.raise_for_status()
|
||
ct = resp.headers.get("content-type", "")
|
||
if "json" in ct:
|
||
return resp.json()
|
||
raw = resp.content
|
||
if raw:
|
||
try:
|
||
return json.loads(raw.decode("utf-8"))
|
||
except Exception:
|
||
pass
|
||
parsed = normalize_im_payload_from_bytes(raw)
|
||
if parsed:
|
||
return {"conversations": parsed}
|
||
return resp.text
|
||
except Exception as e:
|
||
logger.warning(f"IM API {path} failed: {e}")
|
||
return None
|
||
|
||
def _set_error(self, msg: str) -> None:
|
||
self.last_error = msg or ""
|
||
|
||
def _log_send_failure(self, conversation_id: str, detail: str) -> None:
|
||
system_logger.record(
|
||
"私信发送失败",
|
||
detail=f"会话 {conversation_id}：{detail}",
|
||
level="error",
|
||
category="send",
|
||
account_id=self.account_id,
|
||
)
|
||
|
||
def _protobuf_headers(self) -> dict[str, str]:
|
||
return {
|
||
"User-Agent": self.session.user_agent,
|
||
"Cookie": self.session.cookie_header(),
|
||
"Referer": "https://www.douyin.com/",
|
||
"Origin": "https://www.douyin.com",
|
||
"Content-Type": "application/x-protobuf",
|
||
"Accept": "application/x-protobuf",
|
||
}
|
||
|
||
@staticmethod
|
||
def _build_signed_extras(auth, api_path: str, user_agent: str = "") -> tuple[dict, dict]:
|
||
"""生成签名 query 参数(a_bogus) + bd-ticket-guard 请求头。
|
||
|
||
含 Node execjs 调用，属阻塞操作，需放线程执行。
|
||
bd-ticket-guard-client-data 是浏览器对“敏感写接口”（含私信发送）都会带的
|
||
设备票据签名头；DouYin_Spider 的 send_msg 漏带了，这里补齐以通过抖音安全校验。
|
||
|
||
user_agent 必须与本次请求实际发送的 User-Agent 头一致，否则 a_bogus 校验失败 -> 7911。
|
||
"""
|
||
from .dy_util import (
|
||
generate_msToken,
|
||
splice_url,
|
||
generate_a_bogus,
|
||
generate_bd_ticket_client_data,
|
||
generate_ree_key,
|
||
normalize_client_cert,
|
||
DEFAULT_USER_AGENT,
|
||
)
|
||
|
||
ua = user_agent or getattr(auth, "user_agent", "") or DEFAULT_USER_AGENT
|
||
ms_token = ""
|
||
if auth.cookie:
|
||
ms_token = auth.cookie.get("msToken") or ""
|
||
if not ms_token:
|
||
ms_token = getattr(auth, "msToken", "") or generate_msToken()
|
||
|
||
s_v_web_id = auth.cookie.get("s_v_web_id", "") if auth.cookie else ""
|
||
params = {
|
||
"verifyFp": s_v_web_id,
|
||
"fp": s_v_web_id,
|
||
"msToken": ms_token,
|
||
}
|
||
params["a_bogus"] = generate_a_bogus(splice_url(params), user_agent=ua)
|
||
|
||
headers: dict[str, str] = {}
|
||
guard_cert = normalize_client_cert(getattr(auth, "client_cert", "") or "")
|
||
bd_ok = bool(auth.ticket and auth.ts_sign and auth.private_key)
|
||
if bd_ok:
|
||
try:
|
||
headers["bd-ticket-guard-client-data"] = generate_bd_ticket_client_data(
|
||
api_path, auth.ticket, auth.ts_sign, auth.private_key
|
||
)
|
||
headers["bd-ticket-guard-iteration-version"] = "1"
|
||
headers["bd-ticket-guard-ree-public-key"] = generate_ree_key(auth.private_key)
|
||
headers["bd-ticket-guard-version"] = "2"
|
||
headers["bd-ticket-guard-web-version"] = "1"
|
||
if guard_cert:
|
||
headers["bd-ticket-guard-client-cert"] = guard_cert
|
||
except Exception as e:
|
||
logger.warning(f"bd-ticket-guard 头生成失败: {e}")
|
||
bd_ok = False
|
||
|
||
# 签名输入诊断（不打印明文密钥，仅长度/存在性），用于定位 7911 到底缺哪一项
|
||
logger.info(
|
||
"Sign inputs: ua=%r s_v_web_id=%s msToken=%s a_bogus=%s | "
|
||
"ticket=%s ts_sign=%s client_cert=%s private_key=%s device_id=%s bd_guard_headers=%s",
|
||
ua,
|
||
"yes" if s_v_web_id else "MISSING",
|
||
f"len{len(ms_token)}" if ms_token else "MISSING",
|
||
"yes" if params.get("a_bogus") else "MISSING",
|
||
f"len{len(auth.ticket)}" if auth.ticket else "MISSING",
|
||
f"len{len(auth.ts_sign)}" if auth.ts_sign else "MISSING",
|
||
f"len{len(guard_cert)}" if guard_cert else "MISSING",
|
||
f"len{len(auth.private_key)}" if auth.private_key else "MISSING",
|
||
getattr(auth, "device_id", "") or "MISSING",
|
||
"yes" if bd_ok else "MISSING",
|
||
)
|
||
return params, headers
|
||
|
||
async def _post_protobuf(
|
||
self,
|
||
url: str,
|
||
auth,
|
||
payload: bytes,
|
||
signed: bool = False,
|
||
*,
|
||
log_label: str = "",
|
||
proto_hint: dict | None = None,
|
||
):
|
||
"""IM protobuf POST。create/get_info 不带 a_bogus（与 DouYin_Spider 一致）。"""
|
||
params = None
|
||
headers = self._protobuf_headers()
|
||
if signed:
|
||
# execjs 调用 Node 子进程是阻塞的，放到线程池避免卡住事件循环
|
||
api_path = urlparse(url).path
|
||
params, bd_headers = await asyncio.to_thread(
|
||
self._build_signed_extras, auth, api_path, self.session.user_agent
|
||
)
|
||
headers = {**headers, **bd_headers}
|
||
|
||
label = log_label or ("signed" if signed else "unsigned")
|
||
debug_text = _format_im_request_debug(
|
||
label=label,
|
||
url=url,
|
||
params=params,
|
||
headers=headers,
|
||
auth=auth,
|
||
session=self.session,
|
||
payload_len=len(payload or b""),
|
||
proto_hint=proto_hint,
|
||
proxy=self._proxy_url,
|
||
)
|
||
self.last_request_debug = debug_text
|
||
logger.info(debug_text)
|
||
|
||
resp = await self._client.post(
|
||
url,
|
||
params=params,
|
||
headers=headers,
|
||
content=payload,
|
||
cookies=auth.cookie if auth.cookie else None,
|
||
)
|
||
logger.info(
|
||
"[IM响应/%s] status=%s len=%s",
|
||
label,
|
||
resp.status_code,
|
||
len(resp.content or b""),
|
||
)
|
||
return resp
|
||
|
||
def _parse_conversation_body(self, response_proto) -> tuple[str, str, str]:
|
||
body = response_proto.body
|
||
for field in ("create_conversation_v2_body", "get_conversation_info_list_v2_response_body"):
|
||
if body.HasField(field):
|
||
conv_body = getattr(body, field)
|
||
if conv_body.conversation_info_list:
|
||
conv = conv_body.conversation_info_list[0]
|
||
return (
|
||
conv.conversation_id,
|
||
str(conv.conversation_short_id),
|
||
conv.ticket,
|
||
)
|
||
return "", "", ""
|
||
|
||
async def get_conversation_info(
|
||
self,
|
||
auth,
|
||
peer_uid: int,
|
||
my_uid: int,
|
||
conversation_id: str,
|
||
conversation_short_id: int = 0,
|
||
) -> tuple[str, str, str]:
|
||
from .proto_builder import ProtoBuilder
|
||
|
||
request_proto = await asyncio.to_thread(
|
||
ProtoBuilder.build_get_conversation_list_info_request,
|
||
auth, peer_uid, my_uid, conversation_short_id
|
||
)
|
||
if conversation_id:
|
||
request_proto.body.get_conversation_info_list_v2_body.data.conversation_id = conversation_id
|
||
|
||
url = "https://imapi.douyin.com/v2/conversation/get_info_list"
|
||
try:
|
||
resp = await self._post_protobuf(
|
||
url, auth, request_proto.SerializeToString(), signed=False, log_label="get_info"
|
||
)
|
||
resp.raise_for_status()
|
||
from .static import Response_pb2 as ResponseProto
|
||
|
||
response_proto = ResponseProto.Response()
|
||
response_proto.ParseFromString(resp.content)
|
||
if response_proto.error_desc:
|
||
self._set_error(response_proto.error_desc)
|
||
return "", "", ""
|
||
return self._parse_conversation_body(response_proto)
|
||
except Exception as e:
|
||
self._set_error(str(e))
|
||
return "", "", ""
|
||
|
||
async def resolve_conversation_meta(
|
||
self,
|
||
auth,
|
||
conversation_id: str,
|
||
my_uid: int,
|
||
peer_uid: int,
|
||
) -> tuple[str, str, str]:
|
||
"""获取会话 short_id / ticket。
|
||
|
||
注意：ticket 时效很短，缓存的 ticket 用于发送会被接口接受但消息不投递，
|
||
因此每次发送都优先用 get_info 拉取“新鲜”的 ticket，失败再退回创建会话，
|
||
最后才使用缓存值兜底。
|
||
"""
|
||
cached = self.last_send_meta.get(conversation_id, {})
|
||
cached_short = str(cached.get("conversation_short_id") or "")
|
||
cached_ticket = str(cached.get("ticket") or "")
|
||
|
||
conv_id, short_id, ticket = await self.get_conversation_info(
|
||
auth, peer_uid, my_uid, conversation_id, int(cached_short or 0)
|
||
)
|
||
if conv_id and short_id and ticket:
|
||
self._cache_conv_meta(conv_id, short_id, ticket)
|
||
return conv_id, short_id, ticket
|
||
|
||
conv_id, short_id, ticket = await self.create_conversation(peer_uid, my_uid)
|
||
if conv_id and short_id and ticket:
|
||
return conv_id, short_id, ticket
|
||
|
||
if cached_short and cached_ticket:
|
||
logger.info(f"Falling back to cached ticket for {conversation_id}")
|
||
return conversation_id, cached_short, cached_ticket
|
||
|
||
return "", "", ""
|
||
|
||
def _cache_conv_meta(self, conv_id: str, short_id: str, ticket: str) -> None:
|
||
meta = {
|
||
"conversation_short_id": short_id,
|
||
"ticket": ticket,
|
||
}
|
||
self.last_send_meta[conv_id] = meta
|
||
self.session.conv_meta[conv_id] = meta
|
||
|
||
async def verify_messaging_capability(self, auth, my_uid: int) -> tuple[bool, str]:
|
||
"""Verify cached conversation tickets still work (required before IM direct start)."""
|
||
from .conv_util import resolve_peer_uid
|
||
|
||
conv_meta = self.session.conv_meta or {}
|
||
if not conv_meta:
|
||
return False, "缺少 IM 会话 ticket，请用浏览器模式登录并打开私信页"
|
||
|
||
for conv_id, meta in conv_meta.items():
|
||
short_id = str(meta.get("conversation_short_id") or "").strip()
|
||
ticket = str(meta.get("ticket") or "").strip()
|
||
if not short_id or not ticket:
|
||
continue
|
||
peer_uid = resolve_peer_uid(str(conv_id), my_uid)
|
||
if not peer_uid:
|
||
continue
|
||
resolved_id, resolved_short_id, resolved_ticket = await self.get_conversation_info(
|
||
auth,
|
||
peer_uid,
|
||
my_uid,
|
||
str(conv_id),
|
||
int(short_id or 0),
|
||
)
|
||
if resolved_id and resolved_short_id and resolved_ticket:
|
||
self._cache_conv_meta(resolved_id, resolved_short_id, resolved_ticket)
|
||
return True, "凭证有效，可直连 IM 托管（含发送签名）"
|
||
|
||
return (
|
||
False,
|
||
"IM 会话 ticket 已失效，请停止托管后用浏览器模式重新登录并打开私信页",
|
||
)
|
||
|
||
async def get_unread_count(self) -> int:
|
||
data = await self._request("GET", "/v1/client/unread_count")
|
||
if not isinstance(data, dict):
|
||
return 0
|
||
total = 0
|
||
for key, val in data.items():
|
||
if "unread" in str(key).lower():
|
||
try:
|
||
total = max(total, int(val or 0))
|
||
except (TypeError, ValueError):
|
||
pass
|
||
return total
|
||
|
||
async def get_conversations(self) -> list[dict]:
|
||
"""拉取会话列表，返回标准化会话"""
|
||
payloads = [
|
||
{"cursor": 0, "count": 50, "inbox_type": 0},
|
||
{"cursor": 0, "limit": 50},
|
||
{},
|
||
]
|
||
conversations = []
|
||
seen = set()
|
||
|
||
for body in payloads:
|
||
data = await self._request("POST", "/v1/conversation/list", body)
|
||
if not data:
|
||
data = await self._request("GET", "/v1/conversation/list", body)
|
||
if not data:
|
||
continue
|
||
|
||
normalized = normalize_im_payload(data)
|
||
for item in normalized:
|
||
name = item.get("sender_name") or ""
|
||
key = name or item.get("conversation_id") or ""
|
||
if key and key not in seen:
|
||
seen.add(key)
|
||
conversations.append(item)
|
||
|
||
# 也从原始结构提取会话级 unread
|
||
self._extract_conversation_rows(data, conversations, seen)
|
||
|
||
if conversations:
|
||
break
|
||
|
||
logger.info(f"Fetched {len(conversations)} conversations from IM API")
|
||
return conversations
|
||
|
||
def _extract_conversation_rows(self, data: Any, out: list, seen: set, depth: int = 0):
|
||
if depth > 10:
|
||
return
|
||
if isinstance(data, dict):
|
||
name = (
|
||
data.get("nick_name")
|
||
or data.get("nickname")
|
||
or (data.get("core_info") or {}).get("nick_name")
|
||
if isinstance(data.get("core_info"), dict)
|
||
else None
|
||
)
|
||
unread = data.get("unread_count") or data.get("unreadCount") or 0
|
||
conv_id = data.get("conversation_id") or data.get("conversationId") or ""
|
||
preview = ""
|
||
last = data.get("last_message") or data.get("latest_message")
|
||
if isinstance(last, dict):
|
||
preview = last.get("content") or last.get("text") or ""
|
||
elif isinstance(last, str):
|
||
preview = last
|
||
|
||
if isinstance(name, str) and name.strip():
|
||
name = name.strip()
|
||
sender_avatar = _pick_avatar_url(data)
|
||
if name not in seen:
|
||
try:
|
||
unread = int(unread or 0)
|
||
except (TypeError, ValueError):
|
||
unread = 0
|
||
out.append({
|
||
"sender_name": name,
|
||
"sender_avatar": sender_avatar or None,
|
||
"content": str(preview or ""),
|
||
"conversation_id": str(conv_id or ""),
|
||
"unread_count": unread,
|
||
})
|
||
seen.add(name)
|
||
|
||
for v in data.values():
|
||
self._extract_conversation_rows(v, out, seen, depth + 1)
|
||
elif isinstance(data, list):
|
||
for item in data:
|
||
self._extract_conversation_rows(item, out, seen, depth + 1)
|
||
|
||
async def _deliver_single_payload(
|
||
self,
|
||
auth,
|
||
conversation_id: str,
|
||
conv_short_id: str,
|
||
ticket: str,
|
||
reply_spec: dict,
|
||
*,
|
||
log_label: str = "send",
|
||
) -> tuple[dict, dict]:
|
||
"""发送单条 IM 消息并解析响应（无重试）。"""
|
||
from .proto_builder import ProtoBuilder
|
||
from .reply_payload import build_msg_payload
|
||
from .pb_decode import analyze_send_response
|
||
|
||
msg_content, message_type = build_msg_payload(reply_spec)
|
||
from .im_debug_log import log_im_message
|
||
|
||
log_im_message(
|
||
direction="out",
|
||
message_type=message_type,
|
||
conversation_id=conversation_id,
|
||
content=msg_content,
|
||
fields={
|
||
"conversation_id": conversation_id,
|
||
"conversation_short_id": conv_short_id,
|
||
"message_type": message_type,
|
||
"reply_type": reply_spec.get("type"),
|
||
},
|
||
)
|
||
request_proto = await asyncio.to_thread(
|
||
ProtoBuilder.build_send_message_request,
|
||
auth,
|
||
conversation_id,
|
||
conv_short_id,
|
||
ticket,
|
||
msg_content,
|
||
message_type,
|
||
)
|
||
proto_hint = {
|
||
"cmd": request_proto.cmd,
|
||
"device_id": request_proto.device_id,
|
||
"token": _mask_value(request_proto.token, 10),
|
||
"ts_sign": _mask_value(request_proto.ts_sign, 12),
|
||
"sdk_cert": _mask_value(request_proto.sdk_cert, 16),
|
||
"body_ticket": _mask_value(ticket, 10),
|
||
}
|
||
resp = await self._post_protobuf(
|
||
f"{IMAPI_BASE}/v1/message/send",
|
||
auth,
|
||
request_proto.SerializeToString(),
|
||
signed=True,
|
||
log_label=log_label,
|
||
proto_hint=proto_hint,
|
||
)
|
||
resp.raise_for_status()
|
||
return analyze_send_response(resp.content), proto_hint
|
||
|
||
async def _maybe_send_link_card_preface(
|
||
self,
|
||
auth,
|
||
conversation_id: str,
|
||
conv_short_id: str,
|
||
ticket: str,
|
||
*,
|
||
skip: bool,
|
||
) -> None:
|
||
"""成功抓包显示：先发一句引导文字再发 type=70，可显著降低 8004。"""
|
||
import os as _os
|
||
|
||
if skip:
|
||
return
|
||
if _os.getenv("KEFU_HYPERLINK_PREFACE", "1").strip().lower() in ("0", "false", "no"):
|
||
return
|
||
preface_text = (
|
||
_os.getenv("KEFU_HYPERLINK_PREFACE_TEXT", "请查看下方网页链接").strip()
|
||
or "请查看下方网页链接"
|
||
)
|
||
preface_spec = {"type": "text", "text": preface_text}
|
||
try:
|
||
result, _ = await self._deliver_single_payload(
|
||
auth,
|
||
conversation_id,
|
||
conv_short_id,
|
||
ticket,
|
||
preface_spec,
|
||
log_label="send-preface",
|
||
)
|
||
if result.get("ok"):
|
||
delay = float(_os.getenv("KEFU_HYPERLINK_PREFACE_DELAY", "1.2") or "1.2")
|
||
logger.info("Link card preface delivered, waiting %.1fs before type=70", delay)
|
||
await asyncio.sleep(max(0.5, delay))
|
||
else:
|
||
logger.warning(
|
||
"Link card preface failed (status=%s), continuing with type=70",
|
||
result.get("status_code"),
|
||
)
|
||
except Exception as exc:
|
||
logger.warning("Link card preface error: %s", exc)
|
||
|
||
async def _send_url_preview_fallback_bundle(
|
||
self,
|
||
auth,
|
||
conversation_id: str,
|
||
conv_short_id: str,
|
||
ticket: str,
|
||
reply_spec: dict,
|
||
) -> bool:
|
||
"""type=70 网页链接卡被判定 raw_check_code=1（内容风控剥离富媒体）时的兜底：
|
||
|
||
实测已证实：一旦 raw_check_code=1，那条 type=70 消息在对方客户端就是一个
|
||
彻底空白的气泡（无封面/无标题/无文字），且这是服务端已写入的既成事实，
|
||
重发同一张卡片大概率仍会被剥离（还会刷屏出多条空白卡片）。
|
||
|
||
因此这里不重试卡片本身，而是补发两条已验证能正常展示的普通消息：
|
||
1) 封面图（message_type=27，图片消息在抖音客户端始终能正常显示）；
|
||
2) 标题+描述+链接文本（message_type=7），保证对方至少能看到内容和可复制的链接。
|
||
|
||
这样即使卡片被风控剥离，对方也绝不会只收到一个空气泡。
|
||
"""
|
||
import os as _os
|
||
|
||
# 默认关闭：只发原生卡片（message_type=70），不再自动降级成图片+三行文字。
|
||
# 需要兜底时显式设置 KEFU_HYPERLINK_RISK_FALLBACK=1 开启。
|
||
if _os.getenv("KEFU_HYPERLINK_RISK_FALLBACK", "0").strip().lower() not in ("1", "true", "yes"):
|
||
return False
|
||
|
||
sent_any = False
|
||
try:
|
||
image_spec = {**reply_spec, "type": "image"}
|
||
img_result, _ = await self._deliver_single_payload(
|
||
auth,
|
||
conversation_id,
|
||
conv_short_id,
|
||
ticket,
|
||
image_spec,
|
||
log_label="send-risk-fallback-image",
|
||
)
|
||
if img_result.get("ok"):
|
||
sent_any = True
|
||
await asyncio.sleep(0.8)
|
||
else:
|
||
logger.warning(
|
||
"URL preview risk-fallback image not ok: status=%s",
|
||
img_result.get("status_code"),
|
||
)
|
||
except Exception as exc:
|
||
logger.warning("URL preview risk-fallback image failed: %s", exc)
|
||
|
||
try:
|
||
text_spec = {**reply_spec, "type": "hyperlink_multiline"}
|
||
text_result, _ = await self._deliver_single_payload(
|
||
auth,
|
||
conversation_id,
|
||
conv_short_id,
|
||
ticket,
|
||
text_spec,
|
||
log_label="send-risk-fallback-text",
|
||
)
|
||
if text_result.get("ok"):
|
||
sent_any = True
|
||
else:
|
||
logger.warning(
|
||
"URL preview risk-fallback text not ok: status=%s",
|
||
text_result.get("status_code"),
|
||
)
|
||
except Exception as exc:
|
||
logger.warning("URL preview risk-fallback text failed: %s", exc)
|
||
|
||
return sent_any
|
||
|
||
async def send_text_message(
|
||
self,
|
||
conversation_id: str,
|
||
content: str,
|
||
conversation_short_id: str = "",
|
||
) -> bool:
|
||
"""通过 IM API 发送 Protobuf 编码的私信（带接口签名）
|
||
|
||
content 可为纯文本，或 JSON 格式的结构化回复（文本/网址/卡片）。
|
||
"""
|
||
from .auth import DouyinAuth
|
||
from .proto_builder import ProtoBuilder
|
||
from .reply_payload import (
|
||
build_msg_payload,
|
||
parse_reply_content,
|
||
_resolve_card_target_url,
|
||
public_https_url_ok,
|
||
)
|
||
|
||
self._set_error("")
|
||
self.last_send_needs_refresh = False
|
||
self.last_send_risk_notice = ""
|
||
auth = DouyinAuth.from_im_session(self.session)
|
||
my_uid = self.session.my_uid or auth.get_uid()
|
||
if not my_uid:
|
||
self._set_error("无法获取当前账号 UID")
|
||
self._log_send_failure(conversation_id, "无法获取当前账号 UID（Cookie 可能已失效）")
|
||
return False
|
||
if not auth.is_sign_ready():
|
||
self._set_error("缺少 IM 签名密钥，请用浏览器登录补全 localStorage")
|
||
self._log_send_failure(
|
||
conversation_id,
|
||
"缺少 IM 签名密钥（web_protect/keys），请用浏览器模式重新登录并打开私信页采集。",
|
||
)
|
||
return False
|
||
|
||
conversation_id = normalize_conversation_id(conversation_id, my_uid)
|
||
peer_uid = resolve_peer_uid(conversation_id, my_uid)
|
||
if not peer_uid:
|
||
self._set_error("无法解析对方用户 ID")
|
||
self._log_send_failure(conversation_id, "无法从会话 ID 解析对方用户 ID")
|
||
return False
|
||
|
||
cached = self.last_send_meta.get(conversation_id, {})
|
||
conv_short_id = str(conversation_short_id or "").strip()
|
||
|
||
# ticket 时效很短：每次发送都重新解析以拿到新鲜 ticket（避免“接口成功但消息不投递”）
|
||
logger.info(f"Resolving fresh ticket for {conversation_id} (peer={peer_uid})...")
|
||
resolved_id, resolved_short_id, resolved_ticket = await self.resolve_conversation_meta(
|
||
auth, conversation_id, my_uid, peer_uid
|
||
)
|
||
if resolved_id:
|
||
conversation_id = resolved_id
|
||
conv_short_id = resolved_short_id or conv_short_id or str(cached.get("conversation_short_id") or "")
|
||
ticket = resolved_ticket or str(cached.get("ticket") or "")
|
||
|
||
if not conv_short_id or not ticket:
|
||
detail = self.last_error or "无法获取会话 ticket/short_id"
|
||
self._set_error(detail)
|
||
logger.warning(f"Send aborted for {conversation_id}: {detail}")
|
||
self._log_send_failure(conversation_id, f"无法获取会话票据(ticket/short_id)：{detail}")
|
||
return False
|
||
|
||
# 发送目标诊断：确认我们到底发给了哪个账号 / 哪个会话（排查“对方收不到”的关键）
|
||
target = (
|
||
f"my_uid={my_uid} peer_uid={peer_uid} "
|
||
f"conv={conversation_id} short_id={conv_short_id} ticket={ticket[:10]}…"
|
||
)
|
||
logger.info(f"Sending to target: {target}")
|
||
|
||
import os as _os
|
||
from .image_upload import prepare_card_reply_spec, prepare_hyperlink_reply_spec, prepare_image_reply_spec
|
||
|
||
reply_spec = parse_reply_content(content)
|
||
if reply_spec.get("type") == "interaction" and my_uid:
|
||
reply_spec = {**reply_spec, "my_uid": my_uid}
|
||
|
||
upload_dir = _os.path.join(
|
||
_os.path.dirname(_os.path.dirname(_os.path.dirname(__file__))),
|
||
"uploads",
|
||
"messages",
|
||
)
|
||
|
||
# ---- 超链接 / 链接卡片：message_type=70（超链接直连 URL，卡片走落地页）----
|
||
is_hyperlink = reply_spec.get("type") == "hyperlink"
|
||
is_link_card = reply_spec.get("type") == "card"
|
||
is_url_preview = is_hyperlink or is_link_card
|
||
is_interaction = reply_spec.get("type") in ("interaction", "raw_im")
|
||
card_link_url = ""
|
||
card_page_url = ""
|
||
if is_hyperlink:
|
||
reply_spec, upload_err = await asyncio.to_thread(
|
||
prepare_hyperlink_reply_spec, reply_spec, self.session, upload_dir
|
||
)
|
||
if upload_err:
|
||
detail = f"超链接消息无法发送：{upload_err}"
|
||
self._set_error(detail)
|
||
self._log_send_failure(conversation_id, detail)
|
||
return False
|
||
if reply_spec.pop("_cover_fresh_upload", False):
|
||
await asyncio.sleep(_fresh_cover_wait_seconds())
|
||
_preview, _mt = build_msg_payload(reply_spec)
|
||
if not (
|
||
_preview.get("cover_url")
|
||
or (_preview.get("link_info") or {}).get("cover_url")
|
||
):
|
||
detail = "超链接 payload 缺少 cover_url，无法发送网页链接卡（否则会显示 null）"
|
||
self._set_error(detail)
|
||
self._log_send_failure(conversation_id, detail)
|
||
return False
|
||
logger.info(
|
||
"Hyperlink ready: message_type=%s link_url=%s cover_uri=%s",
|
||
_mt,
|
||
(_preview.get("link_url") or "")[:120],
|
||
str(reply_spec.get("cover_uri") or "")[:64],
|
||
)
|
||
card_link_url = (_preview.get("link_url") or "").strip()
|
||
elif is_link_card:
|
||
reply_spec, upload_err = await asyncio.to_thread(
|
||
prepare_card_reply_spec, reply_spec, self.session, upload_dir
|
||
)
|
||
if upload_err:
|
||
detail = f"卡片消息无法发送：{upload_err}"
|
||
self._set_error(detail)
|
||
self._log_send_failure(conversation_id, detail)
|
||
return False
|
||
if reply_spec.pop("_cover_fresh_upload", False):
|
||
await asyncio.sleep(_fresh_cover_wait_seconds())
|
||
_preview, _mt = build_msg_payload(reply_spec)
|
||
logger.info(
|
||
"Link card ready: message_type=%s link_url=%s cover_uri=%s",
|
||
_mt,
|
||
(_preview.get("link_url") or "")[:120],
|
||
str(reply_spec.get("cover_uri") or "")[:64],
|
||
)
|
||
card_link_url = (_preview.get("link_url") or "").strip()
|
||
card_page_url = str(reply_spec.get("url") or reply_spec.get("page_url") or "").strip()
|
||
card_target_url = str(reply_spec.get("target_url") or "").strip()
|
||
if card_page_url.startswith("/"):
|
||
from .image_upload import _public_base_url
|
||
|
||
base = _public_base_url().rstrip("/")
|
||
if base:
|
||
card_page_url = f"{base}{card_page_url}"
|
||
|
||
# 落地页证书无效时，首包即强制用业务 URL，避免抖音校验 link_url 时 8004
|
||
if card_page_url:
|
||
page_ok, page_ssl_reason = public_https_url_ok(card_page_url)
|
||
if not page_ok:
|
||
target = _resolve_card_target_url(reply_spec)
|
||
if target:
|
||
reply_spec["_im_link_url"] = target
|
||
_preview, _mt = build_msg_payload(reply_spec)
|
||
card_link_url = (_preview.get("link_url") or "").strip()
|
||
logger.warning(
|
||
"Card page SSL invalid (%s), IM link_url forced to target: %s",
|
||
page_ssl_reason,
|
||
target[:120],
|
||
)
|
||
|
||
if is_interaction:
|
||
try:
|
||
_preview, _mt = build_msg_payload(reply_spec)
|
||
except ValueError as exc:
|
||
detail = f"互动消息无法发送：{exc}"
|
||
self._set_error(detail)
|
||
self._log_send_failure(conversation_id, detail)
|
||
return False
|
||
logger.info(
|
||
"Interaction message ready: message_type=%s aweType=%s itemId=%s title=%s",
|
||
_mt,
|
||
_preview.get("aweType"),
|
||
_preview.get("itemId"),
|
||
(_preview.get("subject_title") or "")[:40],
|
||
)
|
||
|
||
# ---- 图片类型：上传到抖音 CDN ----
|
||
elif reply_spec.get("type") == "image":
|
||
reply_spec, upload_err = await asyncio.to_thread(
|
||
prepare_image_reply_spec, reply_spec, self.session, upload_dir
|
||
)
|
||
if upload_err:
|
||
logger.warning(
|
||
f"Image upload failed for {conversation_id}: {upload_err}; "
|
||
"falling back to text-only"
|
||
)
|
||
if reply_spec.get("text") == "[图片]":
|
||
return True
|
||
detail = f"图片上传失败：{upload_err}"
|
||
self._set_error(detail)
|
||
self._log_send_failure(conversation_id, detail)
|
||
return False
|
||
|
||
url = "https://imapi.douyin.com/v1/message/send"
|
||
skip_preface = bool(reply_spec.pop("_skip_preface", False))
|
||
if is_url_preview:
|
||
await self._maybe_send_link_card_preface(
|
||
auth,
|
||
conversation_id,
|
||
conv_short_id,
|
||
ticket,
|
||
skip=skip_preface,
|
||
)
|
||
max_attempts = 4 if is_url_preview else 1
|
||
card_retry_delays = (0.0, 2.5, 4.0, 6.0)
|
||
try:
|
||
from .pb_decode import analyze_send_response
|
||
|
||
result: dict = {}
|
||
resp = None
|
||
for attempt in range(max_attempts):
|
||
if attempt > 0:
|
||
delay = card_retry_delays[attempt] if attempt < len(card_retry_delays) else 6.0
|
||
logger.info(
|
||
"Link card send got status_code=%s, retry %s/%s after %.1fs…",
|
||
result.get("status_code"),
|
||
attempt + 1,
|
||
max_attempts,
|
||
delay,
|
||
)
|
||
await asyncio.sleep(delay)
|
||
if is_url_preview and attempt >= 1:
|
||
target = _resolve_card_target_url(reply_spec) if is_link_card else str(
|
||
reply_spec.get("url")
|
||
or reply_spec.get("target_url")
|
||
or reply_spec.get("link_url")
|
||
or ""
|
||
).strip()
|
||
if target and target != card_link_url:
|
||
reply_spec["_im_link_url"] = target
|
||
card_link_url = target
|
||
logger.info(
|
||
"Link card retry using target_url as IM link: %s",
|
||
target[:120],
|
||
)
|
||
|
||
msg_content, message_type = build_msg_payload(reply_spec)
|
||
from .im_debug_log import log_im_message
|
||
|
||
log_im_message(
|
||
direction="out",
|
||
message_type=message_type,
|
||
conversation_id=conversation_id,
|
||
content=msg_content,
|
||
fields={
|
||
"conversation_id": conversation_id,
|
||
"conversation_short_id": conv_short_id,
|
||
"message_type": message_type,
|
||
"reply_type": reply_spec.get("type"),
|
||
"attempt": attempt + 1,
|
||
},
|
||
)
|
||
request_proto = await asyncio.to_thread(
|
||
ProtoBuilder.build_send_message_request,
|
||
auth,
|
||
conversation_id,
|
||
conv_short_id,
|
||
ticket,
|
||
msg_content,
|
||
message_type,
|
||
)
|
||
proto_hint = {
|
||
"cmd": request_proto.cmd,
|
||
"device_id": request_proto.device_id,
|
||
"token": _mask_value(request_proto.token, 10),
|
||
"ts_sign": _mask_value(request_proto.ts_sign, 12),
|
||
"sdk_cert": _mask_value(request_proto.sdk_cert, 16),
|
||
"body_ticket": _mask_value(ticket, 10),
|
||
}
|
||
resp = await self._post_protobuf(
|
||
url,
|
||
auth,
|
||
request_proto.SerializeToString(),
|
||
signed=True,
|
||
log_label="send",
|
||
proto_hint=proto_hint,
|
||
)
|
||
resp.raise_for_status()
|
||
result = analyze_send_response(resp.content)
|
||
if result.get("ok"):
|
||
break
|
||
status_code = result.get("status_code")
|
||
if (
|
||
attempt + 1 < max_attempts
|
||
and is_url_preview
|
||
and status_code in (8003, 8004)
|
||
and result.get("raw_check_code") == 1
|
||
):
|
||
continue
|
||
break
|
||
|
||
# 默认不降级为多行文本（否则对方只看到三行纯文字，不是网页链接卡）
|
||
import os as _os
|
||
|
||
allow_text_fallback = _os.getenv("KEFU_HYPERLINK_TEXT_FALLBACK", "").strip().lower() in (
|
||
"1",
|
||
"true",
|
||
"yes",
|
||
)
|
||
if (
|
||
allow_text_fallback
|
||
and not result.get("ok")
|
||
and is_hyperlink
|
||
and result.get("status_code") in (8003, 8004)
|
||
):
|
||
logger.warning(
|
||
"Hyperlink type=70 rejected (status=%s), fallback to multiline text type=7",
|
||
result.get("status_code"),
|
||
)
|
||
fallback_spec = {**reply_spec, "type": "hyperlink_multiline"}
|
||
request_proto = await asyncio.to_thread(
|
||
ProtoBuilder.build_send_message_request,
|
||
auth,
|
||
conversation_id,
|
||
conv_short_id,
|
||
ticket,
|
||
*build_msg_payload(fallback_spec),
|
||
)
|
||
resp = await self._post_protobuf(
|
||
url,
|
||
auth,
|
||
request_proto.SerializeToString(),
|
||
signed=True,
|
||
log_label="send-fallback",
|
||
proto_hint=proto_hint,
|
||
)
|
||
resp.raise_for_status()
|
||
result = analyze_send_response(resp.content)
|
||
if result.get("ok"):
|
||
self._cache_conv_meta(conversation_id, conv_short_id, ticket)
|
||
logger.info(
|
||
"Hyperlink delivered as multiline text (type=7 fallback) for %s",
|
||
conversation_id,
|
||
)
|
||
system_logger.record(
|
||
"私信已投递（超链接降级为文本）",
|
||
detail=(
|
||
f"{target}；type=70 被拒后已用多行文本送达链接；"
|
||
f"server_message_id={result.get('server_message_id')}"
|
||
),
|
||
level="success",
|
||
category="send",
|
||
account_id=self.account_id,
|
||
)
|
||
return True
|
||
|
||
err_text = result.get("error_desc") or ""
|
||
server_mid = result.get("server_message_id")
|
||
|
||
if result.get("ok"):
|
||
self._cache_conv_meta(conversation_id, conv_short_id, ticket)
|
||
notice = ""
|
||
risk_flagged = False
|
||
if result.get("delivered_with_notice"):
|
||
sc = result.get("status_code")
|
||
rcc = result.get("raw_check_code")
|
||
sr = result.get("status_reason") or ""
|
||
if rcc == 1:
|
||
risk_flagged = True
|
||
notice = (
|
||
f"（已投递但 raw_check_code=1，抖音内容风控未通过 status_code={sc}"
|
||
+ (f"：{sr}" if sr else "")
|
||
+ "，消息已写入会话但富媒体展示（封面/标题）很可能被剥离，"
|
||
"对方看到的可能是空白卡片或纯文字，并非发送失败）"
|
||
)
|
||
else:
|
||
notice = (
|
||
f"（已投递，抖音附带业务提示 status_code={sc}"
|
||
+ (f"：{sr}" if sr else "")
|
||
+ f"，raw_check_code={rcc} 表示已通过风控，对方可正常收到）"
|
||
)
|
||
fallback_sent = False
|
||
if risk_flagged and is_url_preview:
|
||
fallback_sent = await self._send_url_preview_fallback_bundle(
|
||
auth, conversation_id, conv_short_id, ticket, reply_spec
|
||
)
|
||
if fallback_sent:
|
||
notice += (
|
||
"；已按 KEFU_HYPERLINK_RISK_FALLBACK=1 补发【封面图+标题/描述/链接文字】兜底，"
|
||
"对方仍能看到内容，并非彻底空白"
|
||
)
|
||
self.last_send_risk_notice = (
|
||
notice if (risk_flagged and is_url_preview) else ""
|
||
)
|
||
logger.info(
|
||
f"Direct message delivered: {target} server_message_id={server_mid}{notice}"
|
||
)
|
||
if risk_flagged:
|
||
title = (
|
||
"私信已投递（内容风控未通过，已自动补发兜底内容）"
|
||
if fallback_sent
|
||
else "私信已投递（内容风控未通过，卡片可能显示为空白，未补发兜底）"
|
||
)
|
||
else:
|
||
title = "私信已投递"
|
||
system_logger.record(
|
||
title,
|
||
detail=f"{target}；server_message_id={server_mid}{notice}；resp[{result.get('summary')}]",
|
||
level="success" if not risk_flagged else "warning",
|
||
category="send",
|
||
account_id=self.account_id,
|
||
)
|
||
return True
|
||
|
||
status_code = result.get("status_code")
|
||
status_reason = result.get("status_reason") or ""
|
||
|
||
if status_code is not None and status_code != 0:
|
||
# body 内嵌 JSON 给出了明确的 status_code，这是权威失败原因
|
||
hint = _STATUS_CODE_HINTS.get(status_code, "")
|
||
# 8xxx 段未单独建模的，统一归为“业务层拒绝（签名已通过）”
|
||
if not hint and 8000 <= int(status_code) < 9000:
|
||
hint = _BUSINESS_REJECT_FALLBACK
|
||
# 7911 属于“签名凭证失效/安全校验未过”，标记为可刷新后重试
|
||
self.last_send_needs_refresh = status_code in _CREDENTIAL_EXPIRED_CODES
|
||
detail = f"抖音拒绝投递 status_code={status_code}"
|
||
if status_reason:
|
||
detail += f"；抖音提示：{status_reason}"
|
||
if hint:
|
||
detail += f"；可能原因：{hint}"
|
||
if is_hyperlink and result.get("status_code") in (8003, 8004) and not result.get("ok"):
|
||
detail += (
|
||
"；若 message=OK 且含 server_message_id 应判为已投递；"
|
||
"网页链接卡须带封面，勿填抖音后台地址"
|
||
)
|
||
if is_link_card and card_page_url:
|
||
ok, ssl_reason = public_https_url_ok(card_page_url)
|
||
if not ok:
|
||
detail += f"；落地页 HTTPS 检测失败：{ssl_reason}"
|
||
if is_url_preview and card_link_url:
|
||
kind = "超链接" if is_hyperlink else "网页链接卡"
|
||
detail += (
|
||
f"；本次 IM {kind}（message_type=70）link_url={card_link_url[:160]}"
|
||
)
|
||
elif not err_text and result.get("message", "").strip().upper() == "OK" and server_mid is None:
|
||
detail = (
|
||
"抖音接口返回 OK 但响应里没有服务端 message_id，"
|
||
"消息很可能未真正写入对方会话（常见原因：账号被风控限流、"
|
||
"对方关闭了陌生人私信、或非互关导致私信被拦截）"
|
||
)
|
||
else:
|
||
reason_bits = []
|
||
if err_text:
|
||
reason_bits.append(f"error_desc={err_text}")
|
||
if result.get("message"):
|
||
reason_bits.append(f"message={result.get('message')}")
|
||
if result.get("cmd") is not None:
|
||
reason_bits.append(f"cmd={result.get('cmd')}")
|
||
detail = "；".join(reason_bits) or "接口返回但未确认投递（无 server_message_id）"
|
||
|
||
full_detail = f"{detail}；{target}；resp[{result.get('summary')}]"
|
||
if self.last_request_debug:
|
||
full_detail += f"\n--- 请求详情 ---\n{self.last_request_debug}"
|
||
self._set_error(full_detail)
|
||
logger.warning(f"Send not confirmed: {full_detail}")
|
||
self._log_send_failure(conversation_id, full_detail)
|
||
return False
|
||
except Exception as e:
|
||
err_detail = f"发送请求异常：{e}；{target}"
|
||
if self.last_request_debug:
|
||
err_detail += f"\n--- 请求详情 ---\n{self.last_request_debug}"
|
||
self._set_error(err_detail)
|
||
logger.error(f"send_text_message exception: {e}\n{self.last_request_debug}")
|
||
self._log_send_failure(conversation_id, err_detail)
|
||
return False
|
||
|
||
async def create_conversation(self, to_user_id: int, my_uid: int = 0) -> tuple[str, str, str]:
|
||
"""创建/获取私信会话"""
|
||
from .auth import DouyinAuth
|
||
from .proto_builder import ProtoBuilder
|
||
|
||
auth = DouyinAuth.from_im_session(self.session)
|
||
if not my_uid:
|
||
my_uid = self.session.my_uid or auth.get_uid()
|
||
if not my_uid:
|
||
self._set_error("无法获取 my_uid")
|
||
return "", "", ""
|
||
if not auth.is_sign_ready():
|
||
self._set_error("缺少 IM 签名密钥（web_protect/keys）")
|
||
logger.warning("create_conversation: IM sign not ready")
|
||
return "", "", ""
|
||
|
||
request_proto = await asyncio.to_thread(
|
||
ProtoBuilder.build_create_conversation_request, auth, to_user_id, my_uid
|
||
)
|
||
url = "https://imapi.douyin.com/v2/conversation/create"
|
||
try:
|
||
resp = await self._post_protobuf(
|
||
url, auth, request_proto.SerializeToString(), signed=False, log_label="create_conv"
|
||
)
|
||
resp.raise_for_status()
|
||
|
||
from .static import Response_pb2 as ResponseProto
|
||
|
||
response_proto = ResponseProto.Response()
|
||
response_proto.ParseFromString(resp.content)
|
||
if response_proto.error_desc:
|
||
self._set_error(response_proto.error_desc)
|
||
logger.warning(f"create_conversation API error: {response_proto.error_desc}")
|
||
|
||
conv_id, conv_short_id, ticket = self._parse_conversation_body(response_proto)
|
||
if not conv_id:
|
||
msg = response_proto.message or "INVALID_REQUEST"
|
||
self._set_error(msg)
|
||
logger.error(
|
||
"create_conversation missing body: "
|
||
f"cmd={response_proto.cmd}, message={msg}"
|
||
)
|
||
return "", "", ""
|
||
|
||
self._cache_conv_meta(conv_id, conv_short_id, ticket)
|
||
logger.info(
|
||
f"create_conversation ok: id={conv_id}, short_id={conv_short_id}"
|
||
)
|
||
return conv_id, conv_short_id, ticket
|
||
except Exception as e:
|
||
self._set_error(str(e))
|
||
logger.error(f"create_conversation failed for {to_user_id}: {e}")
|
||
return "", "", ""
|