diff --git a/backend/package/yuxi/channel/extensions/ringcentral/__init__.py b/backend/package/yuxi/channel/extensions/ringcentral/__init__.py new file mode 100644 index 00000000..caa1c8f9 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/ringcentral/__init__.py @@ -0,0 +1,409 @@ +from __future__ import annotations + +import logging + +from yuxi.channel.capabilities import ChannelCapabilities +from yuxi.channel.extensions.base import BaseChannelPlugin +from yuxi.channel.extensions.ringcentral.config import RingCentralConfigAdapter +from yuxi.channel.extensions.ringcentral.dedupe import RingCentralDeduplicator +from yuxi.channel.extensions.ringcentral.gateway import RingCentralGatewayAdapter +from yuxi.channel.extensions.ringcentral.mentions import extract_mentions, strip_mentions +from yuxi.channel.extensions.ringcentral.monitor import RingCentralMonitor +from yuxi.channel.extensions.ringcentral.outbound import RingCentralOutboundAdapter +from yuxi.channel.extensions.ringcentral.security import RingCentralSecurityAdapter +from yuxi.channel.extensions.ringcentral.session import RingCentralSessionAdapter +from yuxi.channel.extensions.ringcentral.status import RingCentralStatusAdapter +from yuxi.channel.plugins.registry import ChannelPluginRegistry +from yuxi.channel.sdk.actions import MessageAction, MessageActionRegistry + +logger = logging.getLogger(__name__) + + +class RingCentralPlugin(BaseChannelPlugin): + id = "ringcentral" + name = "RingCentral" + order = 80 + label = "RingCentral (Team Messaging)" + aliases = ["rc", "glip", "ringcentral-glip"] + + def __init__(self): + self._config = RingCentralConfigAdapter() + self._gateway = RingCentralGatewayAdapter() + self._outbound = RingCentralOutboundAdapter() + self._monitor = RingCentralMonitor() + self._status = RingCentralStatusAdapter() + self._security = RingCentralSecurityAdapter() + self._session = RingCentralSessionAdapter() + self._dedup = RingCentralDeduplicator() + self._actions = MessageActionRegistry() + self._actions.register( + MessageAction.SEND, self._handle_send, description="发送文本消息到 RingCentral" + ) + self._actions.register( + MessageAction.SEND_MEDIA, self._handle_send_media, description="发送图片/文件到 RingCentral" + ) + self._actions.register( + MessageAction.EDIT, self._handle_edit, description="编辑已发送的消息" + ) + self._actions.register( + MessageAction.REACT, self._handle_react, description="对消息添加表情反应" + ) + + @property + def capabilities(self) -> ChannelCapabilities: + return ChannelCapabilities( + chat_types=["direct", "group", "team"], + message_types=["text", "image", "file"], + reactions=True, + typing_indicator=False, + threads=False, + edit=True, + unsend=True, + reply=False, + media=True, + streaming=True, + streaming_mode="block", + block_streaming=True, + block_streaming_chunk_min_chars=500, + block_streaming_chunk_max_chars=1000, + block_streaming_coalesce_min_chars=200, + adaptive_cards=True, + tasks=False, + events=False, + notes=False, + ) + + @property + def actions(self): + return self._actions + + # ── ConfigProtocol ──────────────────────────────────── + + def list_account_ids(self, config: dict) -> list[str]: + return self._config.list_account_ids(config) + + async def resolve_account(self, account_id: str) -> dict: + return await self._config.resolve_account(account_id) + + def is_configured(self, account: dict) -> bool: + return self._config.is_configured(account) + + def is_enabled(self, account: dict) -> bool: + return self._config.is_enabled(account) + + def disabled_reason(self, account: dict) -> str: + return self._config.disabled_reason(account) + + async def resolve_allow_from(self, config: dict, account_id: str) -> list[str] | None: + return await self._config.resolve_allow_from(config, account_id) + + def describe_account(self, account: dict) -> dict: + return self._config.describe_account(account) + + # ── GatewayProtocol ─────────────────────────────────── + + async def start(self, ctx) -> object: + return await self._gateway.start(ctx) + + async def stop(self, ctx) -> None: + await self._gateway.stop(ctx) + + # ── OutboundProtocol ────────────────────────────────── + + async def send_text( + self, + target_id: str, + content: str, + *, + reply_to_id: str | None = None, + thread_id: str | None = None, + account_id: str | None = None, + ) -> None: + await self._outbound.send_text( + target_id, + content, + reply_to_id=reply_to_id, + thread_id=thread_id, + account_id=account_id, + ) + + async def send_media( + self, + target_id: str, + media_url: str, + media_type: str, + reply_to_id: str | None = None, + thread_id: str | None = None, + account_id: str | None = None, + ) -> None: + await self._outbound.send_media( + target_id, + media_url, + media_type, + reply_to_id=reply_to_id, + thread_id=thread_id, + account_id=account_id, + ) + + async def edit_message( + self, + target_id: str, + message_id: str, + content: str, + *, + thread_id: str | None = None, + account_id: str | None = None, + ) -> str | None: + return await self._outbound.edit_message( + target_id, + message_id, + content, + thread_id=thread_id, + account_id=account_id, + ) + + def chunker(self, text: str, limit: int, ctx: object | None = None) -> list[str]: + return self._outbound.chunker(text, limit, ctx) + + def sanitize_text(self, text: str, payload: object | None = None) -> str: + return self._outbound.sanitize_text_out(text, payload) + + # ── CardProtocol ────────────────────────────────────── + + async def send_card( + self, + target_id: str, + card_content: dict, + *, + reply_to_id: str | None = None, + account_id: str | None = None, + ) -> str | None: + from yuxi.channel.extensions.ringcentral.adaptive_cards import send_adaptive_card + + client = self._gateway.get_client(account_id or "default") + if not client: + return None + result = await send_adaptive_card(client, target_id, card_content) + return result.get("id") + + async def edit_card( + self, + target_id: str, + message_id: str, + card_content: dict, + *, + account_id: str | None = None, + ) -> str | None: + from yuxi.channel.extensions.ringcentral.adaptive_cards import update_adaptive_card + + client = self._gateway.get_client(account_id or "default") + if not client: + return None + result = await update_adaptive_card(client, message_id, card_content) + return message_id if result else None + + # ── StatusProtocol ──────────────────────────────────── + + async def probe(self, account: dict) -> bool: + return await self._status.probe(account) + + def build_summary(self, snapshot: object) -> dict: + return self._status.build_summary(snapshot) + + # ── SecurityProtocol ────────────────────────────────── + + async def check_allowlist(self, peer_id: str, channel_type: str) -> bool: + try: + accounts = self._config.list_account_ids({}) + for account_id in accounts: + resolved = await self._config.resolve_account(account_id) + account = resolved.get("resolved") + if account and not self._security.check_allowlist(peer_id, account): + return False + except Exception: + pass + return True + + def resolve_dm_policy(self) -> dict: + return self._security.resolve_dm_policy() + + # ── PairingProtocol ─────────────────────────────────── + + async def generate_code(self, peer_id: str) -> str: + from yuxi.channel.extensions.ringcentral.pairing import RingCentralPairingAdapter + + adapter = RingCentralPairingAdapter() + return await adapter.generate_code(peer_id) + + async def verify_code(self, peer_id: str, code: str) -> bool: + from yuxi.channel.extensions.ringcentral.pairing import RingCentralPairingAdapter + + adapter = RingCentralPairingAdapter() + return await adapter.verify_code(peer_id, code) + + # ── MessagingProtocol ───────────────────────────────── + + def resolve_session(self, msg): + return self._session.resolve_session(msg) + + def resolve_reply_to_mode( + self, + config: dict, + account_id: str | None = None, + chat_type: str | None = None, + ) -> str: + return "none" + + # ── ReactionProtocol ────────────────────────────────── + + async def send_reaction( + self, + target_id: str, + message_id: str, + emoji: str, + *, + account_id: str | None = None, + ) -> None: + client = self._gateway.get_client(account_id or "default") + if not client: + return + await client.post( + f"/restapi/v1.0/glip/posts/{message_id}/reactions", + body={"reaction": emoji}, + ) + + async def remove_reaction( + self, + target_id: str, + message_id: str, + emoji: str, + *, + account_id: str | None = None, + ) -> None: + client = self._gateway.get_client(account_id or "default") + if not client: + return + await client.delete(f"/restapi/v1.0/glip/posts/{message_id}/reactions/{emoji}") + + # ── WebhookProtocol ─────────────────────────────────── + + async def handle_webhook(self, request: object, account_id: str | None = None) -> object: + return None + + # ── InboundHandlerProtocol ──────────────────────────── + + async def handle_raw_event(self, event: dict, account: dict) -> object | None: + from yuxi.channel.extensions.ringcentral.types import RingCentralEvent + + rc_event = RingCentralEvent.from_payload(event) + resolved = account.get("resolved") + return await self._monitor.event_to_unified_message(event, rc_event, resolved) + + # ── FormatProtocol ──────────────────────────────────── + + def markdown_to_native(self, md_text: str) -> dict | str: + from yuxi.channel.extensions.ringcentral.formatting import sanitize_text + + return sanitize_text(md_text) + + def native_to_markdown(self, native_content: dict | str) -> str: + if isinstance(native_content, dict): + return native_content.get("text", "") + return str(native_content) + + # ── DedupeProtocol ──────────────────────────────────── + + def is_duplicate(self, key: str) -> bool: + return self._dedup.is_duplicate(key) + + def mark_seen(self, key: str) -> None: + self._dedup.mark_seen(key) + + def reset(self) -> None: + self._dedup.reset() + + @property + def ttl_seconds(self) -> int: + return self._dedup.ttl_seconds + + @property + def max_entries(self) -> int: + return self._dedup.max_entries + + # ── MentionsProtocol ────────────────────────────────── + + def extract_mentions(self, raw_message: dict) -> list[str]: + text = raw_message.get("body", {}).get("text", "") + return [name for name, _ in extract_mentions(text)] + + def strip_mentions(self, text: str, ctx: object | None = None) -> str: + return strip_mentions(text) + + # ── ConfigSchemaProtocol ────────────────────────────── + + def config_schema(self) -> dict: + return self._config.config_schema() + + def collect_warnings(self, config: dict, account_id: str | None = None) -> list[str]: + return self._security.collect_warnings(config, account_id) + + # ── AgentPromptProtocol ─────────────────────────────── + + @property + def channel_format_instructions(self) -> str | None: + return ( + "You are responding via RingCentral Team Messaging. Messages support Markdown " + "(bold, italic, strikethrough, code, code blocks, blockquotes, lists, headings). " + "@mention format: @![DisplayName](extensionId). " + "Messages are limited to 1000 characters and will be automatically chunked if longer. " + "Be concise and avoid excessive formatting." + ) + + def build_system_prompt(self, context) -> str | None: + return ( + "你正在通过 RingCentral Team Messaging (Glip) 与用户交互。" + "RingCentral 支持 Markdown 格式、自适应卡片、表情反应等功能。" + "请使用简洁清晰的格式回复，大字报式内容会自动分段发送。" + ) + + # ── HeartbeatProtocol ───────────────────────────────── + + async def send_typing(self, target_id: str, thread_id: str | None = None) -> None: + pass + + async def clear_typing(self, target_id: str, thread_id: str | None = None) -> None: + pass + + # ── MessageAction handlers ──────────────────────────── + + async def _handle_send( + self, *, content: str, target_id: str, + reply_to_id: str | None = None, thread_id: str | None = None, **kwargs, + ) -> dict: + await self.send_text( + target_id, content, reply_to_id=reply_to_id, thread_id=thread_id, + ) + return {"success": True} + + async def _handle_send_media( + self, *, media_url: str, media_type: str, target_id: str, **kwargs, + ) -> dict: + await self.send_media(target_id, media_url, media_type) + return {"success": True} + + async def _handle_edit( + self, *, message_id: str, content: str, + target_id: str, **kwargs, + ) -> dict: + result = await self.edit_message(target_id, message_id, content) + return {"success": result is not None, "msg_id": result} + + async def _handle_react( + self, *, message_id: str, emoji: str, **kwargs, + ) -> dict: + await self.send_reaction("", message_id, emoji) + return {"success": True} + + +ringcentral_plugin = RingCentralPlugin() +ChannelPluginRegistry.register(ringcentral_plugin) diff --git a/backend/package/yuxi/channel/extensions/ringcentral/adaptive_cards.py b/backend/package/yuxi/channel/extensions/ringcentral/adaptive_cards.py new file mode 100644 index 00000000..560f2951 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/ringcentral/adaptive_cards.py @@ -0,0 +1,40 @@ +from __future__ import annotations + +import logging + +from yuxi.channel.extensions.ringcentral.sdk import AsyncRingCentralClient + +logger = logging.getLogger(__name__) + +TEAM_MESSAGING_BASE = "/team-messaging/v1" + + +async def send_adaptive_card( + client: AsyncRingCentralClient, + chat_id: str, + card: dict, +) -> dict: + return await client.post( + f"{TEAM_MESSAGING_BASE}/chats/{chat_id}/adaptive-cards", + body=card, + ) + + +async def update_adaptive_card( + client: AsyncRingCentralClient, + card_id: str, + card: dict, +) -> dict: + return await client.put( + f"{TEAM_MESSAGING_BASE}/adaptive-cards/{card_id}", + body=card, + ) + + +async def get_adaptive_card( + client: AsyncRingCentralClient, + card_id: str, +) -> dict: + return await client.get( + f"{TEAM_MESSAGING_BASE}/adaptive-cards/{card_id}", + ) diff --git a/backend/package/yuxi/channel/extensions/ringcentral/config.py b/backend/package/yuxi/channel/extensions/ringcentral/config.py new file mode 100644 index 00000000..3cfcaa1b --- /dev/null +++ b/backend/package/yuxi/channel/extensions/ringcentral/config.py @@ -0,0 +1,167 @@ +from __future__ import annotations + +import logging +import os + +from yuxi.channel.extensions.ringcentral.types import ResolvedRingCentralAccount + +logger = logging.getLogger(__name__) + +_DEFAULT_WEBHOOK_PATH = "/api/webhook/ringcentral" + + +class RingCentralConfigAdapter: + DEFAULT_ACCOUNT_ID = "default" + + def list_account_ids(self, config: dict) -> list[str]: + raw = config.get("channels", {}).get("ringcentral", {}) + accounts = raw.get("accounts", {}) + if accounts: + return list(accounts.keys()) + if raw.get("client_id"): + return [self.DEFAULT_ACCOUNT_ID] + return [] + + async def resolve_account(self, account_id: str) -> dict: + from yuxi.channel.runtime.manager import gateway + + config = gateway.global_config if gateway is not None else {} + rc_config = config.get("channels", {}).get("ringcentral", {}) if config else {} + + defaults = rc_config.get("accounts", {}).get("default", {}) + account_cfg = rc_config.get("accounts", {}).get(account_id, {}) + merged = {**rc_config, **defaults, **account_cfg} + + resolved = self._build_resolved(account_id, merged) + + return { + "account_id": account_id, + "name": resolved.name, + "client_id": resolved.client_id, + "client_secret": resolved.client_secret, + "server_url": resolved.server_url, + "jwt_token": resolved.jwt_token, + "webhook_path": resolved.webhook_path, + "webhook_validation_token": resolved.webhook_validation_token, + "bot_person_id": resolved.bot_person_id, + "subscription_id": resolved.subscription_id, + "dm_policy": resolved.dm_policy, + "dm_enabled": resolved.dm_enabled, + "dm_allow_from": resolved.dm_allow_from, + "group_policy": resolved.group_policy, + "require_mention": resolved.require_mention, + "text_chunk_limit": resolved.text_chunk_limit, + "media_max_mb": resolved.media_max_mb, + "enabled": resolved.enabled, + "config": resolved.config, + "resolved": resolved, + } + + def _build_resolved(self, account_id: str, merged: dict) -> ResolvedRingCentralAccount: + client_id = merged.get("client_id", "") or os.getenv("RINGCENTRAL_CLIENT_ID", "") + client_secret = merged.get("client_secret", "") or os.getenv("RINGCENTRAL_CLIENT_SECRET", "") + jwt_token = merged.get("jwt_token", "") or os.getenv("RINGCENTRAL_JWT_TOKEN", "") + + return ResolvedRingCentralAccount( + account_id=account_id, + name=merged.get("name", account_id), + client_id=client_id, + client_secret=client_secret, + server_url=merged.get("server_url", "https://platform.ringcentral.com"), + jwt_token=jwt_token, + webhook_path=merged.get("webhook_path", _DEFAULT_WEBHOOK_PATH), + webhook_validation_token=merged.get("webhook_validation_token", ""), + bot_person_id=merged.get("bot_person_id", ""), + subscription_id=merged.get("subscription_id", ""), + dm_policy=merged.get("dm_policy", "pairing"), + dm_enabled=merged.get("dm_enabled", True), + dm_allow_from=merged.get("dm_allow_from", []), + group_policy=merged.get("group_policy", "allowlist"), + require_mention=merged.get("require_mention", True), + text_chunk_limit=merged.get("text_chunk_limit", 1000), + media_max_mb=merged.get("media_max_mb", 20), + enabled=merged.get("enabled", True), + config=merged, + ) + + def is_configured(self, account: dict) -> bool: + resolved = account.get("resolved") + if isinstance(resolved, ResolvedRingCentralAccount): + return resolved.is_configured + return bool(account.get("client_id") and account.get("jwt_token")) + + def is_enabled(self, account: dict) -> bool: + return bool(account.get("enabled", True)) + + def disabled_reason(self, account: dict) -> str: + if not self.is_configured(account): + return "RingCentral client_id, client_secret and jwt_token are required" + return "" + + async def resolve_allow_from(self, config: dict, account_id: str) -> list[str] | None: + account = await self.resolve_account(account_id) + return account.get("dm_allow_from", []) + + def describe_account(self, account: dict) -> dict: + return { + "account_id": account.get("account_id", ""), + "name": account.get("name", ""), + "configured": self.is_configured(account), + } + + def config_schema(self) -> dict: + return { + "type": "object", + "properties": { + "enabled": {"type": "boolean", "default": True}, + "name": {"type": "string"}, + "clientId": {"type": "string"}, + "clientSecret": {"type": "string"}, + "serverUrl": { + "type": "string", + "default": "https://platform.ringcentral.com", + }, + "jwtToken": {"type": "string"}, + "webhookPath": {"type": "string", "default": _DEFAULT_WEBHOOK_PATH}, + "webhookValidationToken": {"type": "string"}, + "botPersonId": {"type": "string"}, + "subscriptionId": {"type": "string"}, + "dmPolicy": { + "type": "string", + "enum": ["open", "pairing", "allowlist", "disabled"], + "default": "pairing", + }, + "dmEnabled": {"type": "boolean", "default": True}, + "dmAllowFrom": {"type": "array", "items": {"type": "string"}}, + "groupPolicy": { + "type": "string", + "enum": ["open", "allowlist", "disabled"], + "default": "allowlist", + }, + "requireMention": {"type": "boolean", "default": True}, + "textChunkLimit": {"type": "integer", "default": 1000}, + "mediaMaxMb": {"type": "integer", "default": 20}, + "accounts": { + "type": "object", + "additionalProperties": { + "type": "object", + "properties": { + "name": {"type": "string"}, + "clientId": {"type": "string"}, + "clientSecret": {"type": "string"}, + "serverUrl": {"type": "string"}, + "jwtToken": {"type": "string"}, + "webhookPath": {"type": "string"}, + "webhookValidationToken": {"type": "string"}, + "botPersonId": {"type": "string"}, + "subscriptionId": {"type": "string"}, + "dmPolicy": {"type": "string"}, + "dmEnabled": {"type": "boolean"}, + "dmAllowFrom": {"type": "array", "items": {"type": "string"}}, + "groupPolicy": {"type": "string"}, + "requireMention": {"type": "boolean"}, + }, + }, + }, + }, + } diff --git a/backend/package/yuxi/channel/extensions/ringcentral/dedupe.py b/backend/package/yuxi/channel/extensions/ringcentral/dedupe.py new file mode 100644 index 00000000..836e1a73 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/ringcentral/dedupe.py @@ -0,0 +1,36 @@ +from __future__ import annotations + +import time + + +class RingCentralDeduplicator: + def __init__(self, ttl_seconds: int = 300, max_entries: int = 10000): + self._seen: dict[str, float] = {} + self.ttl_seconds = ttl_seconds + self.max_entries = max_entries + + def is_duplicate(self, key: str) -> bool: + self._evict_expired() + return key in self._seen + + def mark_seen(self, key: str) -> None: + self._seen[key] = time.time() + if len(self._seen) > self.max_entries: + self._evict_oldest() + + def reset(self) -> None: + self._seen.clear() + + def _evict_expired(self) -> None: + now = time.time() + expired = [k for k, t in self._seen.items() if now - t > self.ttl_seconds] + for k in expired: + del self._seen[k] + + def _evict_oldest(self) -> None: + excess = len(self._seen) - self.max_entries + if excess <= 0: + return + sorted_keys = sorted(self._seen, key=lambda k: self._seen[k]) + for k in sorted_keys[:excess]: + del self._seen[k] diff --git a/backend/package/yuxi/channel/extensions/ringcentral/errors.py b/backend/package/yuxi/channel/extensions/ringcentral/errors.py new file mode 100644 index 00000000..f53e1787 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/ringcentral/errors.py @@ -0,0 +1,34 @@ +from __future__ import annotations + + +class RingCentralError(Exception): + def __init__(self, status_code: int, message: str): + self.status_code = status_code + self.message = message + super().__init__(f"[{status_code}] {message}") + + +class RingCentralAuthError(RingCentralError): + pass + + +class RingCentralRateLimitError(RingCentralError): + def __init__(self, status_code: int, message: str, retry_after: int = 60): + super().__init__(status_code, message) + self.retry_after = retry_after + + +class RingCentralNotFoundError(RingCentralError): + pass + + +class RingCentralValidationError(RingCentralError): + pass + + +class RingCentralPermissionError(RingCentralError): + pass + + +def is_retryable_error(status_code: int) -> bool: + return status_code == 429 or status_code >= 500 diff --git a/backend/package/yuxi/channel/extensions/ringcentral/events.py b/backend/package/yuxi/channel/extensions/ringcentral/events.py new file mode 100644 index 00000000..5290583c --- /dev/null +++ b/backend/package/yuxi/channel/extensions/ringcentral/events.py @@ -0,0 +1,83 @@ +from __future__ import annotations + +import logging + +from yuxi.channel.extensions.ringcentral.sdk import AsyncRingCentralClient + +logger = logging.getLogger(__name__) + + +async def list_events( + client: AsyncRingCentralClient, + group_id: str, +) -> dict: + return await client.get(f"/restapi/v1.0/glip/groups/{group_id}/events") + + +async def create_event( + client: AsyncRingCentralClient, + group_id: str, + title: str, + start_time: str, + end_time: str, + *, + description: str = "", + location: str = "", + all_day: bool = False, + recurrence: str | None = None, +) -> dict: + payload = { + "title": title, + "startTime": start_time, + "endTime": end_time, + } + if description: + payload["description"] = description + if location: + payload["location"] = location + if all_day: + payload["allDay"] = all_day + if recurrence: + payload["recurrence"] = recurrence + return await client.post( + f"/restapi/v1.0/glip/groups/{group_id}/events", + body=payload, + ) + + +async def get_event( + client: AsyncRingCentralClient, + event_id: str, +) -> dict: + return await client.get(f"/restapi/v1.0/glip/events/{event_id}") + + +async def update_event( + client: AsyncRingCentralClient, + event_id: str, + *, + title: str | None = None, + start_time: str | None = None, + end_time: str | None = None, + description: str | None = None, + location: str | None = None, +) -> dict: + payload = {} + if title is not None: + payload["title"] = title + if start_time is not None: + payload["startTime"] = start_time + if end_time is not None: + payload["endTime"] = end_time + if description is not None: + payload["description"] = description + if location is not None: + payload["location"] = location + return await client.put( + f"/restapi/v1.0/glip/events/{event_id}", + body=payload, + ) + + +async def delete_event(client: AsyncRingCentralClient, event_id: str) -> None: + await client.delete(f"/restapi/v1.0/glip/events/{event_id}") diff --git a/backend/package/yuxi/channel/extensions/ringcentral/formatting.py b/backend/package/yuxi/channel/extensions/ringcentral/formatting.py new file mode 100644 index 00000000..8ba0c5d4 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/ringcentral/formatting.py @@ -0,0 +1,34 @@ +from __future__ import annotations + +import re + +_CONTROL_CHAR_PATTERN = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]") +_MULTI_NEWLINE_PATTERN = re.compile(r"\n{3,}") + + +def sanitize_text(text: str) -> str: + cleaned = _CONTROL_CHAR_PATTERN.sub("", text) + cleaned = _MULTI_NEWLINE_PATTERN.sub("\n\n", cleaned) + return cleaned.strip() + + +def chunk_text(text: str, limit: int = 1000) -> list[str]: + if len(text) <= limit: + return [text] + + chunks = [] + remaining = text + while remaining: + if len(remaining) <= limit: + chunks.append(remaining) + break + + split_at = remaining.rfind("\n", 0, limit) + if split_at == -1 or split_at < limit // 2: + split_at = remaining.rfind(" ", 0, limit) + if split_at == -1: + split_at = limit + + chunks.append(remaining[:split_at]) + remaining = remaining[split_at:].lstrip() + return chunks diff --git a/backend/package/yuxi/channel/extensions/ringcentral/gateway.py b/backend/package/yuxi/channel/extensions/ringcentral/gateway.py new file mode 100644 index 00000000..434b10fb --- /dev/null +++ b/backend/package/yuxi/channel/extensions/ringcentral/gateway.py @@ -0,0 +1,160 @@ +from __future__ import annotations + +import asyncio +import logging +import time + +from yuxi.channel.extensions.ringcentral.config import RingCentralConfigAdapter +from yuxi.channel.extensions.ringcentral.errors import RingCentralAuthError +from yuxi.channel.extensions.ringcentral.sdk import AsyncRingCentralClient +from yuxi.channel.extensions.ringcentral.subscription import ( + create_subscription, + delete_subscription, + renew_subscription, +) +from yuxi.channel.extensions.ringcentral.types import ResolvedRingCentralAccount +from yuxi.channel.extensions.ringcentral.webhook import RingCentralWebhookHandler +from yuxi.channel.gateway.routes import webhook_registry + +logger = logging.getLogger(__name__) + +SUBSCRIPTION_RENEW_BEFORE_SEC = 3600 +RENEW_INTERVAL = 86400 - SUBSCRIPTION_RENEW_BEFORE_SEC + + +class RingCentralGatewayAdapter: + def __init__(self): + self._config_adapter = RingCentralConfigAdapter() + self._webhook_handler = RingCentralWebhookHandler() + self._clients: dict[str, AsyncRingCentralClient] = {} + self._accounts: dict[str, ResolvedRingCentralAccount] = {} + self._tasks: dict[str, asyncio.Task] = {} + self._abort_events: dict[str, asyncio.Event] = {} + self._running: dict[str, bool] = {} + self._status_snapshots: dict[str, dict] = {} + + async def start(self, ctx) -> object: + account_dict = getattr(ctx, "account", {}) or {} + account = account_dict.get("resolved") + if not isinstance(account, ResolvedRingCentralAccount): + account_id = account_dict.get("account_id", "default") + resolved_dict = await self._config_adapter.resolve_account(account_id) + account = resolved_dict.get("resolved") + + if not account or not account.is_configured: + logger.warning("RingCentral account not configured: %s", account.account_id if account else "unknown") + return asyncio.Queue() + + account_id = account.account_id + + client = AsyncRingCentralClient( + client_id=account.client_id, + client_secret=account.client_secret, + server_url=account.server_url, + ) + + try: + await client.login_jwt(account.jwt_token) + except RingCentralAuthError as e: + logger.error("RingCentral auth failed for %s: %s", account_id, e) + return asyncio.Queue() + + self._clients[account_id] = client + self._accounts[account_id] = account + + self._webhook_handler.register_target(account.webhook_path, account) + + webhook_registry.register( + "ringcentral", + self._webhook_handler.handle_webhook, + guard_config=None, + ) + + base_url = getattr(ctx, "base_url", "") if ctx else "" + if base_url and not account.subscription_id: + webhook_url = f"{base_url.rstrip('/')}{account.webhook_path}" + try: + sub_result = await create_subscription(client, account, webhook_url) + account.subscription_id = sub_result.get("id", "") + except Exception: + logger.warning("Failed to create RingCentral subscription for %s", account_id) + + abort_event = asyncio.Event() + self._abort_events[account_id] = abort_event + + task = asyncio.create_task(self._subscription_renew_loop(account_id, account, abort_event)) + self._tasks[account_id] = task + + self._running[account_id] = True + self._status_snapshots[account_id] = { + "running": True, + "webhook_path": account.webhook_path, + "server_url": account.server_url, + "subscription_id": account.subscription_id, + "start_time": time.time(), + } + + logger.info("RingCentral gateway started for account: %s, webhook: %s", account_id, account.webhook_path) + + queue = getattr(ctx, "queue", asyncio.Queue()) + return queue + + async def stop(self, ctx) -> None: + account_dict = getattr(ctx, "account", {}) or {} + account_id = account_dict.get("account_id", "default") + + account = account_dict.get("resolved") + if isinstance(account, ResolvedRingCentralAccount): + self._webhook_handler.unregister_target(account.webhook_path) + + abort = self._abort_events.pop(account_id, None) + if abort: + abort.set() + + task = self._tasks.pop(account_id, None) + if task: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + client = self._clients.pop(account_id, None) + stored_account = self._accounts.pop(account_id, None) + + if client and stored_account and stored_account.subscription_id: + try: + await delete_subscription(client, stored_account.subscription_id) + except Exception: + logger.warning("Failed to delete RingCentral subscription for %s", account_id) + + self._running.pop(account_id, None) + self._status_snapshots[account_id] = {"running": False, "stop_time": time.time()} + + logger.info("RingCentral gateway stopped for account: %s", account_id) + + async def _subscription_renew_loop( + self, + account_id: str, + account: ResolvedRingCentralAccount, + abort_event: asyncio.Event, + ) -> None: + client = self._clients.get(account_id) + if not client: + return + + while not abort_event.is_set(): + try: + await asyncio.sleep(RENEW_INTERVAL) + if abort_event.is_set(): + break + if account.subscription_id: + await renew_subscription(client, account.subscription_id) + except Exception as e: + logger.error("RingCentral subscription renew failed for %s: %s", account_id, e) + + def get_client(self, account_id: str = "default") -> AsyncRingCentralClient | None: + return self._clients.get(account_id) + + def get_status(self, account_id: str) -> dict: + return self._status_snapshots.get(account_id, {}) diff --git a/backend/package/yuxi/channel/extensions/ringcentral/media.py b/backend/package/yuxi/channel/extensions/ringcentral/media.py new file mode 100644 index 00000000..0f913678 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/ringcentral/media.py @@ -0,0 +1,91 @@ +from __future__ import annotations + +import asyncio +import logging + +import httpx + +from yuxi.channel.extensions.ringcentral.sdk import AsyncRingCentralClient + +logger = logging.getLogger(__name__) + + +async def upload_file( + client: AsyncRingCentralClient, + group_id: str, + file_data: bytes, + filename: str, + content_type: str = "application/octet-stream", +) -> dict: + builder = client.platform.create_multipart_builder() + builder.set_body({"groupId": group_id}) + builder.add_file(None, content=file_data, content_type=content_type, file_name=filename) + + def _send(): + request = builder.request("/restapi/v1.0/glip/files") + return client.platform.send_request(request) + + resp = await asyncio.to_thread(_send) + return resp.json() + + +async def download_file_content( + client: AsyncRingCentralClient, + file_id: str, +) -> bytes: + def _download(): + resp = client.platform.get(f"/restapi/v1.0/glip/files/{file_id}/content") + return resp.response.content + + return await asyncio.to_thread(_download) + + +async def send_media_message( + client: AsyncRingCentralClient, + group_id: str, + text: str, + file_info: dict, +) -> dict: + return await client.post( + f"/restapi/v1.0/glip/chats/{group_id}/posts", + body={ + "text": text, + "attachments": [ + { + "type": "File", + "id": file_info.get("id"), + "name": file_info.get("name", ""), + "contentUri": file_info.get("contentUri", ""), + } + ], + }, + ) + + +async def send_media_from_url( + client: AsyncRingCentralClient, + group_id: str, + media_url: str, + media_type: str, +) -> None: + async with httpx.AsyncClient(timeout=httpx.Timeout(60.0)) as http: + resp = await http.get(media_url) + resp.raise_for_status() + content = resp.content + + filename = _extract_filename(media_url, media_type) + upload_result = await upload_file(client, group_id, content, filename, media_type) + await send_media_message(client, group_id, "", upload_result) + + +def _extract_filename(url: str, content_type: str) -> str: + ext_map = { + "image/jpeg": ".jpg", + "image/png": ".png", + "image/gif": ".gif", + "image/webp": ".webp", + "video/mp4": ".mp4", + "application/pdf": ".pdf", + } + ext = ext_map.get(content_type, ".bin") + return f"attachment{ext}" diff --git a/backend/package/yuxi/channel/extensions/ringcentral/mentions.py b/backend/package/yuxi/channel/extensions/ringcentral/mentions.py new file mode 100644 index 00000000..c27506b0 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/ringcentral/mentions.py @@ -0,0 +1,13 @@ +from __future__ import annotations + +import re + +MENTION_PATTERN = re.compile(r"@!\[(.*?)\]\((.*?)\)") + + +def extract_mentions(text: str) -> list[tuple[str, str]]: + return MENTION_PATTERN.findall(text) + + +def strip_mentions(text: str) -> str: + return MENTION_PATTERN.sub("", text).strip() diff --git a/backend/package/yuxi/channel/extensions/ringcentral/monitor.py b/backend/package/yuxi/channel/extensions/ringcentral/monitor.py new file mode 100644 index 00000000..4f4a804f --- /dev/null +++ b/backend/package/yuxi/channel/extensions/ringcentral/monitor.py @@ -0,0 +1,148 @@ +from __future__ import annotations + +import asyncio +import logging +from datetime import UTC, datetime + +from yuxi.channel.extensions.ringcentral.formatting import sanitize_text +from yuxi.channel.extensions.ringcentral.mentions import strip_mentions +from yuxi.channel.extensions.ringcentral.types import RingCentralEvent, ResolvedRingCentralAccount +from yuxi.channel.message.models import ( + GroupContext, + MessageType, + PeerInfo, + UnifiedMessage, +) +from yuxi.channel.routing.models import PeerKind + +logger = logging.getLogger(__name__) + +_PERSON_FETCH_TIMEOUT = 2.0 + + +def _infer_attachment_msg_type(attachment: dict) -> MessageType: + mime = attachment.get("contentType", "").lower() + if mime.startswith("image/"): + return MessageType.IMAGE + if mime.startswith("audio/"): + return MessageType.VOICE + return MessageType.FILE + + +def _build_attachment_text(attachments: list[dict]) -> str: + names = [a.get("name", "unknown") for a in attachments] + return f"[{', '.join(names)}]" + + +class RingCentralMonitor: + async def event_to_unified_message( + self, + raw_payload: dict, + event: RingCentralEvent, + account: ResolvedRingCentralAccount | None = None, + ) -> UnifiedMessage | None: + post = event.body + if not post: + return None + + text = strip_mentions(sanitize_text(post.text or "")) + attachments = post.attachments or [] + + if not text and not attachments: + return None + + if attachments and not text: + msg_type = _infer_attachment_msg_type(attachments[0]) + display_text = _build_attachment_text(attachments) + else: + msg_type = MessageType.TEXT + display_text = text + + chat_type = _infer_chat_type(post.group_id, raw_payload) + is_dm = chat_type == "direct" + + sender_kind = PeerKind.DIRECT if is_dm else PeerKind.GROUP + display_name = None + avatar_url = None + + if post.creator_id: + try: + person = await self._fetch_person_safe(post.creator_id, account) + if person: + display_name = ( + person.get("firstName", "") + " " + person.get("lastName", "") + ).strip() or person.get("name") + avatar_url = person.get("avatar") + except Exception: + pass + + sender_info = PeerInfo( + kind=sender_kind, + id=post.creator_id, + display_name=display_name, + avatar_url=avatar_url, + is_bot=False, + ) + + group = None + if not is_dm: + group = GroupContext( + id=post.group_id, + kind=chat_type, + ) + + metadata = { + "ChatType": chat_type, + "WasMentioned": "@!" in (post.text or ""), + "event_type": event.event, + "subscription_id": event.subscription_id, + } + + return UnifiedMessage( + msg_id=post.id or f"rc:{int(datetime.now(tz=UTC).timestamp())}", + channel_type="ringcentral", + account_id=account.account_id if account else "default", + content=display_text, + sender=sender_info, + message_type=msg_type, + group=group, + raw_payload=raw_payload, + metadata=metadata, + was_mentioned="@!" in (post.text or ""), + ) + + async def _fetch_person_safe( + self, + person_id: str, + account: ResolvedRingCentralAccount | None, + ) -> dict | None: + try: + from yuxi.channel.plugins.registry import ChannelPluginRegistry + + plugin = ChannelPluginRegistry.get("ringcentral") + if plugin and hasattr(plugin, "_outbound"): + return await asyncio.wait_for( + plugin._outbound.fetch_person( + person_id, + account_id=account.account_id if account else None, + ), + timeout=_PERSON_FETCH_TIMEOUT, + ) + except Exception: + pass + return None + + +def _infer_chat_type(group_id: str, raw_payload: dict) -> str: + body = raw_payload.get("body", raw_payload) + mentions = body.get("mentions", {}) + if isinstance(mentions, dict): + for mid, mdata in mentions.items(): + mtype = mdata.get("type", "") if isinstance(mdata, dict) else "" + if mtype == "Team": + return "team" + if mtype == "Group": + return "group" + if mtype == "PersonalChat": + return "direct" + return "direct" diff --git a/backend/package/yuxi/channel/extensions/ringcentral/notes.py b/backend/package/yuxi/channel/extensions/ringcentral/notes.py new file mode 100644 index 00000000..810335f6 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/ringcentral/notes.py @@ -0,0 +1,55 @@ +from __future__ import annotations + +import logging + +from yuxi.channel.extensions.ringcentral.sdk import AsyncRingCentralClient + +logger = logging.getLogger(__name__) + + +async def list_notes( + client: AsyncRingCentralClient, + chat_id: str, +) -> dict: + return await client.get(f"/restapi/v1.0/glip/chats/{chat_id}/notes") + + +async def create_note( + client: AsyncRingCentralClient, + chat_id: str, + title: str, + body: str, +) -> dict: + return await client.post( + f"/restapi/v1.0/glip/chats/{chat_id}/notes", + body={"title": title, "body": body}, + ) + + +async def get_note( + client: AsyncRingCentralClient, + note_id: str, +) -> dict: + return await client.get(f"/restapi/v1.0/glip/notes/{note_id}") + + +async def update_note( + client: AsyncRingCentralClient, + note_id: str, + *, + title: str | None = None, + body: str | None = None, +) -> dict: + payload = {} + if title is not None: + payload["title"] = title + if body is not None: + payload["body"] = body + return await client.patch( + f"/restapi/v1.0/glip/notes/{note_id}", + body=payload, + ) + + +async def delete_note(client: AsyncRingCentralClient, note_id: str) -> None: + await client.delete(f"/restapi/v1.0/glip/notes/{note_id}") diff --git a/backend/package/yuxi/channel/extensions/ringcentral/outbound.py b/backend/package/yuxi/channel/extensions/ringcentral/outbound.py new file mode 100644 index 00000000..94099dd4 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/ringcentral/outbound.py @@ -0,0 +1,324 @@ +from __future__ import annotations + +import asyncio +import logging + +from yuxi.channel.extensions.ringcentral.errors import ( + RingCentralRateLimitError, + is_retryable_error, +) +from yuxi.channel.extensions.ringcentral.formatting import chunk_text, sanitize_text +from yuxi.channel.extensions.ringcentral.sdk import AsyncRingCentralClient +from yuxi.channel.extensions.ringcentral.types import ResolvedRingCentralAccount +from yuxi.channel.protocols import ( + OutboundDeliveryCapabilities, + OutboundDeliveryMode, + OutboundPresentationCapabilities, +) + +logger = logging.getLogger(__name__) + + +class RingCentralOutboundAdapter: + delivery_mode = OutboundDeliveryMode.DIRECT + chunker_mode = "markdown" + text_chunk_limit: int = 1000 + max_retries: int = 3 + retry_base_delay: float = 1.0 + presentation_capabilities = OutboundPresentationCapabilities(supported=False) + delivery_capabilities = OutboundDeliveryCapabilities( + durable_final_text=True, + durable_final_media=False, + ) + + def _get_account(self, account_id: str | None) -> ResolvedRingCentralAccount | None: + if account_id: + from yuxi.channel.plugins.registry import ChannelPluginRegistry + + plugin = ChannelPluginRegistry.get("ringcentral") + if plugin and hasattr(plugin, "_config"): + import asyncio + + loop = asyncio.get_event_loop() + account_dict = loop.run_until_complete(plugin._config.resolve_account(account_id)) + return account_dict.get("resolved") + return None + + def _get_client(self, account_id: str | None = None) -> AsyncRingCentralClient | None: + if account_id: + from yuxi.channel.plugins.registry import ChannelPluginRegistry + + plugin = ChannelPluginRegistry.get("ringcentral") + if plugin and hasattr(plugin, "_gateway"): + return plugin._gateway.get_client(account_id) + return None + + def _require_client(self, account_id: str | None = None) -> AsyncRingCentralClient: + client = self._get_client(account_id) + if not client: + raise RuntimeError(f"RingCentral client not found for account: {account_id}") + return client + + async def send_text( + self, + target_id: str, + content: str, + *, + reply_to_id: str | None = None, + thread_id: str | None = None, + account_id: str | None = None, + ) -> None: + client = self._get_client(account_id) + if not client: + raise RuntimeError(f"RingCentral client not found for account: {account_id}") + + safe = sanitize_text(content) + if not safe: + return + + account = self._get_account(account_id) + limit = account.text_chunk_limit if account else self.text_chunk_limit + + chunks = chunk_text(safe, limit) + for chunk in chunks: + await self._send_post_with_retry(client, target_id, chunk) + + async def _send_post_with_retry( + self, + client: AsyncRingCentralClient, + chat_id: str, + text: str, + ) -> str | None: + last_error = None + for attempt in range(self.max_retries): + try: + result = await client.post( + f"/restapi/v1.0/glip/chats/{chat_id}/posts", + body={"text": text}, + ) + return result.get("id", "") + except RingCentralRateLimitError as e: + delay = max(e.retry_after, self.retry_base_delay * (2**attempt)) + logger.warning("RingCentral rate limited, retrying in %ss (attempt %d)", delay, attempt + 1) + await asyncio.sleep(delay) + last_error = e + except Exception as e: + if not is_retryable_error(getattr(e, "status_code", 500)): + raise + delay = self.retry_base_delay * (2**attempt) + await asyncio.sleep(delay) + last_error = e + if last_error: + raise last_error + return None + + async def send_media( + self, + target_id: str, + media_url: str, + media_type: str, + reply_to_id: str | None = None, + thread_id: str | None = None, + account_id: str | None = None, + ) -> None: + client = self._get_client(account_id) + if not client: + raise RuntimeError(f"RingCentral client not found for account: {account_id}") + + from yuxi.channel.extensions.ringcentral.media import send_media_from_url + + await send_media_from_url(client, target_id, media_url, media_type) + + async def edit_message( + self, + target_id: str, + message_id: str, + content: str, + *, + thread_id: str | None = None, + account_id: str | None = None, + ) -> str | None: + client = self._get_client(account_id) + if not client: + return None + + safe = sanitize_text(content) + result = await client.patch( + f"/restapi/v1.0/glip/chats/{target_id}/posts/{message_id}", + body={"text": safe}, + ) + return result.get("id") if result else None + + async def delete_message( + self, + chat_id: str, + post_id: str, + *, + account_id: str | None = None, + ) -> None: + client = self._get_client(account_id) + if not client: + return + + await client.delete(f"/restapi/v1.0/glip/chats/{chat_id}/posts/{post_id}") + + async def send_reaction( + self, + post_id: str, + emoji: str, + *, + account_id: str | None = None, + ) -> dict | None: + client = self._get_client(account_id) + if not client: + return None + + return await client.post( + f"/restapi/v1.0/glip/posts/{post_id}/reactions", + body={"reaction": emoji}, + ) + + async def remove_reaction( + self, + post_id: str, + emoji: str, + *, + account_id: str | None = None, + ) -> None: + client = self._get_client(account_id) + if not client: + return + + await client.delete( + f"/restapi/v1.0/glip/posts/{post_id}/reactions/{emoji}", + ) + + def chunker(self, text: str, limit: int, ctx: object | None = None) -> list[str]: + return chunk_text(text, limit) + + async def fetch_posts( + self, + chat_id: str, + *, + page: int = 1, + per_page: int = 20, + account_id: str | None = None, + ) -> dict: + client = self._require_client(account_id) + return await client.get( + f"/restapi/v1.0/glip/chats/{chat_id}/posts", + params={"page": page, "perPage": per_page}, + ) + + async def get_post( + self, + chat_id: str, + post_id: str, + *, + account_id: str | None = None, + ) -> dict: + client = self._require_client(account_id) + return await client.get( + f"/restapi/v1.0/glip/chats/{chat_id}/posts/{post_id}", + ) + + async def fetch_person( + self, + person_id: str, + *, + account_id: str | None = None, + ) -> dict: + client = self._require_client(account_id) + return await client.get(f"/restapi/v1.0/glip/persons/{person_id}") + + async def fetch_me( + self, + *, + account_id: str | None = None, + ) -> dict: + return await self.fetch_person("~", account_id=account_id) + + async def mark_chat_read( + self, + chat_id: str, + *, + account_id: str | None = None, + ) -> None: + client = self._require_client(account_id) + await client.post(f"/restapi/v1.0/glip/chats/{chat_id}/read") + + async def favorite_post( + self, + post_id: str, + *, + account_id: str | None = None, + ) -> None: + client = self._require_client(account_id) + await client.post(f"/restapi/v1.0/glip/posts/{post_id}/favorite") + + async def unfavorite_post( + self, + post_id: str, + *, + account_id: str | None = None, + ) -> None: + client = self._require_client(account_id) + await client.delete(f"/restapi/v1.0/glip/posts/{post_id}/favorite") + + async def list_chats( + self, + *, + page: int = 1, + per_page: int = 20, + account_id: str | None = None, + ) -> dict: + client = self._require_client(account_id) + return await client.get( + "/restapi/v1.0/glip/chats", + params={"page": page, "perPage": per_page}, + ) + + async def get_chat( + self, + chat_id: str, + *, + account_id: str | None = None, + ) -> dict: + client = self._require_client(account_id) + return await client.get(f"/restapi/v1.0/glip/chats/{chat_id}") + + async def list_groups( + self, + *, + page: int = 1, + per_page: int = 20, + account_id: str | None = None, + ) -> dict: + client = self._require_client(account_id) + return await client.get( + "/restapi/v1.0/glip/groups", + params={"page": page, "perPage": per_page}, + ) + + async def get_group( + self, + group_id: str, + *, + account_id: str | None = None, + ) -> dict: + client = self._require_client(account_id) + return await client.get(f"/restapi/v1.0/glip/groups/{group_id}") + + def sanitize_text_out(self, text: str, payload: object) -> str: + return sanitize_text(text) + + def resolve_effective_text_chunk_limit( + self, + config: dict, + account_id: str | None = None, + fallback_limit: int | None = None, + ) -> int | None: + return self.text_chunk_limit + + def should_treat_delivered_text_as_visible(self, kind: str, text: str | None = None) -> bool: + return True diff --git a/backend/package/yuxi/channel/extensions/ringcentral/pairing.py b/backend/package/yuxi/channel/extensions/ringcentral/pairing.py new file mode 100644 index 00000000..3201e2f9 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/ringcentral/pairing.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +import logging +import secrets +import time + +logger = logging.getLogger(__name__) + +_CODE_TTL_SEC = 300 +_CODE_LENGTH = 6 + +_pairing_store: dict[str, dict] = {} + + +class RingCentralPairingAdapter: + id_label = "RingCentral User ID" + + async def generate_code(self, peer_id: str) -> str: + code = secrets.token_hex(_CODE_LENGTH // 2)[:_CODE_LENGTH].upper() + _pairing_store[peer_id] = { + "code": code, + "status": "pending", + "created_at": time.time(), + } + return code + + async def verify_code(self, peer_id: str, code: str) -> bool: + stored = _pairing_store.get(peer_id) + if not stored: + return False + if time.time() - stored.get("created_at", 0) > _CODE_TTL_SEC: + del _pairing_store[peer_id] + return False + if stored.get("code") != code: + return False + stored["status"] = "verified" + return True + + def is_paired(self, peer_id: str) -> bool: + stored = _pairing_store.get(peer_id) + return stored is not None and stored.get("status") == "verified" + + def cleanup_expired(self) -> None: + now = time.time() + expired = [k for k, v in _pairing_store.items() if now - v.get("created_at", 0) > _CODE_TTL_SEC] + for k in expired: + del _pairing_store[k] diff --git a/backend/package/yuxi/channel/extensions/ringcentral/plugin.json b/backend/package/yuxi/channel/extensions/ringcentral/plugin.json new file mode 100644 index 00000000..aa04f37b --- /dev/null +++ b/backend/package/yuxi/channel/extensions/ringcentral/plugin.json @@ -0,0 +1,34 @@ +{ + "id": "ringcentral", + "name": "RingCentral", + "version": "0.1.0", + "label": "RingCentral (Team Messaging)", + "aliases": ["rc", "glip", "ringcentral-glip"], + "description": "RingCentral channel plugin for ForcePilot — Team Messaging, Glip API, and RingCentral Bot integration", + "author": "ForcePilot", + "order": 80, + "dependencies": ["ringcentral"], + "capabilities": { + "chat_types": ["direct", "group", "team"], + "message_types": ["text", "image", "file"], + "reactions": true, + "typing_indicator": false, + "threads": false, + "edit": true, + "unsend": true, + "reply": false, + "media": true, + "streaming": true, + "streaming_mode": "block", + "block_streaming": true, + "block_streaming_chunk_min_chars": 500, + "block_streaming_chunk_max_chars": 1000, + "block_streaming_coalesce_min_chars": 200, + "adaptive_cards": true, + "tasks": false, + "events": false, + "notes": false + }, + "enabled": true, + "python_requires": ">=3.12" +} diff --git a/backend/package/yuxi/channel/extensions/ringcentral/reactions.py b/backend/package/yuxi/channel/extensions/ringcentral/reactions.py new file mode 100644 index 00000000..1724b1bf --- /dev/null +++ b/backend/package/yuxi/channel/extensions/ringcentral/reactions.py @@ -0,0 +1,25 @@ +from __future__ import annotations + +import logging + +from yuxi.channel.extensions.ringcentral.sdk import AsyncRingCentralClient + +logger = logging.getLogger(__name__) + + +async def add_reaction(client: AsyncRingCentralClient, post_id: str, emoji: str) -> dict: + return await client.post( + f"/restapi/v1.0/glip/posts/{post_id}/reactions", + body={"reaction": emoji}, + ) + + +async def remove_reaction(client: AsyncRingCentralClient, post_id: str, emoji: str) -> None: + await client.delete( + f"/restapi/v1.0/glip/posts/{post_id}/reactions/{emoji}", + ) + + +async def list_reactions(client: AsyncRingCentralClient, post_id: str) -> list[dict]: + resp = await client.get(f"/restapi/v1.0/glip/posts/{post_id}/reactions") + return resp.get("records", []) diff --git a/backend/package/yuxi/channel/extensions/ringcentral/sdk.py b/backend/package/yuxi/channel/extensions/ringcentral/sdk.py new file mode 100644 index 00000000..f04edbb8 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/ringcentral/sdk.py @@ -0,0 +1,128 @@ +from __future__ import annotations + +import asyncio +import logging + +from ringcentral import SDK +from ringcentral.http.api_exception import ApiException + +from yuxi.channel.extensions.ringcentral.errors import ( + RingCentralAuthError, + RingCentralError, + RingCentralNotFoundError, + RingCentralPermissionError, + RingCentralRateLimitError, + RingCentralValidationError, +) + +logger = logging.getLogger(__name__) + + +class AsyncRingCentralClient: + def __init__(self, client_id: str, client_secret: str, server_url: str): + self._sdk = SDK(client_id, client_secret, server_url) + self._platform = self._sdk.platform() + self.client_id = client_id + self.server_url = server_url + + async def login_jwt(self, jwt_token: str) -> dict: + def _login(): + return self._platform.login(jwt=jwt_token) + + try: + resp = await asyncio.to_thread(_login) + data = resp.json() + logger.info("RingCentral JWT login succeeded, owner_id=%s", data.get("owner_id", "")) + return data + except ApiException as e: + raise RingCentralAuthError(getattr(e, "status_code", 401), str(e)) from e + + async def ensure_authenticated(self, jwt_token: str) -> None: + auth = self._platform.auth() + if auth and auth.access_token_valid(): + return + logger.info("RingCentral access token expired, re-authenticating...") + await self.login_jwt(jwt_token) + + @property + def is_logged_in(self) -> bool: + return self._platform.logged_in() + + @property + def platform(self): + return self._platform + + async def _call( + self, + method: str, + path: str, + body: dict | None = None, + params: dict | None = None, + ) -> dict: + def _do(): + if method == "GET": + return self._platform.get(path, params or {}) + elif method == "POST": + return self._platform.post(path, body) + elif method == "PATCH": + return self._platform.patch(path, body) + elif method == "PUT": + return self._platform.put(path, body) + elif method == "DELETE": + return self._platform.delete(path) + raise ValueError(f"Unknown method: {method}") + + try: + resp = await asyncio.to_thread(_do) + if resp.ok(): + return resp.json() + raise self._parse_error(resp) + except ApiException as e: + raise self._parse_api_exception(e) + + def _parse_error(self, response) -> RingCentralError: + status = response.status + try: + body = response.json() + msg = body.get("message", response.text) + except Exception: + msg = response.text + + if status == 429: + retry_after = response.headers.get("Retry-After", "60") + return RingCentralRateLimitError(status, msg, retry_after=int(retry_after)) + if status == 404: + return RingCentralNotFoundError(status, msg) + if status in (400, 422): + return RingCentralValidationError(status, msg) + if status == 403: + return RingCentralPermissionError(status, msg) + if status in (401,): + return RingCentralAuthError(status, msg) + return RingCentralError(status, msg) + + def _parse_api_exception(self, e: ApiException) -> RingCentralError: + msg = str(e) + if "rate limit" in msg.lower() or "429" in msg: + return RingCentralRateLimitError(429, msg, retry_after=60) + if "unauthorized" in msg.lower() or "token" in msg.lower(): + return RingCentralAuthError(401, msg) + return RingCentralError(500, msg) + + async def get(self, path: str, params: dict | None = None) -> dict: + return await self._call("GET", path, params=params) + + async def post(self, path: str, body: dict | None = None) -> dict: + return await self._call("POST", path, body=body) + + async def patch(self, path: str, body: dict | None = None) -> dict: + return await self._call("PATCH", path, body=body) + + async def put(self, path: str, body: dict | None = None) -> dict: + return await self._call("PUT", path, body=body) + + async def delete(self, path: str) -> dict: + return await self._call("DELETE", path) + + async def close(self) -> None: + pass diff --git a/backend/package/yuxi/channel/extensions/ringcentral/security.py b/backend/package/yuxi/channel/extensions/ringcentral/security.py new file mode 100644 index 00000000..5c7b0ae4 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/ringcentral/security.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +import logging + +logger = logging.getLogger(__name__) + + +class RingCentralSecurityAdapter: + def resolve_dm_policy(self) -> dict: + return {"mode": "pairing", "allow_from": []} + + def check_allowlist(self, sender_id: str, account) -> bool: + if not account: + return True + allow_from = account.dm_allow_from if hasattr(account, "dm_allow_from") else [] + if not allow_from: + return True + return sender_id in allow_from + + def evaluate(self, chat_type: str, sender_id: str, text: str, account) -> tuple[bool, str]: + if chat_type == "direct": + if not account: + return True, "ok" + if not getattr(account, "dm_enabled", True): + return False, "dm_disabled" + dm_policy = getattr(account, "dm_policy", "pairing") + if dm_policy == "disabled": + return False, "dm_disabled" + if dm_policy == "allowlist" and not self.check_allowlist(sender_id, account): + return False, "sender_not_in_allowlist" + if dm_policy == "pairing": + return True, "pairing_check" + + elif chat_type in ("group", "team"): + if not account: + return True, "ok" + group_policy = getattr(account, "group_policy", "allowlist") + if group_policy == "disabled": + return False, "group_disabled" + if getattr(account, "require_mention", True) and "@!" not in text: + return False, "mention_required" + + return True, "ok" + + def collect_warnings(self, config: dict, account_id: str | None = None) -> list[str]: + warnings = [] + rc_config = config.get("channels", {}).get("ringcentral", {}) + + group_policy = rc_config.get("group_policy", "allowlist") + if group_policy == "open": + warnings.append( + "Group policy is 'open' - any RingCentral group can trigger the bot. " + "Consider using 'allowlist' for production." + ) + + dm_policy = rc_config.get("dm_policy", "pairing") + if dm_policy == "open": + warnings.append( + "DM policy is 'open' - anyone can DM the bot. Consider using 'pairing' or 'allowlist' for production." + ) + + return warnings diff --git a/backend/package/yuxi/channel/extensions/ringcentral/session.py b/backend/package/yuxi/channel/extensions/ringcentral/session.py new file mode 100644 index 00000000..8355dbb5 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/ringcentral/session.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +from yuxi.channel.protocols import SessionResolution + + +class RingCentralSessionAdapter: + def resolve_session(self, msg) -> SessionResolution: + from yuxi.channel.routing.models import PeerKind + + if hasattr(msg, "sender") and hasattr(msg.sender, "kind"): + if msg.sender.kind == PeerKind.DIRECT: + chat_id = _extract_chat_id(msg) + return SessionResolution( + kind="direct", + conversation_id=chat_id or msg.sender.id, + ) + + chat_id = _extract_chat_id(msg) + if chat_id: + return SessionResolution(kind="group", conversation_id=chat_id) + + if msg.group and msg.group.id: + return SessionResolution(kind="group", conversation_id=msg.group.id) + + return SessionResolution(kind="group", conversation_id="unknown") + + @staticmethod + def build_dm_session_key(agent_id: str, account_id: str, chat_id: str) -> str: + return f"agent:{agent_id}:ringcentral:direct:{chat_id}" + + @staticmethod + def build_group_session_key(agent_id: str, account_id: str, chat_id: str) -> str: + return f"agent:{agent_id}:ringcentral:group:{chat_id}" + + +def _extract_chat_id(msg) -> str | None: + metadata = getattr(msg, "metadata", {}) or {} + if metadata.get("chat_id"): + return metadata["chat_id"] + + raw = getattr(msg, "raw_payload", {}) or {} + body = raw.get("body", raw) + group_id = body.get("groupId", "") + if group_id: + return group_id + + return None diff --git a/backend/package/yuxi/channel/extensions/ringcentral/status.py b/backend/package/yuxi/channel/extensions/ringcentral/status.py new file mode 100644 index 00000000..b1d4770d --- /dev/null +++ b/backend/package/yuxi/channel/extensions/ringcentral/status.py @@ -0,0 +1,74 @@ +from __future__ import annotations + +import logging + +from yuxi.channel.extensions.ringcentral.types import ResolvedRingCentralAccount + +logger = logging.getLogger(__name__) + + +class RingCentralStatusAdapter: + async def probe(self, account_dict: dict) -> bool: + account = account_dict.get("resolved") + if not isinstance(account, ResolvedRingCentralAccount): + return False + + from yuxi.channel.plugins.registry import ChannelPluginRegistry + + plugin = ChannelPluginRegistry.get("ringcentral") + if not plugin or not hasattr(plugin, "_gateway"): + return False + + client = plugin._gateway.get_client(account.account_id) + if not client or not client.is_logged_in: + return False + + try: + await client.get("/restapi/v1.0/account/~/extension/~") + return True + except Exception: + return False + + def build_summary(self, snapshot) -> dict: + return { + "channel": "ringcentral", + "mode": "webhook", + } + + def build_account_snapshot(self, account_dict: dict) -> dict: + return { + "account_id": account_dict.get("account_id", "default"), + "name": account_dict.get("name", ""), + "configured": bool(account_dict.get("client_id") and account_dict.get("jwt_token")), + "enabled": account_dict.get("enabled", True), + } + + def collect_status_issues(self, accounts: list[dict]) -> list[dict]: + issues = [] + for account in accounts: + if not account: + continue + + if not account.get("client_id"): + issues.append( + { + "channel": "ringcentral", + "account_id": account.get("account_id", "default"), + "kind": "client_id_missing", + "message": "client_id is not configured.", + "fix": "Set channels.ringcentral.client_id.", + } + ) + + if not account.get("jwt_token"): + issues.append( + { + "channel": "ringcentral", + "account_id": account.get("account_id", "default"), + "kind": "jwt_token_missing", + "message": "jwt_token is not configured.", + "fix": "Set channels.ringcentral.jwt_token.", + } + ) + + return issues diff --git a/backend/package/yuxi/channel/extensions/ringcentral/streaming.py b/backend/package/yuxi/channel/extensions/ringcentral/streaming.py new file mode 100644 index 00000000..eb4cc337 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/ringcentral/streaming.py @@ -0,0 +1,51 @@ +from __future__ import annotations + +import logging + +logger = logging.getLogger(__name__) + + +class RingCentralBlockStreamer: + def __init__( + self, + chunk_min: int = 500, + chunk_max: int = 1000, + coalesce_min: int = 200, + coalesce_idle_ms: int = 200, + ): + self.chunk_min = chunk_min + self.chunk_max = chunk_max + self.coalesce_min = coalesce_min + self.coalesce_idle_ms = coalesce_idle_ms + + async def stream( + self, + target_id: str, + content_stream, + outbound_adapter, + account_id: str | None = None, + ) -> list[str]: + message_ids = [] + buffer = "" + + async for chunk in content_stream: + buffer += chunk + if len(buffer) >= self.chunk_min: + to_send = buffer[: self.chunk_max] + await outbound_adapter.send_text( + target_id, + to_send, + account_id=account_id, + ) + message_ids.append(to_send) + buffer = buffer[self.chunk_max :] + + if buffer.strip(): + await outbound_adapter.send_text( + target_id, + buffer, + account_id=account_id, + ) + message_ids.append(buffer) + + return message_ids diff --git a/backend/package/yuxi/channel/extensions/ringcentral/subscription.py b/backend/package/yuxi/channel/extensions/ringcentral/subscription.py new file mode 100644 index 00000000..2e0114bf --- /dev/null +++ b/backend/package/yuxi/channel/extensions/ringcentral/subscription.py @@ -0,0 +1,58 @@ +from __future__ import annotations + +import logging + +from yuxi.channel.extensions.ringcentral.sdk import AsyncRingCentralClient +from yuxi.channel.extensions.ringcentral.types import ResolvedRingCentralAccount + +logger = logging.getLogger(__name__) + +DEFAULT_SUBSCRIPTION_EXPIRES = 86400 * 7 + + +async def create_subscription( + client: AsyncRingCentralClient, + account: ResolvedRingCentralAccount, + webhook_url: str, + expires_in: int = DEFAULT_SUBSCRIPTION_EXPIRES, +) -> dict: + payload = { + "eventFilters": [ + "/restapi/v1.0/glip/posts", + "/restapi/v1.0/glip/groups", + ], + "expiresIn": expires_in, + "deliveryMode": { + "transportType": "WebHook", + "address": webhook_url, + }, + } + + if not any("adaptive-cards" in f for f in payload["eventFilters"]): + payload["eventFilters"].append("/team-messaging/v1/adaptive-cards/action") + + result = await client.post("/restapi/v1.0/subscription", body=payload) + logger.info("RingCentral subscription created: id=%s", result.get("id")) + return result + + +async def renew_subscription( + client: AsyncRingCentralClient, + subscription_id: str, + expires_in: int = DEFAULT_SUBSCRIPTION_EXPIRES, +) -> dict: + result = await client.put( + f"/restapi/v1.0/subscription/{subscription_id}", + body={"expiresIn": expires_in}, + ) + logger.info("RingCentral subscription renewed: id=%s", subscription_id) + return result + + +async def delete_subscription(client: AsyncRingCentralClient, subscription_id: str) -> None: + await client.delete(f"/restapi/v1.0/subscription/{subscription_id}") + logger.info("RingCentral subscription deleted: id=%s", subscription_id) + + +async def get_subscription(client: AsyncRingCentralClient, subscription_id: str) -> dict: + return await client.get(f"/restapi/v1.0/subscription/{subscription_id}") diff --git a/backend/package/yuxi/channel/extensions/ringcentral/tasks.py b/backend/package/yuxi/channel/extensions/ringcentral/tasks.py new file mode 100644 index 00000000..bc107989 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/ringcentral/tasks.py @@ -0,0 +1,77 @@ +from __future__ import annotations + +import logging + +from yuxi.channel.extensions.ringcentral.sdk import AsyncRingCentralClient + +logger = logging.getLogger(__name__) + + +async def list_tasks( + client: AsyncRingCentralClient, + chat_id: str, + *, + page: int = 1, + per_page: int = 20, +) -> dict: + return await client.get( + f"/restapi/v1.0/glip/chats/{chat_id}/tasks", + params={"page": page, "perPage": per_page}, + ) + + +async def create_task( + client: AsyncRingCentralClient, + chat_id: str, + subject: str, + *, + assignees: list[str] | None = None, + body: str | None = None, + due_date: str | None = None, +) -> dict: + payload = {"subject": subject} + if assignees: + payload["assignees"] = [{"id": a} for a in assignees] + if body: + payload["body"] = body + if due_date: + payload["dueDate"] = due_date + return await client.post( + f"/restapi/v1.0/glip/chats/{chat_id}/tasks", + body=payload, + ) + + +async def get_task( + client: AsyncRingCentralClient, + task_id: str, +) -> dict: + return await client.get(f"/restapi/v1.0/glip/tasks/{task_id}") + + +async def update_task( + client: AsyncRingCentralClient, + task_id: str, + *, + subject: str | None = None, + body: str | None = None, + due_date: str | None = None, + status: str | None = None, +) -> dict: + payload = {} + if subject is not None: + payload["subject"] = subject + if body is not None: + payload["body"] = body + if due_date is not None: + payload["dueDate"] = due_date + if status is not None: + payload["status"] = status + return await client.patch( + f"/restapi/v1.0/glip/tasks/{task_id}", + body=payload, + ) + + +async def delete_task(client: AsyncRingCentralClient, task_id: str) -> None: + await client.delete(f"/restapi/v1.0/glip/tasks/{task_id}") diff --git a/backend/package/yuxi/channel/extensions/ringcentral/teams.py b/backend/package/yuxi/channel/extensions/ringcentral/teams.py new file mode 100644 index 00000000..1e7089a1 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/ringcentral/teams.py @@ -0,0 +1,53 @@ +from __future__ import annotations + +import logging + +from yuxi.channel.extensions.ringcentral.sdk import AsyncRingCentralClient + +logger = logging.getLogger(__name__) + + +async def create_team( + client: AsyncRingCentralClient, + name: str, + description: str = "", + members: list[str] | None = None, +) -> dict: + body = {"name": name} + if description: + body["description"] = description + if members: + body["members"] = [{"id": m} for m in members] + return await client.post("/restapi/v1.0/glip/teams", body=body) + + +async def update_team( + client: AsyncRingCentralClient, + team_id: str, + *, + name: str | None = None, + description: str | None = None, +) -> dict: + body = {} + if name is not None: + body["name"] = name + if description is not None: + body["description"] = description + return await client.patch(f"/restapi/v1.0/glip/teams/{team_id}", body=body) + + +async def delete_team(client: AsyncRingCentralClient, team_id: str) -> None: + await client.delete(f"/restapi/v1.0/glip/teams/{team_id}") + + +async def add_team_members( + client: AsyncRingCentralClient, + chat_id: str, + members: list[str], +) -> dict: + body = {"members": [{"id": m} for m in members]} + return await client.post(f"/restapi/v1.0/glip/teams/{chat_id}/add", body=body) + + +async def archive_team(client: AsyncRingCentralClient, chat_id: str) -> None: + await client.post(f"/restapi/v1.0/glip/teams/{chat_id}/archive") diff --git a/backend/package/yuxi/channel/extensions/ringcentral/types.py b/backend/package/yuxi/channel/extensions/ringcentral/types.py new file mode 100644 index 00000000..f4f4f125 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/ringcentral/types.py @@ -0,0 +1,93 @@ +from __future__ import annotations + +from dataclasses import dataclass, field + + +@dataclass +class RingCentralUser: + id: str = "" + name: str = "" + email: str | None = None + avatar_url: str | None = None + + +@dataclass +class RingCentralChat: + id: str = "" + type: str = "" + name: str | None = None + + +@dataclass +class RingCentralPost: + id: str = "" + group_id: str = "" + creator_id: str = "" + text: str = "" + type: str = "TextMessage" + attachments: list[dict] = field(default_factory=list) + creation_time: str = "" + last_modified_time: str = "" + mentions: list[dict] = field(default_factory=list) + + +@dataclass +class RingCentralEvent: + uuid: str = "" + event: str = "" + timestamp: str = "" + subscription_id: str = "" + body: RingCentralPost | None = None + _raw: dict = field(default_factory=dict) + + @classmethod + def from_payload(cls, payload: dict) -> RingCentralEvent: + body = payload.get("body", {}) + post = None + if body: + post = RingCentralPost( + id=body.get("id", ""), + group_id=body.get("groupId", ""), + creator_id=body.get("creatorId", ""), + text=body.get("text", ""), + type=body.get("type", "TextMessage"), + attachments=body.get("attachments", []), + creation_time=body.get("creationTime", ""), + last_modified_time=body.get("lastModifiedTime", ""), + mentions=body.get("mentions", []), + ) + return cls( + uuid=payload.get("uuid", ""), + event=payload.get("event", ""), + timestamp=payload.get("timestamp", ""), + subscription_id=payload.get("subscriptionId", ""), + body=post, + _raw=payload, + ) + + +@dataclass +class ResolvedRingCentralAccount: + account_id: str + name: str = "" + client_id: str = "" + client_secret: str = "" + server_url: str = "https://platform.ringcentral.com" + jwt_token: str = "" + webhook_path: str = "/api/webhook/ringcentral" + webhook_validation_token: str = "" + bot_person_id: str = "" + subscription_id: str = "" + dm_policy: str = "pairing" + dm_enabled: bool = True + dm_allow_from: list[str] = field(default_factory=list) + group_policy: str = "allowlist" + require_mention: bool = True + text_chunk_limit: int = 1000 + media_max_mb: int = 20 + enabled: bool = True + config: dict = field(default_factory=dict) + + @property + def is_configured(self) -> bool: + return bool(self.client_id and self.client_secret and self.jwt_token) diff --git a/backend/package/yuxi/channel/extensions/ringcentral/webhook.py b/backend/package/yuxi/channel/extensions/ringcentral/webhook.py new file mode 100644 index 00000000..6a52440a --- /dev/null +++ b/backend/package/yuxi/channel/extensions/ringcentral/webhook.py @@ -0,0 +1,239 @@ +from __future__ import annotations + +import asyncio +import logging + +from yuxi.channel.extensions.ringcentral.types import RingCentralEvent, ResolvedRingCentralAccount + +logger = logging.getLogger(__name__) + +WEBHOOK_TIMEOUT = 120.0 + + +def _needs_event_body_parsing(event_type: str) -> bool: + return event_type in ("PostAdded", "PostChanged", "PostRemoved") + + +def _is_card_action_event(event_type: str) -> bool: + return event_type in ("adaptiveCard/action", "adaptiveCard/action") + + +def _is_group_event(event_type: str) -> bool: + return event_type in ("GroupAdded", "GroupChanged", "GroupRemoved") + + +class RingCentralWebhookHandler: + def __init__(self): + self._targets: dict[str, ResolvedRingCentralAccount] = {} + + def register_target(self, webhook_path: str, account: ResolvedRingCentralAccount) -> None: + self._targets[webhook_path] = account + + def unregister_target(self, webhook_path: str) -> None: + self._targets.pop(webhook_path, None) + + def resolve_target(self, webhook_path: str) -> ResolvedRingCentralAccount | None: + return self._targets.get(webhook_path) + + async def handle_webhook( + self, + body: dict, + account: ResolvedRingCentralAccount | None = None, + ) -> dict: + raw_event_type = body.get("event", "") + logger.debug("RingCentral webhook event: %s", raw_event_type) + + if _is_card_action_event(raw_event_type): + return await self._handle_card_action(body, account) + + if _is_group_event(raw_event_type): + return await self._handle_group_event(body, raw_event_type, account) + + if _needs_event_body_parsing(raw_event_type): + return await self._handle_post_event(body, account) + + return {"status": "ignored", "reason": f"unhandled event type: {raw_event_type}"} + + async def _handle_card_action( + self, + payload: dict, + account: ResolvedRingCentralAccount | None, + ) -> dict: + action_data = payload.get("body", {}) + action = action_data.get("action", {}) + card = action_data.get("card", {}) + logger.info( + "RingCentral adaptive card action: action_type=%s, card_id=%s", + action.get("type"), + card.get("id"), + ) + + from yuxi.channel.message.models import MessageType, PeerInfo, UnifiedMessage + from yuxi.channel.routing.models import PeerKind + + sender_id = action_data.get("userId", "") or action_data.get("creatorId", "") + unified_msg = UnifiedMessage( + msg_id=payload.get("uuid", f"rc-ca:{payload.get('timestamp', '')}"), + channel_type="ringcentral", + account_id=account.account_id if account else "default", + content=action.get("title", "") or action.get("data", {}).get("value", ""), + sender=PeerInfo(kind=PeerKind.GROUP, id=sender_id, display_name=None), + message_type=MessageType.EVENT, + group=None, + raw_payload=payload, + metadata={ + "card_action": True, + "card_id": card.get("id"), + "action_type": action.get("type"), + "action_data": action.get("data"), + }, + ) + + from yuxi.channel.runtime.manager import gateway + + processor = getattr(gateway, "_processor", None) or getattr(gateway, "processor", None) + if processor: + await asyncio.wait_for(processor.process(unified_msg), timeout=WEBHOOK_TIMEOUT) + return {"status": "processed", "event_type": "card_action"} + + async def _handle_group_event( + self, + payload: dict, + event_type: str, + account: ResolvedRingCentralAccount | None, + ) -> dict: + group_data = payload.get("body", {}) + logger.info( + "RingCentral group event: type=%s, group_id=%s, name=%s", + event_type, + group_data.get("id"), + group_data.get("name"), + ) + return {"status": "acknowledged", "event_type": event_type, "group_id": group_data.get("id")} + + async def _handle_post_event( + self, + body: dict, + account: ResolvedRingCentralAccount | None, + ) -> dict: + event = RingCentralEvent.from_payload(body) + if not event.body: + return {"status": "ignored", "reason": "no event body"} + if account and account.bot_person_id and event.body.creator_id == account.bot_person_id: + return {"status": "ignored", "reason": "self message"} + + text = event.body.text or "" + attachments = getattr(event.body, "attachments", []) or [] + if not text and not attachments: + return {"status": "ignored", "reason": "empty text and no attachments"} + + return await self._process_post_event(body, event, account) + + async def _process_post_event( + self, + raw_payload: dict, + event: RingCentralEvent, + account: ResolvedRingCentralAccount | None, + ) -> dict: + event_type = event.event + if event_type == "PostAdded": + return await self._handle_post_added(raw_payload, event, account) + if event_type == "PostChanged": + return await self._handle_post_changed(raw_payload, event, account) + if event_type == "PostRemoved": + return await self._handle_post_removed(raw_payload, event, account) + return {"status": "ignored", "reason": f"unhandled post event: {event_type}"} + + async def _handle_post_added( + self, + raw_payload: dict, + event: RingCentralEvent, + account: ResolvedRingCentralAccount | None, + ) -> dict: + try: + from yuxi.channel.extensions.ringcentral.monitor import RingCentralMonitor + + monitor = RingCentralMonitor() + unified_msg = await monitor.event_to_unified_message(raw_payload, event, account=account) + if unified_msg is None: + return {"status": "filtered"} + + from yuxi.channel.runtime.manager import gateway + + processor = getattr(gateway, "_processor", None) or getattr(gateway, "processor", None) + if processor: + await asyncio.wait_for(processor.process(unified_msg), timeout=WEBHOOK_TIMEOUT) + + return {"status": "processed"} + except TimeoutError: + logger.error("Agent response timeout for RingCentral message") + return {"status": "timeout"} + except Exception: + logger.exception("Failed to process RingCentral message") + return {"status": "error"} + + async def _handle_post_changed( + self, + raw_payload: dict, + event: RingCentralEvent, + account: ResolvedRingCentralAccount | None, + ) -> dict: + from yuxi.channel.extensions.ringcentral.monitor import RingCentralMonitor + + monitor = RingCentralMonitor() + unified_msg = await monitor.event_to_unified_message(raw_payload, event, account=account) + if unified_msg is None: + return {"status": "filtered"} + + if _contains_reaction_change(event): + unified_msg.metadata["reaction_change"] = True + logger.debug("RingCentral reaction change detected in PostChanged") + + unified_msg.metadata["edit_event"] = True + from yuxi.channel.runtime.manager import gateway + + processor = getattr(gateway, "_processor", None) or getattr(gateway, "processor", None) + if processor: + await asyncio.wait_for(processor.process(unified_msg), timeout=WEBHOOK_TIMEOUT) + return {"status": "processed"} + + async def _handle_post_removed( + self, + raw_payload: dict, + event: RingCentralEvent, + account: ResolvedRingCentralAccount | None, + ) -> dict: + from yuxi.channel.message.models import MessageType, PeerInfo, UnifiedMessage + from yuxi.channel.routing.models import PeerKind + + body = event.body + unified_msg = UnifiedMessage( + msg_id=f"rc-del:{event.uuid}", + channel_type="ringcentral", + account_id=account.account_id if account else "default", + content="", + sender=PeerInfo(kind=PeerKind.GROUP, id="system", display_name="System"), + message_type=MessageType.EVENT, + group=None, + raw_payload=raw_payload, + metadata={ + "delete_event": True, + "deleted_post_id": body.id if body else "", + "event_type": event.event, + }, + ) + from yuxi.channel.runtime.manager import gateway + + processor = getattr(gateway, "_processor", None) or getattr(gateway, "processor", None) + if processor: + await asyncio.wait_for(processor.process(unified_msg), timeout=WEBHOOK_TIMEOUT) + return {"status": "processed"} + + +def _contains_reaction_change(event: RingCentralEvent) -> bool: + if not event.body: + return False + reactions = getattr(event.body, "reactions", None) + if reactions is None: + return False + return len(reactions) > 0