import asyncio
import json
import logging
import re
import time
from typing import Any, Optional
from urllib.parse import urlparse

import httpx

from utils import system_logger
from rpa_engine.egress_channels import (
    EgressChannelUnavailable,
    resolve_fixed_channel,
    resolve_send_channels,
)
from .conv_util import (
    build_conversation_id,
    conversation_belongs_to,
    normalize_conversation_id,
    resolve_peer_uid,
)
from .message_content import format_im_message, serialize_message_content
from .peer_profile import enrich_conversation_item, fetch_peer_profile, is_generic_peer_name
from .protocol import 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
# 「按用户拉取收件箱」cmd：抖音网页版打开私信时用它一次性同步各会话最新消息。
CMD_GET_MESSAGES_BY_USER_INIT = 200

# MessageBody 的字段号（与 Response.proto 的 MessageBody 一致）。
_MSG_FIELD_CONVERSATION_ID = 1
_MSG_FIELD_SERVER_MESSAGE_ID = 3
_MSG_FIELD_CONVERSATION_SHORT_ID = 5
_MSG_FIELD_MESSAGE_TYPE = 6
_MSG_FIELD_SENDER = 7
_MSG_FIELD_CONTENT = 8

_CONVERSATION_ID_RE = re.compile(r"^0:\d+:\d+:\d+$")

# 托管轮询只需覆盖长连接断开的时间窗；游标是微秒时间戳，窗口越小响应越小
# （实测同一账号：游标 0 -> 113KB，回看 30 分钟 -> 约 2KB）。
INBOX_POLL_LOOKBACK_SECONDS = 1800.0

# 会话列表被抖音拒绝时的系统日志节流：托管期间每个账号最多每 30 分钟记一次，
# 既保证「收不到私信」有据可查，又不会被每轮轮询刷屏。
_CONVERSATION_REJECT_LOG_INTERVAL = 1800.0
_conversation_reject_logged_at: dict[tuple[int, str], float] = {}


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 _as_int(value: Any) -> int:
    try:
        return int(str(value or 0))
    except (TypeError, ValueError):
        return 0


def _pb_message_body(buf: bytes) -> Optional[dict]:
    """把一段字节按 MessageBody 解析；形状不像就返回 None。

    判据是「有合法的 conversation_id + server_message_id」，而不是它出现在
    哪个字段号上——收件箱响应里 messages 挂在哪一层随接口而变。
    """
    msg: dict = {}
    try:
        fields = _pb_parse_fields(buf)
    except Exception:
        return None
    for fn, wt, val in fields:
        if fn == _MSG_FIELD_CONVERSATION_ID and wt == 2:
            try:
                conv_id = val.decode("utf-8")
            except Exception:
                return None
            if not _CONVERSATION_ID_RE.match(conv_id):
                return None
            msg["conversation_id"] = conv_id
        elif fn == _MSG_FIELD_SERVER_MESSAGE_ID and wt == 0:
            msg["server_message_id"] = str(val)
        elif fn == _MSG_FIELD_CONVERSATION_SHORT_ID and wt == 0:
            msg["conversation_short_id"] = str(val)
        elif fn == _MSG_FIELD_MESSAGE_TYPE and wt == 0:
            msg["message_type"] = val
        elif fn == _MSG_FIELD_SENDER and wt == 0:
            msg["sender"] = str(val)
        elif fn == _MSG_FIELD_CONTENT and wt == 2:
            msg["content"] = val.decode("utf-8", errors="replace")
    if msg.get("conversation_id") and msg.get("server_message_id"):
        return msg
    return None


def _pb_collect_message_bodies(
    buf: bytes,
    out: list[dict],
    depth: int = 0,
) -> None:
    """递归找出响应体里所有 MessageBody。"""
    if depth > 6:
        return
    parsed = _pb_message_body(buf)
    if parsed is not None:
        out.append(parsed)
        return
    try:
        fields = _pb_parse_fields(buf)
    except Exception:
        return
    for _fn, wt, val in fields:
        if wt == 2 and isinstance(val, bytes) and val:
            _pb_collect_message_bodies(val, out, depth + 1)


def _pb_parse_inbox_messages(content: bytes, cmd: int) -> list[dict]:
    """从 get_by_user_init 响应里取出各会话的最新消息。"""
    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.<cmd>
                continue
            _pb_collect_message_bodies(bval, out)
    return out


def _is_inbox_control_message(msg: dict) -> bool:
    """收件箱里的会话控制/状态帧（不是用户发的消息）。"""
    from .protocol import _is_control_payload

    try:
        message_type = int(msg.get("message_type") or 0)
    except (TypeError, ValueError):
        message_type = 0
    content_json: Any = None
    raw = msg.get("content")
    if raw:
        try:
            content_json = json.loads(raw)
        except Exception:
            content_json = None
    return _is_control_payload(content_json, message_type)


def _pb_parse_inbox_conversations(content: bytes, cmd: int) -> list[dict]:
    """取出 cmd 200 响应里的会话条目。

    响应体除了 messages(字段 1) 还带一组会话条目(字段 6)：
      f1=conversation_short_id(varint)  f4=conversation_id(string)
    游标为 0 时这组条目就是账号的完整会话列表，所以「列全部会话」不必再去拉
    cmd 203 的 1.5MB 全量快照。
    """
    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:
                continue
            for cfn, cwt, cval in _pb_parse_fields(bval):
                if cfn != 6 or cwt != 2:  # repeated conversation entry
                    continue
                short_id = ""
                conv_id = ""
                for efn, ewt, eval_ in _pb_parse_fields(cval):
                    if efn == 1 and ewt == 0:
                        short_id = str(eval_)
                    elif efn == 4 and ewt == 2:
                        try:
                            candidate = eval_.decode("utf-8")
                        except Exception:
                            continue
                        if _CONVERSATION_ID_RE.match(candidate):
                            conv_id = candidate
                if conv_id:
                    out.append(
                        {
                            "conversation_id": conv_id,
                            "conversation_short_id": short_id,
                        }
                    )
    return out


def _pb_parse_inbox_page(content: bytes, cmd: int) -> tuple[int, bool]:
    """返回收件箱这一页的 (next_cursor, has_more)。"""
    next_cursor = 0
    has_more = False
    for fn, wt, val in _pb_parse_fields(content):
        if fn != 6 or wt != 2:
            continue
        for bfn, bwt, bval in _pb_parse_fields(val):
            if bfn != cmd or bwt != 2:
                continue
            for cfn, cwt, cval in _pb_parse_fields(bval):
                if cfn == 2 and cwt == 0:
                    next_cursor = int(cval)
                elif cfn == 3 and cwt == 0:
                    has_more = bool(cval)
    return next_cursor, has_more


def _pb_response_status(content: bytes) -> tuple[Optional[int], str]:
    """返回 IM protobuf 响应的 (status_code, message)。"""
    status: Optional[int] = None
    message = ""
    try:
        for fn, wt, val in _pb_parse_fields(content):
            if fn == 3 and wt == 0:
                status = int(val)
            elif fn == 4 and wt == 2:
                message = val.decode("utf-8", errors="replace")
    except Exception:
        return None, ""
    return status, message


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}

_CHANNEL_RETRYABLE_ERROR_MARKERS = (
    "INVALID_REQUEST",
    "DECISION=KICK",
    "STATUS_CODE=7911",
    "ALL CONNECTION ATTEMPTS FAILED",
    "CANNOT ASSIGN REQUESTED ADDRESS",
    "CONNECTTIMEOUT",
    "CONNECT TIMEOUT",
    "CONNECTION REFUSED",
    "NETWORK IS UNREACHABLE",
    "NO ROUTE TO HOST",
)


def _is_channel_retryable_error_text(detail: str) -> bool:
    upper = str(detail or "").upper()
    return any(marker in upper for marker in _CHANNEL_RETRYABLE_ERROR_MARKERS)


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 = "",
    egress_public_ip: str = "",
    egress_source_ip: 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 egress_public_ip or egress_source_ip:
        lines.append(
            "  egress: "
            f"public_ip={egress_public_ip or '(detecting/default)'} "
            f"source_ip={egress_source_ip or '(default route)'}"
        )
    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,
        *,
        source_ip: Optional[str] = None,
        egress_public_ip: str = "",
    ):
        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
        # Only explicit pre-delivery/security rejection failures may switch to
        # another channel. Ambiguous read timeouts stay false to avoid duplicates.
        self.last_send_channel_retryable: bool = False
        self.last_request_debug: str = ""
        # 会话列表被抖音判定为「请求本身不合法」时置位。换 payload、换 cookie
        # 都修不好，上层据此停掉这轮轮询，别每 120 秒白打一次请求。
        self.conversation_list_unsupported: bool = False
        # 最近一次收件箱响应里的会话条目（只覆盖翻到的那些页）
        self._last_inbox_conversations: list[dict] = []
        # 翻页预算用尽但抖音还说 has_more：这次拿到的会话列表不完整
        self.inbox_truncated: bool = False
        self._proxy_url: str = ""
        self._source_ip_override = str(source_ip or "").strip()
        self._egress_public_ip_override = str(egress_public_ip or "").strip()
        self._source_ip: str = ""
        self._egress_public_ip: str = ""

    async def __aenter__(self):
        from rpa_engine.runtime_config import httpx_proxy

        self._source_ip = self._source_ip_override or str(
            getattr(self.session, "egress_source_ip", "") or ""
        ).strip()
        self._egress_public_ip = self._egress_public_ip_override or str(
            getattr(self.session, "egress_public_ip", "") or ""
        ).strip()
        if self._egress_public_ip and not self._source_ip:
            route = await resolve_fixed_channel(self._egress_public_ip)
            self._source_ip = str(route.source_ip or "")

        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)
        configured_proxy = httpx_proxy()
        # An account-selected source address and a global proxy describe two
        # different exits. The account channel is the more specific setting.
        proxy = None if (self._egress_public_ip or self._source_ip) else configured_proxy
        if configured_proxy and proxy is None:
            logger.info(
                "Account egress channel overrides KEFU_DOUYIN_PROXY for this IM request"
            )
        transport_kwargs: dict[str, Any] = {}
        if self._source_ip:
            transport_kwargs["local_address"] = self._source_ip
        if proxy:
            transport_kwargs["proxy"] = proxy
            self._proxy_url = proxy
            logger.info(f"IM HTTP client using proxy: {_mask_proxy(proxy)}")
        if transport_kwargs:
            client_kwargs["transport"] = httpx.AsyncHTTPTransport(**transport_kwargs)
        elif proxy:
            client_kwargs["proxy"] = proxy
        if self._egress_public_ip or self._source_ip:
            logger.info(
                "IM HTTP client egress: public_ip=%s source_ip=%s",
                self._egress_public_ip or "default",
                self._source_ip or "default",
            )
        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 _report_conversation_list_rejected(self, reason: str) -> None:
        """把「会话列表被抖音拒绝」变成可见故障，而不是静默的空收件箱。"""
        self._set_error(f"会话列表接口被抖音拒绝：{reason}")
        self.conversation_list_unsupported = True
        logger.warning(
            "Conversation list rejected by IM API (account=%s): %s",
            self.account_id,
            reason,
        )
        key = (int(self.account_id or 0), reason)
        now = time.monotonic()
        last_at = _conversation_reject_logged_at.get(key)
        if last_at is not None and now - last_at < _CONVERSATION_REJECT_LOG_INTERVAL:
            return
        _conversation_reject_logged_at[key] = now
        system_logger.record(
            "会话列表接口被抖音拒绝，已停用轮询兜底",
            detail=(
                f"抖音返回：{reason}。该请求本身被判定为不合法，重试也修不好，"
                "本轮托管不再重复调用。私信改为完全依赖实时长连接接收；"
                "长连接断开期间漏收的消息无法再通过轮询补齐。"
            ),
            level="error",
            category="poll",
            account_id=self.account_id,
        )

    def _resolve_authoritative_uid(self, auth) -> int:
        """Resolve a usable UID without overwriting a known IM identity.

        Douyin's query/user ``user_uid`` can differ from the UID used by IM.
        It is therefore only a last-resort value when the session has no UID;
        a profile-verified or collected numeric UID always wins.
        """
        sess = self.session
        auth.source_ip = self._source_ip
        if 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)
            sess.my_uid = resolved
            dev = str(sess.device_id or "")
            if not dev.isdigit():
                sess.device_id = str(resolved)
            return resolved
        return int(sess.my_uid or 0)

    def _peer_matches_expectation(
        self,
        conversation_id: str,
        peer_uid,
        expected_peer_uid: str,
    ) -> bool:
        """收件人必须与调用方指定的 UID 一致，否则拒发。"""
        expected = str(expected_peer_uid or "").strip()
        if not expected:
            return True
        actual = str(peer_uid or "").strip()
        if actual == expected:
            return True
        detail = (
            f"发送目标与预期不一致：会话 {conversation_id} 的对方是 {actual or '未知'}，"
            f"调用方指定的收件人是 {expected}，已拒绝发送以免发错人。"
        )
        self._set_error(detail)
        self.last_send_channel_retryable = False
        self._log_send_failure(conversation_id, detail)
        logger.error(
            "Account %s refused send: peer mismatch conv=%s actual=%s expected=%s",
            self.account_id,
            conversation_id,
            actual,
            expected,
        )
        return False

    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,
            egress_public_ip=self._egress_public_ip,
            egress_source_ip=self._source_ip,
        )
        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:
            # 同样要用 x_tt_token：带 auth.ticket 时抖音回 OK 但正文恒为空，
            # 于是 WS 瘦推送的图片/语音一直补不全真实 content。
            request = await asyncio.to_thread(ProtoBuilder.build_read_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 fetch_inbox_messages(
        self,
        limit: int = 50,
        lookback_seconds: float = INBOX_POLL_LOOKBACK_SECONDS,
        max_pages: int = 1,
    ) -> list[dict]:
        """用 protobuf 拉取收件箱消息，按游标翻页。

        imapi.douyin.com 只接受 protobuf：发 JSON body 会被当成 protobuf 解析，
        固定返回 status_code=1 "unexepcted session length"（与 Cookie 无关，
        实测不带任何 Cookie 也是同一条错误）。这里用与发送/拉消息同一套 Request
        信封，抖音网页版打开私信时用的也是这个 cmd。

        响应是**分页**的：body 的 f2=next_cursor、f3=has_more，随附的会话条目
        只覆盖这一页里出现过的会话。实测 cursor=0 返回 9 个会话且 has_more=1，
        翻 6 页后累计 35 个且仍未翻完——所以「一次请求 = 完整会话列表」是错的，
        翻不完时必须把 inbox_truncated 置位，别把一页伪装成全部。
        """
        from .auth import DouyinAuth
        from .proto_builder import ProtoBuilder

        cmd = CMD_GET_MESSAGES_BY_USER_INIT
        auth = DouyinAuth.from_im_session(self.session)
        auth.source_ip = self._source_ip
        # 字段 1 是游标（微秒时间戳）。托管轮询只回看一个窗口：这条链路只用来
        # 补齐长连接断开期间漏收的消息。lookback_seconds<=0 表示不设游标
        # （游标 0 = 从头翻），别算成「now」，那等于只要比此刻更新的消息。
        if lookback_seconds <= 0:
            cursor = 0
        else:
            cursor = max(0, int((time.time() - lookback_seconds) * 1_000_000))

        messages: list[dict] = []
        conversations: list[dict] = []
        seen_conversations: set[str] = set()
        self.inbox_truncated = False
        pages = max(1, int(max_pages))
        for page in range(pages):
            request = await asyncio.to_thread(
                ProtoBuilder.build_read_request, auth, cmd
            )
            body = _pb_int(1, cursor) + _pb_int(2, int(limit))
            payload = request.SerializeToString() + _pb_msg(8, _pb_msg(cmd, body))
            resp = await self._post_protobuf(
                f"{IMAPI_BASE}/v1/message/get_by_user_init",
                auth,
                payload,
                signed=False,
                log_label="inbox",
            )
            resp.raise_for_status()
            status_code, message = _pb_response_status(resp.content)
            if status_code is not None and status_code != 0:
                self._report_conversation_list_rejected(
                    message or f"status_code={status_code}"
                )
                return []
            if page == 0:
                self._adopt_authoritative_uid(resp.content)
            messages.extend(_pb_parse_inbox_messages(resp.content, cmd))
            for entry in _pb_parse_inbox_conversations(resp.content, cmd):
                conv_id = str(entry.get("conversation_id") or "")
                if conv_id and conv_id not in seen_conversations:
                    seen_conversations.add(conv_id)
                    conversations.append(entry)

            next_cursor, has_more = _pb_parse_inbox_page(resp.content, cmd)
            if not has_more:
                break
            # 游标不前进就停：否则同一页会被无限翻下去。
            if not next_cursor or next_cursor == cursor:
                break
            cursor = next_cursor
            if page == pages - 1:
                self.inbox_truncated = True
                logger.info(
                    "Inbox paging stopped at the %d-page budget for account %s; "
                    "%d conversations so far, more remain",
                    pages,
                    self.account_id,
                    len(conversations),
                )

        self._last_inbox_conversations = conversations
        return messages

    def _adopt_authoritative_uid(self, content: bytes) -> None:
        """响应字段 13 是抖音认定的本账号 IM uid，用它纠正 session.my_uid。

        my_uid 取错时 _is_self_message 拦不住自己发的消息（机器人会自问自答），
        resolve_peer_uid 也会把会话对端认成自己。
        """
        uid = 0
        try:
            for fn, wt, val in _pb_parse_fields(content):
                if fn == 13 and wt == 0:
                    uid = int(val)
                    break
        except Exception:
            return
        if not uid or uid == int(self.session.my_uid or 0):
            return
        logger.warning(
            "Account %s IM uid corrected from %s to %s (imapi response field 13)",
            self.account_id,
            self.session.my_uid,
            uid,
        )
        previous = int(self.session.my_uid or 0)
        self.session.my_uid = uid
        self.session.uid_verified = True
        # 托管注册表按 UID 记录「本系统正在托管谁」。纠正后必须迁移，否则回环
        # 防护会认错人：旧 UID 永远留在表里，真实 UID 从未登记。只迁移确实已登记
        # 的托管身份，避免 API 侧的临时客户端把自己也登记进去。
        from . import hosted_registry

        if hosted_registry.is_hosted(previous):
            hosted_registry.unregister(previous)
            hosted_registry.register(uid)

    async def get_conversations(
        self,
        *,
        enrich_profiles: bool = True,
        lookback_seconds: float = INBOX_POLL_LOOKBACK_SECONDS,
        max_pages: int = 1,
    ) -> list[dict]:
        """拉取会话列表，返回标准化会话。

        lookback_seconds=0 表示从头翻；max_pages 是翻页预算。抖音不提供「一次
        取回全部会话」的接口，翻页预算用完时 inbox_truncated 会被置位——调用方
        必须知道自己拿到的可能只是一部分，不能把一页当成完整会话列表。
        托管轮询用默认的小窗口 + 单页，只为补齐长连接断开期间漏收的消息。
        """
        conversations: list[dict] = []
        try:
            messages = await self.fetch_inbox_messages(
                lookback_seconds=lookback_seconds,
                max_pages=max_pages,
            )
        except Exception as exc:
            self._set_error(str(exc))
            logger.warning("Conversation poll transport failed: %s", exc)
            return []

        # 一个会话只保留最新的一条：server_message_id 单调递增。
        latest: dict[str, dict] = {}
        for msg in messages:
            conv_id = str(msg.get("conversation_id") or "")
            if not conv_id:
                continue
            # 会话状态/已读位等控制帧不是用户消息：既不该当成会话预览，
            # 更不该被 _handle_incoming 拿去匹配自动回复（WS 侧同样过滤）。
            if _is_inbox_control_message(msg):
                continue
            current = latest.get(conv_id)
            if current is None or _as_int(msg.get("server_message_id")) > _as_int(
                current.get("server_message_id")
            ):
                latest[conv_id] = msg

        # 窗口内没有消息、但账号里确实存在的会话也要出现在列表里，
        # 否则「会话列表」会退化成「最近有动静的会话」。
        for entry in self._last_inbox_conversations:
            conv_id = str(entry.get("conversation_id") or "")
            short_id = str(entry.get("conversation_short_id") or "")
            if short_id and short_id != "0":
                self.session.conv_meta.setdefault(conv_id, {})
                self.session.conv_meta[conv_id]["conversation_short_id"] = short_id
            if conv_id and conv_id not in latest:
                latest[conv_id] = {
                    "conversation_id": conv_id,
                    "conversation_short_id": short_id,
                    "server_message_id": "0",
                    "content": "",
                }

        for conv_id, msg in latest.items():
            content = str(msg.get("content") or "")
            try:
                message_type = int(msg.get("message_type") or 0)
            except (TypeError, ValueError):
                message_type = 0
            preview = content
            try:
                parsed = format_im_message(json.loads(content), message_type)
                preview = serialize_message_content(parsed) if parsed else content
            except Exception:
                pass
            short_id = str(msg.get("conversation_short_id") or "")
            if short_id and short_id != "0":
                # 发送私信需要 short_id；从收件箱顺手补上，省掉一次 create。
                self.session.conv_meta.setdefault(conv_id, {})
                self.session.conv_meta[conv_id]["conversation_short_id"] = short_id
            conversations.append(
                {
                    "conversation_id": conv_id,
                    "conversation_short_id": short_id,
                    "sender_name": "",
                    "sender_avatar": None,
                    "content": preview,
                    "raw_content": content,
                    "message_type": message_type,
                    "server_message_id": str(msg.get("server_message_id") or ""),
                    # sender 可能是自己（我方发出的最后一条），不能当 peer；
                    # 交给 enrich_conversation_item 从 conversation_id 推。
                    "sender_uid": str(msg.get("sender") or ""),
                    "unread_count": 0,
                }
            )

        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

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

        content 可为纯文本，或 JSON 格式的结构化回复（文本/网址/卡片）。

        ``expected_peer_uid`` 是调用方认定的收件人 UID。会话归属校验只能保证
        「这条会话是本账号的」，保证不了「这个人就是用户想发的人」——界面按昵称
        兜底匹配会话时可能选中同名的另一个人。带上它，写入点就能在发出去之前
        确认收件人确实是调用方指定的那个。
        """
        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:
                preferred = str(getattr(self.session, "egress_public_ip", "") or "").strip()
                max_attempts = getattr(self.session, "egress_auto_attempts", 1)
                try:
                    routes = await resolve_send_channels(preferred, max_attempts)
                except EgressChannelUnavailable as exc:
                    self._set_error(str(exc))
                    self.last_send_channel_retryable = False
                    self._log_send_failure(conversation_id, str(exc))
                    return False

                kicked_error = ""
                for index, route in enumerate(routes):
                    # Use a client owned by the dispatcher. If an HTTP request
                    # is cancelled while active, the dispatcher still owns the
                    # complete serial ticket/upload/send operation.
                    route_kwargs: dict[str, Any] = {}
                    if route.source_ip or route.public_ip:
                        route_kwargs = {
                            "source_ip": route.source_ip,
                            "egress_public_ip": route.public_ip,
                        }
                    async with DouyinImHttpClient(
                        self.session,
                        account_id=self.account_id,
                        **route_kwargs,
                    ) as queued_http:
                        sent = await queued_http.send_text_message(
                            conversation_id,
                            content,
                            conversation_short_id=conversation_short_id,
                            expected_peer_uid=expected_peer_uid,
                            _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_send_channel_retryable = queued_http.last_send_channel_retryable
                    self.last_request_debug = queued_http.last_request_debug
                    if "DECISION=KICK" in (queued_http.last_error or "").upper():
                        kicked_error = queued_http.last_error
                    if sent:
                        if index:
                            system_logger.record(
                                "公网通道切换后发送成功",
                                detail=(
                                    f"已通过公网 IP {route.public_ip or '默认出口'} 发送；"
                                    f"本次共尝试 {index + 1} 个通道"
                                ),
                                level="success",
                                category="send",
                                account_id=self.account_id,
                            )
                        return True
                    if not queued_http.last_send_channel_retryable or index + 1 >= len(routes):
                        if kicked_error and "DECISION=KICK" not in (self.last_error or "").upper():
                            self.last_error = f"{self.last_error}\n此前通道已返回：{kicked_error}"
                        return False
                    next_route = routes[index + 1]
                    logger.warning(
                        "Account %s send rejected on egress %s; trying %s (%s/%s)",
                        self.account_id,
                        route.public_ip or "default",
                        next_route.public_ip or "default",
                        index + 2,
                        len(routes),
                    )
                    system_logger.record(
                        "发送失败，切换公网通道重试",
                        detail=(
                            f"通道 {route.public_ip or '默认出口'} 明确返回通道/安全校验失败；"
                            f"将串行尝试 {next_route.public_ip or '默认出口'}（{index + 2}/{len(routes)}）"
                        ),
                        level="warning",
                        category="send",
                        account_id=self.account_id,
                    )
                return False

            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
        self.last_send_channel_retryable = 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

        # 跨账号写入闸门：normalize_conversation_id 会把任何会话 ID 改写成
        # 0:1:{本账号}:{末段 UID}，所以一条属于别的账号的会话流到这里会被
        # 静默改写并发给对方的好友。发送前先确认本账号确实是该会话的参与方。
        if not conversation_belongs_to(conversation_id, my_uid):
            detail = (
                f"会话 {conversation_id} 的参与方都不是本账号（uid={my_uid}），"
                "拒绝发送：这条会话属于另一个账号，继续发送会把消息发给别人的好友。"
            )
            self._set_error(detail)
            self.last_send_channel_retryable = False
            self._log_send_failure(conversation_id, detail)
            logger.error(
                "Account %s refused cross-account send to %s (my_uid=%s)",
                self.account_id,
                conversation_id,
                my_uid,
            )
            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
        if not self._peer_matches_expectation(
            conversation_id, peer_uid, expected_peer_uid
        ):
            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

        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
            # 抖音回来的会话 ID 才是真正会被写入的那条：再确认一次收件人没被换掉。
            if not self._peer_matches_expectation(
                conversation_id,
                resolve_peer_uid(conversation_id, my_uid) or peer_uid,
                expected_peer_uid,
            ):
                return False
        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)
            self.last_send_channel_retryable = _is_channel_retryable_error_text(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,
                self._source_ip,
            )
            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 ""
            decision = str(result.get("decision") or "").strip().upper()

            if decision == "KICK":
                # KICK is a terminal, account-session decision.  Retrying the
                # same authenticated write from another source address cannot
                # repair the session and only adds another high-risk request.
                self.last_send_needs_refresh = False
                self.last_send_channel_retryable = False
                detail = (
                    "抖音安全网关返回 decision=KICK，当前登录/安全会话已被服务端踢下线；"
                    "已停止本次发送及公网通道重试，系统正在自动重登录，请留意账号卡片上的二维码并扫码"
                )
            elif decision:
                detail = f"抖音安全网关拒绝发送 decision={decision}"
                if status_reason:
                    detail += f"；抖音提示：{status_reason}"
            elif 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
                # 7911 is a credential/signature problem.  It may be retried
                # once only after refreshing the credentials on the same
                # session; switching egress mid-session makes the fingerprint
                # less consistent and must not be used as the recovery path.
                self.last_send_channel_retryable = False
                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）"

            if "INVALID_REQUEST" in detail.upper():
                # INVALID_REQUEST is a protocol/session rejection, not a
                # transport failure.  A second public IP sends the same invalid
                # request and can invalidate an otherwise recoverable login.
                self.last_send_channel_retryable = False
            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:
            self.last_send_channel_retryable = isinstance(
                e,
                (httpx.ConnectError, httpx.ConnectTimeout, httpx.ProxyError, httpx.PoolTimeout),
            )
            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 "", "", ""