diff --git a/backend/kefu.db-shm b/backend/kefu.db-shm index 405c17a..0ca214a 100644 Binary files a/backend/kefu.db-shm and b/backend/kefu.db-shm differ diff --git a/backend/kefu.db-wal b/backend/kefu.db-wal index 22c0ec3..92b8bbc 100644 Binary files a/backend/kefu.db-wal and b/backend/kefu.db-wal differ diff --git a/backend/main.py b/backend/main.py index 7d7bfe4..e35b1ee 100644 --- a/backend/main.py +++ b/backend/main.py @@ -2,6 +2,7 @@ import os import sys import json import asyncio +import ipaddress import logging import uuid from datetime import datetime, timezone @@ -93,6 +94,10 @@ from utils.cookie_store import ( analyze_cookie, ) from rpa_engine.device_profiles import list_device_profiles, profile_label_for_ua, resolve_user_agent +from rpa_engine.egress_channels import ( + clamp_attempts, + discover_egress_channels, +) logger = logging.getLogger("main") from utils import system_logger @@ -443,6 +448,8 @@ def _build_account_im_session(account: Account) -> DouyinImSession: storage = json.loads(cookie_data) if cookie_data else {} session = build_im_session_from_storage(storage, account.im_session_data) session.user_agent = resolve_user_agent(account.user_agent or session.user_agent) + session.egress_public_ip = str(account.egress_public_ip or "").strip() + session.egress_auto_attempts = clamp_attempts(account.egress_auto_attempts) return session @@ -806,6 +813,8 @@ class AccountResponse(BaseModel): follow_welcome_content: Optional[str] = None user_agent: Optional[str] = None user_agent_label: Optional[str] = None + egress_public_ip: Optional[str] = None + egress_auto_attempts: int = 1 quota_disabled: bool = False class Config: @@ -850,6 +859,10 @@ class AccountUpdate(BaseModel): follow_welcome_enabled: Optional[bool] = None follow_welcome_content: Optional[str] = None user_agent: Optional[str] = None + # 空字符串/null=自动选择；否则保存服务器探测到的公网 IPv4。 + egress_public_ip: Optional[str] = None + # 包含首选通道在内的最大串行尝试数，范围 1~8。 + egress_auto_attempts: Optional[int] = None class ReplyQueueItemResponse(BaseModel): @@ -1074,6 +1087,8 @@ def _build_account_response(account: Account) -> AccountResponse: follow_welcome_content=account.follow_welcome_content or None, user_agent=account.user_agent or None, user_agent_label=profile_label_for_ua(account.user_agent), + egress_public_ip=account.egress_public_ip or None, + egress_auto_attempts=clamp_attempts(account.egress_auto_attempts), quota_disabled=bool(account.quota_disabled), ) @@ -1203,12 +1218,54 @@ class DeviceProfileItem(BaseModel): user_agent: str +class EgressChannelItem(BaseModel): + id: str + public_ip: str + source_ip: Optional[str] = None + interface: str = "" + is_default: bool = False + + +class EgressChannelListResponse(BaseModel): + channels: List[EgressChannelItem] = Field(default_factory=list) + multiple: bool = False + detected_at: datetime + errors: List[str] = Field(default_factory=list) + + @app.get("/api/device-profiles", response_model=List[DeviceProfileItem]) async def get_device_profiles(user: User = Depends(get_current_user)): """可选的伪装设备头（User-Agent）预设列表。""" return list_device_profiles() +@app.get("/api/network/egress-channels", response_model=EgressChannelListResponse) +async def get_egress_channels( + refresh: bool = False, + user: User = Depends(require_accounts_update), +): + """Detect bindable server addresses and the public IPv4 seen through each.""" + + del user + snapshot = await discover_egress_channels(force=refresh) + channels = [ + EgressChannelItem( + id=item.id, + public_ip=item.public_ip, + source_ip=item.source_ip, + interface=item.interface, + is_default=item.is_default, + ) + for item in snapshot.channels + ] + return EgressChannelListResponse( + channels=channels, + multiple=len(channels) > 1, + detected_at=datetime.fromtimestamp(snapshot.detected_at, tz=timezone.utc), + errors=list(snapshot.errors), + ) + + # --- API 路由接口 --- # 1. 账号管理接口 @@ -1663,6 +1720,18 @@ async def update_account( if body.user_agent is not None: ua = (body.user_agent or "").strip() account.user_agent = ua or None + if "egress_public_ip" in body.model_fields_set: + selected_public_ip = str(body.egress_public_ip or "").strip() + if selected_public_ip: + try: + parsed_ip = ipaddress.ip_address(selected_public_ip) + except ValueError: + raise HTTPException(status_code=400, detail="公网通道必须是有效的 IPv4 地址") + if parsed_ip.version != 4: + raise HTTPException(status_code=400, detail="公网通道目前仅支持 IPv4") + account.egress_public_ip = selected_public_ip or None + if body.egress_auto_attempts is not None: + account.egress_auto_attempts = clamp_attempts(body.egress_auto_attempts) account.updated_at = datetime.utcnow() await db.commit() await db.refresh(account) @@ -1671,6 +1740,13 @@ async def update_account( invalidate = getattr(worker, "invalidate_follow_welcome_config", None) if callable(invalidate): invalidate() + worker = manager.workers.get(account_id) + runtime_service = getattr(worker, "_im_service", None) if worker else None + if runtime_service: + runtime_session = runtime_service.session + runtime_session.egress_public_ip = str(account.egress_public_ip or "").strip() + runtime_session.egress_source_ip = "" + runtime_session.egress_auto_attempts = clamp_attempts(account.egress_auto_attempts) return _build_account_response(account) @@ -2912,6 +2988,9 @@ async def send_account_message( db.add(failed_log) await db.commit() + normalized_last_error = (last_error or "").upper() + session_kicked = "DECISION=KICK" in normalized_last_error + invalid_request = "INVALID_REQUEST" in normalized_last_error need_browser = ( not session.keys_str or not session.web_protect_str @@ -2919,11 +2998,14 @@ async def send_account_message( or "ticket" in (last_error or "") or "签名密钥" in (last_error or "") or "web_protect" in (last_error or "") - or last_error == "INVALID_REQUEST" + or invalid_request + or session_kicked ) if need_browser: msg = last_error or "缺少 IM 签名密钥" - if last_error == "INVALID_REQUEST": + if session_kicked: + msg = "抖音已踢下当前 IM 登录态（decision=KICK），请停止托管后用浏览器模式重新登录并打开私信页" + elif invalid_request: msg = "IM 会话创建失败(INVALID_REQUEST)，请停止托管后用浏览器模式重新登录并打开私信页" return SendMessageResponse( success=False, diff --git a/backend/models/db_migrate.py b/backend/models/db_migrate.py index 425eef8..e090640 100644 --- a/backend/models/db_migrate.py +++ b/backend/models/db_migrate.py @@ -114,6 +114,18 @@ def migrate_accounts_table(conn) -> None: "user_agent", {"default": "ALTER TABLE accounts ADD COLUMN user_agent TEXT"}, ) + add_column_if_missing( + conn, + "accounts", + "egress_public_ip", + {"default": "ALTER TABLE accounts ADD COLUMN egress_public_ip VARCHAR(64)"}, + ) + add_column_if_missing( + conn, + "accounts", + "egress_auto_attempts", + {"default": "ALTER TABLE accounts ADD COLUMN egress_auto_attempts INTEGER DEFAULT 1"}, + ) add_column_if_missing( conn, "accounts", diff --git a/backend/models/models.py b/backend/models/models.py index 75a9a59..4e73a16 100644 --- a/backend/models/models.py +++ b/backend/models/models.py @@ -116,6 +116,8 @@ class Account(Base): follow_welcome_enabled = Column(Boolean, default=False) # 新粉丝关注后自动发送欢迎语 follow_welcome_content = Column(Text, nullable=True) # 关注欢迎语内容（空=不发） user_agent = Column(Text, nullable=True) # 伪装设备头（User-Agent），空=默认 + egress_public_ip = Column(String(64), nullable=True) # 指定公网出口；空=自动选择 + egress_auto_attempts = Column(Integer, nullable=False, default=1) # 发送失败时最多串行尝试的出口数 qr_code_base64 = Column(Text, nullable=True) # 当前登录二维码的 base64 字符串 error_message = Column(Text, nullable=True) # 错误信息 quota_disabled = Column(Boolean, default=False, index=True) # 额度不足被停用 diff --git a/backend/rpa_engine/douyin_im/auth.py b/backend/rpa_engine/douyin_im/auth.py index 8e42b5f..659c031 100644 --- a/backend/rpa_engine/douyin_im/auth.py +++ b/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']) diff --git a/backend/rpa_engine/douyin_im/http_client.py b/backend/rpa_engine/douyin_im/http_client.py index 178a027..5db2576 100644 --- a/backend/rpa_engine/douyin_im/http_client.py +++ b/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}" diff --git a/backend/rpa_engine/douyin_im/image_upload.py b/backend/rpa_engine/douyin_im/image_upload.py index cb33c0d..d935086 100644 --- a/backend/rpa_engine/douyin_im/image_upload.py +++ b/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"): diff --git a/backend/rpa_engine/douyin_im/pb_decode.py b/backend/rpa_engine/douyin_im/pb_decode.py index 136258c..47d451d 100644 --- a/backend/rpa_engine/douyin_im/pb_decode.py +++ b/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: diff --git a/backend/rpa_engine/douyin_im/peer_profile.py b/backend/rpa_engine/douyin_im/peer_profile.py index a2e9e8a..ae4ad87 100644 --- a/backend/rpa_engine/douyin_im/peer_profile.py +++ b/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, ) diff --git a/backend/rpa_engine/douyin_im/service.py b/backend/rpa_engine/douyin_im/service.py index 4a4a4f7..051024d 100644 --- a/backend/rpa_engine/douyin_im/service.py +++ b/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", diff --git a/backend/rpa_engine/douyin_im/session.py b/backend/rpa_engine/douyin_im/session.py index 72e3213..5614fee 100644 --- a/backend/rpa_engine/douyin_im/session.py +++ b/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": diff --git a/backend/rpa_engine/douyin_im/ws_client.py b/backend/rpa_engine/douyin_im/ws_client.py index a28183c..7a71e6a 100644 --- a/backend/rpa_engine/douyin_im/ws_client.py +++ b/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 diff --git a/backend/rpa_engine/egress_channels.py b/backend/rpa_engine/egress_channels.py new file mode 100644 index 0000000..372cbd5 --- /dev/null +++ b/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 diff --git a/backend/rpa_engine/playwright_worker.py b/backend/rpa_engine/playwright_worker.py index d2d0b76..d52db98 100644 --- a/backend/rpa_engine/playwright_worker.py +++ b/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(). diff --git a/backend/tests/test_egress_channels.py b/backend/tests/test_egress_channels.py new file mode 100644 index 0000000..e810348 --- /dev/null +++ b/backend/tests/test_egress_channels.py @@ -0,0 +1,113 @@ +from __future__ import annotations + +import os +import sys +import time +import unittest +from pathlib import Path +from unittest.mock import AsyncMock, patch + +from sqlalchemy import create_engine, inspect, text + + +BACKEND_DIR = Path(__file__).resolve().parents[1] +if str(BACKEND_DIR) not in sys.path: + sys.path.insert(0, str(BACKEND_DIR)) + +from rpa_engine.egress_channels import ( + EgressChannel, + EgressChannelUnavailable, + EgressSnapshot, + LocalAddress, + discover_egress_channels, + reset_egress_cache_for_tests, + resolve_send_channels, +) +from models.db_migrate import migrate_accounts_table + + +class EgressChannelTests(unittest.IsolatedAsyncioTestCase): + def setUp(self): + reset_egress_cache_for_tests() + + async def test_discovery_deduplicates_public_ip_and_keeps_bindable_source(self): + candidates = [ + LocalAddress(None, "default", True), + LocalAddress("10.0.0.5", "eth0"), + LocalAddress("10.0.0.6", "eth0:1"), + ] + + async def probe(candidate): + public_ip = "203.0.113.10" if candidate.source_ip != "10.0.0.6" else "203.0.113.11" + return ( + EgressChannel( + public_ip=public_ip, + source_ip=candidate.source_ip, + interface=candidate.interface, + is_default=candidate.is_default, + ), + "", + ) + + with ( + patch( + "rpa_engine.egress_channels.local_address_candidates", + return_value=candidates, + ), + patch( + "rpa_engine.egress_channels._probe_local_address", + AsyncMock(side_effect=probe), + ), + ): + snapshot = await discover_egress_channels(force=True) + + self.assertEqual([item.public_ip for item in snapshot.channels], ["203.0.113.10", "203.0.113.11"]) + self.assertEqual(snapshot.channels[0].source_ip, "10.0.0.5") + self.assertTrue(snapshot.channels[0].is_default) + + async def test_selected_channel_is_first_and_attempt_count_is_bounded(self): + snapshot = EgressSnapshot( + channels=( + EgressChannel("198.51.100.1", "10.0.0.1", "eth0", True), + EgressChannel("198.51.100.2", "10.0.0.2", "eth0:1"), + EgressChannel("198.51.100.3", "10.0.0.3", "eth0:2"), + ), + errors=(), + detected_at=time.time(), + ) + with patch( + "rpa_engine.egress_channels.discover_egress_channels", + AsyncMock(return_value=snapshot), + ): + routes = await resolve_send_channels("198.51.100.2", 2) + + self.assertEqual([item.public_ip for item in routes], ["198.51.100.2", "198.51.100.1"]) + + async def test_missing_selected_channel_fails_closed(self): + snapshot = EgressSnapshot( + channels=(EgressChannel("198.51.100.1", None, "default", True),), + errors=(), + detected_at=time.time(), + ) + with patch( + "rpa_engine.egress_channels.discover_egress_channels", + AsyncMock(return_value=snapshot), + ): + with self.assertRaises(EgressChannelUnavailable): + await resolve_send_channels("198.51.100.99", 2) + + +class EgressMigrationTests(unittest.TestCase): + def test_old_accounts_table_receives_egress_columns(self): + engine = create_engine("sqlite:///:memory:") + with engine.begin() as connection: + connection.execute(text("CREATE TABLE accounts (id INTEGER PRIMARY KEY)")) + migrate_accounts_table(connection) + columns = {item["name"] for item in inspect(connection).get_columns("accounts")} + + self.assertIn("egress_public_ip", columns) + self.assertIn("egress_auto_attempts", columns) + + +if __name__ == "__main__": + unittest.main() diff --git a/backend/tests/test_im_pb_decode.py b/backend/tests/test_im_pb_decode.py new file mode 100644 index 0000000..484fe9f --- /dev/null +++ b/backend/tests/test_im_pb_decode.py @@ -0,0 +1,32 @@ +from __future__ import annotations + +import sys +import unittest +from pathlib import Path + + +BACKEND_DIR = Path(__file__).resolve().parents[1] +if str(BACKEND_DIR) not in sys.path: + sys.path.insert(0, str(BACKEND_DIR)) + +from rpa_engine.douyin_im.pb_decode import analyze_send_response + + +class AnalyzeSendResponseTests(unittest.TestCase): + def test_standalone_kick_json_is_not_decoded_as_protobuf(self): + result = analyze_send_response(b'{"decision": "KICK"}') + + self.assertFalse(result["ok"]) + self.assertEqual(result["decision"], "KICK") + self.assertEqual(result["summary"], "JSON响应 decision=KICK") + self.assertNotIn("unsupported wire type", result["summary"]) + + def test_malformed_json_still_returns_controlled_decode_summary(self): + result = analyze_send_response(b'{"decision":') + + self.assertFalse(result["ok"]) + self.assertIn("解码失败", result["summary"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/backend/tests/test_reply_queue_integration.py b/backend/tests/test_reply_queue_integration.py index de7bf90..84d631e 100644 --- a/backend/tests/test_reply_queue_integration.py +++ b/backend/tests/test_reply_queue_integration.py @@ -114,6 +114,39 @@ def _build_service(delay_seconds: int = 60): class ReplyQueueIntegrationTests(unittest.IsolatedAsyncioTestCase): + async def test_kick_response_takes_account_offline_immediately(self): + callback = AsyncMock() + service, _, _ = _build_service() + service.on_session_invalid = callback + + with unittest.mock.patch( + "rpa_engine.douyin_im.service.system_logger.record", Mock() + ): + await service._note_session_invalid( + "抖音安全网关返回 decision=KICK，当前登录态已失效" + ) + + self.assertFalse(service._running) + self.assertTrue(service._session_invalid_fired) + callback.assert_awaited_once() + self.assertIn("decision=KICK", callback.await_args.args[0]) + + async def test_invalid_request_still_requires_two_consecutive_failures(self): + callback = AsyncMock() + service, _, _ = _build_service() + service.on_session_invalid = callback + + with unittest.mock.patch( + "rpa_engine.douyin_im.service.system_logger.record", Mock() + ): + await service._note_session_invalid("INVALID_REQUEST") + self.assertTrue(service._running) + callback.assert_not_awaited() + await service._note_session_invalid("INVALID_REQUEST") + + self.assertFalse(service._running) + callback.assert_awaited_once() + async def test_same_message_from_ws_and_poll_is_queued_once(self): service, match_reply, _ = _build_service() message = { diff --git a/backend/tests/test_send_entry_and_worker_lifecycle.py b/backend/tests/test_send_entry_and_worker_lifecycle.py index 79eea78..9af7375 100644 --- a/backend/tests/test_send_entry_and_worker_lifecycle.py +++ b/backend/tests/test_send_entry_and_worker_lifecycle.py @@ -20,6 +20,7 @@ if str(BACKEND_DIR) not in sys.path: from rpa_engine.douyin_im import http_client as http_client_module from rpa_engine.douyin_im.http_client import DouyinImHttpClient from rpa_engine.douyin_im.session import DouyinImSession +from rpa_engine.egress_channels import EgressChannel from rpa_engine.playwright_worker import DouyinWorker @@ -85,6 +86,7 @@ class SendTextMessageEntryTests(unittest.IsolatedAsyncioTestCase): last_send_meta=queued_meta, last_error="credential expired", last_send_needs_refresh=True, + last_send_channel_retryable=False, last_request_debug="response status=401", ) queued_context = MagicMock() @@ -131,6 +133,65 @@ class SendTextMessageEntryTests(unittest.IsolatedAsyncioTestCase): self.assertTrue(client.last_send_needs_refresh) self.assertEqual(client.last_request_debug, "response status=401") + async def test_retryable_rejection_switches_channels_serially(self): + client = self._make_client(account_id=89) + client.session.egress_auto_attempts = 2 + routes = [ + EgressChannel("198.51.100.10", "10.0.0.10", "eth0", True), + EgressChannel("198.51.100.11", "10.0.0.11", "eth0:1", False), + ] + + first = SimpleNamespace( + send_text_message=AsyncMock(return_value=False), + last_send_meta={}, + last_error="decision=KICK", + last_send_needs_refresh=False, + last_send_channel_retryable=True, + last_request_debug="first route", + ) + second = SimpleNamespace( + send_text_message=AsyncMock(return_value=True), + last_send_meta={"conv": {"ticket": "ok"}}, + last_error="", + last_send_needs_refresh=False, + last_send_channel_retryable=False, + last_request_debug="second route", + ) + + def context_for(value): + context = MagicMock() + context.__aenter__ = AsyncMock(return_value=value) + context.__aexit__ = AsyncMock(return_value=None) + return context + + queued_factory = MagicMock(side_effect=[context_for(first), context_for(second)]) + + async def execute_submission(account_id, operation, description=""): + self.assertEqual(account_id, 89) + return await operation() + + with ( + patch( + "rpa_engine.douyin_im.traffic_control.submit_outbound", + AsyncMock(side_effect=execute_submission), + ), + patch( + "rpa_engine.douyin_im.http_client.resolve_send_channels", + AsyncMock(return_value=routes), + ), + patch.object(http_client_module, "DouyinImHttpClient", queued_factory), + patch.object(http_client_module.system_logger, "record"), + ): + sent = await client.send_text_message("0:1:10001:20002", "hello") + + self.assertTrue(sent) + self.assertEqual(queued_factory.call_count, 2) + self.assertEqual(queued_factory.call_args_list[0].kwargs["source_ip"], "10.0.0.10") + self.assertEqual(queued_factory.call_args_list[1].kwargs["source_ip"], "10.0.0.11") + first.send_text_message.assert_awaited_once() + second.send_text_message.assert_awaited_once() + self.assertEqual(client.last_request_debug, "second route") + class WorkerLifecycleTests(unittest.IsolatedAsyncioTestCase): async def test_start_saves_task_and_stop_waits_until_it_is_done(self): diff --git a/frontend/src/views/Accounts.vue b/frontend/src/views/Accounts.vue index c8c0bac..e9d298a 100644 --- a/frontend/src/views/Accounts.vue +++ b/frontend/src/views/Accounts.vue @@ -68,9 +68,13 @@ const accountPageSize = ref(9) const accountTotal = ref(0) const rules = ref([]) const deviceProfiles = ref([]) +const egressChannels = ref([]) +const egressChannelsLoading = ref(false) +const egressChannelsError = ref('') const CUSTOM_UA_PROFILE = '__custom__' const loading = ref(false) const batchStarting = ref(false) +const batchDeleting = ref(false) const activeStartBatchId = ref(null) let batchStatusTimer = null let batchStatusRequestActive = false @@ -362,6 +366,8 @@ const editForm = ref({ follow_welcome_content: '', user_agent_profile: 'chrome_win120', user_agent_custom: '', + egress_public_ip: '', + egress_auto_attempts: 1, }) const profileSelectOptions = computed(() => { @@ -373,6 +379,25 @@ const profileSelectOptions = computed(() => { return opts }) +const egressChannelOptions = computed(() => { + const options = [ + { value: '', label: '自动选择（服务器默认公网出口）' } + ] + for (const channel of egressChannels.value || []) { + const source = channel.source_ip ? `本地 ${channel.source_ip}` : '默认路由' + const suffix = channel.is_default ? ' · 当前默认' : '' + options.push({ + value: channel.public_ip, + label: `${channel.public_ip}（${source}${suffix}）` + }) + } + const selected = (editForm.value.egress_public_ip || '').trim() + if (selected && !options.some((item) => item.value === selected)) { + options.push({ value: selected, label: `${selected}（当前未检测到）`, disabled: true }) + } + return options +}) + const accountQuota = computed(() => { const user = auth.user const count = accountTotal.value @@ -446,6 +471,18 @@ const selectedStartableCount = computed(() => startableAccounts.value.filter((a) => selectedIds.value.includes(a.id)).length ) +const selectableAccounts = computed(() => { + if (auth.canDeleteAccounts) return accounts.value + if (auth.canStartAccounts) return startableAccounts.value + return [] +}) + +const selectedAccounts = computed(() => + selectableAccounts.value.filter((a) => selectedIds.value.includes(a.id)) +) + +const selectedAccountCount = computed(() => selectedAccounts.value.length) + const isAccountSelected = (id) => selectedIds.value.includes(id) const toggleAccountSelect = (id) => { @@ -456,8 +493,8 @@ const toggleAccountSelect = (id) => { } } -const selectAllStartable = () => { - selectedIds.value = startableAccounts.value.map((a) => a.id) +const selectAllAccounts = () => { + selectedIds.value = selectableAccounts.value.map((a) => a.id) } const clearSelection = () => { @@ -465,7 +502,7 @@ const clearSelection = () => { } const onSelectAllChange = (e) => { - if (e.target.checked) selectAllStartable() + if (e.target.checked) selectAllAccounts() else clearSelection() } @@ -547,6 +584,26 @@ const fetchDeviceProfiles = async () => { } } +const fetchEgressChannels = async (refresh = false) => { + if (!auth.canUpdateAccounts || egressChannelsLoading.value) return + egressChannelsLoading.value = true + egressChannelsError.value = '' + try { + const res = await api.get('/network/egress-channels', { + params: { refresh }, + timeout: 20000 + }) + egressChannels.value = res.data?.channels || [] + if (!egressChannels.value.length) { + egressChannelsError.value = '未探测到可用公网出口，将继续使用服务器默认路由' + } + } catch (error) { + egressChannelsError.value = error.response?.data?.detail || '公网通道检测失败' + } finally { + egressChannelsLoading.value = false + } +} + const fetchAccounts = async () => { try { loading.value = true @@ -560,6 +617,8 @@ const fetchAccounts = async () => { }) accounts.value = res.data.items || [] accountTotal.value = res.data.total || 0 + const visibleIds = new Set(selectableAccounts.value.map((a) => a.id)) + selectedIds.value = selectedIds.value.filter((id) => visibleIds.has(id)) // 删除/筛选后当前页可能超界，自动回退到最后一页 const maxPage = Math.max(1, Math.ceil(accountTotal.value / accountPageSize.value) || 1) if (accountPage.value > maxPage) { @@ -1063,10 +1122,74 @@ const handleAddAccount = async () => { const handleDeleteAccount = async (id) => { try { await api.delete(`/accounts/${id}`) + selectedIds.value = selectedIds.value.filter((item) => item !== id) message.success('删除成功') - fetchAccounts() + await Promise.all([fetchAccounts(), auth.fetchMe()]) } catch (error) { - message.error('删除账号失败') + message.error(error.response?.data?.detail || '删除账号失败') + } +} + +const handleBatchDeleteAccounts = async () => { + if (batchDeleting.value || batchStarting.value) return + const targets = selectedAccounts.value.map((account) => account.id) + if (!targets.length) { + message.warning('请先勾选要删除的账号') + return + } + + batchDeleting.value = true + const deletedIds = [] + const failures = [] + message.loading({ + content: `正在删除 0/${targets.length} 个账号...`, + key: 'batch_delete', + duration: 0 + }) + try { + // SQLite 下并发执行多个删除事务容易互相抢锁，逐个删除更稳定。 + for (let index = 0; index < targets.length; index += 1) { + const id = targets[index] + try { + await api.delete(`/accounts/${id}`) + deletedIds.push(id) + } catch (error) { + failures.push({ + id, + reason: error.response?.data?.detail || error.message || '删除失败' + }) + } + message.loading({ + content: `正在删除 ${index + 1}/${targets.length} 个账号...`, + key: 'batch_delete', + duration: 0 + }) + } + + const deletedSet = new Set(deletedIds) + selectedIds.value = selectedIds.value.filter((id) => !deletedSet.has(id)) + await Promise.all([fetchAccounts(), auth.fetchMe()]) + + if (!failures.length) { + message.success({ + content: `已删除 ${deletedIds.length} 个账号`, + key: 'batch_delete' + }) + } else if (deletedIds.length) { + message.warning({ + content: `已删除 ${deletedIds.length} 个账号，${failures.length} 个失败，可重新勾选后重试`, + key: 'batch_delete', + duration: 6 + }) + } else { + message.error({ + content: failures[0]?.reason || '批量删除失败', + key: 'batch_delete', + duration: 6 + }) + } + } finally { + batchDeleting.value = false } } @@ -1300,7 +1423,7 @@ const pollBatchStartStatus = (batchId, initialSnapshot = null) => { // 批量启动只提交一个请求；凭证校验和启动由后端小并发队列处理。 const runBatchStart = async ({ accountIds = [], allAccounts = false }) => { - if (batchStarting.value) return + if (batchStarting.value || batchDeleting.value) return batchStarting.value = true stopReplyQueueSummaryPolling() stopBatchStatusPolling() @@ -1336,7 +1459,7 @@ const runBatchStart = async ({ accountIds = [], allAccounts = false }) => { const accountLabel = (acc) => acc?.username || `账号 #${acc?.id}` const batchStartRpa = async () => { - if (batchStarting.value || startingAll.value) return + if (batchStarting.value || batchDeleting.value || startingAll.value) return const targets = startableAccounts.value .filter((a) => selectedIds.value.includes(a.id)) .map((a) => ({ id: a.id, label: accountLabel(a) })) @@ -1351,7 +1474,7 @@ const batchStartRpa = async () => { const startingAll = ref(false) const startAllRpa = async () => { - if (batchStarting.value || startingAll.value) return + if (batchStarting.value || batchDeleting.value || startingAll.value) return startingAll.value = true try { await runBatchStart({ allAccounts: true }) @@ -1601,8 +1724,13 @@ const openEditModal = async (acc) => { follow_welcome_content: acc.follow_welcome_content || '', user_agent_profile: 'chrome_win120', user_agent_custom: '', + egress_public_ip: acc.egress_public_ip || '', + egress_auto_attempts: Math.max(1, Number(acc.egress_auto_attempts) || 1), } initUserAgentFields(acc) + if (auth.canUpdateAccounts) { + fetchEgressChannels(false) + } try { if (auth.canManageCookies) { const res = await api.get(`/accounts/${acc.id}/cookie?purpose=management`) @@ -1690,8 +1818,10 @@ const saveAccountInfo = async () => { follow_welcome_enabled: !!editForm.value.follow_welcome_enabled, follow_welcome_content: (editForm.value.follow_welcome_content || '').trim() || null, user_agent: resolveUserAgentToSave() || null, + egress_public_ip: (editForm.value.egress_public_ip || '').trim() || null, + egress_auto_attempts: Math.max(1, Math.min(8, Number(editForm.value.egress_auto_attempts) || 1)), }) - message.success('账号信息已保存（设备头将在下次启动托管时生效）') + message.success('账号信息已保存（公网发送通道立即生效，设备头下次启动生效）') fetchAccounts() } catch (error) { message.error('保存账号信息失败') @@ -1805,26 +1935,54 @@ onUnmounted(() => {

-
+
- 全选可启动 ({{ startableAccounts.length }}) + {{ auth.canDeleteAccounts ? '全选当前页' : '全选可启动' }} ({{ selectableAccounts.length }}) 批量启动{{ selectedStartableCount ? ` (${selectedStartableCount})` : '' }} - + + + + 批量删除{{ selectedAccountCount ? ` (${selectedAccountCount})` : '' }} + + + 取消选择
@@ -1837,9 +1995,10 @@ onUnmounted(() => { @confirm="startAllRpa" > 一键启动全部 @@ -1933,9 +2092,10 @@ onUnmounted(() => { :class="{ 'account-card-selected': isAccountSelected(acc.id) }" >
+ + +
+ + + 重新检测 + +
+
+ + + +
+
+ {{ egressChannelsError }} +
+ + + + + +
+ 包含首选通道。只有明确收到通道/安全校验失败时才按顺序切换；超时等结果不确定的请求不会重发，避免重复消息。 +
+ + { background: rgba(255, 255, 255, 0.02) !important; } +.batch-toolbar :deep(.batch-delete-btn.ant-btn-dangerous) { + color: #fca5a5 !important; + border-color: rgba(248, 113, 113, 0.45) !important; + background: rgba(239, 68, 68, 0.08) !important; +} + +.batch-toolbar :deep(.batch-delete-btn.ant-btn-dangerous:not(:disabled):hover) { + color: #fecaca !important; + border-color: rgba(252, 165, 165, 0.75) !important; + background: rgba(239, 68, 68, 0.16) !important; +} + +.batch-toolbar :deep(.batch-delete-btn.ant-btn-dangerous:disabled) { + color: rgba(203, 213, 225, 0.45) !important; + border-color: rgba(255, 255, 255, 0.08) !important; + background: rgba(255, 255, 255, 0.02) !important; +} + .batch-toolbar :deep(.batch-clear-btn.ant-btn-default) { color: #cbd5e1 !important; border-color: rgba(255, 255, 255, 0.16) !important; @@ -4364,6 +4595,18 @@ onUnmounted(() => { line-height: 1.55; } +.egress-channel-row { + display: flex; + align-items: center; + gap: 10px; +} + +.egress-channel-error { + margin-top: 6px; + color: #fbbf24; + font-size: 0.78rem; +} + .im-credential-grid { display: grid; grid-template-columns: repeat(2, minmax(0, 1fr));