From ef5483dc1a9670a329cbf12686521abbd37ea85b Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Wed, 13 May 2026 16:13:48 +0800 Subject: [PATCH] =?UTF-8?q?refactor(qqbot):=20=E9=87=8D=E6=9E=84QQ?= =?UTF-8?q?=E6=9C=BA=E5=99=A8=E4=BA=BA=E9=80=82=E9=85=8D=E5=99=A8=E4=BB=A3?= =?UTF-8?q?=E7=A0=81=EF=BC=8C=E4=BC=98=E5=8C=96=E5=A4=9A=E9=A1=B9=E5=8A=9F?= =?UTF-8?q?=E8=83=BD=E4=B8=8E=E7=BB=93=E6=9E=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 主要变更： 1. 修复速率限流器使用setdefault替代重复创建令牌桶 2. 重构交互注册表匹配逻辑，优化精确匹配查找 3. 重构去重缓存逻辑，移到适配器实例方法 4. 重构发送URL解析，增加合法性校验并拆分公共方法 5. 优化流式消息处理逻辑，简化flush_controller调用 6. 重构群聊类型判断代码，简化语法 7. 修复重连管理器对None类型关闭分类的处理 8. 新增消息缓存、线程模拟器、发送初始化模块 9. 重构凭证备份与会话存储逻辑，支持环境变量指定状态目录 10. 新增配置提示与向导二维码绑定功能 11. 优化媒体上传逻辑，增加重试机制与缓存 12. 新增审批键盘模板构建函数 13. 重构消息格式处理，修正媒体发送字段与长度限制 14. 修复令牌过期时间计算，使用time.time替代monotonic 15. 新增群组激活缓冲区与用户追踪器增强功能 16. 修复换行符问题，统一文件结尾格式 --- .../yuxi/channels/adapters/qqbot/adapter.py | 257 ++++++++++-- .../channels/adapters/qqbot/ark_templates.py | 44 ++ .../channels/adapters/qqbot/c2c_stream.py | 32 ++ .../adapters/qqbot/commands/__init__.py | 8 +- .../adapters/qqbot/commands/builtin.py | 4 +- .../adapters/qqbot/commands/streaming_cmd.py | 2 +- .../channels/adapters/qqbot/config_hints.py | 388 ++++++++++++++++++ .../adapters/qqbot/credential_backup.py | 38 +- .../yuxi/channels/adapters/qqbot/format.py | 31 +- .../channels/adapters/qqbot/group_buffer.py | 12 +- .../adapters/qqbot/inbound_pipeline.py | 10 +- .../channels/adapters/qqbot/interaction.py | 12 +- .../channels/adapters/qqbot/known_users.py | 69 +++- .../channels/adapters/qqbot/media_upload.py | 33 +- .../channels/adapters/qqbot/message_cache.py | 94 +++++ .../channels/adapters/qqbot/message_queue.py | 2 +- .../channels/adapters/qqbot/rate_limiter.py | 2 +- .../yuxi/channels/adapters/qqbot/reconnect.py | 4 +- .../yuxi/channels/adapters/qqbot/send.py | 34 +- .../channels/adapters/qqbot/sender_init.py | 48 +++ .../channels/adapters/qqbot/session_store.py | 55 ++- .../channels/adapters/qqbot/setup_wizard.py | 51 +++ .../yuxi/channels/adapters/qqbot/streaming.py | 8 +- .../adapters/qqbot/thread_simulator.py | 64 +++ .../yuxi/channels/adapters/qqbot/token.py | 6 +- .../adapters/qqbot/typing_keepalive.py | 2 +- .../channels/adapters/qqbot/voice_send.py | 21 +- 27 files changed, 1218 insertions(+), 113 deletions(-) create mode 100644 backend/package/yuxi/channels/adapters/qqbot/config_hints.py create mode 100644 backend/package/yuxi/channels/adapters/qqbot/message_cache.py create mode 100644 backend/package/yuxi/channels/adapters/qqbot/sender_init.py create mode 100644 backend/package/yuxi/channels/adapters/qqbot/thread_simulator.py diff --git a/backend/package/yuxi/channels/adapters/qqbot/adapter.py b/backend/package/yuxi/channels/adapters/qqbot/adapter.py index 975da6f1..393bd8ef 100644 --- a/backend/package/yuxi/channels/adapters/qqbot/adapter.py +++ b/backend/package/yuxi/channels/adapters/qqbot/adapter.py @@ -14,6 +14,7 @@ from yuxi.channels.base import BaseChannelAdapter from yuxi.channels.capabilities import ChannelCapabilities from yuxi.channels.exceptions import ( ChannelAuthenticationError, + ChannelException, ChannelNotConnectedError, DeliveryFailedError, MessageFormatError, @@ -56,6 +57,7 @@ from .group_buffer import GroupHistoryBuffer from .inbound_pipeline import QQBotInboundPipeline from .interaction import InteractionContext, InteractionRegistry, InteractionSessionManager from .known_users import KnownUserTracker +from .message_cache import MessageCache from .message_queue import MessageQueue from .probe import health_check_dsm from .rate_limiter import RouteRateLimiter @@ -63,6 +65,7 @@ from .reconnect import QQBotReconnectManager, ReconnectState from .reply_dispatcher import ReplyDispatcher from .security import QQBotSecurityPolicy, verify_webhook_ed25519 from .send import MessageSeqManager, send_with_retry +from .sender_init import SenderInfo, build_sender_headers, init_sender from .session_store import SessionRecord, SessionStore from .streaming import ParagraphChunker, stream_content from .token import QQBotTokenManager @@ -122,7 +125,6 @@ class QQBotAdapter(BaseChannelAdapter): self._sandbox = self.config.get("sandbox", False) self._ws: aiohttp.ClientWebSocketResponse | None = None self._seq: int | None = None - self._session_id: str | None = None self._heartbeat_interval: float = 30.0 self._heartbeat_task: asyncio.Task | None = None self._last_heartbeat_ack: float = 0 @@ -179,6 +181,7 @@ class QQBotAdapter(BaseChannelAdapter): register_streaming_command(self._command_registry) self._interaction_registry = InteractionRegistry() self._interaction_sessions = InteractionSessionManager() + self._inbound_pipeline = QQBotInboundPipeline(self) self._approval_manager = ExecApprovalManager() self._tts_provider = TTSProvider( default_voice=self.config.get("tts_default_voice", "zh-CN-XiaoxiaoNeural"), @@ -187,10 +190,22 @@ class QQBotAdapter(BaseChannelAdapter): self._stt_provider = STTProvider.from_config(self.config) self._audio_format_policy = AudioFormatPolicy.from_config(self.config) self._typing_keepalive = self._init_typing_keepalive() + self._sender_info = self._init_sender_info() + self._message_cache = MessageCache( + max_messages=int(self.config.get("message_cache_max", 500)), + ttl_s=float(self.config.get("message_cache_ttl_s", 3600.0)), + ) + self._group_stream_buffers: dict[str, str] = {} + self._group_stream_senders: dict[str, Any] = {} def _init_typing_keepalive(self) -> TypingKeepAlive: return TypingKeepAlive(send_notify_fn=self._send_input_notify) + def _init_sender_info(self) -> SenderInfo: + plugin_version = self.config.get("plugin_version", "1.0.0") + openclaw_version = self.config.get("openclaw_version", "1.0.0") + return init_sender(plugin_version=plugin_version, openclaw_version=openclaw_version) + async def _send_input_notify(self, chat_id: str) -> None: if not self._http_client or not self._token_manager: return @@ -220,6 +235,8 @@ class QQBotAdapter(BaseChannelAdapter): logger.info(f"[QQBot] Starting channel '{self.channel_id}'...") + await self._inbound_pipeline.initialize() + try: app_id = self.config.get("app_id") or os.environ.get("QQBOT_APP_ID", "") app_secret = ( @@ -255,6 +272,32 @@ class QQBotAdapter(BaseChannelAdapter): hello_data = await self._wait_for_hello() self._heartbeat_interval = hello_data.get("heartbeat_interval", 30000) / 1000.0 + session_id = self._reconnect_manager._session_id + last_seq = self._reconnect_manager._last_seq + if session_id and last_seq is not None: + logger.info("[QQBot] Attempting RESUME with session_id=%s seq=%s", session_id[:16], last_seq) + await self._send_resume() + try: + ready_data = await asyncio.wait_for(self._wait_for_ready(), timeout=5.0) + if ready_data: + logger.info("[QQBot] RESUME successful, session recovered") + self._reconnect_manager.mark_connected() + await self._reconnect_manager.transition(ReconnectState.CONNECTED) + self._status = ChannelStatus.CONNECTED + self._c2c_streaming = C2CStreamingController( + send_message_fn=self._send_stream_message, + flush_interval=self.config.get("stream_flush_interval", 0.6), + ) + self._heartbeat_task = asyncio.create_task(self._heartbeat_loop()) + self._receive_task = asyncio.create_task(self._ws_receive_loop()) + self._backup_credentials() + self._save_session() + self._register_runtime_contexts() + logger.info("[QQBot] Channel '%s' resumed successfully", self.channel_id) + return + except (TimeoutError, Exception): + logger.info("[QQBot] RESUME failed, falling back to IDENTIFY") + await self._send_identify() self._bot_info = { @@ -271,13 +314,15 @@ class QQBotAdapter(BaseChannelAdapter): self._receive_task = asyncio.create_task(self._ws_receive_loop()) self._status = ChannelStatus.CONNECTED - await self._reconnect_manager.on_identify_success(self._session_id or "") + await self._reconnect_manager.on_identify_success(self._reconnect_manager.session_id or "") self._reconnect_manager.mark_connected() await self._reconnect_manager.transition(ReconnectState.CONNECTED) self._backup_credentials() self._save_session() + self._register_runtime_contexts() + logger.info( f"[QQBot] Channel '{self.channel_id}' started " f"(sandbox={self._sandbox}, heartbeat={self._heartbeat_interval:.0f}s)" @@ -338,22 +383,25 @@ class QQBotAdapter(BaseChannelAdapter): try: await self._ws.close() except Exception: - pass + logger.debug("[QQBot] Error closing WebSocket during cleanup", exc_info=True) self._ws = None self._seq = None - self._session_id = None # ========== WebSocket 内部逻辑 ========== async def _wait_for_hello(self) -> dict: if not self._ws: raise ChannelAuthenticationError("WebSocket not connected") - async for msg in self._ws: - data = json.loads(msg.data) - op = data.get("op") - if op == Opcode.HELLO: - return data.get("d", {}) - logger.debug(f"[QQBot] Skipping op={op} before HELLO") + try: + async with asyncio.timeout(30): + async for msg in self._ws: + data = json.loads(msg.data) + op = data.get("op") + if op == Opcode.HELLO: + return data.get("d", {}) + logger.debug(f"[QQBot] Skipping op={op} before HELLO") + except TimeoutError: + raise ChannelAuthenticationError("Timeout waiting for HELLO from QQ Bot Gateway (30s)") async def _send_identify(self) -> None: if not self._ws: @@ -378,12 +426,25 @@ class QQBotAdapter(BaseChannelAdapter): "op": Opcode.RESUME, "d": { "token": f"QQBot {token}", - "session_id": self._session_id, - "seq": self._seq, + "session_id": self._reconnect_manager.session_id, + "seq": self._reconnect_manager.last_seq, }, } await self._ws.send_json(payload) + async def _wait_for_ready(self) -> dict | None: + if not self._ws: + return None + async for msg in self._ws: + data = json.loads(msg.data) + op = data.get("op") + t = data.get("t", "") + if op == Opcode.DISPATCH and t in ("READY", "RESUMED"): + return data.get("d", {}) + if op in (Opcode.INVALID_SESSION, Opcode.RECONNECT): + return None + return None + async def _heartbeat_loop(self) -> None: max_misses = 3 while self._status == ChannelStatus.CONNECTED and self._ws and not self._ws.closed: @@ -393,6 +454,7 @@ class QQBotAdapter(BaseChannelAdapter): if self._heartbeat_miss_count >= max_misses: logger.warning(f"[QQBot] Heartbeat missed {max_misses} consecutive ACKs, disconnecting") self._status = ChannelStatus.RECONNECTING + await self._reconnect_manager.on_disconnect(None) break send_time = time.monotonic() @@ -418,6 +480,7 @@ class QQBotAdapter(BaseChannelAdapter): except Exception as e: logger.warning(f"[QQBot] Heartbeat failed: {e}") self._status = ChannelStatus.RECONNECTING + await self._reconnect_manager.on_disconnect(None) break async def _ws_receive_loop(self) -> None: @@ -474,9 +537,9 @@ class QQBotAdapter(BaseChannelAdapter): self._status = ChannelStatus.ERROR async def _dispatch_event(self, event_type: str, event_data: dict) -> None: - pipeline = QQBotInboundPipeline(self) - await pipeline.initialize() - ctx = await pipeline.process(event_type, event_data) + if not self._inbound_pipeline._stages: + await self._inbound_pipeline.initialize() + await self._inbound_pipeline.process(event_type, event_data) def _prune_old_msg_ids(self, now: float) -> None: expired = [mid for mid, ts in self._recent_msg_ids.items() if now - ts >= self._dedup_window_s] @@ -513,8 +576,22 @@ class QQBotAdapter(BaseChannelAdapter): try: await self._typing_keepalive.start(chat_id) await self._message_handler(msg) + except ChannelException as e: + if e.retryable: + logger.warning( + "[QQBot] Retryable error in message handler (chat=%s): %s", + chat_id, + e, + ) + else: + logger.exception( + "[QQBot] Unrecoverable error in message handler (chat=%s)", + chat_id, + ) + await self._send_error_notification(msg, str(e)) except Exception: - logger.exception("[QQBot] Error in message handler") + logger.exception("[QQBot] Unexpected error in message handler (chat=%s)", chat_id) + await self._send_error_notification(msg, "内部错误，请稍后重试") finally: await self._typing_keepalive.stop(chat_id) @@ -636,7 +713,17 @@ class QQBotAdapter(BaseChannelAdapter): logger.exception("[QQBot] Interaction handling failed") return False - # ========== 消息收发 ========== + # ========== 消息收发 ========== + + async def _send_error_notification(self, msg: ChannelMessage, error_msg: str) -> None: + try: + response = ChannelResponse( + identity=msg.identity, + content=f"抱歉，处理您的消息时遇到了问题：{error_msg}", + ) + await self.send(response) + except Exception: + logger.warning("[QQBot] Failed to send error notification to user", exc_info=True) async def send(self, response: ChannelResponse) -> DeliveryResult: if not self._http_client or not self._token_manager: @@ -666,18 +753,93 @@ class QQBotAdapter(BaseChannelAdapter): chat_id, self.config, token_refresh_cb=_refresh_token, + sender_headers=build_sender_headers(self._sender_info), ) try: await self._rate_limiter.acquire("send_message") result = await self._circuit_breaker.call(_do_send) self._last_outbound_at = time.time() + self._message_cache.record_sent(result, chat_id, content=response.content) return result except CircuitBreakerOpenError: return DeliveryResult(success=False, error="Circuit breaker open") except (DeliveryFailedError, Exception) as e: return DeliveryResult(success=False, error=str(e)) + async def send_stream_chunk( + self, + chat_id: str, + msg_id: str, + chunk: str, + finished: bool = False, + ) -> DeliveryResult: + is_group = chat_id.startswith(GROUP_CHAT_PREFIX) if GROUP_CHAT_PREFIX else False + + if not is_group: + from yuxi.channels.models import ChannelIdentity, ChannelResponse + + identity = ChannelIdentity( + channel_id=self.channel_id, + channel_type=self.channel_type, + channel_chat_id=chat_id, + channel_user_id="", + ) + response = ChannelResponse(identity=identity, content=chunk) + return await self.send(response) + + return await self._send_group_stream_chunk(chat_id, msg_id, chunk, finished) + + async def _send_group_stream_chunk( + self, + chat_id: str, + msg_id: str, + chunk: str, + finished: bool = False, + ) -> DeliveryResult: + existing = self._group_stream_buffers.get(chat_id, "") + accumulated = existing + chunk + + if not finished: + paragraphs = accumulated.split("\n\n") + if len(paragraphs) > 1: + for para in paragraphs[:-1]: + if para.strip(): + await self._send_group_paragraph(chat_id, para) + self._group_stream_buffers[chat_id] = paragraphs[-1] + else: + self._group_stream_buffers[chat_id] = accumulated + return DeliveryResult(success=True) + + self._group_stream_buffers.pop(chat_id, None) + self._group_stream_senders.pop(chat_id, None) + + from yuxi.channels.models import ChannelType + + from .streaming import send_blocks_stream + + await send_blocks_stream( + chat_id=chat_id, + text=accumulated, + send_fn=self.send, + channel_id=self.channel_id, + channel_type=ChannelType.QQ_BOT, + chunk_size=self.text_chunk_limit, + ) + return DeliveryResult(success=True) + + async def _send_group_paragraph(self, chat_id: str, text: str) -> DeliveryResult: + from yuxi.channels.models import ChannelIdentity, ChannelResponse + + identity = ChannelIdentity( + channel_id=self.channel_id, + channel_type=self.channel_type, + channel_chat_id=chat_id, + channel_user_id="", + ) + response = ChannelResponse(identity=identity, content=text.strip()) + return await self.send(response) + async def send_media(self, chat_id: str, media_type: str, data: Any) -> DeliveryResult: if not self._http_client or not self._token_manager: return DeliveryResult(success=False, error="Client not initialized") @@ -774,6 +936,20 @@ class QQBotAdapter(BaseChannelAdapter): prompt: str = "", fields: list[dict] | None = None, ) -> DeliveryResult: + if chat_id.startswith(DM_CHAT_PREFIX) or chat_id.startswith(GROUP_CHAT_PREFIX): + logger.debug("[QQBot] Embed not supported in C2C/Group, degrading to text") + return await self.send( + ChannelResponse( + identity=ChannelIdentity( + channel_id=self.channel_id, + channel_type=self.channel_type, + channel_user_id="", + channel_chat_id=chat_id, + ), + content=f"{title}\n{description}" if title else description, + ) + ) + from .ark_templates import build_embed payload = build_embed(title=title, description=description, prompt=prompt, fields=fields) @@ -911,7 +1087,7 @@ class QQBotAdapter(BaseChannelAdapter): async def receive(self) -> AsyncIterator[ChannelMessage]: while self._status == ChannelStatus.CONNECTED: try: - msg = await self._message_queue.get() + msg = await self._message_queue.dequeue() yield msg except asyncio.CancelledError: break @@ -1124,7 +1300,7 @@ class QQBotAdapter(BaseChannelAdapter): ), access_token=self._token_manager.access_token if self._token_manager else "", expires_at=self._token_manager.expires_at if self._token_manager else 0, - session_id=self._session_id or "", + session_id=self._reconnect_manager.session_id or "", sandbox=self._sandbox, ) self._credential_backup.save(snapshot) @@ -1147,7 +1323,7 @@ class QQBotAdapter(BaseChannelAdapter): if not self.config.get("app_secret") and snapshot.app_secret: logger.info("[QQBot] Restored app_secret from credential backup") - self._session_id = snapshot.session_id + self._reconnect_manager._session_id = snapshot.session_id return True def _init_session_store(self) -> None: @@ -1174,7 +1350,7 @@ class QQBotAdapter(BaseChannelAdapter): return record = SessionRecord( - session_id=self._reconnect_manager._session_id or self._session_id or "", + session_id=self._reconnect_manager._session_id or "", last_seq=self._reconnect_manager._last_seq, last_heartbeat=time.monotonic(), identify_at=time.time(), @@ -1299,8 +1475,6 @@ class QQBotAdapter(BaseChannelAdapter): ) if msg_id: response.reply_to_message_id = msg_id - payload = format_outbound(response) - payload["msg_id"] = msg_id return await self.send(response) async def send_group_message( @@ -1320,8 +1494,6 @@ class QQBotAdapter(BaseChannelAdapter): ) if msg_id: response.reply_to_message_id = msg_id - payload = format_outbound(response) - payload["msg_id"] = msg_id return await self.send(response) async def _send_raw_payload(self, chat_id: str, payload: dict) -> DeliveryResult: @@ -1351,9 +1523,9 @@ class QQBotAdapter(BaseChannelAdapter): return None token = await self._token_manager.get_token() api_base = self._token_manager.api_base - from .send import _resolve_send_url + from .send import resolve_send_url - url = _resolve_send_url(api_base, chat_id) + url = resolve_send_url(api_base, chat_id) headers = { "Authorization": f"QQBot {token}", "Content-Type": "application/json", @@ -1449,9 +1621,7 @@ class QQBotAdapter(BaseChannelAdapter): def is_configured(self) -> bool: app_id = self.config.get("app_id") or os.environ.get("QQBOT_APP_ID", "") app_secret = ( - self.config.get("app_secret") - or os.environ.get("QQBOT_CLIENT_SECRET", "") - or self._read_app_secret_file() + self.config.get("app_secret") or os.environ.get("QQBOT_CLIENT_SECRET", "") or self._read_app_secret_file() ) if app_id and app_secret: return True @@ -1469,9 +1639,7 @@ class QQBotAdapter(BaseChannelAdapter): def unconfigured_reason(self) -> str: app_id = self.config.get("app_id") or os.environ.get("QQBOT_APP_ID", "") app_secret = ( - self.config.get("app_secret") - or os.environ.get("QQBOT_CLIENT_SECRET", "") - or self._read_app_secret_file() + self.config.get("app_secret") or os.environ.get("QQBOT_CLIENT_SECRET", "") or self._read_app_secret_file() ) if not app_id: return "missing_app_id" @@ -1501,7 +1669,9 @@ class QQBotAdapter(BaseChannelAdapter): app_id = self.config.get("app_id") or os.environ.get("QQBOT_APP_ID", "") return { "configured": self.is_configured(), - "tokenSource": "env" if os.environ.get("QQBOT_CLIENT_SECRET") else ("file" if self.config.get("client_secret_file") else "config"), + "tokenSource": "env" + if os.environ.get("QQBOT_CLIENT_SECRET") + else ("file" if self.config.get("client_secret_file") else "config"), "running": self._status == ChannelStatus.CONNECTED, "connected": self._ws is not None and not (self._ws.closed if hasattr(self._ws, "closed") else False), "status": self._status.value, @@ -1515,7 +1685,7 @@ class QQBotAdapter(BaseChannelAdapter): "lastInboundAt": getattr(self, "_last_inbound_at", None), "lastOutboundAt": getattr(self, "_last_outbound_at", None), "wsSeq": self._seq, - "sessionId": self._session_id, + "sessionId": self._reconnect_manager.session_id, "heartbeatLatencyMs": self._heartbeat_latency_ms, "reconnectState": self._reconnect_manager.state.name if self._reconnect_manager else "N/A", } @@ -1532,3 +1702,20 @@ class QQBotAdapter(BaseChannelAdapter): def has_native_approval_routing(self) -> bool: return self.config.get("approval_native_routing", True) + + def _register_runtime_contexts(self) -> None: + self._runtime_contexts = { + "approval.native": { + "enabled": self.has_native_approval_routing(), + "channel_id": self.channel_id, + "suppress_local_prompt": self.should_suppress_local_approval_prompt(), + "keyboard_provider": "qqbot_inline", + }, + } + logger.debug( + "[QQBot] Runtime contexts registered: approval.native=%s", + self._runtime_contexts.get("approval.native", {}).get("enabled"), + ) + + def get_runtime_contexts(self) -> dict: + return getattr(self, "_runtime_contexts", {}) diff --git a/backend/package/yuxi/channels/adapters/qqbot/ark_templates.py b/backend/package/yuxi/channels/adapters/qqbot/ark_templates.py index 216d3d82..7157fe22 100644 --- a/backend/package/yuxi/channels/adapters/qqbot/ark_templates.py +++ b/backend/package/yuxi/channels/adapters/qqbot/ark_templates.py @@ -76,5 +76,49 @@ def build_text_card(text: str, buttons: list[dict] | None = None) -> dict: return payload +def build_approval_keyboard( + request_id: str, + action_label: str = "Approve", + reject_label: str = "Reject", + cancel_label: str = "Cancel", +) -> dict: + buttons = [ + { + "id": f"approval_approve_{request_id}", + "render_data": {"label": action_label, "visited_label": "✅ 已批准"}, + "action": { + "type": 2, + "permission": {"type": 2}, + "click_limit": 1, + "data": f"approve:{request_id}", + }, + "group_id": "approval", + }, + { + "id": f"approval_reject_{request_id}", + "render_data": {"label": reject_label, "visited_label": "❌ 已拒绝"}, + "action": { + "type": 2, + "permission": {"type": 2}, + "click_limit": 1, + "data": f"reject:{request_id}", + }, + "group_id": "approval", + }, + { + "id": f"approval_cancel_{request_id}", + "render_data": {"label": cancel_label, "visited_label": "↩️ 已取消"}, + "action": { + "type": 2, + "permission": {"type": 2}, + "click_limit": 1, + "data": f"cancel:{request_id}", + }, + "group_id": "approval", + }, + ] + return {"content": {"rows": [{"buttons": buttons}]}} + + def _kv(key: str, value: str) -> dict[str, str]: return {"key": key, "value": value} diff --git a/backend/package/yuxi/channels/adapters/qqbot/c2c_stream.py b/backend/package/yuxi/channels/adapters/qqbot/c2c_stream.py index 051ca11e..003d9a63 100644 --- a/backend/package/yuxi/channels/adapters/qqbot/c2c_stream.py +++ b/backend/package/yuxi/channels/adapters/qqbot/c2c_stream.py @@ -43,6 +43,9 @@ class StreamCheckpoint: return time.time() - self.timestamp > 600 +BATCH_AFTER_GAP = 1.5 + + @dataclass class FlushController: strategy: FlushStrategy = FlushStrategy.BACKLOG @@ -54,16 +57,23 @@ class FlushController: _buffer: list[str] = field(default_factory=list, repr=False) _last_flush: float = 0.0 _accumulated: str = "" + _flush_in_progress: bool = False + _needs_reflush: bool = False + _last_chunk_time: float = 0.0 def reset(self) -> None: self._buffer = [] self._last_flush = 0.0 self._accumulated = "" self.seq_counter = 0 + self._flush_in_progress = False + self._needs_reflush = False + self._last_chunk_time = 0.0 def feed(self, chunk: str) -> list[str]: self._buffer.append(chunk) self._accumulated += chunk + self._last_chunk_time = time.monotonic() if self.strategy == FlushStrategy.PER_CHAR: return self._flush_all() @@ -75,6 +85,9 @@ class FlushController: if self.strategy == FlushStrategy.INTERVAL and now - self._last_flush >= self.flush_interval: return self._flush_all() + if self._last_chunk_time > 0 and now - self._last_chunk_time >= BATCH_AFTER_GAP: + return self._flush_all() + return [] def flush_remaining(self) -> tuple[list[str], str]: @@ -83,6 +96,19 @@ class FlushController: batches = self._flush_all() if self._buffer else [] return batches, self._accumulated + def mark_flush_in_progress(self) -> bool: + if self._flush_in_progress: + self._needs_reflush = True + return False + self._flush_in_progress = True + return True + + def mark_flush_done(self) -> bool: + self._flush_in_progress = False + needs = self._needs_reflush + self._needs_reflush = False + return needs + def _flush_all(self) -> list[str]: if not self._buffer: return [] @@ -157,6 +183,8 @@ class C2CStreamingController: collected += segment batches = self._flush_controller.feed(segment) for batch in batches: + if not self._flush_controller.mark_flush_in_progress(): + continue self._flush_controller.seq_counter += 1 success = await self._send_stream_chunk( chat_id, @@ -165,6 +193,7 @@ class C2CStreamingController: self._flush_controller.seq_counter, event_id, ) + _needs_reflush = self._flush_controller.mark_flush_done() if success: self._any_chunk_delivered = True else: @@ -176,6 +205,8 @@ class C2CStreamingController: flush_batches, full = self._flush_controller.flush_remaining() for batch in flush_batches: + if not self._flush_controller.mark_flush_in_progress(): + continue self._flush_controller.seq_counter += 1 success = await self._send_stream_chunk( chat_id, @@ -184,6 +215,7 @@ class C2CStreamingController: self._flush_controller.seq_counter, event_id, ) + _needs_reflush = self._flush_controller.mark_flush_done() if success: self._any_chunk_delivered = True diff --git a/backend/package/yuxi/channels/adapters/qqbot/commands/__init__.py b/backend/package/yuxi/channels/adapters/qqbot/commands/__init__.py index 04790d41..ba1325af 100644 --- a/backend/package/yuxi/channels/adapters/qqbot/commands/__init__.py +++ b/backend/package/yuxi/channels/adapters/qqbot/commands/__init__.py @@ -2,4 +2,10 @@ from .builtin import register_builtin_commands from .framework import CommandContext, CommandRegistry, CommandResult from .streaming_cmd import register_streaming_command -__all__ = ["CommandRegistry", "CommandContext", "CommandResult", "register_builtin_commands", "register_streaming_command"] +__all__ = [ + "CommandRegistry", + "CommandContext", + "CommandResult", + "register_builtin_commands", + "register_streaming_command", +] diff --git a/backend/package/yuxi/channels/adapters/qqbot/commands/builtin.py b/backend/package/yuxi/channels/adapters/qqbot/commands/builtin.py index 8e7232da..b1c53447 100644 --- a/backend/package/yuxi/channels/adapters/qqbot/commands/builtin.py +++ b/backend/package/yuxi/channels/adapters/qqbot/commands/builtin.py @@ -46,9 +46,7 @@ async def _cmd_help(ctx: CommandContext) -> CommandResult: return CommandResult(success=True, message="命令系统未初始化。") is_group = getattr(ctx.msg, "chat_type", None) - is_group_val = str(is_group) == "ChatType.GROUP" or ( - hasattr(is_group, "value") and is_group.value == "group" - ) + is_group_val = str(is_group) == "ChatType.GROUP" or (hasattr(is_group, "value") and is_group.value == "group") lines = ["可用命令："] seen: set[str] = set() diff --git a/backend/package/yuxi/channels/adapters/qqbot/commands/streaming_cmd.py b/backend/package/yuxi/channels/adapters/qqbot/commands/streaming_cmd.py index 23ae3eec..048bf77f 100644 --- a/backend/package/yuxi/channels/adapters/qqbot/commands/streaming_cmd.py +++ b/backend/package/yuxi/channels/adapters/qqbot/commands/streaming_cmd.py @@ -56,4 +56,4 @@ def register_streaming_command(registry: CommandRegistry) -> CommandRegistry: usage="/bot-streaming [on|off|status]", aliases=["streaming"], ) - return registry \ No newline at end of file + return registry diff --git a/backend/package/yuxi/channels/adapters/qqbot/config_hints.py b/backend/package/yuxi/channels/adapters/qqbot/config_hints.py new file mode 100644 index 00000000..ebf5b2c0 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/qqbot/config_hints.py @@ -0,0 +1,388 @@ +from __future__ import annotations + +QQBOT_CONFIG_HINTS: list[dict] = [ + { + "key": "app_id", + "label": "App ID", + "help": "QQ 开放平台应用 ID，从「应用管理」页面获取", + "required": True, + "sensitive": False, + "section": "credentials", + }, + { + "key": "app_secret", + "label": "App Secret", + "help": "QQ 开放平台客户端密钥，从「开发设置」中生成", + "required": True, + "sensitive": True, + "section": "credentials", + }, + { + "key": "client_secret_file", + "label": "密钥文件路径", + "help": "存放 App Secret 的文件路径，优先于环境变量", + "required": False, + "sensitive": False, + "section": "credentials", + }, + { + "key": "sandbox", + "label": "沙箱模式", + "help": "启用沙箱 API 端点 (sandbox.api.sgroup.qq.com)", + "required": False, + "default": False, + "type": "boolean", + "section": "connection", + }, + { + "key": "bot_name", + "label": "Bot 名称", + "help": "机器人显示名称，用于 @提及匹配", + "required": False, + "section": "basic", + }, + { + "key": "intents", + "label": "WebSocket Intents", + "help": "QQ Bot Gateway 意图位掩码，控制订阅的事件类型", + "required": False, + "type": "number", + "section": "connection", + }, + { + "key": "dm_policy", + "label": "DM 策略", + "help": "私信访问策略: open(开放) / allowlist(白名单) / pairing(配对) / disabled(禁用)", + "required": False, + "default": "open", + "type": "enum", + "options": ["open", "allowlist", "pairing", "disabled"], + "section": "security", + }, + { + "key": "group_policy", + "label": "群组策略", + "help": "群组访问策略: open(开放) / allowlist(白名单) / disabled(禁用)", + "required": False, + "default": "open", + "type": "enum", + "options": ["open", "allowlist", "disabled"], + "section": "security", + }, + { + "key": "allow_from", + "label": "DM 白名单", + "help": "允许私信的用户 ID 列表，支持通配符 '*'", + "required": False, + "type": "list", + "section": "security", + }, + { + "key": "group_allow_from", + "label": "群组白名单", + "help": "允许的群组 ID 列表，支持通配符 '*'", + "required": False, + "type": "list", + "section": "security", + }, + { + "key": "group_require_mention", + "label": "群聊 @提及门控", + "help": "群聊中是否需要 @机器人 才响应", + "required": False, + "default": True, + "type": "boolean", + "section": "security", + }, + { + "key": "pairing_enabled", + "label": "配对模式", + "help": "启用配对模式，仅已配对用户可私信", + "required": False, + "default": False, + "type": "boolean", + "section": "security", + }, + { + "key": "group_ignore_other_mentions", + "label": "忽略他人 @", + "help": "当消息 @了其他人但不包含机器人时忽略", + "required": False, + "default": True, + "type": "boolean", + "section": "security", + }, + { + "key": "group_command_bypass_mention", + "label": "命令跳过 @", + "help": "白名单命令可不需要 @机器人", + "required": False, + "default": True, + "type": "boolean", + "section": "security", + }, + { + "key": "use_markdown", + "label": "使用 Markdown", + "help": "使用 Markdown 格式发送消息", + "required": False, + "default": False, + "type": "boolean", + "section": "format", + }, + { + "key": "markdown_template_id", + "label": "Markdown 模板 ID", + "help": "QQ Bot Markdown 消息模板 ID", + "required": False, + "section": "format", + }, + { + "key": "stream_flush_interval", + "label": "流式刷新间隔", + "help": "C2C 流式消息的刷新间隔 (秒)", + "required": False, + "default": 0.6, + "type": "number", + "section": "streaming", + }, + { + "key": "streaming_mode", + "label": "流式模式", + "help": "C2C 流式模式: off(关闭) / block(块流式)", + "required": False, + "default": "block", + "type": "enum", + "options": ["off", "block"], + "section": "streaming", + }, + { + "key": "silent", + "label": "静默发送", + "help": "发送消息时不通知用户 (no_notify)", + "required": False, + "default": False, + "type": "boolean", + "section": "format", + }, + { + "key": "dedup_window_s", + "label": "去重窗口", + "help": "消息去重时间窗口 (秒)", + "required": False, + "default": 60, + "type": "number", + "section": "advanced", + }, + { + "key": "auto_download_attachments", + "label": "自动下载附件", + "help": "自动下载消息中的图片/文件附件", + "required": False, + "default": True, + "type": "boolean", + "section": "advanced", + }, + { + "key": "tts_default_voice", + "label": "TTS 默认语音", + "help": "文字转语音默认发音人", + "required": False, + "default": "zh-CN-XiaoxiaoNeural", + "section": "audio", + }, + { + "key": "tts_default_format", + "label": "TTS 默认格式", + "help": "音频输出格式", + "required": False, + "default": "mp3", + "type": "enum", + "options": ["mp3", "wav", "opus"], + "section": "audio", + }, + { + "key": "queue_global_limit", + "label": "消息队列全局上限", + "help": "消息队列最大容量", + "required": False, + "default": 1000, + "type": "number", + "section": "advanced", + }, + { + "key": "queue_per_user_limit", + "label": "每用户队列上限", + "help": "每个用户的消息队列最大容量", + "required": False, + "default": 20, + "type": "number", + "section": "advanced", + }, + { + "key": "queue_per_group_limit", + "label": "每群组队列上限", + "help": "每个群组的消息队列最大容量", + "required": False, + "default": 50, + "type": "number", + "section": "advanced", + }, + { + "key": "queue_max_concurrency", + "label": "队列最大并发", + "help": "消息队列最大并发处理数", + "required": False, + "default": 10, + "type": "number", + "section": "advanced", + }, + { + "key": "reconnect.max_retries", + "label": "重连次数", + "help": "WebSocket 最大重连次数", + "required": False, + "default": 10, + "type": "number", + "section": "connection", + }, + { + "key": "reconnect.base_delay", + "label": "重连基础延迟", + "help": "重连指数退避基础延迟 (秒)", + "required": False, + "default": 1.0, + "type": "number", + "section": "connection", + }, + { + "key": "reconnect.max_delay", + "label": "重连最大延迟", + "help": "重连指数退避最大延迟 (秒)", + "required": False, + "default": 60.0, + "type": "number", + "section": "connection", + }, + { + "key": "reconnect.jitter", + "label": "重连抖动系数", + "help": "重连延迟随机抖动系数", + "required": False, + "default": 0.3, + "type": "number", + "section": "connection", + }, + { + "key": "retry.attempts", + "label": "发送重试次数", + "help": "消息发送失败最大重试次数", + "required": False, + "default": 3, + "type": "number", + "section": "connection", + }, + { + "key": "credential_backup_dir", + "label": "凭据备份目录", + "help": "凭据备份文件存储目录", + "required": False, + "section": "storage", + }, + { + "key": "session_store_dir", + "label": "会话存储目录", + "help": "WebSocket 会话持久化目录", + "required": False, + "section": "storage", + }, + { + "key": "known_users_max", + "label": "已知用户上限", + "help": "已知用户追踪器最大记录数", + "required": False, + "default": 10000, + "type": "number", + "section": "storage", + }, + { + "key": "known_users_persist_dir", + "label": "已知用户存储目录", + "help": "已知用户持久化文件目录", + "required": False, + "section": "storage", + }, + { + "key": "suppress_local_approval_prompt", + "label": "抑制本地审批提示", + "help": "防止 QQ Bot 本地二次审批提示去重", + "required": False, + "default": True, + "type": "boolean", + "section": "approval", + }, + { + "key": "approval_native_routing", + "label": "审批原生路由", + "help": "审批请求通过 QQ Bot 内联键盘路由", + "required": False, + "default": True, + "type": "boolean", + "section": "approval", + }, + { + "key": "plugin_version", + "label": "插件版本", + "help": "注入 HTTP User-Agent 的插件版本号", + "required": False, + "default": "1.0.0", + "section": "advanced", + }, + { + "key": "openclaw_version", + "label": "OpenClaw 版本", + "help": "注入 HTTP User-Agent 的平台版本号", + "required": False, + "default": "1.0.0", + "section": "advanced", + }, + { + "key": "message_cache_max", + "label": "消息缓存上限", + "help": "发送消息缓存最大记录数", + "required": False, + "default": 500, + "type": "number", + "section": "advanced", + }, + { + "key": "message_cache_ttl_s", + "label": "消息缓存 TTL", + "help": "发送消息缓存过期时间 (秒)", + "required": False, + "default": 3600, + "type": "number", + "section": "advanced", + }, +] + + +def get_config_hints(section: str = "") -> list[dict]: + if section: + return [h for h in QQBOT_CONFIG_HINTS if h.get("section") == section] + return list(QQBOT_CONFIG_HINTS) + + +def get_config_sections() -> list[str]: + seen: set[str] = set() + result: list[str] = [] + for hint in QQBOT_CONFIG_HINTS: + section = hint.get("section", "") + if section and section not in seen: + seen.add(section) + result.append(section) + return result + + +def get_required_config_keys() -> list[str]: + return [h["key"] for h in QQBOT_CONFIG_HINTS if h.get("required")] diff --git a/backend/package/yuxi/channels/adapters/qqbot/credential_backup.py b/backend/package/yuxi/channels/adapters/qqbot/credential_backup.py index 05347509..d6ae34f1 100644 --- a/backend/package/yuxi/channels/adapters/qqbot/credential_backup.py +++ b/backend/package/yuxi/channels/adapters/qqbot/credential_backup.py @@ -14,21 +14,55 @@ logger = logging.getLogger(__name__) DEFAULT_BACKUP_DIR = os.path.join(tempfile.gettempdir(), "yuxi_qqbot_credentials") +def _resolve_backup_dir(backup_dir: str | None = None) -> str: + if backup_dir: + return backup_dir + for env_var in ("OPENCLAW_STATE_DIR", "CLAWDBOT_STATE_DIR"): + env_val = os.environ.get(env_var) + if env_val: + qqbot_dir = os.path.join(env_val, "qqbot", "credentials") + os.makedirs(qqbot_dir, exist_ok=True) + return qqbot_dir + return DEFAULT_BACKUP_DIR + + +_fernet_warned = False + + def _get_fernet(): + global _fernet_warned + try: from cryptography.fernet import Fernet except ImportError: + if not _fernet_warned: + logger.warning( + "QQBot credential encryption unavailable: cryptography not installed. " + "Credentials will be stored in plaintext. " + "Install with: pip install cryptography" + ) + _fernet_warned = True return None key = os.environ.get("QQBOT_CREDENTIAL_ENCRYPTION_KEY") if not key: + if not _fernet_warned: + logger.warning( + "QQBot credential encryption disabled: QQBOT_CREDENTIAL_ENCRYPTION_KEY not set. " + "Credentials will be stored in plaintext at %s", + _resolve_backup_dir(), + ) + _fernet_warned = True return None digest = hashlib.sha256(key.encode()).digest() return Fernet(base64.urlsafe_b64encode(digest)) def _encrypt_field(fernet, value: str) -> str: - if not fernet or not value: + if not value: + return value + if not fernet: + logger.warning("QQBot credential stored in plaintext: encryption key not configured") return value return fernet.encrypt(value.encode()).decode() @@ -65,7 +99,7 @@ class CredentialSnapshot: class CredentialBackup: def __init__(self, app_id: str, backup_dir: str | None = None): self._app_id = app_id - self._backup_dir = backup_dir or DEFAULT_BACKUP_DIR + self._backup_dir = _resolve_backup_dir(backup_dir) self._backup_path = os.path.join(self._backup_dir, f"{app_id}.json") def save(self, snapshot: CredentialSnapshot) -> bool: diff --git a/backend/package/yuxi/channels/adapters/qqbot/format.py b/backend/package/yuxi/channels/adapters/qqbot/format.py index 6fff9de7..f685a939 100644 --- a/backend/package/yuxi/channels/adapters/qqbot/format.py +++ b/backend/package/yuxi/channels/adapters/qqbot/format.py @@ -4,9 +4,15 @@ from yuxi.channels.models import ChannelResponse from .constants import DM_CHAT_PREFIX, GROUP_CHAT_PREFIX +TEXT_CONTENT_LIMIT = 2000 +MARKDOWN_CONTENT_LIMIT = 4096 +EMBED_DESCRIPTION_LIMIT = 4096 +EMBED_PROMPT_LIMIT = 200 +EMBED_TITLE_LIMIT = 200 + def build_text_payload(response: ChannelResponse) -> dict: - content = response.content[:2000] + content = response.content[:TEXT_CONTENT_LIMIT] chat_id = response.identity.channel_chat_id payload: dict = {"content": content} @@ -34,7 +40,7 @@ def build_markdown_payload(response: ChannelResponse, template_id: str | None = "template_id": md_template_id, "params": [ {"key": "title", "values": [response.metadata.get("title", "")]}, - {"key": "content", "values": [response.content[:4096]]}, + {"key": "content", "values": [response.content[:MARKDOWN_CONTENT_LIMIT]]}, ], }, } @@ -42,7 +48,7 @@ def build_markdown_payload(response: ChannelResponse, template_id: str | None = return { "msg_type": 2, "markdown": { - "content": response.content[:4096], + "content": response.content[:MARKDOWN_CONTENT_LIMIT], }, } @@ -66,9 +72,9 @@ def build_embed_payload(response: ChannelResponse) -> dict: return { "msg_type": 4, "embed": { - "title": embed_data.get("title", ""), - "description": response.content[:4096], - "prompt": embed_data.get("prompt", response.content[:200]), + "title": embed_data.get("title", "")[:EMBED_TITLE_LIMIT], + "description": response.content[:EMBED_DESCRIPTION_LIMIT], + "prompt": embed_data.get("prompt", response.content[:EMBED_PROMPT_LIMIT]), "fields": embed_data.get("fields", []), }, } @@ -82,7 +88,7 @@ def build_media_generic_payload(response: ChannelResponse, file_id: str, msg_typ } if response.content: - payload["content"] = response.content[:2000] + payload["content"] = response.content[:TEXT_CONTENT_LIMIT] if chat_id.startswith(GROUP_CHAT_PREFIX): payload["group_openid"] = chat_id.replace(GROUP_CHAT_PREFIX, "") @@ -96,10 +102,12 @@ def build_image_payload(response: ChannelResponse, file_id: str) -> dict: chat_id = response.identity.channel_chat_id payload: dict = { "msg_type": 1, - "image": file_id, - "content": response.content[:2000] if response.content else "", + "file_image": file_id, } + if response.content: + payload["content"] = response.content[:TEXT_CONTENT_LIMIT] + if chat_id.startswith(GROUP_CHAT_PREFIX): payload["group_openid"] = chat_id.replace(GROUP_CHAT_PREFIX, "") elif not chat_id.startswith(DM_CHAT_PREFIX): @@ -114,12 +122,15 @@ def format_outbound( markdown_template_id: str | None = None, ) -> dict: msg_type = response.metadata.get("qq_msg_type", "") + chat_id = response.identity.channel_chat_id if msg_type == "markdown" or (use_markdown and not msg_type): return build_markdown_payload(response, markdown_template_id) elif msg_type == "ark" and response.metadata.get("ark_template_id"): return build_ark_payload(response) elif msg_type == "embed": + if chat_id.startswith(DM_CHAT_PREFIX) or chat_id.startswith(GROUP_CHAT_PREFIX): + return build_markdown_payload(response, markdown_template_id) return build_embed_payload(response) elif msg_type == "image" and response.attachments: file_id = response.attachments[0].file_id or response.attachments[0].url or "" @@ -137,7 +148,7 @@ def format_outbound( class MarkdownChunker: - MAX_CHARS = 5000 + MAX_CHARS = 4096 CHUNK_OVERLAP = 200 def __init__(self, max_chars: int = MAX_CHARS, chunk_overlap: int = CHUNK_OVERLAP): diff --git a/backend/package/yuxi/channels/adapters/qqbot/group_buffer.py b/backend/package/yuxi/channels/adapters/qqbot/group_buffer.py index 936e62d2..5c744e57 100644 --- a/backend/package/yuxi/channels/adapters/qqbot/group_buffer.py +++ b/backend/package/yuxi/channels/adapters/qqbot/group_buffer.py @@ -17,7 +17,7 @@ class GroupMessage: @dataclass class GroupSession: - group_id: str + group_id: str = "" messages: list[GroupMessage] = field(default_factory=list) last_active: float = 0.0 buffer_limit: int = 50 @@ -43,6 +43,16 @@ class GroupHistoryBuffer: self._sessions: dict[str, GroupSession] = defaultdict(GroupSession) self._buffer_limit = buffer_limit self._ttl_seconds = ttl_seconds + self._group_activations: dict[str, str] = {} + + def set_group_activation(self, group_id: str, activation: str) -> None: + self._group_activations[group_id] = activation + + def resolve_group_activation(self, group_id: str) -> str: + return self._group_activations.get(group_id, "") + + def get_all_activations(self) -> dict[str, str]: + return dict(self._group_activations) def record(self, group_id: str, msg: GroupMessage) -> None: session = self._sessions[group_id] diff --git a/backend/package/yuxi/channels/adapters/qqbot/inbound_pipeline.py b/backend/package/yuxi/channels/adapters/qqbot/inbound_pipeline.py index 39ae01ca..26204c23 100644 --- a/backend/package/yuxi/channels/adapters/qqbot/inbound_pipeline.py +++ b/backend/package/yuxi/channels/adapters/qqbot/inbound_pipeline.py @@ -159,11 +159,10 @@ class QQBotInboundPipeline(BaseInboundPipeline): adapter = pipeline.adapter dedup_cache = getattr(adapter, "_recent_msg_ids", None) - dedup_window = getattr(adapter, "_dedup_window_s", 60) if dedup_cache is not None: now = time.monotonic() - self._prune_dedup_cache(dedup_cache, dedup_window, now) + adapter._prune_old_msg_ids(now) if msg_id in dedup_cache: logger.debug("Dedup: %s already processed, skipping", msg_id) ctx.stop("dedup_duplicate") @@ -172,12 +171,6 @@ class QQBotInboundPipeline(BaseInboundPipeline): return ctx - @staticmethod - def _prune_dedup_cache(cache: dict[str, float], window: int, now: float) -> None: - expired = [mid for mid, ts in cache.items() if now - ts >= window] - for mid in expired: - del cache[mid] - @staticmethod def _buffer_group_message(pipeline: BaseInboundPipeline, event_data: dict, ctx: PipelineContext) -> None: adapter = pipeline.adapter @@ -242,7 +235,6 @@ class QQBotInboundPipeline(BaseInboundPipeline): ctx.metadata["group_openid"] = event_data.get("group_openid", "") ctx.metadata["bot_mentioned"] = False ctx.stop("group_message_no_mention") - self._buffer_group_message(pipeline, event_data, ctx) return None elif event_type == "INTERACTION_CREATE": diff --git a/backend/package/yuxi/channels/adapters/qqbot/interaction.py b/backend/package/yuxi/channels/adapters/qqbot/interaction.py index 6aadf378..eb8c787c 100644 --- a/backend/package/yuxi/channels/adapters/qqbot/interaction.py +++ b/backend/package/yuxi/channels/adapters/qqbot/interaction.py @@ -42,12 +42,12 @@ class InteractionRegistry: if not ctx.interaction_id: return False - for action_id, callback in self._callbacks.items(): - if ctx.interaction_id == action_id or ctx.interaction_id.startswith(action_id): - try: - return await callback(ctx) - except Exception: - logger.exception("InteractionRegistry: callback failed for %s", action_id) + callback = self._callbacks.get(ctx.interaction_id) + if callback: + try: + return await callback(ctx) + except Exception: + logger.exception("InteractionRegistry: callback failed for %s", ctx.interaction_id) return False diff --git a/backend/package/yuxi/channels/adapters/qqbot/known_users.py b/backend/package/yuxi/channels/adapters/qqbot/known_users.py index 22a149c7..d25308dc 100644 --- a/backend/package/yuxi/channels/adapters/qqbot/known_users.py +++ b/backend/package/yuxi/channels/adapters/qqbot/known_users.py @@ -14,6 +14,18 @@ logger = logging.getLogger(__name__) DEFAULT_PERSIST_DIR = os.path.join(tempfile.gettempdir(), "yuxi_qqbot_known_users") +def _resolve_known_users_dir(persist_dir: str | None = None) -> str: + if persist_dir: + return persist_dir + for env_var in ("OPENCLAW_STATE_DIR", "CLAWDBOT_STATE_DIR"): + env_val = os.environ.get(env_var) + if env_val: + qqbot_dir = os.path.join(env_val, "qqbot", "known_users") + os.makedirs(qqbot_dir, exist_ok=True) + return qqbot_dir + return DEFAULT_PERSIST_DIR + + @dataclass class UserRecord: user_id: str @@ -66,10 +78,10 @@ class KnownUserTracker: ): self._app_id = app_id self._max_users = max_users - self._persist_dir = persist_dir or DEFAULT_PERSIST_DIR + self._persist_dir = _resolve_known_users_dir(persist_dir) self._persist_path = os.path.join(self._persist_dir, f"{app_id}_users.json") self._persist_interval = persist_interval_s - self._lock = threading.Lock() + self._lock = threading.RLock() self._users: OrderedDict[str, UserRecord] = OrderedDict() self._last_persist: float = 0 self._dirty = False @@ -178,3 +190,56 @@ class KnownUserTracker: def get_recent_users(self, limit: int = 50) -> list[UserRecord]: with self._lock: return list(reversed(self._users.values()))[:limit] + + def list_known_users(self) -> list[dict]: + with self._lock: + return [u.to_dict() for u in self._users.values()] + + def get_known_users_stats(self) -> dict: + now = time.time() + active_24h = 0 + active_7d = 0 + dm_count = 0 + group_count = 0 + with self._lock: + for u in self._users.values(): + if now - u.last_seen <= 86400: + active_24h += 1 + if now - u.last_seen <= 604800: + active_7d += 1 + if "direct" in u.chat_types: + dm_count += 1 + if "group" in u.chat_types: + group_count += 1 + return { + "total": len(self._users), + "active_24h": active_24h, + "active_7d": active_7d, + "dm_users": dm_count, + "group_users": group_count, + } + + def get_user_groups(self, user_id: str) -> list[str]: + record = self.get(user_id) + if record is None: + return [] + return sorted(record.chat_types) + + def get_group_members(self) -> dict[str, list[str]]: + result: dict[str, list[str]] = {} + with self._lock: + for u in self._users.values(): + for ct in u.chat_types: + if ct not in result: + result[ct] = [] + result[ct].append(u.user_id) + return result + + def flush_known_users(self) -> int: + count = 0 + with self._lock: + count = len(self._users) + self._users.clear() + self._dirty = True + self.persist() + return count diff --git a/backend/package/yuxi/channels/adapters/qqbot/media_upload.py b/backend/package/yuxi/channels/adapters/qqbot/media_upload.py index 3bc9952d..b8b9375b 100644 --- a/backend/package/yuxi/channels/adapters/qqbot/media_upload.py +++ b/backend/package/yuxi/channels/adapters/qqbot/media_upload.py @@ -1,6 +1,9 @@ from __future__ import annotations +import asyncio + import aiohttp +from cachetools import LRUCache from yuxi.channels.exceptions import DeliveryFailedError from yuxi.utils.logging_config import logger @@ -47,13 +50,25 @@ async def upload_media( url = f"{api_base}/v2/groups/{group_openid}/files" if group_openid else f"{api_base}/v2/users/@me/files" async def _do_upload(client: aiohttp.ClientSession) -> str: - async with client.post(url, headers=headers, data=form) as resp: - if resp.status != 200: - raise DeliveryFailedError(f"Media upload failed: HTTP {resp.status}") - result = await resp.json() - file_id = result.get("file_uuid", "") or result.get("file_info", "") - logger.debug(f"[QQBot] Media uploaded, file_type={file_type}, file_id={file_id}") - return file_id + last_error = None + for attempt in range(3): + try: + async with client.post(url, headers=headers, data=form) as resp: + if resp.status != 200: + raise DeliveryFailedError(f"Media upload failed: HTTP {resp.status}") + result = await resp.json() + file_id = result.get("file_uuid", "") or result.get("file_info", "") + logger.debug(f"[QQBot] Media uploaded, file_type={file_type}, file_id={file_id}") + return file_id + except DeliveryFailedError: + raise + except Exception as e: + last_error = e + if attempt < 2: + delay = 2**attempt + logger.warning("[QQBot] Media upload retry %d/%d after %.1fs: %s", attempt + 1, 3, delay, e) + await asyncio.sleep(delay) + raise DeliveryFailedError(f"Media upload failed after 3 attempts: {last_error}") if http_client: return await _do_upload(http_client) @@ -119,7 +134,7 @@ def build_media_payload( "msg_type": msg_type, } if msg_type == 1: - payload["image"] = file_id + payload["file_image"] = file_id elif msg_type == 4: payload["file"] = file_id else: @@ -146,7 +161,7 @@ async def download_image( _CHUNK_SIZE = 5 * 1024 * 1024 -_UPLOAD_CACHE: dict[str, str] = {} +_UPLOAD_CACHE = LRUCache(maxsize=1000) def _make_cache_key(data: bytes) -> str: diff --git a/backend/package/yuxi/channels/adapters/qqbot/message_cache.py b/backend/package/yuxi/channels/adapters/qqbot/message_cache.py new file mode 100644 index 00000000..c5e6ff0b --- /dev/null +++ b/backend/package/yuxi/channels/adapters/qqbot/message_cache.py @@ -0,0 +1,94 @@ +from __future__ import annotations + +import logging +import time +from collections import OrderedDict +from dataclasses import dataclass, field + +from yuxi.channels.models import DeliveryResult + +logger = logging.getLogger(__name__) + + +@dataclass +class SentMessage: + message_id: str + chat_id: str + content: str = "" + sent_at: float = field(default_factory=time.time) + updated_at: float = 0.0 + update_count: int = 0 + status: str = "sent" + error: str = "" + + +class MessageCache: + def __init__(self, max_messages: int = 500, ttl_s: float = 3600.0): + self._max = max_messages + self._ttl = ttl_s + self._messages: OrderedDict[str, SentMessage] = OrderedDict() + + def record_sent(self, result: DeliveryResult, chat_id: str, content: str = "") -> SentMessage | None: + if not result.message_id: + return None + + msg = SentMessage( + message_id=result.message_id, + chat_id=chat_id, + content=content[:500], + status="sent" if result.success else "failed", + error=result.error or "", + ) + self._messages[result.message_id] = msg + self._messages.move_to_end(result.message_id) + + while len(self._messages) > self._max: + self._messages.popitem(last=False) + + return msg + + def record_update(self, message_id: str, content: str = "") -> SentMessage | None: + msg = self._messages.get(message_id) + if msg is None: + return None + msg.updated_at = time.time() + msg.update_count += 1 + if content: + msg.content = content[:500] + return msg + + def get(self, message_id: str) -> SentMessage | None: + return self._messages.get(message_id) + + def get_by_chat(self, chat_id: str) -> list[SentMessage]: + return [m for m in self._messages.values() if m.chat_id == chat_id] + + def cleanup_expired(self) -> int: + now = time.time() + expired = [mid for mid, msg in self._messages.items() if now - msg.sent_at > self._ttl] + for mid in expired: + del self._messages[mid] + return len(expired) + + @property + def snapshot(self) -> dict: + return { + "total": len(self._messages), + "max": self._max, + "ttl_s": self._ttl, + "recent": [ + { + "message_id": m.message_id, + "chat_id": m.chat_id, + "sent_at": m.sent_at, + "updates": m.update_count, + "status": m.status, + } + for m in list(self._messages.values())[-20:] + ], + } + + def clear(self) -> int: + count = len(self._messages) + self._messages.clear() + return count diff --git a/backend/package/yuxi/channels/adapters/qqbot/message_queue.py b/backend/package/yuxi/channels/adapters/qqbot/message_queue.py index 09b72b0a..5b833420 100644 --- a/backend/package/yuxi/channels/adapters/qqbot/message_queue.py +++ b/backend/package/yuxi/channels/adapters/qqbot/message_queue.py @@ -114,4 +114,4 @@ class MessageQueue: break self._per_user_counts.clear() self._per_group_counts.clear() - return messages \ No newline at end of file + return messages diff --git a/backend/package/yuxi/channels/adapters/qqbot/rate_limiter.py b/backend/package/yuxi/channels/adapters/qqbot/rate_limiter.py index f9c7f55e..590c8542 100644 --- a/backend/package/yuxi/channels/adapters/qqbot/rate_limiter.py +++ b/backend/package/yuxi/channels/adapters/qqbot/rate_limiter.py @@ -77,7 +77,7 @@ class RouteRateLimiter: def _get_or_create_bucket_sync(self, route: str) -> TokenBucket: if route not in self._buckets: rate, burst = self.defaults.get(route, self.defaults["default"]) - self._buckets[route] = TokenBucket(rate=rate, burst=burst) + self._buckets.setdefault(route, TokenBucket(rate=rate, burst=burst)) return self._buckets[route] def get_stats(self) -> dict[str, dict]: diff --git a/backend/package/yuxi/channels/adapters/qqbot/reconnect.py b/backend/package/yuxi/channels/adapters/qqbot/reconnect.py index 4756ef55..fcc5c245 100644 --- a/backend/package/yuxi/channels/adapters/qqbot/reconnect.py +++ b/backend/package/yuxi/channels/adapters/qqbot/reconnect.py @@ -280,8 +280,8 @@ class QQBotReconnectManager: return base * 1.5 return base * 2.0 - def _check_rapid_disconnect(self, code: int | None, category: CloseCodeCategory, now: float) -> None: - if category == CloseCodeCategory.FATAL: + def _check_rapid_disconnect(self, code: int | None, category: CloseCodeCategory | None, now: float) -> None: + if category is None or category == CloseCodeCategory.FATAL: return if self._last_connect_time == 0: return diff --git a/backend/package/yuxi/channels/adapters/qqbot/send.py b/backend/package/yuxi/channels/adapters/qqbot/send.py index ce9d6ed9..00e5b06c 100644 --- a/backend/package/yuxi/channels/adapters/qqbot/send.py +++ b/backend/package/yuxi/channels/adapters/qqbot/send.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import re from collections.abc import Awaitable, Callable from dataclasses import dataclass, field @@ -64,6 +65,7 @@ async def send_with_retry( config: dict | None = None, token_refresh_cb: Callable[[], Awaitable[str]] | None = None, on_sent: Callable[[DeliveryResult], Awaitable[None]] | None = None, + sender_headers: dict | None = None, ) -> DeliveryResult: cfg = config or {} max_retries = cfg.get("retry", {}).get("attempts", 3) @@ -77,7 +79,7 @@ async def send_with_retry( max_rate_limit_retries = cfg.get("retry", {}).get("max_rate_limit_retries", 3) last_error = None - url = _resolve_send_url(api_base, chat_id) + url = resolve_send_url(api_base, chat_id) for attempt in range(max_retries): try: @@ -85,6 +87,8 @@ async def send_with_retry( "Authorization": f"QQBot {current_token}", "Content-Type": "application/json", } + if sender_headers: + headers.update(sender_headers) async with http_client.post(url, json=payload, headers=headers) as resp: if resp.status == 200: data = await resp.json() @@ -156,7 +160,33 @@ async def send_with_retry( return result -def _resolve_send_url(api_base: str, chat_id: str) -> str: +_ALLOWED_API_HOSTS = frozenset({"api.sgroup.qq.com", "sandbox.api.sgroup.qq.com"}) +_CHAT_ID_PATTERN = re.compile(r"^[a-zA-Z0-9_-]+$") + + +def _validate_api_base(api_base: str) -> None: + from urllib.parse import urlparse + + parsed = urlparse(api_base) + host = parsed.hostname + if host not in _ALLOWED_API_HOSTS: + raise DeliveryFailedError(f"Invalid api_base host: {host}. Allowed: {sorted(_ALLOWED_API_HOSTS)}") + + +def _validate_chat_id(chat_id: str) -> None: + raw = chat_id + if raw.startswith(GROUP_CHAT_PREFIX): + raw = raw[len(GROUP_CHAT_PREFIX) :] + elif raw.startswith(DM_CHAT_PREFIX): + raw = raw[len(DM_CHAT_PREFIX) :] + if raw and not _CHAT_ID_PATTERN.match(raw): + raise DeliveryFailedError(f"Invalid chat_id format: {chat_id}") + + +def resolve_send_url(api_base: str, chat_id: str) -> str: + _validate_api_base(api_base) + _validate_chat_id(chat_id) + if chat_id.startswith(GROUP_CHAT_PREFIX): group_openid = chat_id.replace(GROUP_CHAT_PREFIX, "") return f"{api_base}/v2/groups/{group_openid}/messages" diff --git a/backend/package/yuxi/channels/adapters/qqbot/sender_init.py b/backend/package/yuxi/channels/adapters/qqbot/sender_init.py new file mode 100644 index 00000000..57438489 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/qqbot/sender_init.py @@ -0,0 +1,48 @@ +from __future__ import annotations + +import logging +from dataclasses import dataclass + +logger = logging.getLogger(__name__) + +PLUGIN_VERSION = "1.0.0" +OPENCLAW_VERSION = "1.0.0" + + +@dataclass +class SenderInfo: + plugin_version: str = PLUGIN_VERSION + openclaw_version: str = OPENCLAW_VERSION + channel_id: str = "qqbot" + user_agent: str = "" + + def __post_init__(self): + self.user_agent = f"openclaw-qqbot/{self.plugin_version} openclaw/{self.openclaw_version}" + + +def init_sender( + plugin_version: str = PLUGIN_VERSION, + openclaw_version: str = OPENCLAW_VERSION, +) -> SenderInfo: + info = SenderInfo( + plugin_version=plugin_version, + openclaw_version=openclaw_version, + ) + logger.info( + "Sender initialized: plugin=%s openclaw=%s ua=%s", + info.plugin_version, + info.openclaw_version, + info.user_agent, + ) + return info + + +def build_sender_headers(info: SenderInfo, extra: dict | None = None) -> dict: + headers = { + "User-Agent": info.user_agent, + "X-Plugin-Version": info.plugin_version, + "X-OpenClaw-Version": info.openclaw_version, + } + if extra: + headers.update(extra) + return headers diff --git a/backend/package/yuxi/channels/adapters/qqbot/session_store.py b/backend/package/yuxi/channels/adapters/qqbot/session_store.py index e2c21503..2fb5226b 100644 --- a/backend/package/yuxi/channels/adapters/qqbot/session_store.py +++ b/backend/package/yuxi/channels/adapters/qqbot/session_store.py @@ -12,6 +12,27 @@ logger = logging.getLogger(__name__) DEFAULT_SESSION_DIR = os.path.join(tempfile.gettempdir(), "yuxi_qqbot_sessions") +def _resolve_state_dir(store_dir: str | None = None) -> str: + if store_dir: + return store_dir + for env_var in ("OPENCLAW_STATE_DIR", "CLAWDBOT_STATE_DIR"): + env_val = os.environ.get(env_var) + if env_val: + qqbot_dir = os.path.join(env_val, "qqbot") + os.makedirs(qqbot_dir, exist_ok=True) + return qqbot_dir + return DEFAULT_SESSION_DIR + + +@dataclass +class SessionState: + group_openid: str = "" + group_activation: str = "" + require_mention: bool = True + agent_id: str = "" + last_active: float = 0.0 + + @dataclass class SessionRecord: session_id: str = "" @@ -50,7 +71,7 @@ class SessionRecord: class SessionStore: def __init__(self, app_id: str, store_dir: str | None = None): self._app_id = app_id - self._store_dir = store_dir or DEFAULT_SESSION_DIR + self._store_dir = _resolve_state_dir(store_dir) self._store_path = os.path.join(self._store_dir, f"{app_id}_session.json") def save(self, record: SessionRecord) -> bool: @@ -96,9 +117,39 @@ class SessionStore: logger.exception("SessionStore: failed to clear session") return False + def update_last_seq(self, session_id: str, seq: int) -> bool: + record = self.load() + if record is None: + record = SessionRecord() + record.session_id = session_id + record.last_seq = seq + record.last_heartbeat = time.monotonic() + return self.save(record) + + @staticmethod + def get_all_sessions(store_dir: str | None = None) -> list[dict]: + directory = _resolve_state_dir(store_dir) + if not os.path.exists(directory): + return [] + sessions: list[dict] = [] + try: + for filename in os.listdir(directory): + if not filename.endswith("_session.json"): + continue + filepath = os.path.join(directory, filename) + try: + with open(filepath, encoding="utf-8") as f: + data = json.load(f) + sessions.append(data) + except (OSError, json.JSONDecodeError): + pass + except OSError: + logger.exception("SessionStore: failed to list sessions") + return sessions + @staticmethod def cleanup_expired(store_dir: str | None = None, max_age_s: float = 86400 * 7) -> int: - directory = store_dir or DEFAULT_SESSION_DIR + directory = _resolve_state_dir(store_dir) if not os.path.exists(directory): return 0 diff --git a/backend/package/yuxi/channels/adapters/qqbot/setup_wizard.py b/backend/package/yuxi/channels/adapters/qqbot/setup_wizard.py index 658cc13a..e3e36166 100644 --- a/backend/package/yuxi/channels/adapters/qqbot/setup_wizard.py +++ b/backend/package/yuxi/channels/adapters/qqbot/setup_wizard.py @@ -13,7 +13,9 @@ logger = logging.getLogger(__name__) class WizardStep(Enum): WELCOME = auto() + SETUP_METHOD = auto() APP_CREDENTIALS = auto() + QR_LINK = auto() INTENTS = auto() PERMISSIONS = auto() WEBHOOK_URL = auto() @@ -25,11 +27,14 @@ class WizardStep(Enum): @dataclass class WizardState: step: WizardStep = WizardStep.WELCOME + setup_method: str = "" app_id: str = "" app_secret: str = "" bot_token: str = "" intents: list[str] = None webhook_url: str = "" + qr_url: str = "" + qr_session_id: str = "" verify_result: dict | None = None started_at: float = 0.0 @@ -209,3 +214,49 @@ async def generate_config_yaml(app_id: str, app_secret: str, intents: list[int] lines.append(f"# {mark} {intent_id}: {name} ({desc})") return "\n".join(lines) + + +async def link_via_qr_code(app_id: str) -> dict: + import uuid + + session_id = uuid.uuid4().hex[:16] + api_base = "https://api.sgroup.qq.com" + + qr_url = f"https://q.qq.com/qqbot/link?app_id={app_id}&session_id={session_id}&t={int(time.time())}" + + return { + "success": True, + "session_id": session_id, + "qr_url": qr_url, + "message": "请使用 QQ 扫描二维码完成 Bot 绑定", + "api_base": api_base, + "poll_url": f"{api_base}/oauth2/qr_poll?session_id={session_id}", + } + + +async def check_qr_link_status(session_id: str, app_id: str) -> dict: + import aiohttp + + api_base = "https://api.sgroup.qq.com" + try: + async with aiohttp.ClientSession() as session: + async with session.get( + f"{api_base}/oauth2/qr_poll", + params={"session_id": session_id, "app_id": app_id}, + ) as resp: + if resp.status == 200: + data = await resp.json() + status = data.get("status", "pending") + if status == "confirmed": + return { + "success": True, + "status": "confirmed", + "app_id": app_id, + "data": data, + } + elif status == "expired": + return {"success": False, "status": "expired", "error": "QR code expired"} + return {"success": True, "status": "pending"} + return {"success": False, "error": f"HTTP {resp.status}"} + except Exception as e: + return {"success": False, "error": str(e)} diff --git a/backend/package/yuxi/channels/adapters/qqbot/streaming.py b/backend/package/yuxi/channels/adapters/qqbot/streaming.py index a5fb3feb..0661536c 100644 --- a/backend/package/yuxi/channels/adapters/qqbot/streaming.py +++ b/backend/package/yuxi/channels/adapters/qqbot/streaming.py @@ -65,9 +65,7 @@ async def stream_content( yield para if c2c_ctrl is not None: - c2c_batches = c2c_ctrl.flush_controller.feed(chunk) - for batch in c2c_batches: - pass + c2c_ctrl.flush_controller.feed(chunk) await asyncio.sleep(0) @@ -201,9 +199,7 @@ async def stream_with_media_handling( yield para if c2c_ctrl is not None: - c2c_batches = c2c_ctrl.flush_controller.feed(chunk) - for _batch in c2c_batches: - pass + c2c_ctrl.flush_controller.feed(chunk) await asyncio.sleep(0) diff --git a/backend/package/yuxi/channels/adapters/qqbot/thread_simulator.py b/backend/package/yuxi/channels/adapters/qqbot/thread_simulator.py new file mode 100644 index 00000000..40e526f4 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/qqbot/thread_simulator.py @@ -0,0 +1,64 @@ +from __future__ import annotations + +from collections import OrderedDict +from datetime import datetime +from typing import Any + +from yuxi.channels.models import HistoricalMessage + + +class QQBotThreadSimulator: + def __init__(self, window_size: int = 20): + self._reply_chains: dict[str, list[str]] = {} + self._context_windows: OrderedDict[str, list[dict[str, Any]]] = OrderedDict() + self._window_size = window_size + + def build_thread_context(self, message: dict, session_key: str) -> str: + window = self._context_windows.get(session_key) + if window is None: + window = [] + self._context_windows[session_key] = window + + window.append( + { + "id": message.get("id", ""), + "author": message.get("author", {}).get("username", "unknown"), + "content": message.get("content", ""), + "timestamp": message.get("timestamp", datetime.now()), + } + ) + + if len(window) > self._window_size: + window.pop(0) + + if len(self._context_windows) > 500: + oldest = next(iter(self._context_windows)) + del self._context_windows[oldest] + + context_lines = [f"{msg['author']}: {msg['content']}" for msg in window[:-1]] + return "\n".join(context_lines) + + def resolve_simulated_thread_id(self, message: dict) -> str: + msg_type = message.get("message_type", "c2c") + + if msg_type == "c2c": + openid = message.get("author", {}).get("id", "unknown") + seq = message.get("seq", 0) + return f"qqbot:c2c:{openid}:seq:{seq // 10}" + + group_id = message.get("group_id", "unknown") + seq = message.get("seq", 0) + return f"qqbot:group:{group_id}:seq:{seq // 10}" + + def get_context_as_history(self, session_key: str) -> list[HistoricalMessage]: + window = self._context_windows.get(session_key, []) + return [ + HistoricalMessage( + message_id=msg.get("id", ""), + sender_id=msg.get("author", ""), + sender_name=msg.get("author", "unknown"), + content=msg.get("content", ""), + timestamp=msg.get("timestamp", datetime.now()), + ) + for msg in window + ] diff --git a/backend/package/yuxi/channels/adapters/qqbot/token.py b/backend/package/yuxi/channels/adapters/qqbot/token.py index b72269d8..0d35c0de 100644 --- a/backend/package/yuxi/channels/adapters/qqbot/token.py +++ b/backend/package/yuxi/channels/adapters/qqbot/token.py @@ -83,7 +83,7 @@ class QQBotTokenManager: data = await resp.json() self._access_token = data["access_token"] expires_in = data.get("expires_in", 7200) - self._expires_at = time.monotonic() + expires_in + self._expires_at = time.time() + expires_in logger.info(f"[QQBot] Token refreshed, expires in {expires_in}s (app_id={self.app_id[:6]}...)") finally: if not self._http_client: @@ -106,7 +106,7 @@ class QQBotTokenManager: try: sleep_duration = self._refresh_interval if self._expires_at is not None: - remaining = self._expires_at - time.monotonic() - 300 + remaining = self._expires_at - time.time() - 300 sleep_duration = max(60.0, min(self._refresh_interval, max(remaining, 0.0))) await asyncio.sleep(sleep_duration) @@ -127,4 +127,4 @@ class QQBotTokenManager: def _is_expired(self) -> bool: if self._access_token is None or self._expires_at is None: return True - return time.monotonic() > self._expires_at - 300 + return time.time() > self._expires_at - 300 diff --git a/backend/package/yuxi/channels/adapters/qqbot/typing_keepalive.py b/backend/package/yuxi/channels/adapters/qqbot/typing_keepalive.py index 57695608..a705dd2d 100644 --- a/backend/package/yuxi/channels/adapters/qqbot/typing_keepalive.py +++ b/backend/package/yuxi/channels/adapters/qqbot/typing_keepalive.py @@ -52,4 +52,4 @@ class TypingKeepAlive: except Exception: logger.debug("TypingKeepAlive: notify failed for %s", chat_id, exc_info=True) except asyncio.CancelledError: - pass \ No newline at end of file + pass diff --git a/backend/package/yuxi/channels/adapters/qqbot/voice_send.py b/backend/package/yuxi/channels/adapters/qqbot/voice_send.py index e5884abc..3ccac207 100644 --- a/backend/package/yuxi/channels/adapters/qqbot/voice_send.py +++ b/backend/package/yuxi/channels/adapters/qqbot/voice_send.py @@ -5,13 +5,14 @@ import aiohttp from yuxi.channels.models import DeliveryResult from yuxi.utils.logging_config import logger -from .constants import DM_CHAT_PREFIX, GROUP_CHAT_PREFIX +from .constants import GROUP_CHAT_PREFIX from .media_upload import ( FILE_TYPE_VOICE, build_media_payload, upload_media, validate_media_size, ) +from .send import resolve_send_url async def send_voice( @@ -43,7 +44,7 @@ async def send_voice( payload = build_media_payload(chat_id, file_id, msg_type=7) - url = _resolve_media_send_url(api_base, chat_id) + url = resolve_send_url(api_base, chat_id) headers = { "Authorization": f"QQBot {token}", "Content-Type": "application/json", @@ -94,7 +95,7 @@ async def send_video( payload = build_media_payload(chat_id, file_id, msg_type=7) - url = _resolve_media_send_url(api_base, chat_id) + url = resolve_send_url(api_base, chat_id) headers = { "Authorization": f"QQBot {token}", "Content-Type": "application/json", @@ -145,7 +146,7 @@ async def send_file( payload = build_media_payload(chat_id, file_id, msg_type=7) - url = _resolve_media_send_url(api_base, chat_id) + url = resolve_send_url(api_base, chat_id) headers = { "Authorization": f"QQBot {token}", "Content-Type": "application/json", @@ -163,15 +164,3 @@ async def send_file( except Exception as e: logger.error(f"[QQBot] File send error: {e}") return DeliveryResult(success=False, error=str(e)) - - -def _resolve_media_send_url(api_base: str, chat_id: str) -> str: - if chat_id.startswith(GROUP_CHAT_PREFIX): - group_openid = chat_id.replace(GROUP_CHAT_PREFIX, "") - return f"{api_base}/v2/groups/{group_openid}/messages" - elif chat_id.startswith(DM_CHAT_PREFIX): - openid = chat_id.replace(DM_CHAT_PREFIX, "") - return f"{api_base}/v2/users/{openid}/messages" - elif chat_id: - return f"{api_base}/v2/channels/{chat_id}/messages" - return f"{api_base}/v2/users/@me/messages"