fix: 从 QQ 引用消息的 JSON 卡片中提取文本 | extract text from quoted QQ JSON cards

astrbot/core/utils/quoted_message/chain_parser.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import html
 import json
 import re
 from typing import Any, TypedDict
@@ -10,6 +11,7 @@
     File,
     Forward,
     Image,
+    Json,
     Node,
     Nodes,
     Plain,
@@ -142,6 +144,10 @@ def _extract_text_from_component_chain(
             parts.append(f"[File:{file_name}]")
         elif isinstance(seg, Forward):
             parts.append("[Forward Message]")
+        elif isinstance(seg, Json):
+            card_text = _extract_text_from_json_card(seg.data)
+            if card_text:
+                parts.append(card_text)
         elif isinstance(seg, Reply):
             nested = _extract_text_from_reply_component(
                 seg,
@@ -261,6 +267,85 @@ def _extract_text_from_multimsg_json(raw_json: str) -> str | None:
     return "\n".join(texts).strip() or None
 
 
+# Safety valve: keeps a single card from flooding the LLM context window.
+_MAX_JSON_CARD_TEXT_LENGTH = 1000
+
+
+def _clean_json_card_text(value: Any) -> str | None:
+    if not isinstance(value, str):
+        return None
+    cleaned = html.unescape(value).strip()
+    return cleaned or None
+
+
+def _extract_text_from_json_card(payload: str | dict[str, Any]) -> str | None:
+    """Extract readable text from a generic QQ JSON card (mini-program, music, etc.).
+
+    Keeps the existing behavior for ``com.tencent.multimsg`` (forward) cards by
+    delegating to :func:`_extract_text_from_multimsg_json`; other cards are read
+    from their top-level ``prompt`` and ``meta`` fields.
+    """
+    parsed = payload
+    if isinstance(parsed, str):
+        try:
+            parsed = json.loads(parsed)
+        except Exception:
+            return None
+    if not isinstance(parsed, dict):
+        return None
+
+    # Unwrap the double-encoded {"data": "<nested json>"} form when the outer
+    # dict carries no app identifier of its own.
+    nested = parsed.get("data")
+    if isinstance(nested, str) and parsed.get("app") is None:
+        try:
+            nested_parsed = json.loads(nested)
+            if isinstance(nested_parsed, dict):
+                parsed = nested_parsed
+        except Exception:
+            pass
+
+    if parsed.get("app") == "com.tencent.multimsg":
+        return _extract_text_from_multimsg_json(json.dumps(parsed))
+
+    parts: list[str] = []
+    seen: set[str] = set()
+
+    def _append(value: str) -> None:
+        if value and value not in seen:
+            seen.add(value)
+            parts.append(value)
+
+    def _append_field(label: str, value: Any) -> None:
+        cleaned = _clean_json_card_text(value)
+        if cleaned:
+            _append(f"{label}: {cleaned}")
+
+    prompt = _clean_json_card_text(parsed.get("prompt"))
+    if prompt:
+        _append(prompt)
+
+    meta = parsed.get("meta")
+    if isinstance(meta, dict):
+        for detail in meta.values():
+            if not isinstance(detail, dict):
+                continue
+            _append_field("Title", detail.get("title"))
+            _append_field("Description", detail.get("desc"))
+            url = _clean_json_card_text(
+                detail.get("qqdocurl") or detail.get("jumpUrl") or detail.get("url")
+            )
+            if url:
+                _append(f"URL: {url}")
+
+    if not parts:
+        return None
+    text = "\n".join(parts).strip()
+    if len(text) > _MAX_JSON_CARD_TEXT_LENGTH:
+        text = text[:_MAX_JSON_CARD_TEXT_LENGTH].rstrip() + "..."
+    return text
+
+
 def _parse_onebot_segments(
     segments: list[Any],
     *,
@@ -335,9 +420,9 @@ def _parse_onebot_segments(
             raw_json = seg_data.get("data")
             if isinstance(raw_json, str) and raw_json.strip():
                 raw_json = raw_json.replace("&#44;", ",")
-                multimsg_text = _extract_text_from_multimsg_json(raw_json)
-                if multimsg_text:
-                    text_parts.append(multimsg_text)
+                card_text = _extract_text_from_json_card(raw_json)
+                if card_text:
+                    text_parts.append(card_text)
 
     return _build_parsed_payload(
         _join_text_parts(text_parts),

tests/test_quoted_message_parser.py

@@ -2,7 +2,7 @@
 
 import pytest
 
-from astrbot.core.message.components import Image, Plain, Reply
+from astrbot.core.message.components import Image, Json, Plain, Reply
 from astrbot.core.utils.quoted_message_parser import (
     extract_quoted_message_images,
     extract_quoted_message_text,
@@ -514,3 +514,192 @@ async def test_extract_quoted_message_nested_forward_id_is_resolved():
 
     images = await extract_quoted_message_images(event)
     assert images == [nested_image]
+
+
+@pytest.mark.asyncio
+async def test_extract_quoted_message_text_json_card_in_reply_chain():
+    reply = Reply(
+        id="500",
+        chain=[
+            Json(
+                data={
+                    "app": "com.tencent.miniapp_01",
+                    "prompt": "[QQ小程序]无穷小亮正式加入《洛克王国：世界》精灵调查团！",
+                    "meta": {
+                        "detail_1": {
+                            "title": "哔哩哔哩",
+                            "desc": "无穷小亮正式加入《洛克王国：世界》精灵调查团！",
+                            "qqdocurl": "https://b23.tv/2PFeKuq",
+                        }
+                    },
+                }
+            )
+        ],
+        message_str="",
+    )
+    event = SimpleNamespace(
+        message_obj=SimpleNamespace(message=[reply]),
+        bot=SimpleNamespace(api=_FailIfCalledAPI()),
+        get_group_id=lambda: "",
+    )
+
+    text = await extract_quoted_message_text(event)
+    assert text is not None
+    assert "无穷小亮正式加入《洛克王国：世界》精灵调查团！" in text
+    assert "Title: 哔哩哔哩" in text
+    assert "URL: https://b23.tv/2PFeKuq" in text
+
+
+@pytest.mark.asyncio
+async def test_extract_quoted_message_text_json_card_via_get_msg_fallback():
+    reply = Reply(id="501", chain=None, message_str="")
+    event = _make_event(
+        reply,
+        responses={
+            ("get_msg", "501"): {
+                "data": {
+                    "message": [
+                        {
+                            "type": "json",
+                            "data": {
+                                "data": (
+                                    '{"app":"com.tencent.miniapp_01",'
+                                    '"prompt":"[QQ小程序]卡片标题",'
+                                    '"meta":{"detail_1":{"title":"某App",'
+                                    '"desc":"卡片描述","jumpUrl":"https://example.com/x"}}}'
+                                )
+                            },
+                        }
+                    ]
+                }
+            }
+        },
+    )
+
+    text = await extract_quoted_message_text(event)
+    assert text is not None
+    assert "卡片标题" in text
+    assert "Title: 某App" in text
+    assert "卡片描述" in text
+    assert "https://example.com/x" in text
+
+
+@pytest.mark.asyncio
+async def test_extract_quoted_message_text_json_card_multimsg_still_extracts_news():
+    reply = Reply(id="502", chain=None, message_str="")
+    event = _make_event(
+        reply,
+        responses={
+            ("get_msg", "502"): {
+                "data": {
+                    "message": [
+                        {
+                            "type": "json",
+                            "data": {
+                                "data": (
+                                    '{"app":"com.tencent.multimsg",'
+                                    '"config":{"forward":1},'
+                                    '"meta":{"detail":{"news":['
+                                    '{"text":"Alice: hello"},{"text":"Bob: world"}]}}}'
+                                )
+                            },
+                        }
+                    ]
+                }
+            }
+        },
+    )
+
+    text = await extract_quoted_message_text(event)
+    assert text is not None
+    assert "Alice: hello" in text
+    assert "Bob: world" in text
+
+
+@pytest.mark.asyncio
+async def test_extract_quoted_message_text_json_card_html_entities_unescaped():
+    reply = Reply(id="503", chain=None, message_str="")
+    event = _make_event(
+        reply,
+        responses={
+            ("get_msg", "503"): {
+                "data": {
+                    "message": [
+                        {
+                            "type": "json",
+                            "data": {
+                                "data": (
+                                    '{"app":"com.tencent.miniapp_01",'
+                                    '"prompt":"[QQ小程序]实体测试",'
+                                    '"meta":{"detail_1":{"title":"T&A"}}}'
+                                )
+                            },
+                        }
+                    ]
+                }
+            }
+        },
+    )
+
+    text = await extract_quoted_message_text(event)
+    assert text is not None
+    assert "[QQ小程序]实体测试" in text
+    assert "Title: T&A" in text
+    assert "[" not in text
+    assert "&" not in text
+
+
+@pytest.mark.asyncio
+async def test_extract_quoted_message_text_json_card_double_encoded_wrapper():
+    reply = Reply(
+        id="505",
+        chain=[
+            Json(
+                data={
+                    "data": (
+                        '{"app":"com.tencent.miniapp_01",'
+                        '"prompt":"包裹卡片","meta":{"detail_1":{"desc":"内层描述"}}}'
+                    )
+                }
+            )
+        ],
+        message_str="",
+    )
+    event = SimpleNamespace(
+        message_obj=SimpleNamespace(message=[reply]),
+        bot=SimpleNamespace(api=_FailIfCalledAPI()),
+        get_group_id=lambda: "",
+    )
+
+    text = await extract_quoted_message_text(event)
+    assert text is not None
+    assert "包裹卡片" in text
+    assert "内层描述" in text
+
+
+@pytest.mark.asyncio
+async def test_extract_quoted_message_text_json_card_length_capped():
+    long_desc = "字" * 5000
+    reply = Reply(
+        id="504",
+        chain=[
+            Json(
+                data={
+                    "app": "com.tencent.miniapp_01",
+                    "meta": {"detail_1": {"desc": long_desc}},
+                }
+            )
+        ],
+        message_str="",
+    )
+    event = SimpleNamespace(
+        message_obj=SimpleNamespace(message=[reply]),
+        bot=SimpleNamespace(api=_FailIfCalledAPI()),
+        get_group_id=lambda: "",
+    )
+
+    text = await extract_quoted_message_text(event)
+    assert text is not None
+    assert text.endswith("...")
+    assert len(text) < 1500
+    assert long_desc not in text