系统设置增加中控企业微信告警，后台巡检离线/故障并冷却推送。
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
+72
-55
@@ -1,55 +1,72 @@
|
||||
"""Hub .env 读写(仅中控账号等少量键)."""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Dict, List
|
||||
|
||||
_KEY_RE = re.compile(r"^([A-Za-z_][A-Za-z0-9_]*)=(.*)$")
|
||||
|
||||
|
||||
def _parse_line(line: str) -> tuple[str | None, str]:
|
||||
raw = line.rstrip("\n\r")
|
||||
if not raw.strip() or raw.lstrip().startswith("#"):
|
||||
return None, raw
|
||||
m = _KEY_RE.match(raw)
|
||||
if not m:
|
||||
return None, raw
|
||||
return m.group(1), raw
|
||||
|
||||
|
||||
def apply_env_updates(path: Path | str, updates: Dict[str, str]) -> List[str]:
|
||||
"""更新或追加 .env 键;返回实际写入的 key 列表.不删其它行."""
|
||||
env_path = Path(path)
|
||||
wanted = {str(k).strip(): str(v) for k, v in updates.items() if str(k).strip()}
|
||||
if not wanted:
|
||||
return []
|
||||
|
||||
lines: list[str] = []
|
||||
if env_path.is_file():
|
||||
lines = env_path.read_text(encoding="utf-8").splitlines(keepends=True)
|
||||
|
||||
seen: set[str] = set()
|
||||
out: list[str] = []
|
||||
changed: list[str] = []
|
||||
|
||||
for line in lines:
|
||||
key, _ = _parse_line(line)
|
||||
if key and key in wanted:
|
||||
out.append(f"{key}={wanted[key]}\n")
|
||||
seen.add(key)
|
||||
changed.append(key)
|
||||
else:
|
||||
if line.endswith("\n") or line.endswith("\r"):
|
||||
out.append(line if line.endswith("\n") else line + "\n")
|
||||
else:
|
||||
out.append(line + "\n")
|
||||
|
||||
for key, val in wanted.items():
|
||||
if key not in seen:
|
||||
out.append(f"{key}={val}\n")
|
||||
changed.append(key)
|
||||
|
||||
env_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
env_path.write_text("".join(out), encoding="utf-8")
|
||||
return changed
|
||||
"""Hub .env 读写(仅中控账号等少量键)."""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Dict, List
|
||||
|
||||
_KEY_RE = re.compile(r"^([A-Za-z_][A-Za-z0-9_]*)=(.*)$")
|
||||
|
||||
|
||||
def _parse_line(line: str) -> tuple[str | None, str]:
|
||||
raw = line.rstrip("\n\r")
|
||||
if not raw.strip() or raw.lstrip().startswith("#"):
|
||||
return None, raw
|
||||
m = _KEY_RE.match(raw)
|
||||
if not m:
|
||||
return None, raw
|
||||
return m.group(1), raw
|
||||
|
||||
|
||||
def apply_env_updates(path: Path | str, updates: Dict[str, str]) -> List[str]:
|
||||
"""更新或追加 .env 键;返回实际写入的 key 列表.不删其它行."""
|
||||
env_path = Path(path)
|
||||
wanted = {str(k).strip(): str(v) for k, v in updates.items() if str(k).strip()}
|
||||
if not wanted:
|
||||
return []
|
||||
|
||||
lines: list[str] = []
|
||||
if env_path.is_file():
|
||||
lines = env_path.read_text(encoding="utf-8").splitlines(keepends=True)
|
||||
|
||||
seen: set[str] = set()
|
||||
out: list[str] = []
|
||||
changed: list[str] = []
|
||||
|
||||
for line in lines:
|
||||
key, _ = _parse_line(line)
|
||||
if key and key in wanted:
|
||||
out.append(f"{key}={wanted[key]}\n")
|
||||
seen.add(key)
|
||||
changed.append(key)
|
||||
else:
|
||||
if line.endswith("\n") or line.endswith("\r"):
|
||||
out.append(line if line.endswith("\n") else line + "\n")
|
||||
else:
|
||||
out.append(line + "\n")
|
||||
|
||||
for key, val in wanted.items():
|
||||
if key not in seen:
|
||||
out.append(f"{key}={val}\n")
|
||||
changed.append(key)
|
||||
|
||||
env_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
env_path.write_text("".join(out), encoding="utf-8")
|
||||
return changed
|
||||
|
||||
|
||||
def env_get(path: Path | str, key: str, default: str = "") -> str:
|
||||
env_path = Path(path)
|
||||
if not env_path.is_file():
|
||||
return default
|
||||
want = (key or "").strip()
|
||||
if not want:
|
||||
return default
|
||||
for line in env_path.read_text(encoding="utf-8").splitlines():
|
||||
k, _ = _parse_line(line)
|
||||
if k == want:
|
||||
m = _KEY_RE.match(line.rstrip("\n\r"))
|
||||
if m:
|
||||
return m.group(2)
|
||||
return default
|
||||
return default
|
||||
|
||||
@@ -0,0 +1,129 @@
|
||||
"""客户端在线健康巡检与企业微信告警(冷却去重)."""
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
from lib.wechat_lib import (
|
||||
hub_alert_cooldown_sec,
|
||||
hub_health_fail_threshold,
|
||||
hub_wechat_enabled,
|
||||
send_hub_wechat,
|
||||
)
|
||||
|
||||
_lock = threading.Lock()
|
||||
# instance_id -> {online, fail_count, last_alert_at, last_reason, name}
|
||||
_state: dict[Any, dict[str, Any]] = {}
|
||||
|
||||
|
||||
def _now() -> float:
|
||||
return time.time()
|
||||
|
||||
|
||||
def reset_health_state() -> None:
|
||||
with _lock:
|
||||
_state.clear()
|
||||
|
||||
|
||||
def inspect_overview_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||||
"""根据总览拉取结果更新状态并按需告警.返回本次新发出的告警摘要."""
|
||||
if not rows:
|
||||
return []
|
||||
if not hub_wechat_enabled():
|
||||
# 仍更新在线状态,便于开关打开后立刻有基线
|
||||
with _lock:
|
||||
for row in rows:
|
||||
iid = row.get("id")
|
||||
if iid is None:
|
||||
continue
|
||||
online = bool(row.get("online"))
|
||||
st = _state.get(iid) or {}
|
||||
st.update(
|
||||
{
|
||||
"online": online,
|
||||
"fail_count": 0 if online else int(st.get("fail_count") or 0) + 1,
|
||||
"name": row.get("name") or st.get("name") or str(iid),
|
||||
"last_reason": "" if online else (row.get("error") or "离线"),
|
||||
}
|
||||
)
|
||||
_state[iid] = st
|
||||
return []
|
||||
|
||||
threshold = hub_health_fail_threshold()
|
||||
cooldown = hub_alert_cooldown_sec()
|
||||
sent: list[dict[str, Any]] = []
|
||||
now = _now()
|
||||
|
||||
with _lock:
|
||||
for row in rows:
|
||||
iid = row.get("id")
|
||||
if iid is None:
|
||||
continue
|
||||
name = str(row.get("name") or iid)
|
||||
online = bool(row.get("online"))
|
||||
err = str(row.get("error") or "").strip()
|
||||
st = _state.get(iid) or {
|
||||
"online": None,
|
||||
"fail_count": 0,
|
||||
"last_alert_at": 0.0,
|
||||
"last_reason": "",
|
||||
"name": name,
|
||||
}
|
||||
prev_online = st.get("online")
|
||||
st["name"] = name
|
||||
|
||||
if online:
|
||||
# 从离线恢复
|
||||
if prev_online is False:
|
||||
if now - float(st.get("last_alert_at") or 0) >= cooldown:
|
||||
msg = f"客户端恢复在线\n名称:{name}\nID:{iid}"
|
||||
result = send_hub_wechat(msg)
|
||||
if result.get("ok"):
|
||||
st["last_alert_at"] = now
|
||||
sent.append({"instance_id": iid, "kind": "online", "msg": msg})
|
||||
st["online"] = True
|
||||
st["fail_count"] = 0
|
||||
st["last_reason"] = ""
|
||||
else:
|
||||
reason = err or "离线"
|
||||
st["fail_count"] = int(st.get("fail_count") or 0) + 1
|
||||
st["online"] = False
|
||||
st["last_reason"] = reason
|
||||
# 首次已知状态若就是离线,也累计;达到阈值才告警
|
||||
if st["fail_count"] >= threshold:
|
||||
if now - float(st.get("last_alert_at") or 0) >= cooldown:
|
||||
kind = "fault" if reason and reason not in ("离线", "超时") else "offline"
|
||||
title = "客户端故障" if kind == "fault" else "客户端离线"
|
||||
msg = "\n".join(
|
||||
[
|
||||
title,
|
||||
f"名称:{name}",
|
||||
f"ID:{iid}",
|
||||
f"原因:{reason}",
|
||||
f"连续失败:{st['fail_count']} 次",
|
||||
]
|
||||
)
|
||||
result = send_hub_wechat(msg)
|
||||
if result.get("ok"):
|
||||
st["last_alert_at"] = now
|
||||
sent.append({"instance_id": iid, "kind": kind, "msg": msg})
|
||||
_state[iid] = st
|
||||
return sent
|
||||
|
||||
|
||||
def health_snapshot() -> list[dict[str, Any]]:
|
||||
with _lock:
|
||||
out = []
|
||||
for iid, st in _state.items():
|
||||
out.append(
|
||||
{
|
||||
"id": iid,
|
||||
"name": st.get("name"),
|
||||
"online": st.get("online"),
|
||||
"fail_count": st.get("fail_count") or 0,
|
||||
"last_reason": st.get("last_reason") or "",
|
||||
"last_alert_at": st.get("last_alert_at") or 0,
|
||||
}
|
||||
)
|
||||
return out
|
||||
@@ -0,0 +1,68 @@
|
||||
"""中控企业微信 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 "发送失败"}
|
||||
Reference in New Issue
Block a user