From 87fc26cd15064005e87c091c06964122036a25c4 Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Wed, 13 May 2026 16:09:31 +0800 Subject: [PATCH] =?UTF-8?q?refactor(googlechat):=20=E6=95=B4=E7=90=86?= =?UTF-8?q?=E5=B9=B6=E4=BC=98=E5=8C=96Google=20Chat=E9=80=82=E9=85=8D?= =?UTF-8?q?=E5=99=A8=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 调整依赖导入顺序与清理冗余空行 - 修复setup wizard步骤存储逻辑 - 新增消息编辑/转发标记、卡片控件支持 - 完善目标校验、媒体上传、配对存储功能 - 优化流式响应与错误处理逻辑 - 增强SSRF防护与配置灵活性 - 新增/扩展内置命令与卡片处理能力 - 调整媒体大小限制与类型参数 --- .../channels/adapters/googlechat/adapter.py | 219 +++++++++++++----- .../channels/adapters/googlechat/cards.py | 47 ++++ .../channels/adapters/googlechat/directory.py | 26 ++- .../channels/adapters/googlechat/formatter.py | 5 +- .../channels/adapters/googlechat/media.py | 54 ----- .../adapters/googlechat/normalizer.py | 2 + .../channels/adapters/googlechat/pairing.py | 71 ++++++ .../channels/adapters/googlechat/policy.py | 2 +- .../channels/adapters/googlechat/pubsub_.py | 2 +- .../yuxi/channels/adapters/googlechat/send.py | 55 ++++- .../channels/adapters/googlechat/setup.py | 5 +- .../adapters/googlechat/slash_commands.py | 1 - .../yuxi/channels/adapters/googlechat/ssrf.py | 95 ++++---- .../channels/adapters/googlechat/streaming.py | 7 +- .../channels/adapters/googlechat/target.py | 17 +- 15 files changed, 438 insertions(+), 170 deletions(-) diff --git a/backend/package/yuxi/channels/adapters/googlechat/adapter.py b/backend/package/yuxi/channels/adapters/googlechat/adapter.py index d56bb413..1ae3ada6 100644 --- a/backend/package/yuxi/channels/adapters/googlechat/adapter.py +++ b/backend/package/yuxi/channels/adapters/googlechat/adapter.py @@ -5,7 +5,7 @@ import json import os import time from collections import defaultdict -from collections.abc import AsyncIterator +from collections.abc import AsyncIterator, Callable from datetime import UTC from typing import Any, ClassVar @@ -25,11 +25,13 @@ from yuxi.channels.models import ( ChannelStatus, ChannelType, DeliveryResult, + EventType, HealthStatus, ) from yuxi.channels.registry import register_builtin_adapter from yuxi.utils.logging_config import logger +from . import auth, proxy from .formatter import format_outbound from .media import download_media as _download_media from .normalizer import is_bot_message, normalize_inbound @@ -46,8 +48,6 @@ from .send import ( upload_image_message, ) from .streaming import StreamManager -from . import auth -from . import proxy _CONNECT_TIMEOUT_S = 30.0 _PREAUTH_BODY_MAX_BYTES = 16 * 1024 @@ -106,7 +106,7 @@ class GoogleChatAdapter(BaseChannelAdapter): supports_markdown: ClassVar[bool] = True supports_streaming: ClassVar[bool] = True streaming_modes: ClassVar[list[str]] = ["off", "partial", "block"] - max_media_size_mb: ClassVar[int] = 20 + max_media_size_mb: ClassVar[int] = 200 min_send_interval_ms: ClassVar[int] = 1000 capabilities = ChannelCapabilities( @@ -121,7 +121,7 @@ class GoogleChatAdapter(BaseChannelAdapter): supports_streaming=True, streaming_modes=["off", "partial", "block"], text_chunk_limit=4000, - max_media_size_mb=20, + max_media_size_mb=200, ) meta = ChannelMeta( id="googlechat", @@ -144,10 +144,30 @@ class GoogleChatAdapter(BaseChannelAdapter): self._service_account_email: str = "" self._pubsub_subscription: str = "" self._connected_at: float | None = None + + stream_cfg: dict = {} + if config: + stream_cfg = config.get("streaming", {}) if isinstance(config.get("streaming"), dict) else {} + block_cfg = stream_cfg.get("block", {}) if isinstance(stream_cfg, dict) else {} + if not isinstance(block_cfg, dict): + block_cfg = {} + + update_interval = (config.get("streamUpdateIntervalMs") if config else None) or block_cfg.get( + "update_interval_ms", 800 + ) + + coalesce_min = (config.get("blockStreamingCoalesceMinChars") if config else None) or block_cfg.get( + "coalesce_min_chars", 1500 + ) + + coalesce_idle = (config.get("blockStreamingCoalesceIdleMs") if config else None) or block_cfg.get( + "coalesce_idle_ms", 1000 + ) + self._stream_mgr = StreamManager( - update_interval_ms=config.get("streamUpdateIntervalMs", 800) if config else 800, - coalesce_min_chars=config.get("blockStreamingCoalesceMinChars", 1500) if config else 1500, - coalesce_idle_ms=config.get("blockStreamingCoalesceIdleMs", 1000) if config else 1000, + update_interval_ms=update_interval, + coalesce_min_chars=coalesce_min, + coalesce_idle_ms=coalesce_idle, ) self._rate_limiter = ChatRateLimiter(ops_per_second=(config or {}).get("minSendIntervalMs", 1000) / 1000.0) self._circuit_breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60.0) @@ -175,6 +195,7 @@ class GoogleChatAdapter(BaseChannelAdapter): self._session_contexts: dict[str, tuple[float, dict]] = {} self._session_ttl_s = config.get("sessionTtlS", 1800) if config else 1800 self._session_cleanup_task: asyncio.Task | None = None + self._card_handlers: dict[str, Callable] = {} cfg = config or {} self._connect_timeout_s = cfg.get("connectTimeout", _CONNECT_TIMEOUT_S) self._preauth_body_max_bytes = cfg.get("preauthBodyMaxBytes", _PREAUTH_BODY_MAX_BYTES) @@ -208,6 +229,7 @@ class GoogleChatAdapter(BaseChannelAdapter): status = 0 try: import googleapiclient.errors + if isinstance(e, googleapiclient.errors.HttpError): status = e.resp.status except Exception: @@ -257,6 +279,7 @@ class GoogleChatAdapter(BaseChannelAdapter): self._chat_service = None self._credentials = None self._session_contexts.clear() + self._card_handlers.clear() if self._session_cleanup_task: self._session_cleanup_task.cancel() self._session_cleanup_task = None @@ -284,6 +307,9 @@ class GoogleChatAdapter(BaseChannelAdapter): error_detail = f"HTTP {status_code}: {error_detail}" return {"ok": False, "status": "error", "error": error_detail, "status_code": status_code} + def register_card_handler(self, action_name: str, handler: Callable) -> None: + self._card_handlers[action_name] = handler + async def send(self, response: ChannelResponse) -> DeliveryResult: chat_id = response.identity.channel_chat_id space_name = chat_id.split("/threads/")[0] @@ -349,13 +375,24 @@ class GoogleChatAdapter(BaseChannelAdapter): except CircuitBreakerOpenError: return DeliveryResult(success=False, error="Circuit breaker open") - async def edit_message(self, chat_id: str, msg_id: str, content: str) -> DeliveryResult: + async def edit_message( + self, + chat_id: str, + msg_id: str, + content: str, + cards: list[dict] | None = None, + update_fields: list[str] | None = None, + ) -> DeliveryResult: space_name = chat_id.split("/threads/")[0] await self._rate_limiter.acquire(space_name) body: dict[str, Any] = {"text": content} + if cards: + body["cards_v2"] = cards + update_fields = list(set((update_fields or ["text"]) + ["cardsV2"])) + mask = ",".join(update_fields or ["text"]) async def _do_send(): - return await update_message(self._chat_service, msg_id, body) + return await update_message(self._chat_service, msg_id, body, update_mask=mask) try: return await self._circuit_breaker.call(_do_send) @@ -374,7 +411,7 @@ class GoogleChatAdapter(BaseChannelAdapter): except CircuitBreakerOpenError: return DeliveryResult(success=False, error="Circuit breaker open") - async def send_reaction(self, chat_id: str, msg_id: str, emoji: str) -> DeliveryResult: + async def send_reaction(self, chat_id: str, msg_id: str, emoji: str, custom_emoji_uid: str = "") -> DeliveryResult: reactions_enabled = self._actions.get("reactions", True) if not reactions_enabled: return DeliveryResult(success=False, error="Agent reactions disabled by actions.reactions config") @@ -383,7 +420,7 @@ class GoogleChatAdapter(BaseChannelAdapter): await self._rate_limiter.acquire(space_name) async def _do_send(): - return await send_reaction(self._chat_service, msg_id, emoji) + return await send_reaction(self._chat_service, msg_id, emoji, custom_emoji_uid=custom_emoji_uid) try: return await self._circuit_breaker.call(_do_send) @@ -410,25 +447,10 @@ class GoogleChatAdapter(BaseChannelAdapter): async def send_typing_indicator(self, chat_id: str) -> DeliveryResult: space_name = chat_id.split("/threads/")[0] await self._rate_limiter.acquire(space_name) - body = format_outbound( - ChannelResponse( - identity=ChannelIdentity( - channel_id=self.channel_id, - channel_type=self.channel_type, - channel_user_id="", - channel_chat_id=chat_id, - ), - content="", - ) - ) - - async def _do_send(): - return await send_message(self._chat_service, chat_id, body) - - try: - return await self._circuit_breaker.call(_do_send) - except CircuitBreakerOpenError: - return DeliveryResult(success=False, error="Circuit breaker open") + msg_id = await self._stream_mgr.create_typing_message(self._chat_service, chat_id, self._bot_user or "Bot") + if msg_id: + return DeliveryResult(success=True, message_id=msg_id) + return DeliveryResult(success=False, error="Failed to create typing indicator") async def send_stream_chunk( self, @@ -876,6 +898,29 @@ class GoogleChatAdapter(BaseChannelAdapter): raise ChannelNotConnectedError() return await _download_media(self._chat_service, file_id) + async def get_message(self, msg_id: str) -> dict | None: + if not self._chat_service: + raise ChannelNotConnectedError() + from .send import get_message as _get_message + + return await _get_message(self._chat_service, msg_id) + + async def list_messages( + self, space_name: str, page_size: int = 50, page_token: str | None = None + ) -> tuple[list[dict], str | None]: + if not self._chat_service: + raise ChannelNotConnectedError() + from .send import list_messages as _list_messages + + return await _list_messages(self._chat_service, space_name, page_size, page_token) + + async def list_spaces(self, page_size: int = 50) -> tuple[list[dict], str | None]: + if not self._chat_service: + raise ChannelNotConnectedError() + from .directory import list_spaces as _list_spaces + + return await _list_spaces(self._chat_service, page_size) + async def _refresh_token_if_needed(self) -> bool: if not self._credentials or not self._credentials.valid: return False @@ -947,6 +992,16 @@ class GoogleChatAdapter(BaseChannelAdapter): queue.task_done() continue + if msg.event_type == EventType.CARD_ACTION: + action_data = json.loads(msg.content) if msg.content else {} + card_handler = self._card_handlers.get(action_data.get("function", "")) + if card_handler: + await card_handler(msg, action_data) + else: + logger.warning(f"No handler for card action: {action_data}") + queue.task_done() + continue + if msg.message_type == "command" and msg.metadata.get("slash_command"): handled = await self._handle_slash_command(msg, space_name) if handled: @@ -974,10 +1029,7 @@ class GoogleChatAdapter(BaseChannelAdapter): while self._status == ChannelStatus.CONNECTED: await asyncio.sleep(300) now = time.monotonic() - expired = [ - k for k, (ts, _) in self._session_contexts.items() - if now - ts > self._session_ttl_s - ] + expired = [k for k, (ts, _) in self._session_contexts.items() if now - ts > self._session_ttl_s] for k in expired: del self._session_contexts[k] if expired: @@ -996,22 +1048,24 @@ class GoogleChatAdapter(BaseChannelAdapter): async def _handle_slash_command(self, msg: ChannelMessage, space_name: str) -> bool: command = msg.metadata.get("slash_command", "") + if not command: + return False + chat_id = msg.identity.channel_chat_id + def _make_identity(): + return ChannelIdentity( + channel_id=self.channel_id, + channel_type=self.channel_type, + channel_user_id="", + channel_chat_id=chat_id, + ) + if command == "/help": from .slash_commands import get_command_help help_text = get_command_help() - response = ChannelResponse( - identity=ChannelIdentity( - channel_id=self.channel_id, - channel_type=self.channel_type, - channel_user_id="", - channel_chat_id=chat_id, - ), - content=help_text, - ) - await self.send(response) + await self.send(ChannelResponse(identity=_make_identity(), content=help_text)) return True if command == "/status": @@ -1024,19 +1078,57 @@ class GoogleChatAdapter(BaseChannelAdapter): f"• 待处理流消息: {self._stream_mgr.pending_count}\n" f"• 熔断器: {self._circuit_breaker.state}" ) - response = ChannelResponse( - identity=ChannelIdentity( - channel_id=self.channel_id, - channel_type=self.channel_type, - channel_user_id="", - channel_chat_id=chat_id, - ), - content=status_text, - ) - await self.send(response) + await self.send(ChannelResponse(identity=_make_identity(), content=status_text)) return True - return False + if command == "/reset": + thread_id = msg.metadata.get("thread_id", chat_id) + self._session_contexts.pop(thread_id, None) + self._touch_session(thread_id) + await self.send( + ChannelResponse( + identity=_make_identity(), + content="会话上下文已清除，可以开始新的对话。", + ) + ) + return True + + if command == "/history": + thread_id = msg.metadata.get("thread_id", chat_id) + _, ctx = self._session_contexts.get(thread_id, (0, {})) + history = ctx.get("history_summary", "暂无对话历史") + await self.send( + ChannelResponse( + identity=_make_identity(), + content=f"*对话历史摘要*\n{history}", + ) + ) + return True + + if command == "/context": + thread_id = msg.metadata.get("thread_id", chat_id) + _, ctx = self._session_contexts.get(thread_id, (0, {})) + ctx_info = json.dumps(ctx, ensure_ascii=False, indent=2) + await self.send( + ChannelResponse( + identity=_make_identity(), + content=f"*当前上下文*\n```json\n{ctx_info}\n```", + ) + ) + return True + + if command == "/summary": + msg.metadata["request_summary"] = True + await self._handle_message(msg) + return True + + await self.send( + ChannelResponse( + identity=_make_identity(), + content=f"命令 `{command}` 暂未实现，即将支持。发送 `/help` 查看可用命令。", + ) + ) + return True def _convert_addon_event(self, common_event: dict) -> dict: event_type = common_event.get("type", "MESSAGE") @@ -1431,6 +1523,21 @@ class GoogleChatAdapter(BaseChannelAdapter): push_config = pubsub_v1.PushConfig(push_endpoint=webhook_url) topic_path = subscriber.topic_path(project_id, topic) + publisher = pubsub_v1.PublisherClient(credentials=self._credentials) + try: + publisher.get_topic(topic=topic_path) + logger.info(f"Pub/Sub topic exists: {topic_path}") + except Exception: + try: + publisher.create_topic(name=topic_path) + logger.info(f"Created Pub/Sub topic: {topic_path}") + except Exception as topic_err: + if hasattr(topic_err, "code") and getattr(topic_err, "code") == 409: + logger.info(f"Pub/Sub topic already exists (race): {topic_path}") + else: + logger.error(f"Failed to create Pub/Sub topic {topic}: {topic_err}") + return + try: subscriber.create_subscription( name=sub_path, diff --git a/backend/package/yuxi/channels/adapters/googlechat/cards.py b/backend/package/yuxi/channels/adapters/googlechat/cards.py index 2a1a8691..cc46c7b5 100644 --- a/backend/package/yuxi/channels/adapters/googlechat/cards.py +++ b/backend/package/yuxi/channels/adapters/googlechat/cards.py @@ -90,6 +90,15 @@ def build_form_card(title: str, fields: list[dict], submit_action: str) -> dict: } } ) + elif field_input_type in ("DATE_AND_TIME", "DATE_ONLY", "TIME_ONLY"): + widgets.append( + _make_date_time_picker( + name=field.get("name", ""), + label=field.get("label", ""), + type_=field_input_type, + value_ms_epoch=field.get("value_ms_epoch"), + ) + ) elif field_input_type == "DIVIDER": widgets.append({"divider": {}}) else: @@ -216,3 +225,41 @@ def _make_button(text: str, action: str, green: bool = False) -> dict: "color": color, "onClick": {"action": {"function": action}}, } + + +def _make_date_time_picker( + name: str, + label: str = "", + type_: str = "DATE_AND_TIME", + value_ms_epoch: int | None = None, +) -> dict: + widget = { + "dateTimePicker": { + "name": name, + "label": label, + "type": type_, + } + } + if value_ms_epoch is not None: + widget["dateTimePicker"]["valueMsEpoch"] = str(value_ms_epoch) + return widget + + +def _make_grid(column_count: int, items: list[dict]) -> dict: + return { + "grid": { + "columnCount": column_count, + "items": [{"widgets": [item]} for item in items], + } + } + + +def _make_columns(column_items: list[list[dict]]) -> dict: + return {"columns": {"columnItems": [{"widgets": col_widgets} for col_widgets in column_items]}} + + +def _make_link_button(text: str, url: str) -> dict: + return { + "text": text, + "onClick": {"openLink": {"url": url}}, + } diff --git a/backend/package/yuxi/channels/adapters/googlechat/directory.py b/backend/package/yuxi/channels/adapters/googlechat/directory.py index 590bcaec..328dfd31 100644 --- a/backend/package/yuxi/channels/adapters/googlechat/directory.py +++ b/backend/package/yuxi/channels/adapters/googlechat/directory.py @@ -1,6 +1,30 @@ from __future__ import annotations -from typing import Any +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from googleapiclient.discovery import Resource + +from yuxi.utils.logging_config import logger + + +async def list_spaces( + chat_service: Resource, + page_size: int = 50, + page_token: str | None = None, + filter_str: str | None = None, +) -> tuple[list[dict], str | None]: + kwargs: dict[str, Any] = {"pageSize": page_size} + if page_token: + kwargs["pageToken"] = page_token + if filter_str: + kwargs["filter"] = filter_str + try: + result = chat_service.spaces().list(**kwargs).execute() + return result.get("spaces", []), result.get("nextPageToken") + except Exception as e: + logger.warning(f"list_spaces failed: {e}") + return [], None def list_peers(allow_from: list[str]) -> list[dict[str, Any]]: diff --git a/backend/package/yuxi/channels/adapters/googlechat/formatter.py b/backend/package/yuxi/channels/adapters/googlechat/formatter.py index eef24b8b..d3dfd96e 100644 --- a/backend/package/yuxi/channels/adapters/googlechat/formatter.py +++ b/backend/package/yuxi/channels/adapters/googlechat/formatter.py @@ -23,7 +23,10 @@ def format_outbound(response: ChannelResponse) -> dict[str, Any]: for att in response.attachments: if att.url and att.type in ("image", "IMAGE", "video", "VIDEO"): body.setdefault("cards_v2", []) - body["cards_v2"].append({"card": {"sections": [{"widgets": [{"image": {"imageUrl": att.url}}]}]}}) + img_widget = {"image": {"imageUrl": att.url}} + if att.filename: + img_widget["image"]["altText"] = att.filename + body["cards_v2"].append({"card": {"sections": [{"widgets": [img_widget]}]}}) if response.metadata and response.metadata.get("thread_key"): body["messageReplyOption"] = "REPLY_MESSAGE_FALLBACK_TO_NEW_THREAD" diff --git a/backend/package/yuxi/channels/adapters/googlechat/media.py b/backend/package/yuxi/channels/adapters/googlechat/media.py index ac3bbbeb..c097565f 100644 --- a/backend/package/yuxi/channels/adapters/googlechat/media.py +++ b/backend/package/yuxi/channels/adapters/googlechat/media.py @@ -8,65 +8,11 @@ if TYPE_CHECKING: from googleapiclient.discovery import Resource from yuxi.channels.exceptions import DeliveryFailedError -from yuxi.channels.models import DeliveryResult from yuxi.utils.logging_config import logger _MAX_DOWNLOAD_SIZE_BYTES = 100 * 1024 * 1024 -async def upload_image( - chat_service: Resource, - chat_id: str, - image_data: bytes, - filename: str = "image.png", - caption: str = "", -) -> DeliveryResult: - from googleapiclient.http import MediaIoBaseUpload - - try: - media = MediaIoBaseUpload( - io.BytesIO(image_data), - mimetype="image/png", - resumable=True, - ) - body = {"text": caption or " "} - result = chat_service.spaces().messages().create(parent=chat_id, body=body, media_body=media).execute() - return DeliveryResult( - success=True, - message_id=result.get("name", ""), - ) - except Exception as e: - logger.error(f"upload_image failed: {e}") - return DeliveryResult(success=False, error=str(e)) - - -async def upload_file( - chat_service: Resource, - chat_id: str, - file_data: bytes, - filename: str, - mime_type: str = "application/octet-stream", - caption: str = "", -) -> DeliveryResult: - from googleapiclient.http import MediaIoBaseUpload - - try: - media = MediaIoBaseUpload( - io.BytesIO(file_data), - mimetype=mime_type, - resumable=True, - ) - body = {"text": caption or " "} - result = chat_service.spaces().messages().create(parent=chat_id, body=body, media_body=media).execute() - return DeliveryResult( - success=True, - message_id=result.get("name", ""), - ) - except Exception as e: - logger.error(f"upload_file failed: {e}") - return DeliveryResult(success=False, error=str(e)) - - async def download_media(chat_service: Resource, file_id: str) -> bytes: try: file_info = json.loads(file_id) diff --git a/backend/package/yuxi/channels/adapters/googlechat/normalizer.py b/backend/package/yuxi/channels/adapters/googlechat/normalizer.py index 33fb198d..bb2c6d64 100644 --- a/backend/package/yuxi/channels/adapters/googlechat/normalizer.py +++ b/backend/package/yuxi/channels/adapters/googlechat/normalizer.py @@ -90,6 +90,8 @@ def normalize_inbound( "route_envelope": route_envelope, "slash_command": command, "slash_args": command_args, + "is_edited": detect_message_edit(message), + "is_forwarded": is_forwarded_message(message), }, ) diff --git a/backend/package/yuxi/channels/adapters/googlechat/pairing.py b/backend/package/yuxi/channels/adapters/googlechat/pairing.py index 503128e1..6817356b 100644 --- a/backend/package/yuxi/channels/adapters/googlechat/pairing.py +++ b/backend/package/yuxi/channels/adapters/googlechat/pairing.py @@ -1,5 +1,8 @@ from __future__ import annotations +import json +import os +import threading from typing import TYPE_CHECKING if TYPE_CHECKING: @@ -16,6 +19,74 @@ _PAIRING_CHALLENGE_MESSAGE = ( ) +class PairingStore: + def __init__(self, storage_path: str = ""): + self._paired_users: dict[str, str] = {} + self._lock = threading.Lock() + self._storage_path = storage_path or os.path.join( + os.path.expanduser("~"), ".forcepilot", "googlechat_pairing.json" + ) + self._load() + + def _load(self) -> None: + try: + if os.path.exists(self._storage_path): + with open(self._storage_path, encoding="utf-8") as f: + self._paired_users = json.load(f) + logger.debug(f"Loaded {len(self._paired_users)} paired users from {self._storage_path}") + except Exception as e: + logger.warning(f"Failed to load pairing store: {e}") + self._paired_users = {} + + def _save(self) -> None: + try: + os.makedirs(os.path.dirname(self._storage_path), exist_ok=True) + with open(self._storage_path, "w", encoding="utf-8") as f: + json.dump(self._paired_users, f, indent=2) + except Exception as e: + logger.warning(f"Failed to save pairing store: {e}") + + def is_paired(self, space_id: str) -> bool: + with self._lock: + return space_id in self._paired_users + + def pair(self, space_id: str, user_id: str) -> None: + with self._lock: + self._paired_users[space_id] = user_id + self._save() + + def unpair(self, space_id: str) -> bool: + with self._lock: + if space_id in self._paired_users: + del self._paired_users[space_id] + self._save() + return True + return False + + def get_paired_user(self, space_id: str) -> str | None: + with self._lock: + return self._paired_users.get(space_id) + + def list_paired_spaces(self) -> list[str]: + with self._lock: + return list(self._paired_users.keys()) + + def clear(self) -> None: + with self._lock: + self._paired_users.clear() + self._save() + + +_global_pairing_store: PairingStore | None = None + + +def get_pairing_store(storage_path: str = "") -> PairingStore: + global _global_pairing_store + if _global_pairing_store is None: + _global_pairing_store = PairingStore(storage_path) + return _global_pairing_store + + async def send_pairing_challenge( chat_service: Resource, space_name: str, diff --git a/backend/package/yuxi/channels/adapters/googlechat/policy.py b/backend/package/yuxi/channels/adapters/googlechat/policy.py index 46cd1c50..45e13d2f 100644 --- a/backend/package/yuxi/channels/adapters/googlechat/policy.py +++ b/backend/package/yuxi/channels/adapters/googlechat/policy.py @@ -4,7 +4,7 @@ from dataclasses import dataclass, field from enum import StrEnum from typing import Any -from yuxi.channels.models import ChatType, ChannelMessage +from yuxi.channels.models import ChannelMessage, ChatType from yuxi.utils.logging_config import logger diff --git a/backend/package/yuxi/channels/adapters/googlechat/pubsub_.py b/backend/package/yuxi/channels/adapters/googlechat/pubsub_.py index a34f6263..a4f3d8c5 100644 --- a/backend/package/yuxi/channels/adapters/googlechat/pubsub_.py +++ b/backend/package/yuxi/channels/adapters/googlechat/pubsub_.py @@ -1,7 +1,7 @@ from __future__ import annotations -import json import base64 +import json from yuxi.utils.logging_config import logger diff --git a/backend/package/yuxi/channels/adapters/googlechat/send.py b/backend/package/yuxi/channels/adapters/googlechat/send.py index bc9175bb..50819092 100644 --- a/backend/package/yuxi/channels/adapters/googlechat/send.py +++ b/backend/package/yuxi/channels/adapters/googlechat/send.py @@ -55,6 +55,9 @@ def _classify_error(e: Exception, account_id: str = "default") -> None: if isinstance(e, HttpError): status = e.resp.status if status == 429: + retry_after = _extract_retry_after(e) + if retry_after: + raise ChannelRateLimitError(f"Rate limited, retry after {retry_after}s") raise ChannelRateLimitError() if status in (401, 403): reason = _extract_rate_limit_reason(e) @@ -81,6 +84,19 @@ def _extract_rate_limit_reason(e: Exception) -> bool: return False +def _extract_retry_after(e: Exception) -> int | None: + try: + resp = e.resp if hasattr(e, "resp") else None + if resp is None: + return None + retry_after = resp.get("Retry-After") or resp.get("retry-after") + if retry_after: + return int(retry_after) + except (ValueError, TypeError): + pass + return None + + def _backoff_with_jitter(retry_count: int) -> float: delay = min((2**retry_count) + random.uniform(0, 1), _MAX_BACKOFF_S) return delay @@ -195,10 +211,14 @@ async def send_reaction( chat_service: Resource, message_id: str, emoji: str, + custom_emoji_uid: str = "", account_id: str = "default", ) -> DeliveryResult: async def _do_send(): - body = {"emoji": {"unicode": emoji}} + if custom_emoji_uid: + body = {"emoji": {"customEmoji": {"uid": custom_emoji_uid}}} + else: + body = {"emoji": {"unicode": emoji}} (chat_service.spaces().messages().reactions().create(parent=message_id, body=body).execute()) return DeliveryResult(success=True) @@ -302,6 +322,39 @@ async def find_direct_message_space( return None +async def get_message( + chat_service: Resource, + message_id: str, +) -> dict | None: + try: + result = chat_service.spaces().messages().get(name=message_id).execute() + return result + except Exception as e: + logger.warning(f"get_message failed for {message_id}: {e}") + return None + + +async def list_messages( + chat_service: Resource, + space_name: str, + page_size: int = 50, + page_token: str | None = None, + filter_str: str | None = None, +) -> tuple[list[dict], str | None]: + kwargs: dict[str, Any] = {"parent": space_name, "pageSize": page_size} + if page_token: + kwargs["pageToken"] = page_token + if filter_str: + kwargs["filter"] = filter_str + + try: + result = chat_service.spaces().messages().list(**kwargs).execute() + return result.get("messages", []), result.get("nextPageToken") + except Exception as e: + logger.warning(f"list_messages failed for {space_name}: {e}") + return [], None + + async def resolve_outbound_space( chat_service: Resource, target: str, diff --git a/backend/package/yuxi/channels/adapters/googlechat/setup.py b/backend/package/yuxi/channels/adapters/googlechat/setup.py index 48a54c81..05976bc6 100644 --- a/backend/package/yuxi/channels/adapters/googlechat/setup.py +++ b/backend/package/yuxi/channels/adapters/googlechat/setup.py @@ -2,7 +2,6 @@ from __future__ import annotations from typing import Any - _SETUP_GUIDE = """ Google Chat Setup Guide ======================= @@ -64,8 +63,8 @@ class SetupWizard: return {"step": self._current_step, "steps": len(self._steps), "data": self.current_step} def next(self, answer: Any = None) -> dict[str, Any]: - if self._current_step > 0 and answer is not None: - self._results[self._steps[self._current_step - 1]["id"]] = answer + if answer is not None: + self._results[self._steps[self._current_step]["id"]] = answer self._current_step += 1 if self._current_step >= len(self._steps): diff --git a/backend/package/yuxi/channels/adapters/googlechat/slash_commands.py b/backend/package/yuxi/channels/adapters/googlechat/slash_commands.py index 4ae435c3..588ceeba 100644 --- a/backend/package/yuxi/channels/adapters/googlechat/slash_commands.py +++ b/backend/package/yuxi/channels/adapters/googlechat/slash_commands.py @@ -1,6 +1,5 @@ from __future__ import annotations - SLASH_COMMAND_MAP: dict[str, str] = { "/reset": "清除当前会话上下文，重新开始对话", "/history": "查看当前会话的对话历史摘要", diff --git a/backend/package/yuxi/channels/adapters/googlechat/ssrf.py b/backend/package/yuxi/channels/adapters/googlechat/ssrf.py index aab2d08e..495c0861 100644 --- a/backend/package/yuxi/channels/adapters/googlechat/ssrf.py +++ b/backend/package/yuxi/channels/adapters/googlechat/ssrf.py @@ -2,70 +2,69 @@ from __future__ import annotations from typing import Any -from yuxi.utils.logging_config import logger - -_UNSAFE_TRANSPORT_FIELDS = frozenset( - { - "agent", - "cert", - "cert_file", - "key", - "key_file", - "dispatcher", - "proxy", - "session", - } +from yuxi.channels.auth.ssrf_guard import ( + apply_ssrf_guard_defaults, +) +from yuxi.channels.auth.ssrf_guard import ( + check_response_size as _check_response_size, +) +from yuxi.channels.auth.ssrf_guard import ( + check_response_text as _check_response_text, +) +from yuxi.channels.auth.ssrf_guard import ( + is_private_url as _is_private_url, +) +from yuxi.channels.auth.ssrf_guard import ( + sanitize_transport_kwargs as _sanitize_transport_kwargs, +) +from yuxi.channels.auth.ssrf_guard import ( + validate_url_with_whitelist as _validate_url_with_whitelist, ) -_AUTH_RESPONSE_MAX_BYTES = 1 * 1024 * 1024 +_GOOGLE_CHAT_ALLOWED_HOSTS = [ + "*.googleapis.com", + "*.google.com", + "chat.googleapis.com", + "oauth2.googleapis.com", +] def sanitize_google_auth_init(kwargs: dict[str, Any]) -> dict[str, Any]: cleaned: dict[str, Any] = {} - removed: list[str] = [] for key, value in kwargs.items(): - if key in _UNSAFE_TRANSPORT_FIELDS: - removed.append(key) - continue - cleaned[key] = value - if removed: - logger.warning(f"SSRF guard: removed unsafe transport fields: {removed}") - return cleaned + if key not in apply_ssrf_guard_defaults(): + cleaned[key] = value + safe = _sanitize_transport_kwargs(cleaned) + unsafe_defaults = apply_ssrf_guard_defaults() + for k, v in unsafe_defaults.items(): + safe.setdefault(k, v) + return safe def sanitize_request_kwargs(kwargs: dict[str, Any]) -> dict[str, Any]: - cleaned: dict[str, Any] = {} - removed: list[str] = [] - for key, value in kwargs.items(): - if key in _UNSAFE_TRANSPORT_FIELDS: - removed.append(key) - continue - cleaned[key] = value - if removed: - logger.warning(f"SSRF guard: removed unsafe request fields: {removed}") - return cleaned + return _sanitize_transport_kwargs(kwargs) def apply_ssrf_guard() -> dict[str, Any]: - return { - "agent": None, - "cert": None, - "cert_file": None, - "key": None, - "key_file": None, - } + return apply_ssrf_guard_defaults() def check_auth_response_size(data: bytes) -> bytes: - if len(data) > _AUTH_RESPONSE_MAX_BYTES: - logger.warning( - f"Google Auth response exceeds size limit: {len(data)} bytes > {_AUTH_RESPONSE_MAX_BYTES} bytes, truncating" - ) - return data[:_AUTH_RESPONSE_MAX_BYTES] - return data + return _check_response_size(data) def check_auth_response_text(text: str) -> str: - data = text.encode("utf-8", errors="replace") - truncated = check_auth_response_size(data) - return truncated.decode("utf-8", errors="replace") + return _check_response_text(text) + + +def validate_outbound_url(url: str, allow_private: bool = False) -> str: + return _validate_url_with_whitelist( + url, + allowed_hosts=_GOOGLE_CHAT_ALLOWED_HOSTS, + enforce_https=True, + allow_private=allow_private, + ) + + +def is_private_target(url: str) -> bool: + return _is_private_url(url) diff --git a/backend/package/yuxi/channels/adapters/googlechat/streaming.py b/backend/package/yuxi/channels/adapters/googlechat/streaming.py index a36ced29..ad7a6b59 100644 --- a/backend/package/yuxi/channels/adapters/googlechat/streaming.py +++ b/backend/package/yuxi/channels/adapters/googlechat/streaming.py @@ -10,7 +10,7 @@ if TYPE_CHECKING: from yuxi.channels.models import DeliveryResult from yuxi.utils.logging_config import logger -from .send import update_message, send_message, delete_message +from .send import delete_message, send_message, update_message DEFAULT_UPDATE_INTERVAL_MS = 800 DEFAULT_COALESCE_MIN_CHARS = 1500 @@ -183,7 +183,10 @@ class StreamManager: self._typing_messages.pop(chat_id, None) self._typing_replaced.add(chat_id) - body = {"text": text if finished else text + " ⏳"} + if self._coalesce_min_chars > 0 and not finished: + body = {"text": text + "\n\n_正在生成中..._"} + else: + body = {"text": text if finished else text + " ⏳"} result = await update_message(chat_service, message_name, body) if result.success: diff --git a/backend/package/yuxi/channels/adapters/googlechat/target.py b/backend/package/yuxi/channels/adapters/googlechat/target.py index dd8e6f08..9ec3f787 100644 --- a/backend/package/yuxi/channels/adapters/googlechat/target.py +++ b/backend/package/yuxi/channels/adapters/googlechat/target.py @@ -3,9 +3,24 @@ from __future__ import annotations import re from typing import Any - _CHANNEL_PREFIXES = ("googlechat:", "gchat:", "google-chat:") +_VALID_GC_RESOURCE_PREFIXES = frozenset({"spaces/", "users/"}) +_MESSAGES_SUFFIX_RE = re.compile(r"/messages/[^/]+$") + + +def strip_message_suffix(target: str) -> tuple[str, bool]: + stripped = _MESSAGES_SUFFIX_RE.sub("", target) + return stripped, stripped != target + + +def detect_deprecated_target(target: str) -> str | None: + cleaned = _strip_channel_prefix(target) + for prefix in _VALID_GC_RESOURCE_PREFIXES: + if cleaned.startswith(prefix): + return None + return f"Target '{target}' does not use a valid Google Chat resource prefix (spaces/ or users/)" + def normalize_googlechat_target(target: str) -> dict[str, Any]: if not target: