feat(feishu): auto-convert markdown tables to native table component

gateway/platforms/feishu.py

@@ -155,7 +155,8 @@
     re.MULTILINE,
 )
 # Detect markdown tables: a line starting with | followed by a separator line.
-# Feishu post-type 'md' elements do not render tables, so we force text mode.
+# Table content is converted to an interactive card; post-type 'md' elements
+# cannot render tables.
 _MARKDOWN_TABLE_RE = re.compile(r"^\|.*\|\n\|[-|: ]+\|", re.MULTILINE)
 _MARKDOWN_LINK_RE = re.compile(r"\[([^\]]+)\]\(([^)]+)\)")
 _MARKDOWN_FENCE_OPEN_RE = re.compile(r"^```([^\n`]*)\s*$")
@@ -560,6 +561,169 @@ def _build_markdown_post_payload(content: str) -> str:
     )
 
 
+_TABLE_MAX_ROWS = 50
+
+
+def _build_table_card_payload(content: str) -> Optional[str]:
+    """Convert markdown-table content into a Feishu interactive card JSON string.
+
+    Uses the native Feishu card v2 ``table`` component so tables render as
+    proper interactive tables instead of raw pipe-character text.
+
+    Returns ``None`` when the content cannot be parsed (caller should fall back
+    to plain text).
+    """
+    try:
+        segments = _parse_table_segments(content)
+    except Exception:
+        return None
+
+    if not segments:
+        return None
+
+    # Determine card title from first table's first column value.
+    title = "📊 数据"
+    for seg in segments:
+        if seg["type"] == "table":
+            title = f"📊 {seg['header'][0]}" if seg["header"] else "📊 数据"
+            break
+
+    elements = []  # type: List[Dict[str, Any]]
+    for seg in segments:
+        if seg["type"] == "text":
+            text = seg["content"].strip()
+            if text:
+                elements.append({"tag": "markdown", "content": text})
+        else:
+            table_elem = _build_feishu_table_element(seg)
+            if table_elem:
+                elements.append(table_elem)
+
+    if not elements:
+        return None
+
+    card = {
+        "schema": "2.0",
+        "config": {"wide_screen_mode": True},
+        "header": {
+            "title": {"content": title, "tag": "plain_text"},
+            "template": "blue",
+        },
+        "body": {"elements": elements},
+    }
+    return json.dumps(card, ensure_ascii=False)
+
+
+def _build_feishu_table_element(seg: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+    """Build a Feishu card v2 ``table`` component from a parsed table segment.
+
+    Returns a dict suitable for inclusion in ``body.elements[]`` of a v2 card.
+    """
+    header = seg.get("header", [])
+    rows = seg.get("rows", [])
+    if not header:
+        return None
+
+    columns = []
+    for idx, col_name in enumerate(header):
+        col_key = f"col{idx}"
+        columns.append({
+            "name": col_key,
+            "display_name": col_name,
+            "data_type": "text",
+            "width": "auto",
+        })
+
+    row_data = []
+    for row in rows:
+        padded = (row + [""] * len(header))[: len(header)]
+        row_obj = {}
+        for idx, val in enumerate(padded):
+            row_obj[f"col{idx}"] = val
+        row_data.append(row_obj)
+
+    return {
+        "tag": "table",
+        "page_size": min(len(row_data), 10) if row_data else 5,
+        "row_height": "auto",
+        "header_style": {
+            "text_align": "left",
+            "background_style": "grey",
+            "bold": True,
+        },
+        "columns": columns,
+        "rows": row_data,
+    }
+
+
+def _parse_table_segments(content: str) -> List[Dict[str, Any]]:
+    """Split *content* into alternating text/table segments.
+
+    Each segment is ``{"type": "text", "content": str}`` or
+    ``{"type": "table", "header": [...], "rows": [[...], ...]}``.
+    Raises ``ValueError`` on malformed tables so the caller can fall back.
+    """
+    lines = content.split("\n")
+    segments = []  # type: List[Dict[str, Any]]
+    buf = []  # type: List[str]
+    i = 0
+
+    def _flush_buf() -> None:
+        text = "\n".join(buf)
+        buf.clear()
+        if text.strip():
+            segments.append({"type": "text", "content": text})
+
+    while i < len(lines):
+        line = lines[i]
+        # Look for the start of a table: data row followed by separator row.
+        if (
+            line.startswith("|")
+            and i + 1 < len(lines)
+            and re.match(r"^\|[-|: ]+\|$", lines[i + 1].strip())
+        ):
+            _flush_buf()
+            header = [c.strip() for c in line.strip().strip("|").split("|")]
+            i += 2  # skip header + separator
+
+            rows = []  # type: List[List[str]]
+            while i < len(lines) and lines[i].startswith("|") and "|" in lines[i][1:]:
+                cells = [c.strip() for c in lines[i].strip().strip("|").split("|")]
+                rows.append(cells)
+                i += 1
+
+            if len(rows) > _TABLE_MAX_ROWS:
+                rows = rows[:_TABLE_MAX_ROWS]
+                # Append a truncation note as a following text segment.
+                segments.append({"type": "table", "header": header, "rows": rows})
+                segments.append({
+                    "type": "text",
+                    "content": f"⚠️ 表格过长，已截断至前 {_TABLE_MAX_ROWS} 行",
+                })
+            else:
+                segments.append({"type": "table", "header": header, "rows": rows})
+            continue
+
+        buf.append(line)
+        i += 1
+
+    _flush_buf()
+    return segments
+
+
+def _format_table_md(seg: Dict[str, Any]) -> str:
+    """Render a table segment dict back to markdown table syntax."""
+    header = seg["header"]
+    rows = seg["rows"]
+    parts = ["| " + " | ".join(header) + " |"]
+    parts.append("| " + " | ".join("---" for _ in header) + " |")
+    for row in rows:
+        # Pad/truncate row to match header column count.
+        padded = (row + [""] * len(header))[: len(header)]
+        parts.append("| " + " | ".join(padded) + " |")
+    return "\n".join(parts)
+
+
 def _build_markdown_post_rows(content: str) -> List[List[Dict[str, str]]]:
     """Build Feishu post rows while isolating fenced code blocks.
 
@@ -4374,10 +4538,11 @@ def _is_duplicate(self, message_id: str) -> bool:
     # =========================================================================
 
     def _build_outbound_payload(self, content: str) -> tuple[str, str]:
-        # Feishu post-type 'md' elements do not render markdown tables; sending
-        # table content as post causes the message to appear blank on the client.
-        # Force plain text for anything that looks like a markdown table.
         if _MARKDOWN_TABLE_RE.search(content):
+            card_json = _build_table_card_payload(content)
+            if card_json is not None:
+                return "interactive", card_json
+            # Card building failed – fall back to plain text.
             text_payload = {"text": content}
             return "text", json.dumps(text_payload, ensure_ascii=False)
         if _MARKDOWN_HINT_RE.search(content):

tests/gateway/test_feishu.py

@@ -4944,3 +4944,170 @@ async def _run():
                 held.release()
 
         asyncio.run(_run())
+
+
+class TestBuildTableCardPayload(unittest.TestCase):
+    """_build_table_card_payload converts markdown tables to Feishu cards."""
+
+    # -- helper -----------------------------------------------------------
+
+    def _parse_card(self, content):
+        """Run _build_table_card_payload and return parsed card dict."""
+        from gateway.platforms.feishu import _build_table_card_payload
+
+        raw = _build_table_card_payload(content)
+        self.assertIsNotNone(raw, "card payload should not be None")
+        return json.loads(raw)
+
+    # -- single table -----------------------------------------------------
+
+    def test_single_table_becomes_card(self):
+        table = "| Name | Age |\n| --- | --- |\n| Alice | 30 |\n| Bob | 25 |"
+        card = self._parse_card(table)
+
+        self.assertEqual(card["config"]["wide_screen_mode"], True)
+        self.assertEqual(card["header"]["template"], "blue")
+        self.assertIn("📊", card["header"]["title"]["content"])
+        elements = card["body"]["elements"]
+        self.assertEqual(len(elements), 1)
+        self.assertEqual(elements[0]["tag"], "table")
+        # Verify table structure
+        col_names = [c["display_name"] for c in elements[0]["columns"]]
+        self.assertEqual(col_names, ["Name", "Age"])
+        self.assertEqual(len(elements[0]["rows"]), 2)
+        self.assertEqual(elements[0]["rows"][0]["col0"], "Alice")
+        self.assertEqual(elements[0]["rows"][1]["col1"], "25")
+
+    # -- mixed text + table -----------------------------------------------
+
+    def test_text_before_and_after_table(self):
+        content = (
+            "Here is the report:\n"
+            "\n"
+            "| Col |\n| --- |\n| val |\n"
+            "\n"
+            "End of report."
+        )
+        card = self._parse_card(content)
+        elements = card["body"]["elements"]
+        # Should have: text-before, table, text-after
+        self.assertEqual(len(elements), 3)
+        self.assertEqual(elements[0]["tag"], "markdown")
+        self.assertIn("Here is the report", elements[0]["content"])
+        self.assertEqual(elements[1]["tag"], "table")
+        self.assertEqual(elements[1]["columns"][0]["display_name"], "Col")
+        self.assertEqual(elements[1]["rows"][0]["col0"], "val")
+        self.assertEqual(elements[2]["tag"], "markdown")
+        self.assertIn("End of report", elements[2]["content"])
+
+    # -- multiple tables --------------------------------------------------
+
+    def test_two_tables_in_one_message(self):
+        content = (
+            "| A |\n| --- |\n| 1 |\n"
+            "\nSome text\n\n"
+            "| B |\n| --- |\n| 2 |"
+        )
+        card = self._parse_card(content)
+        elements = card["body"]["elements"]
+        # table1, text, table2
+        self.assertEqual(len(elements), 3)
+        self.assertEqual(elements[0]["tag"], "table")
+        self.assertEqual(elements[0]["columns"][0]["display_name"], "A")
+        self.assertEqual(elements[0]["rows"][0]["col0"], "1")
+        self.assertEqual(elements[1]["tag"], "markdown")
+        self.assertIn("Some text", elements[1]["content"])
+        self.assertEqual(elements[2]["tag"], "table")
+        self.assertEqual(elements[2]["columns"][0]["display_name"], "B")
+        self.assertEqual(elements[2]["rows"][0]["col0"], "2")
+
+    # -- large table truncation -------------------------------------------
+
+    def test_large_table_truncated_to_50_rows(self):
+        rows = "\n".join(f"| row{i} |" for i in range(80))
+        content = f"| Id |\n| --- |\n{rows}"
+        card = self._parse_card(content)
+        elements = card["body"]["elements"]
+        # table + truncation note
+        self.assertTrue(len(elements) >= 2)
+        table_elem = elements[0]
+        self.assertEqual(table_elem["tag"], "table")
+        # Native table component has rows directly.
+        self.assertEqual(len(table_elem["rows"]), 50)
+        # Truncation warning
+        last = elements[-1]
+        self.assertEqual(last["tag"], "markdown")
+        self.assertIn("截断", last["content"])
+
+    # -- malformed table falls back ---------------------------------------
+
+    def test_pipe_line_without_separator_produces_text_card(self):
+        from gateway.platforms.feishu import _build_table_card_payload
+
+        # A string that looks like a pipe line but has no separator row.
+        # _parse_table_segments treats it as plain text, so
+        # _build_table_card_payload still returns a valid card (text-only).
+        result = _build_table_card_payload("| just a pipe line |\nnot a table")
+        card = json.loads(result)
+        self.assertEqual(card["body"]["elements"][0]["tag"], "markdown")
+
+    def test_empty_content_returns_none(self):
+        from gateway.platforms.feishu import _build_table_card_payload
+
+        result = _build_table_card_payload("")
+        self.assertIsNone(result)
+
+    def test_whitespace_only_returns_none(self):
+        from gateway.platforms.feishu import _build_table_card_payload
+
+        result = _build_table_card_payload("   \n  \n  ")
+        self.assertIsNone(result)
+
+    # -- alignment markers stripped ---------------------------------------
+
+    def test_alignment_markers_are_replaced(self):
+        table = "| L | C | R |\n| :--- | :---: | ---: |\n| a | b | c |"
+        card = self._parse_card(table)
+        elem = card["body"]["elements"][0]
+        self.assertEqual(elem["tag"], "table")
+        # Alignment markers are stripped; column names are clean.
+        col_names = [c["display_name"] for c in elem["columns"]]
+        self.assertEqual(col_names, ["L", "C", "R"])
+        # Row data should contain the values, not alignment markers.
+        self.assertEqual(elem["rows"][0], {"col0": "a", "col1": "b", "col2": "c"})
+
+    # -- _build_outbound_payload integration ------------------------------
+
+    @patch.dict(os.environ, {}, clear=True)
+    def test_build_outbound_payload_returns_interactive_for_table(self):
+        from gateway.config import PlatformConfig
+        from gateway.platforms.feishu import FeishuAdapter
+
+        adapter = FeishuAdapter(PlatformConfig())
+        content = "| X |\n| --- |\n| 1 |"
+        msg_type, payload = adapter._build_outbound_payload(content)
+        self.assertEqual(msg_type, "interactive")
+        card = json.loads(payload)
+        self.assertEqual(card["config"]["wide_screen_mode"], True)
+        self.assertIn("X", card["header"]["title"]["content"])
+
+    @patch.dict(os.environ, {}, clear=True)
+    def test_build_outbound_payload_falls_back_for_plain_text(self):
+        from gateway.config import PlatformConfig
+        from gateway.platforms.feishu import FeishuAdapter
+
+        adapter = FeishuAdapter(PlatformConfig())
+        content = "no tables here"
+        msg_type, payload = adapter._build_outbound_payload(content)
+        self.assertEqual(msg_type, "text")
+        self.assertEqual(json.loads(payload)["text"], content)
+
+    @patch.dict(os.environ, {}, clear=True)
+    def test_build_outbound_payload_falls_back_for_markdown_only(self):
+        from gateway.config import PlatformConfig
+        from gateway.platforms.feishu import FeishuAdapter
+
+        adapter = FeishuAdapter(PlatformConfig())
+        content = "**bold** and _italic_"
+        msg_type, payload = adapter._build_outbound_payload(content)
+        self.assertEqual(msg_type, "post")