From 9ab904c5bda27ee29a7c14686d6a126785154bff Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Wed, 13 May 2026 16:07:32 +0800 Subject: [PATCH] =?UTF-8?q?refactor(discord-adapter):=20=E6=95=B4=E7=90=86?= =?UTF-8?q?=E4=BB=A3=E7=A0=81=E6=A0=BC=E5=BC=8F=E5=B9=B6=E4=BF=AE=E5=A4=8D?= =?UTF-8?q?Discord=E4=BA=A4=E4=BA=92=E7=9B=B8=E5=85=B3=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 本次提交包含多项优化与修复： 1. 清理导入语句并调整导入顺序，修复datetime导入顺序 2. 重构并发配置函数的换行格式，提升可读性 3. 修复poll模块的API调用方式，改用timedelta设置时长并调整答案添加逻辑 4. 重构REST调度器的协程执行逻辑，支持工厂模式创建协程 5. 新增discord webhook签名验证与交互类型定义 6. 新增线程管理相关的消息动作配置 7. 扩展安全策略的禁用策略支持，优化日志与函数参数格式 8. 重构事件队列的初始化参数格式，优化日志输出 9. 修复组件模块的类型错误：替换弃用的StringSelect为Select，新增提及选择器，修复按钮样式解析，添加视图超时处理 10. 新增丰富的Discord slash命令：模型管理组与用户信息查询命令 11. 新增多账号配置相关的数据类与工具方法 12. 新增Discord消息与交互事件的规范化处理逻辑 --- .../channels/adapters/discord/accounts.py | 106 +++ .../yuxi/channels/adapters/discord/adapter.py | 749 ++++++++++++++++-- .../yuxi/channels/adapters/discord/chunker.py | 4 +- .../channels/adapters/discord/commands.py | 73 ++ .../channels/adapters/discord/components.py | 37 +- .../channels/adapters/discord/event_queue.py | 19 +- .../channels/adapters/discord/formatter.py | 3 +- .../adapters/discord/message_actions.py | 35 + .../channels/adapters/discord/normalizer.py | 186 +++++ .../yuxi/channels/adapters/discord/poll.py | 12 +- .../adapters/discord/rest_scheduler.py | 10 +- .../channels/adapters/discord/security.py | 14 +- .../yuxi/channels/adapters/discord/send.py | 5 +- .../yuxi/channels/adapters/discord/webhook.py | 55 ++ 14 files changed, 1195 insertions(+), 113 deletions(-) create mode 100644 backend/package/yuxi/channels/adapters/discord/accounts.py create mode 100644 backend/package/yuxi/channels/adapters/discord/webhook.py diff --git a/backend/package/yuxi/channels/adapters/discord/accounts.py b/backend/package/yuxi/channels/adapters/discord/accounts.py new file mode 100644 index 00000000..af39204d --- /dev/null +++ b/backend/package/yuxi/channels/adapters/discord/accounts.py @@ -0,0 +1,106 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + + +@dataclass +class DiscordAccountConfig: + account_id: str + token: str = "" + token_file: str = "" + label: str = "" + enabled: bool = True + weight: int = 1 + priority: int = 0 + config: dict[str, Any] = field(default_factory=dict) + + @property + def resolved_token(self) -> str | None: + if self.token: + return self.token.strip().strip('"').strip("'") + if self.token_file: + import os + + if os.path.isfile(self.token_file): + try: + with open(self.token_file, encoding="utf-8") as f: + return f.read().strip() + except (OSError, UnicodeDecodeError): + pass + return None + + @property + def configured(self) -> bool: + return bool(self.resolved_token) + + +DEFAULT_ACCOUNT_ID = "default" + + +@dataclass +class DiscordMultiAccountConfig: + accounts: dict[str, DiscordAccountConfig] = field(default_factory=dict) + default_account_id: str = DEFAULT_ACCOUNT_ID + + @classmethod + def from_config(cls, config: dict | None) -> DiscordMultiAccountConfig: + if not config: + return cls() + + accounts_data = config.get("accounts", {}) + if not accounts_data: + token = config.get("token", "") + token_file = config.get("tokenFile", "") + if token or token_file: + account = DiscordAccountConfig( + account_id=DEFAULT_ACCOUNT_ID, + token=token, + token_file=token_file, + label="Default", + config=config, + ) + return cls( + accounts={DEFAULT_ACCOUNT_ID: account}, + default_account_id=DEFAULT_ACCOUNT_ID, + ) + + accounts: dict[str, DiscordAccountConfig] = {} + for aid, entry in accounts_data.items(): + if not isinstance(entry, dict): + continue + accounts[aid] = DiscordAccountConfig( + account_id=aid, + token=entry.get("token", ""), + token_file=entry.get("tokenFile", entry.get("token_file", "")), + label=entry.get("label", aid), + enabled=entry.get("enabled", True), + weight=entry.get("weight", 1), + priority=entry.get("priority", 0), + config=entry, + ) + + default_id = config.get("default_account_id", DEFAULT_ACCOUNT_ID) + if default_id not in accounts and accounts: + default_id = next(iter(accounts)) + + return cls(accounts=accounts, default_account_id=default_id) + + def resolve_account(self, account_id: str | None = None) -> DiscordAccountConfig | None: + if account_id and account_id in self.accounts: + acc = self.accounts[account_id] + return acc if acc.enabled else None + if self.default_account_id in self.accounts: + acc = self.accounts[self.default_account_id] + return acc if acc.enabled else None + return None + + def list_enabled(self) -> list[DiscordAccountConfig]: + return [a for a in self.accounts.values() if a.enabled and a.configured] + + def list_account_ids(self) -> list[str]: + return list(self.accounts.keys()) + + @property + def is_multi(self) -> bool: + return len(self.list_enabled()) > 1 diff --git a/backend/package/yuxi/channels/adapters/discord/adapter.py b/backend/package/yuxi/channels/adapters/discord/adapter.py index a798fbec..0f708d3f 100644 --- a/backend/package/yuxi/channels/adapters/discord/adapter.py +++ b/backend/package/yuxi/channels/adapters/discord/adapter.py @@ -11,12 +11,14 @@ import aiohttp import discord from yuxi.channels.base import BaseChannelAdapter -from yuxi.channels.infra.circuit_breaker import CircuitBreaker, CircuitBreakerOpenError +from yuxi.channels.capabilities import ChannelCapabilities from yuxi.channels.exceptions import ( ChannelAuthenticationError, ChannelException, ChannelNotConnectedError, ) +from yuxi.channels.infra.circuit_breaker import CircuitBreaker, CircuitBreakerOpenError +from yuxi.channels.meta import ChannelMeta from yuxi.channels.models import ( ChannelIdentity, ChannelMessage, @@ -28,21 +30,21 @@ from yuxi.channels.models import ( EventType, HealthStatus, ) -from yuxi.channels.capabilities import ChannelCapabilities -from yuxi.channels.meta import ChannelMeta from yuxi.channels.registry import register_builtin_adapter from yuxi.utils.logging_config import logger +from .accounts import DiscordAccountConfig, DiscordMultiAccountConfig from .commands import register_slash_commands from .dedupe import ReplayGuard from .formatter import DiscordMessageFormatter from .message_actions import DISCORD_MESSAGE_ACTIONS from .normalizer import DiscordMessageNormalizer +from .poll import create_poll from .probe import check_health, probe_token from .security import DiscordSecurityPolicy from .send import chunk_text, send_media_file, send_stream_edit, send_with_retry -from .poll import create_poll from .session import parse_chat_id +from .webhook import InteractionType, handle_ping_interaction, verify_ed25519_signature GATEWAY_CLOSE_CODES: dict[int, tuple[str, bool]] = { 4000: ("Unknown Error", True), @@ -111,13 +113,13 @@ class DiscordAdapter(BaseChannelAdapter): "mute": "mute", } - webhook_path: ClassVar[str | None] = None + webhook_path: ClassVar[str | None] = "discord" def __init__(self, config: dict[str, Any] | None = None): super().__init__(config) self._status = ChannelStatus.DISCONNECTED - self._client: discord.Client | None = None - self._connect_task: asyncio.Task | None = None + self._clients: dict[str, discord.Client] = {} + self._connect_tasks: dict[str, asyncio.Task] = {} self._ready_event = asyncio.Event() self._ready_at: float | None = None self._circuit_breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60) @@ -140,6 +142,26 @@ class DiscordAdapter(BaseChannelAdapter): self._error_count: int = 0 self._chunk_mode: str = self.config.get("chunkMode", "newline") self._actions = {k for k, v in DISCORD_MESSAGE_ACTIONS.items() if v.get("status") == "implemented"} + self._account_configs = DiscordMultiAccountConfig.from_config(config or {}) + self._account_route: dict[str, str] = {} + self._ready_by_account: dict[str, asyncio.Event] = {} + self._webhook_public_key: str = self.config.get("webhook_public_key", "") + self._webhook_enabled: bool = self.config.get("webhook_enabled", False) + self._webhook_signature_tolerance: int = self.config.get("webhook_signature_tolerance_seconds", 5) + self._privileged_intents_enabled: bool = self.config.get("privileged_intents_enabled", False) + + def _resolve_intents(self) -> discord.Intents: + intents = discord.Intents( + guilds=True, + messages=True, + message_content=True, + guild_messages=True, + direct_messages=True, + ) + if self._privileged_intents_enabled: + intents.members = True + intents.presences = True + return intents def validate_capabilities(self) -> dict[str, Any]: cap_actions = set(self.capabilities.supported_actions()) @@ -154,6 +176,161 @@ class DiscordAdapter(BaseChannelAdapter): issues.append(f"Implemented but not declared: {sorted(extra)}") return {"valid": len(issues) == 0, "issues": issues} + @property + def _client(self) -> discord.Client | None: + default_acc = self._account_configs.default_account_id + return self._clients.get(default_acc) + + @_client.setter + def _client(self, client: discord.Client | None) -> None: + default_acc = self._account_configs.default_account_id + if default_acc: + if client is None: + self._clients.pop(default_acc, None) + else: + self._clients[default_acc] = client + + @property + def _is_multi_account(self) -> bool: + return self._account_configs.is_multi + + def _get_client(self, account_id: str | None = None) -> discord.Client | None: + aid = account_id or self._account_configs.default_account_id + return self._clients.get(aid) + + def _get_any_client(self) -> discord.Client | None: + for client in self._clients.values(): + return client + return None + + def list_account_ids(self) -> list[str]: + return self._account_configs.list_account_ids() + + def resolve_account(self, account_id: str) -> dict | None: + acc = self._account_configs.resolve_account(account_id) + if acc is None: + return None + return {"account_id": acc.account_id, "label": acc.label, "enabled": acc.enabled} + + async def add_account(self, account_id: str, config: dict[str, Any]) -> bool: + existing = self._account_configs.accounts.get(account_id) + if existing and existing.enabled: + logger.warning(f"[Discord] Account '{account_id}' already exists and is enabled") + return False + + new_acc = DiscordAccountConfig( + account_id=account_id, + token=config.get("token", ""), + token_file=config.get("token_file", ""), + label=config.get("label", account_id), + enabled=True, + config=config, + ) + self._account_configs.accounts[account_id] = new_acc + if not self._account_configs.default_account_id: + self._account_configs.default_account_id = account_id + + if self._status == ChannelStatus.CONNECTED: + token = new_acc.resolved_token + if not token: + logger.warning(f"[Discord] Account '{account_id}' has no valid token, skip connect") + return True + + intents = self._resolve_intents() + client = discord.Client(intents=intents, max_messages=1000) + self._clients[account_id] = client + self._ready_by_account[account_id] = asyncio.Event() + self._register_events_for_client(client, account_id) + task = asyncio.create_task(client.start(token)) + self._connect_tasks[account_id] = task + logger.info(f"[Discord] Dynamically connected account '{account_id}'") + + logger.info(f"[Discord] Added account '{account_id}'") + return True + + async def remove_account(self, account_id: str) -> bool: + acc = self._account_configs.accounts.pop(account_id, None) + if acc is None: + logger.warning(f"[Discord] Account '{account_id}' not found") + return False + + client = self._clients.pop(account_id, None) + task = self._connect_tasks.pop(account_id, None) + ready_evt = self._ready_by_account.pop(account_id, None) + + if task and not task.done(): + task.cancel() + try: + await task + except (asyncio.CancelledError, Exception): + pass + + if client: + try: + await client.close() + except Exception: + pass + + if ready_evt: + ready_evt.clear() + + gw_key = f"gateway_session:{account_id}" + await self.state_set(gw_key, None, namespace="connection") + + if self._account_configs.default_account_id == account_id: + remaining = self._account_configs.list_enabled() + self._account_configs.default_account_id = remaining[0].account_id if remaining else "" + + logger.info(f"[Discord] Removed account '{account_id}'") + return True + + async def delete_account(self, account_id: str) -> bool: + return await self.remove_account(account_id) + + async def update_account(self, account_id: str, config: dict[str, Any]) -> bool: + acc = self._account_configs.accounts.get(account_id) + if acc is None: + logger.warning(f"[Discord] Account '{account_id}' not found") + return False + + old_token = acc.resolved_token + acc.token = config.get("token", acc.token) + acc.token_file = config.get("token_file", acc.token_file) + acc.label = config.get("label", acc.label) + acc.enabled = config.get("enabled", acc.enabled) + acc.config = config + + new_token = acc.resolved_token + if new_token and new_token != old_token and self._status == ChannelStatus.CONNECTED: + old_client = self._clients.pop(account_id, None) + old_task = self._connect_tasks.pop(account_id, None) + if old_task and not old_task.done(): + old_task.cancel() + try: + await old_task + except (asyncio.CancelledError, Exception): + pass + if old_client: + try: + await old_client.close() + except Exception: + pass + + intents = self._resolve_intents() + new_client = discord.Client(intents=intents, max_messages=1000) + self._clients[account_id] = new_client + self._ready_by_account[account_id] = asyncio.Event() + self._register_events_for_client(new_client, account_id) + new_task = asyncio.create_task(new_client.start(new_token)) + self._connect_tasks[account_id] = new_task + logger.info(f"[Discord] Reconnected account '{account_id}' with new token") + + logger.info(f"[Discord] Updated account '{account_id}'") + return True + + def _resolve_guild_account(self, guild_id: int) -> str: + return self._account_route.get(str(guild_id), self._account_configs.default_account_id) + async def connect(self) -> None: if self._status == ChannelStatus.CONNECTED: return @@ -164,48 +341,70 @@ class DiscordAdapter(BaseChannelAdapter): self._reconnect_attempts = 0 logger.info(f"[Discord] Starting channel '{self.config.get('name', self.channel_id)}'") - token = self._resolve_token() - if not token: - raise ChannelAuthenticationError() + enabled_accounts = self._account_configs.list_enabled() + if not enabled_accounts: + raise ChannelAuthenticationError("No enabled Discord accounts configured") self._ready_event.clear() delay_ms = self.config.get("identify_delay_ms", 0) - if delay_ms > 0: - await asyncio.sleep(delay_ms / 1000) - intents = discord.Intents( - guilds=True, - messages=True, - message_content=True, - guild_messages=True, - direct_messages=True, - ) - self._client = discord.Client(intents=intents, max_messages=1000) - self._register_events() + for idx, account_config in enumerate(enabled_accounts): + token = account_config.resolved_token + if not token: + logger.warning(f"[Discord] Account '{account_config.account_id}' has no valid token, skipping") + continue + + if idx > 0 and delay_ms > 0: + await asyncio.sleep(delay_ms / 1000) + + intents = self._resolve_intents() + client = discord.Client(intents=intents, max_messages=1000) + self._clients[account_config.account_id] = client + + gw_session = await self.state_get(f"gateway_session:{account_config.account_id}", namespace="connection") + if gw_session and isinstance(gw_session, dict): + conn = getattr(client, "_connection", None) + if conn: + conn._session_id = gw_session.get("session_id") + seq = gw_session.get("sequence") + conn.sequence = seq if seq is not None else 0 + logger.info( + f"[Discord] Restored gateway session for '{account_config.account_id}', will attempt resume" + ) + + self._ready_by_account[account_config.account_id] = asyncio.Event() + self._register_events_for_client(client, account_config.account_id) + + task = asyncio.create_task(client.start(token)) + self._connect_tasks[account_config.account_id] = task - self._connect_task = asyncio.create_task(self._client.start(token)) try: await asyncio.wait_for(self._ready_event.wait(), timeout=30.0) except TimeoutError: self._status = ChannelStatus.ERROR raise ChannelException("Discord READY timeout", retryable=True) - self._normalizer = DiscordMessageNormalizer(bot_user_id=self._client.user.id) + primary_client = self._client + if primary_client and primary_client.user: + self._normalizer = DiscordMessageNormalizer(bot_user_id=primary_client.user.id) - asyncio.create_task( - register_slash_commands( - self._client, - self._message_handler, - self.fetch_channel_history, - get_status_fn=self._get_health_status, + self._account_configs = DiscordMultiAccountConfig.from_config(self.config) + + asyncio.create_task( + register_slash_commands( + primary_client, + self._message_handler, + self.fetch_channel_history, + get_status_fn=self._get_health_status, + ) ) - ) self._status = ChannelStatus.CONNECTED + connected_count = len([c for c in self._clients.values()]) logger.info( - f"[Discord] Bot '{self._client.user}' connected " - f"({len(self._client.guilds)} guilds, latency: {self._client.latency * 1000:.0f}ms)" + f"[Discord] {connected_count} bot(s) connected, " + f"primary: '{primary_client.user if primary_client else 'N/A'}'" ) async def disconnect(self) -> None: @@ -214,6 +413,15 @@ class DiscordAdapter(BaseChannelAdapter): logger.info(f"[Discord] Stopping channel '{self.config.get('name', self.channel_id)}'") + for account_id, client in list(self._clients.items()): + conn = getattr(client, "_connection", None) + if conn: + gw_data = { + "session_id": getattr(conn, "_session_id", None), + "sequence": getattr(conn, "sequence", None), + } + await self.state_set(f"gateway_session:{account_id}", gw_data, namespace="connection") + self._should_reconnect = False self._status = ChannelStatus.DISCONNECTED self._ready_event.clear() @@ -226,20 +434,22 @@ class DiscordAdapter(BaseChannelAdapter): pass self._reconnect_task = None - if self._client: + for account_id, client in list(self._clients.items()): try: - await self._client.close() + await client.close() except Exception: - logger.debug("[Discord] Error closing client", exc_info=True) + logger.debug(f"[Discord] Error closing client '{account_id}'", exc_info=True) - if self._connect_task and not self._connect_task.done(): - self._connect_task.cancel() - try: - await self._connect_task - except (asyncio.CancelledError, Exception): - pass + for account_id, task in list(self._connect_tasks.items()): + if not task.done(): + task.cancel() + try: + await task + except (asyncio.CancelledError, Exception): + pass - self._client = None + self._clients.clear() + self._connect_tasks.clear() self._normalizer = None async def send(self, response: ChannelResponse, silent: bool = False) -> DeliveryResult: @@ -270,10 +480,7 @@ class DiscordAdapter(BaseChannelAdapter): chunk_response = copy.copy(response) chunk_response.content = chunk if original_meta: - chunk_response.metadata = { - k: v for k, v in original_meta.items() - if k not in embed_keys - } + chunk_response.metadata = {k: v for k, v in original_meta.items() if k not in embed_keys} chunk_result = await self._send_single(chunk_response, silent, channel) if not chunk_result.success: logger.warning(f"[Discord] Chunk send failed: {chunk_result.error}") @@ -952,6 +1159,220 @@ class DiscordAdapter(BaseChannelAdapter): except Exception as e: return DeliveryResult(success=False, error=str(e)) + async def unarchive_thread(self, chat_id: str) -> DeliveryResult: + if not self._client: + return DeliveryResult(success=False, error="Not connected") + + kind, target_id = parse_chat_id(chat_id) + if kind != "thread": + return DeliveryResult(success=False, error="Not a thread") + + thread = self._client.get_channel(target_id) + if not thread or not isinstance(thread, discord.Thread): + try: + thread = await self._client.fetch_channel(target_id) + except (discord.NotFound, discord.Forbidden, discord.HTTPException): + return DeliveryResult(success=False, error="Thread not found") + + if not isinstance(thread, discord.Thread): + return DeliveryResult(success=False, error="Channel is not a thread") + + try: + await thread.edit(archived=False) + return DeliveryResult( + success=True, + message_id=chat_id, + metadata={"archived": False}, + ) + except discord.Forbidden: + return DeliveryResult(success=False, error="Forbidden") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def unlock_thread(self, chat_id: str) -> DeliveryResult: + if not self._client: + return DeliveryResult(success=False, error="Not connected") + + kind, target_id = parse_chat_id(chat_id) + if kind != "thread": + return DeliveryResult(success=False, error="Not a thread") + + thread = self._client.get_channel(target_id) + if not thread or not isinstance(thread, discord.Thread): + try: + thread = await self._client.fetch_channel(target_id) + except (discord.NotFound, discord.Forbidden, discord.HTTPException): + return DeliveryResult(success=False, error="Thread not found") + + if not isinstance(thread, discord.Thread): + return DeliveryResult(success=False, error="Channel is not a thread") + + try: + await thread.edit(locked=False) + return DeliveryResult( + success=True, + message_id=chat_id, + metadata={"locked": False}, + ) + except discord.Forbidden: + return DeliveryResult(success=False, error="Forbidden") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def join_thread(self, chat_id: str) -> DeliveryResult: + if not self._client: + return DeliveryResult(success=False, error="Not connected") + + kind, target_id = parse_chat_id(chat_id) + if kind != "thread": + return DeliveryResult(success=False, error="Not a thread") + + thread = self._client.get_channel(target_id) + if not thread or not isinstance(thread, discord.Thread): + try: + thread = await self._client.fetch_channel(target_id) + except (discord.NotFound, discord.Forbidden, discord.HTTPException): + return DeliveryResult(success=False, error="Thread not found") + + if not isinstance(thread, discord.Thread): + return DeliveryResult(success=False, error="Channel is not a thread") + + try: + await thread.join() + return DeliveryResult( + success=True, + message_id=chat_id, + metadata={"joined": True}, + ) + except discord.Forbidden: + return DeliveryResult(success=False, error="Forbidden") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def leave_thread(self, chat_id: str) -> DeliveryResult: + if not self._client: + return DeliveryResult(success=False, error="Not connected") + + kind, target_id = parse_chat_id(chat_id) + if kind != "thread": + return DeliveryResult(success=False, error="Not a thread") + + thread = self._client.get_channel(target_id) + if not thread or not isinstance(thread, discord.Thread): + try: + thread = await self._client.fetch_channel(target_id) + except (discord.NotFound, discord.Forbidden, discord.HTTPException): + return DeliveryResult(success=False, error="Thread not found") + + if not isinstance(thread, discord.Thread): + return DeliveryResult(success=False, error="Channel is not a thread") + + try: + await thread.leave() + return DeliveryResult( + success=True, + message_id=chat_id, + metadata={"left": True}, + ) + except discord.Forbidden: + return DeliveryResult(success=False, error="Forbidden") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def add_thread_member(self, chat_id: str, user_id: int) -> DeliveryResult: + if not self._client: + return DeliveryResult(success=False, error="Not connected") + + kind, target_id = parse_chat_id(chat_id) + if kind != "thread": + return DeliveryResult(success=False, error="Not a thread") + + thread = self._client.get_channel(target_id) + if not thread or not isinstance(thread, discord.Thread): + try: + thread = await self._client.fetch_channel(target_id) + except (discord.NotFound, discord.Forbidden, discord.HTTPException): + return DeliveryResult(success=False, error="Thread not found") + + if not isinstance(thread, discord.Thread): + return DeliveryResult(success=False, error="Channel is not a thread") + + try: + await thread.add_user(discord.Object(id=user_id)) + return DeliveryResult( + success=True, + message_id=chat_id, + metadata={"member_id": str(user_id), "added": True}, + ) + except discord.Forbidden: + return DeliveryResult(success=False, error="Forbidden") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def remove_thread_member(self, chat_id: str, user_id: int) -> DeliveryResult: + if not self._client: + return DeliveryResult(success=False, error="Not connected") + + kind, target_id = parse_chat_id(chat_id) + if kind != "thread": + return DeliveryResult(success=False, error="Not a thread") + + thread = self._client.get_channel(target_id) + if not thread or not isinstance(thread, discord.Thread): + try: + thread = await self._client.fetch_channel(target_id) + except (discord.NotFound, discord.Forbidden, discord.HTTPException): + return DeliveryResult(success=False, error="Thread not found") + + if not isinstance(thread, discord.Thread): + return DeliveryResult(success=False, error="Channel is not a thread") + + try: + await thread.remove_user(discord.Object(id=user_id)) + return DeliveryResult( + success=True, + message_id=chat_id, + metadata={"member_id": str(user_id), "removed": True}, + ) + except discord.Forbidden: + return DeliveryResult(success=False, error="Forbidden") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def list_active_threads(self, guild_id: int) -> DeliveryResult: + if not self._client: + return DeliveryResult(success=False, error="Not connected") + + guild = self._client.get_guild(guild_id) + if not guild: + try: + guild = await self._client.fetch_guild(guild_id) + except (discord.NotFound, discord.Forbidden, discord.HTTPException): + return DeliveryResult(success=False, error="Guild not found") + + try: + active_threads: list[dict[str, Any]] = [] + for thread in guild.threads: + active_threads.append( + { + "id": str(thread.id), + "name": thread.name, + "parent_id": str(thread.parent_id) if thread.parent_id else None, + "member_count": thread.member_count, + "message_count": getattr(thread, "message_count", 0), + "archived": thread.archived, + "locked": getattr(thread, "locked", False), + } + ) + return DeliveryResult( + success=True, + metadata={"threads": active_threads, "count": len(active_threads)}, + ) + except discord.Forbidden: + return DeliveryResult(success=False, error="Forbidden") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + async def send_poll( self, chat_id: str, @@ -1221,6 +1642,9 @@ class DiscordAdapter(BaseChannelAdapter): channel_type: str = "text", parent_id: str | None = None, topic: str | None = None, + bitrate: int | None = None, + user_limit: int | None = None, + permission_overwrites: list[dict[str, Any]] | None = None, ) -> DeliveryResult: if not self._client: return DeliveryResult(success=False, error="Not connected") @@ -1252,9 +1676,28 @@ class DiscordAdapter(BaseChannelAdapter): if topic and channel_type in ("text", "announcement"): kwargs["topic"] = topic + if channel_type == "voice" and bitrate is not None: + kwargs["bitrate"] = max(8000, min(bitrate, 384000)) + if channel_type == "voice" and user_limit is not None: + kwargs["user_limit"] = max(0, min(user_limit, 99)) + + if permission_overwrites: + overwrites: dict[discord.abc.Snowflake, discord.PermissionOverwrite] = {} + for ow in permission_overwrites: + target_id = int(ow.get("id", 0)) + ow_type = ow.get("type", "role") + allow = discord.Permissions(ow.get("allow", 0)) + deny = discord.Permissions(ow.get("deny", 0)) + if ow_type == "member": + target = guild.get_member(target_id) or discord.Object(id=target_id) + else: + target = guild.get_role(target_id) or discord.Object(id=target_id) + overwrites[target] = discord.PermissionOverwrite.from_pair(allow, deny) + kwargs["overwrites"] = overwrites + try: if ct == discord.ChannelType.category: - channel = await guild.create_category(name=name) + channel = await guild.create_category(name=name, overwrites=kwargs.get("overwrites")) elif ct == discord.ChannelType.voice: channel = await guild.create_voice_channel(**kwargs) else: @@ -1292,6 +1735,9 @@ class DiscordAdapter(BaseChannelAdapter): name: str | None = None, topic: str | None = None, position: int | None = None, + bitrate: int | None = None, + user_limit: int | None = None, + permission_overwrites: list[dict[str, Any]] | None = None, ) -> DeliveryResult: if not self._client: return DeliveryResult(success=False, error="Not connected") @@ -1312,6 +1758,25 @@ class DiscordAdapter(BaseChannelAdapter): kwargs["topic"] = topic if position is not None: kwargs["position"] = position + if bitrate is not None and isinstance(channel, discord.VoiceChannel): + kwargs["bitrate"] = max(8000, min(bitrate, 384000)) + if user_limit is not None and isinstance(channel, discord.VoiceChannel): + kwargs["user_limit"] = max(0, min(user_limit, 99)) + + if permission_overwrites is not None and hasattr(channel, "guild"): + guild = channel.guild + overwrites: dict[discord.abc.Snowflake, discord.PermissionOverwrite] = {} + for ow in permission_overwrites: + target_id = int(ow.get("id", 0)) + ow_type = ow.get("type", "role") + allow = discord.Permissions(ow.get("allow", 0)) + deny = discord.Permissions(ow.get("deny", 0)) + if ow_type == "member": + target = guild.get_member(target_id) or discord.Object(id=target_id) + else: + target = guild.get_role(target_id) or discord.Object(id=target_id) + overwrites[target] = discord.PermissionOverwrite.from_pair(allow, deny) + kwargs["overwrites"] = overwrites if not kwargs: return DeliveryResult(success=False, error="No fields to update") @@ -1541,36 +2006,38 @@ class DiscordAdapter(BaseChannelAdapter): logger.warning(f"[Discord] Failed to read token file: {path}") return "" - def _register_events(self) -> None: - if not self._client: - return - - @self._client.event + def _register_events_for_client(self, client: discord.Client, account_id: str) -> None: + @client.event async def on_ready(): self._ready_at = time.time() self._ready_event.set() + ready_evt = self._ready_by_account.get(account_id) + if ready_evt: + ready_evt.set() self._consecutive_401_count = 0 self._401_backoff_until = None - logger.info(f"[Discord] READY: {self._client.user} (latency: {self._client.latency * 1000:.0f}ms)") + logger.info(f"[Discord/{account_id}] READY: {client.user} (latency: {client.latency * 1000:.0f}ms)") - @self._client.event + @client.event async def on_disconnect(): self._status = ChannelStatus.RECONNECTING self._reconnect_attempts += 1 self._last_disconnect_at = time.time() - close_code = self._capture_close_code() + close_code = self._capture_close_code_for_client(client) if close_code is not None: self._last_close_code = close_code desc, recoverable = GATEWAY_CLOSE_CODES.get(close_code, (f"Unknown code {close_code}", True)) logger.warning( - f"[Discord] Gateway disconnected with close code {close_code} ({desc}), " + f"[Discord/{account_id}] Gateway disconnected with close code {close_code} ({desc}), " f"recoverable={recoverable} (attempt {self._reconnect_attempts})" ) if not recoverable: self._should_reconnect = False self._status = ChannelStatus.ERROR - logger.error(f"[Discord] Non-recoverable close code {close_code} ({desc}), reconnect disabled") + logger.error( + f"[Discord/{account_id}] Non-recoverable close code {close_code} ({desc}), reconnect disabled" + ) return if close_code == 4004: @@ -1578,27 +2045,28 @@ class DiscordAdapter(BaseChannelAdapter): backoff_seconds = min(300 * self._consecutive_401_count, 1800) self._401_backoff_until = time.time() + backoff_seconds logger.warning( - f"[Discord] 401 Authentication Failed detected, " + f"[Discord/{account_id}] 401 Authentication Failed detected, " f"backoff for {backoff_seconds}s (consecutive 401s: {self._consecutive_401_count})" ) else: logger.warning( - f"[Discord] Gateway disconnected (attempt {self._reconnect_attempts}), scheduling reconnect..." + f"[Discord/{account_id}] Gateway disconnected " + f"(attempt {self._reconnect_attempts}), scheduling reconnect..." ) if self._should_reconnect and not (self._reconnect_task and not self._reconnect_task.done()): self._reconnect_task = asyncio.create_task(self._reconnect()) - @self._client.event + @client.event async def on_resumed(): self._status = ChannelStatus.CONNECTED self._reconnect_attempts = 0 self._ready_at = time.time() logger.info("[Discord] Gateway session resumed") - @self._client.event + @client.event async def on_message(message: discord.Message): - if message.author == self._client.user: + if message.author == client.user: return if not message.content and not message.attachments: return @@ -1650,7 +2118,7 @@ class DiscordAdapter(BaseChannelAdapter): except Exception: logger.exception("[Discord] Error handling on_message") - @self._client.event + @client.event async def on_message_edit(after: discord.Message): if after.author == self._client.user: return @@ -1665,7 +2133,7 @@ class DiscordAdapter(BaseChannelAdapter): except Exception: logger.exception("[Discord] Error handling on_message_edit") - @self._client.event + @client.event async def on_raw_message_delete(payload: discord.RawMessageDeleteEvent): if self._message_handler is None: return @@ -1689,7 +2157,7 @@ class DiscordAdapter(BaseChannelAdapter): except Exception: logger.exception("[Discord] Error handling on_raw_message_delete") - @self._client.event + @client.event async def on_guild_join(guild: discord.Guild): if self._message_handler is None: return @@ -1711,7 +2179,7 @@ class DiscordAdapter(BaseChannelAdapter): except Exception: logger.exception("[Discord] Error handling on_guild_join") - @self._client.event + @client.event async def on_guild_remove(guild: discord.Guild): if self._message_handler is None: return @@ -1733,7 +2201,7 @@ class DiscordAdapter(BaseChannelAdapter): except Exception: logger.exception("[Discord] Error handling on_guild_remove") - @self._client.event + @client.event async def on_raw_bulk_message_delete(payload: discord.RawBulkMessageDeleteEvent): if self._message_handler is None: return @@ -1743,7 +2211,7 @@ class DiscordAdapter(BaseChannelAdapter): except Exception: logger.exception("[Discord] Error handling on_raw_bulk_message_delete") - @self._client.event + @client.event async def on_raw_reaction_add(payload: discord.RawReactionActionEvent): if self._message_handler is None: return @@ -1753,7 +2221,7 @@ class DiscordAdapter(BaseChannelAdapter): except Exception: logger.exception("[Discord] Error handling on_raw_reaction_add") - @self._client.event + @client.event async def on_raw_reaction_remove(payload: discord.RawReactionActionEvent): if self._message_handler is None: return @@ -1763,7 +2231,7 @@ class DiscordAdapter(BaseChannelAdapter): except Exception: logger.exception("[Discord] Error handling on_raw_reaction_remove") - @self._client.event + @client.event async def on_typing(channel, user, when): if self._message_handler is None: return @@ -1773,7 +2241,7 @@ class DiscordAdapter(BaseChannelAdapter): except Exception: logger.exception("[Discord] Error handling on_typing") - @self._client.event + @client.event async def on_member_join(member: discord.Member): if self._message_handler is None: return @@ -1783,7 +2251,7 @@ class DiscordAdapter(BaseChannelAdapter): except Exception: logger.exception("[Discord] Error handling on_member_join") - @self._client.event + @client.event async def on_member_remove(member: discord.Member): if self._message_handler is None: return @@ -1793,7 +2261,7 @@ class DiscordAdapter(BaseChannelAdapter): except Exception: logger.exception("[Discord] Error handling on_member_remove") - @self._client.event + @client.event async def on_member_update(before: discord.Member, after: discord.Member): if self._message_handler is None: return @@ -1803,7 +2271,7 @@ class DiscordAdapter(BaseChannelAdapter): except Exception: logger.exception("[Discord] Error handling on_member_update") - @self._client.event + @client.event async def on_guild_role_create(role: discord.Role): if self._message_handler is None: return @@ -1813,7 +2281,7 @@ class DiscordAdapter(BaseChannelAdapter): except Exception: logger.exception("[Discord] Error handling on_guild_role_create") - @self._client.event + @client.event async def on_guild_role_delete(role: discord.Role): if self._message_handler is None: return @@ -1823,7 +2291,7 @@ class DiscordAdapter(BaseChannelAdapter): except Exception: logger.exception("[Discord] Error handling on_guild_role_delete") - @self._client.event + @client.event async def on_guild_role_update(before: discord.Role, after: discord.Role): if self._message_handler is None: return @@ -1833,6 +2301,69 @@ class DiscordAdapter(BaseChannelAdapter): except Exception: logger.exception("[Discord] Error handling on_guild_role_update") + @client.event + async def on_interaction(interaction: discord.Interaction): + if self._message_handler is None: + return + + self._last_event_at = time.time() + + try: + if interaction.type == discord.InteractionType.ping: + await interaction.response.pong() + return + + channel_msg = DiscordMessageNormalizer.normalize_interaction(interaction) + + if interaction.type == discord.InteractionType.application_command: + logger.info( + f"[Discord/{account_id}] Slash command: {interaction.data.get('name')} by {interaction.user}" + ) + elif interaction.type == discord.InteractionType.component: + logger.debug( + f"[Discord/{account_id}] Component interaction: " + f"{interaction.data.get('custom_id')} by {interaction.user}" + ) + elif interaction.type == discord.InteractionType.modal_submit: + logger.debug( + f"[Discord/{account_id}] Modal submit: " + f"{interaction.data.get('custom_id')} by {interaction.user}" + ) + + await self._message_handler(channel_msg) + except Exception: + logger.exception("[Discord] Error handling on_interaction") + + @client.event + async def on_guild_channel_create(channel: discord.abc.GuildChannel): + if self._message_handler is None: + return + channel_msg = DiscordMessageNormalizer.normalize_channel_event(channel, EventType.CHANNEL_CREATED) + try: + await self._message_handler(channel_msg) + except Exception: + logger.exception("[Discord] Error handling on_guild_channel_create") + + @client.event + async def on_guild_channel_update(before: discord.abc.GuildChannel, after: discord.abc.GuildChannel): + if self._message_handler is None: + return + channel_msg = DiscordMessageNormalizer.normalize_channel_update(before, after) + try: + await self._message_handler(channel_msg) + except Exception: + logger.exception("[Discord] Error handling on_guild_channel_update") + + @client.event + async def on_guild_channel_delete(channel: discord.abc.GuildChannel): + if self._message_handler is None: + return + channel_msg = DiscordMessageNormalizer.normalize_channel_event(channel, EventType.CHANNEL_DELETED) + try: + await self._message_handler(channel_msg) + except Exception: + logger.exception("[Discord] Error handling on_guild_channel_delete") + async def _resolve_raw_chat_id(self, payload: discord.RawMessageDeleteEvent) -> str: if payload.guild_id: return f"guild_{payload.guild_id}_channel_{payload.channel_id}" @@ -1863,17 +2394,18 @@ class DiscordAdapter(BaseChannelAdapter): return metadata - async def _resolve_discord_channel(self, chat_id: str): - if not self._client: + async def _resolve_discord_channel(self, chat_id: str, account_id: str | None = None): + client = self._get_client(account_id) + if not client: return None kind, target_id = parse_chat_id(chat_id) if kind == "dm": - user = self._client.get_user(target_id) + user = client.get_user(target_id) if not user: try: - user = await self._client.fetch_user(target_id) + user = await client.fetch_user(target_id) except discord.NotFound: return None except (discord.HTTPException, discord.Forbidden) as e: @@ -1888,7 +2420,7 @@ class DiscordAdapter(BaseChannelAdapter): return None if kind in ("guild_channel", "thread"): - return self._client.get_channel(target_id) + return client.get_channel(target_id) return None @@ -1992,10 +2524,7 @@ class DiscordAdapter(BaseChannelAdapter): return int(val) return None - def _capture_close_code(self) -> int | None: - client = self._client - if client is None: - return None + def _capture_close_code_for_client(self, client: discord.Client) -> int | None: try: ws = getattr(getattr(client, "_connection", None), "_websocket", None) if ws is not None: @@ -2017,3 +2546,53 @@ class DiscordAdapter(BaseChannelAdapter): "base_delay": float(retry.get("minDelayMs", 1000)) / 1000.0, "jitter": float(retry.get("jitter", 0.1)), } + + async def verify_webhook_signature(self, headers: dict[str, str], body: bytes) -> bool: + if not self._webhook_enabled or not self._webhook_public_key: + return True + + signature = headers.get("x-signature-ed25519", "") + timestamp = headers.get("x-signature-timestamp", "") + + if not signature or not timestamp: + logger.warning("[Discord/Webhook] Missing signature headers") + return False + + try: + ts = int(timestamp) + current_time = int(time.time()) + if abs(current_time - ts) > self._webhook_signature_tolerance: + logger.warning( + f"[Discord/Webhook] Timestamp out of tolerance: " + f"delta={abs(current_time - ts)}s > {self._webhook_signature_tolerance}s" + ) + return False + except ValueError: + logger.warning("[Discord/Webhook] Invalid timestamp format") + return False + + body_str = body.decode("utf-8") if isinstance(body, bytes) else str(body) + valid = verify_ed25519_signature( + self._webhook_public_key, + signature, + timestamp, + body_str, + ) + if not valid: + logger.warning("[Discord/Webhook] Ed25519 signature verification failed") + return valid + + async def handle_webhook(self, body_data: dict) -> None: + interaction_type = body_data.get("type", 0) + + if interaction_type == InteractionType.PING: + return handle_ping_interaction() + + normalized_msg = DiscordMessageNormalizer.normalize_interaction_raw(body_data) + + if self._message_handler: + try: + await self._message_handler(normalized_msg) + except Exception: + logger.exception("[Discord/Webhook] Error handling webhook interaction") + return None diff --git a/backend/package/yuxi/channels/adapters/discord/chunker.py b/backend/package/yuxi/channels/adapters/discord/chunker.py index e70d4451..f9e2f597 100644 --- a/backend/package/yuxi/channels/adapters/discord/chunker.py +++ b/backend/package/yuxi/channels/adapters/discord/chunker.py @@ -1,10 +1,8 @@ from __future__ import annotations import re -from typing import Literal - from collections.abc import Iterator - +from typing import Literal ChunkMode = Literal["newline", "length"] diff --git a/backend/package/yuxi/channels/adapters/discord/commands.py b/backend/package/yuxi/channels/adapters/discord/commands.py index eddf32e0..90ba4ca5 100644 --- a/backend/package/yuxi/channels/adapters/discord/commands.py +++ b/backend/package/yuxi/channels/adapters/discord/commands.py @@ -174,6 +174,79 @@ async def register_slash_commands( else: await interaction.followup.send(f"切换失败：模型 `{model_name}` 不可用") + model_group = app_commands.Group(name="ai-model", description="AI 模型管理") + + @model_group.command(name="list", description="查看当前可用的 AI 模型列表") + async def model_list(interaction: discord.Interaction): + await interaction.response.defer(ephemeral=True) + + if get_models_fn is None: + await interaction.followup.send("模型管理功能未启用") + return + + models = await get_models_fn() + if models: + model_list = "\n".join(f"• `{m}`" for m in models) + await interaction.followup.send(f"**可用模型:**\n{model_list}") + else: + await interaction.followup.send("当前无可用模型") + + async def model_autocomplete( + _interaction: discord.Interaction, + current: str, + ) -> list[app_commands.Choice[str]]: + if get_models_fn is None: + return [] + try: + models = await get_models_fn() + except Exception: + return [] + return [app_commands.Choice(name=m, value=m) for m in models if current.lower() in m.lower()][:25] + + @model_group.command(name="switch", description="切换当前使用的 AI 模型") + @app_commands.describe(model_name="要切换到的模型名称") + @app_commands.autocomplete(model_name=model_autocomplete) + async def model_switch(interaction: discord.Interaction, model_name: str): + await interaction.response.defer(ephemeral=True) + + if set_model_fn is None: + await interaction.followup.send("模型管理功能未启用") + return + + ok = await set_model_fn(str(interaction.user.id), model_name) + if ok: + await interaction.followup.send(f"已切换模型到 `{model_name}`") + else: + await interaction.followup.send(f"切换失败：模型 `{model_name}` 不可用") + + tree.add_command(model_group) + + @tree.command(name="user-info", description="查看指定用户的 Discord 信息") + @app_commands.describe(user="要查询的用户") + async def user_info(interaction: discord.Interaction, user: discord.User): + await interaction.response.defer(ephemeral=True) + + lines = [ + "**用户信息:**", + f"• 用户名: `{user.name}`", + f"• 显示名: `{user.display_name}`", + f"• ID: `{user.id}`", + f"• Bot: {'是' if user.bot else '否'}", + f"• 创建时间: {user.created_at.strftime('%Y-%m-%d %H:%M:%S')}", + ] + + if interaction.guild: + member = interaction.guild.get_member(user.id) + if member: + lines.append(f"• 昵称: `{member.nick or '无'}`") + joined_str = member.joined_at.strftime("%Y-%m-%d %H:%M:%S") if member.joined_at else "未知" + lines.append(f"• 加入时间: {joined_str}") + role_names = [r.name for r in member.roles if r.name != "@everyone"] + if role_names: + lines.append(f"• 角色: {', '.join(f'`{r}`' for r in role_names[:10])}") + + await interaction.followup.send("\n".join(lines)) + try: synced = await tree.sync() logger.info(f"Discord slash commands synced: {len(synced)} commands") diff --git a/backend/package/yuxi/channels/adapters/discord/components.py b/backend/package/yuxi/channels/adapters/discord/components.py index f9c4830d..8000c6da 100644 --- a/backend/package/yuxi/channels/adapters/discord/components.py +++ b/backend/package/yuxi/channels/adapters/discord/components.py @@ -5,6 +5,8 @@ from typing import Any import discord +from yuxi.utils.logging_config import logger + class ButtonStyle(Enum): PRIMARY = discord.ButtonStyle.primary @@ -27,7 +29,8 @@ def create_button( emoji: str | None = None, disabled: bool = False, ) -> discord.ui.Button: - kwargs: dict[str, Any] = {"label": label, "style": style, "disabled": disabled} + resolved_style = style.value if isinstance(style, ButtonStyle) else style + kwargs: dict[str, Any] = {"label": label, "style": resolved_style, "disabled": disabled} if custom_id is not None: kwargs["custom_id"] = custom_id if url is not None and style in (ButtonStyle.LINK, discord.ButtonStyle.link): @@ -44,7 +47,7 @@ def create_string_select( min_values: int = 1, max_values: int = 1, disabled: bool = False, -) -> discord.ui.StringSelect: +) -> discord.ui.Select: select_options = [ discord.SelectOption( label=opt.get("label", ""), @@ -55,7 +58,7 @@ def create_string_select( ) for opt in options ] - return discord.ui.StringSelect( + return discord.ui.Select( custom_id=custom_id, options=select_options, placeholder=placeholder, @@ -113,6 +116,22 @@ def create_channel_select( ) +def create_mentionable_select( + custom_id: str, + placeholder: str | None = None, + min_values: int = 1, + max_values: int = 1, + disabled: bool = False, +) -> discord.ui.MentionableSelect: + return discord.ui.MentionableSelect( + custom_id=custom_id, + placeholder=placeholder, + min_values=min_values, + max_values=max_values, + disabled=disabled, + ) + + def create_text_input( label: str, custom_id: str, @@ -123,10 +142,11 @@ def create_text_input( max_length: int | None = None, default: str | None = None, ) -> discord.ui.TextInput: + resolved_style = style.value if isinstance(style, TextInputStyle) else style kwargs: dict[str, Any] = { "label": label, "custom_id": custom_id, - "style": style, + "style": resolved_style, "required": required, } if placeholder is not None: @@ -157,6 +177,15 @@ def build_view( timeout: float | None = None, ) -> discord.ui.View: view = discord.ui.View(timeout=timeout) + + async def _on_timeout(): + logger.debug(f"Discord View timeout: custom_id={view.id}") + for child in view.children: + child.disabled = True + view.stop() + + view.on_timeout = _on_timeout + for row in action_rows: for item in row: view.add_item(item) diff --git a/backend/package/yuxi/channels/adapters/discord/event_queue.py b/backend/package/yuxi/channels/adapters/discord/event_queue.py index c4e2047f..7015e1d7 100644 --- a/backend/package/yuxi/channels/adapters/discord/event_queue.py +++ b/backend/package/yuxi/channels/adapters/discord/event_queue.py @@ -12,9 +12,13 @@ class EventQueue: DEFAULT_FORCE_FLUSH_THRESHOLD = 5 DEFAULT_FORCE_FLUSH_MIN_AGE_MS = 200 - def __init__(self, max_queue_size: int = 1000, ordering_window_ms: int = 500, - force_flush_threshold: int = DEFAULT_FORCE_FLUSH_THRESHOLD, - force_flush_min_age_ms: int = DEFAULT_FORCE_FLUSH_MIN_AGE_MS): + def __init__( + self, + max_queue_size: int = 1000, + ordering_window_ms: int = 500, + force_flush_threshold: int = DEFAULT_FORCE_FLUSH_THRESHOLD, + force_flush_min_age_ms: int = DEFAULT_FORCE_FLUSH_MIN_AGE_MS, + ): self._max_queue_size = max_queue_size self._ordering_window_ms = ordering_window_ms self._force_flush_threshold = force_flush_threshold @@ -44,8 +48,7 @@ class EventQueue: if len(self._pending) >= self._max_queue_size: self._pending.popitem(last=False) self._dropped_count += 1 - logger.warning(f"[EventQueue] Queue full, dropping oldest event " - f"(total dropped: {self._dropped_count})") + logger.warning(f"[EventQueue] Queue full, dropping oldest event (total dropped: {self._dropped_count})") key = f"{sequence:020d}_{event_id}" self._pending[key] = { @@ -92,8 +95,10 @@ class EventQueue: if item["enqueued_at"] <= min_age_cutoff: ready_keys.append(key) if ready_keys: - logger.debug(f"[EventQueue] Force flush: {len(ready_keys)} events " - f"(pending={len(self._pending)}, threshold={self._force_flush_threshold})") + logger.debug( + f"[EventQueue] Force flush: {len(ready_keys)} events " + f"(pending={len(self._pending)}, threshold={self._force_flush_threshold})" + ) for key in ready_keys: item = self._pending.pop(key) diff --git a/backend/package/yuxi/channels/adapters/discord/formatter.py b/backend/package/yuxi/channels/adapters/discord/formatter.py index f80f0977..235105ba 100644 --- a/backend/package/yuxi/channels/adapters/discord/formatter.py +++ b/backend/package/yuxi/channels/adapters/discord/formatter.py @@ -1,13 +1,12 @@ from __future__ import annotations import re -from datetime import datetime, UTC +from datetime import UTC, datetime from typing import Any import discord from yuxi.channels.models import ChannelResponse -from yuxi.utils.logging_config import logger _SANITIZE_PATTERNS = [ re.compile(r"[\s\S]*?", re.IGNORECASE), diff --git a/backend/package/yuxi/channels/adapters/discord/message_actions.py b/backend/package/yuxi/channels/adapters/discord/message_actions.py index d0c787db..79badcca 100644 --- a/backend/package/yuxi/channels/adapters/discord/message_actions.py +++ b/backend/package/yuxi/channels/adapters/discord/message_actions.py @@ -103,6 +103,41 @@ DISCORD_MESSAGE_ACTIONS: dict[str, dict[str, Any]] = { "description": "Lock a thread to prevent further replies", "api": "modify_thread", }, + "unarchiveThread": { + "status": "implemented", + "description": "Unarchive a thread to make it active again", + "api": "modify_thread", + }, + "unlockThread": { + "status": "implemented", + "description": "Unlock a thread to allow replies again", + "api": "modify_thread", + }, + "joinThread": { + "status": "implemented", + "description": "Join a thread as the bot user", + "api": "join_thread", + }, + "leaveThread": { + "status": "implemented", + "description": "Leave a thread as the bot user", + "api": "leave_thread", + }, + "addThreadMember": { + "status": "implemented", + "description": "Add a user to a private thread", + "api": "add_thread_member", + }, + "removeThreadMember": { + "status": "implemented", + "description": "Remove a user from a private thread", + "api": "remove_thread_member", + }, + "listActiveThreads": { + "status": "implemented", + "description": "List all active threads in a guild", + "api": "list_active_threads", + }, "fetchMessage": { "status": "implemented", "description": "Fetch a single message by ID from a channel", diff --git a/backend/package/yuxi/channels/adapters/discord/normalizer.py b/backend/package/yuxi/channels/adapters/discord/normalizer.py index f7b00d72..6c185223 100644 --- a/backend/package/yuxi/channels/adapters/discord/normalizer.py +++ b/backend/package/yuxi/channels/adapters/discord/normalizer.py @@ -11,6 +11,7 @@ from yuxi.channels.models import ( ChannelIdentity, ChannelMessage, ChannelType, + ChatType, EventType, MentionsInfo, MessageType, @@ -335,3 +336,188 @@ class DiscordMessageNormalizer: if changes: msg.metadata["changes"] = changes return msg + + @staticmethod + def normalize_interaction(interaction: discord.Interaction) -> ChannelMessage: + guild = interaction.guild + channel = interaction.channel + + if guild and channel: + if isinstance(channel, discord.Thread): + chat_id = f"thread_{channel.id}" + chat_type = ChatType.THREAD + else: + chat_id = f"guild_{guild.id}_channel_{channel.id}" + chat_type = ChatType.GUILD_CHANNEL + else: + chat_id = f"dm_{interaction.user.id}" + chat_type = ChatType.DIRECT + + interaction_data = getattr(interaction, "data", {}) or {} + metadata: dict[str, Any] = { + "interaction_type": str(interaction.type), + "interaction_id": str(interaction.id), + "token": interaction.token, + "custom_id": interaction_data.get("custom_id"), + "command_name": interaction_data.get("name"), + } + + if guild: + metadata["guild_id"] = str(guild.id) + metadata["guild_name"] = guild.name + if channel: + metadata["channel_id"] = str(channel.id) + metadata["channel_name"] = getattr(channel, "name", None) + + if interaction.type == discord.InteractionType.component: + metadata["component_type"] = str(interaction_data.get("component_type", "unknown")) + elif interaction.type == discord.InteractionType.modal_submit: + components = interaction_data.get("components", []) + input_values: dict[str, str] = {} + for comp in components: + for sub in comp.get("components", []): + input_values[sub.get("custom_id", "")] = sub.get("value", "") + metadata["modal_values"] = input_values + + content = "" + if interaction.type == discord.InteractionType.component: + content = f"component:{interaction_data.get('custom_id', '')}" + elif interaction.type == discord.InteractionType.modal_submit: + content = f"modal:{interaction_data.get('custom_id', '')}" + elif interaction.type == discord.InteractionType.application_command: + content = f"/{interaction_data.get('name', '')}" + + return ChannelMessage( + identity=ChannelIdentity( + channel_id="discord", + channel_type=ChannelType.DISCORD, + channel_user_id=str(interaction.user.id), + channel_chat_id=chat_id, + channel_message_id=str(interaction.id), + ), + event_type=EventType.INTERACTION, + chat_type=chat_type, + content=content, + metadata=metadata, + ) + + @staticmethod + def normalize_channel_event(channel: discord.abc.GuildChannel, event_type: EventType) -> ChannelMessage: + guild = channel.guild + guild_id = str(guild.id) if guild else "" + channel_id = str(channel.id) + + metadata: dict[str, Any] = { + "guild_id": guild_id, + "channel_id": channel_id, + "channel_name": channel.name, + "channel_type": str(channel.type), + } + + if guild: + metadata["guild_name"] = guild.name + + if hasattr(channel, "position"): + metadata["position"] = channel.position + if hasattr(channel, "parent_id") and channel.parent_id: + metadata["parent_id"] = str(channel.parent_id) + + return ChannelMessage( + identity=ChannelIdentity( + channel_id="discord", + channel_type=ChannelType.DISCORD, + channel_user_id="", + channel_chat_id=guild_id, + channel_message_id=channel_id, + ), + event_type=event_type, + content=channel.name, + metadata=metadata, + ) + + @staticmethod + def normalize_channel_update(before: discord.abc.GuildChannel, after: discord.abc.GuildChannel) -> ChannelMessage: + msg = DiscordMessageNormalizer.normalize_channel_event(after, EventType.CHANNEL_UPDATED) + changes: dict[str, Any] = {} + if before.name != after.name: + changes["name_before"] = before.name + changes["name_after"] = after.name + if hasattr(before, "position") and hasattr(after, "position") and before.position != after.position: + changes["position_before"] = before.position + changes["position_after"] = after.position + if hasattr(before, "topic") and hasattr(after, "topic") and before.topic != after.topic: + changes["topic_before"] = before.topic + changes["topic_after"] = after.topic + if hasattr(before, "nsfw") and hasattr(after, "nsfw") and before.nsfw != after.nsfw: + changes["nsfw_before"] = before.nsfw + changes["nsfw_after"] = after.nsfw + if changes: + msg.metadata["changes"] = changes + return msg + + @staticmethod + def normalize_interaction_raw(data: dict) -> ChannelMessage: + interaction_type = data.get("type", 0) + user_data = data.get("user", {}) or data.get("member", {}).get("user", {}) + user_id = str(user_data.get("id", "")) + interaction_id = str(data.get("id", "")) + interaction_token = data.get("token", "") + + guild_id = data.get("guild_id") + channel_id = data.get("channel_id") + + if guild_id and channel_id: + chat_id = f"guild_{guild_id}_channel_{channel_id}" + chat_type = ChatType.GUILD_CHANNEL + elif channel_id: + chat_id = f"dm_{user_id}" + chat_type = ChatType.DIRECT + else: + chat_id = f"dm_{user_id}" + chat_type = ChatType.DIRECT + + interaction_data = data.get("data", {}) or {} + metadata: dict[str, Any] = { + "interaction_type": str(interaction_type), + "interaction_id": interaction_id, + "token": interaction_token, + "custom_id": interaction_data.get("custom_id"), + "command_name": interaction_data.get("name"), + } + + if guild_id: + metadata["guild_id"] = str(guild_id) + if channel_id: + metadata["channel_id"] = str(channel_id) + + if interaction_type == 3: + metadata["component_type"] = str(interaction_data.get("component_type", "unknown")) + elif interaction_type == 5: + components = interaction_data.get("components", []) + input_values: dict[str, str] = {} + for comp in components: + for sub in comp.get("components", []): + input_values[sub.get("custom_id", "")] = sub.get("value", "") + metadata["modal_values"] = input_values + + content = "" + if interaction_type == 3: + content = f"component:{interaction_data.get('custom_id', '')}" + elif interaction_type == 5: + content = f"modal:{interaction_data.get('custom_id', '')}" + elif interaction_type == 2: + content = f"/{interaction_data.get('name', '')}" + + return ChannelMessage( + identity=ChannelIdentity( + channel_id="discord", + channel_type=ChannelType.DISCORD, + channel_user_id=user_id, + channel_chat_id=chat_id, + channel_message_id=interaction_id, + ), + event_type=EventType.INTERACTION, + chat_type=chat_type, + content=content, + metadata=metadata, + ) diff --git a/backend/package/yuxi/channels/adapters/discord/poll.py b/backend/package/yuxi/channels/adapters/discord/poll.py index 14cdd02a..55530c40 100644 --- a/backend/package/yuxi/channels/adapters/discord/poll.py +++ b/backend/package/yuxi/channels/adapters/discord/poll.py @@ -1,5 +1,7 @@ from __future__ import annotations +from datetime import timedelta + import discord from yuxi.channels.models import DeliveryResult @@ -20,13 +22,13 @@ async def create_poll( return DeliveryResult(success=False, error="Question too long (max 300 characters)") try: - poll_answers = [discord.PollAnswer(text=opt[:55]) for opt in options] poll = discord.Poll( - question=discord.PollMedia(text=question[:300]), - answers=poll_answers, - duration=duration_hours, - allow_multiselect=allow_multiselect, + question=question[:300], + duration=timedelta(hours=duration_hours), + multiple=allow_multiselect, ) + for opt in options: + poll.add_answer(text=opt[:55]) msg = await channel.send(poll=poll) return DeliveryResult( success=True, diff --git a/backend/package/yuxi/channels/adapters/discord/rest_scheduler.py b/backend/package/yuxi/channels/adapters/discord/rest_scheduler.py index e5fa9bb4..6384acae 100644 --- a/backend/package/yuxi/channels/adapters/discord/rest_scheduler.py +++ b/backend/package/yuxi/channels/adapters/discord/rest_scheduler.py @@ -60,11 +60,17 @@ def _get_global_limiter() -> _GlobalRateLimit: class RESTRequest: def __init__(self, route: str, coro, priority: int = 0): self.route = route - self.coro = coro + self._coro_factory = coro if callable(coro) else None + self._coro = None if callable(coro) else coro self.priority = priority self.created_at = time.monotonic() self.future: asyncio.Future = asyncio.Future() + async def execute_coro(self): + if self._coro_factory is not None: + self._coro = self._coro_factory() + return await self._coro + def __lt__(self, other: RESTRequest) -> bool: if self.priority != other.priority: return self.priority > other.priority @@ -122,7 +128,7 @@ class RESTScheduler: last_error = None for attempt in range(self._retry_attempts): try: - result = await request.coro + result = await request.execute_coro() if not request.future.done(): request.future.set_result(result) return diff --git a/backend/package/yuxi/channels/adapters/discord/security.py b/backend/package/yuxi/channels/adapters/discord/security.py index 8905e29c..49b98dc1 100644 --- a/backend/package/yuxi/channels/adapters/discord/security.py +++ b/backend/package/yuxi/channels/adapters/discord/security.py @@ -9,7 +9,7 @@ _PolicyPairing = "pairing" _PolicyAllowlist = "allowlist" _PolicyDisabled = "disabled" -DM_POLICIES = {_PolicyOpen, _PolicyPairing, _PolicyAllowlist} +DM_POLICIES = {_PolicyOpen, _PolicyPairing, _PolicyAllowlist, _PolicyDisabled} GROUP_POLICIES = {_PolicyOpen, _PolicyAllowlist, _PolicyDisabled} @@ -37,7 +37,9 @@ class DiscordSecurityPolicy: self._guild_configs = guilds_config if self._dangerously_allow_name_matching: - logger.warning("[Discord/Security] Name-based guild matching is active — use numeric guild IDs in production") + logger.warning( + "[Discord/Security] Name-based guild matching is active — use numeric guild IDs in production" + ) def is_dm_allowed(self, user_id: str) -> bool: if self.dm_policy == _PolicyOpen: @@ -87,7 +89,9 @@ class DiscordSecurityPolicy: return True, "" allowed_channels = guild_config.get("allowChannels", []) - if self._match_channel(channel_id, channel_name, allowed_channels, allow_name_match=self._dangerously_allow_name_matching): + if self._match_channel( + channel_id, channel_name, allowed_channels, allow_name_match=self._dangerously_allow_name_matching + ): return True, "" allowed_roles = guild_config.get("allowRoles", []) @@ -125,7 +129,9 @@ class DiscordSecurityPolicy: return {} @staticmethod - def _match_channel(channel_id: str, channel_name: str | None, allowed_channels: list[str], *, allow_name_match: bool = False) -> bool: + def _match_channel( + channel_id: str, channel_name: str | None, allowed_channels: list[str], *, allow_name_match: bool = False + ) -> bool: if channel_id in allowed_channels: return True if allow_name_match and channel_name: diff --git a/backend/package/yuxi/channels/adapters/discord/send.py b/backend/package/yuxi/channels/adapters/discord/send.py index 4fd7548e..8018a073 100644 --- a/backend/package/yuxi/channels/adapters/discord/send.py +++ b/backend/package/yuxi/channels/adapters/discord/send.py @@ -24,7 +24,9 @@ _global_semaphore: asyncio.Semaphore | None = None _PRE_ROUTE_SEMAPHORES: dict[str, asyncio.Semaphore] = {} -def configure_concurrency(global_limit: int = _DEFAULT_GLOBAL_CONCURRENCY, route_limit: int = _DEFAULT_ROUTE_CONCURRENCY) -> None: +def configure_concurrency( + global_limit: int = _DEFAULT_GLOBAL_CONCURRENCY, route_limit: int = _DEFAULT_ROUTE_CONCURRENCY +) -> None: global _global_semaphore _global_semaphore = asyncio.Semaphore(global_limit) _PRE_ROUTE_SEMAPHORES.clear() @@ -37,6 +39,7 @@ def _get_global_semaphore() -> asyncio.Semaphore: _global_semaphore = asyncio.Semaphore(_DEFAULT_GLOBAL_CONCURRENCY) return _global_semaphore + ChunkMode = Literal["newline", "length"] diff --git a/backend/package/yuxi/channels/adapters/discord/webhook.py b/backend/package/yuxi/channels/adapters/discord/webhook.py new file mode 100644 index 00000000..6867230c --- /dev/null +++ b/backend/package/yuxi/channels/adapters/discord/webhook.py @@ -0,0 +1,55 @@ +from __future__ import annotations + +from enum import IntEnum + +try: + import nacl.bindings + import nacl.exceptions +except ImportError: + nacl = None + + +class InteractionType(IntEnum): + PING = 1 + APPLICATION_COMMAND = 2 + MESSAGE_COMPONENT = 3 + APPLICATION_COMMAND_AUTOCOMPLETE = 4 + MODAL_SUBMIT = 5 + + +class InteractionCallbackType(IntEnum): + PONG = 1 + CHANNEL_MESSAGE_WITH_SOURCE = 4 + DEFERRED_CHANNEL_MESSAGE_WITH_SOURCE = 5 + DEFERRED_UPDATE_MESSAGE = 6 + UPDATE_MESSAGE = 7 + APPLICATION_COMMAND_AUTOCOMPLETE_RESULT = 8 + MODAL = 9 + + +def verify_ed25519_signature( + public_key_hex: str, + signature_hex: str, + timestamp: str, + body: str, +) -> bool: + if nacl is None: + raise ImportError("PyNaCl is required for Ed25519 signature verification. Install it with: pip install pynacl") + + try: + public_key_bytes = bytes.fromhex(public_key_hex) + signature_bytes = bytes.fromhex(signature_hex) + except ValueError: + return False + + message = timestamp.encode() + body.encode() + + try: + nacl.bindings.crypto_sign_verify(signature_bytes, message, public_key_bytes) + return True + except nacl.exceptions.BadSignatureError: + return False + + +def handle_ping_interaction() -> dict[str, int]: + return {"type": InteractionCallbackType.PONG}