feat: send clickable Feishu alert cards

AI-Co-Authored-By: Codex

tests/test_business_assistant.py

@@ -128,23 +128,22 @@ class FakeFeishuWebhook:
             }
         )
 
-    async def send_post(
+    async def send_card(
         self,
         webhook_url: str,
         *,
         title: str,
-        content: list[list[dict[str, str]]],
+        lines: list[str],
+        link_url: str = "",
         signing_secret: str = "",
     ) -> None:
         self.sent.append(
             {
                 "webhook_url": webhook_url,
                 "title": title,
-                "content": content,
-                "text": "\n".join(
-                    "".join(segment.get("text", "") for segment in line)
-                    for line in content
-                ),
+                "lines": lines,
+                "text": "\n".join(lines),
+                "link_url": link_url,
                 "signing_secret": signing_secret,
             }
         )
@@ -637,7 +636,13 @@ async def test_every_user_message_notifies_feishu_once(app_modules):
     runtime.feishu = feishu
 
     await runtime.process_update(
-        {"update_id": 101, "business_message": customer_message(message_id=1)}
+        {
+            "update_id": 101,
+            "business_message": customer_message(
+                message_id=1,
+                sent_at=datetime(2026, 8, 24, 2, 3, 4, tzinfo=UTC),
+            ),
+        }
     )
     await runtime.process_update(
         {"update_id": 102, "business_message": customer_message(message_id=1)}
@@ -650,10 +655,9 @@ async def test_every_user_message_notifies_feishu_once(app_modules):
     assert "用户：" in feishu.sent[0]["text"]
     assert "客户：" not in feishu.sent[0]["text"]
     assert "消息标识：" not in feishu.sent[0]["text"]
-    assert feishu.sent[0]["content"][1] == [
-        {"tag": "text", "text": "用户："},
-        {"tag": "a", "text": "客户", "href": "https://t.me/user501"},
-    ]
+    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"]
 
     await dbassistant.update_account_settings(
         "conn-a", {"feishu_message_preview_enabled": True}
@@ -696,10 +700,9 @@ async def test_every_user_message_notifies_feishu_once(app_modules):
         }
     )
     assert len(feishu.sent) == 4
-    assert feishu.sent[3]["content"][1] == [
-        {"tag": "text", "text": "用户："},
-        {"tag": "text", "text": "无用户名用户"},
-    ]
+    assert "用户：无用户名用户" in feishu.sent[3]["lines"]
+    assert feishu.sent[3]["link_url"] == ""
+    assert "点击卡片打开 Telegram 对话" not in feishu.sent[3]["lines"]
 
     await runtime.process_update(
         {
@@ -988,26 +991,51 @@ async def test_feishu_webhook_signature_and_success_contract(app_modules):
     assert session.payload["timestamp"]
     assert session.payload["sign"]
 
-    post_content = [
-        [
-            {"tag": "text", "text": "用户："},
-            {"tag": "a", "text": "测试用户", "href": "https://t.me/test_user"},
-        ]
-    ]
-    await client.send_post(
+    card_lines = ["账号：店主", "用户：测试用户"]
+    await client.send_card(
         "https://open.feishu.cn/open-apis/bot/v2/hook/12345678-abcd-4321-abcd-123456789abc",
         title="消息提醒",
-        content=post_content,
+        lines=card_lines,
+        link_url="https://t.me/test_user",
     )
     assert session.payload == {
-        "msg_type": "post",
-        "content": {
-            "post": {
-                "zh_cn": {
-                    "title": "消息提醒",
-                    "content": post_content,
+        "msg_type": "interactive",
+        "card": {
+            "config": {"wide_screen_mode": True},
+            "header": {
+                "template": "blue",
+                "title": {"tag": "plain_text", "content": "消息提醒"},
+            },
+            "elements": [
+                {
+                    "tag": "div",
+                    "text": {"tag": "plain_text", "content": line},
                 }
-            }
+                for line in card_lines
+            ],
+            "card_link": {"url": "https://t.me/test_user"},
+        },
+    }
+
+    await client.send_card(
+        "https://open.feishu.cn/open-apis/bot/v2/hook/12345678-abcd-4321-abcd-123456789abc",
+        title="消息提醒",
+        lines=["无链接"],
+    )
+    assert session.payload == {
+        "msg_type": "interactive",
+        "card": {
+            "config": {"wide_screen_mode": True},
+            "header": {
+                "template": "blue",
+                "title": {"tag": "plain_text", "content": "消息提醒"},
+            },
+            "elements": [
+                {
+                    "tag": "div",
+                    "text": {"tag": "plain_text", "content": "无链接"},
+                }
+            ],
         },
     }
 

wbb/services/business_assistant.py

@@ -9,6 +9,7 @@ import re
 from contextlib import suppress
 from datetime import UTC, datetime
 from typing import Any
+from zoneinfo import ZoneInfo
 
 from aiohttp import ClientError, ClientSession, ClientTimeout
 
@@ -66,6 +67,7 @@ BUSINESS_ALLOWED_UPDATES = [
     "deleted_business_messages",
 ]
 BUSINESS_REPLY_WINDOW_SECONDS = 24 * 60 * 60
+BEIJING_TIMEZONE = ZoneInfo("Asia/Shanghai")
 
 
 class TelegramBotApiError(RuntimeError):
@@ -108,24 +110,34 @@ class FeishuWebhookClient:
             webhook_url, payload, signing_secret=signing_secret
         )
 
-    async def send_post(
+    async def send_card(
         self,
         webhook_url: str,
         *,
         title: str,
-        content: list[list[dict[str, str]]],
+        lines: list[str],
+        link_url: str = "",
         signing_secret: str = "",
     ) -> None:
-        payload: dict[str, Any] = {
-            "msg_type": "post",
-            "content": {
-                "post": {
-                    "zh_cn": {
-                        "title": str(title)[:100],
-                        "content": content,
-                    }
-                }
+        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
+            ],
+        }
+        if link_url:
+            card["card_link"] = {"url": link_url}
+        payload: dict[str, Any] = {
+            "msg_type": "interactive",
+            "card": card,
         }
         await self._send_payload(
             webhook_url, payload, signing_secret=signing_secret
@@ -1147,43 +1159,36 @@ class BusinessAssistantRuntime:
         preview_text = str(
             message.get("text") or message.get("caption") or ""
         ).strip()
-        content: list[list[dict[str, str]]] = [
-            [{"tag": "text", "text": f"账号：{account_name}"}],
-            [{"tag": "text", "text": "用户："}],
-            [{"tag": "text", "text": f"类型：{message_type}"}],
+        lines = [
+            f"账号：{account_name}",
+            f"用户：{user_name}",
+            f"类型：{message_type}",
         ]
         telegram_username = str(sender.get("username") or "").strip().lstrip("@")
-        if re.fullmatch(r"[A-Za-z0-9_]{1,64}", telegram_username):
-            content[1].append(
-                {
-                    "tag": "a",
-                    "text": user_name,
-                    "href": f"https://t.me/{telegram_username}",
-                }
-            )
-        else:
-            content[1].append({"tag": "text", "text": user_name})
+        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"):
-            content.append(
-                [
-                    {
-                        "tag": "text",
-                        "text": f"内容：{preview_text[:500] if preview_text else '[非文本消息]'}",
-                    }
-                ]
+            lines.append(
+                f"内容：{preview_text[:500] if preview_text else '[非文本消息]'}"
             )
         try:
             message_time = datetime.fromtimestamp(
                 int(message.get("date") or 0), UTC
-            ).strftime("%Y-%m-%d %H:%M:%S UTC")
+            ).astimezone(BEIJING_TIMEZONE).strftime("%Y-%m-%d %H:%M:%S")
         except (OSError, OverflowError, TypeError, ValueError):
             message_time = "未知"
-        content.append([{"tag": "text", "text": f"时间：{message_time}"}])
+        lines.append(f"时间（北京时间）：{message_time}")
+        if telegram_chat_url:
+            lines.append("点击卡片打开 Telegram 对话")
         try:
-            await self.feishu.send_post(
+            await self.feishu.send_card(
                 webhook_url,
                 title="消息提醒",
-                content=content,
+                lines=lines,
+                link_url=telegram_chat_url,
                 signing_secret=str(settings.get("feishu_webhook_signing_secret") or ""),
             )
         except FeishuWebhookError as exc:
@@ -1222,15 +1227,10 @@ class BusinessAssistantRuntime:
             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 账号")
-        await self.feishu.send(
+        await self.feishu.send_card(
             webhook_url,
-            "\n".join(
-                [
-                    "消息提醒",
-                    "飞书群机器人连接测试成功。",
-                    f"账号：{account_name}",
-                ]
-            ),
+            title="消息提醒",
+            lines=["飞书群机器人连接测试成功。", f"账号：{account_name}"],
             signing_secret=str(settings.get("feishu_webhook_signing_secret") or ""),
         )
         return {"ok": True, "account": account_name}