kris/ForcePilot
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat(imessage): 新增iMessage渠道插件完整实现

Browse Source 
该提交新增了基于BlueBubbles的iMessage渠道插件，支持单聊和群组消息，包含文本、图片、语音、文件和视频消息收发，支持消息编辑、撤回、回复、 reactions和输入状态提示，同时实现了账号配置、安全校验、配对授权、消息格式化与分片等完整功能。
This commit is contained in:
Kris 2026-05-21 10:50:15 +08:00
parent 432d5db970
commit 4d52f634fe
16 changed files with 2209 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

285
backend/package/yuxi/channel/extensions/imessage/__init__.py Normal file
View File

@ -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)

153
backend/package/yuxi/channel/extensions/imessage/config.py Normal file
View File

@ -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},
},
},
},
}

38
backend/package/yuxi/channel/extensions/imessage/echo.py Normal file
View File

@ -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]

35
backend/package/yuxi/channel/extensions/imessage/errors.py Normal file
View File

@ -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

77
backend/package/yuxi/channel/extensions/imessage/format.py Normal file
View File

@ -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)

148
backend/package/yuxi/channel/extensions/imessage/gateway.py Normal file
View File

@ -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

104
backend/package/yuxi/channel/extensions/imessage/media.py Normal file
View File

@ -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")

265
backend/package/yuxi/channel/extensions/imessage/monitor.py Normal file
View File

@ -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)

680
backend/package/yuxi/channel/extensions/imessage/outbound.py Normal file
View File

@ -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

68
backend/package/yuxi/channel/extensions/imessage/pairing.py Normal file
View File

@ -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")

26
backend/package/yuxi/channel/extensions/imessage/plugin.json Normal file
View File

@ -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
}
}

111
backend/package/yuxi/channel/extensions/imessage/security.py Normal file
View File

@ -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

25
backend/package/yuxi/channel/extensions/imessage/session.py Normal file
View File

@ -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)

47
backend/package/yuxi/channel/extensions/imessage/status.py Normal file
View File

@ -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"

55
backend/package/yuxi/channel/extensions/imessage/targets.py Normal file
View File

@ -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:]),
}

92
backend/package/yuxi/channel/extensions/imessage/types.py Normal file
View File

@ -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"