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 格式的结构化回复（文本/网址/卡片）。
        所有出站发送（自动回复/手动 API/脚本）都经过全局限流器，
        防止大量托管账号同时发送占满出站带宽或触发平台级风控。
        """
        from .rate_limit import get_send_limiter

        async with get_send_limiter():
            return await self._send_text_message_unthrottled(
                conversation_id, content, conversation_short_id
            )

    async def _send_text_message_unthrottled(
        self,
        conversation_id: str,
        content: str,
        conversation_short_id: str = "",
    ) -> bool:
        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 "", "", ""