Kris
1f78c44b03
这是一个批量整理提交，包含以下主要改动： 1. 删除多处冗余的空行和未使用的导入 2. 修复文件末尾缺少换行符的问题 3. 调整部分模块的导入顺序与代码排版 4. 修复部分配置默认值与策略逻辑 5. 新增多个功能模块与辅助工具 6. 完善异常处理与日志记录 7. 修复速率限制、消息缓存、权限校验等逻辑bug 8. 废弃部分旧有API与配置项并添加警告提示
2377 lines
97 KiB
Python
2377 lines
97 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import hmac
|
|
import io
|
|
import os
|
|
import re
|
|
import time
|
|
from collections.abc import AsyncIterator
|
|
from collections import OrderedDict
|
|
from datetime import datetime, UTC
|
|
from typing import Any, ClassVar
|
|
|
|
from slack_sdk.http_retry.builtin_async_handlers import AsyncRateLimitErrorRetryHandler
|
|
from slack_sdk.socket_mode.aiohttp import SocketModeClient
|
|
from slack_sdk.socket_mode.request import SocketModeRequest
|
|
from slack_sdk.errors import SlackApiError
|
|
from slack_sdk.web.async_client import AsyncWebClient
|
|
|
|
from yuxi.channels.base import BaseChannelAdapter
|
|
from yuxi.channels.capabilities import ChannelCapabilities, TTSCapabilities, TTSVoiceCapabilities
|
|
from yuxi.channels.meta import ChannelMeta
|
|
from yuxi.channels.exceptions import (
|
|
ChannelAuthenticationError,
|
|
ChannelException,
|
|
ChannelNotConnectedError,
|
|
)
|
|
from yuxi.channels.models import (
|
|
Attachment,
|
|
ChannelIdentity,
|
|
ChannelMessage,
|
|
ChannelResponse,
|
|
ChannelStatus,
|
|
ChannelType,
|
|
DeliveryResult,
|
|
EventType,
|
|
HealthStatus,
|
|
MentionsInfo,
|
|
MessageType,
|
|
)
|
|
from yuxi.channels.registry import register_builtin_adapter
|
|
from yuxi.channels.adapters.slack.session import resolve_chat_id, resolve_chat_type
|
|
from yuxi.channels.adapters.slack.security import (
|
|
DmPolicy,
|
|
SecurityConfig,
|
|
SecurityDecision,
|
|
)
|
|
from yuxi.channels.adapters.slack.security_audit import (
|
|
audit_security_config,
|
|
auto_fix_security,
|
|
)
|
|
from yuxi.channels.adapters.slack.allowlist import AllowlistManager
|
|
from yuxi.channels.adapters.slack.pairing import PairingManager
|
|
from yuxi.channels.adapters.slack.poll_manager import PollManager
|
|
from yuxi.channels.adapters.slack.commands import SlackCommandRegistry
|
|
from yuxi.channels.adapters.slack.approval import ApprovalManager
|
|
from yuxi.channels.adapters.slack.reaction_notify import ReactionNotifyConfig
|
|
from yuxi.channels.adapters.slack.voice.tts import SlackTTSConfig, synthesize_slack_tts
|
|
from yuxi.channels.adapters.slack.vision.vision import SlackVisionConfig
|
|
from yuxi.channels.adapters.slack.chunker import ChunkMode, resolve_text_chunks
|
|
from yuxi.channels.adapters.slack.sent_cache import SentMessageCache
|
|
from yuxi.channels.adapters.slack.reconnect import Http401BackoffState
|
|
from yuxi.channels.adapters.slack.normalizer import strip_mentions, normalize_slack_text
|
|
from yuxi.channels.adapters.slack.interactive_replies import compile_interactive_replies, has_interactive_syntax
|
|
from yuxi.channels.adapters.slack.agent_prompt import SlackPromptHints
|
|
from yuxi.channels.adapters.slack.streaming_compat import StreamingConfig
|
|
from yuxi.utils.logging_config import logger
|
|
|
|
|
|
@register_builtin_adapter
|
|
class SlackAdapter(BaseChannelAdapter):
|
|
channel_id: ClassVar[str] = "slack"
|
|
channel_type: ClassVar[ChannelType] = ChannelType.SLACK
|
|
webhook_path: ClassVar[str | None] = "/api/webhook/slack"
|
|
|
|
text_chunk_limit: ClassVar[int] = 8000
|
|
supports_markdown: ClassVar[bool] = True
|
|
supports_streaming: ClassVar[bool] = True
|
|
streaming_modes: ClassVar[list[str]] = ["off", "partial", "block", "progress"]
|
|
max_media_size_mb: ClassVar[int] = 100
|
|
|
|
capabilities = ChannelCapabilities(
|
|
chat_types=["direct", "channel", "thread"],
|
|
reactions=True,
|
|
edit=True,
|
|
unsend=True,
|
|
reply=True,
|
|
threads=True,
|
|
media=True,
|
|
pin=True,
|
|
unpin=True,
|
|
list_pins=True,
|
|
native_commands=True,
|
|
supports_markdown=True,
|
|
supports_streaming=True,
|
|
streaming_modes=["off", "partial", "block", "progress"],
|
|
text_chunk_limit=8000,
|
|
max_media_size_mb=100,
|
|
polls=True,
|
|
group_management=True,
|
|
tts=TTSCapabilities(voice=TTSVoiceCapabilities(enabled=True, synthesis_target="voice-note")),
|
|
vision=True,
|
|
approval=True,
|
|
typing=True,
|
|
)
|
|
meta = ChannelMeta(
|
|
id="slack",
|
|
label="Slack",
|
|
aliases=["slack", "slackbot"],
|
|
markdown_capable=True,
|
|
selection_label="Slack (Socket Mode / HTTP)",
|
|
)
|
|
|
|
STREAM_UPDATE_MIN_INTERVAL_MS = 500
|
|
STREAM_TTL_SECONDS = 300
|
|
SOCKET_CONNECT_TIMEOUT_S = 30.0
|
|
MAX_RETRY_COUNT = 3
|
|
CHANNEL_EVENT_HEARTBEAT_TIMEOUT_S = 600
|
|
MAX_PROCESSED_EVENT_IDS = 1000
|
|
HTTP_REQUEST_TIMEOUT_S = 30.0
|
|
HTTP_MAX_BODY_BYTES = 1_048_576
|
|
PAIRING_TTL_SECONDS = 300.0
|
|
COALESCE_MIN_CHARS: ClassVar[int] = 1500
|
|
COALESCE_IDLE_MS: ClassVar[int] = 1000
|
|
|
|
_APP_TOKEN_PATTERN: ClassVar = re.compile(r"^xapp-\d-[a-z0-9]+-", re.IGNORECASE)
|
|
|
|
def __init__(self, config: dict[str, Any] | None = None):
|
|
super().__init__(config)
|
|
self._status: ChannelStatus = ChannelStatus.DISCONNECTED
|
|
self._client: AsyncWebClient | None = None
|
|
self._socket_handler: SocketModeClient | None = None
|
|
self._socket_task: asyncio.Task | None = None
|
|
self._connected_event = asyncio.Event()
|
|
self._connected_at: datetime | None = None
|
|
self._bot_user_id: str = ""
|
|
self._bot_id: str = ""
|
|
self._team: str = ""
|
|
self._team_id: str = ""
|
|
self._streaming_messages: dict[str, dict] = {}
|
|
self._stream_lock = asyncio.Lock()
|
|
self._stream_cleanup_task: asyncio.Task | None = None
|
|
self._stream_cleanup_interval_s = 60
|
|
self._last_channel_event_at: float = 0.0
|
|
self._channel_heartbeat_task: asyncio.Task | None = None
|
|
self._processed_event_ids: OrderedDict[str, None] = OrderedDict()
|
|
self._reconnect_count = 0
|
|
self._http_401_backoff = Http401BackoffState()
|
|
|
|
self._shutting_down = False
|
|
self._mode: str = config.get("mode", "socket") if config else "socket"
|
|
self._signing_secret = (
|
|
config.get("signing_secret", "") or os.getenv("SLACK_SIGNING_SECRET", "")
|
|
if config
|
|
else os.getenv("SLACK_SIGNING_SECRET", "")
|
|
)
|
|
|
|
self._send_username: str = config.get("username", "") if config else ""
|
|
self._send_icon_url: str = config.get("icon_url", "") if config else ""
|
|
self._send_icon_emoji: str = config.get("icon_emoji", "") if config else ""
|
|
|
|
self._bot_token_source: str = "none"
|
|
self._app_token_source: str = "none"
|
|
self._user_token_source: str = "none"
|
|
self._user_token: str = ""
|
|
|
|
self._security_config = SecurityConfig.from_config(config)
|
|
self._allowlist_mgr = AllowlistManager(lambda: self._client)
|
|
self._pairing_mgr = PairingManager(ttl_seconds=self.PAIRING_TTL_SECONDS)
|
|
self._sent_cache = SentMessageCache()
|
|
self._poll_mgr = PollManager()
|
|
self._cmd_registry = SlackCommandRegistry()
|
|
self._cmd_registry.register_defaults()
|
|
self._approval_mgr = ApprovalManager()
|
|
self._reaction_notify_cfg = ReactionNotifyConfig.from_config(config)
|
|
self._tts_cfg = SlackTTSConfig.from_config(config)
|
|
self._vision_cfg = SlackVisionConfig.from_config(config)
|
|
self._prompt_hints = SlackPromptHints.from_config(config)
|
|
self._streaming_config = StreamingConfig.from_config(config)
|
|
self._dm_scope: str = config.get("dm_scope", "per-user") if config else "per-user"
|
|
self._native_stream_mgr: Any = None
|
|
|
|
self._inbound_queue: asyncio.Queue[ChannelMessage] = asyncio.Queue(maxsize=256)
|
|
self._shutdown_event = asyncio.Event()
|
|
|
|
async def pre_connect(self) -> dict:
|
|
bot_token = self._resolve_bot_token()
|
|
app_token = self._resolve_app_token()
|
|
if not bot_token:
|
|
return {"status": "error", "message": "Missing bot_token (SLACK_BOT_TOKEN)"}
|
|
if not app_token:
|
|
return {"status": "error", "message": "Missing app_token (SLACK_APP_TOKEN)"}
|
|
if not bot_token.startswith("xoxb-"):
|
|
return {"status": "error", "message": "bot_token must start with 'xoxb-'"}
|
|
if not app_token.startswith("xapp-"):
|
|
return {"status": "error", "message": "app_token must start with 'xapp-'"}
|
|
|
|
try:
|
|
client = AsyncWebClient(token=bot_token)
|
|
auth = await client.auth_test()
|
|
if not auth.get("ok"):
|
|
return {"status": "error", "message": f"auth.test failed: {auth.get('error')}"}
|
|
return {
|
|
"status": "ok",
|
|
"bot_id": auth.get("bot_id", ""),
|
|
"bot_user_id": auth.get("user_id", ""),
|
|
"team": auth.get("team", ""),
|
|
"team_id": auth.get("team_id", ""),
|
|
}
|
|
except SlackApiError as e:
|
|
return {"status": "error", "message": str(e)}
|
|
except Exception as e:
|
|
return {"status": "error", "message": str(e)}
|
|
|
|
async def connect(self) -> None:
|
|
bot_token = self._resolve_bot_token()
|
|
app_token = self._resolve_app_token()
|
|
|
|
if self._mode == "http":
|
|
await self._connect_http(bot_token)
|
|
return
|
|
|
|
if not bot_token or not app_token:
|
|
raise ChannelAuthenticationError(
|
|
"Slack bot_token and app_token must be configured.\n"
|
|
" bot_token: SLACK_BOT_TOKEN env or config (xoxb-...)\n"
|
|
" app_token: SLACK_APP_TOKEN env or config (xapp-...)"
|
|
)
|
|
|
|
self._status = ChannelStatus.CONNECTING
|
|
self._connected_event.clear()
|
|
|
|
self._client = AsyncWebClient(token=bot_token)
|
|
self._client.retry_handlers.append(AsyncRateLimitErrorRetryHandler(max_retry_count=self.MAX_RETRY_COUNT))
|
|
|
|
audit_result = audit_security_config(self._security_config)
|
|
if audit_result.has_critical:
|
|
logger.error(f"Slack security audit failed: {audit_result.to_dict()}")
|
|
self._security_config = auto_fix_security(self._security_config)
|
|
logger.warning(
|
|
f"Security config auto-fixed. Review allowlist settings. New config: {self._security_config.to_dict()}"
|
|
)
|
|
|
|
try:
|
|
auth = await self._client.auth_test()
|
|
if not auth.get("ok"):
|
|
raise ChannelAuthenticationError(f"auth.test failed: {auth.get('error')}")
|
|
except SlackApiError as e:
|
|
raise ChannelAuthenticationError(str(e)) from e
|
|
|
|
self._verify_token_app_id(bot_token, app_token, auth)
|
|
|
|
self._bot_user_id = auth["user_id"]
|
|
self._bot_id = auth.get("bot_id", "")
|
|
self._team = auth["team"]
|
|
self._team_id = auth.get("team_id", "")
|
|
logger.info(f"Slack bot @{auth['user']} authenticated, workspace: '{auth['team']}'")
|
|
|
|
self._socket_handler = SocketModeClient(
|
|
app_token=app_token,
|
|
web_client=self._client,
|
|
auto_reconnect_enabled=True,
|
|
)
|
|
|
|
self._register_socket_events()
|
|
|
|
self._socket_task = asyncio.create_task(self._socket_handler.connect_async())
|
|
self._connected_at = datetime.now(UTC)
|
|
|
|
try:
|
|
await asyncio.wait_for(
|
|
self._connected_event.wait(),
|
|
timeout=self.SOCKET_CONNECT_TIMEOUT_S,
|
|
)
|
|
except TimeoutError:
|
|
self._status = ChannelStatus.ERROR
|
|
raise ChannelException(
|
|
"Slack Socket Mode connection timed out",
|
|
retryable=True,
|
|
retry_after_ms=5000,
|
|
)
|
|
|
|
self._status = ChannelStatus.CONNECTED
|
|
logger.info(f"Slack Socket Mode connected, bot: @{auth['user']}, team: {auth['team']}")
|
|
|
|
self._stream_cleanup_task = asyncio.create_task(self._stream_cleanup_loop())
|
|
self._last_channel_event_at = time.monotonic()
|
|
self._channel_heartbeat_task = asyncio.create_task(self._channel_heartbeat_loop())
|
|
|
|
async def disconnect(self) -> None:
|
|
self._shutting_down = True
|
|
self._shutdown_event.set()
|
|
self._status = ChannelStatus.DISCONNECTED
|
|
self._streaming_messages.clear()
|
|
self._connected_event.clear()
|
|
|
|
if self._channel_heartbeat_task and not self._channel_heartbeat_task.done():
|
|
self._channel_heartbeat_task.cancel()
|
|
try:
|
|
await self._channel_heartbeat_task
|
|
except (asyncio.CancelledError, Exception):
|
|
pass
|
|
|
|
if self._stream_cleanup_task and not self._stream_cleanup_task.done():
|
|
self._stream_cleanup_task.cancel()
|
|
try:
|
|
await self._stream_cleanup_task
|
|
except (asyncio.CancelledError, Exception):
|
|
pass
|
|
|
|
if self._socket_handler:
|
|
try:
|
|
self._socket_handler.disconnect()
|
|
except Exception:
|
|
pass
|
|
|
|
if self._socket_task and not self._socket_task.done():
|
|
self._socket_task.cancel()
|
|
try:
|
|
await self._socket_task
|
|
except (asyncio.CancelledError, Exception):
|
|
pass
|
|
|
|
async with self._stream_lock:
|
|
self._streaming_messages.clear()
|
|
|
|
self._client = None
|
|
self._socket_handler = None
|
|
self._socket_task = None
|
|
self._connected_at = None
|
|
logger.info("Slack adapter disconnected")
|
|
|
|
async def send(self, response: ChannelResponse) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
payload = self.format_outbound(response)
|
|
chat_id = response.identity.channel_chat_id
|
|
thread_ts = response.metadata.get("thread_ts")
|
|
|
|
cached_ts = await self._sent_cache.get(chat_id)
|
|
is_dm = chat_id.startswith("D")
|
|
if cached_ts and not thread_ts and not response.reply_to_message_id and is_dm:
|
|
thread_ts = cached_ts
|
|
|
|
params: dict[str, Any] = {"channel": chat_id}
|
|
text = payload.get("text", "")
|
|
blocks = payload.get("blocks")
|
|
|
|
if response.reply_to_message_id:
|
|
params["thread_ts"] = response.reply_to_message_id
|
|
elif thread_ts:
|
|
params["thread_ts"] = thread_ts
|
|
|
|
if blocks:
|
|
params["blocks"] = blocks
|
|
|
|
for key in ("username", "icon_url", "icon_emoji"):
|
|
if payload.get(key):
|
|
params[key] = payload[key]
|
|
|
|
if len(text) <= self.text_chunk_limit:
|
|
params["text"] = text or "(empty message)"
|
|
params["mrkdwn"] = payload.get("mrkdwn", True)
|
|
result = await self._client.chat_postMessage(**params)
|
|
if result.get("ok"):
|
|
ts = result.get("ts")
|
|
if ts:
|
|
await self._sent_cache.put(chat_id, ts)
|
|
return DeliveryResult(success=True, message_id=ts)
|
|
return DeliveryResult(success=False, error=result.get("error", "Unknown error"))
|
|
|
|
chunks = resolve_text_chunks(text, self.text_chunk_limit, mode=ChunkMode.NEWLINE)
|
|
first_result = None
|
|
base_thread_ts = params.get("thread_ts")
|
|
failed_chunks: list[dict[str, Any]] = []
|
|
|
|
for chunk in chunks:
|
|
chunk_params: dict[str, Any] = {"channel": chat_id, "text": chunk.text, "mrkdwn": True}
|
|
if base_thread_ts:
|
|
chunk_params["thread_ts"] = base_thread_ts
|
|
if blocks and chunk.index == 0:
|
|
chunk_params["blocks"] = blocks
|
|
if chunk.index > 0 and first_result and first_result.message_id:
|
|
chunk_params["thread_ts"] = first_result.message_id
|
|
|
|
result = await self._client.chat_postMessage(**chunk_params)
|
|
if result.get("ok"):
|
|
ts = result.get("ts")
|
|
if chunk.index == 0:
|
|
first_result = DeliveryResult(success=True, message_id=ts)
|
|
if ts:
|
|
await self._sent_cache.put(chat_id, ts)
|
|
elif chunk.index == 0:
|
|
return DeliveryResult(success=False, error=result.get("error", "Unknown error"))
|
|
else:
|
|
failed_chunks.append({"index": chunk.index, "error": result.get("error", "Unknown error")})
|
|
|
|
metadata: dict[str, Any] = {}
|
|
if failed_chunks:
|
|
metadata["failed_chunks"] = failed_chunks
|
|
metadata["partial_success"] = True
|
|
logger.warning(f"Slack send: {len(failed_chunks)}/{len(chunks)} chunks failed: {failed_chunks}")
|
|
|
|
if first_result:
|
|
if metadata:
|
|
first_result.metadata = {**first_result.metadata, **metadata}
|
|
return first_result
|
|
return DeliveryResult(success=False, error="No chunks to send")
|
|
|
|
except SlackApiError as e:
|
|
err = e.response.get("error", str(e))
|
|
return DeliveryResult(success=False, error=err, metadata={"error_type": "slack_api_error"})
|
|
except (TypeError, ValueError, AttributeError):
|
|
logger.exception("Programming error in send()")
|
|
raise
|
|
except Exception as e:
|
|
return DeliveryResult(success=False, error=str(e), metadata={"error_type": "unexpected_error"})
|
|
|
|
async def receive(self) -> AsyncIterator[ChannelMessage]:
|
|
while not self._shutdown_event.is_set():
|
|
try:
|
|
msg = await asyncio.wait_for(self._inbound_queue.get(), timeout=1.0)
|
|
yield msg
|
|
except TimeoutError:
|
|
continue
|
|
except asyncio.CancelledError:
|
|
break
|
|
except Exception:
|
|
logger.debug("receive() iteration error", exc_info=True)
|
|
await asyncio.sleep(0.1)
|
|
|
|
def normalize_inbound(self, raw: dict) -> ChannelMessage:
|
|
event = raw.get("event", {})
|
|
channel = event.get("channel", "")
|
|
user = event.get("user", "")
|
|
ts = event.get("ts", "")
|
|
text = event.get("text", "")
|
|
thread_ts = event.get("thread_ts")
|
|
|
|
stripped_text = normalize_slack_text(text) if text else ""
|
|
|
|
chat_type = resolve_chat_type(channel, event)
|
|
channel_chat_id = resolve_chat_id(channel, event)
|
|
|
|
is_at_bot = self._check_bot_mention(text)
|
|
mentioned_ids = self._extract_mention_ids(text) if text else []
|
|
|
|
msg_type = MessageType.TEXT
|
|
attachments: list[Attachment] = []
|
|
if event.get("files"):
|
|
files = event["files"]
|
|
msg_type = self._resolve_file_message_type(files)
|
|
attachments = self._extract_attachments(files)
|
|
|
|
evt = EventType.MESSAGE_RECEIVED
|
|
metadata: dict[str, Any] = {}
|
|
if event.get("type") == "interactive" or raw.get("interaction"):
|
|
evt = EventType.CARD_ACTION
|
|
interaction = raw.get("interaction", {})
|
|
metadata["interaction_type"] = interaction.get("type", "")
|
|
metadata["interaction_payload"] = interaction
|
|
|
|
reply_to = None
|
|
if thread_ts and thread_ts != ts:
|
|
reply_to = thread_ts
|
|
|
|
return ChannelMessage(
|
|
identity=ChannelIdentity(
|
|
channel_id=self.channel_id,
|
|
channel_type=self.channel_type,
|
|
channel_user_id=user,
|
|
channel_chat_id=channel_chat_id,
|
|
channel_message_id=f"{channel}:{ts}",
|
|
),
|
|
event_type=evt,
|
|
message_type=msg_type,
|
|
chat_type=chat_type,
|
|
content=stripped_text or text,
|
|
attachments=attachments,
|
|
mentions=MentionsInfo(
|
|
mentioned_user_ids=mentioned_ids,
|
|
is_bot_mentioned=is_at_bot,
|
|
),
|
|
reply_to_message_id=reply_to,
|
|
metadata={
|
|
**metadata,
|
|
"channel": channel,
|
|
"ts": ts,
|
|
"thread_ts": thread_ts,
|
|
"team_id": raw.get("team_id", self._team_id),
|
|
},
|
|
)
|
|
|
|
def format_outbound(self, response: ChannelResponse) -> dict[str, Any]:
|
|
content = response.content
|
|
blocks = response.metadata.get("blocks")
|
|
|
|
if not blocks and has_interactive_syntax(content):
|
|
clean_text, compiled_blocks = compile_interactive_replies(content)
|
|
if compiled_blocks:
|
|
content = clean_text or "(interactive message)"
|
|
blocks = compiled_blocks
|
|
|
|
payload: dict[str, Any] = {
|
|
"text": content,
|
|
"mrkdwn": True,
|
|
}
|
|
|
|
if blocks:
|
|
payload["blocks"] = blocks
|
|
|
|
if response.attachments:
|
|
payload["attachments"] = [
|
|
{
|
|
"attachment_type": att.type,
|
|
"url": att.url,
|
|
"filename": att.filename,
|
|
"size": att.size_bytes,
|
|
}
|
|
for att in response.attachments
|
|
]
|
|
|
|
if response.reply_to_message_id:
|
|
payload["thread_ts"] = response.reply_to_message_id
|
|
|
|
if response.metadata.get("thread_ts"):
|
|
payload["thread_ts"] = response.metadata["thread_ts"]
|
|
|
|
identity = response.metadata.get("identity", {})
|
|
username = identity.get("username") or self._send_username
|
|
if username:
|
|
payload["username"] = username
|
|
icon_url = identity.get("icon_url") or self._send_icon_url
|
|
if icon_url:
|
|
payload["icon_url"] = icon_url
|
|
icon_emoji = identity.get("icon_emoji") or self._send_icon_emoji
|
|
if icon_emoji:
|
|
payload["icon_emoji"] = icon_emoji
|
|
|
|
return payload
|
|
|
|
async def health_check(self) -> HealthStatus:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return HealthStatus(
|
|
status="unhealthy",
|
|
last_error="Socket Mode not connected",
|
|
)
|
|
|
|
start = time.monotonic()
|
|
try:
|
|
auth = await self._client.auth_test()
|
|
latency_ms = (time.monotonic() - start) * 1000
|
|
|
|
if not auth.get("ok"):
|
|
return HealthStatus(
|
|
status="degraded",
|
|
last_error=f"auth.test: {auth.get('error')}",
|
|
)
|
|
|
|
return HealthStatus(
|
|
status="healthy",
|
|
latency_ms=latency_ms,
|
|
last_connected_at=self._connected_at,
|
|
metadata={
|
|
"bot_id": self._bot_user_id,
|
|
"team": self._team,
|
|
"team_id": self._team_id,
|
|
},
|
|
)
|
|
except SlackApiError as e:
|
|
return HealthStatus(status="unhealthy", last_error=str(e))
|
|
except Exception as e:
|
|
return HealthStatus(status="unhealthy", last_error=str(e))
|
|
|
|
async def send_stream_chunk(self, chat_id: str, msg_id: str, chunk: str, finished: bool) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
now = time.monotonic()
|
|
full_text: str = ""
|
|
|
|
async with self._stream_lock:
|
|
last = self._streaming_messages.get(chat_id, {}).get("_last_update", 0)
|
|
elapsed_ms = (now - last) * 1000
|
|
if elapsed_ms < self.STREAM_UPDATE_MIN_INTERVAL_MS and not finished:
|
|
if chat_id in self._streaming_messages:
|
|
self._streaming_messages[chat_id]["_pending"] = (
|
|
self._streaming_messages[chat_id].get("_pending", "") + chunk
|
|
)
|
|
return DeliveryResult(success=True, message_id=msg_id)
|
|
|
|
pending = self._streaming_messages.get(chat_id, {}).pop("_pending", "")
|
|
full_text = pending + chunk if pending else chunk
|
|
|
|
try:
|
|
if not msg_id:
|
|
result = await self._client.chat_postMessage(
|
|
channel=chat_id,
|
|
text=full_text + (" ..." if not finished else ""),
|
|
mrkdwn=True,
|
|
)
|
|
ts = result.get("ts")
|
|
if ts:
|
|
async with self._stream_lock:
|
|
self._streaming_messages[chat_id] = {"ts": ts, "text": full_text, "_last_update": now}
|
|
return DeliveryResult(
|
|
success=result.get("ok", False),
|
|
message_id=ts,
|
|
error=result.get("error"),
|
|
)
|
|
else:
|
|
result = await self._client.chat_update(
|
|
channel=chat_id,
|
|
ts=msg_id,
|
|
text=full_text + (" ..." if not finished else ""),
|
|
mrkdwn=True,
|
|
)
|
|
async with self._stream_lock:
|
|
if finished:
|
|
self._streaming_messages.pop(chat_id, None)
|
|
else:
|
|
self._streaming_messages[chat_id] = {"ts": msg_id, "text": full_text, "_last_update": now}
|
|
return DeliveryResult(
|
|
success=result.get("ok", False),
|
|
message_id=result.get("ts", msg_id),
|
|
error=result.get("error"),
|
|
)
|
|
except SlackApiError as e:
|
|
err = e.response.get("error", str(e))
|
|
if err == "message_not_found":
|
|
resp = ChannelResponse(
|
|
identity=ChannelIdentity(
|
|
channel_id=self.channel_id,
|
|
channel_type=self.channel_type,
|
|
channel_user_id="",
|
|
channel_chat_id=chat_id,
|
|
),
|
|
content=full_text,
|
|
)
|
|
return await self.send(resp)
|
|
return DeliveryResult(success=False, error=err)
|
|
except (TypeError, ValueError, AttributeError):
|
|
logger.exception("Programming error in send_stream_chunk()")
|
|
raise
|
|
except Exception as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
async def send_media(
|
|
self, chat_id: str, media_type: str, data: Any, *, thread_ts: str | None = None
|
|
) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
filename = f"upload.{media_type}" if media_type else "upload"
|
|
kwargs: dict[str, Any] = {
|
|
"channel": chat_id,
|
|
"file": io.BytesIO(data) if isinstance(data, bytes) else data,
|
|
"filename": filename,
|
|
}
|
|
if thread_ts:
|
|
kwargs["thread_ts"] = thread_ts
|
|
result = await self._client.files_upload_v2(**kwargs)
|
|
if result.get("ok"):
|
|
file_info = result.get("file", {})
|
|
return DeliveryResult(
|
|
success=True,
|
|
message_id=file_info.get("id"),
|
|
)
|
|
return DeliveryResult(
|
|
success=False,
|
|
error=result.get("error", "files.upload failed"),
|
|
)
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
async def send_tts(self, chat_id: str, text: str, *, thread_ts: str | None = None) -> DeliveryResult:
|
|
if not self._tts_cfg.enabled:
|
|
return DeliveryResult(success=False, error="TTS not enabled")
|
|
|
|
audio_data = await synthesize_slack_tts(text, self._tts_cfg)
|
|
if not audio_data:
|
|
return DeliveryResult(success=False, error="TTS synthesis failed")
|
|
|
|
suffix = "mp3" if self._tts_cfg.response_format == "mp3" else "ogg"
|
|
filename = f"tts_{chat_id}.{suffix}"
|
|
return await self.send_media(chat_id, filename, audio_data, thread_ts=thread_ts)
|
|
|
|
async def send_approval(self, chat_id: str, title: str, detail: str, command: str) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
from yuxi.channels.adapters.slack.blocks import build_approval_blocks
|
|
|
|
try:
|
|
blocks = build_approval_blocks(title, detail)
|
|
result = await self._client.chat_postMessage(
|
|
channel=chat_id,
|
|
text=title,
|
|
blocks=blocks,
|
|
)
|
|
if not result.get("ok"):
|
|
return DeliveryResult(success=False, error=result.get("error", "Unknown error"))
|
|
|
|
ts = result.get("ts")
|
|
approval = self._approval_mgr.create_approval(
|
|
title=title,
|
|
detail=detail,
|
|
command=command,
|
|
chat_id=chat_id,
|
|
message_ts=ts or "",
|
|
)
|
|
return DeliveryResult(
|
|
success=True,
|
|
message_id=ts,
|
|
metadata={"approval_id": approval.approval_id, "status": approval.status.value},
|
|
)
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
async def handle_approval_callback(
|
|
self, action_id: str, user_id: str, message_ts: str, chat_id: str
|
|
) -> DeliveryResult:
|
|
pending = self._approval_mgr.list_pending(chat_id)
|
|
target = None
|
|
for p in pending:
|
|
req = self._approval_mgr.get_approval(p["approval_id"])
|
|
if req and req.message_ts == message_ts:
|
|
target = req
|
|
break
|
|
|
|
if not target:
|
|
return DeliveryResult(success=False, error="No matching pending approval found")
|
|
|
|
if not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
status_text = ""
|
|
if action_id == "exec_approve":
|
|
self._approval_mgr.approve(target.approval_id, user_id)
|
|
exec_result = await self._approval_mgr.execute_approval(target.approval_id)
|
|
if exec_result.get("success"):
|
|
status_text = f"已执行 ✅\n```{str(exec_result.get('result', ''))[:500]}```"
|
|
else:
|
|
status_text = f"执行失败 ❌\n`{exec_result.get('error', '')}`"
|
|
elif action_id == "exec_reject":
|
|
self._approval_mgr.reject(target.approval_id)
|
|
status_text = "已取消 ❌"
|
|
|
|
await self._client.chat_update(
|
|
channel=chat_id,
|
|
ts=message_ts,
|
|
text=f"*{target.title}*\n{target.detail}\n\n{status_text}",
|
|
blocks=None,
|
|
)
|
|
return DeliveryResult(success=True, metadata={"status": status_text})
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
async def send_poll(
|
|
self, chat_id: str, question: str, options: list[str], *, thread_ts: str | None = None
|
|
) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
if len(options) < 2:
|
|
return DeliveryResult(success=False, error="Poll requires at least 2 options")
|
|
if len(options) > 10:
|
|
return DeliveryResult(success=False, error="Poll supports at most 10 options")
|
|
|
|
from yuxi.channels.adapters.slack.blocks import build_poll_blocks
|
|
|
|
poll_msg_params: dict[str, Any] = {"channel": chat_id, "text": question}
|
|
if thread_ts:
|
|
poll_msg_params["thread_ts"] = thread_ts
|
|
|
|
try:
|
|
poll = self._poll_mgr.create_poll(question, options, chat_id, "", "")
|
|
blocks = build_poll_blocks(question, options, poll.poll_id)
|
|
poll_msg_params["blocks"] = blocks
|
|
result = await self._client.chat_postMessage(**poll_msg_params)
|
|
if result.get("ok"):
|
|
ts = result.get("ts")
|
|
poll.message_ts = ts or ""
|
|
return DeliveryResult(
|
|
success=True,
|
|
message_id=ts,
|
|
metadata={"poll_id": poll.poll_id, "results": poll.results()},
|
|
)
|
|
return DeliveryResult(success=False, error=result.get("error", "Unknown error"))
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
def _build_media_filename(self, media_type: str) -> str:
|
|
return f"upload.{media_type}" if media_type else "upload"
|
|
|
|
async def send_ephemeral(
|
|
self,
|
|
chat_id: str,
|
|
user_id: str,
|
|
text: str,
|
|
*,
|
|
blocks: list | None = None,
|
|
thread_ts: str | None = None,
|
|
) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
params: dict[str, Any] = {
|
|
"channel": chat_id,
|
|
"user": user_id,
|
|
"text": text,
|
|
}
|
|
if blocks:
|
|
params["blocks"] = blocks
|
|
if thread_ts:
|
|
params["thread_ts"] = thread_ts
|
|
result = await self._client.chat_postEphemeral(**params)
|
|
ok = result.get("ok", False)
|
|
return DeliveryResult(
|
|
success=ok,
|
|
message_id=result.get("message_ts"),
|
|
error=result.get("error") if not ok else None,
|
|
)
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
except (TypeError, ValueError, AttributeError):
|
|
logger.exception("Programming error in send_ephemeral()")
|
|
raise
|
|
except Exception as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
async def edit_message(self, chat_id: str, msg_id: str, content: str) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
result = await self._client.chat_update(
|
|
channel=chat_id,
|
|
ts=msg_id,
|
|
text=content,
|
|
mrkdwn=True,
|
|
)
|
|
return DeliveryResult(
|
|
success=result.get("ok", False),
|
|
message_id=result.get("ts"),
|
|
error=result.get("error"),
|
|
)
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
async def delete_message(self, chat_id: str, msg_id: str) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
result = await self._client.chat_delete(
|
|
channel=chat_id,
|
|
ts=msg_id,
|
|
)
|
|
return DeliveryResult(
|
|
success=result.get("ok", False),
|
|
error=result.get("error"),
|
|
)
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
async def send_reaction(self, chat_id: str, msg_id: str, emoji: str) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
name = emoji.replace(":", "")
|
|
result = await self._client.reactions_add(
|
|
channel=chat_id,
|
|
name=name,
|
|
timestamp=msg_id,
|
|
)
|
|
return DeliveryResult(
|
|
success=result.get("ok", False),
|
|
error=result.get("error"),
|
|
)
|
|
except SlackApiError as e:
|
|
err = e.response.get("error", str(e))
|
|
if err == "already_reacted":
|
|
return DeliveryResult(success=True, error="already_reacted")
|
|
return DeliveryResult(success=False, error=err)
|
|
|
|
async def get_user_info(self, channel_user_id: str) -> dict[str, Any]:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
raise ChannelNotConnectedError()
|
|
|
|
try:
|
|
result = await self._client.users_info(user=channel_user_id)
|
|
user = result.get("user", {})
|
|
return {
|
|
"id": user.get("id"),
|
|
"username": user.get("name"),
|
|
"display_name": user.get("profile", {}).get("display_name", ""),
|
|
"real_name": user.get("real_name", ""),
|
|
"email": user.get("profile", {}).get("email", ""),
|
|
"tz": user.get("tz", ""),
|
|
"is_admin": user.get("is_admin", False),
|
|
"is_bot": user.get("is_bot", False),
|
|
}
|
|
except SlackApiError as e:
|
|
raise ChannelException(f"Failed to get user info: {e}") from e
|
|
|
|
async def download_media(self, file_id: str) -> bytes:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
raise ChannelNotConnectedError()
|
|
|
|
try:
|
|
import aiohttp
|
|
|
|
file_info = await self._client.files_info(file=file_id)
|
|
if not file_info.get("ok"):
|
|
raise ChannelException(f"files.info failed: {file_info.get('error')}")
|
|
|
|
url_private = file_info.get("file", {}).get("url_private", "")
|
|
if not url_private:
|
|
raise ChannelException(f"File {file_id} has no downloadable URL")
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(
|
|
url_private,
|
|
headers={"Authorization": f"Bearer {self._resolve_bot_token()}"},
|
|
) as resp:
|
|
return await resp.read()
|
|
|
|
except SlackApiError as e:
|
|
raise ChannelException(f"Failed to download file {file_id}: {e}") from e
|
|
except Exception as e:
|
|
raise ChannelException(f"Failed to download file {file_id}: {e}") from e
|
|
|
|
# ========== 内部辅助方法 ==========
|
|
|
|
def _resolve_bot_token(self) -> str:
|
|
token = self.config.get("bot_token", "")
|
|
if token:
|
|
self._bot_token_source = "config"
|
|
else:
|
|
token = os.getenv("SLACK_BOT_TOKEN", "")
|
|
if token:
|
|
self._bot_token_source = "env"
|
|
else:
|
|
self._bot_token_source = "none"
|
|
if token and not token.startswith("xoxb-"):
|
|
logger.warning(f"SLACK_BOT_TOKEN does not start with 'xoxb-', got prefix: {token[:5]}...")
|
|
return token
|
|
|
|
def _resolve_app_token(self) -> str:
|
|
token = self.config.get("app_token", "")
|
|
if token:
|
|
self._app_token_source = "config"
|
|
else:
|
|
token = os.getenv("SLACK_APP_TOKEN", "")
|
|
if token:
|
|
self._app_token_source = "env"
|
|
else:
|
|
self._app_token_source = "none"
|
|
if token and not token.startswith("xapp-"):
|
|
logger.warning(f"SLACK_APP_TOKEN does not start with 'xapp-', got prefix: {token[:5]}...")
|
|
if token and token.startswith("xapp-") and not self._APP_TOKEN_PATTERN.match(token):
|
|
logger.warning(f"SLACK_APP_TOKEN does not match expected format xapp-\\d-[a-z0-9]+-, got: {token[:12]}...")
|
|
return token
|
|
|
|
def _resolve_user_token(self) -> str:
|
|
token = self.config.get("user_token", "")
|
|
if token:
|
|
self._user_token_source = "config"
|
|
else:
|
|
token = os.getenv("SLACK_USER_TOKEN", "")
|
|
if token:
|
|
self._user_token_source = "env"
|
|
else:
|
|
self._user_token_source = "none"
|
|
if token and not token.startswith("xoxp-"):
|
|
logger.warning(f"SLACK_USER_TOKEN does not start with 'xoxp-', got prefix: {token[:5]}...")
|
|
return token
|
|
|
|
async def _connect_http(self, bot_token: str) -> None:
|
|
if not bot_token:
|
|
raise ChannelAuthenticationError(
|
|
"Slack bot_token must be configured for HTTP mode.\n"
|
|
" bot_token: SLACK_BOT_TOKEN env or config (xoxb-...)"
|
|
)
|
|
if not self._signing_secret:
|
|
raise ChannelAuthenticationError(
|
|
"Slack signing_secret must be configured for HTTP mode.\n"
|
|
" signing_secret: SLACK_SIGNING_SECRET env or config"
|
|
)
|
|
|
|
self._status = ChannelStatus.CONNECTING
|
|
self._client = AsyncWebClient(token=bot_token)
|
|
self._client.retry_handlers.append(AsyncRateLimitErrorRetryHandler(max_retry_count=self.MAX_RETRY_COUNT))
|
|
|
|
try:
|
|
auth = await self._client.auth_test()
|
|
if not auth.get("ok"):
|
|
raise ChannelAuthenticationError(f"auth.test failed: {auth.get('error')}")
|
|
except SlackApiError as e:
|
|
raise ChannelAuthenticationError(str(e)) from e
|
|
|
|
self._bot_user_id = auth["user_id"]
|
|
self._bot_id = auth.get("bot_id", "")
|
|
self._team = auth["team"]
|
|
self._team_id = auth.get("team_id", "")
|
|
logger.info(f"Slack bot @{auth['user']} authenticated (HTTP mode), workspace: '{auth['team']}'")
|
|
|
|
self._status = ChannelStatus.CONNECTED
|
|
self._connected_at = datetime.now(UTC)
|
|
self._connected_event.set()
|
|
|
|
def _verify_token_app_id(self, bot_token: str, app_token: str, auth: dict) -> None:
|
|
bot_app_id = self._extract_app_id_from_token(bot_token, "xoxb-")
|
|
app_app_id = self._extract_app_id_from_token(app_token, "xapp-")
|
|
|
|
if bot_app_id and app_app_id and bot_app_id != app_app_id:
|
|
logger.warning(
|
|
f"Slack bot_token app_id ({bot_app_id}) does not match "
|
|
f"app_token app_id ({app_app_id}), ensure tokens belong to the same app"
|
|
)
|
|
|
|
@staticmethod
|
|
def _extract_app_id_from_token(token: str, prefix: str) -> str:
|
|
if not token.startswith(prefix):
|
|
return ""
|
|
parts = token[len(prefix) :].split("-", 1)
|
|
return parts[0] if parts else ""
|
|
|
|
async def _handle_http_event(self, payload: dict) -> None:
|
|
event = payload.get("event", {})
|
|
event_type = payload.get("event", {}).get("type", "")
|
|
subtype = event.get("subtype", "")
|
|
|
|
if subtype == "message_changed":
|
|
await self._handle_http_message_changed(event, payload)
|
|
return
|
|
if subtype == "message_deleted":
|
|
await self._handle_http_message_deleted(event, payload)
|
|
return
|
|
|
|
if subtype and subtype not in ("file_share", "thread_broadcast"):
|
|
return
|
|
if event.get("bot_id") and not subtype:
|
|
return
|
|
|
|
msg = self.normalize_inbound(
|
|
{
|
|
"event": event,
|
|
"event_type": event_type,
|
|
"team_id": payload.get("team_id", ""),
|
|
}
|
|
)
|
|
await self._handle_message_with_security(msg)
|
|
|
|
async def _handle_http_message_changed(self, event: dict, payload: dict) -> None:
|
|
msg_data = event.get("message", {})
|
|
previous = event.get("previous_message", {})
|
|
msg = self.normalize_inbound(
|
|
{
|
|
"event": {**previous, "channel": event.get("channel", ""), "ts": previous.get("ts", "")},
|
|
"event_type": "message",
|
|
"team_id": payload.get("team_id", ""),
|
|
}
|
|
)
|
|
msg.event_type = EventType.MESSAGE_UPDATED
|
|
msg.content = msg_data.get("text", "")
|
|
msg.metadata["edited_message"] = msg_data
|
|
msg.metadata["previous_message"] = previous
|
|
await self._handle_message_with_security(msg)
|
|
|
|
async def _handle_http_message_deleted(self, event: dict, payload: dict) -> None:
|
|
ts = event.get("deleted_ts", "")
|
|
channel = event.get("channel", "")
|
|
msg = ChannelMessage(
|
|
identity=ChannelIdentity(
|
|
channel_id=self.channel_id,
|
|
channel_type=self.channel_type,
|
|
channel_user_id="",
|
|
channel_chat_id=channel,
|
|
channel_message_id=ts,
|
|
),
|
|
event_type=EventType.MESSAGE_DELETED,
|
|
chat_type="group" if not channel.startswith("D") else "direct",
|
|
content="",
|
|
metadata={"deleted_ts": ts, "team_id": payload.get("team_id", self._team_id)},
|
|
)
|
|
await self._handle_message_with_security(msg)
|
|
|
|
async def pin_message(self, chat_id: str, msg_id: str) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
result = await self._client.pins_add(channel=chat_id, timestamp=msg_id)
|
|
return DeliveryResult(
|
|
success=result.get("ok", False),
|
|
error=result.get("error"),
|
|
)
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
async def unpin_message(self, chat_id: str, msg_id: str) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
result = await self._client.pins_remove(channel=chat_id, timestamp=msg_id)
|
|
return DeliveryResult(
|
|
success=result.get("ok", False),
|
|
error=result.get("error"),
|
|
)
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
async def list_pins(self, chat_id: str) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
result = await self._client.pins_list(channel=chat_id)
|
|
items = result.get("items", [])
|
|
return DeliveryResult(
|
|
success=result.get("ok", False),
|
|
metadata={"pins": items, "count": len(items)},
|
|
error=result.get("error"),
|
|
)
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
async def send_typing(self, chat_id: str) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
result = await self._client.api_call(
|
|
api_method="chat.sendTyping",
|
|
http_verb="POST",
|
|
params={"channel": chat_id},
|
|
)
|
|
return DeliveryResult(
|
|
success=result.get("ok", False),
|
|
error=result.get("error"),
|
|
)
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
async def remove_reaction(self, chat_id: str, msg_id: str, emoji: str) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
name = emoji.replace(":", "")
|
|
result = await self._client.reactions_remove(
|
|
channel=chat_id,
|
|
name=name,
|
|
timestamp=msg_id,
|
|
)
|
|
return DeliveryResult(
|
|
success=result.get("ok", False),
|
|
error=result.get("error"),
|
|
)
|
|
except SlackApiError as e:
|
|
err = e.response.get("error", str(e))
|
|
if err == "no_reaction":
|
|
return DeliveryResult(success=True, error="no_reaction")
|
|
return DeliveryResult(success=False, error=err)
|
|
|
|
async def list_reactions(self, chat_id: str, msg_id: str) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
result = await self._client.reactions_get(
|
|
channel=chat_id,
|
|
timestamp=msg_id,
|
|
full=True,
|
|
)
|
|
if not result.get("ok"):
|
|
return DeliveryResult(success=False, error=result.get("error"))
|
|
|
|
message = result.get("message", {})
|
|
reactions = message.get("reactions", [])
|
|
return DeliveryResult(
|
|
success=True,
|
|
metadata={"reactions": reactions},
|
|
)
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
async def remove_own_reactions(self, chat_id: str, msg_id: str) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
result = await self._client.reactions_get(
|
|
channel=chat_id,
|
|
timestamp=msg_id,
|
|
full=True,
|
|
)
|
|
if not result.get("ok"):
|
|
return DeliveryResult(success=False, error=result.get("error"))
|
|
|
|
message = result.get("message", {})
|
|
reactions = message.get("reactions", [])
|
|
removed = 0
|
|
for reaction in reactions:
|
|
users = reaction.get("users", [])
|
|
if self._bot_user_id in users:
|
|
name = reaction.get("name", "")
|
|
try:
|
|
await self._client.reactions_remove(
|
|
channel=chat_id,
|
|
name=name,
|
|
timestamp=msg_id,
|
|
)
|
|
removed += 1
|
|
except SlackApiError:
|
|
pass
|
|
|
|
return DeliveryResult(success=True, metadata={"removed_count": removed})
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
async def read_messages(
|
|
self,
|
|
chat_id: str,
|
|
*,
|
|
limit: int = 100,
|
|
cursor: str = "",
|
|
thread_ts: str | None = None,
|
|
) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
if thread_ts:
|
|
result = await self._client.conversations_replies(
|
|
channel=chat_id,
|
|
ts=thread_ts,
|
|
limit=min(limit, 1000),
|
|
cursor=cursor or None,
|
|
)
|
|
messages = result.get("messages", [])
|
|
else:
|
|
result = await self._client.conversations_history(
|
|
channel=chat_id,
|
|
limit=min(limit, 1000),
|
|
cursor=cursor or None,
|
|
)
|
|
messages = result.get("messages", [])
|
|
|
|
return DeliveryResult(
|
|
success=result.get("ok", False),
|
|
metadata={
|
|
"messages": messages,
|
|
"count": len(messages),
|
|
"has_more": result.get("has_more", False),
|
|
"next_cursor": result.get("response_metadata", {}).get("next_cursor", ""),
|
|
},
|
|
error=result.get("error"),
|
|
)
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
async def list_emojis(self) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
result = await self._client.emoji_list()
|
|
emoji_data = result.get("emoji", {})
|
|
return DeliveryResult(
|
|
success=result.get("ok", False),
|
|
metadata={"emojis": emoji_data, "count": len(emoji_data)},
|
|
error=result.get("error"),
|
|
)
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
async def verify_webhook_signature(self, headers: dict, body: bytes) -> bool:
|
|
if not self._signing_secret:
|
|
return False
|
|
|
|
timestamp = headers.get("x-slack-request-timestamp", "")
|
|
signature = headers.get("x-slack-signature", "")
|
|
|
|
if not timestamp or not signature:
|
|
return False
|
|
|
|
try:
|
|
ts = int(timestamp)
|
|
except (ValueError, TypeError):
|
|
return False
|
|
|
|
now = int(time.time())
|
|
if abs(now - ts) > 300:
|
|
logger.warning("Slack webhook: request timestamp too old")
|
|
return False
|
|
|
|
sig_basestring = f"v0:{timestamp}:{body.decode('utf-8')}"
|
|
computed = (
|
|
"v0="
|
|
+ hmac.new(
|
|
self._signing_secret.encode("utf-8"),
|
|
sig_basestring.encode("utf-8"),
|
|
hashlib.sha256,
|
|
).hexdigest()
|
|
)
|
|
|
|
return hmac.compare_digest(computed, signature)
|
|
|
|
async def _check_security(self, msg: ChannelMessage) -> SecurityDecision:
|
|
if msg.chat_type == "direct":
|
|
if not self._security_config.is_allowed_dm(msg.identity.channel_user_id):
|
|
if self._security_config.dm_policy == DmPolicy.PAIRING:
|
|
decision = await self._pairing_mgr.generate_pairing(msg.identity.channel_user_id)
|
|
if decision.requires_pairing:
|
|
return decision
|
|
return SecurityDecision(
|
|
allowed=False,
|
|
reason=f"dm_policy={self._security_config.dm_policy.value}",
|
|
)
|
|
return SecurityDecision(allowed=True)
|
|
if msg.chat_type in ("group", "channel"):
|
|
if not self._security_config.is_allowed_group(msg.identity.channel_chat_id):
|
|
return SecurityDecision(
|
|
allowed=False,
|
|
reason=f"group_policy={self._security_config.group_policy.value}",
|
|
)
|
|
if self._security_config.require_mention and not msg.mentions.is_bot_mentioned:
|
|
return SecurityDecision(
|
|
allowed=False,
|
|
reason="require_mention: bot not mentioned in channel message",
|
|
)
|
|
return SecurityDecision(allowed=True)
|
|
return SecurityDecision(allowed=True)
|
|
|
|
async def _enqueue_message(self, msg: ChannelMessage) -> None:
|
|
try:
|
|
self._inbound_queue.put_nowait(msg)
|
|
except asyncio.QueueFull:
|
|
logger.warning("Slack inbound queue full, dropping oldest message")
|
|
try:
|
|
self._inbound_queue.get_nowait()
|
|
self._inbound_queue.put_nowait(msg)
|
|
except (asyncio.QueueEmpty, asyncio.QueueFull):
|
|
pass
|
|
|
|
async def _handle_message_with_security(self, msg: ChannelMessage) -> None:
|
|
decision = await self._check_security(msg)
|
|
if not decision.allowed:
|
|
if decision.requires_pairing and decision.pairing_code:
|
|
logger.info(
|
|
f"Slack pairing required for user {msg.identity.channel_user_id}, code: {decision.pairing_code}"
|
|
)
|
|
await self._send_pairing_message(msg, decision.pairing_code)
|
|
else:
|
|
logger.debug(f"Slack security blocked message from {msg.identity.channel_user_id}: {decision.reason}")
|
|
return
|
|
|
|
await self._enqueue_message(msg)
|
|
await self._handle_message(msg)
|
|
|
|
async def _send_pairing_message(self, msg: ChannelMessage, code: str) -> None:
|
|
if not self._client:
|
|
return
|
|
try:
|
|
await self._client.chat_postMessage(
|
|
channel=msg.identity.channel_chat_id,
|
|
text=(
|
|
f"要通过此 Bot 进行对话，请先完成配对验证。\n配对码：`{code}`\n请将此配对码发送给管理员以完成配对。"
|
|
),
|
|
mrkdwn=True,
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Failed to send pairing message: {e}")
|
|
|
|
async def create_channel(self, name: str, *, is_private: bool = False) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
result = await self._client.conversations_create(name=name, is_private=is_private)
|
|
channel = result.get("channel", {})
|
|
return DeliveryResult(
|
|
success=result.get("ok", False),
|
|
message_id=channel.get("id"),
|
|
error=result.get("error"),
|
|
)
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
async def archive_channel(self, chat_id: str) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
result = await self._client.conversations_archive(channel=chat_id)
|
|
return DeliveryResult(
|
|
success=result.get("ok", False),
|
|
error=result.get("error"),
|
|
)
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
async def unarchive_channel(self, chat_id: str) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
result = await self._client.conversations_unarchive(channel=chat_id)
|
|
return DeliveryResult(
|
|
success=result.get("ok", False),
|
|
error=result.get("error"),
|
|
)
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
async def rename_channel(self, chat_id: str, name: str) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
result = await self._client.conversations_rename(channel=chat_id, name=name)
|
|
return DeliveryResult(
|
|
success=result.get("ok", False),
|
|
error=result.get("error"),
|
|
)
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
async def invite_users(self, chat_id: str, users: list[str]) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
result = await self._client.conversations_invite(channel=chat_id, users=users)
|
|
return DeliveryResult(
|
|
success=result.get("ok", False),
|
|
error=result.get("error"),
|
|
)
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
async def kick_user(self, chat_id: str, user_id: str) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
result = await self._client.conversations_kick(channel=chat_id, user=user_id)
|
|
return DeliveryResult(
|
|
success=result.get("ok", False),
|
|
error=result.get("error"),
|
|
)
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
async def join_channel(self, chat_id: str) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
result = await self._client.conversations_join(channel=chat_id)
|
|
return DeliveryResult(
|
|
success=result.get("ok", False),
|
|
error=result.get("error"),
|
|
)
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
async def leave_channel(self, chat_id: str) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
result = await self._client.conversations_leave(channel=chat_id)
|
|
return DeliveryResult(
|
|
success=result.get("ok", False),
|
|
error=result.get("error"),
|
|
)
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
async def get_channel_info(self, chat_id: str) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
result = await self._client.conversations_info(channel=chat_id)
|
|
channel = result.get("channel", {})
|
|
return DeliveryResult(
|
|
success=result.get("ok", False),
|
|
metadata={"channel": channel},
|
|
error=result.get("error"),
|
|
)
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
async def list_channel_members(self, chat_id: str) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
result = await self._client.conversations_members(channel=chat_id)
|
|
members = result.get("members", [])
|
|
return DeliveryResult(
|
|
success=result.get("ok", False),
|
|
metadata={"members": members, "count": len(members)},
|
|
error=result.get("error"),
|
|
)
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
async def set_channel_topic(self, chat_id: str, topic: str) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
result = await self._client.conversations_setTopic(channel=chat_id, topic=topic)
|
|
return DeliveryResult(
|
|
success=result.get("ok", False),
|
|
error=result.get("error"),
|
|
)
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
async def set_channel_purpose(self, chat_id: str, purpose: str) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
result = await self._client.conversations_setPurpose(channel=chat_id, purpose=purpose)
|
|
return DeliveryResult(
|
|
success=result.get("ok", False),
|
|
error=result.get("error"),
|
|
)
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
async def unfurl_link(self, chat_id: str, ts: str, unfurls: dict[str, Any]) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
result = await self._client.chat_unfurl(channel=chat_id, ts=ts, unfurls=unfurls)
|
|
return DeliveryResult(
|
|
success=result.get("ok", False),
|
|
error=result.get("error"),
|
|
)
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
async def list_users(self, *, limit: int = 200, cursor: str = "") -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
result = await self._client.users_list(limit=limit, cursor=cursor or None)
|
|
members = result.get("members", [])
|
|
return DeliveryResult(
|
|
success=result.get("ok", False),
|
|
metadata={
|
|
"members": members,
|
|
"count": len(members),
|
|
"next_cursor": result.get("response_metadata", {}).get("next_cursor", ""),
|
|
},
|
|
error=result.get("error"),
|
|
)
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
async def list_channels(self, *, limit: int = 200, cursor: str = "") -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
result = await self._client.conversations_list(
|
|
limit=limit,
|
|
cursor=cursor or None,
|
|
types="public_channel,private_channel",
|
|
)
|
|
channels = result.get("channels", [])
|
|
return DeliveryResult(
|
|
success=result.get("ok", False),
|
|
metadata={
|
|
"channels": channels,
|
|
"count": len(channels),
|
|
"next_cursor": result.get("response_metadata", {}).get("next_cursor", ""),
|
|
},
|
|
error=result.get("error"),
|
|
)
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
async def close_poll(self, poll_id: str) -> DeliveryResult:
|
|
poll = self._poll_mgr.get_poll(poll_id)
|
|
if not poll:
|
|
return DeliveryResult(success=False, error=f"Poll not found: {poll_id}")
|
|
if poll.is_closed:
|
|
return DeliveryResult(success=False, error="Poll already closed")
|
|
|
|
self._poll_mgr.close_poll(poll_id)
|
|
|
|
if self._status == ChannelStatus.CONNECTED and self._client and poll.message_ts:
|
|
try:
|
|
results = poll.results()
|
|
result_lines = [f"*📊 投票已关闭: {poll.question}*", ""]
|
|
for opt in results["options"]:
|
|
bar = "█" * opt["count"] if opt["count"] > 0 else "▁"
|
|
result_lines.append(f"• {opt['text']}: {opt['count']} 票 {bar}")
|
|
result_lines.append(f"\n_总计 {results['total_votes']} 票_")
|
|
await self._client.chat_update(
|
|
channel=poll.chat_id,
|
|
ts=poll.message_ts,
|
|
text="\n".join(result_lines),
|
|
blocks=None,
|
|
)
|
|
except SlackApiError as e:
|
|
logger.warning(f"Failed to update closed poll message: {e}")
|
|
|
|
return DeliveryResult(success=True, metadata={"results": poll.results()})
|
|
|
|
async def list_polls(self, chat_id: str | None = None) -> DeliveryResult:
|
|
polls = self._poll_mgr.list_polls(chat_id)
|
|
return DeliveryResult(success=True, metadata={"polls": polls, "count": len(polls)})
|
|
|
|
async def open_modal(self, trigger_id: str, view: dict) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
result = await self._client.views_open(trigger_id=trigger_id, view=view)
|
|
ok = result.get("ok", False)
|
|
return DeliveryResult(
|
|
success=ok,
|
|
error=result.get("error") if not ok else None,
|
|
metadata={"view": result.get("view", {}) if ok else None},
|
|
)
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
async def update_modal(self, view_id: str, view: dict, *, hash_val: str = "") -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
params: dict[str, Any] = {"view_id": view_id, "view": view}
|
|
if hash_val:
|
|
params["hash"] = hash_val
|
|
result = await self._client.views_update(**params)
|
|
ok = result.get("ok", False)
|
|
return DeliveryResult(
|
|
success=ok,
|
|
error=result.get("error") if not ok else None,
|
|
metadata={"view": result.get("view", {}) if ok else None},
|
|
)
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
async def push_modal(self, trigger_id: str, view: dict) -> DeliveryResult:
|
|
if self._status != ChannelStatus.CONNECTED or not self._client:
|
|
return DeliveryResult(success=False, error="Slack not connected")
|
|
|
|
try:
|
|
result = await self._client.views_push(trigger_id=trigger_id, view=view)
|
|
ok = result.get("ok", False)
|
|
return DeliveryResult(
|
|
success=ok,
|
|
error=result.get("error") if not ok else None,
|
|
metadata={"view": result.get("view", {}) if ok else None},
|
|
)
|
|
except SlackApiError as e:
|
|
return DeliveryResult(success=False, error=str(e))
|
|
|
|
def _check_bot_mention(self, text: str) -> bool:
|
|
if not text or not self._bot_user_id:
|
|
return False
|
|
return f"<@{self._bot_user_id}>" in text
|
|
|
|
def _extract_mention_ids(self, text: str) -> list[str]:
|
|
if not text:
|
|
return []
|
|
return re.findall(r"<@([A-Z0-9]+)>", text)
|
|
|
|
def _resolve_file_message_type(self, files: list) -> MessageType:
|
|
if not files:
|
|
return MessageType.TEXT
|
|
f = files[0]
|
|
mimetype = f.get("mimetype", "")
|
|
if mimetype.startswith("image/"):
|
|
return MessageType.IMAGE
|
|
if mimetype.startswith("video/"):
|
|
return MessageType.VIDEO
|
|
if mimetype.startswith("audio/"):
|
|
return MessageType.AUDIO
|
|
return MessageType.FILE
|
|
|
|
def _extract_attachments(self, files: list) -> list[Attachment]:
|
|
result = []
|
|
for f in files:
|
|
mt = f.get("mimetype", "application/octet-stream")
|
|
att_type = "file"
|
|
if mt.startswith("image/"):
|
|
att_type = "image"
|
|
elif mt.startswith("video/"):
|
|
att_type = "video"
|
|
elif mt.startswith("audio/"):
|
|
att_type = "audio"
|
|
|
|
result.append(
|
|
Attachment(
|
|
type=att_type,
|
|
url=f.get("url_private", ""),
|
|
mime_type=mt,
|
|
filename=f.get("name", ""),
|
|
size_bytes=f.get("size", 0),
|
|
)
|
|
)
|
|
return result
|
|
|
|
def _register_socket_events(self) -> None:
|
|
if not self._socket_handler:
|
|
return
|
|
|
|
handler = self._socket_handler
|
|
|
|
@handler.on("message")
|
|
async def _on_message(client: SocketModeClient, req: SocketModeRequest):
|
|
event = req.payload.get("event", {})
|
|
subtype = event.get("subtype")
|
|
|
|
if subtype == "message_changed":
|
|
await self._handle_message_changed(event, req)
|
|
return
|
|
if subtype == "message_deleted":
|
|
await self._handle_message_deleted(event, req)
|
|
return
|
|
if subtype in ("channel_join", "group_join"):
|
|
await self._handle_member_event(event, req, EventType.MEMBER_JOINED)
|
|
return
|
|
if subtype in ("channel_leave", "group_leave"):
|
|
await self._handle_member_event(event, req, EventType.MEMBER_LEFT)
|
|
return
|
|
if subtype and subtype not in ("file_share", "thread_broadcast"):
|
|
return
|
|
if event.get("bot_id") and not subtype:
|
|
return
|
|
|
|
try:
|
|
event_id = event.get("event_ts", event.get("ts", ""))
|
|
if event_id and self._is_duplicate_event(event_id):
|
|
return
|
|
|
|
channel = event.get("channel", "")
|
|
if channel and not channel.startswith("D"):
|
|
self._last_channel_event_at = time.monotonic()
|
|
|
|
msg = self.normalize_inbound(
|
|
{
|
|
"event": event,
|
|
"event_type": req.payload.get("event", {}).get("type", "unknown"),
|
|
"team_id": req.payload.get("team_id", ""),
|
|
}
|
|
)
|
|
await self._handle_message_with_security(msg)
|
|
except Exception as e:
|
|
logger.error(f"Slack message event error: {e}", exc_info=True)
|
|
|
|
@handler.on("reaction_added")
|
|
async def _on_reaction_added(client: SocketModeClient, req: SocketModeRequest):
|
|
try:
|
|
event_data = {
|
|
"reaction": req.payload.get("reaction", ""),
|
|
"user": req.payload.get("user", ""),
|
|
"item": req.payload.get("item", {}),
|
|
"event_ts": req.payload.get("event_ts", ""),
|
|
}
|
|
item = req.payload.get("item", {})
|
|
msg = self._build_system_event(
|
|
req,
|
|
event_type=EventType.MESSAGE_UPDATED,
|
|
content=f"reaction_added:{event_data['reaction']}",
|
|
channel=item.get("channel", ""),
|
|
ts=item.get("ts", ""),
|
|
user=req.payload.get("user", ""),
|
|
metadata={"reaction_event": event_data},
|
|
)
|
|
await self._handle_message_with_security(msg)
|
|
|
|
if self._reaction_notify_cfg.should_notify(event_data["reaction"]):
|
|
await self._notify_bot_owner_of_reaction(event_data["reaction"], req)
|
|
except Exception as e:
|
|
logger.error(f"Slack reaction_added error: {e}", exc_info=True)
|
|
|
|
@handler.on("reaction_removed")
|
|
async def _on_reaction_removed(client: SocketModeClient, req: SocketModeRequest):
|
|
try:
|
|
item = req.payload.get("item", {})
|
|
event_data = {
|
|
"reaction": req.payload.get("reaction", ""),
|
|
"user": req.payload.get("user", ""),
|
|
"item": item,
|
|
"event_ts": req.payload.get("event_ts", ""),
|
|
}
|
|
msg = self._build_system_event(
|
|
req,
|
|
event_type=EventType.MESSAGE_UPDATED,
|
|
content=f"reaction_removed:{event_data['reaction']}",
|
|
channel=item.get("channel", ""),
|
|
ts=item.get("ts", ""),
|
|
user=req.payload.get("user", ""),
|
|
metadata={"reaction_event": event_data},
|
|
)
|
|
await self._handle_message_with_security(msg)
|
|
except Exception as e:
|
|
logger.error(f"Slack reaction_removed error: {e}", exc_info=True)
|
|
|
|
@handler.on("interactive")
|
|
async def _on_interactive(client: SocketModeClient, req: SocketModeRequest):
|
|
payload = req.payload
|
|
try:
|
|
actions = payload.get("actions", [])
|
|
action_id = actions[0].get("action_id", "") if actions else ""
|
|
|
|
if action_id.startswith("poll_"):
|
|
await self._handle_poll_vote(payload, action_id, req)
|
|
return
|
|
|
|
if action_id in ("exec_approve", "exec_reject"):
|
|
await self._handle_approval_interaction(payload, action_id)
|
|
return
|
|
|
|
msg = self.normalize_inbound(
|
|
{
|
|
"event": {
|
|
"type": "interactive",
|
|
"user": payload.get("user", {}).get("id", ""),
|
|
"channel": payload.get("channel", {}).get("id", ""),
|
|
"ts": payload.get("message", {}).get("ts", ""),
|
|
},
|
|
"event_type": "interactive",
|
|
"team_id": payload.get("team", {}).get("id", ""),
|
|
"interaction": payload,
|
|
}
|
|
)
|
|
msg.event_type = EventType.CARD_ACTION
|
|
msg.metadata["interaction_type"] = payload.get("type", "")
|
|
await self._handle_message_with_security(msg)
|
|
except Exception as e:
|
|
logger.error(f"Slack interactive event error: {e}", exc_info=True)
|
|
|
|
@handler.on("view_submission")
|
|
async def _on_view_submission(client: SocketModeClient, req: SocketModeRequest):
|
|
payload = req.payload
|
|
try:
|
|
view = payload.get("view", {})
|
|
user = payload.get("user", {})
|
|
view_state = view.get("state", {}).get("values", {})
|
|
|
|
msg = self.normalize_inbound(
|
|
{
|
|
"event": {
|
|
"type": "view_submission",
|
|
"user": user.get("id", ""),
|
|
"channel": user.get("id", ""),
|
|
"ts": view.get("id", ""),
|
|
},
|
|
"event_type": "view_submission",
|
|
"team_id": payload.get("team", {}).get("id", ""),
|
|
"interaction": payload,
|
|
}
|
|
)
|
|
msg.event_type = EventType.CARD_ACTION
|
|
msg.metadata["interaction_type"] = "view_submission"
|
|
msg.metadata["view"] = view
|
|
msg.metadata["view_state"] = view_state
|
|
msg.metadata["callback_id"] = view.get("callback_id", "")
|
|
await self._handle_message_with_security(msg)
|
|
except Exception as e:
|
|
logger.error(f"Slack view_submission error: {e}", exc_info=True)
|
|
|
|
@handler.on("error")
|
|
async def _on_error(client: SocketModeClient, req: SocketModeRequest):
|
|
err_msg = str(req.payload)
|
|
logger.error(f"Slack Socket Mode error: {err_msg}")
|
|
|
|
if "too_many_websockets" in err_msg:
|
|
logger.warning(
|
|
f"too_many_websockets detected, "
|
|
f"waiting for Slack server-side GC (reconnect #{self._reconnect_count})"
|
|
)
|
|
self._status = ChannelStatus.RECONNECTING
|
|
return
|
|
|
|
if self._status == ChannelStatus.CONNECTED:
|
|
self._status = ChannelStatus.RECONNECTING
|
|
|
|
@handler.on("connected")
|
|
async def _on_connected(client: SocketModeClient, req: SocketModeRequest):
|
|
logger.info("Slack Socket Mode WebSocket connected")
|
|
self._status = ChannelStatus.CONNECTED
|
|
self._connected_at = datetime.now(UTC)
|
|
self._connected_event.set()
|
|
self._last_channel_event_at = time.monotonic()
|
|
self._reconnect_count = 0
|
|
|
|
@handler.on("disconnect")
|
|
async def _on_disconnect(client: SocketModeClient, req: SocketModeRequest):
|
|
logger.warning("Slack Socket Mode WebSocket disconnected")
|
|
if self._shutting_down:
|
|
return
|
|
if self._status == ChannelStatus.CONNECTED:
|
|
self._status = ChannelStatus.RECONNECTING
|
|
self._reconnect_count += 1
|
|
|
|
@handler.on("slash_commands")
|
|
async def _on_slash_command(client: SocketModeClient, req: SocketModeRequest):
|
|
try:
|
|
command = req.payload.get("command", "")
|
|
text = req.payload.get("text", "")
|
|
user_id = req.payload.get("user_id", "")
|
|
channel_id = req.payload.get("channel_id", "")
|
|
team_id = req.payload.get("team_id", "")
|
|
trigger_id = req.payload.get("trigger_id", "")
|
|
|
|
if not command:
|
|
return
|
|
|
|
result = await self._cmd_registry.handle_command(command, text, user_id, channel_id)
|
|
|
|
response_url = req.payload.get("response_url", "")
|
|
if response_url and self._client:
|
|
try:
|
|
import aiohttp
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
await session.post(response_url, json=result)
|
|
except Exception as e:
|
|
logger.error(f"Failed to send slash command response: {e}")
|
|
|
|
msg = self._build_system_event(
|
|
req,
|
|
event_type=EventType.SYSTEM_EVENT,
|
|
content=f"slash_command:{command} {text}",
|
|
channel=channel_id,
|
|
ts=req.payload.get("event_ts", ""),
|
|
user=user_id,
|
|
metadata={
|
|
"slash_command": command,
|
|
"text": text,
|
|
"trigger_id": trigger_id,
|
|
"team_id": team_id,
|
|
},
|
|
)
|
|
await self._handle_message_with_security(msg)
|
|
except Exception as e:
|
|
logger.error(f"Slack slash_command error: {e}", exc_info=True)
|
|
|
|
@handler.on("pin_added")
|
|
async def _on_pin_added(client: SocketModeClient, req: SocketModeRequest):
|
|
try:
|
|
item = req.payload.get("item", {})
|
|
msg = self._build_system_event(
|
|
req,
|
|
event_type=EventType.MESSAGE_UPDATED,
|
|
content="pin_added",
|
|
channel=item.get("channel", ""),
|
|
ts=item.get("message", {}).get("ts", ""),
|
|
user=req.payload.get("user", ""),
|
|
metadata={
|
|
"pin_event": "pin_added",
|
|
"item": item,
|
|
"event_ts": req.payload.get("event_ts", ""),
|
|
},
|
|
)
|
|
await self._handle_message_with_security(msg)
|
|
except Exception as e:
|
|
logger.error(f"Slack pin_added error: {e}", exc_info=True)
|
|
|
|
@handler.on("pin_removed")
|
|
async def _on_pin_removed(client: SocketModeClient, req: SocketModeRequest):
|
|
try:
|
|
item = req.payload.get("item", {})
|
|
msg = self._build_system_event(
|
|
req,
|
|
event_type=EventType.MESSAGE_UPDATED,
|
|
content="pin_removed",
|
|
channel=item.get("channel", ""),
|
|
ts=item.get("message", {}).get("ts", ""),
|
|
user=req.payload.get("user", ""),
|
|
metadata={
|
|
"pin_event": "pin_removed",
|
|
"item": item,
|
|
"event_ts": req.payload.get("event_ts", ""),
|
|
},
|
|
)
|
|
await self._handle_message_with_security(msg)
|
|
except Exception as e:
|
|
logger.error(f"Slack pin_removed error: {e}", exc_info=True)
|
|
|
|
@handler.on("channel_created")
|
|
async def _on_channel_created(client: SocketModeClient, req: SocketModeRequest):
|
|
try:
|
|
channel_data = req.payload.get("channel", {})
|
|
msg = self._build_system_event(
|
|
req,
|
|
event_type=EventType.SYSTEM_EVENT,
|
|
content=f"channel_created:{channel_data.get('name', '')}",
|
|
channel=channel_data.get("id", ""),
|
|
ts=req.payload.get("event_ts", ""),
|
|
user=channel_data.get("creator", ""),
|
|
metadata={
|
|
"channel_event": "channel_created",
|
|
"channel_data": channel_data,
|
|
},
|
|
)
|
|
await self._handle_message_with_security(msg)
|
|
except Exception as e:
|
|
logger.error(f"Slack channel_created error: {e}", exc_info=True)
|
|
|
|
@handler.on("channel_rename")
|
|
async def _on_channel_rename(client: SocketModeClient, req: SocketModeRequest):
|
|
try:
|
|
channel_data = req.payload.get("channel", {})
|
|
msg = self._build_system_event(
|
|
req,
|
|
event_type=EventType.SYSTEM_EVENT,
|
|
content=f"channel_rename:{channel_data.get('name', '')}",
|
|
channel=channel_data.get("id", ""),
|
|
ts=req.payload.get("event_ts", ""),
|
|
metadata={
|
|
"channel_event": "channel_rename",
|
|
"channel_data": channel_data,
|
|
},
|
|
)
|
|
await self._handle_message_with_security(msg)
|
|
except Exception as e:
|
|
logger.error(f"Slack channel_rename error: {e}", exc_info=True)
|
|
|
|
@handler.on("channel_archive")
|
|
async def _on_channel_archive(client: SocketModeClient, req: SocketModeRequest):
|
|
try:
|
|
channel = req.payload.get("channel", "")
|
|
msg = self._build_system_event(
|
|
req,
|
|
event_type=EventType.SYSTEM_EVENT,
|
|
content=f"channel_archive:{channel}",
|
|
channel=channel,
|
|
ts=req.payload.get("event_ts", ""),
|
|
user=req.payload.get("user", ""),
|
|
metadata={
|
|
"channel_event": "channel_archive",
|
|
"channel_id": channel,
|
|
},
|
|
)
|
|
await self._handle_message_with_security(msg)
|
|
except Exception as e:
|
|
logger.error(f"Slack channel_archive error: {e}", exc_info=True)
|
|
|
|
@handler.on("channel_unarchive")
|
|
async def _on_channel_unarchive(client: SocketModeClient, req: SocketModeRequest):
|
|
try:
|
|
channel = req.payload.get("channel", "")
|
|
msg = self._build_system_event(
|
|
req,
|
|
event_type=EventType.SYSTEM_EVENT,
|
|
content=f"channel_unarchive:{channel}",
|
|
channel=channel,
|
|
ts=req.payload.get("event_ts", ""),
|
|
user=req.payload.get("user", ""),
|
|
metadata={
|
|
"channel_event": "channel_unarchive",
|
|
"channel_id": channel,
|
|
},
|
|
)
|
|
await self._handle_message_with_security(msg)
|
|
except Exception as e:
|
|
logger.error(f"Slack channel_unarchive error: {e}", exc_info=True)
|
|
|
|
@handler.on("app_home_opened")
|
|
async def _on_app_home_opened(client: SocketModeClient, req: SocketModeRequest):
|
|
try:
|
|
user = req.payload.get("user", "")
|
|
msg = self._build_system_event(
|
|
req,
|
|
event_type=EventType.SYSTEM_EVENT,
|
|
content="app_home_opened",
|
|
channel=user,
|
|
ts=req.payload.get("event_ts", ""),
|
|
user=user,
|
|
metadata={
|
|
"forum_event": "app_home_opened",
|
|
"tab": req.payload.get("tab", "home"),
|
|
"view": req.payload.get("view", {}),
|
|
},
|
|
)
|
|
await self._handle_message_with_security(msg)
|
|
except Exception as e:
|
|
logger.error(f"Slack app_home_opened error: {e}", exc_info=True)
|
|
|
|
@handler.on("link_shared")
|
|
async def _on_link_shared(client: SocketModeClient, req: SocketModeRequest):
|
|
try:
|
|
links = req.payload.get("links", [])
|
|
user = req.payload.get("user", "")
|
|
channel = req.payload.get("channel", "")
|
|
message_ts = req.payload.get("message_ts", "")
|
|
link_urls = [link.get("url", "") for link in links]
|
|
msg = self._build_system_event(
|
|
req,
|
|
event_type=EventType.MESSAGE_RECEIVED,
|
|
content=" ".join(link_urls),
|
|
channel=channel,
|
|
ts=message_ts,
|
|
user=user,
|
|
metadata={
|
|
"forum_event": "link_shared",
|
|
"links": links,
|
|
"unfurl_id": req.payload.get("unfurl_id", ""),
|
|
},
|
|
)
|
|
await self._handle_message_with_security(msg)
|
|
except Exception as e:
|
|
logger.error(f"Slack link_shared error: {e}", exc_info=True)
|
|
|
|
@handler.on("thread_broadcast")
|
|
async def _on_thread_broadcast(client: SocketModeClient, req: SocketModeRequest):
|
|
try:
|
|
channel = req.payload.get("channel", "")
|
|
thread_ts = req.payload.get("thread_ts", req.payload.get("message", {}).get("thread_ts", ""))
|
|
msg = self._build_system_event(
|
|
req,
|
|
event_type=EventType.MESSAGE_UPDATED,
|
|
content=f"thread_broadcast:{thread_ts}",
|
|
channel=channel,
|
|
ts=req.payload.get("ts", req.payload.get("event_ts", "")),
|
|
user=req.payload.get("user", ""),
|
|
metadata={
|
|
"thread_event": "thread_broadcast",
|
|
"thread_ts": thread_ts,
|
|
"root_message": req.payload.get("root_message", req.payload.get("message", {})),
|
|
},
|
|
)
|
|
await self._handle_message_with_security(msg)
|
|
except Exception as e:
|
|
logger.error(f"Slack thread_broadcast error: {e}", exc_info=True)
|
|
|
|
async def _handle_message_changed(self, event: dict, req: SocketModeRequest) -> None:
|
|
msg_data = event.get("message", {})
|
|
previous = event.get("previous_message", {})
|
|
original_msg = self.normalize_inbound(
|
|
{
|
|
"event": {**previous, "channel": event.get("channel", ""), "ts": previous.get("ts", "")},
|
|
"event_type": "message",
|
|
"team_id": req.payload.get("team_id", ""),
|
|
}
|
|
)
|
|
original_msg.event_type = EventType.MESSAGE_UPDATED
|
|
original_msg.content = msg_data.get("text", "")
|
|
original_msg.metadata["edited_message"] = msg_data
|
|
original_msg.metadata["previous_message"] = previous
|
|
await self._handle_message_with_security(original_msg)
|
|
|
|
async def _handle_message_deleted(self, event: dict, req: SocketModeRequest) -> None:
|
|
ts = event.get("deleted_ts", "")
|
|
channel = event.get("channel", "")
|
|
msg = self._build_system_event(
|
|
req,
|
|
event_type=EventType.MESSAGE_DELETED,
|
|
content="",
|
|
channel=channel,
|
|
ts=ts,
|
|
)
|
|
msg.metadata["deleted_ts"] = ts
|
|
await self._handle_message_with_security(msg)
|
|
|
|
async def _handle_member_event(self, event: dict, req: SocketModeRequest, event_type: EventType) -> None:
|
|
joined_user = event.get("user", event.get("inviter", ""))
|
|
channel = event.get("channel", "")
|
|
msg = self._build_system_event(
|
|
req,
|
|
event_type=event_type,
|
|
content=joined_user,
|
|
channel=channel,
|
|
ts=event.get("ts", ""),
|
|
user=joined_user,
|
|
)
|
|
await self._handle_message_with_security(msg)
|
|
|
|
async def _handle_poll_vote(self, payload: dict, action_id: str, req: SocketModeRequest) -> None:
|
|
parts = action_id.split("_", 2)
|
|
if len(parts) < 3:
|
|
return
|
|
poll_id = parts[1]
|
|
try:
|
|
option_index = int(parts[2])
|
|
except ValueError:
|
|
return
|
|
|
|
user_id = payload.get("user", {}).get("id", "")
|
|
channel_id = payload.get("channel", {}).get("id", "")
|
|
|
|
success = self._poll_mgr.vote_poll(poll_id, option_index, user_id)
|
|
if not success:
|
|
return
|
|
|
|
poll = self._poll_mgr.get_poll(poll_id)
|
|
if not poll:
|
|
return
|
|
|
|
results = poll.results()
|
|
result_lines = [f"*📊 投票: {poll.question}*", ""]
|
|
for opt in results["options"]:
|
|
bar = "█" * opt["count"] if opt["count"] > 0 else "▁"
|
|
result_lines.append(f"• {opt['text']}: {opt['count']} 票 {bar}")
|
|
result_lines.append(f"\n_总计 {results['total_votes']} 票_")
|
|
|
|
if self._client and channel_id and poll.message_ts:
|
|
try:
|
|
await self._client.chat_update(
|
|
channel=channel_id,
|
|
ts=poll.message_ts,
|
|
text="\n".join(result_lines),
|
|
blocks=None,
|
|
)
|
|
except SlackApiError as e:
|
|
logger.warning(f"Failed to update poll results: {e}")
|
|
|
|
response_url = payload.get("response_url", "")
|
|
if response_url and self._client:
|
|
try:
|
|
import aiohttp
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
await session.post(
|
|
response_url,
|
|
json={"text": f"已投票: {poll.options[option_index].text}", "replace_original": False},
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
async def _handle_approval_interaction(self, payload: dict, action_id: str) -> None:
|
|
user_id = payload.get("user", {}).get("id", "")
|
|
channel_id = payload.get("channel", {}).get("id", "")
|
|
message_ts = payload.get("message", {}).get("ts", "")
|
|
|
|
result = await self.handle_approval_callback(action_id, user_id, message_ts, channel_id)
|
|
if not result.success:
|
|
logger.warning(f"Approval callback failed: {result.error}")
|
|
|
|
async def _notify_bot_owner_of_reaction(self, emoji: str, req: SocketModeRequest) -> None:
|
|
if not self._client or not self._bot_user_id:
|
|
return
|
|
try:
|
|
user = req.payload.get("user", "")
|
|
item = req.payload.get("item", {})
|
|
channel = item.get("channel", "unknown")
|
|
await self._client.chat_postMessage(
|
|
channel=self._bot_user_id,
|
|
text=f":{emoji}: 你的消息在 <#{channel}> 中收到了一条 {emoji} 反应，来自 <@{user}>",
|
|
)
|
|
except Exception as e:
|
|
logger.debug(f"Failed to notify bot owner of reaction: {e}")
|
|
|
|
def _build_system_event(
|
|
self,
|
|
req: SocketModeRequest,
|
|
*,
|
|
event_type: EventType,
|
|
content: str,
|
|
channel: str = "",
|
|
ts: str = "",
|
|
user: str = "",
|
|
metadata: dict | None = None,
|
|
) -> ChannelMessage:
|
|
event = req.payload.get("event", {})
|
|
channel = channel or event.get("channel", "")
|
|
user = user or event.get("user", "")
|
|
ts = ts or event.get("ts", "")
|
|
|
|
chat_type = resolve_chat_type(channel, event)
|
|
channel_chat_id = resolve_chat_id(channel, event)
|
|
|
|
return ChannelMessage(
|
|
identity=ChannelIdentity(
|
|
channel_id=self.channel_id,
|
|
channel_type=self.channel_type,
|
|
channel_user_id=user,
|
|
channel_chat_id=channel_chat_id,
|
|
channel_message_id=f"{channel}:{ts}" if ts else None,
|
|
),
|
|
event_type=event_type,
|
|
chat_type=chat_type,
|
|
content=content,
|
|
metadata={
|
|
**(metadata or {}),
|
|
"channel": channel,
|
|
"ts": ts,
|
|
"team_id": req.payload.get("team_id", self._team_id),
|
|
},
|
|
)
|
|
|
|
def _cleanup_stale_streams(self, now: float) -> None:
|
|
stale_keys = [
|
|
k for k, v in self._streaming_messages.items() if now - v.get("_last_update", 0) > self.STREAM_TTL_SECONDS
|
|
]
|
|
for k in stale_keys:
|
|
self._streaming_messages.pop(k, None)
|
|
logger.debug(f"Cleaned up stale streaming entry: {k}")
|
|
|
|
async def _stream_cleanup_loop(self) -> None:
|
|
while self._status in (ChannelStatus.CONNECTED, ChannelStatus.CONNECTING):
|
|
try:
|
|
await asyncio.sleep(self._stream_cleanup_interval_s)
|
|
async with self._stream_lock:
|
|
self._cleanup_stale_streams(time.monotonic())
|
|
except asyncio.CancelledError:
|
|
break
|
|
except Exception:
|
|
logger.debug("Stream cleanup iteration failed", exc_info=True)
|
|
|
|
async def _channel_heartbeat_loop(self) -> None:
|
|
while self._status in (ChannelStatus.CONNECTED, ChannelStatus.RECONNECTING):
|
|
try:
|
|
await asyncio.sleep(60)
|
|
now = time.monotonic()
|
|
elapsed = now - self._last_channel_event_at
|
|
if (
|
|
self._status == ChannelStatus.CONNECTED
|
|
and elapsed > self.CHANNEL_EVENT_HEARTBEAT_TIMEOUT_S
|
|
and self._last_channel_event_at > 0
|
|
):
|
|
logger.warning(
|
|
f"No channel events for {elapsed:.0f}s, connection may be degraded — triggering reconnect"
|
|
)
|
|
self._status = ChannelStatus.RECONNECTING
|
|
if self._socket_handler:
|
|
try:
|
|
self._socket_handler.disconnect()
|
|
except Exception:
|
|
pass
|
|
self._connected_event.clear()
|
|
if self._socket_task and not self._socket_task.done():
|
|
self._socket_task.cancel()
|
|
try:
|
|
await self._socket_task
|
|
except (asyncio.CancelledError, Exception):
|
|
pass
|
|
self._socket_task = asyncio.create_task(self._socket_handler.connect_async())
|
|
except asyncio.CancelledError:
|
|
break
|
|
except Exception:
|
|
logger.debug("Channel heartbeat iteration failed", exc_info=True)
|
|
|
|
def _is_duplicate_event(self, event_id: str) -> bool:
|
|
if event_id in self._processed_event_ids:
|
|
self._processed_event_ids.move_to_end(event_id)
|
|
logger.debug(f"Skipping duplicate event: {event_id}")
|
|
return True
|
|
self._processed_event_ids[event_id] = None
|
|
if len(self._processed_event_ids) > self.MAX_PROCESSED_EVENT_IDS:
|
|
self._processed_event_ids.popitem(last=False)
|
|
return False
|
|
|
|
def should_recover_thread(self, peer_kind: str = "direct") -> bool:
|
|
if self._dm_scope == "main" and peer_kind == "direct":
|
|
return False
|
|
return True
|