diff --git a/backend/package/yuxi/channel/extensions/douyin/__init__.py b/backend/package/yuxi/channel/extensions/douyin/__init__.py new file mode 100644 index 00000000..c80eb7ae --- /dev/null +++ b/backend/package/yuxi/channel/extensions/douyin/__init__.py @@ -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()) diff --git a/backend/package/yuxi/channel/extensions/douyin/config.py b/backend/package/yuxi/channel/extensions/douyin/config.py new file mode 100644 index 00000000..9f5a78b8 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/douyin/config.py @@ -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"], + } diff --git a/backend/package/yuxi/channel/extensions/douyin/dedupe.py b/backend/package/yuxi/channel/extensions/douyin/dedupe.py new file mode 100644 index 00000000..81628bd9 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/douyin/dedupe.py @@ -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() diff --git a/backend/package/yuxi/channel/extensions/douyin/gateway.py b/backend/package/yuxi/channel/extensions/douyin/gateway.py new file mode 100644 index 00000000..43b29b79 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/douyin/gateway.py @@ -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") diff --git a/backend/package/yuxi/channel/extensions/douyin/media.py b/backend/package/yuxi/channel/extensions/douyin/media.py new file mode 100644 index 00000000..d7733e09 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/douyin/media.py @@ -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 diff --git a/backend/package/yuxi/channel/extensions/douyin/outbound.py b/backend/package/yuxi/channel/extensions/douyin/outbound.py new file mode 100644 index 00000000..145257a3 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/douyin/outbound.py @@ -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 diff --git a/backend/package/yuxi/channel/extensions/douyin/pairing.py b/backend/package/yuxi/channel/extensions/douyin/pairing.py new file mode 100644 index 00000000..cfe826c6 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/douyin/pairing.py @@ -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)) diff --git a/backend/package/yuxi/channel/extensions/douyin/plugin.json b/backend/package/yuxi/channel/extensions/douyin/plugin.json new file mode 100644 index 00000000..11970f6d --- /dev/null +++ b/backend/package/yuxi/channel/extensions/douyin/plugin.json @@ -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" +} \ No newline at end of file diff --git a/backend/package/yuxi/channel/extensions/douyin/security.py b/backend/package/yuxi/channel/extensions/douyin/security.py new file mode 100644 index 00000000..925b8e7b --- /dev/null +++ b/backend/package/yuxi/channel/extensions/douyin/security.py @@ -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) diff --git a/backend/package/yuxi/channel/extensions/douyin/status.py b/backend/package/yuxi/channel/extensions/douyin/status.py new file mode 100644 index 00000000..60f2eca2 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/douyin/status.py @@ -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 diff --git a/backend/package/yuxi/channel/extensions/douyin/streaming.py b/backend/package/yuxi/channel/extensions/douyin/streaming.py new file mode 100644 index 00000000..f95d2b65 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/douyin/streaming.py @@ -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[⏳]" diff --git a/backend/package/yuxi/channel/extensions/douyin/types.py b/backend/package/yuxi/channel/extensions/douyin/types.py new file mode 100644 index 00000000..a16ec7d2 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/douyin/types.py @@ -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 diff --git a/backend/package/yuxi/channel/extensions/douyin/webhook.py b/backend/package/yuxi/channel/extensions/douyin/webhook.py new file mode 100644 index 00000000..fb2e74c3 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/douyin/webhook.py @@ -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) diff --git a/backend/package/yuxi/channel/extensions/douyin/window.py b/backend/package/yuxi/channel/extensions/douyin/window.py new file mode 100644 index 00000000..fa8dea75 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/douyin/window.py @@ -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)