diff --git a/gateway/platforms/feishu.py b/gateway/platforms/feishu.py index 351337e827..6e27d33e09 100644 --- a/gateway/platforms/feishu.py +++ b/gateway/platforms/feishu.py @@ -430,23 +430,71 @@ def _coerce_required_int(value: Any, default: int, min_value: int = 0) -> int: def _build_markdown_post_payload(content: str) -> str: + rows = _build_markdown_post_rows(content) return json.dumps( { "zh_cn": { - "content": [ - [ - { - "tag": "md", - "text": content, - } - ] - ], + "content": rows, } }, ensure_ascii=False, ) +def _build_markdown_post_rows(content: str) -> List[List[Dict[str, str]]]: + """Build Feishu post rows while isolating fenced code blocks. + + Feishu's `md` renderer can swallow trailing content when a fenced code block + appears inside one large markdown element. Splitting the reply at code + fences preserves the surrounding markdown while keeping the code block in a + dedicated row. + """ + if not content: + return [[{"tag": "md", "text": ""}]] + if "```" not in content: + return [[{"tag": "md", "text": content}]] + + rows: List[List[Dict[str, str]]] = [] + current: List[str] = [] + in_code_block = False + + for raw_line in content.splitlines(): + line = raw_line.rstrip() + is_fence = line.strip().startswith("```") + + if is_fence: + if not in_code_block and current: + segment = "\n".join(current).strip() + if segment: + rows.append([{"tag": "md", "text": segment}]) + current = [] + current.append(line) + in_code_block = not in_code_block + if not in_code_block: + segment = "\n".join(current).strip() + if segment: + rows.append([{"tag": "md", "text": segment}]) + current = [] + continue + + current.append(line) + + if current: + segment = "\n".join(current).strip() + if segment: + rows.append([{"tag": "md", "text": segment}]) + + return rows or [[{"tag": "md", "text": content}]] + + +def parse_feishu_post_content(raw_content: str) -> FeishuPostParseResult: + try: + parsed = json.loads(raw_content) if raw_content else {} + except json.JSONDecodeError: + return FeishuPostParseResult(text_content=FALLBACK_POST_TEXT) + return parse_feishu_post_payload(parsed) + + def parse_feishu_post_payload(payload: Any) -> FeishuPostParseResult: resolved = _resolve_post_payload(payload) if not resolved: diff --git a/tests/gateway/test_feishu.py b/tests/gateway/test_feishu.py index 661e37ec1a..47e5a94966 100644 --- a/tests/gateway/test_feishu.py +++ b/tests/gateway/test_feishu.py @@ -2370,6 +2370,69 @@ class TestAdapterBehavior(unittest.TestCase): elements = payload["zh_cn"]["content"][0] self.assertEqual(elements, [{"tag": "md", "text": "可以用 **粗体** 和 *斜体*。"}]) + @patch.dict(os.environ, {}, clear=True) + def test_send_splits_fenced_code_blocks_into_separate_post_rows(self): + from gateway.config import PlatformConfig + from gateway.platforms.feishu import FeishuAdapter + + adapter = FeishuAdapter(PlatformConfig()) + captured = {} + + class _MessageAPI: + def create(self, request): + captured["request"] = request + return SimpleNamespace( + success=lambda: True, + data=SimpleNamespace(message_id="om_codeblock"), + ) + + adapter._client = SimpleNamespace( + im=SimpleNamespace( + v1=SimpleNamespace( + message=_MessageAPI(), + ) + ) + ) + + async def _direct(func, *args, **kwargs): + return func(*args, **kwargs) + + content = ( + "确认已入库 ✓\n" + "文件路径:`/root/.hermes/profiles/agent_cto/cron/jobs.json`\n" + "**解码后的内容:**\n" + "```json\n" + '{"cron": "list"}\n' + "```\n" + "后续说明仍应保留。" + ) + + with patch("gateway.platforms.feishu.asyncio.to_thread", side_effect=_direct): + result = asyncio.run( + adapter.send( + chat_id="oc_chat", + content=content, + ) + ) + + self.assertTrue(result.success) + self.assertEqual(captured["request"].request_body.msg_type, "post") + payload = json.loads(captured["request"].request_body.content) + rows = payload["zh_cn"]["content"] + self.assertEqual( + rows, + [ + [ + { + "tag": "md", + "text": "确认已入库 ✓\n文件路径:`/root/.hermes/profiles/agent_cto/cron/jobs.json`\n**解码后的内容:**", + } + ], + [{"tag": "md", "text": "```json\n{\"cron\": \"list\"}\n```"}], + [{"tag": "md", "text": "后续说明仍应保留。"}], + ], + ) + @patch.dict(os.environ, {}, clear=True) def test_send_falls_back_to_text_when_post_payload_is_rejected(self): from gateway.config import PlatformConfig