mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-04-25 00:51:20 +00:00
d97f6cec7f
Adds Apple iMessage as a gateway platform via BlueBubbles macOS server. Architecture: - Webhook-based inbound (event-driven, no polling/dedup needed) - Email/phone → chat GUID resolution for user-friendly addressing - Private API safety (checks helper_connected before tapback/typing) - Inbound attachment downloading (images, audio, documents cached locally) - Markdown stripping for clean iMessage delivery - Smart progress suppression for platforms without message editing Based on PR #5869 by @benjaminsehl (webhook architecture, GUID resolution, Private API safety, progress suppression) with inbound attachment downloading from PR #4588 by @1960697431 (attachment cache routing). Integration points: Platform enum, env config, adapter factory, auth maps, cron delivery, send_message routing, channel directory, platform hints, toolset definition, setup wizard, status display. 27 tests covering config, adapter, webhook parsing, GUID resolution, attachment download routing, toolset consistency, and prompt hints.
828 lines
30 KiB
Python
828 lines
30 KiB
Python
"""BlueBubbles iMessage platform adapter.
|
|
|
|
Uses the local BlueBubbles macOS server for outbound REST sends and inbound
|
|
webhooks. Supports text messaging, media attachments (images, voice, video,
|
|
documents), tapback reactions, typing indicators, and read receipts.
|
|
|
|
Architecture based on PR #5869 (benjaminsehl) with inbound attachment
|
|
downloading from PR #4588 (YuhangLin).
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import uuid
|
|
from datetime import datetime
|
|
from typing import Any, Dict, List, Optional
|
|
from urllib.parse import quote
|
|
|
|
import httpx
|
|
|
|
from gateway.config import Platform, PlatformConfig
|
|
from gateway.platforms.base import (
|
|
BasePlatformAdapter,
|
|
MessageEvent,
|
|
MessageType,
|
|
SendResult,
|
|
cache_image_from_bytes,
|
|
cache_audio_from_bytes,
|
|
cache_document_from_bytes,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Constants
|
|
# ---------------------------------------------------------------------------
|
|
|
|
DEFAULT_WEBHOOK_HOST = "127.0.0.1"
|
|
DEFAULT_WEBHOOK_PORT = 8645
|
|
DEFAULT_WEBHOOK_PATH = "/bluebubbles-webhook"
|
|
MAX_TEXT_LENGTH = 4000
|
|
|
|
# Tapback reaction codes (BlueBubbles associatedMessageType values)
|
|
_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",
|
|
}
|
|
|
|
# Webhook event types that carry user messages
|
|
_MESSAGE_EVENTS = {"new-message", "message", "updated-message"}
|
|
|
|
# Log redaction patterns
|
|
_PHONE_RE = re.compile(r"\+?\d{7,15}")
|
|
_EMAIL_RE = re.compile(r"[\w.+-]+@[\w-]+\.[\w.]+")
|
|
|
|
|
|
def _redact(text: str) -> str:
|
|
"""Redact phone numbers and emails from log output."""
|
|
text = _PHONE_RE.sub("[REDACTED]", text)
|
|
text = _EMAIL_RE.sub("[REDACTED]", text)
|
|
return text
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def check_bluebubbles_requirements() -> bool:
|
|
try:
|
|
import aiohttp # noqa: F401
|
|
import httpx as _httpx # noqa: F401
|
|
except ImportError:
|
|
return False
|
|
return True
|
|
|
|
|
|
def _normalize_server_url(raw: str) -> str:
|
|
value = (raw or "").strip()
|
|
if not value:
|
|
return ""
|
|
if not re.match(r"^https?://", value, flags=re.I):
|
|
value = f"http://{value}"
|
|
return value.rstrip("/")
|
|
|
|
|
|
def _strip_markdown(text: str) -> str:
|
|
"""Strip common markdown formatting for iMessage plain-text delivery."""
|
|
text = re.sub(r"\*\*(.+?)\*\*", r"\1", text, flags=re.DOTALL)
|
|
text = re.sub(r"\*(.+?)\*", r"\1", text, flags=re.DOTALL)
|
|
text = re.sub(r"__(.+?)__", r"\1", text, flags=re.DOTALL)
|
|
text = re.sub(r"_(.+?)_", r"\1", text, flags=re.DOTALL)
|
|
text = re.sub(r"```[a-zA-Z0-9_+-]*\n?", "", text)
|
|
text = re.sub(r"`(.+?)`", r"\1", text)
|
|
text = re.sub(r"^#{1,6}\s+", "", text, flags=re.MULTILINE)
|
|
text = re.sub(r"\[([^\]]+)\]\(([^\)]+)\)", r"\1", text)
|
|
text = re.sub(r"\n{3,}", "\n\n", text)
|
|
return text.strip()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Adapter
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class BlueBubblesAdapter(BasePlatformAdapter):
|
|
platform = Platform.BLUEBUBBLES
|
|
MAX_MESSAGE_LENGTH = MAX_TEXT_LENGTH
|
|
|
|
def __init__(self, config: PlatformConfig):
|
|
super().__init__(config, Platform.BLUEBUBBLES)
|
|
extra = config.extra or {}
|
|
self.server_url = _normalize_server_url(
|
|
extra.get("server_url") or os.getenv("BLUEBUBBLES_SERVER_URL", "")
|
|
)
|
|
self.password = extra.get("password") or os.getenv("BLUEBUBBLES_PASSWORD", "")
|
|
self.webhook_host = (
|
|
extra.get("webhook_host")
|
|
or os.getenv("BLUEBUBBLES_WEBHOOK_HOST", DEFAULT_WEBHOOK_HOST)
|
|
)
|
|
self.webhook_port = int(
|
|
extra.get("webhook_port")
|
|
or os.getenv("BLUEBUBBLES_WEBHOOK_PORT", str(DEFAULT_WEBHOOK_PORT))
|
|
)
|
|
self.webhook_path = (
|
|
extra.get("webhook_path")
|
|
or os.getenv("BLUEBUBBLES_WEBHOOK_PATH", DEFAULT_WEBHOOK_PATH)
|
|
)
|
|
if not str(self.webhook_path).startswith("/"):
|
|
self.webhook_path = f"/{self.webhook_path}"
|
|
self.send_read_receipts = bool(extra.get("send_read_receipts", True))
|
|
self.client: Optional[httpx.AsyncClient] = None
|
|
self._runner = None
|
|
self._private_api_enabled: Optional[bool] = None
|
|
self._helper_connected: bool = False
|
|
self._guid_cache: Dict[str, str] = {}
|
|
|
|
# ------------------------------------------------------------------
|
|
# API helpers
|
|
# ------------------------------------------------------------------
|
|
|
|
def _api_url(self, path: str) -> str:
|
|
sep = "&" if "?" in path else "?"
|
|
return f"{self.server_url}{path}{sep}password={quote(self.password, safe='')}"
|
|
|
|
async def _api_get(self, path: str) -> Dict[str, Any]:
|
|
assert self.client is not None
|
|
res = await self.client.get(self._api_url(path))
|
|
res.raise_for_status()
|
|
return res.json()
|
|
|
|
async def _api_post(self, path: str, payload: Dict[str, Any]) -> Dict[str, Any]:
|
|
assert self.client is not None
|
|
res = await self.client.post(self._api_url(path), json=payload)
|
|
res.raise_for_status()
|
|
return res.json()
|
|
|
|
# ------------------------------------------------------------------
|
|
# Lifecycle
|
|
# ------------------------------------------------------------------
|
|
|
|
async def connect(self) -> bool:
|
|
if not self.server_url or not self.password:
|
|
logger.error(
|
|
"[bluebubbles] BLUEBUBBLES_SERVER_URL and BLUEBUBBLES_PASSWORD are required"
|
|
)
|
|
return False
|
|
from aiohttp import web
|
|
|
|
self.client = httpx.AsyncClient(timeout=30.0)
|
|
try:
|
|
await self._api_get("/api/v1/ping")
|
|
info = await self._api_get("/api/v1/server/info")
|
|
server_data = (info or {}).get("data", {})
|
|
self._private_api_enabled = bool(server_data.get("private_api"))
|
|
self._helper_connected = bool(server_data.get("helper_connected"))
|
|
logger.info(
|
|
"[bluebubbles] connected to %s (private_api=%s, helper=%s)",
|
|
self.server_url,
|
|
self._private_api_enabled,
|
|
self._helper_connected,
|
|
)
|
|
except Exception as exc:
|
|
logger.error(
|
|
"[bluebubbles] cannot reach server at %s: %s", self.server_url, exc
|
|
)
|
|
if self.client:
|
|
await self.client.aclose()
|
|
self.client = None
|
|
return False
|
|
|
|
app = web.Application()
|
|
app.router.add_get("/health", lambda _: web.Response(text="ok"))
|
|
app.router.add_post(self.webhook_path, self._handle_webhook)
|
|
self._runner = web.AppRunner(app)
|
|
await self._runner.setup()
|
|
site = web.TCPSite(self._runner, self.webhook_host, self.webhook_port)
|
|
await site.start()
|
|
self._mark_connected()
|
|
logger.info(
|
|
"[bluebubbles] webhook listening on http://%s:%s%s",
|
|
self.webhook_host,
|
|
self.webhook_port,
|
|
self.webhook_path,
|
|
)
|
|
return True
|
|
|
|
async def disconnect(self) -> None:
|
|
if self.client:
|
|
await self.client.aclose()
|
|
self.client = None
|
|
if self._runner:
|
|
await self._runner.cleanup()
|
|
self._runner = None
|
|
self._mark_disconnected()
|
|
|
|
# ------------------------------------------------------------------
|
|
# Chat GUID resolution
|
|
# ------------------------------------------------------------------
|
|
|
|
async def _resolve_chat_guid(self, target: str) -> Optional[str]:
|
|
"""Resolve an email/phone to a BlueBubbles chat GUID.
|
|
|
|
If *target* already contains a semicolon (raw GUID format like
|
|
``iMessage;-;user@example.com``), it is returned as-is. Otherwise
|
|
the adapter queries the BlueBubbles chat list and matches on
|
|
``chatIdentifier`` or participant address.
|
|
"""
|
|
target = (target or "").strip()
|
|
if not target:
|
|
return None
|
|
# Already a raw GUID
|
|
if ";" in target:
|
|
return target
|
|
if target in self._guid_cache:
|
|
return self._guid_cache[target]
|
|
try:
|
|
payload = await self._api_post(
|
|
"/api/v1/chat/query",
|
|
{"limit": 100, "offset": 0, "with": ["participants"]},
|
|
)
|
|
for chat in payload.get("data", []) or []:
|
|
guid = chat.get("guid") or chat.get("chatGuid")
|
|
identifier = chat.get("chatIdentifier") or chat.get("identifier")
|
|
if identifier == target:
|
|
if guid:
|
|
self._guid_cache[target] = guid
|
|
return guid
|
|
for part in chat.get("participants", []) or []:
|
|
if (part.get("address") or "").strip() == target and guid:
|
|
self._guid_cache[target] = guid
|
|
return guid
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
async def _create_chat_for_handle(
|
|
self, address: str, message: str
|
|
) -> SendResult:
|
|
"""Create a new chat by sending the first message to *address*."""
|
|
payload = {
|
|
"addresses": [address],
|
|
"message": message,
|
|
"tempGuid": f"temp-{datetime.utcnow().timestamp()}",
|
|
}
|
|
try:
|
|
res = await self._api_post("/api/v1/chat/new", payload)
|
|
data = res.get("data") or {}
|
|
msg_id = data.get("guid") or data.get("messageGuid") or "ok"
|
|
return SendResult(success=True, message_id=str(msg_id), raw_response=res)
|
|
except Exception as exc:
|
|
return SendResult(success=False, error=str(exc))
|
|
|
|
# ------------------------------------------------------------------
|
|
# Text sending
|
|
# ------------------------------------------------------------------
|
|
|
|
async def send(
|
|
self,
|
|
chat_id: str,
|
|
content: str,
|
|
reply_to: Optional[str] = None,
|
|
metadata: Optional[Dict[str, Any]] = None,
|
|
) -> SendResult:
|
|
text = _strip_markdown(content or "")
|
|
if not text:
|
|
return SendResult(success=False, error="BlueBubbles send requires text")
|
|
chunks = self.truncate_message(text, max_length=self.MAX_MESSAGE_LENGTH)
|
|
last = SendResult(success=True)
|
|
for chunk in chunks:
|
|
guid = await self._resolve_chat_guid(chat_id)
|
|
if not guid:
|
|
# If the target looks like an address, try creating a new chat
|
|
if self._private_api_enabled and (
|
|
"@" in chat_id or re.match(r"^\+\d+", chat_id)
|
|
):
|
|
return await self._create_chat_for_handle(chat_id, chunk)
|
|
return SendResult(
|
|
success=False,
|
|
error=f"BlueBubbles chat not found for target: {chat_id}",
|
|
)
|
|
payload: Dict[str, Any] = {
|
|
"chatGuid": guid,
|
|
"tempGuid": f"temp-{datetime.utcnow().timestamp()}",
|
|
"message": chunk,
|
|
}
|
|
if reply_to and self._private_api_enabled and self._helper_connected:
|
|
payload["method"] = "private-api"
|
|
payload["selectedMessageGuid"] = reply_to
|
|
payload["partIndex"] = 0
|
|
try:
|
|
res = await self._api_post("/api/v1/message/text", payload)
|
|
data = res.get("data") or {}
|
|
msg_id = data.get("guid") or data.get("messageGuid") or "ok"
|
|
last = SendResult(
|
|
success=True, message_id=str(msg_id), raw_response=res
|
|
)
|
|
except Exception as exc:
|
|
return SendResult(success=False, error=str(exc))
|
|
return last
|
|
|
|
# ------------------------------------------------------------------
|
|
# Media sending (outbound)
|
|
# ------------------------------------------------------------------
|
|
|
|
async def _send_attachment(
|
|
self,
|
|
chat_id: str,
|
|
file_path: str,
|
|
filename: Optional[str] = None,
|
|
caption: Optional[str] = None,
|
|
is_audio_message: bool = False,
|
|
) -> SendResult:
|
|
"""Send a file attachment via BlueBubbles multipart upload."""
|
|
if not self.client:
|
|
return SendResult(success=False, error="Not connected")
|
|
if not os.path.isfile(file_path):
|
|
return SendResult(success=False, error=f"File not found: {file_path}")
|
|
|
|
guid = await self._resolve_chat_guid(chat_id)
|
|
if not guid:
|
|
return SendResult(success=False, error=f"Chat not found: {chat_id}")
|
|
|
|
fname = filename or os.path.basename(file_path)
|
|
try:
|
|
with open(file_path, "rb") as f:
|
|
files = {"attachment": (fname, f, "application/octet-stream")}
|
|
data: Dict[str, str] = {
|
|
"chatGuid": guid,
|
|
"name": fname,
|
|
"tempGuid": uuid.uuid4().hex,
|
|
}
|
|
if is_audio_message:
|
|
data["isAudioMessage"] = "true"
|
|
res = await self.client.post(
|
|
self._api_url("/api/v1/message/attachment"),
|
|
files=files,
|
|
data=data,
|
|
timeout=120,
|
|
)
|
|
res.raise_for_status()
|
|
result = res.json()
|
|
|
|
if caption:
|
|
await self.send(chat_id, caption)
|
|
|
|
if result.get("status") == 200:
|
|
rdata = result.get("data") or {}
|
|
msg_id = rdata.get("guid") if isinstance(rdata, dict) else None
|
|
return SendResult(
|
|
success=True, message_id=msg_id, raw_response=result
|
|
)
|
|
return SendResult(
|
|
success=False,
|
|
error=result.get("message", "Attachment upload failed"),
|
|
)
|
|
except Exception as e:
|
|
return SendResult(success=False, error=str(e))
|
|
|
|
async def send_image(
|
|
self,
|
|
chat_id: str,
|
|
image_url: str,
|
|
caption: Optional[str] = None,
|
|
reply_to: Optional[str] = None,
|
|
metadata: Optional[Dict[str, Any]] = None,
|
|
) -> SendResult:
|
|
try:
|
|
from gateway.platforms.base import cache_image_from_url
|
|
|
|
local_path = await cache_image_from_url(image_url)
|
|
return await self._send_attachment(chat_id, local_path, caption=caption)
|
|
except Exception:
|
|
return await super().send_image(chat_id, image_url, caption, reply_to)
|
|
|
|
async def send_image_file(
|
|
self,
|
|
chat_id: str,
|
|
image_path: str,
|
|
caption: Optional[str] = None,
|
|
reply_to: Optional[str] = None,
|
|
**kwargs,
|
|
) -> SendResult:
|
|
return await self._send_attachment(chat_id, image_path, caption=caption)
|
|
|
|
async def send_voice(
|
|
self,
|
|
chat_id: str,
|
|
audio_path: str,
|
|
caption: Optional[str] = None,
|
|
reply_to: Optional[str] = None,
|
|
**kwargs,
|
|
) -> SendResult:
|
|
return await self._send_attachment(
|
|
chat_id, audio_path, caption=caption, is_audio_message=True
|
|
)
|
|
|
|
async def send_video(
|
|
self,
|
|
chat_id: str,
|
|
video_path: str,
|
|
caption: Optional[str] = None,
|
|
reply_to: Optional[str] = None,
|
|
**kwargs,
|
|
) -> SendResult:
|
|
return await self._send_attachment(chat_id, video_path, caption=caption)
|
|
|
|
async def send_document(
|
|
self,
|
|
chat_id: str,
|
|
file_path: str,
|
|
caption: Optional[str] = None,
|
|
file_name: Optional[str] = None,
|
|
reply_to: Optional[str] = None,
|
|
**kwargs,
|
|
) -> SendResult:
|
|
return await self._send_attachment(
|
|
chat_id, file_path, filename=file_name, caption=caption
|
|
)
|
|
|
|
async def send_animation(
|
|
self,
|
|
chat_id: str,
|
|
animation_url: str,
|
|
caption: Optional[str] = None,
|
|
reply_to: Optional[str] = None,
|
|
metadata: Optional[Dict[str, Any]] = None,
|
|
) -> SendResult:
|
|
return await self.send_image(
|
|
chat_id, animation_url, caption, reply_to, metadata
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Typing indicators
|
|
# ------------------------------------------------------------------
|
|
|
|
async def send_typing(self, chat_id: str, metadata=None) -> None:
|
|
if not self._private_api_enabled or not self._helper_connected or not self.client:
|
|
return
|
|
try:
|
|
guid = await self._resolve_chat_guid(chat_id)
|
|
if guid:
|
|
encoded = quote(guid, safe="")
|
|
await self.client.post(
|
|
self._api_url(f"/api/v1/chat/{encoded}/typing"), timeout=5
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
async def stop_typing(self, chat_id: str) -> None:
|
|
if not self._private_api_enabled or not self._helper_connected or not self.client:
|
|
return
|
|
try:
|
|
guid = await self._resolve_chat_guid(chat_id)
|
|
if guid:
|
|
encoded = quote(guid, safe="")
|
|
await self.client.delete(
|
|
self._api_url(f"/api/v1/chat/{encoded}/typing"), timeout=5
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
# ------------------------------------------------------------------
|
|
# Read receipts
|
|
# ------------------------------------------------------------------
|
|
|
|
async def mark_read(self, chat_id: str) -> bool:
|
|
if not self._private_api_enabled or not self._helper_connected or not self.client:
|
|
return False
|
|
try:
|
|
guid = await self._resolve_chat_guid(chat_id)
|
|
if guid:
|
|
encoded = quote(guid, safe="")
|
|
await self.client.post(
|
|
self._api_url(f"/api/v1/chat/{encoded}/read"), timeout=5
|
|
)
|
|
return True
|
|
except Exception:
|
|
pass
|
|
return False
|
|
|
|
# ------------------------------------------------------------------
|
|
# Tapback reactions
|
|
# ------------------------------------------------------------------
|
|
|
|
async def send_reaction(
|
|
self,
|
|
chat_id: str,
|
|
message_guid: str,
|
|
reaction: str,
|
|
part_index: int = 0,
|
|
) -> SendResult:
|
|
"""Send a tapback reaction (requires Private API helper)."""
|
|
if not self._private_api_enabled or not self._helper_connected:
|
|
return SendResult(
|
|
success=False, error="Private API helper not connected"
|
|
)
|
|
guid = await self._resolve_chat_guid(chat_id)
|
|
if not guid:
|
|
return SendResult(success=False, error=f"Chat not found: {chat_id}")
|
|
try:
|
|
res = await self._api_post(
|
|
"/api/v1/message/react",
|
|
{
|
|
"chatGuid": guid,
|
|
"selectedMessageGuid": message_guid,
|
|
"reaction": reaction,
|
|
"partIndex": part_index,
|
|
},
|
|
)
|
|
return SendResult(success=True, raw_response=res)
|
|
except Exception as exc:
|
|
return SendResult(success=False, error=str(exc))
|
|
|
|
# ------------------------------------------------------------------
|
|
# Chat info
|
|
# ------------------------------------------------------------------
|
|
|
|
async def get_chat_info(self, chat_id: str) -> Dict[str, Any]:
|
|
is_group = ";+;" in (chat_id or "")
|
|
info: Dict[str, Any] = {
|
|
"name": chat_id,
|
|
"type": "group" if is_group else "dm",
|
|
}
|
|
try:
|
|
guid = await self._resolve_chat_guid(chat_id)
|
|
if guid:
|
|
encoded = quote(guid, safe="")
|
|
res = await self._api_get(
|
|
f"/api/v1/chat/{encoded}?with=participants"
|
|
)
|
|
data = (res or {}).get("data", {})
|
|
display_name = (
|
|
data.get("displayName")
|
|
or data.get("chatIdentifier")
|
|
or chat_id
|
|
)
|
|
participants = []
|
|
for p in data.get("participants", []) or []:
|
|
addr = (p.get("address") or "").strip()
|
|
if addr:
|
|
participants.append(addr)
|
|
info["name"] = display_name
|
|
if participants:
|
|
info["participants"] = participants
|
|
except Exception:
|
|
pass
|
|
return info
|
|
|
|
def format_message(self, content: str) -> str:
|
|
return _strip_markdown(content)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Inbound attachment downloading (from #4588)
|
|
# ------------------------------------------------------------------
|
|
|
|
async def _download_attachment(
|
|
self, att_guid: str, att_meta: Dict[str, Any]
|
|
) -> Optional[str]:
|
|
"""Download an attachment from BlueBubbles and cache it locally.
|
|
|
|
Returns the local file path on success, None on failure.
|
|
"""
|
|
if not self.client:
|
|
return None
|
|
try:
|
|
encoded = quote(att_guid, safe="")
|
|
resp = await self.client.get(
|
|
self._api_url(f"/api/v1/attachment/{encoded}/download"),
|
|
timeout=60,
|
|
follow_redirects=True,
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.content
|
|
|
|
mime = (att_meta.get("mimeType") or "").lower()
|
|
transfer_name = att_meta.get("transferName", "")
|
|
|
|
if mime.startswith("image/"):
|
|
ext_map = {
|
|
"image/jpeg": ".jpg",
|
|
"image/png": ".png",
|
|
"image/gif": ".gif",
|
|
"image/webp": ".webp",
|
|
"image/heic": ".jpg",
|
|
"image/heif": ".jpg",
|
|
"image/tiff": ".jpg",
|
|
}
|
|
ext = ext_map.get(mime, ".jpg")
|
|
return cache_image_from_bytes(data, ext)
|
|
|
|
if mime.startswith("audio/"):
|
|
ext_map = {
|
|
"audio/mp3": ".mp3",
|
|
"audio/mpeg": ".mp3",
|
|
"audio/ogg": ".ogg",
|
|
"audio/wav": ".wav",
|
|
"audio/x-caf": ".mp3",
|
|
"audio/mp4": ".m4a",
|
|
"audio/aac": ".m4a",
|
|
}
|
|
ext = ext_map.get(mime, ".mp3")
|
|
return cache_audio_from_bytes(data, ext)
|
|
|
|
# Videos, documents, and everything else
|
|
filename = transfer_name or f"file_{uuid.uuid4().hex[:8]}"
|
|
return cache_document_from_bytes(data, filename)
|
|
|
|
except Exception as exc:
|
|
logger.warning(
|
|
"[bluebubbles] failed to download attachment %s: %s",
|
|
_redact(att_guid),
|
|
exc,
|
|
)
|
|
return None
|
|
|
|
# ------------------------------------------------------------------
|
|
# Webhook handling
|
|
# ------------------------------------------------------------------
|
|
|
|
def _extract_payload_record(
|
|
self, payload: Dict[str, Any]
|
|
) -> Optional[Dict[str, Any]]:
|
|
data = payload.get("data")
|
|
if isinstance(data, dict):
|
|
return data
|
|
if isinstance(data, list):
|
|
for item in data:
|
|
if isinstance(item, dict):
|
|
return item
|
|
if isinstance(payload.get("message"), dict):
|
|
return payload.get("message")
|
|
return payload if isinstance(payload, dict) else None
|
|
|
|
@staticmethod
|
|
def _value(*candidates: Any) -> Optional[str]:
|
|
for candidate in candidates:
|
|
if isinstance(candidate, str) and candidate.strip():
|
|
return candidate.strip()
|
|
return None
|
|
|
|
async def _handle_webhook(self, request):
|
|
from aiohttp import web
|
|
|
|
token = (
|
|
request.query.get("password")
|
|
or request.query.get("guid")
|
|
or request.headers.get("x-password")
|
|
or request.headers.get("x-guid")
|
|
or request.headers.get("x-bluebubbles-guid")
|
|
)
|
|
if token != self.password:
|
|
return web.json_response({"error": "unauthorized"}, status=401)
|
|
try:
|
|
raw = await request.read()
|
|
body = raw.decode("utf-8", errors="replace")
|
|
try:
|
|
payload = json.loads(body)
|
|
except Exception:
|
|
from urllib.parse import parse_qs
|
|
|
|
form = parse_qs(body)
|
|
payload_str = (
|
|
form.get("payload")
|
|
or form.get("data")
|
|
or form.get("message")
|
|
or [""]
|
|
)[0]
|
|
payload = json.loads(payload_str) if payload_str else {}
|
|
except Exception as exc:
|
|
logger.error("[bluebubbles] webhook parse error: %s", exc)
|
|
return web.json_response({"error": "invalid payload"}, status=400)
|
|
|
|
event_type = self._value(payload.get("type"), payload.get("event")) or ""
|
|
# Only process message events; silently acknowledge everything else
|
|
if event_type and event_type not in _MESSAGE_EVENTS:
|
|
return web.Response(text="ok")
|
|
|
|
record = self._extract_payload_record(payload) or {}
|
|
is_from_me = bool(
|
|
record.get("isFromMe")
|
|
or record.get("fromMe")
|
|
or record.get("is_from_me")
|
|
)
|
|
if is_from_me:
|
|
return web.Response(text="ok")
|
|
|
|
# Skip tapback reactions delivered as messages
|
|
assoc_type = record.get("associatedMessageType")
|
|
if isinstance(assoc_type, int) and assoc_type in {
|
|
**_TAPBACK_ADDED,
|
|
**_TAPBACK_REMOVED,
|
|
}:
|
|
return web.Response(text="ok")
|
|
|
|
text = (
|
|
self._value(
|
|
record.get("text"), record.get("message"), record.get("body")
|
|
)
|
|
or ""
|
|
)
|
|
|
|
# --- Inbound attachment handling ---
|
|
attachments = record.get("attachments") or []
|
|
media_urls: List[str] = []
|
|
media_types: List[str] = []
|
|
msg_type = MessageType.TEXT
|
|
|
|
for att in attachments:
|
|
att_guid = att.get("guid", "")
|
|
if not att_guid:
|
|
continue
|
|
cached = await self._download_attachment(att_guid, att)
|
|
if cached:
|
|
mime = (att.get("mimeType") or "").lower()
|
|
media_urls.append(cached)
|
|
media_types.append(mime)
|
|
if mime.startswith("image/"):
|
|
msg_type = MessageType.PHOTO
|
|
elif mime.startswith("audio/") or (att.get("uti") or "").endswith(
|
|
"caf"
|
|
):
|
|
msg_type = MessageType.VOICE
|
|
elif mime.startswith("video/"):
|
|
msg_type = MessageType.VIDEO
|
|
else:
|
|
msg_type = MessageType.DOCUMENT
|
|
|
|
# With multiple attachments, prefer PHOTO if any images present
|
|
if len(media_urls) > 1:
|
|
mime_prefixes = {(m or "").split("/")[0] for m in media_types}
|
|
if "image" in mime_prefixes:
|
|
msg_type = MessageType.PHOTO
|
|
|
|
if not text and media_urls:
|
|
text = "(attachment)"
|
|
# --- End attachment handling ---
|
|
|
|
chat_guid = self._value(
|
|
record.get("chatGuid"),
|
|
payload.get("chatGuid"),
|
|
record.get("chat_guid"),
|
|
payload.get("chat_guid"),
|
|
payload.get("guid"),
|
|
)
|
|
chat_identifier = self._value(
|
|
record.get("chatIdentifier"),
|
|
record.get("identifier"),
|
|
payload.get("chatIdentifier"),
|
|
payload.get("identifier"),
|
|
)
|
|
sender = (
|
|
self._value(
|
|
record.get("handle", {}).get("address")
|
|
if isinstance(record.get("handle"), dict)
|
|
else None,
|
|
record.get("sender"),
|
|
record.get("from"),
|
|
record.get("address"),
|
|
)
|
|
or chat_identifier
|
|
or chat_guid
|
|
)
|
|
if not (chat_guid or chat_identifier) and sender:
|
|
chat_identifier = sender
|
|
if not sender or not (chat_guid or chat_identifier) or not text:
|
|
return web.json_response({"error": "missing message fields"}, status=400)
|
|
|
|
session_chat_id = chat_guid or chat_identifier
|
|
is_group = bool(record.get("isGroup")) or (";+;" in (chat_guid or ""))
|
|
source = self.build_source(
|
|
chat_id=session_chat_id,
|
|
chat_name=chat_identifier or sender,
|
|
chat_type="group" if is_group else "dm",
|
|
user_id=sender,
|
|
user_name=sender,
|
|
chat_id_alt=chat_identifier,
|
|
)
|
|
event = MessageEvent(
|
|
text=text,
|
|
message_type=msg_type,
|
|
source=source,
|
|
raw_message=payload,
|
|
message_id=self._value(
|
|
record.get("guid"),
|
|
record.get("messageGuid"),
|
|
record.get("id"),
|
|
),
|
|
reply_to_message_id=self._value(
|
|
record.get("threadOriginatorGuid"),
|
|
record.get("associatedMessageGuid"),
|
|
),
|
|
media_urls=media_urls,
|
|
media_types=media_types,
|
|
)
|
|
task = asyncio.create_task(self.handle_message(event))
|
|
self._background_tasks.add(task)
|
|
task.add_done_callback(self._background_tasks.discard)
|
|
|
|
# Fire-and-forget read receipt
|
|
if self.send_read_receipts and session_chat_id:
|
|
asyncio.create_task(self.mark_read(session_chat_id))
|
|
|
|
return web.Response(text="ok")
|