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 .peer_profile import enrich_conversation_item, fetch_peer_profile, is_generic_peer_name
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"

# 抖音 IM「按会话拉取消息」cmd（与电商/web 一致）；body 字段号 == cmd。
CMD_GET_MESSAGES_BY_CONVERSATION = 301


def _pb_varint(n: int) -> bytes:
    out = b""
    while True:
        b = n & 0x7F
        n >>= 7
        if n:
            out += bytes([b | 0x80])
        else:
            return out + bytes([b])


def _pb_tag(field: int, wire: int) -> bytes:
    return _pb_varint((field << 3) | wire)


def _pb_str(field: int, val: str) -> bytes:
    b = (val or "").encode("utf-8")
    return _pb_tag(field, 2) + _pb_varint(len(b)) + b


def _pb_int(field: int, n: int) -> bytes:
    return _pb_tag(field, 0) + _pb_varint(int(n))


def _pb_msg(field: int, mb: bytes) -> bytes:
    return _pb_tag(field, 2) + _pb_varint(len(mb)) + mb


def _pb_read_varint(buf: bytes, i: int) -> tuple[int, int]:
    shift = 0
    result = 0
    while True:
        b = buf[i]
        i += 1
        result |= (b & 0x7F) << shift
        if not (b & 0x80):
            return result, i
        shift += 7


def _pb_parse_fields(buf: bytes) -> list[tuple[int, int, Any]]:
    i = 0
    n = len(buf)
    out: list[tuple[int, int, Any]] = []
    while i < n:
        try:
            key, i = _pb_read_varint(buf, i)
        except IndexError:
            break
        fn = key >> 3
        wt = key & 7
        if wt == 0:
            val, i = _pb_read_varint(buf, i)
            out.append((fn, wt, val))
        elif wt == 2:
            ln, i = _pb_read_varint(buf, i)
            out.append((fn, wt, buf[i:i + ln]))
            i += ln
        elif wt == 5:
            out.append((fn, wt, buf[i:i + 4]))
            i += 4
        elif wt == 1:
            out.append((fn, wt, buf[i:i + 8]))
            i += 8
        else:
            break
    return out


def _pb_parse_conversation_messages(content: bytes, cmd: int) -> list[dict]:
    """解析 get_by_conversation 的 protobuf 响应，返回消息列表。"""
    out: list[dict] = []
    for fn, wt, val in _pb_parse_fields(content):
        if fn != 6 or wt != 2:  # Response.body
            continue
        for bfn, bwt, bval in _pb_parse_fields(val):
            if bfn != cmd or bwt != 2:  # ResponseBody.conversation_messages
                continue
            for cfn, cwt, cval in _pb_parse_fields(bval):
                if cfn != 1 or cwt != 2:  # repeated ConversationMessage
                    continue
                msg: dict = {}
                for mfn, mwt, mval in _pb_parse_fields(cval):
                    if mfn == 3 and mwt == 0:
                        msg["server_message_id"] = str(mval)
                    elif mfn == 6 and mwt == 0:
                        msg["message_type"] = mval
                    elif mfn == 7 and mwt == 0:
                        msg["sender"] = str(mval)
                    elif mfn == 8 and mwt == 2:
                        msg["content"] = mval.decode("utf-8", errors="replace")
                if msg.get("content") is not None:
                    out.append(msg)
    return out

# 抖音 web IM 发送私信的内嵌 status_code 提示（非官方开放平台错误码，依据实测经验）
# 注意：7911 是风控/安全校验类错误，抖音返回“系统繁忙，重新登录后可以正常使用私信功能”，
# 真实含义是“本次发送请求未通过抖音的安全校验”——通常是签名/凭证失效，需要重新登录刷新。
_STATUS_CODE_HINTS = {
    7911: "私信请求未通过抖音安全校验（风控）。抖音提示“重新登录后可正常使用”，"
          "通常是 IM 签名/凭证失效或与浏览器环境不一致：请停止托管后用『浏览器模式』"
          "重新登录该账号并打开一次私信页，刷新 cookie、web_protect、keys、msToken/a_bogus 后再试；"
          "同时放慢自动回复频率，避免触发风控。",
    7913: "私信被风控拦截（发送过于频繁/内容触发风控），建议降低自动回复频率、放慢节奏。",
    7919: "对方设置了不接收陌生人私信。",
    7423: "你与对方未【互相关注】，抖音仅允许互关用户发送该类型消息（图片/表情/卡片等富媒体）。"
          "请改用纯文字回复，或先与对方互相关注后再发送图片；也可让对方先主动私信你以打开会话窗口。",
    8003: "发送频率或会话状态受限（可能触发频控、陌生人私信窗口已过期，或对方未互关）。"
          "图片消息另需确认已上传到抖音 CDN（payload 含 uri），且勿使用本机 /api/media 地址。"
          "建议放慢自动回复节奏，让对方先发一条文字消息后再回复。",
    60021: "图片消息被拒绝：通常是因为图片未上传到抖音 CDN（使用了 localhost/本机地址），"
           "或图片资源无效。请重新选择图片发送；若仍失败，请用浏览器模式重新登录后再试。",
    8004: "抖音业务规则拒绝投递（raw_check_code=1，非签名/凭证问题）。常见原因："
          "① 对方只发了表情/点赞类互动（如[赞]），平台不允许对此类消息回复；"
          "② 陌生人私信回复窗口已过期或未互关；"
          "③ 账号对该用户的发送被临时限流。可让对方发一条正常文字后再试，或降低自动回复频率。",
    8101: "抖音业务规则拒绝投递（raw_check_code=1，签名/凭证已正常）。这通常是"
          "“关系/隐私/频率”限制，而非程序问题：① 你与对方非互关，陌生人主动私信有条数上限"
          "（常为很少几条，发完即被拦）；② 对方隐私设置为“不接收陌生人私信”；"
          "③ 短时间内对同一会话发送过多被临时限制。建议：用一个与该账号【互相关注】、"
          "或【对方先主动发起会话】的真实用户来测试，不要用两个互不关注的小号互发。",
}

# 8xxx 段普遍是“业务/风控/关系”层面的拒绝（签名已通过），统一兜底文案
_BUSINESS_REJECT_FALLBACK = (
    "抖音业务层拒绝投递（签名与凭证已通过安全校验，请求已到达抖音服务端）。"
    "多为关系链/隐私设置/陌生人私信条数或频率限制所致，并非发送程序的 bug。"
    "请改用与该账号互关、或对方先主动私信过的真实用户进行测试。"
)

# 这些 status_code 表示“签名凭证失效/安全校验未通过”，可通过重新采集 web_protect 后重试
_CREDENTIAL_EXPIRED_CODES = {7911}


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._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 _resolve_authoritative_uid(self, auth) -> int:
        """用 query/user 接口核验当前账号真实 UID，并回写 session.my_uid。

        采集端从 tea_cache 推断的 my_uid 可能取到访客/对方 id（导致 cmd=609
        INVALID_REQUEST、会话列表为 0）。query/user 返回的 user_uid 才是权威值。
        核验成功后写回 session 并打标，避免每次发送都请求接口。
        """
        sess = self.session
        if getattr(sess, "uid_verified", False) and sess.my_uid:
            return int(sess.my_uid)
        resolved = None
        try:
            resolved = auth.get_uid()
        except Exception as e:
            logger.warning(f"query/user 解析 my_uid 失败: {e}")
        if resolved and str(resolved).isdigit():
            resolved = int(resolved)
            old = int(sess.my_uid or 0)
            if old and old != resolved:
                logger.warning("my_uid 修正(query/user)：采集值 %s -> 权威值 %s", old, resolved)
            sess.my_uid = resolved
            # 本系统里 device_id 等同账号 uid（采集端常与 my_uid 一起取错，导致会话列表为 0）。
            # device_id 为空 / 非数字 / 等于旧的错误 my_uid 时，一并修正为权威 uid。
            dev = str(sess.device_id or "")
            if (not dev.isdigit()) or (old and dev == str(old)):
                sess.device_id = str(resolved)
            sess.uid_verified = True
            return resolved
        return int(sess.my_uid or 0)

    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)
                logger.warning(
                    "get_conversation_info error: cmd=%s message=%r error_desc=%r",
                    response_proto.cmd, response_proto.message, response_proto.error_desc,
                )
                return "", "", ""
            conv_id, short_id, ticket = self._parse_conversation_body(response_proto)
            if not (conv_id and short_id and ticket):
                # 诊断：get_info 没拿到完整 ticket 时记录，便于判断是否退回到 create_conversation
                logger.info(
                    "get_conversation_info incomplete: cmd=%s message=%r conv_id=%s short_id=%s has_ticket=%s",
                    response_proto.cmd, response_proto.message, conv_id, short_id, bool(ticket),
                )
            return conv_id, short_id, ticket
        except Exception as e:
            self._set_error(str(e))
            logger.warning(f"get_conversation_info exception: {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 "", "", ""

    async def get_conversation_messages(
        self,
        auth,
        conversation_id: str,
        conversation_short_id: int = 0,
        limit: int = 20,
        direction: int = 1,
        anchor_index: int = 0,
    ) -> list[dict]:
        """按会话拉取最近消息（含相册图片/语音等真实 content + URL）。

        抖音相册图片(type 27)等通过 WS 推送时 content 为空，需用本接口补全。
        cmd=301，body 字段号==301，读接口 signed=False 即可。
        """
        from .proto_builder import ProtoBuilder

        cmd = CMD_GET_MESSAGES_BY_CONVERSATION
        try:
            request = await asyncio.to_thread(ProtoBuilder.build_normal_request, auth, cmd)
            conv_req = (
                _pb_str(1, conversation_id)
                + _pb_int(2, 1)
                + _pb_int(3, int(conversation_short_id or 0))
                + _pb_int(4, int(direction))
                + _pb_int(5, int(anchor_index))
                + _pb_int(6, int(limit))
            )
            payload = request.SerializeToString() + _pb_msg(8, _pb_msg(cmd, conv_req))
            resp = await self._post_protobuf(
                "https://imapi.douyin.com/v1/message/get_by_conversation",
                auth,
                payload,
                signed=False,
                log_label="get_messages",
            )
            resp.raise_for_status()
            messages = _pb_parse_conversation_messages(resp.content, cmd)
            logger.info(
                "get_conversation_messages: conv=%s got=%d", conversation_id, len(messages)
            )
            return messages
        except Exception as e:
            self._set_error(str(e))
            logger.warning("get_conversation_messages failed: %s", e)
            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, *, enrich_profiles: bool = True) -> 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)
            # Only a transport/parse failure warrants trying GET.  An empty
            # JSON object/list can be a perfectly valid empty inbox response.
            if data is None:
                data = await self._request("GET", "/v1/conversation/list", body)
            if data is None:
                # Payload variants only help with schema compatibility.  They
                # cannot repair a network outage, so stop after POST + GET
                # both fail instead of occupying a scarce global slot for up
                # to four more full request timeouts.
                logger.warning("Conversation poll transport failed; skipping payload fallbacks")
                break

            status_code = data.get("status_code") if isinstance(data, dict) else None
            error_text = ""
            if isinstance(data, dict):
                error_text = str(
                    data.get("error_desc")
                    or data.get("message")
                    or data.get("error")
                    or ""
                ).strip().lower()
            explicit_success = status_code in (0, "0")
            structured_without_status = (
                isinstance(data, (dict, list)) and status_code is None
            )
            terminal_credential_error = (
                status_code not in (None, 0, "0")
                and any(
                    marker in error_text
                    for marker in (
                        "empty token",
                        "invalid token",
                        "token expired",
                        "credential expired",
                        "authentication",
                        "unauthorized",
                        "not login",
                        "not logged",
                    )
                )
            )

            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)

            # Compatibility payloads are alternatives, not pagination.  Stop
            # after a successful empty response as well as a non-empty one;
            # otherwise every idle account issues three identical endpoint
            # calls on every poll.  Credential errors cannot be repaired by
            # changing only the JSON shape, so do not amplify those either.
            if (
                conversations
                or explicit_success
                or structured_without_status
                or terminal_credential_error
            ):
                break

        my_uid = int(self.session.my_uid or 0)
        enriched: list[dict] = []
        for item in conversations:
            conv = enrich_conversation_item(item, my_uid)
            if not enrich_profiles:
                enriched.append(conv)
                continue
            peer_uid = str(conv.get("peer_uid") or "")
            name = (conv.get("sender_name") or "").strip()
            avatar = str(conv.get("sender_avatar") or "").strip()
            if peer_uid and (is_generic_peer_name(name, peer_uid) or not avatar):
                profile = await fetch_peer_profile(self.session, peer_uid, self.account_id or 0)
                if profile.get("nickname"):
                    conv["sender_name"] = profile["nickname"]
                if profile.get("avatar_url"):
                    conv["sender_avatar"] = profile["avatar_url"]
                if profile.get("uid"):
                    conv["peer_uid"] = str(profile["uid"])
                    conv["sender_id"] = str(profile["uid"])
            enriched.append(conv)

        logger.info(f"Fetched {len(enriched)} conversations from IM API")
        return enriched

    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 send_text_message(
        self,
        conversation_id: str,
        content: str,
        conversation_short_id: str = "",
        _bypass_global_queue: bool = False,
    ) -> bool:
        """通过 IM API 发送 Protobuf 编码的私信（带接口签名）

        content 可为纯文本，或 JSON 格式的结构化回复（文本/网址/卡片）。
        """
        if not _bypass_global_queue:
            # This is the common write entry point used by automatic replies,
            # manual sends, follow welcomes and the no-worker API fallback.
            # Queue the whole ticket-resolution/upload/send transaction so no
            # caller can accidentally bypass the cross-account bandwidth cap.
            from .traffic_control import submit_outbound

            conversation_hint = str(conversation_id or "")[-12:]

            async def _queued_send() -> bool:
                # Use a client owned by the dispatcher.  If an HTTP request is
                # cancelled while this job is already active, the request-level
                # context may close, but the dispatcher must finish the active
                # upload/send before it starts another bandwidth-heavy job.
                async with DouyinImHttpClient(
                    self.session,
                    account_id=self.account_id,
                ) as queued_http:
                    sent = await queued_http.send_text_message(
                        conversation_id,
                        content,
                        conversation_short_id=conversation_short_id,
                        _bypass_global_queue=True,
                    )
                    self.last_send_meta = dict(queued_http.last_send_meta)
                    self.last_error = queued_http.last_error
                    self.last_send_needs_refresh = queued_http.last_send_needs_refresh
                    self.last_request_debug = queued_http.last_request_debug
                    return sent

            return await submit_outbound(
                int(self.account_id or 0),
                _queued_send,
                description=f"IM send {conversation_hint}",
            )

        import asyncio
        import os

        from .auth import DouyinAuth
        from .image_upload import prepare_image_reply_spec
        from .proto_builder import ProtoBuilder
        from .reply_payload import build_msg_payload, parse_reply_content

        self._set_error("")
        self.last_send_needs_refresh = False
        auth = DouyinAuth.from_im_session(self.session)
        my_uid = await asyncio.to_thread(self._resolve_authoritative_uid, auth)
        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}")

        reply_spec = parse_reply_content(content)
        if reply_spec.get("type") == "image":
            upload_dir = os.path.join(
                os.path.dirname(os.path.dirname(os.path.dirname(__file__))),
                "uploads",
                "messages",
            )
            reply_spec, upload_err = await asyncio.to_thread(
                prepare_image_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

        request_proto = await asyncio.to_thread(
            ProtoBuilder.build_send_message_request,
            auth,
            conversation_id,
            conv_short_id,
            ticket,
            *build_msg_payload(reply_spec),
        )
        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),
        }
        url = "https://imapi.douyin.com/v1/message/send"
        try:
            resp = await self._post_protobuf(
                url,
                auth,
                request_proto.SerializeToString(),
                signed=True,
                log_label="send",
                proto_hint=proto_hint,
            )
            resp.raise_for_status()

            from .pb_decode import analyze_send_response

            # 官方 Response.proto 未建模“发送响应”，仅 error_desc 为空/message=OK 都不代表投递成功，
            # 真正成功的强信号是 body 内带服务端分配的 server_message_id。
            result = analyze_send_response(resp.content)
            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 = ""
                if result.get("delivered_with_notice"):
                    sc = result.get("status_code")
                    sr = result.get("status_reason") or ""
                    notice = (
                        f"（已投递，抖音附带业务提示 status_code={sc}"
                        + (f"：{sr}" if sr else "")
                        + "，raw_check_code=0 表示已通过风控，对方可正常收到）"
                    )
                logger.info(
                    f"Direct message delivered: {target} server_message_id={server_mid}{notice}"
                )
                system_logger.record(
                    "私信已投递",
                    detail=f"{target}；server_message_id={server_mid}{notice}；resp[{result.get('summary')}]",
                    level="success",
                    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}"
            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 = await asyncio.to_thread(self._resolve_authoritative_uid, auth)
        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)
                # 诊断：把抖音原始响应（cmd/message/error_desc + hex 片段）落到系统日志，
                # 用于区分「登录/凭证失效」(message/error_desc 有提示) vs「接口需 a_bogus 签名」(多为空)。
                raw_hex = (resp.content or b"")[:160].hex()
                diag = (
                    f"创建会话失败：cmd={response_proto.cmd} "
                    f"message={msg!r} error_desc={response_proto.error_desc!r} "
                    f"http={resp.status_code} resp_len={len(resp.content or b'')} "
                    f"hex={raw_hex}"
                )
                logger.error(f"create_conversation missing body: {diag}")
                system_logger.record(
                    "创建会话失败(INVALID_REQUEST)",
                    detail=diag
                    + "；若 message/error_desc 含登录/凭证提示=登录失效需重登；"
                    "若均为空=接口可能需 a_bogus 签名，请把本行发给开发定位。",
                    level="error",
                    category="send",
                    account_id=self.account_id,
                )
                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 "", "", ""