"""NapCat 入站 JSON 卡片解析辅助。""" from __future__ import annotations from typing import TYPE_CHECKING, Any, List, Mapping, Optional import hashlib import json import re from ...qq_emoji_list import QQ_FACE from ...types import NapCatSegment, NapCatSegments if TYPE_CHECKING: from ...services import NapCatQueryService class NapCatInboundCardMixin: """封装入站 JSON 卡片与预览内容转换逻辑。""" if TYPE_CHECKING: _query_service: NapCatQueryService @staticmethod def _build_text_segment(text: str) -> NapCatSegment: ... @staticmethod def _encode_binary(binary_data: bytes) -> str: ... async def _build_json_segments(self, segment_data: Mapping[str, Any]) -> NapCatSegments: """将 JSON 卡片最佳努力转换为消息段列表。 Args: segment_data: OneBot ``json`` 段的 ``data`` 字典。 Returns: NapCatSegments: 转换后的消息段列表。 """ json_data = str(segment_data.get("data") or "").strip() if not json_data: return [self._build_text_segment("[json]")] try: parsed_json = json.loads(json_data) except Exception: return [self._build_text_segment("[json]")] if not isinstance(parsed_json, Mapping): return [self._build_text_segment("[json]")] app_name = str(parsed_json.get("app") or "").strip() meta = parsed_json.get("meta", {}) if not isinstance(meta, Mapping): meta = {} if app_name == "com.tencent.mannounce": return [self._build_mannounce_segment(meta)] if app_name in {"com.tencent.music.lua", "com.tencent.structmsg"}: music_segments = self._build_music_card_segments(meta) if music_segments: return music_segments if app_name == "com.tencent.miniapp_01": return await self._build_preview_text_segments( self._build_miniapp_text(meta), self._extract_preview_url(meta, "detail_1"), ) if app_name == "com.tencent.giftmall.giftark": gift_text = self._build_gift_text(meta) if gift_text: return [self._build_text_segment(gift_text)] if app_name == "com.tencent.contact.lua": return [self._build_text_segment(self._build_contact_text(meta, "推荐联系人"))] if app_name == "com.tencent.troopsharecard": return [self._build_text_segment(self._build_contact_text(meta, "推荐群聊"))] if app_name == "com.tencent.tuwen.lua": return await self._build_preview_text_segments( self._build_news_text(meta, default_tag="图文分享"), self._extract_preview_url(meta, "news"), ) if app_name == "com.tencent.feed.lua": return await self._build_preview_text_segments( self._build_feed_text(meta), self._extract_preview_url(meta, "feed", field_name="cover"), ) if app_name == "com.tencent.template.qqfavorite.share": return await self._build_preview_text_segments( self._build_favorite_text(meta), self._extract_preview_url(meta, "news"), ) if app_name == "com.tencent.miniapp.lua": return await self._build_preview_text_segments( self._build_simple_title_text(meta, "miniapp", "QQ空间"), self._extract_preview_url(meta, "miniapp"), ) if app_name == "com.tencent.forum": forum_segments = await self._build_forum_segments(meta) if forum_segments: return forum_segments if app_name == "com.tencent.map": location_text = self._build_location_text(meta) if location_text: return [self._build_text_segment(location_text)] if app_name == "com.tencent.together": together_text = self._build_together_text(meta) if together_text: return [self._build_text_segment(together_text)] prompt = str(parsed_json.get("prompt") or "").strip() if not prompt and isinstance(meta, Mapping): prompt = str(meta.get("prompt") or "").strip() text = prompt or app_name or "json" return [self._build_text_segment(f"[json:{text}]")] def _build_mannounce_segment(self, meta: Mapping[str, Any]) -> NapCatSegment: """构造群公告文本段。 Args: meta: JSON 卡片 ``meta`` 数据。 Returns: NapCatSegment: 群公告文本段。 """ mannounce = meta.get("mannounce", {}) if not isinstance(mannounce, Mapping): mannounce = {} title = str(mannounce.get("title") or "").strip() text = str(mannounce.get("text") or "").strip() encode_flag = mannounce.get("encode") if encode_flag == 1: title = self._safe_base64_decode(title) text = self._safe_base64_decode(text) if title and text: content = f"[{title}]:{text}" elif title: content = f"[{title}]" elif text: content = text else: content = "[群公告]" return self._build_text_segment(content) def _build_music_card_segments(self, meta: Mapping[str, Any]) -> NapCatSegments: """构造音乐卡片文本段。 Args: meta: JSON 卡片 ``meta`` 数据。 Returns: NapCatSegments: 音乐卡片转换后的消息段列表。 """ music = meta.get("music", {}) if not isinstance(music, Mapping): return [] title = str(music.get("title") or "").strip() singer = str(music.get("desc") or music.get("singer") or "").strip() tag = str(music.get("tag") or "音乐分享").strip() text_parts: List[str] = [f"[{tag}]"] if title: text_parts.append(title) if singer: text_parts.append(f"- {singer}") content = " ".join(text_parts).strip() or "[音乐分享]" return [self._build_text_segment(content)] async def _build_preview_text_segments( self, text: str, preview_url: str, ) -> NapCatSegments: """构造“文本 + 预览图”消息段列表。 Args: text: 主文本内容。 preview_url: 预览图地址。 Returns: NapCatSegments: 转换后的消息段列表。 """ segments: NapCatSegments = [self._build_text_segment(text or "[卡片消息]")] image_segment = await self._build_remote_image_segment(preview_url) if image_segment is not None: segments.append(image_segment) return segments async def _build_remote_image_segment(self, image_url: str) -> Optional[NapCatSegment]: """从远端图片地址构造图片消息段。 Args: image_url: 图片地址。 Returns: Optional[NapCatSegment]: 成功时返回图片消息段,否则返回 ``None``。 """ normalized_url = str(image_url or "").strip() if not normalized_url: return None binary_data = await self._query_service.download_binary(normalized_url) if not binary_data: return None return { "type": "image", "data": "", "hash": hashlib.sha256(binary_data).hexdigest(), "binary_data_base64": self._encode_binary(binary_data), } def _build_miniapp_text(self, meta: Mapping[str, Any]) -> str: """构造小程序分享文本。 Args: meta: JSON 卡片 ``meta`` 数据。 Returns: str: 小程序分享文本。 """ detail = meta.get("detail_1", {}) if not isinstance(detail, Mapping): return "[小程序]" title = str(detail.get("title") or "").strip() desc = str(detail.get("desc") or "").strip() if title and desc: return f"[小程序] {title}:{desc}" if title: return f"[小程序] {title}" if desc: return f"[小程序] {desc}" return "[小程序]" def _build_gift_text(self, meta: Mapping[str, Any]) -> str: """构造礼物卡片文本。 Args: meta: JSON 卡片 ``meta`` 数据。 Returns: str: 礼物卡片文本。 """ giftark = meta.get("giftark", {}) if not isinstance(giftark, Mapping): return "[赠送礼物]" gift_name = str(giftark.get("title") or "礼物").strip() desc = str(giftark.get("desc") or "").strip() if desc: return f"[赠送礼物: {gift_name}] {desc}" return f"[赠送礼物: {gift_name}]" def _build_contact_text(self, meta: Mapping[str, Any], default_tag: str) -> str: """构造推荐联系人或群聊文本。 Args: meta: JSON 卡片 ``meta`` 数据。 default_tag: 默认标签文本。 Returns: str: 推荐对象文本。 """ contact = meta.get("contact", {}) if not isinstance(contact, Mapping): return f"[{default_tag}]" name = str(contact.get("nickname") or "未知对象").strip() tag = str(contact.get("tag") or default_tag).strip() or default_tag return f"[{tag}] {name}" def _build_news_text(self, meta: Mapping[str, Any], default_tag: str) -> str: """构造图文分享文本。 Args: meta: JSON 卡片 ``meta`` 数据。 default_tag: 默认标签文本。 Returns: str: 图文分享文本。 """ news = meta.get("news", {}) if not isinstance(news, Mapping): return f"[{default_tag}]" title = str(news.get("title") or "未知标题").strip() desc = str(news.get("desc") or "").replace("[图片]", "").strip() tag = str(news.get("tag") or default_tag).strip() or default_tag if tag and title and tag in title: title = self._trim_card_title(title.replace(tag, "", 1)) if desc: return f"[{tag}] {title}:{desc}" return f"[{tag}] {title}".strip() def _build_feed_text(self, meta: Mapping[str, Any]) -> str: """构造群相册分享文本。 Args: meta: JSON 卡片 ``meta`` 数据。 Returns: str: 群相册分享文本。 """ feed = meta.get("feed", {}) if not isinstance(feed, Mapping): return "[群相册]" title = str(feed.get("title") or "群相册").strip() tag = str(feed.get("tagName") or "群相册").strip() or "群相册" desc = str(feed.get("forwardMessage") or "").strip() if tag and title and tag in title: title = self._trim_card_title(title.replace(tag, "", 1)) if desc: return f"[{tag}] {title}:{desc}" return f"[{tag}] {title}".strip() def _build_favorite_text(self, meta: Mapping[str, Any]) -> str: """构造 QQ 收藏分享文本。 Args: meta: JSON 卡片 ``meta`` 数据。 Returns: str: QQ 收藏分享文本。 """ news = meta.get("news", {}) if not isinstance(news, Mapping): return "[QQ收藏]" desc = str(news.get("desc") or "").replace("[图片]", "").strip() tag = str(news.get("tag") or "QQ收藏").strip() or "QQ收藏" if desc: return f"[{tag}] {desc}" return f"[{tag}]" def _build_simple_title_text( self, meta: Mapping[str, Any], key: str, default_tag: str, ) -> str: """构造简单标题类卡片文本。 Args: meta: JSON 卡片 ``meta`` 数据。 key: 子对象键名。 default_tag: 默认标签文本。 Returns: str: 简单标题文本。 """ nested_payload = meta.get(key, {}) if not isinstance(nested_payload, Mapping): return f"[{default_tag}]" title = str(nested_payload.get("title") or "未知标题").strip() tag = str(nested_payload.get("tag") or default_tag).strip() or default_tag return f"[{tag}] {title}".strip() async def _build_forum_segments(self, meta: Mapping[str, Any]) -> NapCatSegments: """构造 QQ 频道帖子消息段。 Args: meta: JSON 卡片 ``meta`` 数据。 Returns: NapCatSegments: 频道帖子转换后的消息段列表。 """ detail = meta.get("detail", {}) if not isinstance(detail, Mapping): return [] feed = detail.get("feed", {}) poster = detail.get("poster", {}) channel_info = detail.get("channel_info", {}) if not isinstance(feed, Mapping) or not isinstance(poster, Mapping) or not isinstance(channel_info, Mapping): return [] guild_name = str(channel_info.get("guild_name") or "").strip() nick = str(poster.get("nick") or "QQ用户").strip() or "QQ用户" title = self._extract_forum_title(feed) face_content = self._extract_forum_face_text(feed) text_prefix = "[频道帖子]" if guild_name: text_prefix = f"{text_prefix} [{guild_name}]" text_content = f"{text_prefix}{nick}:{title}{face_content}" segments: NapCatSegments = [self._build_text_segment(text_content)] images = feed.get("images", []) if not isinstance(images, list): return segments for image_item in images: if not isinstance(image_item, Mapping): continue image_segment = await self._build_remote_image_segment(str(image_item.get("pic_url") or "").strip()) if image_segment is not None: segments.append(image_segment) return segments def _extract_forum_title(self, feed: Mapping[str, Any]) -> str: """提取 QQ 频道帖子标题。 Args: feed: 频道帖子 ``feed`` 数据。 Returns: str: 帖子标题。 """ title_payload = feed.get("title", {}) if not isinstance(title_payload, Mapping): return "帖子" contents = title_payload.get("contents", []) if not isinstance(contents, list) or not contents: return "帖子" first_content = contents[0] if not isinstance(first_content, Mapping): return "帖子" text_content = first_content.get("text_content", {}) if not isinstance(text_content, Mapping): return "帖子" return str(text_content.get("text") or "帖子").strip() or "帖子" def _extract_forum_face_text(self, feed: Mapping[str, Any]) -> str: """提取 QQ 频道帖子中的表情文本。 Args: feed: 频道帖子 ``feed`` 数据。 Returns: str: 合并后的表情文本。 """ contents_payload = feed.get("contents", {}) if not isinstance(contents_payload, Mapping): return "" contents = contents_payload.get("contents", []) if not isinstance(contents, list): return "" face_text_parts: List[str] = [] for item in contents: if not isinstance(item, Mapping): continue emoji_content = item.get("emoji_content", {}) if not isinstance(emoji_content, Mapping): continue emoji_id = str(emoji_content.get("id") or "").strip() if emoji_id in QQ_FACE: face_text_parts.append(QQ_FACE[emoji_id]) return "".join(face_text_parts) def _build_location_text(self, meta: Mapping[str, Any]) -> str: """构造位置分享文本。 Args: meta: JSON 卡片 ``meta`` 数据。 Returns: str: 位置分享文本。 """ location = meta.get("Location.Search", {}) if not isinstance(location, Mapping): return "[位置]" name = str(location.get("name") or "未知地点").strip() address = str(location.get("address") or "").strip() if address: return f"[位置] {address} · {name}" return f"[位置] {name}" def _build_together_text(self, meta: Mapping[str, Any]) -> str: """构造“一起听歌”文本。 Args: meta: JSON 卡片 ``meta`` 数据。 Returns: str: 一起听歌文本。 """ invite = meta.get("invite", {}) if not isinstance(invite, Mapping): return "[一起听歌]" title = str(invite.get("title") or "一起听歌").strip() or "一起听歌" summary = str(invite.get("summary") or "").strip() if summary: return f"[{title}] {summary}" return f"[{title}]" def _extract_preview_url( self, meta: Mapping[str, Any], key: str, field_name: str = "preview", ) -> str: """从卡片元数据中提取预览图地址。 Args: meta: JSON 卡片 ``meta`` 数据。 key: 子对象键名。 field_name: 预览图字段名。 Returns: str: 预览图地址;不存在时返回空字符串。 """ nested_payload = meta.get(key, {}) if not isinstance(nested_payload, Mapping): return "" return str(nested_payload.get(field_name) or "").strip() @staticmethod def _trim_card_title(title: str) -> str: """清理卡片标题两侧的常见分隔符。 Args: title: 原始标题文本。 Returns: str: 清理后的标题文本。 """ return re.sub(r"^[::\s\-—]+|[::\s\-—]+$", "", str(title or "").strip()) @staticmethod def _safe_base64_decode(encoded_text: str) -> str: """安全地解码 Base64 文本。 Args: encoded_text: 待解码的 Base64 文本。 Returns: str: 解码结果;失败时返回原始文本。 """ normalized_text = str(encoded_text or "").strip() if not normalized_text: return "" try: import base64 return base64.b64decode(normalized_text).decode("utf-8", errors="ignore") except Exception: return normalized_text