yeuimu
9b130097f9
- wecom.py 重写: HTTP POST webhook(替代 aibot WebSocket 按需连接), 接口 send_markdown/send_text 不变, Secret Block wecom-webhook 存 URL - webhook markdown 支持标题/引用/font 三色, 报告改为 '## 标题 + 单行结论 + 引用详情' 紧凑排版, 结论行用 info/warning 色 - alert_failure 改标题式排版; device_check 结论用 font 色, 去掉负载行 - 移除 wecom-aibot-sdk 依赖与 wecom_capture_chatid.py(镜像下次构建生效, 无需立即重建)
84 lines
2.8 KiB
Python
84 lines
2.8 KiB
Python
"""企微群机器人通知层(消息推送 webhook,HTTP POST,无状态)。
|
|
|
|
文档:https://developer.work.weixin.qq.com/document/path/99110
|
|
凭据存 Prefect Secret Block `wecom-webhook`(值为完整 webhook URL,含 key);
|
|
本地调试可用环境变量 WECOM_WEBHOOK_URL 覆盖。
|
|
markdown 支持:1-6 级标题/加粗/链接/引用/<font color="info|comment|warning">(不支持表格);
|
|
单条 ≤4096 字节,超长自动按行分段;频率限制 20 条/分钟。
|
|
|
|
所有函数容错——发送失败只 print 警告并返回 False,绝不中断 flow(对齐 artifacts.py 风格)。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
|
|
import httpx
|
|
|
|
WEBHOOK_BLOCK = "wecom-webhook"
|
|
MAX_MARKDOWN_BYTES = 4000 # 官方上限 4096,留余量
|
|
MAX_TEXT_BYTES = 2000 # text 类型上限 2048,留余量
|
|
|
|
|
|
def load_webhook_url() -> str:
|
|
url = os.environ.get("WECOM_WEBHOOK_URL", "").strip()
|
|
if not url:
|
|
from prefect.blocks.system import Secret
|
|
|
|
url = (Secret.load(WEBHOOK_BLOCK).get() or "").strip()
|
|
if not url:
|
|
raise ValueError(f"webhook 地址为空(Secret Block {WEBHOOK_BLOCK} 或 WECOM_WEBHOOK_URL)")
|
|
return url
|
|
|
|
|
|
def _split_bytes(content: str, limit: int) -> list[str]:
|
|
"""按字节预算切段,优先在换行处切。"""
|
|
if len(content.encode("utf-8")) <= limit:
|
|
return [content]
|
|
parts: list[str] = []
|
|
cur = ""
|
|
for line in content.splitlines(keepends=True):
|
|
if cur and len((cur + line).encode("utf-8")) > limit:
|
|
parts.append(cur)
|
|
cur = line
|
|
else:
|
|
cur += line
|
|
if cur:
|
|
parts.append(cur)
|
|
return parts
|
|
|
|
|
|
def _post(url: str, payload: dict) -> bool:
|
|
resp = httpx.post(url, json=payload, timeout=15)
|
|
try:
|
|
data = resp.json()
|
|
except ValueError:
|
|
data = {}
|
|
if resp.status_code != 200 or data.get("errcode") != 0:
|
|
print(f"[wecom] 发送失败: HTTP {resp.status_code} {data or resp.text[:100]}")
|
|
return False
|
|
return True
|
|
|
|
|
|
def _send(content: str, msgtype: str, limit: int) -> bool:
|
|
try:
|
|
url = load_webhook_url()
|
|
parts = _split_bytes(content, limit)
|
|
ok = all(_post(url, {"msgtype": msgtype, msgtype: {"content": part}}) for part in parts)
|
|
if ok and len(parts) > 1:
|
|
print(f"[wecom] {msgtype} 已发送 ({len(parts)} 段)")
|
|
return ok
|
|
except Exception as err: # noqa: BLE001
|
|
print(f"[wecom] {msgtype} 发送失败: {err}")
|
|
return False
|
|
|
|
|
|
def send_markdown(content: str) -> bool:
|
|
"""发送 markdown 到群(超长自动分段)。失败返回 False,不抛异常。"""
|
|
return _send(content, "markdown", MAX_MARKDOWN_BYTES)
|
|
|
|
|
|
def send_text(content: str) -> bool:
|
|
"""发送纯文本到群。失败返回 False,不抛异常。"""
|
|
return _send(content, "text", MAX_TEXT_BYTES)
|