from __future__ import annotations import re from typing import Any _SHIP_PATTERN = re.compile(r"^~[a-z]{3,}(-[a-z]{3,})*$", re.IGNORECASE) def is_valid_ship(ship: str) -> bool: return bool(_SHIP_PATTERN.match(ship)) def strip_ship_tilde(ship: str) -> str: return ship.lstrip("~") def ensure_ship_tilde(ship: str) -> str: return ship if ship.startswith("~") else f"~{ship}" def build_dm_channel_id(target_ship: str, sender_ship: str) -> str: t = strip_ship_tilde(target_ship) s = strip_ship_tilde(sender_ship) return f"~{t}/dm--{s}" def build_group_channel_id(host_ship: str, group_name: str, channel_type: str = "chat") -> str: h = strip_ship_tilde(host_ship) return f"~{h}/{group_name}/{channel_type}" def build_channel_path(chat_type: str, **kwargs) -> str: if chat_type == "direct": target = strip_ship_tilde(kwargs["target_ship"]) return f"/dm/~{target}" group_name = kwargs["group_name"] ch_type = kwargs.get("channel_type", "chat") return f"/{ch_type}/{group_name}" def parse_graph_update(raw: dict[str, Any]) -> dict[str, Any]: graph_update = raw.get("graph-update", {}) if not isinstance(graph_update, dict) or not graph_update: return {} result: dict[str, Any] = {} resource = graph_update.get("resource", {}) result["resource_path"] = resource.get("path", "") result["resource_type"] = resource.get("type", "chat") result["ship"] = graph_update.get("ship", "") result["index"] = graph_update.get("index", "") result["time"] = graph_update.get("time") additions = graph_update.get("additions", {}) if additions: result["content"] = _extract_graph_content(additions) if "reaction" in graph_update: result["reaction"] = graph_update["reaction"] return result def _extract_graph_content(additions: dict[str, Any]) -> str: parts: list[str] = [] for node_data in additions.values(): post = node_data.get("post", {}) contents = post.get("contents", []) for item in contents: if "text" in item: parts.append(item["text"]) elif "mention" in item: parts.append(f"~{item['mention']}") elif "url" in item: parts.append(item["url"]) elif "code" in item: code_block = item["code"] lang = code_block.get("lang", "") code_text = code_block.get("code", "") if isinstance(code_block, dict) else str(code_block) if lang: parts.append(f"```{lang}\n{code_text}\n```\n") else: parts.append(f"```\n{code_text}\n```\n") elif "blockquote" in item: quoted = item["blockquote"] if isinstance(quoted, list): quoted_text = _extract_content_list(quoted) parts.append(f"> {quoted_text}\n") else: parts.append(f"> {quoted}\n") elif "image" in item: img = item["image"] img_url = img.get("src", "") if isinstance(img, dict) else str(img) if img_url: parts.append(f"[Image: {img_url}]") elif "ship" in item: parts.append(f"~{item['ship']}") elif isinstance(item, str): parts.append(item) return "".join(parts) def _extract_content_list(contents: list[Any]) -> str: parts: list[str] = [] for item in contents: if isinstance(item, dict): if "text" in item: parts.append(item["text"]) elif "mention" in item: parts.append(f"~{item['mention']}") elif "url" in item: parts.append(item["url"]) elif "code" in item: code_block = item["code"] code_text = code_block.get("code", "") if isinstance(code_block, dict) else str(code_block) parts.append(code_text) elif "ship" in item: parts.append(f"~{item['ship']}") elif isinstance(item, str): parts.append(item) return "".join(parts) def extract_graph_mentions(additions: dict[str, Any]) -> list[str]: mentions: list[str] = [] for node_data in additions.values(): post = node_data.get("post", {}) contents = post.get("contents", []) for item in contents: if "mention" in item: mentions.append(item["mention"]) return mentions def extract_graph_attachments(additions: dict[str, Any]) -> list[dict[str, Any]]: from yuxi.channels.models import Attachment attachments: list[dict[str, Any]] = [] for node_data in additions.values(): post = node_data.get("post", {}) contents = post.get("contents", []) for item in contents: if "image" in item: img = item["image"] src = img.get("src", "") if isinstance(img, dict) else str(img) if src: attachment = Attachment( type="image", url=src, mime_type="image/*", ) attachments.append(attachment.model_dump()) elif "file" in item: file_info = item["file"] if isinstance(file_info, dict): attachment = Attachment( type="file", url=file_info.get("src", ""), filename=file_info.get("name"), size_bytes=file_info.get("size"), ) attachments.append(attachment.model_dump()) return attachments def build_poke_payload( host_ship: str, content: str, channel_path: str, *, continuation: bool = False, reply_to: str | None = None, media_content: list[dict[str, Any]] | None = None, chat_type: str = "group", ) -> dict[str, Any]: _MAX_CONTENT_LENGTH = 100_000 if len(content) > _MAX_CONTENT_LENGTH: raise ValueError(f"Content length {len(content)} exceeds maximum {_MAX_CONTENT_LENGTH}") mark = "chat-dm-action" if chat_type == "direct" else "chat-message" if media_content: content_list = media_content else: from .story import markdown_to_story story = markdown_to_story(content) content_list = story payload: dict[str, Any] = { "action": "poke", "ship": host_ship, "app": "chat", "mark": mark, "json": { "path": channel_path, "envelope": { "content": content_list, "author": f"~{host_ship}", }, }, } if continuation: payload["json"]["envelope"]["continuation"] = True if reply_to: payload["json"]["envelope"]["reply_to"] = reply_to return payload def build_chat_action_payload( host_ship: str, channel_path: str, action: str, msg_id: str, *, content: str | None = None, emoji: str | None = None, ) -> dict[str, Any]: action_data: dict[str, Any] = {} match action: case "add-react": action_data = {"add-react": {"react": emoji, "msg": msg_id}} case "del-react": action_data = {"del-react": {"react": emoji, "msg": msg_id}} case "edit": action_data = {"edit": {"msg": msg_id, "content": [{"text": content}]}} case "del": action_data = {"del": {"msg": msg_id}} case "typing": action_data = {"typing": None} case _: raise ValueError(f"Unknown chat action: {action}") return { "action": "poke", "ship": host_ship, "app": "chat", "mark": "chat-action", "json": { "path": channel_path, "action": action_data, }, } def build_media_content(media_type: str, data: str) -> list[dict[str, Any]]: match media_type: case "image": return [{"image": {"src": data, "width": 0, "height": 0}}] case "video": return [{"video": {"src": data}}] case "audio": return [{"audio": {"src": data}}] case "url": return [{"url": data}] case "reference": return [{"reference": {"text": data}}] case _: return [{"text": data}]