2260eed2
 tangwang
推送报警到微信群webhook
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
|
#!/usr/bin/env python
"""
Lightweight Enterprise WeChat webhook sender for service monitor alerts.
This module is intentionally small and focused so that Bash-based monitors
can invoke it without pulling in the full application stack.
Usage example:
python scripts/wechat_alert.py --service backend --level error --message "backend restarted"
"""
import argparse
import json
import os
import sys
from datetime import datetime
import requests
def get_webhook_url() -> str:
"""
Resolve webhook URL from environment, with optional default.
Priority:
1. SERVICE_MONITOR_WECHAT_WEBHOOK
2. Built-in default (provided by ops)
"""
env_url = os.getenv("SERVICE_MONITOR_WECHAT_WEBHOOK", "").strip()
if env_url:
return env_url
# Fallback to the URL provided in ops configuration.
return (
"https://qyapi.weixin.qq.com/cgi-bin/webhook/send?"
"key=2d9e38ef-9242-4e2e-82cc-fab060322871"
)
def build_text_payload(message: str) -> dict:
return {
"msgtype": "text",
"text": {
"content": message,
},
}
def send_wechat_message(message: str) -> None:
url = get_webhook_url()
if not url:
# No webhook configured; fail silently to avoid breaking callers.
return
payload = build_text_payload(message)
headers = {"Content-Type": "application/json"}
try:
resp = requests.post(url, headers=headers, data=json.dumps(payload), timeout=5)
except Exception:
# Swallow all exceptions to avoid impacting the caller.
return
try:
if resp.status_code != 200:
return
data = resp.json()
# errcode == 0 means success per WeCom docs
if int(data.get("errcode", -1)) != 0:
return
except Exception:
return
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="Send Enterprise WeChat alert message")
parser.add_argument("--service", help="service name", default="")
parser.add_argument("--level", help="alert level (info|warn|error)", default="info")
parser.add_argument(
"--message",
required=True,
help="alert message body (short, human-readable)",
)
args = parser.parse_args(argv)
ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
parts = [
f"【服务监控告警】",
f"时间: {ts}",
]
if args.service:
parts.append(f"服务: {args.service}")
if args.level:
parts.append(f"级别: {args.level}")
parts.append(f"详情: {args.message}")
full_message = "\n".join(parts)
send_wechat_message(full_message)
return 0
if __name__ == "__main__":
raise SystemExit(main())
|