mirror of
https://ghfast.top/https://github.com/aeroxw/tick-stock-panel.git
synced 2026-09-12 22:34:18 +08:00
shy3130
37934091ac
新增 webhook_adapter 飞书群机器人推送 (含签名校验);preferences/settings 增加 feishu_webhook_url/secret 与 webhook_enabled_default (新建规则默认推送) 配置与端点;Monitoring 设置页重构为「推送通知」卡片 (渠道列表+飞书地址折叠 +默认渠道勾选);RuleEditor 规则编辑器支持单条规则勾选飞书。 QMT/ptrade 渠道文案统一改为「待定」。
107 lines
4.0 KiB
Python
107 lines
4.0 KiB
Python
"""Webhook 推送适配器 — 把告警事件推送到外部 IM / 量化软件。
|
|
|
|
职责: 把后端产生的告警事件, 通过用户配置的 Webhook 地址推送到外部。
|
|
目前支持飞书群机器人; QMT / ptrade 等量化通道为待定。
|
|
|
|
飞书自定义机器人接入:
|
|
1. 飞书群 → 群设置 → 群机器人 → 添加「自定义机器人」
|
|
2. 复制生成的 Webhook 地址 (形如 https://open.feishu.cn/open-apis/bot/v2/hook/xxx)
|
|
3. (可选) 安全设置 → 启用「签名校验」, 记录签名密钥(secret)
|
|
4. 填入设置页「飞书 Webhook」配置
|
|
|
|
设计: 失败静默降级, 绝不因推送失败阻断告警主流程 (落盘 / SSE 推送)。
|
|
去重不在本层做, 复用 MonitorRuleEngine 的 cooldown。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import logging
|
|
import time
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# 单次推送最长字符 (飞书单条文本消息上限 30KB, 这里保守截断避免刷屏)
|
|
_MAX_LEN = 500
|
|
|
|
# 飞书自定义机器人 Webhook 前缀 (用于 URL 合法性校验)
|
|
FEISHU_HOOK_PREFIX = "https://open.feishu.cn/open-apis/bot/v2/hook/"
|
|
|
|
|
|
def _truncate(text: str) -> str:
|
|
"""截断超长文本。"""
|
|
text = (text or "").strip()
|
|
return text[:_MAX_LEN] + ("…" if len(text) > _MAX_LEN else "")
|
|
|
|
|
|
def is_valid_feishu_url(url: str) -> bool:
|
|
"""校验是否为合法的飞书自定义机器人 Webhook 地址。"""
|
|
return bool(url) and url.startswith(FEISHU_HOOK_PREFIX)
|
|
|
|
|
|
def _gen_sign(timestamp: str, secret: str) -> str:
|
|
"""计算飞书自定义机器人签名。
|
|
|
|
算法 (官方): 把 `timestamp + "\\n" + secret` 作为签名字符串 (key),
|
|
用 HmacSHA256 计算空字符串的签名结果, 再 Base64 编码。
|
|
"""
|
|
string_to_sign = f"{timestamp}\n{secret}"
|
|
hmac_code = hmac.new(
|
|
string_to_sign.encode("utf-8"),
|
|
digestmod=hashlib.sha256,
|
|
).digest()
|
|
return base64.b64encode(hmac_code).decode("utf-8")
|
|
|
|
|
|
def send_feishu(webhook_url: str, title: str, body: str, secret: str = "") -> bool:
|
|
"""推送一条文本消息到飞书群机器人。
|
|
|
|
Args:
|
|
webhook_url: 飞书自定义机器人 Webhook 地址
|
|
title: 消息标题 (与正文拼接为一条文本)
|
|
body: 消息正文
|
|
secret: 签名密钥 (机器人启用了「签名校验」时必填; 留空则不带签名)
|
|
|
|
Returns:
|
|
True=成功送达, False=失败或 URL 非法。
|
|
失败静默, 不抛异常 (Webhook 是辅助通道, 不能阻断告警主流程)。
|
|
"""
|
|
if not is_valid_feishu_url(webhook_url):
|
|
return False
|
|
|
|
text = _truncate(f"{title}\n{body}".strip())
|
|
if not text:
|
|
return False
|
|
|
|
try:
|
|
import httpx
|
|
|
|
payload: dict = {"msg_type": "text", "content": {"text": text}}
|
|
# 启用签名校验时, 请求体须带 timestamp + sign (秒级时间戳)
|
|
if secret:
|
|
timestamp = str(int(time.time()))
|
|
payload["timestamp"] = timestamp
|
|
payload["sign"] = _gen_sign(timestamp, secret)
|
|
|
|
resp = httpx.post(webhook_url, json=payload, timeout=5.0)
|
|
# 飞书成功响应: {"code":0,"msg":"success"} (或 StatusCode 200 + Extra)
|
|
if resp.status_code == 200:
|
|
try:
|
|
data = resp.json()
|
|
# code=0 表示飞书业务侧成功; 部分版本无 code 字段则按 msg 判断
|
|
if isinstance(data, dict):
|
|
code = data.get("code", data.get("StatusCode", 0))
|
|
if code == 0:
|
|
return True
|
|
logger.debug("飞书推送业务失败: %s", data)
|
|
return False
|
|
except ValueError:
|
|
# 非 JSON 响应但 HTTP 200, 视为成功
|
|
return True
|
|
logger.debug("飞书推送 HTTP %s: %s", resp.status_code, resp.text[:200])
|
|
return False
|
|
except Exception as e: # noqa: BLE001
|
|
logger.debug("飞书 Webhook 推送失败: %s", e)
|
|
return False
|