kris/ForcePilot
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat(douyin): 新增抖音渠道插件，支持私信收发与相关能力

Browse Source 
该提交实现了完整的抖音开放平台IM渠道插件，包含：
1. 基础配置与账号管理能力
2. 消息去重、安全策略校验
3. 流式回复、多媒体消息发送
4. Webhook回调处理与事件解析
5. 配对认证与流量限流机制
This commit is contained in:
Kris 2026-05-21 10:45:37 +08:00
parent e29cc2d611
commit 4552bde837
14 changed files with 1741 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

250
backend/package/yuxi/channel/extensions/douyin/__init__.py Normal file
View File

@ -0,0 +1,250 @@
from yuxi.channel.capabilities import ChannelCapabilities
from yuxi.channel.extensions.base import BaseChannelPlugin
from yuxi.channel.extensions.douyin.config import DouyinConfig
from yuxi.channel.extensions.douyin.dedupe import MessageDeduplicator
from yuxi.channel.extensions.douyin.gateway import DouyinGateway
from yuxi.channel.extensions.douyin.media import DouyinMedia
from yuxi.channel.extensions.douyin.outbound import DouyinOutbound
from yuxi.channel.extensions.douyin.pairing import DouyinPairing
from yuxi.channel.extensions.douyin.security import DouyinSecurity
from yuxi.channel.extensions.douyin.status import DouyinStatus
from yuxi.channel.extensions.douyin.streaming import DouyinStreaming
from yuxi.channel.plugins.registry import ChannelPluginRegistry
from yuxi.channel.protocols import ChannelAccountSnapshot
class DouyinPlugin(BaseChannelPlugin):
id = "douyin"
name = "抖音"
order = 50
label = "抖音 (Douyin)"
aliases = ["douyin", "dy", "抖音企业号", "抖音私信"]
resolve_reply_to_mode = "off"
def __init__(self):
super().__init__()
self._config_adapter = DouyinConfig()
self._gateway: DouyinGateway | None = None
self._outbound: DouyinOutbound | None = None
self._status: DouyinStatus | None = None
self._security: DouyinSecurity | None = None
self._pairing = DouyinPairing()
self._streaming = DouyinStreaming()
self._media: DouyinMedia | None = None
self._deduplicator = MessageDeduplicator()
self._remove_markdown: bool = True
@property
def capabilities(self) -> ChannelCapabilities:
return ChannelCapabilities(
chat_types=["direct"],
message_types=["text", "image", "video", "card"],
reactions=False,
typing_indicator=False,
threads=False,
edit=False,
unsend=True,
reply=False,
media=True,
effects=False,
native_commands=False,
polls=False,
group_management=False,
streaming=True,
streaming_mode="block",
block_streaming=True,
block_streaming_chunk_min_chars=300,
block_streaming_chunk_max_chars=800,
block_streaming_chunk_break_preference="paragraph",
block_streaming_coalesce_min_chars=80,
block_streaming_coalesce_max_chars=400,
block_streaming_coalesce_idle_ms=500,
tts=None,
)
def list_account_ids(self, config: dict | None = None) -> list[str]:
return self._config_adapter.list_account_ids(config)
async def resolve_account(self, account_id: str = "default") -> dict:
account = self._config_adapter.resolve_account(account_id)
return {
"account_id": account.account_id,
"client_key": account.client_key,
"client_secret": account.client_secret,
"dm_policy": account.dm_policy,
"allow_from": account.allow_from,
"streaming": account.streaming,
"remove_markdown": account.remove_markdown,
"welcome_text": account.welcome_text,
}
def is_configured(self, account: dict | None = None) -> bool:
return self._config_adapter.is_configured(account)
def is_enabled(self, account: dict | None = None) -> bool:
return self._config_adapter.is_enabled(account)
def disabled_reason(self, account: dict | None = None) -> str:
return self._config_adapter.disabled_reason(account)
def describe_account(self, account: dict | None = None) -> dict:
return self._config_adapter.describe_account(account)
def config_schema(self) -> dict:
return self._config_adapter.config_schema()
async def start(self, ctx) -> object:
self._gateway = DouyinGateway(config_adapter=self._config_adapter)
result = await self._gateway.start(ctx)
self._outbound = DouyinOutbound(self._gateway)
self._media = DouyinMedia(lambda: self._gateway.client_token if self._gateway else None)
self._status = DouyinStatus(self._gateway)
account = self._config_adapter.resolve_account()
self._security = DouyinSecurity(account)
self._remove_markdown = account.remove_markdown
from yuxi.channel.extensions.douyin.webhook import set_plugin
set_plugin(self)
return result
async def stop(self, ctx) -> None:
import logging as _logging
_logger = _logging.getLogger(__name__)
if self._gateway:
token_status = "has_biz_token" if self._gateway.business_token else "no_biz_token"
_logger.info(
"Douyin audit: gateway_stop token=%s account=%s",
token_status,
getattr(self._gateway, "_account", None),
)
from yuxi.channel.extensions.douyin.webhook import clear_plugin
clear_plugin()
if self._media:
await self._media.close()
self._media = None
if self._gateway:
await self._gateway.stop(ctx)
self._gateway = None
self._outbound = None
self._status = None
async def on_config_changed(self, prev_cfg: dict, next_cfg: dict, account_id: str) -> None:
if prev_cfg != next_cfg:
import logging as _logging
_logger = _logging.getLogger(__name__)
_logger.info("Douyin config changed, reloading...")
from yuxi.channel.context import ChannelContext
ctx = ChannelContext(channel_type="douyin", account_id=account_id, config=next_cfg)
await self.stop(ctx)
await self.start(ctx)
async def send_text(
self,
target_id: str,
content: str,
*,
reply_to_id: str | None = None,
thread_id: str | None = None,
account_id: str | None = None,
) -> None:
if not content or not self._outbound:
return
await self._outbound.send_text(
target_id, content, reply_to_id=reply_to_id, thread_id=thread_id, account_id=account_id
)
async def send_media(
self,
target_id: str,
media_url: str,
media_type: str,
reply_to_id: str | None = None,
thread_id: str | None = None,
) -> None:
if not self._outbound:
return
if media_type == "image":
media_id = await self._outbound.upload_image(media_url)
if media_id:
await self._outbound.send_image(target_id, media_id)
elif media_type == "video":
await self._outbound.send_video(target_id, media_url)
async def send_card(
self,
target_id: str,
card_template_id: str,
card_data: dict | None = None,
) -> None:
if not self._outbound:
return
await self._outbound.send_card(target_id, card_template_id, card_data)
async def send_guide_card(
self,
target_id: str,
questions: list[dict],
card_type: int = 204,
) -> None:
if not self._outbound:
return
await self._outbound.send_guide_card(target_id, questions, card_type)
async def recall_msg(self, open_id: str, server_message_id: str) -> bool:
if not self._outbound:
return False
return await self._outbound.recall_msg(open_id, server_message_id)
async def probe(self, account: dict | None = None) -> bool:
if self._status:
return await self._status.probe(account)
return False
def build_summary(self, snapshot: object) -> dict:
if self._status:
return self._status.build_summary(snapshot)
return {}
def build_account_snapshot(
self,
account: dict,
config: dict,
runtime: ChannelAccountSnapshot | None = None,
probe_result: object | None = None,
audit: object | None = None,
) -> ChannelAccountSnapshot:
if self._status:
return self._status.build_account_snapshot(account, config, runtime, probe_result, audit)
return ChannelAccountSnapshot(account_id="default")
def resolve_dm_policy(self) -> dict:
if self._security:
return {"mode": self._security.resolve_dm_policy(), "allow_from": []}
return {"mode": "open", "allow_from": []}
async def check_allowlist(self, peer_id: str, channel_type: str) -> bool:
if self._security:
return self._security.check_allowlist(peer_id, channel_type)
return True
async def generate_code(self, peer_id: str) -> str:
code = self._pairing.generate_code(peer_id)
return code or ""
async def verify_code(self, peer_id: str, code: str) -> bool:
return self._pairing.verify(peer_id, code)
ChannelPluginRegistry.register(DouyinPlugin())

134
backend/package/yuxi/channel/extensions/douyin/config.py Normal file
View File

@ -0,0 +1,134 @@
import os
from yuxi.channel.extensions.douyin.types import DouyinAccount
class DouyinConfig:
ENV_MAP = {
"client_key": "DOUYIN_CLIENT_KEY",
"client_secret": "DOUYIN_CLIENT_SECRET",
"dm_policy": "DOUYIN_DM_POLICY",
"streaming": "DOUYIN_STREAMING",
"remove_markdown": "DOUYIN_REMOVE_MD",
"welcome_text": "DOUYIN_WELCOME_TEXT",
}
def __init__(self, config: dict | None = None):
self._config = config or {}
def list_account_ids(self, config: dict | None = None) -> list[str]:
cfg = config or self._config
accounts = cfg.get("accounts", [])
if accounts:
return [a.get("account_id", "default") for a in accounts]
if self._resolve_client_key(cfg):
return ["default"]
return []
def resolve_account(self, account_id: str = "default", config: dict | None = None) -> DouyinAccount:
cfg = config or self._config
accounts = cfg.get("accounts", [])
account_data = {}
for a in accounts:
if a.get("account_id") == account_id:
account_data = a
break
return DouyinAccount(
account_id=account_id,
client_key=account_data.get("client_key") or self._resolve_client_key(cfg) or "",
client_secret=account_data.get("client_secret") or self._env_or_config("client_secret", cfg) or "",
dm_policy=account_data.get("dm_policy") or self._env_or_config("dm_policy", cfg) or "open",
streaming=self._resolve_bool(account_data, "streaming", cfg, True),
remove_markdown=self._resolve_bool(account_data, "remove_markdown", cfg, True),
welcome_text=account_data.get("welcome_text")
or self._env_or_config("welcome_text", cfg)
or "你好！有什么可以帮助你的？",
)
def is_configured(self, account: dict | None = None) -> bool:
if account:
return bool(account.get("client_key") and account.get("client_secret"))
return bool(self._resolve_client_key(self._config) and self._env_or_config("client_secret", self._config))
def is_enabled(self, account: dict | None = None, config: dict | None = None) -> bool:
cfg = config or self._config
if account:
return account.get("enabled", True)
return cfg.get("enabled", True)
def disabled_reason(self, account: dict | None = None) -> str:
if account and account.get("enabled") is False:
return "Account explicitly disabled"
return ""
def describe_account(self, account: dict | None = None) -> dict:
if account:
return {"account_id": account.get("account_id", "default")}
return {"account_id": "default"}
def _resolve_client_key(self, config: dict | None = None) -> str | None:
return self._env_or_config("client_key", config)
def _env_or_config(self, key: str, config: dict | None = None) -> str | None:
cfg = config or self._config
if key in cfg:
val = cfg[key]
if val:
return str(val)
env_key = self.ENV_MAP.get(key, "")
if env_key:
val = os.getenv(env_key)
if val:
return val
return None
def _resolve_bool(self, account_data: dict, key: str, config: dict | None, default: bool) -> bool:
if key in account_data:
val = account_data[key]
if isinstance(val, bool):
return val
if isinstance(val, str):
return val.lower() != "false"
env_val = self._env_or_config(key, config)
if env_val is not None:
return env_val.lower() != "false"
return default
def config_schema(self) -> dict:
return {
"$schema": "https://json-schema.org/draft-07/schema#",
"type": "object",
"title": "抖音渠道配置",
"properties": {
"client_key": {
"type": "string",
"title": "Client Key",
"description": "抖音开放平台小程序应用的 Client Key",
},
"client_secret": {
"type": "string",
"title": "Client Secret",
"x-ui-password": True,
"description": "抖音开放平台小程序应用的 Client Secret",
},
"dm_policy": {
"type": "string",
"enum": ["open", "pairing", "allowlist", "disabled"],
"default": "open",
"title": "DM 策略",
},
"streaming": {
"type": "boolean",
"default": True,
"title": "启用流式输出",
},
"remove_markdown": {
"type": "boolean",
"default": True,
"title": "移除 Markdown 格式",
"description": "自动移除 AI 回复中的 Markdown 符号和外部 URL 链接",
},
},
"required": ["client_key", "client_secret"],
}

43
backend/package/yuxi/channel/extensions/douyin/dedupe.py Normal file
View File

@ -0,0 +1,43 @@
import time
from collections import OrderedDict
class MessageDeduplicator:
MAX_SIZE = 10000
TTL_SECONDS = 300
def __init__(self, max_size: int = MAX_SIZE, ttl: int = TTL_SECONDS):
self._cache: OrderedDict[str, float] = OrderedDict()
self._max_size = max_size
self._ttl = ttl
def is_duplicate(self, msg_id: str) -> bool:
if not msg_id:
return False
self._evict_expired()
if msg_id in self._cache:
return True
self._cache[msg_id] = time.time()
self._cache.move_to_end(msg_id)
while len(self._cache) > self._max_size:
self._cache.popitem(last=False)
return False
def mark_seen(self, msg_id: str) -> None:
if not msg_id:
return
self._cache[msg_id] = time.time()
def _evict_expired(self):
now = time.time()
expired = [k for k, ts in self._cache.items() if now - ts > self._ttl]
for k in expired:
del self._cache[k]
def reset(self):
self._cache.clear()

236
backend/package/yuxi/channel/extensions/douyin/gateway.py Normal file
View File

@ -0,0 +1,236 @@
import asyncio
import logging
import time
import httpx
logger = logging.getLogger(__name__)
TOKEN_URL = "https://open.douyin.com/oauth/client_token/"
BUSINESS_TOKEN_URL = "https://open.douyin.com/oauth/business_token/"
REFRESH_BIZ_TOKEN_URL = "https://open.douyin.com/oauth/refresh_biz_token/"
REFRESH_INTERVAL = 3600
BIZ_REFRESH_INTERVAL = 29 * 86400
RETRY_MAX = 3
RETRY_BASE_DELAY = 2
class DouyinGateway:
def __init__(self, config_adapter=None):
self._config = config_adapter
self._account = None
self._client_token: str | None = None
self._token_expires_at: float = 0
self._refresh_task: asyncio.Task | None = None
self._business_token: str | None = None
self._biz_token_expires_at: float = 0
self._biz_refresh_token: str | None = None
self._biz_refresh_task: asyncio.Task | None = None
self._cancel_event = asyncio.Event()
self.http: httpx.AsyncClient | None = None
self._running = False
self._token_lock = asyncio.Lock()
self._biz_token_lock = asyncio.Lock()
async def start(self, ctx) -> object:
from yuxi.channel.extensions.douyin.config import DouyinConfig
config_adapter = self._config or DouyinConfig()
account = config_adapter.resolve_account()
if not account.is_configured():
logger.warning("Douyin account not configured, skipping start")
return {"running": False, "reason": "not-configured"}
self._account = account
self._cancel_event.clear()
self.http = httpx.AsyncClient(
base_url="https://open.douyin.com",
timeout=15.0,
)
try:
await self._fetch_client_token()
await self._fetch_business_token()
self._refresh_task = asyncio.create_task(self._token_refresh_loop())
self._biz_refresh_task = asyncio.create_task(self._biz_token_refresh_loop())
self._running = True
logger.info("Douyin gateway started for account %s", account.account_id)
return {"running": True, "account_id": account.account_id}
except Exception:
logger.exception("Douyin gateway failed to start")
await self.stop(ctx)
raise
async def stop(self, ctx) -> None:
self._running = False
self._cancel_event.set()
for task in (self._refresh_task, self._biz_refresh_task):
if task and not task.done():
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
if self.http:
await self.http.aclose()
self.http = None
logger.info("Douyin gateway stopped")
@property
def client_token(self) -> str | None:
if self._client_token and time.time() < self._token_expires_at:
return self._client_token
return None
@property
def business_token(self) -> str | None:
if self._business_token and time.time() < self._biz_token_expires_at:
return self._business_token
return None
async def _fetch_client_token(self) -> str:
async with self._token_lock:
if self._client_token and time.time() < self._token_expires_at:
return self._client_token
for attempt in range(RETRY_MAX):
try:
resp = await self.http.post(
"/oauth/client_token/",
json={
"client_key": self._account.client_key,
"client_secret": self._account.client_secret,
"grant_type": "client_credential",
},
)
data = resp.json()
inner = data.get("data", {})
if inner.get("error_code", -1) != 0:
raise ConnectionError(f"Failed to get client_token: {data}")
self._client_token = inner["access_token"]
expires_in = inner.get("expires_in", 7200)
self._token_expires_at = time.time() + expires_in - 300
logger.info("Douyin client_token refreshed, expires_in=%ds", expires_in)
return self._client_token
except Exception:
if attempt == RETRY_MAX - 1:
raise
delay = RETRY_BASE_DELAY * (2**attempt)
await asyncio.sleep(delay)
raise ConnectionError("Failed to get client_token after retries")
async def _fetch_business_token(self) -> str:
async with self._biz_token_lock:
if self._business_token and time.time() < self._biz_token_expires_at:
return self._business_token
client_token = self._client_token
if not client_token:
client_token = await self._fetch_client_token()
for attempt in range(RETRY_MAX):
try:
resp = await self.http.post(
"/oauth/business_token/",
json={
"client_key": self._account.client_key,
"client_secret": self._account.client_secret,
"grant_type": "client_credential",
},
headers={"access-token": client_token},
)
data = resp.json()
inner = data.get("data", {})
if inner.get("error_code", -1) != 0:
err_msg = data.get("message", inner.get("description", str(data)))
logger.error("Failed to get business_token: %s", err_msg)
if attempt == RETRY_MAX - 1:
raise ConnectionError(f"Failed to get business_token: {data}")
delay = RETRY_BASE_DELAY * (2**attempt)
await asyncio.sleep(delay)
continue
self._business_token = inner["access_token"]
self._biz_refresh_token = inner.get("refresh_token")
expires_in = inner.get("expires_in", 2592000)
self._biz_token_expires_at = time.time() + expires_in - 86400
logger.info(
"Douyin business_token refreshed, expires_in=%ds, has_refresh=%s",
expires_in,
bool(self._biz_refresh_token),
)
return self._business_token
except Exception:
if attempt == RETRY_MAX - 1:
raise
delay = RETRY_BASE_DELAY * (2**attempt)
await asyncio.sleep(delay)
raise ConnectionError("Failed to get business_token after retries")
async def _refresh_biz_token(self) -> str:
if not self._biz_refresh_token:
return await self._fetch_business_token()
for attempt in range(RETRY_MAX):
try:
resp = await self.http.post(
"/oauth/refresh_biz_token/",
json={
"client_key": self._account.client_key,
"grant_type": "refresh_biz_token",
"refresh_token": self._biz_refresh_token,
},
)
data = resp.json()
inner = data.get("data", {})
if inner.get("error_code", -1) != 0:
logger.warning("business_token refresh failed, retrying full fetch")
return await self._fetch_business_token()
self._business_token = inner["access_token"]
self._biz_refresh_token = inner.get("refresh_token")
expires_in = inner.get("expires_in", 2592000)
self._biz_token_expires_at = time.time() + expires_in - 86400
logger.info("Douyin business_token refreshed via refresh_token, expires_in=%ds", expires_in)
return self._business_token
except Exception:
if attempt == RETRY_MAX - 1:
logger.exception("business_token refresh exhausted retries")
return await self._fetch_business_token()
delay = RETRY_BASE_DELAY * (2**attempt)
await asyncio.sleep(delay)
return await self._fetch_business_token()
async def _token_refresh_loop(self):
while not self._cancel_event.is_set():
try:
await asyncio.sleep(REFRESH_INTERVAL)
if not self._cancel_event.is_set():
await self._fetch_client_token()
except asyncio.CancelledError:
break
except Exception:
logger.exception("Douyin client_token refresh failed in loop")
async def _biz_token_refresh_loop(self):
while not self._cancel_event.is_set():
try:
await asyncio.sleep(BIZ_REFRESH_INTERVAL)
if not self._cancel_event.is_set():
async with self._biz_token_lock:
await self._refresh_biz_token()
except asyncio.CancelledError:
break
except Exception:
logger.exception("Douyin business_token refresh failed in loop")

69
backend/package/yuxi/channel/extensions/douyin/media.py Normal file
View File

@ -0,0 +1,69 @@
import asyncio
import logging
import httpx
logger = logging.getLogger(__name__)
DOWNLOAD_URL = "https://open.douyin.com/api/apps/v1/developer_toolbox/image_material/download/"
RETRY_MAX = 3
RETRY_BASE_DELAY = 2
class DouyinMedia:
def __init__(self, token_provider):
self._token_provider = token_provider
self._http: httpx.AsyncClient | None = None
async def _client(self) -> httpx.AsyncClient:
if self._http is None:
self._http = httpx.AsyncClient(timeout=30.0)
return self._http
async def download(self, media_id: str) -> dict:
token = self._token_provider()
if not token:
return {"success": False, "error": "no access_token"}
url = f"{DOWNLOAD_URL}?access_token={token}&media_id={media_id}"
for attempt in range(RETRY_MAX):
try:
client = await self._client()
resp = await client.get(url)
content_type = resp.headers.get("content-type", "")
if "application/json" in content_type or resp.text.startswith("{"):
data = resp.json()
if data.get("data", {}).get("error_code", -1) != 0:
logger.warning(
"Douyin media download failed (attempt %d): error_code=%s",
attempt + 1,
data.get("data", {}).get("error_code"),
)
await asyncio.sleep(RETRY_BASE_DELAY * (2**attempt))
continue
return {"success": False, "error": "unexpected json response"}
return {
"success": True,
"data": resp.content,
"content_type": content_type,
}
except Exception as e:
logger.warning(
"Douyin media download exception (attempt %d/%d): %s",
attempt + 1,
RETRY_MAX,
e,
)
await asyncio.sleep(RETRY_BASE_DELAY * (2**attempt))
logger.error("Douyin media download exhausted retries for media_id=%s", media_id)
return {"success": False, "error": "download retries exhausted"}
async def close(self):
if self._http:
await self._http.aclose()
self._http = None

331
backend/package/yuxi/channel/extensions/douyin/outbound.py Normal file
View File

@ -0,0 +1,331 @@
import asyncio
import logging
import re
import httpx
from yuxi.channel.extensions.douyin.window import DouyinWindowTracker
logger = logging.getLogger(__name__)
SEND_MSG_PATH = "/im/send/msg/"
UPLOAD_IMAGE_PATH = "/api/apps/v1/developer_toolbox/image_material/upload/"
RECALL_MSG_PATH = "/im/recall/msg/"
MAX_TEXT_LEN = 1000
def clean_for_douyin(text: str) -> str:
text = re.sub(r"```\w*\n?", "", text)
text = re.sub(r"`([^`]+)`", r"\1", text)
text = re.sub(r"\*\*([^*]+)\*\*", r"\1", text)
text = re.sub(r"\*([^*]+)\*", r"\1", text)
text = re.sub(r"__([^_]+)__", r"\1", text)
text = re.sub(r"_([^_]+)_", r"\1", text)
text = re.sub(r"(?m)^#{1,6}\s+", "", text)
text = re.sub(r"(?m)^[-*_]{3,}\s*$", "", text)
text = re.sub(r"https?://\S+", "[链接]", text)
return text.strip()
class DouyinOutbound:
def __init__(self, gateway=None):
self._gateway = gateway
self._send_context: dict[str, dict] = {}
self._window_tracker = DouyinWindowTracker()
@property
def window_tracker(self) -> DouyinWindowTracker:
return self._window_tracker
def set_send_context(self, open_id: str, conversation_id: str, server_message_id: str) -> None:
self._send_context[open_id] = {
"conversation_id": conversation_id,
"server_message_id": server_message_id,
}
def _http(self) -> httpx.AsyncClient | None:
if self._gateway and self._gateway.http:
return self._gateway.http
return None
async def send_text(
self,
to_user_id: str,
content: str,
*,
reply_to_id: str | None = None,
thread_id: str | None = None,
account_id: str | None = None,
) -> bool:
if not content:
return False
if not self._window_tracker.can_reply(to_user_id):
return False
content = clean_for_douyin(content)
if len(content) > MAX_TEXT_LEN:
content = content[:MAX_TEXT_LEN]
result = await self._send_msg(
to_user_id,
{
"msg_type": 1,
"text": {"text": content},
},
)
if result:
self._window_tracker.record_send(to_user_id)
return result
async def send_image(
self,
to_user_id: str,
media_id: str,
) -> bool:
if not media_id:
return False
if not self._window_tracker.can_reply(to_user_id):
return False
result = await self._send_msg(
to_user_id,
{
"msg_type": 2,
"image": {"media_id": media_id},
},
)
if result:
self._window_tracker.record_send(to_user_id)
return result
async def send_video(
self,
to_user_id: str,
item_id: str,
) -> bool:
if not item_id:
return False
if not self._window_tracker.can_reply(to_user_id):
return False
result = await self._send_msg(
to_user_id,
{
"msg_type": 3,
"video": {"item_id": item_id},
},
)
if result:
self._window_tracker.record_send(to_user_id)
return result
async def send_card(
self,
to_user_id: str,
card_template_id: str,
card_data: dict | None = None,
) -> bool:
if not card_template_id:
return False
if not self._window_tracker.can_reply(to_user_id):
return False
payload: dict = {
"msg_type": 10,
"applet_card": {
"card_template_id": card_template_id,
"card_data": card_data or {},
},
}
result = await self._send_msg(to_user_id, payload)
if result:
self._window_tracker.record_send(to_user_id)
return result
async def send_guide_card(
self,
to_user_id: str,
questions: list[dict],
card_type: int = 204,
) -> bool:
if not questions or card_type not in (204, 205):
return False
if not self._window_tracker.can_reply(to_user_id):
return False
result = await self._send_msg(
to_user_id,
{
"msg_type": card_type,
"guide_card": {"questions": questions},
},
)
if result:
self._window_tracker.record_send(to_user_id)
return result
async def recall_msg(
self,
open_id: str,
server_message_id: str,
) -> bool:
if not open_id or not server_message_id:
return False
token = self._gateway.business_token if self._gateway else None
if not token:
logger.error("No valid business_token for douyin recall")
return False
url = f"{RECALL_MSG_PATH}?open_id={open_id}"
headers = {
"access-token": token,
"Content-Type": "application/json",
}
client = self._http()
if client is None:
logger.error("No HTTP client available for douyin recall")
return False
for attempt in range(3):
try:
resp = await client.post(
url,
json={
"msg_id": server_message_id,
},
headers=headers,
)
data = resp.json()
err_code = data.get("data", {}).get("error_code", data.get("error_code", -1))
if err_code == 0:
logger.info(
"Douyin message recalled: open_id=%s, msg_id=%s",
open_id,
server_message_id,
)
return True
logger.warning(
"Douyin recall failed (attempt %d): error_code=%s, error_msg=%s",
attempt + 1,
err_code,
data.get("data", {}).get("description", data.get("message", "")),
)
except Exception as e:
logger.warning("Douyin recall attempt %d failed: %s", attempt + 1, e)
await asyncio.sleep(attempt + 1)
return False
async def _send_msg(self, to_user_id: str, content: dict) -> bool:
token = self._gateway.business_token if self._gateway else None
if not token:
logger.error("No valid business_token for douyin send")
return False
ctx = self._send_context.pop(to_user_id, {})
conversation_id = ctx.get("conversation_id", "")
server_message_id = ctx.get("server_message_id", "")
payload = {
"to_user_id": to_user_id,
"scene": "im_reply_msg",
**content,
}
if conversation_id:
payload["conversation_id"] = conversation_id
if server_message_id:
payload["msg_id"] = server_message_id
url = f"{SEND_MSG_PATH}?open_id={to_user_id}"
headers = {
"access-token": token,
"Content-Type": "application/json",
}
client = self._http()
if client is None:
logger.error("No HTTP client available for douyin send")
return False
for attempt in range(3):
try:
resp = await client.post(url, json=payload, headers=headers)
data = resp.json()
err_code = data.get("data", {}).get("error_code", data.get("error_code", -1))
if err_code == 0:
return True
logger.warning(
"Douyin send failed (attempt %d): error_code=%s, error_msg=%s",
attempt + 1,
err_code,
data.get("data", {}).get("description", data.get("message", "")),
)
if err_code == 2190001:
logger.error("Douyin access_token expired")
elif err_code == 2190008:
logger.warning("Douyin rate limit hit")
except Exception as e:
logger.warning("Douyin send attempt %d failed: %s", attempt + 1, e)
await asyncio.sleep(attempt + 1)
return False
async def upload_image(self, image_url: str) -> str | None:
token = self._gateway.client_token if self._gateway else None
if not token:
return None
client = self._http()
if client is None:
logger.error("No HTTP client available for douyin upload")
return None
headers = {
"access-token": token,
}
for attempt in range(3):
try:
resp = await client.post(
UPLOAD_IMAGE_PATH,
json={"image_material_url": image_url},
headers=headers,
)
data = resp.json()
inner = data.get("data", {})
if inner.get("error_code", -1) == 0:
return inner.get("image_id") or inner.get("media_id")
logger.warning("Douyin image upload failed (attempt %d): %s", attempt + 1, data)
except Exception:
logger.exception("Douyin image upload exception (attempt %d)", attempt + 1)
await asyncio.sleep(attempt + 1)
return None

57
backend/package/yuxi/channel/extensions/douyin/pairing.py Normal file
View File

@ -0,0 +1,57 @@
import random
import time
CODE_LENGTH = 6
CODE_TTL_SECONDS = 600
PENDING_MAX = 3
RATE_LIMIT_SECONDS = 3600
class DouyinPairing:
def __init__(self):
self._pending: dict[str, dict] = {}
self._openid_last_code_at: dict[str, float] = {}
def generate_code(self, open_id: str) -> str | None:
now = time.time()
if open_id in self._openid_last_code_at:
elapsed = now - self._openid_last_code_at[open_id]
if elapsed < RATE_LIMIT_SECONDS:
return None
if len(self._pending) >= PENDING_MAX:
oldest = min(self._pending.values(), key=lambda p: p["created_at"])
if now - oldest["created_at"] < CODE_TTL_SECONDS:
return None
expired_openid = next(k for k, v in self._pending.items() if v == oldest)
del self._pending[expired_openid]
code = _generate_code()
self._pending[open_id] = {"code": code, "created_at": now}
self._openid_last_code_at[open_id] = now
return code
def verify(self, open_id: str, code: str) -> bool:
entry = self._pending.get(open_id)
if not entry:
return False
if time.time() - entry["created_at"] > CODE_TTL_SECONDS:
del self._pending[open_id]
return False
if entry["code"] != code.upper().strip():
return False
del self._pending[open_id]
return True
def get_pending_count(self) -> int:
return len(self._pending)
def _generate_code() -> str:
no_ambiguous = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789"
return "".join(random.choices(no_ambiguous, k=CODE_LENGTH))

31
backend/package/yuxi/channel/extensions/douyin/plugin.json Normal file
View File

@ -0,0 +1,31 @@
{
"id": "douyin",
"name": "抖音",
"version": "0.1.0",
"label": "Douyin",
"aliases": ["douyin", "dy"],
"description": "抖音渠道插件，支持抖音小程序 IM 私信 Webhook 回调模式的消息收发",
"author": "ForcePilot Team",
"order": 50,
"dependencies": ["httpx"],
"capabilities": {
"chat_types": ["direct"],
"message_types": ["text", "image", "video", "card"],
"reactions": false,
"typing_indicator": false,
"threads": false,
"edit": false,
"unsend": true,
"reply": false,
"media": true,
"effects": false,
"native_commands": false,
"polls": false,
"group_management": false,
"streaming": true,
"streaming_mode": "block",
"block_streaming": true
},
"enabled": true,
"python_requires": ">=3.12"
}

36
backend/package/yuxi/channel/extensions/douyin/security.py Normal file
View File

@ -0,0 +1,36 @@
import logging
from yuxi.channel.extensions.douyin.types import DouyinAccount
logger = logging.getLogger(__name__)
class DouyinSecurity:
VALID_POLICIES = {"open", "pairing", "allowlist", "disabled"}
def __init__(self, account: DouyinAccount | None = None):
self._account = account
if account and account.dm_policy in self.VALID_POLICIES:
self._policy = account.dm_policy
else:
self._policy = "open"
self._allowlist: set[str] = set(account.allow_from) if account else set()
def resolve_dm_policy(self) -> str:
return self._policy
def check_allowlist(self, open_id: str, channel_type: str = "douyin") -> bool:
if self._policy == "open":
return True
if self._policy == "disabled":
return False
if not self._allowlist:
logger.warning("allowlist is empty with policy=%s", self._policy)
return False
return open_id in self._allowlist
def add_to_allowlist(self, open_id: str):
self._allowlist.add(open_id)
def remove_from_allowlist(self, open_id: str):
self._allowlist.discard(open_id)

104
backend/package/yuxi/channel/extensions/douyin/status.py Normal file
View File

@ -0,0 +1,104 @@
import logging
import httpx
from yuxi.channel.protocols import ChannelAccountSnapshot
logger = logging.getLogger(__name__)
PROBE_URL = "https://open.douyin.com/oauth/client_token/"
class DouyinStatus:
def __init__(self, gateway=None):
self._gateway = gateway
async def probe(self, account: dict | None = None) -> bool:
gateway = self._gateway
if gateway is None:
return False
token = gateway.client_token
if not token:
return False
url = f"{PROBE_URL}?access_token={token}"
try:
client = gateway.http
if client is None:
client = httpx.AsyncClient(timeout=10.0)
try:
resp = await client.get(url)
data = resp.json()
finally:
await client.aclose()
else:
resp = await client.get(url)
data = resp.json()
if "data" in data and data.get("data", {}).get("error_code", -1) == 0:
return True
logger.warning("Douyin probe failed: %s", data.get("message", "unknown error"))
return False
except Exception:
logger.exception("Douyin probe failed")
return False
def build_summary(self, snapshot: object) -> dict:
if not isinstance(snapshot, ChannelAccountSnapshot):
return {}
return {
"account_id": snapshot.account_id,
"running": snapshot.running,
"connected": snapshot.connected,
"health_state": snapshot.health_state,
}
def build_account_snapshot(
self,
account: dict,
config: dict,
runtime: ChannelAccountSnapshot | None = None,
probe_result: object | None = None,
audit: object | None = None,
) -> ChannelAccountSnapshot:
gateway = self._gateway
token = gateway.client_token if gateway else None
return ChannelAccountSnapshot(
account_id=account.get("account_id", "default"),
name=account.get("name", ""),
enabled=account.get("enabled", True),
configured=bool(account.get("client_key") and account.get("client_secret")),
status_state="linked" if token else "not-linked",
running=getattr(runtime, "running", False) if runtime else False,
connected=bool(token),
last_message_at=getattr(runtime, "last_message_at", None) if runtime else None,
health_state="ok" if probe_result else "unknown",
dm_policy=account.get("dm_policy", "open"),
)
def collect_status_issues(self, accounts: list[ChannelAccountSnapshot]) -> list:
issues = []
for acc in accounts:
if not acc.configured:
issues.append(
{
"channel": "douyin",
"account_id": acc.account_id,
"kind": "not-configured",
"message": "Douyin credentials not configured",
"fix": "Set DOUYIN_CLIENT_KEY and DOUYIN_CLIENT_SECRET",
}
)
elif not acc.connected:
issues.append(
{
"channel": "douyin",
"account_id": acc.account_id,
"kind": "not-connected",
"message": "Douyin gateway not running",
"fix": "Check environment variables and gateway status",
}
)
return issues

25
backend/package/yuxi/channel/extensions/douyin/streaming.py Normal file
View File

@ -0,0 +1,25 @@
class DouyinStreaming:
streaming_mode = "block"
preview_stream_throttle_ms = 160
preview_min_initial_chars = 18
block_streaming_enabled = True
block_streaming_break = "text_end"
block_streaming_chunk_min_chars = 300
block_streaming_chunk_max_chars = 900
block_streaming_chunk_break_preference = "paragraph"
block_streaming_coalesce_defaults = {
"min_chars": 80,
"max_chars": 400,
"idle_ms": 500,
}
@classmethod
def should_stream(cls, estimated_length: int) -> bool:
return estimated_length > cls.block_streaming_chunk_min_chars
@classmethod
def build_chunk(cls, text: str, is_final: bool) -> str:
if is_final:
return text
return f"{text}\n[⏳]"

38
backend/package/yuxi/channel/extensions/douyin/types.py Normal file
View File

@ -0,0 +1,38 @@
from dataclasses import dataclass, field
@dataclass
class DouyinAccount:
account_id: str = "default"
client_key: str = ""
client_secret: str = ""
dm_policy: str = "open"
allow_from: list[str] = field(default_factory=list)
streaming: bool = True
remove_markdown: bool = True
welcome_text: str = "你好！有什么可以帮助你的？"
def is_configured(self) -> bool:
return bool(self.client_key and self.client_secret)
@dataclass
class InboundDouyinEvent:
event: str
client_key: str
from_user_id: str
to_user_id: str
log_id: str
msg_id: str
msg_type: str
content: str
create_time: int
conversation_short_id: str
raw: dict = field(default_factory=dict)
@dataclass
class OutboundResult:
success: bool
error: str | None = None
detail: str | None = None

281
backend/package/yuxi/channel/extensions/douyin/webhook.py Normal file
View File

@ -0,0 +1,281 @@
import asyncio
import hashlib
import hmac
import json
import logging
from datetime import UTC, datetime
from fastapi import APIRouter, Request
from fastapi.responses import JSONResponse, PlainTextResponse
from yuxi.channel.extensions.douyin.outbound import DouyinOutbound, clean_for_douyin
from yuxi.channel.extensions.douyin.security import DouyinSecurity
from yuxi.channel.message.models import MessageType, PeerInfo, UnifiedMessage
from yuxi.channel.routing.models import PeerKind
from yuxi.channel.runtime.manager import gateway
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/webhook/douyin", tags=["douyin"])
_plugin = None
def set_plugin(plugin) -> None:
global _plugin
_plugin = plugin
def clear_plugin() -> None:
global _plugin
_plugin = None
def verify_signature(client_secret: str, body_bytes: bytes, signature: str) -> bool:
data = client_secret.encode() + body_bytes
computed = hashlib.sha1(data).hexdigest()
return hmac.compare_digest(computed, signature)
@router.post("/callback")
async def douyin_webhook_receive(request: Request):
body_bytes = await request.body()
body_str = body_bytes.decode("utf-8")
signature = request.headers.get("X-Douyin-Signature", "")
plugin = _plugin
if not plugin:
logger.error("Douyin plugin not initialized, rejecting webhook")
return JSONResponse({"status": "plugin_not_ready"}, status_code=503)
config_adapter = plugin._config_adapter
account = config_adapter.resolve_account()
if not verify_signature(account.client_secret, body_bytes, signature):
logger.warning("Douyin webhook: invalid X-Douyin-Signature")
return JSONResponse({"status": "signature_invalid"}, status_code=403)
try:
event = json.loads(body_str)
except json.JSONDecodeError:
logger.warning("Douyin webhook: invalid JSON body")
return PlainTextResponse("", status_code=400)
if event.get("event") == "verify_webhook":
challenge = event.get("content", {}).get("challenge")
if challenge:
return JSONResponse({"echostr": challenge})
return PlainTextResponse("", status_code=400)
event_type = event.get("event", "")
if event_type == "im_send_message_failed":
logger.error("Douyin message send failed: %s", json.dumps(event, ensure_ascii=False))
return PlainTextResponse("success")
if event_type == "im_send_msg":
logger.info(
"Douyin message sent callback: msg_id=%s, to_user=%s",
event.get("content", {}).get("server_message_id", ""),
event.get("to_user_id", ""),
)
return PlainTextResponse("success")
if event_type == "im_recall_msg":
logger.info(
"Douyin message recalled: msg_id=%s, from_user=%s",
event.get("content", {}).get("server_message_id", ""),
event.get("from_user_id", ""),
)
return PlainTextResponse("success")
if event_type == "im_msg_read":
logger.info(
"Douyin message read: msg_id=%s, from_user=%s",
event.get("content", {}).get("server_message_id", ""),
event.get("from_user_id", ""),
)
return PlainTextResponse("success")
if event_type not in ("im_receive_msg", "im_enter_direct_msg"):
return PlainTextResponse("success")
msg_id = request.headers.get("Msg-Id", event.get("log_id", ""))
if msg_id and plugin._deduplicator.is_duplicate(msg_id):
return PlainTextResponse("success")
content_data = event.get("content", {})
msg_type = content_data.get("message_type", "")
if msg_type == "text":
content = content_data.get("text", "")
elif msg_type in ("image", "user_local_image"):
content = "[图片]"
elif msg_type in ("video", "user_local_video"):
content = "[视频]"
elif msg_type == "emoji":
emoji_info = content_data.get("emoji", {})
emoji_text = emoji_info.get("text", "") or emoji_info.get("resource_url", "") or "[表情]"
content = f"[表情: {emoji_text}]"
elif msg_type == "retain_consult_card":
content = "[留资卡片]"
else:
content = f"[不支持的消息类型: {msg_type}]"
if not content:
return PlainTextResponse("success")
from_user_id = event.get("from_user_id", "")
user_infos = event.get("user_infos", [])
nick_name = from_user_id
avatar = ""
if user_infos:
user_info = user_infos[0]
nick_name = user_info.get("nick_name", from_user_id)
avatar = user_info.get("avatar", "")
outbound = plugin._outbound
if outbound:
outbound.window_tracker.record(from_user_id)
if event_type == "im_enter_direct_msg":
await _handle_enter_direct_msg(event, account, plugin)
return PlainTextResponse("success")
conversation_short_id = content_data.get("conversation_short_id", "")
server_message_id = content_data.get("server_message_id", "")
if (conversation_short_id or server_message_id) and outbound:
outbound.set_send_context(from_user_id, conversation_short_id, server_message_id)
security = plugin._security
if security is None:
security = DouyinSecurity(account)
if plugin._remove_markdown:
agent_content = clean_for_douyin(content)
else:
agent_content = content
policy = security.resolve_dm_policy()
if policy == "disabled":
return PlainTextResponse("success")
if policy == "pairing":
if not security.check_allowlist(from_user_id):
handled = await _handle_pairing(from_user_id, agent_content, plugin)
if handled:
return PlainTextResponse("success")
return PlainTextResponse("success")
if policy == "allowlist":
if not security.check_allowlist(from_user_id):
return PlainTextResponse("success")
unified = UnifiedMessage(
msg_id=msg_id,
channel_type="douyin",
account_id="default",
content=content,
message_type=MessageType.TEXT if msg_type == "text" else MessageType.IMAGE,
sender=PeerInfo(
id=from_user_id,
kind=PeerKind.DIRECT,
display_name=nick_name,
),
timestamp=datetime.fromtimestamp(content_data.get("create_time", 0) / 1000, tz=UTC),
raw_payload=event,
body_for_agent=agent_content,
metadata={
"conversation_short_id": conversation_short_id,
"message_type": msg_type,
"open_id": from_user_id,
"server_message_id": server_message_id,
"avatar": avatar,
"nick_name": nick_name,
},
)
asyncio.create_task(
_dispatch_to_agent(unified),
name=f"douyin-dispatch-{from_user_id}",
)
return PlainTextResponse("success")
async def _dispatch_to_agent(msg: UnifiedMessage) -> None:
processor = gateway._processor
if processor is None:
logger.warning("Message processor not available, cannot dispatch Douyin message")
return
try:
await asyncio.wait_for(processor.process(msg), timeout=120.0)
except TimeoutError:
logger.error("Agent response timeout for douyin user %s", msg.sender.id)
except Exception:
logger.exception("Failed to process Douyin message for user %s", msg.sender.id)
async def _handle_enter_direct_msg(event: dict, account, plugin) -> None:
from_user_id = event.get("from_user_id", "")
logger.info("Douyin user %s entered direct message session", from_user_id)
security = plugin._security
if security is None:
security = DouyinSecurity(account)
policy = security.resolve_dm_policy()
if policy == "disabled":
return
outbound = plugin._outbound
if outbound and not outbound.window_tracker.can_enter_dm_reply(from_user_id):
logger.info("Douyin enter-dm rate limited for user %s", from_user_id)
return
welcome_text = account.welcome_text or "你好！有什么可以帮助你的？"
await _send_douyin_text(from_user_id, welcome_text, plugin)
if outbound:
outbound.window_tracker.record_enter_dm_reply(from_user_id)
async def _handle_pairing(from_user_id: str, content: str, plugin) -> bool:
if content.strip().startswith("配对 "):
code_input = content.strip()[3:].strip()
if plugin._pairing.verify(from_user_id, code_input):
security = plugin._security
if security:
security.add_to_allowlist(from_user_id)
await _send_douyin_text(from_user_id, "配对成功！现在可以开始对话了。", plugin)
return True
else:
await _send_douyin_text(from_user_id, "配对码无效或已过期，请重新发送消息获取配对码。", plugin)
return True
else:
code = plugin._pairing.generate_code(from_user_id)
if code:
await _send_douyin_text(
from_user_id,
f"首次对话需要验证身份，请输入以下配对码：\n\n配对 {code}\n\n（配对码有效期 10 分钟）",
plugin,
)
else:
await _send_douyin_text(from_user_id, "配对请求过于频繁，请稍后再试。", plugin)
return True
return False
async def _send_douyin_text(to_user_id: str, content: str, plugin) -> None:
ob = plugin._outbound
if ob is None:
gw = plugin._gateway
if gw is None:
logger.warning("No Douyin gateway available for sending text")
return
ob = DouyinOutbound(gw)
try:
await ob.send_text(to_user_id, content)
except Exception:
logger.exception("Failed to send Douyin text to %s", to_user_id)

106
backend/package/yuxi/channel/extensions/douyin/window.py Normal file
View File

@ -0,0 +1,106 @@
import logging
import time
logger = logging.getLogger(__name__)
_WINDOW_SECONDS = 24 * 3600
_MAX_MSG_PER_WINDOW = 6
_ENTER_DM_REPLY_WINDOW = 30
_ENTER_DM_MAX_REPLIES_PER_WEBHOOK = 3
_ENTER_DM_MAX_REPLIES_PER_DAY = 3
_ENTER_DM_COOLDOWN_SECONDS = 3600
class DouyinWindowTracker:
def __init__(self):
self._last_msg_time: dict[str, float] = {}
self._msg_count: dict[str, int] = {}
self._last_send_time: dict[str, float] = {}
self._enter_dm_reply_count: dict[str, int] = {}
self._enter_dm_last_reply_time: dict[str, float] = {}
def record(self, open_id: str) -> None:
self._last_msg_time[open_id] = time.time()
def record_send(self, open_id: str) -> None:
now = time.time()
if open_id not in self._last_msg_time:
self._msg_count[open_id] = 0
return
if now - self._last_msg_time[open_id] > _WINDOW_SECONDS:
self._msg_count[open_id] = 0
self._msg_count[open_id] = self._msg_count.get(open_id, 0) + 1
self._last_send_time[open_id] = now
def can_reply(self, open_id: str) -> bool:
last = self._last_msg_time.get(open_id, 0)
now = time.time()
if now - last > _WINDOW_SECONDS:
logger.info(
"Douyin 24h reply window expired for user %s, last_msg_at=%s",
open_id,
last,
)
return False
count = self._msg_count.get(open_id, 0)
if count >= _MAX_MSG_PER_WINDOW:
logger.warning(
"Douyin max message count reached for user %s: %d/%d",
open_id,
count,
_MAX_MSG_PER_WINDOW,
)
return False
return True
def can_enter_dm_reply(self, open_id: str) -> bool:
now = time.time()
day_key = f"{open_id}:{int(now / 86400)}"
daily_count = self._enter_dm_reply_count.get(day_key, 0)
if daily_count >= _ENTER_DM_MAX_REPLIES_PER_DAY:
logger.info(
"Douyin enter-dm daily limit reached for user %s: %d",
open_id,
daily_count,
)
return False
last_reply = self._enter_dm_last_reply_time.get(open_id, 0)
if last_reply and now - last_reply < _ENTER_DM_COOLDOWN_SECONDS:
logger.info(
"Douyin enter-dm cooldown active for user %s, remaining=%ds",
open_id,
int(_ENTER_DM_COOLDOWN_SECONDS - (now - last_reply)),
)
return False
return True
def record_enter_dm_reply(self, open_id: str) -> None:
now = time.time()
day_key = f"{open_id}:{int(now / 86400)}"
self._enter_dm_reply_count[day_key] = self._enter_dm_reply_count.get(day_key, 0) + 1
self._enter_dm_last_reply_time[open_id] = now
def remaining_seconds(self, open_id: str) -> float:
last = self._last_msg_time.get(open_id, 0)
if last == 0:
return 0
elapsed = time.time() - last
remaining = _WINDOW_SECONDS - elapsed
return max(0.0, remaining)
def cleanup(self):
now = time.time()
expired = [k for k, ts in self._last_msg_time.items() if now - ts > _WINDOW_SECONDS]
for k in expired:
del self._last_msg_time[k]
self._msg_count.pop(k, None)
self._last_send_time.pop(k, None)
stale_enter = [
k for k, ts in self._enter_dm_last_reply_time.items() if now - ts > _ENTER_DM_COOLDOWN_SECONDS * 2
]
for k in stale_enter:
self._enter_dm_last_reply_time.pop(k, None)