feat(feishu): support embedding images in interactive cards and sending mixed image+file messages

- Enhance Feishu card builder to embed images using remote URLs or uploaded files
- Add logic to send image card first, then file attachment when both image and file are present
- Update card schema to 2.0 and use new button behaviors for callbacks and URLs
- Improve callback data extraction for both new and legacy card actions
- Extend tests to cover image embedding, mixed messages, and new card structure

app/modules/feishu/__init__.py

@@ -92,7 +92,24 @@ class FeishuModule(_ModuleBase, _MessageBase[Feishu]):
             userid, chat_id, receive_id_type = self._resolve_message_target(message)
             client: Feishu = self.get_instance(conf.name)
             if client:
-                if message.file_path:
+                if message.image and message.file_path:
+                    # 普通文件无法嵌入卡片，先发送图文卡片，再单独发送附件，避免图片被 file_path 分支吞掉。
+                    client.send_notification(
+                        message=message.model_copy(update={"file_path": None, "file_name": None}),
+                        userid=userid,
+                        chat_id=chat_id,
+                        receive_id_type=receive_id_type,
+                        original_message_id=str(message.original_message_id) if message.original_message_id else None,
+                    )
+                    client.send_file(
+                        file_path=message.file_path,
+                        userid=userid,
+                        chat_id=chat_id,
+                        file_name=message.file_name,
+                        receive_id_type=receive_id_type,
+                        original_message_id=str(message.original_message_id) if message.original_message_id else None,
+                    )
+                elif message.file_path:
                     client.send_file(
                         file_path=message.file_path,
                         userid=userid,
@@ -186,7 +203,24 @@ class FeishuModule(_ModuleBase, _MessageBase[Feishu]):
             client: Feishu = self.get_instance(conf.name)
             if not client:
                 continue
-            if message.file_path:
+            if message.image and message.file_path:
+                result = client.send_notification(
+                    message=message.model_copy(update={"file_path": None, "file_name": None}),
+                    userid=userid,
+                    chat_id=chat_id,
+                    receive_id_type=receive_id_type,
+                    original_message_id=str(message.original_message_id) if message.original_message_id else None,
+                )
+                if result and result.get("success"):
+                    client.send_file(
+                        file_path=message.file_path,
+                        userid=userid,
+                        chat_id=chat_id,
+                        file_name=message.file_name,
+                        receive_id_type=receive_id_type,
+                        original_message_id=str(message.original_message_id) if message.original_message_id else None,
+                    )
+            elif message.file_path:
                 result = client.send_file(
                     file_path=message.file_path,
                     userid=userid,

app/modules/feishu/feishu.py

@@ -1,9 +1,11 @@
 import asyncio
 import json
+import tempfile
 import threading
 import uuid
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Tuple
+from urllib.parse import urlparse
 
 import lark_oapi as lark
 import lark_oapi.ws.client as lark_ws_client_module
@@ -61,6 +63,7 @@ class Feishu:
     PROCESSING_REACTION_EMOJI = "GLANCE"
     STREAM_CARD_TITLE_ELEMENT_ID = "mp_stream_title"
     STREAM_CARD_BODY_ELEMENT_ID = "mp_stream_body"
+    IMAGE_SUFFIXES = {".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".ico", ".tiff", ".heic"}
 
     def __init__(
             self,
@@ -314,12 +317,10 @@ class Feishu:
         operator = getattr(event, "operator", None)
         action = getattr(event, "action", None)
         context = getattr(event, "context", None)
-        value = getattr(action, "value", None) or {}
+        callback_data = self._extract_card_callback_data(
-        callback_data = None
+            value=getattr(action, "value", None),
-        if isinstance(value, dict):
+            name=getattr(action, "name", None),
-            callback_data = value.get("callback_data") or value.get("value")
+        )
-        if not callback_data:
-            callback_data = getattr(action, "name", None)
 
         payload = {
             "type": "cardAction",
@@ -575,6 +576,73 @@ class Feishu:
             parts.append(f"[查看详情]({link.strip()})")
         return "\n\n".join(part for part in parts if part)
 
+    @staticmethod
+    def _guess_image_suffix(image_url: str, content_type: Optional[str] = None) -> str:
+        """根据 URL 或响应 Content-Type 推断临时图片后缀。"""
+        content_type = (content_type or "").split(";", 1)[0].strip().lower()
+        suffix_map = {
+            "image/png": ".png",
+            "image/jpeg": ".jpg",
+            "image/jpg": ".jpg",
+            "image/gif": ".gif",
+            "image/webp": ".webp",
+            "image/bmp": ".bmp",
+            "image/tiff": ".tiff",
+            "image/heic": ".heic",
+        }
+        if content_type in suffix_map:
+            return suffix_map[content_type]
+        path_suffix = Path(urlparse(image_url).path).suffix.lower()
+        if path_suffix in Feishu.IMAGE_SUFFIXES:
+            return path_suffix
+        return ".jpg"
+
+    def _upload_remote_image(self, image_url: Optional[str]) -> Optional[str]:
+        """下载远程图片并上传到飞书，返回可用于卡片的 image_key。"""
+        image_url = (image_url or "").strip()
+        if not image_url:
+            return None
+        if image_url.startswith("feishu://image/"):
+            resource_path = image_url.replace("feishu://image/", "", 1)
+            return resource_path.rsplit("/", 1)[-1].strip() or None
+
+        response = None
+        temp_path = None
+        try:
+            response = RequestUtils(timeout=30, ua=settings.USER_AGENT).get_res(image_url)
+            if not response or not getattr(response, "content", None):
+                logger.warning(f"飞书图片下载失败：{image_url}")
+                return None
+            content_type = response.headers.get("Content-Type") if response.headers else None
+            suffix = self._guess_image_suffix(image_url=image_url, content_type=content_type)
+            with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as fp:
+                fp.write(response.content)
+                temp_path = Path(fp.name)
+            return self._upload_image(temp_path)
+        except Exception as err:
+            logger.error(f"飞书远程图片上传失败：{err}")
+            return None
+        finally:
+            if response is not None:
+                response.close()
+            if temp_path:
+                try:
+                    temp_path.unlink(missing_ok=True)
+                except Exception as err:
+                    logger.debug(f"删除飞书临时图片失败：{err}")
+
+    @staticmethod
+    def _extract_card_callback_data(value: Any, name: Optional[str] = None) -> Optional[str]:
+        """从新版/旧版飞书卡片回调中提取统一的 callback_data。"""
+        callback_data = None
+        if isinstance(value, dict):
+            callback_data = value.get("callback_data") or value.get("value")
+        elif isinstance(value, str):
+            callback_data = value
+        if not callback_data:
+            callback_data = name
+        return str(callback_data).strip() if callback_data else None
+
     @staticmethod
     def _card_actions(buttons: Optional[List[List[dict]]]) -> List[dict]:
         """将统一按钮结构转换为飞书卡片按钮配置。"""
@@ -582,34 +650,71 @@ class Feishu:
             return []
         card_rows = []
         for row in buttons[:8]:
-            elements = []
+            columns = []
             for button in row[:3]:
                 text = (button or {}).get("text")
                 if not text:
                     continue
                 url = (button or {}).get("url")
                 callback_data = (button or {}).get("callback_data")
-                value = {"callback_data": callback_data} if callback_data else {"value": text}
+                behaviors = []
+                # 长连接模式不支持旧版消息卡片回传，必须使用新版 behaviors callback。
+                if callback_data:
+                    behaviors.append(
+                        {
+                            "type": "callback",
+                            "value": {"callback_data": str(callback_data)},
+                        }
+                    )
+                if url:
+                    behaviors.append(
+                        {
+                            "type": "open_url",
+                            "default_url": str(url),
+                            "pc_url": str(url),
+                            "android_url": str(url),
+                            "ios_url": str(url),
+                        }
+                    )
+                if not behaviors:
+                    behaviors.append(
+                        {
+                            "type": "callback",
+                            "value": {"callback_data": str(text)},
+                        }
+                    )
                 element = {
                     "tag": "button",
                     "text": {"tag": "plain_text", "content": text[:20]},
                     "type": "default",
-                    "value": value,
+                    "behaviors": behaviors,
                 }
-                if url:
+                columns.append(
-                    element["multi_url"] = {
+                    {
-                        "url": url,
+                        "tag": "column",
-                        "pc_url": url,
+                        "width": "weighted",
-                        "android_url": url,
+                        "weight": 1,
-                        "ios_url": url,
+                        "elements": [element],
                     }
-                elements.append(element)
+                )
-            if elements:
+            if columns:
-                card_rows.append({"tag": "action", "actions": elements})
+                card_rows.append(
+                    {
+                        "tag": "column_set",
+                        "flex_mode": "none",
+                        "columns": columns,
+                    }
+                )
         return card_rows
 
-    def _build_card(self, title: Optional[str], text: Optional[str], link: Optional[str],
+    def _build_card(
-                    buttons: Optional[List[List[dict]]]) -> Dict[str, Any]:
+            self,
+            title: Optional[str],
+            text: Optional[str],
+            link: Optional[str],
+            buttons: Optional[List[List[dict]]],
+            image_key: Optional[str] = None,
+    ) -> Dict[str, Any]:
         """构建飞书交互卡片结构。"""
         elements: List[dict] = []
         title_section = self._build_markdown_section(title, text_size="heading")
@@ -621,15 +726,35 @@ class Feishu:
             elements.append(title_section)
         if body_section:
             elements.append(body_section)
+        if image_key:
+            elements.append(
+                {
+                    "tag": "img",
+                    "img_key": image_key,
+                    "alt": {
+                        "tag": "plain_text",
+                        "content": title or "图片",
+                    },
+                    "mode": "fit_horizontal",
+                }
+            )
         elements.extend(self._card_actions(buttons))
         return {
             # 飞书卡片消息要支持后续 PATCH 更新，发送和更新时都必须显式声明 update_multi。
+            "schema": "2.0",
             "config": {
                 "wide_screen_mode": True,
                 "enable_forward": True,
                 "update_multi": True,
+                "summary": {
+                    "content": title or "MoviePilot",
+                },
+            },
+            "body": {
+                "direction": "vertical",
+                "padding": "12px 12px 12px 12px",
+                "elements": elements,
             },
-            "elements": elements,
         }
 
     def _build_streaming_card_payload(
@@ -1060,17 +1185,24 @@ class Feishu:
             return {"success": False}
 
         suffix = local_file.suffix.lower()
-        is_image = suffix in {".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".ico", ".tiff", ".heic"}
+        is_image = suffix in self.IMAGE_SUFFIXES
         try:
             if is_image:
                 image_key = self._upload_image(local_file)
                 if not image_key:
                     return {"success": False}
+                payload = self._build_card(
+                    title=title,
+                    text=text,
+                    link=None,
+                    buttons=None,
+                    image_key=image_key,
+                )
                 if original_message_id:
                     result = self._reply_message(
                         message_id=original_message_id,
-                        msg_type="image",
+                        msg_type="interactive",
-                        content={"image_key": image_key},
+                        content=payload,
                     )
                 else:
                     receive_id, resolved_receive_id_type = self._resolve_target(
@@ -1081,8 +1213,8 @@ class Feishu:
                     result = self._send_message(
                         receive_id,
                         resolved_receive_id_type,
-                        "image",
+                        "interactive",
-                        {"image_key": image_key},
+                        payload,
                     )
             else:
                 file_key = self._upload_file(local_file, file_name=file_name)
@@ -1106,7 +1238,7 @@ class Feishu:
                         "file",
                         {"file_key": file_key},
                     )
-            if result and (title or text):
+            if result and (title or text) and not is_image:
                 self.send_text(
                     self._build_message_text(title=title, text=text),
                     userid=userid,
@@ -1212,11 +1344,13 @@ class Feishu:
                 userid or "") or self._default_chat_id
             return result
 
+        image_key = self._upload_remote_image(message.image)
         payload = self._build_card(
             title=message.title,
             text=message.text,
             link=message.link,
             buttons=message.buttons,
+            image_key=image_key,
         )
         try:
             if original_message_id:

tests/test_feishu.py

@@ -144,6 +144,24 @@ class TestFeishu(unittest.TestCase):
         self.assertTrue(result.is_callback)
         self.assertEqual(result.chat_id, "oc_123")
 
+    def test_extract_card_callback_data_supports_new_and_legacy_values(self):
+        self.assertEqual(
+            Feishu._extract_card_callback_data({"callback_data": "approve"}),
+            "approve",
+        )
+        self.assertEqual(
+            Feishu._extract_card_callback_data({"value": "legacy"}),
+            "legacy",
+        )
+        self.assertEqual(
+            Feishu._extract_card_callback_data("direct"),
+            "direct",
+        )
+        self.assertEqual(
+            Feishu._extract_card_callback_data({}, name="fallback"),
+            "fallback",
+        )
+
     def test_build_event_handler_registers_common_im_events(self):
         registered = []
 
@@ -221,9 +239,54 @@ class TestFeishu(unittest.TestCase):
 
         content = json.loads(request.request_body.content)
         self.assertNotIn("card", content)
+        self.assertEqual(content["schema"], "2.0")
         self.assertTrue(content["config"]["update_multi"])
-        self.assertEqual(content["elements"][0]["text_size"], "heading")
+        self.assertEqual(content["body"]["elements"][0]["text_size"], "heading")
-        self.assertEqual(content["elements"][0]["tag"], "markdown")
+        self.assertEqual(content["body"]["elements"][0]["tag"], "markdown")
+        button = content["body"]["elements"][-1]["columns"][0]["elements"][0]
+        self.assertEqual(button["tag"], "button")
+        self.assertNotIn("value", button)
+        self.assertEqual(
+            button["behaviors"],
+            [{"type": "callback", "value": {"callback_data": "confirm"}}],
+        )
+
+    def test_send_notification_embeds_remote_image_in_card(self):
+        client = self._build_client()
+        image_upload_response = MagicMock()
+        image_upload_response.success.return_value = True
+        image_upload_response.data = SimpleNamespace(image_key="img_v2_remote")
+        client._api_client, message_api = self._build_message_api(
+            create_response=self._success_response(),
+            image_create_response=image_upload_response,
+        )
+
+        response = MagicMock()
+        response.content = b"png-bytes"
+        response.headers = {"Content-Type": "image/png"}
+
+        with patch("app.modules.feishu.feishu.RequestUtils") as request_utils:
+            request_utils.return_value.get_res.return_value = response
+            result = client.send_notification(
+                Notification(
+                    title="测试标题",
+                    text="测试正文",
+                    image="https://example.com/poster.png",
+                    buttons=[[{"text": "确认", "callback_data": "confirm"}]],
+                ),
+                userid="ou_user_img",
+            )
+
+        self.assertTrue(result["success"])
+        response.close.assert_called_once()
+        client._api_client.im.v1.image.create.assert_called_once()
+        request = message_api.create.call_args.args[0]
+        self.assertEqual(request.request_body.msg_type, "interactive")
+        content = json.loads(request.request_body.content)
+        image_element = content["body"]["elements"][2]
+        self.assertEqual(image_element["tag"], "img")
+        self.assertEqual(image_element["img_key"], "img_v2_remote")
+        self.assertEqual(content["body"]["elements"][-1]["tag"], "column_set")
 
     def test_send_notification_supports_user_id_target(self):
         client = self._build_client()
@@ -261,8 +324,14 @@ class TestFeishu(unittest.TestCase):
         self.assertEqual(request.message_id, "om_456")
         content = json.loads(request.request_body.content)
         self.assertNotIn("card", content)
+        self.assertEqual(content["schema"], "2.0")
         self.assertTrue(content["config"]["update_multi"])
-        self.assertEqual(content["elements"][0]["tag"], "markdown")
+        self.assertEqual(content["body"]["elements"][0]["tag"], "markdown")
+        button = content["body"]["elements"][-1]["columns"][0]["elements"][0]
+        self.assertEqual(
+            button["behaviors"],
+            [{"type": "callback", "value": {"callback_data": "confirm"}}],
+        )
 
     def test_send_notification_replies_when_original_message_id_is_present(self):
         client = self._build_client()
@@ -488,7 +557,7 @@ class TestFeishu(unittest.TestCase):
             )
         )
 
-    def test_send_file_uploads_image_then_sends_image_message(self):
+    def test_send_file_uploads_image_then_sends_mixed_card(self):
         client = self._build_client()
         image_upload_response = MagicMock()
         image_upload_response.success.return_value = True
@@ -501,13 +570,50 @@ class TestFeishu(unittest.TestCase):
         with tempfile.NamedTemporaryFile(suffix=".png") as fp:
             fp.write(b"png-bytes")
             fp.flush()
-            result = client.send_file(file_path=fp.name, userid="ou_user_7")
+            result = client.send_file(
+                file_path=fp.name,
+                userid="ou_user_7",
+                title="图片标题",
+                text="图片说明",
+            )
 
         self.assertTrue(result["success"])
         client._api_client.im.v1.image.create.assert_called_once()
         request = message_api.create.call_args.args[0]
-        self.assertEqual(request.request_body.msg_type, "image")
+        self.assertEqual(request.request_body.msg_type, "interactive")
-        self.assertEqual(json.loads(request.request_body.content)["image_key"], "img_v2_uploaded")
+        content = json.loads(request.request_body.content)
+        self.assertEqual(content["body"]["elements"][0]["content"], "图片标题")
+        self.assertEqual(content["body"]["elements"][1]["content"], "图片说明")
+        self.assertEqual(content["body"]["elements"][2]["img_key"], "img_v2_uploaded")
+
+    def test_send_file_keeps_non_image_file_message_and_caption(self):
+        client = self._build_client()
+        file_upload_response = MagicMock()
+        file_upload_response.success.return_value = True
+        file_upload_response.data = SimpleNamespace(file_key="file_doc")
+        client._api_client, message_api = self._build_message_api(
+            create_response=self._success_response(message_id="om_file"),
+            file_create_response=file_upload_response,
+        )
+
+        with tempfile.NamedTemporaryFile(suffix=".txt") as fp, patch.object(
+            client, "send_text", return_value={"success": True}
+        ) as send_text:
+            fp.write(b"text-bytes")
+            fp.flush()
+            result = client.send_file(
+                file_path=fp.name,
+                userid="ou_user_7",
+                title="文件标题",
+                text="文件说明",
+            )
+
+        self.assertTrue(result["success"])
+        client._api_client.im.v1.file.create.assert_called_once()
+        request = message_api.create.call_args.args[0]
+        self.assertEqual(request.request_body.msg_type, "file")
+        self.assertEqual(json.loads(request.request_body.content)["file_key"], "file_doc")
+        send_text.assert_called_once()
 
     def test_send_voice_uploads_audio_file_and_optionally_sends_caption(self):
         client = self._build_client()
@@ -708,6 +814,66 @@ class TestFeishu(unittest.TestCase):
         client.send_file.assert_called_once()
         client.send_voice.assert_called_once()
 
+    def test_module_post_message_sends_image_card_before_file_attachment(self):
+        module = FeishuModule()
+        conf = SimpleNamespace(name="feishu-main")
+        client = MagicMock()
+
+        with patch.object(module, "get_configs", return_value={"feishu-main": conf}), patch.object(
+            module, "check_message", return_value=True
+        ), patch.object(module, "get_instance", return_value=client):
+            module.post_message(
+                Notification(
+                    file_path="/tmp/demo.txt",
+                    file_name="demo.txt",
+                    image="https://example.com/poster.png",
+                    text="说明",
+                    title="标题",
+                    userid="ou_user",
+                )
+            )
+
+        client.send_notification.assert_called_once()
+        sent_message = client.send_notification.call_args.kwargs["message"]
+        self.assertEqual(sent_message.image, "https://example.com/poster.png")
+        self.assertIsNone(sent_message.file_path)
+        client.send_file.assert_called_once()
+        self.assertIsNone(client.send_file.call_args.kwargs.get("title"))
+        self.assertIsNone(client.send_file.call_args.kwargs.get("text"))
+
+    def test_module_send_direct_message_sends_image_card_before_file_attachment(self):
+        module = FeishuModule()
+        module._channel = MessageChannel.Feishu
+        conf = SimpleNamespace(name="feishu-main")
+        client = MagicMock()
+        client.send_notification.return_value = {
+            "success": True,
+            "message_id": "om_card",
+            "chat_id": "oc_card",
+        }
+        client.send_file.return_value = {"success": True, "message_id": "om_file"}
+
+        with patch.object(module, "get_configs", return_value={"feishu-main": conf}), patch.object(
+            module, "check_message", return_value=True
+        ), patch.object(module, "get_instance", return_value=client):
+            response = module.send_direct_message(
+                Notification(
+                    channel=MessageChannel.Feishu,
+                    source="feishu-main",
+                    file_path="/tmp/demo.txt",
+                    file_name="demo.txt",
+                    image="https://example.com/poster.png",
+                    text="说明",
+                    title="标题",
+                    userid="ou_user",
+                )
+            )
+
+        self.assertTrue(response.success)
+        self.assertEqual(response.message_id, "om_card")
+        client.send_notification.assert_called_once()
+        client.send_file.assert_called_once()
+
     def test_module_post_message_passes_original_message_id_for_reply(self):
         module = FeishuModule()
         conf = SimpleNamespace(name="feishu-main")