Forráskód Böngészése

feat: refine Feishu message alert cards

AI-Co-Authored-By: Codex
chendeben 1 hónapja
szülő
commit
5cf78debc7

+ 1 - 1
admin-web/src/pages/BusinessAssistant.tsx

@@ -692,7 +692,7 @@ export default function BusinessAssistantPage() {
                           </Form.Item>
                         </Col>
                         <Col xs={24} md={8}>
-                          <Form.Item name="feishu_message_preview_enabled" label="包含消息内容" valuePropName="checked" extra="关闭时只发送账号、客户、消息类型和标识。">
+                          <Form.Item name="feishu_message_preview_enabled" label="包含消息内容" valuePropName="checked" extra="关闭时卡片会显示“消息已隐藏”。">
                             <Switch checkedChildren="包含" unCheckedChildren="隐藏" />
                           </Form.Item>
                         </Col>

+ 24 - 11
tests/test_business_assistant.py

@@ -651,13 +651,12 @@ async def test_every_user_message_notifies_feishu_once(app_modules):
     assert feishu.sent[0]["webhook_url"] == webhook_url
     assert feishu.sent[0]["signing_secret"] == "secret"
     assert "营业时间是什么" not in feishu.sent[0]["text"]
-    assert feishu.sent[0]["title"] == "消息提醒"
-    assert "用户:" in feishu.sent[0]["text"]
-    assert "客户:" not in feishu.sent[0]["text"]
-    assert "消息标识:" not in feishu.sent[0]["text"]
+    assert feishu.sent[0]["title"] == "客户"
+    assert feishu.sent[0]["lines"][0] == "消息已隐藏"
+    assert "账号:" not in feishu.sent[0]["text"]
+    assert "用户:" not in feishu.sent[0]["text"]
     assert feishu.sent[0]["link_url"] == "https://t.me/user501"
-    assert "时间(北京时间):2026-08-24 10:03:04" in feishu.sent[0]["lines"]
-    assert "点击卡片打开 Telegram 对话" in feishu.sent[0]["lines"]
+    assert "2026-08-24 10:03:04" in feishu.sent[0]["lines"]
 
     await dbassistant.update_account_settings(
         "conn-a", {"feishu_message_preview_enabled": True}
@@ -685,8 +684,7 @@ async def test_every_user_message_notifies_feishu_once(app_modules):
         }
     )
     assert len(feishu.sent) == 3
-    assert "类型:非文本" in feishu.sent[2]["text"]
-    assert "内容:图片说明" in feishu.sent[2]["text"]
+    assert "图片说明" in feishu.sent[2]["text"]
 
     await runtime.process_update(
         {
@@ -700,9 +698,8 @@ async def test_every_user_message_notifies_feishu_once(app_modules):
         }
     )
     assert len(feishu.sent) == 4
-    assert "用户:无用户名用户" in feishu.sent[3]["lines"]
+    assert feishu.sent[3]["title"] == "无用户名用户"
     assert feishu.sent[3]["link_url"] == ""
-    assert "点击卡片打开 Telegram 对话" not in feishu.sent[3]["lines"]
 
     await runtime.process_update(
         {
@@ -1012,8 +1009,24 @@ async def test_feishu_webhook_signature_and_success_contract(app_modules):
                     "text": {"tag": "plain_text", "content": line},
                 }
                 for line in card_lines
+            ]
+            + [
+                {"tag": "hr"},
+                {
+                    "tag": "action",
+                    "actions": [
+                        {
+                            "tag": "button",
+                            "text": {
+                                "tag": "plain_text",
+                                "content": "打开 Telegram 对话",
+                            },
+                            "type": "primary",
+                            "url": "https://t.me/test_user",
+                        }
+                    ],
+                },
             ],
-            "card_link": {"url": "https://t.me/test_user"},
         },
     }
 

+ 37 - 35
wbb/services/business_assistant.py

@@ -119,22 +119,41 @@ class FeishuWebhookClient:
         link_url: str = "",
         signing_secret: str = "",
     ) -> None:
+        elements: list[dict[str, Any]] = [
+            {
+                "tag": "div",
+                "text": {"tag": "plain_text", "content": str(line)[:1000]},
+            }
+            for line in lines
+        ]
+        if link_url:
+            elements.extend(
+                [
+                    {"tag": "hr"},
+                    {
+                        "tag": "action",
+                        "actions": [
+                            {
+                                "tag": "button",
+                                "text": {
+                                    "tag": "plain_text",
+                                    "content": "打开 Telegram 对话",
+                                },
+                                "type": "primary",
+                                "url": link_url,
+                            }
+                        ],
+                    },
+                ]
+            )
         card: dict[str, Any] = {
             "config": {"wide_screen_mode": True},
             "header": {
                 "template": "blue",
                 "title": {"tag": "plain_text", "content": str(title)[:100]},
             },
-            "elements": [
-                {
-                    "tag": "div",
-                    "text": {"tag": "plain_text", "content": str(line)[:1000]},
-                }
-                for line in lines
-            ],
+            "elements": elements,
         }
-        if link_url:
-            card["card_link"] = {"url": link_url}
         payload: dict[str, Any] = {
             "msg_type": "interactive",
             "card": card,
@@ -1010,12 +1029,10 @@ class BusinessAssistantRuntime:
             metadata={"content_type": "text" if text else "unsupported"},
         )
         await self._notify_feishu_customer_message(
-            connection=connection,
             conversation=conversation,
             message=message,
             sender=sender,
             settings=settings,
-            text=text,
         )
         if not connection.get("is_enabled") or not settings.get("assistant_enabled"):
             return
@@ -1124,12 +1141,10 @@ class BusinessAssistantRuntime:
     async def _notify_feishu_customer_message(
         self,
         *,
-        connection: dict[str, Any],
         conversation: dict[str, Any],
         message: dict[str, Any],
         sender: dict[str, Any],
         settings: dict[str, Any],
-        text: str,
     ) -> None:
         webhook_url = str(settings.get("feishu_webhook_url") or "")
         if not settings.get("feishu_webhook_enabled") or not webhook_url:
@@ -1149,44 +1164,31 @@ class BusinessAssistantRuntime:
                 if sender.get("username")
                 else str((message.get("chat") or {}).get("id") or "未知")
             )
-        account = connection.get("user") or {}
-        account_name = " ".join(
-            str(item).strip()
-            for item in (account.get("first_name"), account.get("last_name"))
-            if str(item or "").strip()
-        ) or (f"@{account.get('username')}" if account.get("username") else "Business 账号")
-        message_type = "文本" if text else "非文本"
         preview_text = str(
             message.get("text") or message.get("caption") or ""
         ).strip()
-        lines = [
-            f"账号:{account_name}",
-            f"用户:{user_name}",
-            f"类型:{message_type}",
-        ]
+        if settings.get("feishu_message_preview_enabled"):
+            message_content = preview_text[:500] if preview_text else "[非文本消息]"
+        else:
+            message_content = "消息已隐藏"
+        lines = [message_content]
         telegram_username = str(sender.get("username") or "").strip().lstrip("@")
         telegram_chat_url = (
             f"https://t.me/{telegram_username}"
             if re.fullmatch(r"[A-Za-z0-9_]{1,64}", telegram_username)
             else ""
         )
-        if settings.get("feishu_message_preview_enabled"):
-            lines.append(
-                f"内容:{preview_text[:500] if preview_text else '[非文本消息]'}"
-            )
         try:
             message_time = datetime.fromtimestamp(
                 int(message.get("date") or 0), UTC
             ).astimezone(BEIJING_TIMEZONE).strftime("%Y-%m-%d %H:%M:%S")
         except (OSError, OverflowError, TypeError, ValueError):
             message_time = "未知"
-        lines.append(f"时间(北京时间):{message_time}")
-        if telegram_chat_url:
-            lines.append("点击卡片打开 Telegram 对话")
+        lines.append(message_time)
         try:
             await self.feishu.send_card(
                 webhook_url,
-                title="消息提醒",
+                title=user_name,
                 lines=lines,
                 link_url=telegram_chat_url,
                 signing_secret=str(settings.get("feishu_webhook_signing_secret") or ""),
@@ -1229,8 +1231,8 @@ class BusinessAssistantRuntime:
         ) or (f"@{account.get('username')}" if account.get("username") else "Business 账号")
         await self.feishu.send_card(
             webhook_url,
-            title="消息提醒",
-            lines=["飞书群机器人连接测试成功。", f"账号:{account_name}"],
+            title="测试消息",
+            lines=["飞书群机器人连接测试成功。"],
             signing_secret=str(settings.get("feishu_webhook_signing_secret") or ""),
         )
         return {"ok": True, "account": account_name}