From 4d52f634fe23c3e308186b5d3342535dbf279330 Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Thu, 21 May 2026 10:50:15 +0800 Subject: [PATCH] =?UTF-8?q?feat(imessage):=20=E6=96=B0=E5=A2=9EiMessage?= =?UTF-8?q?=E6=B8=A0=E9=81=93=E6=8F=92=E4=BB=B6=E5=AE=8C=E6=95=B4=E5=AE=9E?= =?UTF-8?q?=E7=8E=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 该提交新增了基于BlueBubbles的iMessage渠道插件，支持单聊和群组消息，包含文本、图片、语音、文件和视频消息收发，支持消息编辑、撤回、回复、 reactions和输入状态提示，同时实现了账号配置、安全校验、配对授权、消息格式化与分片等完整功能。 --- .../channel/extensions/imessage/__init__.py | 285 ++++++++ .../channel/extensions/imessage/config.py | 153 ++++ .../yuxi/channel/extensions/imessage/echo.py | 38 + .../channel/extensions/imessage/errors.py | 35 + .../channel/extensions/imessage/format.py | 77 ++ .../channel/extensions/imessage/gateway.py | 148 ++++ .../yuxi/channel/extensions/imessage/media.py | 104 +++ .../channel/extensions/imessage/monitor.py | 265 +++++++ .../channel/extensions/imessage/outbound.py | 680 ++++++++++++++++++ .../channel/extensions/imessage/pairing.py | 68 ++ .../channel/extensions/imessage/plugin.json | 26 + .../channel/extensions/imessage/security.py | 111 +++ .../channel/extensions/imessage/session.py | 25 + .../channel/extensions/imessage/status.py | 47 ++ .../channel/extensions/imessage/targets.py | 55 ++ .../yuxi/channel/extensions/imessage/types.py | 92 +++ 16 files changed, 2209 insertions(+) create mode 100644 backend/package/yuxi/channel/extensions/imessage/__init__.py create mode 100644 backend/package/yuxi/channel/extensions/imessage/config.py create mode 100644 backend/package/yuxi/channel/extensions/imessage/echo.py create mode 100644 backend/package/yuxi/channel/extensions/imessage/errors.py create mode 100644 backend/package/yuxi/channel/extensions/imessage/format.py create mode 100644 backend/package/yuxi/channel/extensions/imessage/gateway.py create mode 100644 backend/package/yuxi/channel/extensions/imessage/media.py create mode 100644 backend/package/yuxi/channel/extensions/imessage/monitor.py create mode 100644 backend/package/yuxi/channel/extensions/imessage/outbound.py create mode 100644 backend/package/yuxi/channel/extensions/imessage/pairing.py create mode 100644 backend/package/yuxi/channel/extensions/imessage/plugin.json create mode 100644 backend/package/yuxi/channel/extensions/imessage/security.py create mode 100644 backend/package/yuxi/channel/extensions/imessage/session.py create mode 100644 backend/package/yuxi/channel/extensions/imessage/status.py create mode 100644 backend/package/yuxi/channel/extensions/imessage/targets.py create mode 100644 backend/package/yuxi/channel/extensions/imessage/types.py diff --git a/backend/package/yuxi/channel/extensions/imessage/__init__.py b/backend/package/yuxi/channel/extensions/imessage/__init__.py new file mode 100644 index 00000000..b5fde61c --- /dev/null +++ b/backend/package/yuxi/channel/extensions/imessage/__init__.py @@ -0,0 +1,285 @@ +from yuxi.channel.capabilities import ChannelCapabilities +from yuxi.channel.extensions.base import BaseChannelPlugin +from yuxi.channel.extensions.imessage.config import IMessageConfigAdapter +from yuxi.channel.extensions.imessage.echo import EchoCache +from yuxi.channel.extensions.imessage.gateway import IMessageGatewayAdapter +from yuxi.channel.extensions.imessage.monitor import IMessageMonitor +from yuxi.channel.extensions.imessage.outbound import IMessageOutboundAdapter +from yuxi.channel.extensions.imessage.pairing import IMessagePairingAdapter +from yuxi.channel.extensions.imessage.security import IMessageSecurityAdapter +from yuxi.channel.extensions.imessage.status import IMessageStatusAdapter +from yuxi.channel.extensions.imessage.format import IMessageFormatAdapter +from yuxi.channel.extensions.imessage.session import IMessageSessionAdapter +from yuxi.channel.plugins.registry import ChannelPluginRegistry + +import asyncio + + +class IMessagePlugin(BaseChannelPlugin): + id = "imessage" + name = "iMessage" + order = 90 + label = "iMessage" + aliases = ["imsg"] + + def __init__(self): + self._echo_cache = EchoCache() + self._config = IMessageConfigAdapter() + self._monitor = IMessageMonitor(echo_cache=self._echo_cache) + self._gateway = IMessageGatewayAdapter(monitor=self._monitor) + self._outbound = IMessageOutboundAdapter(echo_cache=self._echo_cache) + self._status = IMessageStatusAdapter() + self._security = IMessageSecurityAdapter(config_adapter=self._config) + self._format = IMessageFormatAdapter() + self._session = IMessageSessionAdapter() + self._pairing = IMessagePairingAdapter() + self._raw_config: dict = {} + self._current_account_id: str = "default" + + @property + def capabilities(self) -> ChannelCapabilities: + return ChannelCapabilities( + chat_types=["direct", "group"], + message_types=["text", "image", "voice", "file", "video"], + interactions=[], + reactions=True, + typing_indicator=True, + threads=False, + edit=True, + unsend=True, + reply=True, + media=True, + native_commands=False, + polls=False, + streaming=False, + block_streaming=False, + block_streaming_chunk_min_chars=2000, + block_streaming_chunk_max_chars=4000, + ) + + def list_account_ids(self, config: dict) -> list[str]: + return self._config.list_account_ids(config) + + async def resolve_account(self, account_id: str = "default") -> dict: + return await self._config.resolve_account(self._raw_config, account_id) or {} + + 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) + + async def start(self, ctx) -> object: + self._raw_config = getattr(ctx, "config", {}) or {} + self._current_account_id = getattr(ctx, "account_id", None) or "default" + return await self._gateway.start(ctx) + + async def stop(self, ctx) -> None: + await self._gateway.stop(ctx) + + 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: + aid = account_id or self._current_account_id + _account = await self._config.resolve_account(self._raw_config, aid) + await self._outbound.send_text( + target_id, content, + reply_to_id=reply_to_id, thread_id=thread_id, account_id=aid, + _account=_account, + ) + + 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, + ) -> None: + _account = await self._config.resolve_account(self._raw_config, self._current_account_id) + await self._outbound.send_media( + target_id, media_url, media_type, + reply_to_id=reply_to_id, thread_id=thread_id, + _account=_account, + ) + + async def send_reaction( + self, + target_id: str, + message_id: str, + emoji: str, + ) -> None: + _account = await self._config.resolve_account(self._raw_config, self._current_account_id) + await self._outbound.send_reaction( + target_id, message_id, emoji, + _account=_account, + ) + + 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: + aid = account_id or self._current_account_id + _account = await self._config.resolve_account(self._raw_config, aid) + return await self._outbound.edit_message( + target_id, message_id, content, + thread_id=thread_id, account_id=aid, _account=_account, + ) + + async def unsend_message( + self, + target_id: str, + message_id: str, + *, + account_id: str | None = None, + ) -> bool: + aid = account_id or self._current_account_id + _account = await self._config.resolve_account(self._raw_config, aid) + return await self._outbound.unsend_message( + target_id, message_id, + account_id=aid, _account=_account, + ) + + async def fetch_chat_history( + self, + chat_guid: str, + *, + limit: int = 50, + before: int | None = None, + after: int | None = None, + ) -> list[dict]: + _account = await self._config.resolve_account(self._raw_config, self._current_account_id) + return await self._outbound.fetch_chat_history( + chat_guid, limit=limit, before=before, after=after, _account=_account, + ) + + async def send_typing(self, target_id: str, thread_id: str | None = None) -> None: + _account = await self._config.resolve_account(self._raw_config, self._current_account_id) + await self._outbound.send_typing_indicator(target_id, _account=_account) + + async def clear_typing(self, target_id: str, thread_id: str | None = None) -> None: + pass + + async def start_typing(self, target_id: str, account: dict) -> object: + await self._outbound.send_typing_indicator(target_id, _account=account) + return target_id + + async def stop_typing(self, handle: object) -> None: + pass + + async def typing_indicator(self, target_id: str, account: dict, _stop_event) -> None: + await self._outbound.send_typing_indicator(target_id, _account=account) + while not _stop_event.is_set(): + try: + await asyncio.wait_for(_stop_event.wait(), timeout=10.0) + break + except TimeoutError: + await self._outbound.send_typing_indicator(target_id, _account=account) + + async def list_groups(self) -> list[dict]: + _account = await self._config.resolve_account(self._raw_config, self._current_account_id) + return await self._outbound.list_chats(_account=_account) + + def extract_thread_id(self, msg: object) -> str | None: + if hasattr(msg, "message_thread_id"): + return msg.message_thread_id + return None + + def extract_mentions(self, raw_message: dict) -> list[str]: + mentions = raw_message.get("mentions") or [] + return [m.get("handle", "") for m in mentions if isinstance(m, dict)] + + async def on_config_changed(self, prev_cfg: dict, next_cfg: dict, account_id: str) -> None: + self._raw_config = next_cfg + + async def on_account_removed(self, account_id: str) -> None: + self._echo_cache.clear() + + async def on_retire(self) -> None: + self._echo_cache.clear() + if self._outbound._http: + await self._outbound._http.aclose() + self._outbound._http = None + + async def list_allow_entries(self, config: dict, scope: str) -> list[str]: + channel_config = self._config._get_channel_config(config) + if scope == "dm": + return list(channel_config.get("allow_from", [])) + if scope == "group": + return list(channel_config.get("group_allow_from", [])) + return [] + + async def add_allow_entry(self, config: dict, scope: str, entry: str) -> bool: + if not self._security.is_valid_allow_entry(entry): + return False + channel_config = config.setdefault("channels", {}).setdefault(self.id, {}) + key = "allow_from" if scope == "dm" else "group_allow_from" + entries: list = channel_config.setdefault(key, []) + if entry not in entries: + entries.append(entry) + return True + + async def remove_allow_entry(self, config: dict, scope: str, entry: str) -> bool: + channel_config = config.get("channels", {}).get(self.id, {}) + key = "allow_from" if scope == "dm" else "group_allow_from" + entries: list = channel_config.get(key, []) + if entry in entries: + entries.remove(entry) + return True + return False + + 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) + + async def check_allowlist(self, peer_id: str, channel_type: str) -> bool: + return await self._security.check_allowlist(peer_id, channel_type, self._raw_config) + + def resolve_dm_policy(self) -> dict: + return self._security.resolve_dm_policy(self._raw_config) + + async def generate_code(self, peer_id: str) -> str: + return self._pairing.generate_code(peer_id) + + async def verify_code(self, peer_id: str, code: str) -> bool: + return self._pairing.verify_code(peer_id, code) + + def resolve_session(self, msg: object): + return self._session.resolve_session(msg) + + def config_schema(self) -> dict: + return self._config.config_schema() + + @property + def channel_format_instructions(self) -> str | None: + return ( + "iMessage only supports plain text. Do not use Markdown formatting. " + "Use numbered lists, bullets with '-' prefix, and uppercase for emphasis. " + "Keep responses concise. Messages longer than 4000 characters will be split." + ) + + +imessage_plugin = IMessagePlugin() +ChannelPluginRegistry.register(imessage_plugin) diff --git a/backend/package/yuxi/channel/extensions/imessage/config.py b/backend/package/yuxi/channel/extensions/imessage/config.py new file mode 100644 index 00000000..dcb129fd --- /dev/null +++ b/backend/package/yuxi/channel/extensions/imessage/config.py @@ -0,0 +1,153 @@ +from __future__ import annotations + +import os + + +class IMessageConfigAdapter: + CHANNEL_KEY = "imessage" + + def _get_channel_config(self, config: dict) -> dict: + return config.get("channels", {}).get(self.CHANNEL_KEY, {}) + + def list_account_ids(self, config: dict) -> list[str]: + channel_config = self._get_channel_config(config) + accounts = channel_config.get("accounts", {}) + return list(accounts.keys()) if accounts else ["default"] + + async def resolve_account(self, config: dict, account_id: str = "default") -> dict | None: + channel_config = self._get_channel_config(config) + accounts = channel_config.get("accounts", {}) + account = accounts.get(account_id, {}) + + server_url = ( + account.get("server_url") + or channel_config.get("server_url") + or os.getenv("IMESSAGE_SERVER_URL", "") + ) + password = ( + account.get("password") + or channel_config.get("password") + or os.getenv("IMESSAGE_SERVER_PASSWORD", "") + ) + + if not server_url or not password: + return None + + return { + "account_id": account_id, + "server_url": server_url, + "password": password, + "name": account.get("name", account_id), + "dm_policy": account.get("dm_policy", channel_config.get("dm_policy", "pairing")), + "allow_from": account.get("allow_from", channel_config.get("allow_from", [])), + "group_policy": account.get("group_policy", channel_config.get("group_policy", "allowlist")), + "group_allow_from": account.get("group_allow_from", channel_config.get("group_allow_from", [])), + "media_max_mb": account.get("media_max_mb", channel_config.get("media_max_mb", 16)), + "text_chunk_limit": account.get("text_chunk_limit", channel_config.get("text_chunk_limit", 4000)), + "streaming": account.get("streaming", channel_config.get("streaming", {})), + "config": {**channel_config, **account}, + } + + def is_configured(self, account: dict | None) -> bool: + return bool(account and account.get("server_url") and account.get("password")) + + def is_enabled(self, account: dict | None) -> bool: + if account is None: + return False + return account.get("config", {}).get("enabled", True) if account else False + + def disabled_reason(self, account: dict | None) -> str: + if not account: + return "Not configured" + if not account.get("server_url"): + return "Missing server URL" + if not account.get("password"): + return "Missing password" + return "" + + async def resolve_allow_from(self, config: dict, account_id: str) -> list[str] | None: + account = await self.resolve_account(config, account_id) + if not account: + return None + return account.get("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), + "server_url": account.get("server_url", ""), + } + + def config_schema(self) -> dict: + return { + "type": "object", + "properties": { + "enabled": {"type": "boolean", "default": True}, + "accounts": { + "type": "object", + "additionalProperties": { + "type": "object", + "properties": { + "server_url": {"type": "string", "description": "BlueBubbles server URL"}, + "password": {"type": "string", "description": "BlueBubbles server password"}, + "name": {"type": "string"}, + "dm_policy": { + "type": "string", + "enum": ["pairing", "allowlist", "open", "disabled"], + "default": "pairing", + }, + "allow_from": { + "type": "array", + "items": {"type": "string"}, + "description": "E.164 numbers or emails allowed to DM", + }, + "group_policy": { + "type": "string", + "enum": ["open", "allowlist", "disabled"], + "default": "allowlist", + }, + "group_allow_from": { + "type": "array", + "items": {"type": "string"}, + }, + "media_max_mb": {"type": "integer", "default": 16}, + "text_chunk_limit": {"type": "integer", "default": 4000}, + "streaming": { + "type": "object", + "properties": { + "mode": {"type": "string", "enum": ["block"], "default": "block"}, + "min_chars": {"type": "integer", "default": 2000}, + "coalesce_min_chars": {"type": "integer", "default": 400}, + }, + }, + }, + }, + }, + "default_account": {"type": "string", "default": "default"}, + "groups": { + "type": "object", + "description": "Per-group config overrides keyed by chat GUID", + "additionalProperties": { + "type": "object", + "properties": { + "require_mention": {"type": "boolean", "description": "Require @mention to trigger"}, + "group_policy": { + "type": "string", + "enum": ["open", "allowlist", "disabled"], + }, + "system_prompt": {"type": "string", "description": "Group-specific system prompt"}, + }, + }, + }, + "streaming": { + "type": "object", + "description": "Default streaming settings", + "properties": { + "mode": {"type": "string", "enum": ["block"], "default": "block"}, + "min_chars": {"type": "integer", "default": 2000}, + "coalesce_min_chars": {"type": "integer", "default": 400}, + }, + }, + }, + } diff --git a/backend/package/yuxi/channel/extensions/imessage/echo.py b/backend/package/yuxi/channel/extensions/imessage/echo.py new file mode 100644 index 00000000..4b484619 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/imessage/echo.py @@ -0,0 +1,38 @@ +from __future__ import annotations + +import hashlib +import time + + +class EchoCache: + def __init__(self, ttl_ms: int = 10_000): + self._cache: dict[str, float] = {} + self._ttl = ttl_ms + + def remember(self, chat_guid: str, text: str) -> None: + key = self._build_key(chat_guid, text) + self._cache[key] = time.monotonic() + + def is_echo(self, chat_guid: str, text: str) -> bool: + key = self._build_key(chat_guid, text) + ts = self._cache.get(key) + if ts and (time.monotonic() - ts) * 1000 < self._ttl: + return True + return False + + def cleanup(self) -> None: + now = time.monotonic() + expired = [ + k for k, ts in self._cache.items() + if (now - ts) * 1000 >= self._ttl + ] + for k in expired: + del self._cache[k] + + def clear(self) -> None: + self._cache.clear() + + @staticmethod + def _build_key(chat_guid: str, text: str) -> str: + raw = f"{chat_guid}:{text}" + return hashlib.sha256(raw.encode()).hexdigest()[:16] diff --git a/backend/package/yuxi/channel/extensions/imessage/errors.py b/backend/package/yuxi/channel/extensions/imessage/errors.py new file mode 100644 index 00000000..ebb3f796 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/imessage/errors.py @@ -0,0 +1,35 @@ +from __future__ import annotations + + +class IMessageError(Exception): + pass + + +class IMessageAuthError(IMessageError): + pass + + +class IMessageConnectionError(IMessageError): + pass + + +class IMessageSendError(IMessageError): + pass + + +class IMessageMediaError(IMessageError): + pass + + +def classify_http_error(status_code: int, body: str = "") -> IMessageError: + if status_code == 401 or status_code == 403: + return IMessageAuthError(f"HTTP {status_code}: {body}") + if status_code >= 500: + return IMessageConnectionError(f"HTTP {status_code}: {body}") + return IMessageError(f"HTTP {status_code}: {body}") + + +def get_retry_after_ms(error: Exception) -> int | None: + if isinstance(error, IMessageConnectionError): + return 5000 + return None diff --git a/backend/package/yuxi/channel/extensions/imessage/format.py b/backend/package/yuxi/channel/extensions/imessage/format.py new file mode 100644 index 00000000..56b4ca36 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/imessage/format.py @@ -0,0 +1,77 @@ +from __future__ import annotations + +import re + +_RE_STRIP_MARKDOWN_LINK = re.compile(r"\[([^\]]*)\]\([^)]+\)") +_RE_STRIP_MARKDOWN_IMAGE = re.compile(r"!\[([^\]]*)\]\([^)]+\)") + + +def markdown_to_plaintext(md_text: str) -> str: + result = md_text + result = re.sub(r"\*\*(.+?)\*\*", lambda m: m.group(1).upper(), result) + result = re.sub(r"__([^_]+)__", lambda m: m.group(1).upper(), result) + result = re.sub(r"\*([^*]+)\*", r"_\1_", result) + result = re.sub(r"_([^_]+)_", r"_\1_", result) + result = re.sub(r"~~(.+?)~~", "", result) + result = _RE_STRIP_MARKDOWN_IMAGE.sub(r"[\1]", result) + result = _RE_STRIP_MARKDOWN_LINK.sub(lambda m: f"{m.group(1)} ({m.group(0).split('](')[1].rstrip(')')})", result) + result = re.sub(r"```(\w*)\n(.*?)```", r"[CODE]\n\2\n[/CODE]", result, flags=re.DOTALL) + result = result.replace(" & ", " & ") + result = result.replace(" < ", " < ") + result = result.replace(" > ", " > ") + return result.strip() + + +def convert_table_to_bullets(md_text: str) -> str: + lines = md_text.split("\n") + result: list[str] = [] + in_table = False + headers: list[str] = [] + + for line in lines: + stripped = line.strip() + if stripped.startswith("|") and stripped.endswith("|"): + cells = [c.strip() for c in stripped[1:-1].split("|")] + if not in_table: + headers = cells + in_table = True + continue + if all(re.match(r"^[-:]+$", c) for c in cells): + continue + for i, header in enumerate(headers): + value = cells[i] if i < len(cells) else "" + result.append(f" {header}: {value}") + else: + if in_table: + in_table = False + result.append(line) + + return "\n".join(result) + + +def split_plaintext_chunks(text: str, limit: int = 4000) -> list[str]: + if len(text) <= limit: + return [text] + + chunks: list[str] = [] + lines = text.split("\n") + current = "" + for line in lines: + if len(current) + len(line) + 1 > limit and current: + chunks.append(current.rstrip()) + current = line + else: + current = current + "\n" + line if current else line + if current: + chunks.append(current.rstrip()) + return chunks + + +class IMessageFormatAdapter: + + def format_for_imessage(self, md_text: str) -> str: + text = convert_table_to_bullets(md_text) + return markdown_to_plaintext(text) + + def chunk_text(self, text: str, limit: int = 4000) -> list[str]: + return split_plaintext_chunks(text, limit) diff --git a/backend/package/yuxi/channel/extensions/imessage/gateway.py b/backend/package/yuxi/channel/extensions/imessage/gateway.py new file mode 100644 index 00000000..281a0d11 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/imessage/gateway.py @@ -0,0 +1,148 @@ +from __future__ import annotations + +import asyncio +import json +import logging + +import websockets + +from yuxi.channel.extensions.imessage.errors import IMessageAuthError +from yuxi.channel.extensions.imessage.monitor import IMessageMonitor +from yuxi.channel.extensions.imessage.types import IMessageGatewayContext + +logger = logging.getLogger(__name__) + + +class IMessageGatewayAdapter: + def __init__(self, monitor=None): + self._monitor = monitor or IMessageMonitor() + self._ws = None + self._ctx: IMessageGatewayContext | None = None + self._receive_task: asyncio.Task | None = None + self._ping_task: asyncio.Task | None = None + self._running = False + self._reconnect_attempt = 0 + + async def start(self, ctx) -> object: + account = getattr(ctx, "account", {}) or {} + server_url = account.get("server_url", "") + password = account.get("password", "") + account_id = account.get("account_id", "default") + + self._ctx = IMessageGatewayContext( + server_url=server_url, + password=password, + account_id=account_id, + ) + + if not server_url or not password: + raise IMessageAuthError("BlueBubbles server URL and password are required") + + self._running = True + + queue = getattr(ctx, "queue", asyncio.Queue()) + + self._receive_task = asyncio.create_task(self._ws_loop(queue)) + self._ping_task = asyncio.create_task(self._ping_loop()) + + return queue + + async def stop(self, ctx) -> None: + self._running = False + for task in [self._receive_task, self._ping_task]: + if task and not task.done(): + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + self._ws = None + + async def _ws_loop(self, queue: asyncio.Queue) -> None: + while self._running: + try: + await self._connect_and_listen(queue) + except asyncio.CancelledError: + break + except Exception as e: + logger.warning("WebSocket loop error: %s", e) + + if not self._running: + break + + delay = self._reconnect_delay() + logger.info("Reconnecting in %.1fs (attempt %d)", delay, self._reconnect_attempt) + await asyncio.sleep(delay) + + async def _connect_and_listen(self, queue: asyncio.Queue) -> None: + if not self._ctx: + return + + ws_url = self._ctx.ws_url + if self._ctx.password: + ws_url = f"{ws_url}?password={self._ctx.password}" + + self._reconnect_attempt += 1 + + async with websockets.connect(ws_url, ping_interval=30, ping_timeout=10, close_timeout=5) as ws: + self._ws = ws + self._reconnect_attempt = 0 + logger.info("Connected to BlueBubbles WebSocket: %s", self._ctx.server_url) + + async for raw in ws: + if not self._running: + break + + try: + event = json.loads(raw) + except json.JSONDecodeError: + continue + + event_type = event.get("type", "") + if event_type == "new-message": + unified_msg = self._monitor.parse_event(event, self._ctx.account_id) + if unified_msg: + await queue.put(unified_msg) + elif event_type == "updated-message": + await self._monitor.handle_updated_message(event, self._ctx.account_id, queue) + elif event_type == "message-send-error": + self._monitor.handle_send_error(event) + elif event_type in ( + "participant-removed", "participant-added", "participant-left", + "group-name-changed", "group-icon-changed", "group-icon-removed", + ): + self._monitor.handle_group_event(event, event_type, self._ctx.account_id, queue) + elif event_type == "typing-indicator": + self._monitor.handle_typing_indicator(event, self._ctx.account_id, queue) + elif event_type == "chat-read-status-changed": + self._monitor.handle_chat_read_status_changed(event, self._ctx.account_id, queue) + elif event_type == "imessage-alias-removed": + self._monitor.handle_alias_removed(event, self._ctx.account_id, queue) + else: + unified_msg = self._monitor.parse_event(event, self._ctx.account_id) + if unified_msg: + await queue.put(unified_msg) + + self._ws = None + + def _reconnect_delay(self) -> float: + ctx = self._ctx + if not ctx: + return 60.0 + if self._reconnect_attempt > ctx.max_reconnect_attempts: + return ctx.reconnect_max_delay + delay = min( + ctx.reconnect_min_delay * (2 ** (self._reconnect_attempt - 1)), + ctx.reconnect_max_delay, + ) + return delay + + async def _ping_loop(self) -> None: + while self._running: + await asyncio.sleep(30) + if not self._ws or not self._running: + continue + try: + await self._ws.ping() + except Exception: + pass diff --git a/backend/package/yuxi/channel/extensions/imessage/media.py b/backend/package/yuxi/channel/extensions/imessage/media.py new file mode 100644 index 00000000..41405c68 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/imessage/media.py @@ -0,0 +1,104 @@ +from __future__ import annotations + +import logging +import os +import tempfile +from pathlib import Path + +import httpx + +from yuxi.channel.extensions.imessage.errors import IMessageMediaError + +logger = logging.getLogger(__name__) + +DEFAULT_MEDIA_MAX_MB = 16 + + +async def download_media( + media_url: str, + max_mb: int = DEFAULT_MEDIA_MAX_MB, + temp_dir: str | None = None, +) -> tuple[bytes, str, str]: + async with httpx.AsyncClient(timeout=httpx.Timeout(60.0)) as client: + resp = await client.get(media_url) + resp.raise_for_status() + data = resp.content + + size_mb = len(data) / (1024 * 1024) + if size_mb > max_mb: + raise IMessageMediaError( + f"Media size {size_mb:.1f}MB exceeds limit of {max_mb}MB" + ) + + content_type = resp.headers.get("content-type", "application/octet-stream") + ext = _guess_extension(content_type) + filename = f"imessage_media_{os.urandom(8).hex()}{ext}" + + write_dir = temp_dir or tempfile.gettempdir() + filepath = os.path.join(write_dir, filename) + Path(filepath).write_bytes(data) + + return data, content_type, filepath + + +def cleanup_temp_file(filepath: str) -> None: + try: + os.unlink(filepath) + except OSError: + pass + + +async def download_attachment_via_api( + attachment_guid: str, + server_url: str, + password: str, + max_mb: int = DEFAULT_MEDIA_MAX_MB, +) -> tuple[bytes, str]: + async with httpx.AsyncClient(timeout=httpx.Timeout(60.0)) as client: + resp = await client.get( + f"{server_url}/api/v1/attachment/{attachment_guid}/download", + headers={"Password": password}, + ) + resp.raise_for_status() + data = resp.content + + size_mb = len(data) / (1024 * 1024) + if size_mb > max_mb: + raise IMessageMediaError( + f"Attachment size {size_mb:.1f}MB exceeds limit of {max_mb}MB" + ) + + content_type = resp.headers.get("content-type", "application/octet-stream") + return data, content_type + + +async def download_attachment_chunked( + attachment_guid: str, + server_url: str, + password: str, + chunk_size: int = 1024 * 1024, +) -> bytes: + async with httpx.AsyncClient(timeout=httpx.Timeout(300.0)) as client: + resp = await client.get( + f"{server_url}/api/v1/attachment/{attachment_guid}/chunk", + headers={"Password": password}, + ) + resp.raise_for_status() + return resp.content + + +def _guess_extension(content_type: str) -> str: + ext_map = { + "image/jpeg": ".jpg", + "image/png": ".png", + "image/gif": ".gif", + "image/heic": ".heic", + "image/heif": ".heif", + "audio/mpeg": ".mp3", + "audio/aac": ".aac", + "audio/wav": ".wav", + "video/mp4": ".mp4", + "video/quicktime": ".mov", + "application/pdf": ".pdf", + } + return ext_map.get(content_type, ".bin") diff --git a/backend/package/yuxi/channel/extensions/imessage/monitor.py b/backend/package/yuxi/channel/extensions/imessage/monitor.py new file mode 100644 index 00000000..6c135ffa --- /dev/null +++ b/backend/package/yuxi/channel/extensions/imessage/monitor.py @@ -0,0 +1,265 @@ +from __future__ import annotations + +import logging +from datetime import datetime + +from yuxi.channel.extensions.imessage.types import BlueBubblesMessageEvent +from yuxi.channel.message.models import ( + GroupContext, + MessageType, + PeerInfo, + UnifiedMessage, +) +from yuxi.channel.routing.models import PeerKind + +logger = logging.getLogger(__name__) + +_TAPBACK_ADDED = { + 2000: "love", 2001: "like", 2002: "dislike", + 2003: "laugh", 2004: "emphasize", 2005: "question", +} +_TAPBACK_REMOVED = { + 3000: "love", 3001: "like", 3002: "dislike", + 3003: "laugh", 3004: "emphasize", 3005: "question", +} +_TAPBACK_CODES = _TAPBACK_ADDED | _TAPBACK_REMOVED + + +class IMessageMonitor: + def __init__(self, echo_cache=None): + self._echo_cache = echo_cache + + def parse_event(self, raw: dict, account_id: str) -> UnifiedMessage | None: + event = BlueBubblesMessageEvent.from_json(raw) + + if event.is_from_me: + return None + + group_action = event.group_action_type + if group_action and group_action in _TAPBACK_CODES: + return self._parse_tapback(raw, event, account_id) + + if not event.text and not event.attachments: + return None + + if self._echo_cache and self._echo_cache.is_echo(event.chat_guid or "", event.text or ""): + return None + + msg_type = self._resolve_message_type(event) + sender = PeerInfo( + kind=PeerKind.DIRECT, + id=event.sender_handle or "unknown", + display_name=event.sender_name, + is_bot=False, + is_self=event.is_from_me, + ) + + group: GroupContext | None = None + is_group = event.chat_identifier and ";" in (event.chat_identifier or "") + if is_group and event.chat_guid: + group = GroupContext( + id=event.chat_guid, + name=event.chat_display_name, + ) + + media_urls: list[str] = [] + media_types: list[str] = [] + for att in event.attachments: + att_guid = att.get("guid") or att.get("attachmentGuid") + mime = att.get("mimeType", "application/octet-stream") + if att_guid: + media_urls.append(att_guid) + media_types.append(mime) + else: + path = att.get("path") or att.get("filePath") + if path: + media_urls.append(path) + media_types.append(mime) + + timestamp = None + if event.date_created: + timestamp = datetime.fromtimestamp(event.date_created / 1000) + + return UnifiedMessage( + msg_id=f"im:{event.guid}", + channel_type="imessage", + account_id=account_id, + content=event.text or "", + sender=sender, + message_type=msg_type, + media_urls=media_urls, + media_types=media_types, + group=group, + timestamp=timestamp, + raw_payload=raw, + message_thread_id=event.thread_originator_guid, + metadata={ + "is_from_me": event.is_from_me, + "chat_identifier": event.chat_identifier, + }, + ) + + @staticmethod + def _resolve_message_type(event: BlueBubblesMessageEvent) -> MessageType: + if not event.attachments: + return MessageType.TEXT + mime = event.attachments[0].get("mimeType", "") + if mime.startswith("image/"): + return MessageType.IMAGE + if mime.startswith("audio/"): + return MessageType.VOICE + if mime.startswith("video/"): + return MessageType.VIDEO + return MessageType.FILE + + @staticmethod + def _parse_tapback(raw: dict, event, account_id: str) -> UnifiedMessage | None: + reaction_name = _TAPBACK_CODES.get(event.group_action_type) + if not reaction_name: + return None + + action = "added" if event.group_action_type < 3000 else "removed" + associated_guid = raw.get("associatedMessageGuid") or "" + + is_group = event.chat_identifier and ";" in (event.chat_identifier or "") + group_info = None + if is_group and event.chat_guid: + group_info = GroupContext( + id=event.chat_guid, + name=event.chat_display_name, + ) + + return UnifiedMessage( + msg_id=f"im:reaction:{event.guid}", + channel_type="imessage", + account_id=account_id, + content=reaction_name, + message_type=MessageType.TEXT, + sender=PeerInfo( + kind=PeerKind.DIRECT if not is_group else PeerKind.GROUP, + id=event.sender_handle or "", + display_name=event.sender_name or event.sender_handle or "", + ), + group=group_info, + raw_payload={ + "reaction": reaction_name, + "action": action, + "associated_message_guid": associated_guid, + "tapback_code": event.group_action_type, + }, + metadata={"event_type": "tapback"}, + ) + + async def handle_updated_message(self, raw: dict, account_id: str, queue) -> None: + event = BlueBubblesMessageEvent.from_json(raw) + if not event.guid: + return + unified = UnifiedMessage( + msg_id=f"im:update:{event.guid}", + channel_type="imessage", + account_id=account_id, + content=event.text or "", + message_type=MessageType.EVENT, + sender=PeerInfo( + kind=PeerKind.CHANNEL, + id="system", + display_name="iMessage", + ), + raw_payload={ + "action": "message_updated", + "original_guid": event.guid, + "updated_text": event.text, + "chat_guid": event.chat_guid, + "is_from_me": event.is_from_me, + }, + metadata={"event_type": "updated-message"}, + ) + await queue.put(unified) + + def handle_send_error(self, raw: dict) -> None: + error_message = raw.get("error", "Unknown send error") + chat_guid = raw.get("chatGuid", "unknown") + logger.error( + "iMessage send error: chat=%s, error=%s, raw=%s", + chat_guid, error_message, raw, + ) + + async def handle_group_event(self, raw: dict, event_type: str, account_id: str, queue) -> None: + chat = raw.get("chat") or {} + chat_guid = chat.get("guid", "") + participant = raw.get("participant", "") + + action_map = { + "participant-removed": "member_removed", + "participant-added": "member_added", + "participant-left": "member_left", + "group-name-changed": "group_renamed", + "group-icon-changed": "group_icon_changed", + "group-icon-removed": "group_icon_removed", + } + action = action_map.get(event_type, "group_event") + + content = "" + if event_type == "group-name-changed": + content = raw.get("newName", "") + elif participant: + content = participant + + unified = UnifiedMessage( + msg_id=f"im:group:{chat_guid}", + channel_type="imessage", + account_id=account_id, + content=content, + message_type=MessageType.EVENT, + sender=PeerInfo(kind=PeerKind.CHANNEL, id="system", display_name="iMessage"), + group=GroupContext(id=chat_guid, name=chat.get("displayName")), + raw_payload={"action": action, "chat_guid": chat_guid, "participant": participant}, + metadata={"event_type": event_type}, + ) + await queue.put(unified) + + def handle_typing_indicator(self, raw: dict, account_id: str, queue) -> None: + sender_handle = raw.get("sender", "") + chat_guid = raw.get("chatGuid", "") + display_name = raw.get("displayName", sender_handle) + is_typing = raw.get("typing", False) + + unified = UnifiedMessage( + msg_id=f"im:typing:{chat_guid}:{sender_handle}", + channel_type="imessage", + account_id=account_id, + content="", + message_type=MessageType.EVENT, + sender=PeerInfo(kind=PeerKind.DIRECT, id=sender_handle, display_name=display_name), + raw_payload={"action": "typing", "chat_guid": chat_guid, "typing": is_typing}, + metadata={"event_type": "typing-indicator", "is_typing": is_typing}, + ) + queue.put_nowait(unified) + + def handle_chat_read_status_changed(self, raw: dict, account_id: str, queue) -> None: + chat_guid = raw.get("chatGuid", "") + unified = UnifiedMessage( + msg_id=f"im:read:{chat_guid}", + channel_type="imessage", + account_id=account_id, + content="", + message_type=MessageType.EVENT, + sender=PeerInfo(kind=PeerKind.CHANNEL, id="system", display_name="iMessage"), + raw_payload={"action": "read_status_changed", "chat_guid": chat_guid}, + metadata={"event_type": "chat-read-status-changed"}, + ) + queue.put_nowait(unified) + + def handle_alias_removed(self, raw: dict, account_id: str, queue) -> None: + alias = raw.get("alias", "") + unified = UnifiedMessage( + msg_id=f"im:alias-removed:{alias}", + channel_type="imessage", + account_id=account_id, + content="", + message_type=MessageType.EVENT, + sender=PeerInfo(kind=PeerKind.CHANNEL, id="system", display_name="iMessage"), + raw_payload={"action": "alias_removed", "alias": alias}, + metadata={"event_type": "imessage-alias-removed"}, + ) + queue.put_nowait(unified) diff --git a/backend/package/yuxi/channel/extensions/imessage/outbound.py b/backend/package/yuxi/channel/extensions/imessage/outbound.py new file mode 100644 index 00000000..f77ff43f --- /dev/null +++ b/backend/package/yuxi/channel/extensions/imessage/outbound.py @@ -0,0 +1,680 @@ +from __future__ import annotations + +import base64 +import logging +import os + +import httpx + +from yuxi.channel.extensions.imessage.errors import IMessageSendError, classify_http_error +from yuxi.channel.extensions.imessage.format import IMessageFormatAdapter +from yuxi.channel.protocols import OutboundDeliveryMode, OutboundDeliveryCapabilities, OutboundPresentationCapabilities + +logger = logging.getLogger(__name__) + +DEFAULT_CHUNK_LIMIT = 4000 + + +class IMessageOutboundAdapter: + delivery_mode = OutboundDeliveryMode.DIRECT + chunker_mode = "newline" + text_chunk_limit = DEFAULT_CHUNK_LIMIT + poll_max_options = None + supports_poll_duration_seconds = False + supports_anonymous_polls = False + extract_markdown_images = False + presentation_capabilities = OutboundPresentationCapabilities( + supported=False, + buttons=False, + selects=False, + context=False, + divider=False, + ) + delivery_capabilities = OutboundDeliveryCapabilities( + pin=False, + durable_final_text=True, + durable_final_media=False, + ) + + def __init__(self, echo_cache=None): + self._format = IMessageFormatAdapter() + self._echo_cache = echo_cache + self._http: httpx.AsyncClient | None = None + + async def _get_http(self) -> httpx.AsyncClient: + if self._http is None: + self._http = httpx.AsyncClient(timeout=httpx.Timeout(30.0)) + return self._http + + def chunker(self, text: str, limit: int, ctx: object | None = None) -> list[str]: + return self._format.chunk_text(text, limit) + + def sanitize_text(self, text: str, payload: object) -> str: + return text + + def should_skip_plain_text_sanitization(self, payload: object) -> bool: + return True + + def normalize_payload(self, payload: object, config: dict, account_id: str | None = None) -> object | None: + return payload + + 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 + + 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, + _account: dict | None = None, + ) -> None: + http = await self._get_http() + server_url, password = await self._resolve_server(account_id, _account) + + formatted = self._format.format_for_imessage(content) + if self._echo_cache: + self._echo_cache.remember(target_id, formatted) + + payload = { + "text": formatted, + "chatGuid": target_id, + } + if reply_to_id: + payload["replyToGuid"] = reply_to_id + + try: + resp = await http.post( + f"{server_url}/api/v1/message/text", + json=payload, + headers={"Password": password}, + ) + if resp.status_code >= 400: + raise classify_http_error(resp.status_code, resp.text) + except httpx.RequestError as e: + raise IMessageSendError(f"Failed to send text: {e}") from e + + 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, + _account: dict | None = None, + ) -> None: + http = await self._get_http() + server_url, password = await self._resolve_server(account_id, _account) + + try: + media_resp = await http.get(media_url) + media_resp.raise_for_status() + media_data = media_resp.content + except Exception as e: + raise IMessageSendError(f"Failed to download media: {e}") from e + + media_b64 = base64.b64encode(media_data).decode("utf-8") + content_type = media_resp.headers.get("content-type", media_type) + + payload = { + "chatGuid": target_id, + "base64Data": media_b64, + "mimeType": content_type, + } + + try: + resp = await http.post( + f"{server_url}/api/v1/message/attachment", + json=payload, + headers={"Password": password}, + ) + if resp.status_code >= 400: + raise classify_http_error(resp.status_code, resp.text) + except httpx.RequestError as e: + raise IMessageSendError(f"Failed to send media: {e}") from e + + async def send_reaction( + self, + target_id: str, + message_id: str, + reaction: str, + *, + account_id: str | None = None, + _account: dict | None = None, + ) -> None: + http = await self._get_http() + server_url, password = await self._resolve_server(account_id, _account) + + payload = { + "chatGuid": target_id, + "reaction": reaction, + } + + try: + resp = await http.post( + f"{server_url}/api/v1/message/{message_id}/reaction", + json=payload, + headers={"Password": password}, + ) + if resp.status_code >= 400: + raise classify_http_error(resp.status_code, resp.text) + except httpx.RequestError as e: + raise IMessageSendError(f"Failed to send reaction: {e}") from e + + async def edit_message( + self, + target_id: str, + message_id: str, + content: str, + *, + thread_id: str | None = None, + account_id: str | None = None, + _account: dict | None = None, + ) -> str | None: + http = await self._get_http() + server_url, password = await self._resolve_server(account_id, _account) + + formatted = self._format.format_for_imessage(content) + payload = {"text": formatted} + + try: + resp = await http.patch( + f"{server_url}/api/v1/message/{message_id}", + json=payload, + headers={"Password": password}, + ) + if resp.status_code >= 400: + raise classify_http_error(resp.status_code, resp.text) + return message_id + except httpx.RequestError as e: + raise IMessageSendError(f"Failed to edit message: {e}") from e + + async def unsend_message( + self, + target_id: str, + message_id: str, + *, + account_id: str | None = None, + _account: dict | None = None, + ) -> bool: + http = await self._get_http() + server_url, password = await self._resolve_server(account_id, _account) + + try: + resp = await http.delete( + f"{server_url}/api/v1/message/{message_id}", + headers={"Password": password}, + ) + if resp.status_code >= 400: + raise classify_http_error(resp.status_code, resp.text) + return True + except httpx.RequestError as e: + raise IMessageSendError(f"Failed to unsend message: {e}") from e + + async def fetch_chat_history( + self, + chat_guid: str, + *, + limit: int = 50, + before: int | None = None, + after: int | None = None, + account_id: str | None = None, + _account: dict | None = None, + ) -> list[dict]: + http = await self._get_http() + server_url, password = await self._resolve_server(account_id, _account) + + params: dict = {"limit": limit} + if before: + params["before"] = before + if after: + params["after"] = after + + try: + resp = await http.get( + f"{server_url}/api/v1/chat/{chat_guid}/message", + params=params, + headers={"Password": password}, + ) + if resp.status_code >= 400: + raise classify_http_error(resp.status_code, resp.text) + data = resp.json() + return data.get("data", []) + except httpx.RequestError as e: + raise IMessageSendError(f"Failed to fetch chat history: {e}") from e + + async def send_typing_indicator( + self, chat_guid: str, + *, account_id: str | None = None, _account: dict | None = None, + ) -> None: + http = await self._get_http() + server_url, password = await self._resolve_server(account_id, _account) + try: + await http.post( + f"{server_url}/api/v1/chat/{chat_guid}/typing", + headers={"Password": password}, + ) + except httpx.RequestError as e: + logger.warning("Failed to send typing indicator: %s", e) + + async def send_read_receipt( + self, chat_guid: str, + *, account_id: str | None = None, _account: dict | None = None, + ) -> None: + http = await self._get_http() + server_url, password = await self._resolve_server(account_id, _account) + try: + await http.post( + f"{server_url}/api/v1/chat/{chat_guid}/readreceipt", + headers={"Password": password}, + ) + except httpx.RequestError as e: + logger.warning("Failed to send read receipt: %s", e) + + async def rename_group( + self, chat_guid: str, name: str, + *, account_id: str | None = None, _account: dict | None = None, + ) -> bool: + http = await self._get_http() + server_url, password = await self._resolve_server(account_id, _account) + try: + resp = await http.patch( + f"{server_url}/api/v1/chat/{chat_guid}/rename", + json={"name": name}, + headers={"Password": password}, + ) + return resp.status_code < 400 + except httpx.RequestError as e: + raise IMessageSendError(f"Failed to rename group: {e}") from e + + async def add_participant( + self, chat_guid: str, handle: str, + *, account_id: str | None = None, _account: dict | None = None, + ) -> bool: + http = await self._get_http() + server_url, password = await self._resolve_server(account_id, _account) + try: + resp = await http.post( + f"{server_url}/api/v1/chat/{chat_guid}/addparticipant", + json={"address": handle}, + headers={"Password": password}, + ) + return resp.status_code < 400 + except httpx.RequestError as e: + raise IMessageSendError(f"Failed to add participant: {e}") from e + + async def remove_participant( + self, chat_guid: str, handle: str, + *, account_id: str | None = None, _account: dict | None = None, + ) -> bool: + http = await self._get_http() + server_url, password = await self._resolve_server(account_id, _account) + try: + resp = await http.post( + f"{server_url}/api/v1/chat/{chat_guid}/removeparticipant", + json={"address": handle}, + headers={"Password": password}, + ) + return resp.status_code < 400 + except httpx.RequestError as e: + raise IMessageSendError(f"Failed to remove participant: {e}") from e + + async def leave_group( + self, chat_guid: str, + *, account_id: str | None = None, _account: dict | None = None, + ) -> bool: + http = await self._get_http() + server_url, password = await self._resolve_server(account_id, _account) + try: + resp = await http.post( + f"{server_url}/api/v1/chat/{chat_guid}/leave", + headers={"Password": password}, + ) + return resp.status_code < 400 + except httpx.RequestError as e: + raise IMessageSendError(f"Failed to leave group: {e}") from e + + async def list_chats( + self, *, account_id: str | None = None, _account: dict | None = None, + ) -> list[dict]: + http = await self._get_http() + server_url, password = await self._resolve_server(account_id, _account) + try: + resp = await http.get( + f"{server_url}/api/v1/chat", + headers={"Password": password}, + ) + if resp.status_code >= 400: + raise classify_http_error(resp.status_code, resp.text) + data = resp.json() + return data.get("data", []) + except httpx.RequestError as e: + raise IMessageSendError(f"Failed to list chats: {e}") from e + + async def get_chat( + self, chat_guid: str, + *, account_id: str | None = None, _account: dict | None = None, + ) -> dict | None: + http = await self._get_http() + server_url, password = await self._resolve_server(account_id, _account) + try: + resp = await http.get( + f"{server_url}/api/v1/chat/{chat_guid}", + headers={"Password": password}, + ) + if resp.status_code >= 400: + raise classify_http_error(resp.status_code, resp.text) + return resp.json() + except httpx.RequestError as e: + raise IMessageSendError(f"Failed to get chat: {e}") from e + + async def send_sms( + self, target_id: str, content: str, + *, account_id: str | None = None, _account: dict | None = None, + ) -> None: + http = await self._get_http() + server_url, password = await self._resolve_server(account_id, _account) + + formatted = self._format.format_for_imessage(content) + payload = {"text": formatted, "chatGuid": target_id, "method": "sms"} + + try: + resp = await http.post( + f"{server_url}/api/v1/message/text", + json=payload, + headers={"Password": password}, + ) + if resp.status_code >= 400: + raise classify_http_error(resp.status_code, resp.text) + except httpx.RequestError as e: + raise IMessageSendError(f"Failed to send SMS: {e}") from e + + async def send_mention( + self, chat_guid: str, message_id: str, mentions: list[str], + *, account_id: str | None = None, _account: dict | None = None, + ) -> None: + http = await self._get_http() + server_url, password = await self._resolve_server(account_id, _account) + try: + resp = await http.post( + f"{server_url}/api/v1/message/{message_id}/sendmention", + json={"mentions": mentions}, + headers={"Password": password}, + ) + if resp.status_code >= 400: + raise classify_http_error(resp.status_code, resp.text) + except httpx.RequestError as e: + raise IMessageSendError(f"Failed to send mention: {e}") from e + + async def send_with_effect( + self, chat_guid: str, message_id: str, effect: str, + *, account_id: str | None = None, _account: dict | None = None, + ) -> None: + http = await self._get_http() + server_url, password = await self._resolve_server(account_id, _account) + valid_effects = {"slam", "loud", "gentle", "invisibleink"} + if effect not in valid_effects: + raise ValueError(f"Invalid effect: {effect}. Must be one of {valid_effects}") + try: + resp = await http.post( + f"{server_url}/api/v1/message/{message_id}/sendwitheffect", + json={"effect": effect}, + headers={"Password": password}, + ) + if resp.status_code >= 400: + raise classify_http_error(resp.status_code, resp.text) + except httpx.RequestError as e: + raise IMessageSendError(f"Failed to send effect: {e}") from e + + async def send_with_subject( + self, chat_guid: str, message_id: str, subject: str, + *, account_id: str | None = None, _account: dict | None = None, + ) -> None: + http = await self._get_http() + server_url, password = await self._resolve_server(account_id, _account) + try: + resp = await http.post( + f"{server_url}/api/v1/message/{message_id}/sendwithsubject", + json={"subject": subject}, + headers={"Password": password}, + ) + if resp.status_code >= 400: + raise classify_http_error(resp.status_code, resp.text) + except httpx.RequestError as e: + raise IMessageSendError(f"Failed to send subject: {e}") from e + + async def list_handles( + self, *, account_id: str | None = None, _account: dict | None = None, + ) -> list[dict]: + http = await self._get_http() + server_url, password = await self._resolve_server(account_id, _account) + resp = await http.get( + f"{server_url}/api/v1/handle", + headers={"Password": password}, + ) + if resp.status_code >= 400: + raise classify_http_error(resp.status_code, resp.text) + return resp.json().get("data", []) + + async def get_handle( + self, address: str, + *, account_id: str | None = None, _account: dict | None = None, + ) -> dict | None: + http = await self._get_http() + server_url, password = await self._resolve_server(account_id, _account) + resp = await http.get( + f"{server_url}/api/v1/handle/{address}", + headers={"Password": password}, + ) + if resp.status_code >= 400: + raise classify_http_error(resp.status_code, resp.text) + return resp.json() + + async def check_focus_status( + self, *, account_id: str | None = None, _account: dict | None = None, + ) -> dict: + http = await self._get_http() + server_url, password = await self._resolve_server(account_id, _account) + resp = await http.get( + f"{server_url}/api/v1/focus", + headers={"Password": password}, + ) + if resp.status_code >= 400: + raise classify_http_error(resp.status_code, resp.text) + return resp.json() + + async def check_facetime_status( + self, *, account_id: str | None = None, _account: dict | None = None, + ) -> dict: + http = await self._get_http() + server_url, password = await self._resolve_server(account_id, _account) + resp = await http.get( + f"{server_url}/api/v1/facetime", + headers={"Password": password}, + ) + if resp.status_code >= 400: + raise classify_http_error(resp.status_code, resp.text) + return resp.json() + + async def create_chat( + self, handles: list[str], + *, account_id: str | None = None, _account: dict | None = None, + ) -> dict | None: + http = await self._get_http() + server_url, password = await self._resolve_server(account_id, _account) + try: + resp = await http.post( + f"{server_url}/api/v1/chat/new", + json={"addresses": handles}, + headers={"Password": password}, + ) + if resp.status_code >= 400: + raise classify_http_error(resp.status_code, resp.text) + return resp.json() + except httpx.RequestError as e: + raise IMessageSendError(f"Failed to create chat: {e}") from e + + async def delete_chat( + self, chat_guid: str, + *, account_id: str | None = None, _account: dict | None = None, + ) -> bool: + http = await self._get_http() + server_url, password = await self._resolve_server(account_id, _account) + try: + resp = await http.delete( + f"{server_url}/api/v1/chat/{chat_guid}", + headers={"Password": password}, + ) + return resp.status_code < 400 + except httpx.RequestError as e: + raise IMessageSendError(f"Failed to delete chat: {e}") from e + + async def mark_chat_read( + self, chat_guid: str, + *, account_id: str | None = None, _account: dict | None = None, + ) -> bool: + http = await self._get_http() + server_url, password = await self._resolve_server(account_id, _account) + try: + resp = await http.post( + f"{server_url}/api/v1/chat/{chat_guid}/markread", + headers={"Password": password}, + ) + return resp.status_code < 400 + except httpx.RequestError as e: + logger.warning("Failed to mark chat read: %s", e) + return False + + async def mark_chat_unread( + self, chat_guid: str, + *, account_id: str | None = None, _account: dict | None = None, + ) -> bool: + http = await self._get_http() + server_url, password = await self._resolve_server(account_id, _account) + try: + resp = await http.post( + f"{server_url}/api/v1/chat/{chat_guid}/markunread", + headers={"Password": password}, + ) + return resp.status_code < 400 + except httpx.RequestError as e: + logger.warning("Failed to mark chat unread: %s", e) + return False + + async def force_notify( + self, message_id: str, + *, account_id: str | None = None, _account: dict | None = None, + ) -> bool: + http = await self._get_http() + server_url, password = await self._resolve_server(account_id, _account) + try: + resp = await http.post( + f"{server_url}/api/v1/message/{message_id}/forcenotify", + headers={"Password": password}, + ) + return resp.status_code < 400 + except httpx.RequestError as e: + raise IMessageSendError(f"Failed to force notify: {e}") from e + + async def set_group_icon( + self, chat_guid: str, icon_path: str, + *, account_id: str | None = None, _account: dict | None = None, + ) -> bool: + http = await self._get_http() + server_url, password = await self._resolve_server(account_id, _account) + try: + resp = await http.post( + f"{server_url}/api/v1/chat/{chat_guid}/seticon", + json={"path": icon_path}, + headers={"Password": password}, + ) + return resp.status_code < 400 + except httpx.RequestError as e: + raise IMessageSendError(f"Failed to set group icon: {e}") from e + + async def get_digital_touch( + self, message_id: str, + *, account_id: str | None = None, _account: dict | None = None, + ) -> dict | None: + http = await self._get_http() + server_url, password = await self._resolve_server(account_id, _account) + resp = await http.get( + f"{server_url}/api/v1/message/{message_id}/digitaltouch", + headers={"Password": password}, + ) + if resp.status_code >= 400: + raise classify_http_error(resp.status_code, resp.text) + return resp.json() + + async def send_card( + self, + target_id: str, + card_content: dict, + *, + reply_to_id: str | None = None, + thread_id: str | None = None, + account_id: str | None = None, + ) -> str | None: + return None + + async def edit_card( + self, + target_id: str, + message_id: str, + card_content: dict, + *, + thread_id: str | None = None, + account_id: str | None = None, + ) -> str | None: + return None + + async def send_payload(self, ctx: object) -> object: + return None + + async def send_poll(self, ctx: object) -> object: + return None + + async def render_presentation(self, payload: object, presentation: object, ctx: object) -> object | None: + return None + + async def pin_delivered_message(self, config: dict, target_ref: object, message_id: str, pin: object) -> None: + pass + + async def before_deliver_payload( + self, config: dict, target_ref: object, payload: object, hint: object | None = None + ) -> None: + pass + + async def after_deliver_payload(self, config: dict, target_ref: object, payload: object, results: list) -> None: + pass + + def resolve_target( + self, + to: str | None = None, + *, + config: dict | None = None, + allow_from: list[str] | None = None, + account_id: str | None = None, + mode: str | None = None, + ) -> tuple[bool, str]: + if not to: + return False, "Target is required" + return True, to + + def should_treat_delivered_text_as_visible(self, kind: str, text: str | None = None) -> bool: + return True + + async def _resolve_server(self, account_id: str | None = None, _account: dict | None = None) -> tuple[str, str]: + if _account: + server_url = _account.get("server_url", "") + password = _account.get("password", "") + if server_url and password: + return server_url, password + server_url = os.getenv("IMESSAGE_SERVER_URL", "http://localhost:1234") + password = os.getenv("IMESSAGE_SERVER_PASSWORD", "") + return server_url, password diff --git a/backend/package/yuxi/channel/extensions/imessage/pairing.py b/backend/package/yuxi/channel/extensions/imessage/pairing.py new file mode 100644 index 00000000..467255a0 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/imessage/pairing.py @@ -0,0 +1,68 @@ +from __future__ import annotations + +import secrets +import time +import logging + +logger = logging.getLogger(__name__) + +PAIRING_CODE_LENGTH = 8 +PAIRING_CODE_EXPIRY_SECONDS = 600 +PAIRING_CODE_CHARS = "ABCDEFGHJKMNPQRSTUVWXYZ23456789" # 无歧义: 排除 0/O/1/I + +PAIRING_APPROVED_MESSAGE = ( + "Your access has been approved. " + "You can now interact with this iMessage bot." +) + + +class IMessagePairingAdapter: + def __init__(self): + self._pending: dict[str, dict] = {} + self._approved: set[str] = set() + + def generate_code(self, peer_id: str) -> str: + self._cleanup_expired() + code = "".join(secrets.choice(PAIRING_CODE_CHARS) for _ in range(PAIRING_CODE_LENGTH)) + self._pending[peer_id] = { + "code": code, + "created_at": time.monotonic(), + "status": "pending", + } + return code + + def verify_code(self, peer_id: str, code: str) -> bool: + entry = self._pending.get(peer_id) + if not entry: + return False + + if time.monotonic() - entry["created_at"] > PAIRING_CODE_EXPIRY_SECONDS: + del self._pending[peer_id] + return False + + if secrets.compare_digest(entry["code"], code.upper().strip()): + entry["status"] = "approved" + self._approved.add(peer_id) + return True + + return False + + def is_approved(self, peer_id: str) -> bool: + return peer_id in self._approved + + def revoke(self, peer_id: str) -> None: + self._approved.discard(peer_id) + self._pending.pop(peer_id, None) + + def _cleanup_expired(self): + now = time.monotonic() + expired = [ + pid for pid, entry in self._pending.items() + if (now - entry["created_at"]) > PAIRING_CODE_EXPIRY_SECONDS + ] + for pid in expired: + del self._pending[pid] + + def pending_count(self) -> int: + self._cleanup_expired() + return sum(1 for v in self._pending.values() if v["status"] == "pending") diff --git a/backend/package/yuxi/channel/extensions/imessage/plugin.json b/backend/package/yuxi/channel/extensions/imessage/plugin.json new file mode 100644 index 00000000..44be4ea7 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/imessage/plugin.json @@ -0,0 +1,26 @@ +{ + "id": "imessage", + "name": "iMessage", + "version": "1.0.0", + "description": "Apple iMessage channel plugin for direct and group messaging with text, image, voice, file and video support", + "author": "ForcePilot", + "order": 90, + "enabled": true, + "capabilities": { + "chat_types": ["direct", "group"], + "message_types": ["text", "image", "voice", "file", "video"], + "reactions": true, + "typing_indicator": true, + "threads": false, + "edit": true, + "unsend": true, + "reply": true, + "media": true, + "native_commands": false, + "polls": false, + "streaming": false, + "block_streaming": false, + "block_streaming_chunk_min_chars": 2000, + "block_streaming_chunk_max_chars": 4000 + } +} diff --git a/backend/package/yuxi/channel/extensions/imessage/security.py b/backend/package/yuxi/channel/extensions/imessage/security.py new file mode 100644 index 00000000..568b9a58 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/imessage/security.py @@ -0,0 +1,111 @@ +from __future__ import annotations + +import logging + +logger = logging.getLogger(__name__) + + +class IMessageSecurityAdapter: + DM_POLICY_OPTIONS = ["pairing", "allowlist", "open", "disabled"] + GROUP_POLICY_OPTIONS = ["open", "allowlist", "disabled"] + + def __init__(self, config_adapter=None): + self._config_adapter = config_adapter + + def _get_channel_config(self, config: dict) -> dict: + if self._config_adapter: + return self._config_adapter._get_channel_config(config) + return config.get("channels", {}).get("imessage", {}) + + def resolve_dm_policy(self, config: dict | None = None) -> dict: + if config: + channel_config = self._get_channel_config(config) + return { + "mode": channel_config.get("dm_policy", "pairing"), + "allow_from": channel_config.get("allow_from", []), + } + return {"mode": "pairing", "allow_from": []} + + async def check_allowlist(self, peer_id: str, channel_type: str, config: dict | None = None) -> bool: + if not config: + return False + + channel_config = self._get_channel_config(config) + dm_policy = channel_config.get("dm_policy", "pairing") + + if dm_policy == "open": + return True + if dm_policy == "disabled": + return False + + allow_from = channel_config.get("allow_from", []) + if "*" in allow_from: + return True + if peer_id in allow_from: + return True + + if dm_policy == "pairing": + return False + + return False + + def resolve_group_policy(self, config: dict | None = None) -> dict: + if config: + channel_config = self._get_channel_config(config) + return { + "mode": channel_config.get("group_policy", "allowlist"), + "group_allow_from": channel_config.get("group_allow_from", []), + } + return {"mode": "allowlist", "group_allow_from": []} + + async def check_group_access(self, config: dict, chat_guid: str, sender_handle: str) -> bool: + channel_config = self._get_channel_config(config) + group_policy = channel_config.get("group_policy", "allowlist") + + if group_policy == "open": + return True + if group_policy == "disabled": + return False + + group_allow_from = channel_config.get("group_allow_from", []) + if sender_handle in group_allow_from: + return True + + groups = channel_config.get("groups", {}) + if chat_guid in groups: + return True + + return False + + def is_allowed_dm(self, peer_id: str, allow_from: list[str], policy: str) -> bool: + if policy == "open": + return True + if policy == "disabled": + return False + if policy == "allowlist": + return peer_id in allow_from or "*" in allow_from + if policy == "pairing": + return peer_id in allow_from + return False + + def is_allowed_group(self, group_id: str, group_allow_from: list[str], policy: str) -> bool: + if policy == "open": + return True + if policy == "disabled": + return False + if policy == "allowlist": + return group_id in group_allow_from + return False + + def is_valid_allow_entry(self, entry: str) -> bool: + if not entry or entry == "*": + return True + if "@" in entry: + return True + if entry.startswith("+"): + return True + if entry.startswith("chat_guid:"): + return len(entry) > len("chat_guid:") + if entry.startswith("chat_identifier:"): + return len(entry) > len("chat_identifier:") + return False diff --git a/backend/package/yuxi/channel/extensions/imessage/session.py b/backend/package/yuxi/channel/extensions/imessage/session.py new file mode 100644 index 00000000..407b46ec --- /dev/null +++ b/backend/package/yuxi/channel/extensions/imessage/session.py @@ -0,0 +1,25 @@ +from __future__ import annotations + +from yuxi.channel.extensions.imessage.targets import build_session_key +from yuxi.channel.message.models import PeerKind +from yuxi.channel.protocols import SessionResolution + + +class IMessageSessionAdapter: + def resolve_session(self, msg: object) -> SessionResolution: + if hasattr(msg, "sender") and hasattr(msg.sender, "kind"): + if msg.sender.kind == PeerKind.DIRECT: + sender_id = msg.sender.id if hasattr(msg.sender, "id") else "unknown" + return SessionResolution(kind="direct", conversation_id=sender_id) + + gid = "unknown" + if hasattr(msg, "group") and msg.group: + gid = msg.group.id or "unknown" + + return SessionResolution(kind="group", conversation_id=gid) + + def build_dm_session_key(self, account_id: str, handle: str) -> str: + return build_session_key("imessage", account_id, "direct", handle) + + def build_group_session_key(self, account_id: str, chat_guid: str) -> str: + return build_session_key("imessage", account_id, "group", chat_guid) diff --git a/backend/package/yuxi/channel/extensions/imessage/status.py b/backend/package/yuxi/channel/extensions/imessage/status.py new file mode 100644 index 00000000..dd86815a --- /dev/null +++ b/backend/package/yuxi/channel/extensions/imessage/status.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +import logging + +import httpx + +logger = logging.getLogger(__name__) + + +class IMessageStatusAdapter: + async def probe(self, account: dict | None = None) -> bool: + if not account: + return False + + server_url = account.get("server_url", "") + password = account.get("password", "") + if not server_url or not password: + return False + + try: + async with httpx.AsyncClient(timeout=httpx.Timeout(5.0)) as client: + resp = await client.get( + f"{server_url}/api/v1/server/info", + headers={"Password": password}, + ) + if resp.status_code == 200: + data = resp.json() + return data.get("data", {}).get("imessage_ready", False) + return False + except Exception as e: + logger.warning("iMessage probe failed: %s", e) + return False + + def build_summary(self, snapshot: object) -> dict: + return { + "channel": "imessage", + "mode": "bluebubbles", + } + + def resolve_account_state( + self, account: dict, config: dict, configured: bool, enabled: bool + ) -> str: + if not configured: + return "unconfigured" + if not enabled: + return "disabled" + return "ok" diff --git a/backend/package/yuxi/channel/extensions/imessage/targets.py b/backend/package/yuxi/channel/extensions/imessage/targets.py new file mode 100644 index 00000000..d74d6664 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/imessage/targets.py @@ -0,0 +1,55 @@ +from __future__ import annotations + +import re + +_E164_CLEAN = re.compile(r"[^\d+]") + + +def normalize_imessage_handle(handle: str) -> str: + handle = handle.strip() + for prefix in ("imessage:", "sms:", "auto:"): + if handle.lower().startswith(prefix): + handle = handle[len(prefix):] + if "@" in handle: + return handle.lower().strip() + return normalize_e164(handle) + + +def normalize_e164(handle: str) -> str: + cleaned = _E164_CLEAN.sub("", handle) + if not cleaned: + return handle + if not cleaned.startswith("+"): + cleaned = "+" + cleaned + return cleaned + + +def parse_target(target: str) -> tuple[str, str | None]: + """Parse a target string into (format, value).""" + target = target.strip() + if target.startswith("chat_guid:"): + return ("chat_guid", target[len("chat_guid:"):]) + if target.startswith("chat_identifier:"): + return ("chat_identifier", target[len("chat_identifier:"):]) + return ("handle", normalize_imessage_handle(target)) + + +def build_session_key( + channel_type: str, + account_id: str, + chat_type: str, + identifier: str, +) -> str: + return f"{channel_type}:{account_id}:{chat_type}:{identifier}" + + +def parse_session_key(session_key: str) -> dict | None: + parts = session_key.split(":") + if len(parts) < 4: + return None + return { + "channel_type": parts[0], + "account_id": parts[1], + "chat_type": parts[2], + "identifier": ":".join(parts[3:]), + } diff --git a/backend/package/yuxi/channel/extensions/imessage/types.py b/backend/package/yuxi/channel/extensions/imessage/types.py new file mode 100644 index 00000000..538699a3 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/imessage/types.py @@ -0,0 +1,92 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import StrEnum + + +class IMessageChatType(StrEnum): + DIRECT = "direct" + GROUP = "group" + + +class IMessageTargetFormat(StrEnum): + HANDLE = "handle" + CHAT_GUID = "chat_guid" + CHAT_IDENTIFIER = "chat_identifier" + + +@dataclass +class BlueBubblesChatInfo: + guid: str + chat_identifier: str | None = None + display_name: str | None = None + participants: list[str] = field(default_factory=list) + + +@dataclass +class BlueBubblesMessageEvent: + guid: str + text: str | None = None + subject: str | None = None + sender_handle: str | None = None + sender_name: str | None = None + chat_guid: str | None = None + chat_identifier: str | None = None + chat_display_name: str | None = None + is_from_me: bool = False + date_created: int = 0 + attachments: list[dict] = field(default_factory=list) + thread_originator_guid: str | None = None + group_action_type: int | None = None + + @classmethod + def from_json(cls, data: dict) -> BlueBubblesMessageEvent: + sender = data.get("sender") or {} + chat = data.get("chat") or {} + return cls( + guid=data.get("guid", ""), + text=data.get("text"), + subject=data.get("subject"), + sender_handle=( + ",".join(sender.get("handle", [])) + if isinstance(sender.get("handle"), list) + else sender.get("handle") + ), + sender_name=sender.get("displayName"), + chat_guid=chat.get("guid"), + chat_identifier=chat.get("chatIdentifier"), + chat_display_name=chat.get("displayName"), + is_from_me=bool(data.get("isFromMe", False)), + date_created=data.get("dateCreated", 0), + attachments=data.get("attachments", []), + thread_originator_guid=data.get("threadOriginatorGuid"), + group_action_type=data.get("groupActionType"), + ) + + +@dataclass +class ResolvedTarget: + target_id: str + chat_guid: str | None = None + chat_type: IMessageChatType = IMessageChatType.DIRECT + + +@dataclass +class IMessageGatewayContext: + server_url: str + password: str + account_id: str + ping_interval: int = 30 + ping_timeout: int = 10 + reconnect_min_delay: float = 1.0 + reconnect_max_delay: float = 60.0 + max_reconnect_attempts: int = 10 + + @property + def ws_url(self) -> str: + base = self.server_url.rstrip("/") + if base.startswith("https://"): + return f"wss://{base[8:]}/api/v1/ws" + elif base.startswith("http://"): + return f"ws://{base[7:]}/api/v1/ws" + return f"{base}/api/v1/ws"