72 lines
2.7 KiB
Python
72 lines
2.7 KiB
Python
"""Feishu webhook transport for system error alerts."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
|
|
import requests
|
|
|
|
|
|
def _webhook_url() -> str:
|
|
return os.getenv("ALPHAX_SYSTEM_ERROR_FEISHU_WEBHOOK", "").strip()
|
|
|
|
|
|
def _runtime_env() -> str:
|
|
return str(os.getenv("ALPHAX_ENV") or "dev").strip().lower() or "dev"
|
|
|
|
|
|
def _enabled() -> bool:
|
|
raw = os.getenv("ALPHAX_SYSTEM_ERROR_FEISHU_ENABLED", "1")
|
|
return str(raw).strip().lower() in ("1", "true", "yes", "on")
|
|
|
|
|
|
def push_system_error_alert(item: dict) -> tuple[bool, object]:
|
|
"""Send one system error alert. This function must never raise."""
|
|
try:
|
|
if not _enabled():
|
|
return False, "system error alert disabled"
|
|
webhook = _webhook_url()
|
|
if not webhook:
|
|
return False, "ALPHAX_SYSTEM_ERROR_FEISHU_WEBHOOK not configured"
|
|
title = f"AlphaX 系统错误 #{item.get('id') or '--'}"
|
|
env = _runtime_env()
|
|
if env not in ("prod", "production"):
|
|
title = f"[{env.upper()}] {title}"
|
|
message = str(item.get("message") or "--")
|
|
stack = str(item.get("stack_trace") or "")
|
|
if len(stack) > 900:
|
|
stack = stack[:900] + "\n..."
|
|
fields = [
|
|
("级别", item.get("level") or "--"),
|
|
("来源", item.get("source") or "--"),
|
|
("类型", item.get("error_type") or "--"),
|
|
("状态", item.get("status_code") or "--"),
|
|
("路径", item.get("request_path") or "--"),
|
|
("时间", item.get("created_at") or "--"),
|
|
("指纹", item.get("fingerprint") or "--"),
|
|
]
|
|
content = "\n".join(f"**{label}**: {value}" for label, value in fields)
|
|
card = {
|
|
"config": {"wide_screen_mode": True},
|
|
"header": {
|
|
"template": "red" if str(item.get("level") or "") == "error" else "yellow",
|
|
"title": {"tag": "plain_text", "content": title},
|
|
},
|
|
"elements": [
|
|
{"tag": "div", "text": {"tag": "lark_md", "content": content}},
|
|
{"tag": "div", "text": {"tag": "lark_md", "content": f"**消息**: {message[:900]}"}},
|
|
{"tag": "note", "elements": [{"tag": "plain_text", "content": "请到日志中心 /logs 查看完整上下文与堆栈。"}]},
|
|
],
|
|
}
|
|
if stack:
|
|
card["elements"].append({"tag": "div", "text": {"tag": "lark_md", "content": f"```text\n{stack}\n```"}})
|
|
resp = requests.post(webhook, json={"msg_type": "interactive", "card": card}, timeout=10)
|
|
result = resp.json()
|
|
ok = resp.status_code == 200 and result.get("StatusCode") == 0
|
|
return ok, result
|
|
except Exception as exc:
|
|
return False, str(exc)
|
|
|
|
|
|
__all__ = ["push_system_error_alert"]
|