diff --git a/backend/rpa_engine/douyin_im/http_client.py b/backend/rpa_engine/douyin_im/http_client.py
index 73ff628..5aee173 100644
--- a/backend/rpa_engine/douyin_im/http_client.py
+++ b/backend/rpa_engine/douyin_im/http_client.py
@@ -818,7 +818,22 @@ class DouyinImHttpClient:
"""通过 IM API 发送 Protobuf 编码的私信（带接口签名）
content 可为纯文本，或 JSON 格式的结构化回复（文本/网址/卡片）。
+ 所有出站发送（自动回复/手动 API/脚本）都经过全局限流器，
+ 防止大量托管账号同时发送占满出站带宽或触发平台级风控。
"""
+ from .rate_limit import get_send_limiter
+
+ async with get_send_limiter():
+ return await self._send_text_message_unthrottled(
+ conversation_id, content, conversation_short_id
+ )
+
+ async def _send_text_message_unthrottled(
+ self,
+ conversation_id: str,
+ content: str,
+ conversation_short_id: str = "",
+ ) -> bool:
from .auth import DouyinAuth
from .proto_builder import ProtoBuilder
from .reply_payload import (
diff --git a/backend/rpa_engine/douyin_im/rate_limit.py b/backend/rpa_engine/douyin_im/rate_limit.py
new file mode 100644
index 0000000..d3b4180
--- /dev/null
+++ b/backend/rpa_engine/douyin_im/rate_limit.py
@@ -0,0 +1,117 @@
+"""全局出站发送限流器。
+
+目的：托管账号数量多（如 1000 个）时，避免自动回复/手动发送在同一时刻
+大量并发打向抖音接口，占满服务器出站带宽或触发平台级风控。
+
+机制（两层，先到先过）：
+ 1. 并发上限（Semaphore）：同时在途的发送请求数不超过 KEFU_SEND_MAX_CONCURRENCY；
+ 2. 令牌桶（QPS）：平均发送速率不超过 KEFU_SEND_RATE_PER_SEC，允许 KEFU_SEND_BURST 的突发。
+
+限流器按事件循环惰性创建（backend 主循环一个实例；独立脚本各自一个），
+所有账号共享同一个实例，因此是跨账号的全局闸门。
+
+环境变量：
+ KEFU_SEND_MAX_CONCURRENCY 同时在途发送数上限，默认 20，<=0 表示不限
+ KEFU_SEND_RATE_PER_SEC 平均每秒发送数上限，默认 10，<=0 表示不限
+ KEFU_SEND_BURST 令牌桶容量（突发上限），默认与并发上限相同
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+import os
+import time
+
+logger = logging.getLogger("douyin_im.rate_limit")
+
+
+def _env_float(name: str, default: float) -> float:
+ try:
+ return float(os.getenv(name, "") or default)
+ except ValueError:
+ return default
+
+
+class GlobalSendLimiter:
+ """并发上限 + 令牌桶。用法：async with limiter: await 发送。"""
+
+ def __init__(
+ self,
+ max_concurrency: int = 20,
+ rate_per_sec: float = 10.0,
+ burst: float | None = None,
+ ):
+ self.max_concurrency = int(max_concurrency)
+ self.rate_per_sec = float(rate_per_sec)
+ self.burst = float(burst if burst is not None else max(1, max_concurrency))
+ self._sem = (
+ asyncio.Semaphore(self.max_concurrency) if self.max_concurrency > 0 else None
+ )
+ self._tokens = self.burst
+ self._last_refill = time.monotonic()
+ self._token_lock = asyncio.Lock()
+ # 统计：便于日志观察限流是否生效
+ self._waited_total = 0.0
+ self._acquired_count = 0
+
+ async def _take_token(self) -> None:
+ if self.rate_per_sec <= 0:
+ return
+ async with self._token_lock:
+ while True:
+ now = time.monotonic()
+ self._tokens = min(
+ self.burst, self._tokens + (now - self._last_refill) * self.rate_per_sec
+ )
+ self._last_refill = now
+ if self._tokens >= 1.0:
+ self._tokens -= 1.0
+ return
+ await asyncio.sleep((1.0 - self._tokens) / self.rate_per_sec)
+
+ async def __aenter__(self) -> "GlobalSendLimiter":
+ start = time.monotonic()
+ if self._sem is not None:
+ await self._sem.acquire()
+ try:
+ await self._take_token()
+ except BaseException:
+ if self._sem is not None:
+ self._sem.release()
+ raise
+ waited = time.monotonic() - start
+ self._waited_total += waited
+ self._acquired_count += 1
+ if waited > 1.0:
+ logger.info(
+ "send throttled: waited %.1fs (in-flight cap=%s, rate=%s/s)",
+ waited, self.max_concurrency or "∞", self.rate_per_sec or "∞",
+ )
+ return self
+
+ async def __aexit__(self, *exc) -> None:
+ if self._sem is not None:
+ self._sem.release()
+
+
+# 每个事件循环一个实例（backend 主循环即全局唯一；脚本自用循环互不影响）
+_limiters: dict[int, GlobalSendLimiter] = {}
+
+
+def get_send_limiter() -> GlobalSendLimiter:
+ loop = asyncio.get_running_loop()
+ key = id(loop)
+ limiter = _limiters.get(key)
+ if limiter is None:
+ max_conc = int(_env_float("KEFU_SEND_MAX_CONCURRENCY", 20))
+ rate = _env_float("KEFU_SEND_RATE_PER_SEC", 10.0)
+ burst = _env_float("KEFU_SEND_BURST", max(1, max_conc))
+ limiter = GlobalSendLimiter(max_conc, rate, burst)
+ _limiters[key] = limiter
+ logger.info(
+ "global send limiter ready: concurrency=%s, rate=%s/s, burst=%s",
+ max_conc if max_conc > 0 else "unlimited",
+ rate if rate > 0 else "unlimited",
+ burst,
+ )
+ return limiter
diff --git a/backend/rpa_engine/douyin_im/service.py b/backend/rpa_engine/douyin_im/service.py
index 3d4773d..5fdb1bf 100644
--- a/backend/rpa_engine/douyin_im/service.py
+++ b/backend/rpa_engine/douyin_im/service.py
@@ -1,5 +1,7 @@
import asyncio
import logging
+import os
+import random
import time
from typing import Awaitable, Callable, Optional
@@ -374,6 +376,22 @@ class DouyinImService:
)
await self._ws_client.start()
+ # 轮询错峰：多账号同时托管时，若所有账号按同一节奏轮询，请求会在同一
+ # 时刻叠峰。这里给每个账号随机相位偏移 + 每轮 ±20% 抖动，把请求摊平。
+ # WS 可用时轮询只是兜底，可以适当放缓（KEFU_IM_POLL_INTERVAL_SECONDS 可调）。
+ try:
+ poll_interval = float(os.getenv("KEFU_IM_POLL_INTERVAL_SECONDS", "") or 15)
+ except ValueError:
+ poll_interval = 15.0
+ poll_interval = max(5.0, poll_interval)
+ if has_ws:
+ poll_interval = max(poll_interval, 30.0)
+
+ # 首轮轮询前的随机延迟（相位偏移），批量启动时错开各账号的首波请求
+ await asyncio.sleep(random.uniform(0.5, min(10.0, poll_interval)))
+ if not self._running:
+ return
+
try:
await self._poll_conversations()
except Exception as e:
@@ -387,10 +405,12 @@ class DouyinImService:
)
loop_count = 0
+ next_poll = time.monotonic() + poll_interval * random.uniform(0.8, 1.2)
while self._running:
loop_count += 1
try:
- if loop_count % 3 == 1:
+ if time.monotonic() >= next_poll:
+ next_poll = time.monotonic() + poll_interval * random.uniform(0.8, 1.2)
await self._poll_conversations()
if loop_count % 6 == 1:
logger.info(f"IM direct tick #{loop_count} account={self.account_id}")
diff --git a/backend/rpa_engine/playwright_worker.py b/backend/rpa_engine/playwright_worker.py
index 04c8edd..eb9f01f 100644
--- a/backend/rpa_engine/playwright_worker.py
+++ b/backend/rpa_engine/playwright_worker.py
@@ -49,6 +49,36 @@ async def _launch_chromium(pw, args: list[str], headless: Optional[bool] = None)
return await pw.chromium.launch(**launch_kwargs)
+# ---------------------------------------------------------------------------
+# 启动错峰门：批量启动大量账号时，把各 worker 的启动时刻按固定间隔排开，
+# 避免同一瞬间大量凭证校验/WS 建连/首轮拉取叠峰。
+# 空闲时单个账号启动无需等待；只有短时间内大量启动才会排队。
+# KEFU_WORKER_START_INTERVAL_SECONDS：相邻两个 worker 启动的最小间隔，默认 1.5s，<=0 关闭。
+# ---------------------------------------------------------------------------
+_start_gate = {"lock": None, "next_at": 0.0}
+
+
+async def _startup_stagger(account_id: int) -> None:
+ try:
+ interval = float(os.getenv("KEFU_WORKER_START_INTERVAL_SECONDS", "") or 1.5)
+ except ValueError:
+ interval = 1.5
+ if interval <= 0:
+ return
+ if _start_gate["lock"] is None:
+ _start_gate["lock"] = asyncio.Lock()
+ async with _start_gate["lock"]:
+ now = time.monotonic()
+ wait = max(0.0, _start_gate["next_at"] - now)
+ _start_gate["next_at"] = max(now, _start_gate["next_at"]) + interval
+ if wait > 0:
+ if wait > 5:
+ logger.info(
+ f"Account {account_id}: start queued, waiting {wait:.1f}s to smooth batch startup"
+ )
+ await asyncio.sleep(wait)
+
+
def format_error(exc: BaseException) -> str:
message = str(exc).strip()
if "Target page, context or browser has been closed" in message:
@@ -995,6 +1025,9 @@ class DouyinWorker:
logger.info(f"Starting worker loop for account {self.account_id}")
try:
+ await _startup_stagger(self.account_id)
+ if self.stopping:
+ return
storage_state = await self._load_storage_state()
cookie_info = analyze_cookie(
json.dumps(storage_state, ensure_ascii=False) if storage_state else None
diff --git a/douyin-login-launcher/desktop_app.py b/douyin-login-launcher/desktop_app.py
index 742d8b1..c88e7b5 100644
--- a/douyin-login-launcher/desktop_app.py
+++ b/douyin-login-launcher/desktop_app.py
@@ -1,10 +1,12 @@
"""抖音托管客服 · 桌面整合版
一个软件搞定一切：
- 1) 启动后默认打开云端网站（dev.zhenyangtang.com.cn）；
- 2) 页面右下角自动出现「一键本地登录」悬浮按钮；
- 3) 点它 → 弹出你的托管账号列表 → 选一个 → 本机直接打开一个
- 已登录该托管账号的浏览器，进入抖音。
+ 1) 启动后先显示「站点选择」界面（内置 dev / dev1，可自行添加更多站点，
+ 可勾选“记住选择”下次直接进入）；
+ 2) 进入站点后，页面右下角自动出现「一键本地登录」悬浮按钮，
+ 以及「切换站点」按钮可随时换环境；
+ 3) 点「一键本地登录」→ 弹出你的托管账号列表 → 选一个 → 本机直接打开
+ 一个已登录该托管账号的浏览器，进入抖音。
原理：桌面壳（pywebview）把云端网页装进原生窗口，并注入一段脚本。该脚本
用网页里已有的登录令牌(localStorage.kefu_token)调用云端接口取账号与凭证，
@@ -18,6 +20,7 @@
from __future__ import annotations
import ctypes
+import json
import os
import subprocess
import threading
@@ -29,10 +32,276 @@ from browser_open import normalize_storage_state, open_logged_in_browser
from updater import check_update, download_installer
from version import __version__
-CLOUD_URL = "https://dev.zhenyangtang.com.cn/"
-CLOUD_HOST = urlparse(CLOUD_URL).hostname or ""
WINDOW_TITLE = f"抖音托管客服 · 桌面版 v{__version__}"
+# ---------------------------------------------------------------------------
+# 多站点配置：内置站点 + 用户自定义站点（保存在用户目录，升级软件不丢失）。
+# 以后要新增内置环境，往 BUILTIN_SITES 里加一行即可。
+# ---------------------------------------------------------------------------
+BUILTIN_SITES = [
+ {"name": "节点1", "url": "https://dev.zhenyangtang.com.cn/"},
+ {"name": "节点2", "url": "https://dev1.zhenyangtang.com.cn/"},
+]
+CONFIG_PATH = os.path.join(os.path.expanduser("~"), ".douyin_desktop_config.json")
+
+
+def load_config() -> dict:
+ try:
+ with open(CONFIG_PATH, "r", encoding="utf-8") as f:
+ data = json.load(f)
+ return data if isinstance(data, dict) else {}
+ except Exception: # noqa: BLE001
+ return {}
+
+
+def save_config(cfg: dict) -> None:
+ try:
+ with open(CONFIG_PATH, "w", encoding="utf-8") as f:
+ json.dump(cfg, f, ensure_ascii=False, indent=2)
+ except Exception: # noqa: BLE001
+ pass
+
+
+def normalize_site_url(url: str) -> str:
+ url = (url or "").strip()
+ if not url:
+ return ""
+ if not url.startswith(("http://", "https://")):
+ url = "https://" + url
+ if not url.endswith("/"):
+ url += "/"
+ return url
+
+
+def get_all_sites() -> list[dict]:
+ """内置站点 + 自定义站点（去重，按 URL）。"""
+ cfg = load_config()
+ sites: list[dict] = []
+ seen: set[str] = set()
+ for s in BUILTIN_SITES:
+ u = normalize_site_url(s["url"])
+ sites.append({"name": s["name"], "url": u, "builtin": True})
+ seen.add(u)
+ for s in cfg.get("custom_sites", []):
+ u = normalize_site_url(s.get("url", ""))
+ if u and u not in seen:
+ sites.append({"name": s.get("name") or urlparse(u).hostname, "url": u, "builtin": False})
+ seen.add(u)
+ return sites
+
+
+def site_hosts() -> set[str]:
+ return {urlparse(s["url"]).hostname or "" for s in get_all_sites()} - {""}
+
+
+# ---------------------------------------------------------------------------
+# 站点选择界面（软件启动首页）。与注入浮层同一套视觉语言：#fe2c55 主色 + 玻璃拟态。
+# ---------------------------------------------------------------------------
+PICKER_HTML = r"""
+
+
+
+选择站点
+
+
+
+
+
+
+
+
+
抖音托管客服 · 桌面版
+
请选择要进入的站点，也可以在下方添加新站点
+
+
+
+
+
+