97 lines
2.8 KiB
Python
97 lines
2.8 KiB
Python
|
"""飞书机器人通知 — 通过 Webhook 推送消息到飞书群聊"""
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import logging
|
|||
|
|
from typing import Optional
|
|||
|
|
|
|||
|
|
import httpx
|
|||
|
|
|
|||
|
|
logger = logging.getLogger(__name__)
|
|||
|
|
|
|||
|
|
FEISHU_TIMEOUT_SEC = 10
|
|||
|
|
|
|||
|
|
|
|||
|
|
def send_feishu_text(webhook_url: str, text: str) -> bool:
|
|||
|
|
"""发送纯文本消息到飞书群机器人。
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
webhook_url: 飞书机器人 webhook 地址
|
|||
|
|
text: 消息文本
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
是否发送成功
|
|||
|
|
"""
|
|||
|
|
payload = {"msg_type": "text", "content": {"text": text}}
|
|||
|
|
return _do_send(webhook_url, payload)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def send_feishu_card(
|
|||
|
|
webhook_url: str,
|
|||
|
|
title: str,
|
|||
|
|
body: str,
|
|||
|
|
status: str = "info",
|
|||
|
|
detail_link: Optional[str] = None,
|
|||
|
|
) -> bool:
|
|||
|
|
"""发送消息卡片到飞书群机器人。
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
webhook_url: 飞书机器人 webhook 地址
|
|||
|
|
title: 卡片标题
|
|||
|
|
body: 卡片正文（支持 Markdown）
|
|||
|
|
status: 状态 — info / success / failed
|
|||
|
|
detail_link: 详情链接（可选）
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
是否发送成功
|
|||
|
|
"""
|
|||
|
|
color_map = {"success": "green", "failed": "red", "info": "blue"}
|
|||
|
|
color = color_map.get(status, "blue")
|
|||
|
|
|
|||
|
|
elements = [
|
|||
|
|
{"tag": "markdown", "content": body},
|
|||
|
|
]
|
|||
|
|
if detail_link:
|
|||
|
|
elements.append({
|
|||
|
|
"tag": "action",
|
|||
|
|
"actions": [{"tag": "button", "text": {"tag": "plain_text", "content": "查看详情"}, "url": detail_link, "type": "default"}],
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
payload = {
|
|||
|
|
"msg_type": "interactive",
|
|||
|
|
"card": {
|
|||
|
|
"header": {
|
|||
|
|
"title": {"tag": "plain_text", "content": title},
|
|||
|
|
"template": color,
|
|||
|
|
},
|
|||
|
|
"elements": elements,
|
|||
|
|
},
|
|||
|
|
}
|
|||
|
|
return _do_send(webhook_url, payload)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _do_send(webhook_url: str, payload: dict) -> bool:
|
|||
|
|
"""底层 POST 发送，统一异常处理。"""
|
|||
|
|
if not webhook_url or not webhook_url.startswith("https://open.feishu.cn/"):
|
|||
|
|
logger.warning("飞书 webhook URL 无效: %s", webhook_url[:50] if webhook_url else "None")
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
with httpx.Client(timeout=FEISHU_TIMEOUT_SEC) as client:
|
|||
|
|
resp = client.post(webhook_url, json=payload)
|
|||
|
|
result = resp.json()
|
|||
|
|
if resp.is_success and result.get("code") == 0:
|
|||
|
|
logger.info("飞书通知发送成功: %s", result.get("msg"))
|
|||
|
|
return True
|
|||
|
|
else:
|
|||
|
|
logger.warning(
|
|||
|
|
"飞书通知发送失败: status=%s code=%s msg=%s",
|
|||
|
|
resp.status_code, result.get("code"), result.get("msg"),
|
|||
|
|
)
|
|||
|
|
return False
|
|||
|
|
except httpx.TimeoutException:
|
|||
|
|
logger.warning("飞书通知发送超时: %s", webhook_url[:50])
|
|||
|
|
return False
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.warning("飞书通知发送异常: %s", e)
|
|||
|
|
return False
|