更新

backend/rpa_engine/douyin_im/auth.py

@@ -1,7 +1,7 @@
 import base64
 import json
 import logging
-import requests
+from rpa_engine.egress_channels import source_bound_requests_session
 from .dy_util import (
     trans_cookies,
     generate_msToken,
@@ -63,6 +63,7 @@ class DouyinAuth:
         self.uid = None
         self.msToken = None
         self.web_id = None
+        self.source_ip = ""
 
     def perepare_auth(self, cookieStr: str, web_protect_: str = "", keys_: str = ""):
         self.cookie = trans_cookies(cookieStr)
@@ -158,6 +159,14 @@ class DouyinAuth:
         abogus = generate_a_bogus(query, user_agent=DEFAULT_USER_AGENT)
         params['a_bogus'] = abogus
         
-        resp = requests.get(url, params=params, headers=headers, cookies=self.cookie, verify=False, timeout=10)
+        with source_bound_requests_session(self.source_ip) as client:
+            resp = client.get(
+                url,
+                params=params,
+                headers=headers,
+                cookies=self.cookie,
+                verify=False,
+                timeout=10,
+            )
         resp_json = resp.json()
         return int(resp_json['user_uid'])

backend/rpa_engine/douyin_im/http_client.py

@@ -7,6 +7,11 @@ 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, 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
@@ -153,6 +158,24 @@ _BUSINESS_REJECT_FALLBACK = (
 # 这些 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 中的用户名/密码，仅用于日志展示。"""
@@ -204,6 +227,8 @@ def _format_im_request_debug(
     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}"]
@@ -266,6 +291,12 @@ def _format_im_request_debug(
     )
     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:
@@ -300,7 +331,14 @@ def format_session_credential_summary(session: DouyinImSession) -> str:
 class DouyinImHttpClient:
     """抖音 IM HTTP API 客户端（基于 Cookie 鉴权）"""
 
-    def __init__(self, session: DouyinImSession, account_id: Optional[int] = None):
+    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
@@ -308,12 +346,29 @@ class DouyinImHttpClient:
         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 = ""
         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(),
@@ -328,11 +383,31 @@ class DouyinImHttpClient:
             "follow_redirects": True,
         }
         # 配置 KEFU_DOUYIN_PROXY 时让全部抖音 IM 请求走住宅代理，绕开机房 IP 风控(7911)
-        proxy = httpx_proxy()
+        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:
-            client_kwargs["proxy"] = 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
 
@@ -381,6 +456,7 @@ class DouyinImHttpClient:
         核验成功后写回 session 并打标，避免每次发送都请求接口。
         """
         sess = self.session
+        auth.source_ip = self._source_ip
         if getattr(sess, "uid_verified", False) and sess.my_uid:
             return int(sess.my_uid)
         resolved = None
@@ -524,6 +600,8 @@ class DouyinImHttpClient:
             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)
@@ -905,25 +983,82 @@ class DouyinImHttpClient:
             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,
-                    )
+                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,
+                            _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
-                    return sent
+                    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),
@@ -941,6 +1076,7 @@ class DouyinImHttpClient:
 
         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:
@@ -978,6 +1114,7 @@ class DouyinImHttpClient:
         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
@@ -997,7 +1134,11 @@ class DouyinImHttpClient:
                 "messages",
             )
             reply_spec, upload_err = await asyncio.to_thread(
-                prepare_image_reply_spec, reply_spec, self.session, upload_dir
+                prepare_image_reply_spec,
+                reply_spec,
+                self.session,
+                upload_dir,
+                self._source_ip,
             )
             if upload_err:
                 detail = f"图片上传失败：{upload_err}"
@@ -1066,8 +1207,19 @@ class DouyinImHttpClient:
 
             status_code = result.get("status_code")
             status_reason = result.get("status_reason") or ""
+            decision = str(result.get("decision") or "").strip().upper()
 
-            if status_code is not None and status_code != 0:
+            if decision == "KICK":
+                self.last_send_channel_retryable = True
+                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 段未单独建模的，统一归为“业务层拒绝（签名已通过）”
@@ -1075,6 +1227,7 @@ class DouyinImHttpClient:
                     hint = _BUSINESS_REJECT_FALLBACK
                 # 7911 属于“签名凭证失效/安全校验未过”，标记为可刷新后重试
                 self.last_send_needs_refresh = status_code in _CREDENTIAL_EXPIRED_CODES
+                self.last_send_channel_retryable = self.last_send_needs_refresh
                 detail = f"抖音拒绝投递 status_code={status_code}"
                 if status_reason:
                     detail += f"；抖音提示：{status_reason}"
@@ -1096,6 +1249,8 @@ class DouyinImHttpClient:
                     reason_bits.append(f"cmd={result.get('cmd')}")
                 detail = "；".join(reason_bits) or "接口返回但未确认投递（无 server_message_id）"
 
+            if "INVALID_REQUEST" in detail.upper():
+                self.last_send_channel_retryable = True
             full_detail = f"{detail}；{target}；resp[{result.get('summary')}]"
             if self.last_request_debug:
                 full_detail += f"\n--- 请求详情 ---\n{self.last_request_debug}"
@@ -1104,6 +1259,10 @@ class DouyinImHttpClient:
             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}"

backend/rpa_engine/douyin_im/image_upload.py

@@ -31,6 +31,8 @@ import zlib
 from typing import Any
 from urllib.parse import urlencode
 
+from rpa_engine.egress_channels import source_bound_requests_session
+
 logger = logging.getLogger("douyin_im.image_upload")
 
 _LOCAL_URL_RE = re.compile(
@@ -269,10 +271,8 @@ def _decode_sts(sts_token: str) -> tuple[str, str]:
         return "", ""
 
 
-def _fetch_im_upload_sts(session) -> tuple[str, str, str, str]:
+def _fetch_im_upload_sts(session, source_ip: str = "") -> tuple[str, str, str, str]:
     """返回 (access_key_id, secret_access_key, sts_token, space_name)。"""
-    import requests
-
     from .auth import DouyinAuth
     from .dy_util import (
         DEFAULT_USER_AGENT,
@@ -328,15 +328,16 @@ def _fetch_im_upload_sts(session) -> tuple[str, str, str, str]:
         "Referer": "https://www.douyin.com/",
         "Accept": "application/json, text/plain, */*",
     }
-    resp = requests.get(
-        IM_UPLOAD_CONFIG_URL,
-        params=params,
-        headers=headers,
-        cookies=auth.cookie,
-        timeout=20,
-        verify=False,
-        proxies=_requests_proxies(),
-    )
+    with source_bound_requests_session(source_ip) as client:
+        resp = client.get(
+            IM_UPLOAD_CONFIG_URL,
+            params=params,
+            headers=headers,
+            cookies=auth.cookie,
+            timeout=20,
+            verify=False,
+            proxies=None if source_ip else _requests_proxies(),
+        )
     data = _safe_json(resp)
     if data.get("error"):
         raise RuntimeError(f"获取 IM 上传配置失败：{data['error']}")
@@ -453,10 +454,8 @@ def _extract_apply_inner(data: dict[str, Any]) -> tuple[str, str, str, str]:
 
 
 def _vod_apply_upload_inner(
-    ak: str, sk: str, token: str, space: str, file_size: int
+    ak: str, sk: str, token: str, space: str, file_size: int, source_ip: str = ""
 ) -> tuple[str, str, str, str]:
-    import requests
-
     from .dy_util import DEFAULT_USER_AGENT
 
     now = datetime.datetime.utcnow()
@@ -483,20 +482,21 @@ def _vod_apply_upload_inner(
         secret_access_key=sk,
         service=VOD_SERVICE,
     )
-    resp = requests.get(
-        f"{VOD_HOST}?{qs}",
-        headers={
-            "accept": "*/*",
-            "authorization": authorization,
-            "user-agent": DEFAULT_USER_AGENT,
-            "x-amz-date": amz_date,
-            "x-amz-security-token": token,
-            "Referer": "https://www.douyin.com/",
-        },
-        timeout=30,
-        verify=False,
-        proxies=_requests_proxies(),
-    )
+    with source_bound_requests_session(source_ip) as client:
+        resp = client.get(
+            f"{VOD_HOST}?{qs}",
+            headers={
+                "accept": "*/*",
+                "authorization": authorization,
+                "user-agent": DEFAULT_USER_AGENT,
+                "x-amz-date": amz_date,
+                "x-amz-security-token": token,
+                "Referer": "https://www.douyin.com/",
+            },
+            timeout=30,
+            verify=False,
+            proxies=None if source_ip else _requests_proxies(),
+        )
     data = _safe_json(resp)
     if data.get("error"):
         raise RuntimeError(f"申请上传地址失败：{data['error']}")
@@ -512,10 +512,14 @@ def _vod_apply_upload_inner(
 # ---------------------------------------------------------------------------
 
 def _vod_upload_binary(
-    host: str, store_uri: str, jwt_auth: str, user_id: str, raw: bytes, session=None
+    host: str,
+    store_uri: str,
+    jwt_auth: str,
+    user_id: str,
+    raw: bytes,
+    session=None,
+    source_ip: str = "",
 ) -> None:
-    import requests
-
     from .dy_util import DEFAULT_USER_AGENT
 
     crc32 = format(zlib.crc32(raw) & 0xFFFFFFFF, "08x")
@@ -530,14 +534,15 @@ def _vod_upload_binary(
     }
     if user_id:
         headers["X-Storage-U"] = str(user_id)
-    resp = requests.post(
-        url,
-        headers=headers,
-        data=raw,
-        timeout=60,
-        verify=False,
-        proxies=_requests_proxies(),
-    )
+    with source_bound_requests_session(source_ip) as client:
+        resp = client.post(
+            url,
+            headers=headers,
+            data=raw,
+            timeout=60,
+            verify=False,
+            proxies=None if source_ip else _requests_proxies(),
+        )
     data = _safe_json(resp)
     if data.get("error"):
         raise RuntimeError(f"上传图片数据失败：{data['error']}")
@@ -550,10 +555,8 @@ def _vod_upload_binary(
 # ---------------------------------------------------------------------------
 
 def _vod_commit_upload_inner(
-    ak: str, sk: str, token: str, space: str, session_key: str
+    ak: str, sk: str, token: str, space: str, session_key: str, source_ip: str = ""
 ) -> dict[str, Any]:
-    import requests
-
     from .dy_util import DEFAULT_USER_AGENT
 
     now = datetime.datetime.utcnow()
@@ -580,23 +583,24 @@ def _vod_commit_upload_inner(
         signed_headers=signed_headers,
         service=VOD_SERVICE,
     )
-    resp = requests.post(
-        f"{VOD_HOST}?{qs}",
-        data=body,
-        headers={
-            "accept": "*/*",
-            "authorization": authorization,
-            "content-type": "application/json",
-            "user-agent": DEFAULT_USER_AGENT,
-            "x-amz-content-sha256": payload_hash,
-            "x-amz-date": amz_date,
-            "x-amz-security-token": token,
-            "Referer": "https://www.douyin.com/",
-        },
-        timeout=30,
-        verify=False,
-        proxies=_requests_proxies(),
-    )
+    with source_bound_requests_session(source_ip) as client:
+        resp = client.post(
+            f"{VOD_HOST}?{qs}",
+            data=body,
+            headers={
+                "accept": "*/*",
+                "authorization": authorization,
+                "content-type": "application/json",
+                "user-agent": DEFAULT_USER_AGENT,
+                "x-amz-content-sha256": payload_hash,
+                "x-amz-date": amz_date,
+                "x-amz-security-token": token,
+                "Referer": "https://www.douyin.com/",
+            },
+            timeout=30,
+            verify=False,
+            proxies=None if source_ip else _requests_proxies(),
+        )
     data = _safe_json(resp)
     if data.get("error"):
         raise RuntimeError(f"确认上传失败：{data['error']}")
@@ -613,6 +617,7 @@ def upload_im_image(
     *,
     filename: str = "image.jpg",
     content_type: str = "image/jpeg",
+    source_ip: str = "",
 ) -> dict[str, Any]:
     """上传图片到抖音 IM 私信图床（VOD/zhenzhen 空间）。
 
@@ -622,16 +627,16 @@ def upload_im_image(
     if not raw:
         return {"error": "图片为空"}
     try:
-        ak, sk, token, space = _fetch_im_upload_sts(session)
+        ak, sk, token, space = _fetch_im_upload_sts(session, source_ip)
         host, store_uri, jwt_auth, session_key = _vod_apply_upload_inner(
-            ak, sk, token, space, len(raw)
+            ak, sk, token, space, len(raw), source_ip
         )
         if not host or not store_uri or not jwt_auth:
             return {"error": "申请上传地址失败：缺少 UploadHost/StoreUri/Auth"}
 
         user_id = str(getattr(session, "my_uid", "") or "")
-        _vod_upload_binary(host, store_uri, jwt_auth, user_id, raw, session)
-        _vod_commit_upload_inner(ak, sk, token, space, session_key)
+        _vod_upload_binary(host, store_uri, jwt_auth, user_id, raw, session, source_ip)
+        _vod_commit_upload_inner(ak, sk, token, space, session_key, source_ip)
 
         uri = store_uri.lstrip("/")
         out: dict[str, Any] = {"uri": uri, "md5": hashlib.md5(raw).hexdigest()}
@@ -650,7 +655,12 @@ def upload_im_image(
         return {"error": str(exc)}
 
 
-def prepare_image_reply_spec(spec: dict[str, Any], session, upload_dir: str) -> tuple[dict[str, Any], str]:
+def prepare_image_reply_spec(
+    spec: dict[str, Any],
+    session,
+    upload_dir: str,
+    source_ip: str = "",
+) -> tuple[dict[str, Any], str]:
     """若图片仍是本地地址，则上传到抖音 CDN 并补全 uri。返回 (spec, error)。"""
     if spec.get("type") != "image":
         return spec, ""
@@ -705,7 +715,13 @@ def prepare_image_reply_spec(spec: dict[str, Any], session, upload_dir: str) ->
             return spec, "图片地址必须是抖音 CDN 或本地上传后的地址，外部 URL 无法用于 IM 发送"
         return spec, "缺少可上传的图片数据"
 
-    uploaded = upload_im_image(session, raw, filename=filename, content_type=content_type)
+    uploaded = upload_im_image(
+        session,
+        raw,
+        filename=filename,
+        content_type=content_type,
+        source_ip=source_ip,
+    )
     if uploaded.get("error"):
         return spec, uploaded["error"]
     if not uploaded.get("uri"):

backend/rpa_engine/douyin_im/pb_decode.py

@@ -160,6 +160,7 @@ def analyze_send_response(raw: bytes) -> dict:
         "raw_check_code": None,
         "delivered_with_notice": False,
         "status_reason": "",
+        "decision": "",
         "message": "",
         "error_desc": "",
         "server_message_id": None,
@@ -169,6 +170,45 @@ def analyze_send_response(raw: bytes) -> dict:
     if not raw:
         info["summary"] = "空响应"
         return info
+
+    # 风控/登录网关有时不返回 protobuf，而是直接返回 JSON，例如：
+    # {"decision":"KICK"}。若继续按 protobuf 解码，JSON 的首字节“{”会被
+    # 误读为 wire type 3，只留下 unsupported wire type 3 这种次生错误。
+    stripped = raw.lstrip()
+    if stripped.startswith(b"{"):
+        try:
+            payload = json.loads(stripped.decode("utf-8"))
+        except (UnicodeDecodeError, json.JSONDecodeError):
+            payload = None
+        if isinstance(payload, dict):
+            decision = str(
+                payload.get("decision") or payload.get("decision_type") or ""
+            ).strip()
+            info["decision"] = decision
+            info["status_code"] = payload.get("status_code")
+            info["raw_check_code"] = payload.get("raw_check_code")
+            info["message"] = str(payload.get("message") or "")
+            info["error_desc"] = str(
+                payload.get("error_desc") or payload.get("error") or ""
+            )
+            info["status_reason"] = str(
+                payload.get("tips") or payload.get("reason") or ""
+            )
+            summary_parts = ["JSON响应"]
+            if decision:
+                summary_parts.append(f"decision={decision}")
+            if info["status_code"] is not None:
+                summary_parts.append(f"status_code={info['status_code']}")
+            if info["raw_check_code"] is not None:
+                summary_parts.append(f"raw_check_code={info['raw_check_code']}")
+            if info["message"]:
+                summary_parts.append(f"message={info['message']}")
+            if info["error_desc"]:
+                summary_parts.append(f"error_desc={info['error_desc']}")
+            info["summary"] = " ".join(summary_parts)
+            # /message/send 的正常成功响应是 protobuf；独立 JSON 是网关级响应，
+            # 不能据此确认消息已经写入会话。
+            return info
     try:
         fields = decode_fields(raw)
     except Exception as e:

backend/rpa_engine/douyin_im/peer_profile.py

@@ -7,9 +7,8 @@ import logging
 import time
 from typing import Any, Optional
 
-import requests
-
 from rpa_engine.device_profiles import resolve_user_agent
+from rpa_engine.egress_channels import resolve_fixed_channel, source_bound_requests_session
 from .auth import DouyinAuth
 from .conv_util import resolve_peer_uid
 from .dy_util import (
@@ -99,6 +98,7 @@ def fetch_peer_profile_sync(
     session: DouyinImSession,
     peer_uid: int | str,
     account_id: int = 0,
+    source_ip: str = "",
 ) -> dict[str, str]:
     uid = str(peer_uid or "").strip()
     if not uid.isdigit():
@@ -156,21 +156,22 @@ def fetch_peer_profile_sync(
         "https://www.douyin.com/aweme/v1/web/im/user/info/",
     ]
 
-    proxies = _requests_proxies()
+    proxies = None if source_ip else _requests_proxies()
     for url in endpoints:
         try:
             params = dict(base_params)
             query = splice_url(params)
             params["a_bogus"] = generate_a_bogus(query, user_agent=ua)
-            resp = requests.get(
-                url,
-                params=params,
-                headers=headers,
-                cookies=auth.cookie,
-                verify=False,
-                timeout=12,
-                proxies=proxies,
-            )
+            with source_bound_requests_session(source_ip) as client:
+                resp = client.get(
+                    url,
+                    params=params,
+                    headers=headers,
+                    cookies=auth.cookie,
+                    verify=False,
+                    timeout=12,
+                    proxies=proxies,
+                )
             data = resp.json()
             extracted = _extract_profile_from_payload(data)
             if extracted.get("uid") and not result["uid"]:
@@ -199,12 +200,21 @@ async def fetch_peer_profile(
     from .traffic_control import get_traffic_controller
 
     controller = get_traffic_controller()
+    source_ip = str(getattr(session, "egress_source_ip", "") or "").strip()
+    selected_public_ip = str(getattr(session, "egress_public_ip", "") or "").strip()
+    if selected_public_ip and not source_ip:
+        try:
+            route = await resolve_fixed_channel(selected_public_ip)
+            source_ip = str(route.source_ip or "")
+        except Exception as exc:
+            logger.debug("peer profile egress resolution failed: %s", exc)
     async with controller.background_slot(account_id, "peer profile"):
         return await asyncio.to_thread(
             fetch_peer_profile_sync,
             session,
             peer_uid,
             account_id,
+            source_ip,
         )
 
 

backend/rpa_engine/douyin_im/service.py

@@ -331,7 +331,7 @@ class DouyinImService:
         self.account_id = account_id
         # 由 worker 注入：周期性检测新粉丝并发送关注欢迎语（约每 60s 触发一次）
         self.follow_tick = follow_tick
-        # 由 worker 注入：检测到 IM 登录失效（INVALID_REQUEST）时回调，用于自动下线
+        # 由 worker 注入：检测到 IM 登录失效（INVALID_REQUEST/KICK）时回调，用于自动下线
         self.on_session_invalid = on_session_invalid
         self._on_ready = on_ready
         self._ready_notified = False
@@ -1449,26 +1449,36 @@ class DouyinImService:
         return False, None
 
     async def _note_session_invalid(self, error: str) -> None:
-        """根据发送失败原因判断 IM 是否已退出登录；连续 INVALID_REQUEST 即触发自动下线。
+        """根据发送失败原因判断 IM 是否已退出登录，并触发自动下线。
 
         INVALID_REQUEST 来自 create_conversation/发送：会话/签名被抖音判为无效，强相关于「登录失效」。
+        decision=KICK 是安全网关明确要求终止当前登录态，一次即可确认，无需等待第二次发送。
         而 8xxx/7xxx 等业务错误（关系/频控/内容）说明请求已到达抖音、登录仍有效，重置计数。
         """
         err = error or ""
-        if "INVALID_REQUEST" not in err:
+        upper_err = err.upper()
+        is_kicked = "DECISION=KICK" in upper_err
+        is_invalid_request = "INVALID_REQUEST" in upper_err
+        if not is_invalid_request and not is_kicked:
             self._session_invalid_strikes = 0
             return
         self._session_invalid_strikes += 1
-        if self._session_invalid_strikes < 2 or self._session_invalid_fired:
+        threshold = 1 if is_kicked else 2
+        if self._session_invalid_strikes < threshold or self._session_invalid_fired:
             return
         self._session_invalid_fired = True
-        reason = "IM 会话失效(INVALID_REQUEST)，登录可能已退出"
+        if is_kicked:
+            reason = "抖音安全网关已踢下线(decision=KICK)"
+            failure_detail = "发送接口返回 decision=KICK"
+        else:
+            reason = "IM 会话失效(INVALID_REQUEST)，登录可能已退出"
+            failure_detail = f"连续 {self._session_invalid_strikes} 次发送返回 INVALID_REQUEST"
         logger.warning(
-            f"Account {self.account_id} {reason}；连续 {self._session_invalid_strikes} 次 -> 自动下线"
+            f"Account {self.account_id} {reason} -> 自动下线"
         )
         system_logger.record(
             "IM 登录失效，自动下线",
-            detail=f"{reason}（连续 {self._session_invalid_strikes} 次发送返回 INVALID_REQUEST）。"
+            detail=f"{reason}（{failure_detail}）。"
             "请停止托管后用浏览器模式重新登录并打开私信页，再重新启动托管。",
             level="error",
             category="auth",

backend/rpa_engine/douyin_im/session.py

@@ -40,6 +40,11 @@ class DouyinImSession:
     # 方案 A：直接复用浏览器抓到的真实 frontier 连接凭证（绕开我们自己推导 token/access_key 不准的问题）
     sdk_cert: str = ""          # bd-ticket-guard 客户端证书（frontier sdk_cert / HTTP client-cert）
     frontier_ts_sign: str = ""  # 抓包得到的新鲜 ts_sign（覆盖 web_protect 里可能已过期的）
+    # 账号级公网出口配置来自 accounts 表，不写回 im_session_data，避免网络配置
+    # 与登录凭证重复存储。egress_source_ip 是当前服务器探测出的本地绑定地址。
+    egress_public_ip: str = ""
+    egress_source_ip: str = ""
+    egress_auto_attempts: int = 1
 
     @classmethod
     def from_storage_state(cls, data: dict, extra: Optional[dict] = None) -> "DouyinImSession":

backend/rpa_engine/douyin_im/ws_client.py

@@ -343,6 +343,8 @@ class DouyinImWsClient:
         loop = asyncio.get_running_loop()
         connected_at: float | None = None
         connection: Optional[WebSocketClientProtocol] = None
+        source_ip = str(getattr(self.session, "egress_source_ip", "") or "").strip()
+        connect_kwargs = {"local_addr": (source_ip, 0)} if source_ip else {}
         try:
             async with websocket_connect(
                 url,
@@ -363,6 +365,7 @@ class DouyinImWsClient:
                 # receive memory genuinely bounded across hundreds of peers.
                 max_size=_INCOMING_MAX_SIZE,
                 max_queue=_TRANSPORT_MAX_QUEUE,
+                **connect_kwargs,
             ) as websocket:
                 connection = websocket
                 self._connection = websocket

backend/rpa_engine/egress_channels.py

@@ -0,0 +1,339 @@
+"""Discover and select server egress channels for account-bound IM traffic.
+
+One public address may be reached through a private address on the host (for
+example, an ECS secondary private IP mapped to an EIP).  A channel therefore
+keeps both values: ``source_ip`` is bound on the socket and ``public_ip`` is
+what the remote service observes.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import ipaddress
+import json
+import logging
+import os
+import socket
+import subprocess
+import threading
+import time
+from dataclasses import dataclass
+from typing import Iterable
+
+import httpx
+import requests
+from requests.adapters import HTTPAdapter
+
+logger = logging.getLogger("rpa_engine.egress")
+
+_DISCOVERY_TTL_SECONDS = 300.0
+_PROBE_TIMEOUT_SECONDS = 6.0
+_MAX_CHANNEL_ATTEMPTS = 8
+_PROBE_URLS = (
+    "https://www.cloudflare.com/cdn-cgi/trace",
+    "https://api64.ipify.org?format=json",
+)
+
+
+@dataclass(frozen=True)
+class LocalAddress:
+    source_ip: str | None
+    interface: str
+    is_default: bool = False
+
+
+@dataclass(frozen=True)
+class EgressChannel:
+    public_ip: str
+    source_ip: str | None
+    interface: str = ""
+    is_default: bool = False
+
+    @property
+    def id(self) -> str:
+        return self.public_ip
+
+
+@dataclass(frozen=True)
+class EgressSnapshot:
+    channels: tuple[EgressChannel, ...]
+    errors: tuple[str, ...]
+    detected_at: float
+
+
+class EgressChannelUnavailable(RuntimeError):
+    pass
+
+
+_cache_lock = threading.Lock()
+_cached_snapshot: EgressSnapshot | None = None
+
+
+def clamp_attempts(value: int | None) -> int:
+    try:
+        parsed = int(value or 1)
+    except (TypeError, ValueError):
+        parsed = 1
+    return max(1, min(_MAX_CHANNEL_ATTEMPTS, parsed))
+
+
+def _usable_source_ip(value: str) -> bool:
+    try:
+        addr = ipaddress.ip_address(str(value or "").strip())
+    except ValueError:
+        return False
+    return bool(
+        addr.version == 4
+        and not addr.is_loopback
+        and not addr.is_link_local
+        and not addr.is_multicast
+        and not addr.is_unspecified
+    )
+
+
+def _linux_local_addresses() -> list[LocalAddress]:
+    if os.name != "posix":
+        return []
+    try:
+        proc = subprocess.run(
+            ["ip", "-j", "-4", "addr", "show", "scope", "global"],
+            capture_output=True,
+            text=True,
+            timeout=3,
+            check=False,
+        )
+        payload = json.loads(proc.stdout or "[]") if proc.returncode == 0 else []
+    except (OSError, subprocess.SubprocessError, json.JSONDecodeError):
+        return []
+
+    found: list[LocalAddress] = []
+    for item in payload if isinstance(payload, list) else []:
+        interface = str(item.get("ifname") or "")
+        for info in item.get("addr_info") or []:
+            source_ip = str(info.get("local") or "").strip()
+            if _usable_source_ip(source_ip):
+                found.append(LocalAddress(source_ip, interface))
+    return found
+
+
+def _socket_local_addresses() -> list[LocalAddress]:
+    found: list[LocalAddress] = []
+    names = {socket.gethostname(), socket.getfqdn()}
+    for name in names:
+        try:
+            records = socket.getaddrinfo(name, None, socket.AF_INET, socket.SOCK_STREAM)
+        except OSError:
+            continue
+        for record in records:
+            source_ip = str(record[4][0] or "").strip()
+            if _usable_source_ip(source_ip):
+                found.append(LocalAddress(source_ip, name))
+    return found
+
+
+def local_address_candidates() -> list[LocalAddress]:
+    """Return the default route plus each bindable global/private IPv4."""
+
+    candidates = [LocalAddress(None, "default", True)]
+    seen: set[str] = set()
+    for item in [*_linux_local_addresses(), *_socket_local_addresses()]:
+        source_ip = str(item.source_ip or "")
+        if not source_ip or source_ip in seen:
+            continue
+        seen.add(source_ip)
+        candidates.append(item)
+    return candidates
+
+
+def _extract_public_ip(response: httpx.Response) -> str:
+    text = response.text.strip()
+    content_type = response.headers.get("content-type", "").lower()
+    candidate = ""
+    if "json" in content_type or text.startswith("{"):
+        try:
+            candidate = str(response.json().get("ip") or "").strip()
+        except (ValueError, AttributeError):
+            candidate = ""
+    if not candidate:
+        for line in text.splitlines():
+            if line.startswith("ip="):
+                candidate = line.partition("=")[2].strip()
+                break
+    if not candidate and "\n" not in text and len(text) <= 64:
+        candidate = text
+    try:
+        addr = ipaddress.ip_address(candidate)
+    except ValueError:
+        return ""
+    return str(addr) if addr.version == 4 else ""
+
+
+async def _probe_local_address(candidate: LocalAddress) -> tuple[EgressChannel | None, str]:
+    transport = httpx.AsyncHTTPTransport(
+        local_address=candidate.source_ip,
+        retries=0,
+    )
+    last_error = ""
+    try:
+        async with httpx.AsyncClient(
+            transport=transport,
+            timeout=httpx.Timeout(_PROBE_TIMEOUT_SECONDS),
+            follow_redirects=True,
+            trust_env=False,
+        ) as client:
+            for url in _PROBE_URLS:
+                try:
+                    response = await client.get(url, headers={"Accept": "text/plain, application/json"})
+                    response.raise_for_status()
+                    public_ip = _extract_public_ip(response)
+                    if public_ip:
+                        return (
+                            EgressChannel(
+                                public_ip=public_ip,
+                                source_ip=candidate.source_ip,
+                                interface=candidate.interface,
+                                is_default=candidate.is_default,
+                            ),
+                            "",
+                        )
+                    last_error = "探测响应中没有 IPv4"
+                except Exception as exc:  # one endpoint may be unavailable
+                    last_error = str(exc) or type(exc).__name__
+    finally:
+        await transport.aclose()
+    label = candidate.source_ip or "默认路由"
+    return None, f"{label}: {last_error or '无法访问公网探测服务'}"
+
+
+def _dedupe_channels(channels: Iterable[EgressChannel]) -> tuple[EgressChannel, ...]:
+    by_public_ip: dict[str, EgressChannel] = {}
+    order: list[str] = []
+    for channel in channels:
+        existing = by_public_ip.get(channel.public_ip)
+        if existing is None:
+            by_public_ip[channel.public_ip] = channel
+            order.append(channel.public_ip)
+            continue
+        # Keep an explicit bindable source when possible, while preserving the
+        # fact that this is also the host's default public route.
+        if existing.source_ip is None and channel.source_ip:
+            by_public_ip[channel.public_ip] = EgressChannel(
+                public_ip=channel.public_ip,
+                source_ip=channel.source_ip,
+                interface=channel.interface,
+                is_default=existing.is_default or channel.is_default,
+            )
+        elif channel.is_default and not existing.is_default:
+            by_public_ip[channel.public_ip] = EgressChannel(
+                public_ip=existing.public_ip,
+                source_ip=existing.source_ip,
+                interface=existing.interface,
+                is_default=True,
+            )
+    return tuple(by_public_ip[key] for key in order)
+
+
+async def discover_egress_channels(*, force: bool = False) -> EgressSnapshot:
+    global _cached_snapshot
+
+    now = time.time()
+    with _cache_lock:
+        cached = _cached_snapshot
+    if not force and cached and now - cached.detected_at < _DISCOVERY_TTL_SECONDS:
+        return cached
+
+    candidates = await asyncio.to_thread(local_address_candidates)
+    results = await asyncio.gather(*(_probe_local_address(item) for item in candidates))
+    channels = _dedupe_channels(item[0] for item in results if item[0] is not None)
+    errors = tuple(item[1] for item in results if item[1])
+    snapshot = EgressSnapshot(channels=channels, errors=errors, detected_at=time.time())
+    with _cache_lock:
+        _cached_snapshot = snapshot
+    return snapshot
+
+
+async def resolve_fixed_channel(public_ip: str) -> EgressChannel:
+    selected = str(public_ip or "").strip()
+    if not selected:
+        return EgressChannel(public_ip="", source_ip=None, interface="default", is_default=True)
+    snapshot = await discover_egress_channels()
+    for channel in snapshot.channels:
+        if channel.public_ip == selected:
+            return channel
+    raise EgressChannelUnavailable(
+        f"指定公网通道 {selected} 当前不可用；请在账号编辑中重新检测并选择可用通道"
+    )
+
+
+async def resolve_send_channels(
+    preferred_public_ip: str = "",
+    max_attempts: int = 1,
+) -> list[EgressChannel]:
+    """Order channels for one serial send operation.
+
+    The ordinary one-channel automatic mode deliberately avoids discovery so
+    a temporary outage of the probe service never blocks existing sends.
+    """
+
+    preferred = str(preferred_public_ip or "").strip()
+    attempts = clamp_attempts(max_attempts)
+    if not preferred and attempts == 1:
+        return [EgressChannel(public_ip="", source_ip=None, interface="default", is_default=True)]
+
+    snapshot = await discover_egress_channels()
+    channels = list(snapshot.channels)
+    if not channels:
+        if preferred:
+            raise EgressChannelUnavailable(
+                f"指定公网通道 {preferred} 无法探测；请检查服务器网卡、路由或公网访问"
+            )
+        return [EgressChannel(public_ip="", source_ip=None, interface="default", is_default=True)]
+
+    ordered: list[EgressChannel] = []
+    if preferred:
+        selected = next((item for item in channels if item.public_ip == preferred), None)
+        if selected is None:
+            raise EgressChannelUnavailable(
+                f"指定公网通道 {preferred} 当前不可用；请在账号编辑中重新检测"
+            )
+        ordered.append(selected)
+    else:
+        default = next((item for item in channels if item.is_default), None)
+        if default is not None:
+            ordered.append(default)
+
+    ordered.extend(item for item in channels if item not in ordered)
+    return ordered[:attempts]
+
+
+class _SourceAddressAdapter(HTTPAdapter):
+    """Requests adapter that binds outgoing sockets to one local IPv4."""
+
+    def __init__(self, source_ip: str, *args, **kwargs):
+        self._source_address = (source_ip, 0)
+        super().__init__(*args, **kwargs)
+
+    def init_poolmanager(self, connections, maxsize, block=False, **pool_kwargs):
+        pool_kwargs["source_address"] = self._source_address
+        return super().init_poolmanager(connections, maxsize, block=block, **pool_kwargs)
+
+    def proxy_manager_for(self, proxy, **proxy_kwargs):
+        proxy_kwargs["source_address"] = self._source_address
+        return super().proxy_manager_for(proxy, **proxy_kwargs)
+
+
+def source_bound_requests_session(source_ip: str | None = None) -> requests.Session:
+    client = requests.Session()
+    source = str(source_ip or "").strip()
+    if source:
+        client.trust_env = False
+        adapter = _SourceAddressAdapter(source)
+        client.mount("http://", adapter)
+        client.mount("https://", adapter)
+    return client
+
+
+def reset_egress_cache_for_tests() -> None:
+    global _cached_snapshot
+    with _cache_lock:
+        _cached_snapshot = None

backend/rpa_engine/playwright_worker.py

@@ -33,6 +33,7 @@ from rpa_engine.runtime_config import (
     ensure_browser_display,
     playwright_proxy,
 )
+from rpa_engine.egress_channels import clamp_attempts, resolve_fixed_channel
 
 logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
 logger = logging.getLogger("rpa_engine")
@@ -909,6 +910,42 @@ class DouyinWorker:
 
     async def _run_im_direct_service(self, session: DouyinImSession):
         """运行 IM API + WebSocket 直连自动回复"""
+        # 公网通道配置独立存于账号表。固定通道在启动时解析一次供 WS 使用；
+        # HTTP 发送仍会在每次建连时校验，账号编辑后的配置无需重启即可生效。
+        row = None
+        db = await self.get_db()
+        try:
+            try:
+                row = (
+                    await db.execute(
+                        select(
+                            Account.egress_public_ip,
+                            Account.egress_auto_attempts,
+                        ).where(Account.id == self.account_id)
+                    )
+                ).one_or_none()
+            except Exception as exc:
+                # A worker may be created by an isolated test or during a
+                # rolling deployment before the startup migration finishes.
+                logger.debug("load account egress config failed: %s", exc)
+        finally:
+            await db.close()
+        session.egress_public_ip = str((row.egress_public_ip if row else "") or "").strip()
+        session.egress_auto_attempts = clamp_attempts(
+            row.egress_auto_attempts if row else 1
+        )
+        session.egress_source_ip = ""
+        if session.egress_public_ip:
+            try:
+                route = await resolve_fixed_channel(session.egress_public_ip)
+                session.egress_source_ip = str(route.source_ip or "")
+            except Exception as exc:
+                logger.warning(
+                    "Account %s selected egress %s is not currently resolvable: %s",
+                    self.account_id,
+                    session.egress_public_ip,
+                    exc,
+                )
         # Cache the only account fields needed by the follow-welcome timer.
         # Disabled accounts subsequently avoid the old full Account query on
         # every minute tick.
@@ -925,7 +962,7 @@ class DouyinWorker:
             reply_delay_resolver=self.resolve_reply_delay_seconds,
             # 关注欢迎语：周期性检测新粉丝并自动私信（约每 60s）
             follow_tick=self.follow_welcome_tick,
-            # IM 登录失效（INVALID_REQUEST）时自动下线
+            # IM 登录失效（INVALID_REQUEST/KICK）时自动下线
             on_session_invalid=self.on_im_session_invalid,
             # Batch admission waits for UID/frontier/WS/first-poll completion;
             # it no longer releases its slot immediately after create_task().