dekun
69 lines
2.0 KiB
Python
69 lines
2.0 KiB
Python
"""中控企业微信 Webhook 推送."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from typing import Any
|
|
|
|
import requests
|
|
|
|
|
|
def hub_wechat_enabled() -> bool:
|
|
raw = (os.getenv("HUB_WECHAT_ENABLED") or "").strip().lower()
|
|
if raw in ("0", "false", "no", "off"):
|
|
return False
|
|
if raw in ("1", "true", "yes", "on"):
|
|
return True
|
|
# 未显式开关时:有有效 webhook 即视为开启
|
|
return bool(hub_wechat_webhook())
|
|
|
|
|
|
def hub_wechat_webhook() -> str:
|
|
return (os.getenv("HUB_WECHAT_WEBHOOK") or "").strip()
|
|
|
|
|
|
def hub_alert_cooldown_sec() -> int:
|
|
try:
|
|
return max(60, int(float(os.getenv("HUB_ALERT_COOLDOWN_SEC") or 600)))
|
|
except (TypeError, ValueError):
|
|
return 600
|
|
|
|
|
|
def hub_health_fail_threshold() -> int:
|
|
try:
|
|
return max(1, int(float(os.getenv("HUB_HEALTH_FAIL_THRESHOLD") or 2)))
|
|
except (TypeError, ValueError):
|
|
return 2
|
|
|
|
|
|
def send_hub_wechat(content: str, *, timeout: float = 8.0) -> dict[str, Any]:
|
|
"""发送文本消息;返回 {ok, msg}."""
|
|
if not hub_wechat_enabled():
|
|
return {"ok": False, "msg": "企业微信推送未启用"}
|
|
url = hub_wechat_webhook()
|
|
if not url or "replace-me" in url:
|
|
return {"ok": False, "msg": "未配置有效 Webhook"}
|
|
body = str(content or "").strip()
|
|
if not body:
|
|
return {"ok": False, "msg": "消息为空"}
|
|
full = f"【期权中控】\n{body}"
|
|
try:
|
|
resp = requests.post(
|
|
url,
|
|
json={"msgtype": "text", "text": {"content": full}},
|
|
timeout=timeout,
|
|
)
|
|
data = {}
|
|
try:
|
|
data = resp.json()
|
|
except Exception:
|
|
data = {}
|
|
if resp.status_code == 200 and int(data.get("errcode", -1)) == 0:
|
|
return {"ok": True, "msg": "已发送"}
|
|
return {
|
|
"ok": False,
|
|
"msg": (data.get("errmsg") if isinstance(data, dict) else None)
|
|
or f"HTTP {resp.status_code}",
|
|
}
|
|
except Exception as e:
|
|
return {"ok": False, "msg": str(e) or "发送失败"}
|