573 lines
17 KiB
Python
Executable File
573 lines
17 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Transition-based health monitor for the local WeChat GUI automation chain."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
HOME = Path.home()
|
|
STATE_DIR = HOME / ".local" / "state" / "wechat-health-watch"
|
|
STATE_FILE = STATE_DIR / "state.json"
|
|
NOTIFICATION_CONFIG = Path(
|
|
os.environ.get(
|
|
"JWEBOT_HEALTH_NOTIFICATION_CONFIG",
|
|
str(
|
|
HOME
|
|
/ ".config"
|
|
/ "jwebot"
|
|
/ "current"
|
|
/ "generated"
|
|
/ "notification.json"
|
|
),
|
|
)
|
|
).expanduser()
|
|
RECORDER_STATE = (
|
|
HOME / ".local" / "state" / "wechat-adaptive-ocr" / "recorder-state.json"
|
|
)
|
|
FAILED_DIR = HOME / ".local" / "state" / "wechat-adaptive-ocr" / "failed"
|
|
DISPLAY = ":1"
|
|
XAUTHORITY = HOME / ".Xauthority"
|
|
XDG_RUNTIME_DIR = Path(f"/run/user/{os.getuid()}")
|
|
DBUS_ADDRESS = f"unix:path={XDG_RUNTIME_DIR}/bus"
|
|
|
|
SYSTEM_SERVICES = (
|
|
"tigervncserver@:1.service",
|
|
"wechat-novnc.service",
|
|
"openclaw-tunnel.service",
|
|
)
|
|
USER_SERVICES = ("wechat-ocr-adaptive-recorder.service",)
|
|
SERVICE_CODES = {
|
|
"tigervncserver@:1.service": "vnc_service",
|
|
"wechat-novnc.service": "novnc_service",
|
|
"openclaw-tunnel.service": "tunnel_service",
|
|
}
|
|
USER_SERVICE_CODES = {
|
|
"wechat-ocr-adaptive-recorder.service": "adaptive_recorder_service",
|
|
}
|
|
CODE_LABELS = {
|
|
"vnc_service": "VNC 服务",
|
|
"novnc_service": "noVNC 服务",
|
|
"adaptive_recorder_service": "自适应 OCR 录制器",
|
|
"tunnel_service": "OpenClaw 隧道",
|
|
"wechat_process": "微信客户端进程",
|
|
"wechat_window": "微信主窗口",
|
|
"wechat_login_required": "微信登录状态",
|
|
"recorder_state": "OCR 录制状态",
|
|
"recorder_stale": "OCR 捕获心跳",
|
|
"responder_failed": "消息回复投递",
|
|
}
|
|
|
|
RECORDER_FRESHNESS_SECONDS = 180
|
|
MIN_READY_WINDOW_WIDTH = 700
|
|
MIN_READY_WINDOW_HEIGHT = 500
|
|
START_COOLDOWN_SECONDS = 15 * 60
|
|
NOTIFICATION_COOLDOWN_SECONDS = 15 * 60
|
|
REQUIRED_FAILURE_SCANS = 2
|
|
|
|
|
|
class NotificationDeliveryUncertain(RuntimeError):
|
|
"""The platform may have accepted a notification before CLI confirmation."""
|
|
|
|
|
|
def run(
|
|
command: list[str],
|
|
*,
|
|
timeout: int = 30,
|
|
data: str | None = None,
|
|
) -> subprocess.CompletedProcess[str]:
|
|
return subprocess.run(
|
|
command,
|
|
input=data,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
timeout=timeout,
|
|
)
|
|
|
|
|
|
def load_json(path: Path) -> dict[str, Any]:
|
|
try:
|
|
value = json.loads(path.read_text(encoding="utf-8"))
|
|
except (FileNotFoundError, OSError, json.JSONDecodeError):
|
|
return {}
|
|
return value if isinstance(value, dict) else {}
|
|
|
|
|
|
def atomic_json(path: Path, value: dict[str, Any]) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
|
|
temporary.write_text(
|
|
json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
os.chmod(temporary, 0o600)
|
|
temporary.replace(path)
|
|
|
|
|
|
def parse_timestamp(value: object) -> float | None:
|
|
if not isinstance(value, str) or not value:
|
|
return None
|
|
try:
|
|
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
except ValueError:
|
|
return None
|
|
if parsed.tzinfo is None:
|
|
return parsed.timestamp()
|
|
return parsed.timestamp()
|
|
|
|
|
|
def is_active(unit: str) -> bool:
|
|
try:
|
|
result = run(["/usr/bin/systemctl", "is-active", "--quiet", unit], timeout=10)
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
return False
|
|
return result.returncode == 0
|
|
|
|
|
|
def is_user_active(unit: str) -> bool:
|
|
try:
|
|
result = run(
|
|
["/usr/bin/systemctl", "--user", "is-active", "--quiet", unit],
|
|
timeout=10,
|
|
)
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
return False
|
|
return result.returncode == 0
|
|
|
|
|
|
def wechat_process_present() -> bool:
|
|
try:
|
|
result = run(["/usr/bin/pgrep", "-u", str(os.getuid()), "-x", "wechat"], timeout=10)
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
return False
|
|
return result.returncode == 0
|
|
|
|
|
|
def weixin_window_geometry() -> tuple[int, int] | None:
|
|
environment = os.environ | {
|
|
"DISPLAY": DISPLAY,
|
|
"XAUTHORITY": str(XAUTHORITY),
|
|
"XDG_RUNTIME_DIR": str(XDG_RUNTIME_DIR),
|
|
"DBUS_SESSION_BUS_ADDRESS": DBUS_ADDRESS,
|
|
}
|
|
try:
|
|
search = subprocess.run(
|
|
["/usr/bin/xdotool", "search", "--onlyvisible", "--class", "wechat"],
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
timeout=10,
|
|
env=environment,
|
|
)
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
return False
|
|
if search.returncode != 0:
|
|
return False
|
|
for identifier in search.stdout.splitlines():
|
|
candidate = identifier.strip()
|
|
if not candidate:
|
|
continue
|
|
try:
|
|
title = subprocess.run(
|
|
["/usr/bin/xdotool", "getwindowname", candidate],
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
timeout=10,
|
|
env=environment,
|
|
)
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
continue
|
|
if title.returncode == 0 and title.stdout.strip() == "Weixin":
|
|
try:
|
|
geometry = subprocess.run(
|
|
[
|
|
"/usr/bin/xdotool",
|
|
"getwindowgeometry",
|
|
"--shell",
|
|
candidate,
|
|
],
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
timeout=10,
|
|
env=environment,
|
|
)
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
return None
|
|
if geometry.returncode != 0:
|
|
return None
|
|
values = dict(
|
|
line.split("=", 1)
|
|
for line in geometry.stdout.splitlines()
|
|
if "=" in line
|
|
)
|
|
try:
|
|
return int(values["WIDTH"]), int(values["HEIGHT"])
|
|
except (KeyError, ValueError):
|
|
return None
|
|
return None
|
|
|
|
|
|
def recorder_is_fresh(now: float) -> tuple[bool, str]:
|
|
state = load_json(RECORDER_STATE)
|
|
if not state:
|
|
return False, "recorder_state"
|
|
captured_at = parse_timestamp(state.get("last_capture_at"))
|
|
if captured_at is None:
|
|
return False, "recorder_state"
|
|
if now - captured_at > RECORDER_FRESHNESS_SECONDS:
|
|
return False, "recorder_stale"
|
|
return True, ""
|
|
|
|
|
|
def has_unacknowledged_failed_delivery() -> bool:
|
|
try:
|
|
return any(FAILED_DIR.glob("*.json"))
|
|
except OSError:
|
|
return False
|
|
|
|
|
|
def health_codes(now: float) -> list[str]:
|
|
codes = [
|
|
SERVICE_CODES[unit]
|
|
for unit in SYSTEM_SERVICES
|
|
if not is_active(unit)
|
|
]
|
|
codes.extend(
|
|
USER_SERVICE_CODES[unit]
|
|
for unit in USER_SERVICES
|
|
if not is_user_active(unit)
|
|
)
|
|
if not wechat_process_present():
|
|
codes.append("wechat_process")
|
|
geometry = weixin_window_geometry()
|
|
if geometry is None:
|
|
codes.append("wechat_window")
|
|
elif (
|
|
geometry[0] < MIN_READY_WINDOW_WIDTH
|
|
or geometry[1] < MIN_READY_WINDOW_HEIGHT
|
|
):
|
|
codes.append("wechat_login_required")
|
|
recorder_ok, recorder_code = recorder_is_fresh(now)
|
|
if not recorder_ok:
|
|
codes.append(recorder_code)
|
|
if has_unacknowledged_failed_delivery():
|
|
codes.append("responder_failed")
|
|
return codes
|
|
|
|
|
|
def should_retry(value: object, now: float, cooldown_seconds: int) -> bool:
|
|
try:
|
|
previous = float(value)
|
|
except (TypeError, ValueError):
|
|
return True
|
|
return now - previous >= cooldown_seconds
|
|
|
|
|
|
def start_wechat_once(state: dict[str, Any], now: float, *, dry_run: bool) -> str:
|
|
if not should_retry(state.get("last_start_attempt_at"), now, START_COOLDOWN_SECONDS):
|
|
return "cooldown"
|
|
if dry_run:
|
|
return "would_start"
|
|
|
|
state["last_start_attempt_at"] = now
|
|
try:
|
|
result = run(
|
|
[
|
|
"/usr/bin/systemctl",
|
|
"--user",
|
|
"start",
|
|
"app-wechat@autostart.service",
|
|
],
|
|
timeout=30,
|
|
)
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
state["last_start_result"] = "error"
|
|
return "error"
|
|
state["last_start_result"] = "started" if result.returncode == 0 else "error"
|
|
return str(state["last_start_result"])
|
|
|
|
|
|
def update_incident(
|
|
state: dict[str, Any],
|
|
*,
|
|
codes: list[str],
|
|
now: float,
|
|
) -> str | None:
|
|
incident = state.setdefault("wechat_chain", {})
|
|
active = bool(incident.get("active"))
|
|
|
|
if codes:
|
|
incident["consecutive_failures"] = int(incident.get("consecutive_failures", 0)) + 1
|
|
incident["last_codes"] = list(codes)
|
|
incident["last_seen_at"] = now
|
|
if not active and incident["consecutive_failures"] >= REQUIRED_FAILURE_SCANS:
|
|
incident["active"] = True
|
|
incident["started_at"] = now
|
|
incident["incident_pending"] = True
|
|
active = True
|
|
if active and not bool(incident.get("incident_sent")):
|
|
incident["incident_pending"] = True
|
|
if (
|
|
active
|
|
and bool(incident.get("incident_pending"))
|
|
and should_retry(
|
|
incident.get("incident_last_notification_attempt_at"),
|
|
now,
|
|
NOTIFICATION_COOLDOWN_SECONDS,
|
|
)
|
|
):
|
|
return "incident"
|
|
return None
|
|
|
|
incident["consecutive_failures"] = 0
|
|
incident["last_codes"] = []
|
|
incident["last_healthy_at"] = now
|
|
if active:
|
|
incident["active"] = False
|
|
incident["incident_pending"] = False
|
|
incident["recovered_at"] = now
|
|
if bool(incident.get("incident_sent")):
|
|
incident["recovery_pending"] = True
|
|
if (
|
|
bool(incident.get("recovery_pending"))
|
|
and should_retry(
|
|
incident.get("recovery_last_notification_attempt_at"),
|
|
now,
|
|
NOTIFICATION_COOLDOWN_SECONDS,
|
|
)
|
|
):
|
|
return "recovery"
|
|
return None
|
|
|
|
|
|
def notification_config() -> dict[str, Any]:
|
|
value = load_json(NOTIFICATION_CONFIG)
|
|
enabled = value.get("enabled", False)
|
|
command = value.get("command", [])
|
|
timeout = value.get("timeout_seconds", 30)
|
|
if not isinstance(enabled, bool):
|
|
raise RuntimeError("invalid notification enabled flag")
|
|
if (
|
|
not isinstance(command, list)
|
|
or any(not isinstance(argument, str) or not argument for argument in command)
|
|
or len(command) > 32
|
|
):
|
|
raise RuntimeError("invalid notification command")
|
|
if enabled and (not command or not command[0].startswith("/")):
|
|
raise RuntimeError("notification command must be absolute")
|
|
if not isinstance(timeout, int) or not 1 <= timeout <= 120:
|
|
raise RuntimeError("invalid notification timeout")
|
|
return {
|
|
"enabled": enabled,
|
|
"command": command,
|
|
"timeout_seconds": timeout,
|
|
}
|
|
|
|
|
|
def notification_failure_code(result: subprocess.CompletedProcess[str]) -> str:
|
|
output = f"{result.stdout}\n{result.stderr}".casefold()
|
|
for code in (
|
|
"provider_prepare_failed",
|
|
"provider_context_rejected",
|
|
"channel_unavailable",
|
|
"delivery_timeout",
|
|
"delivery_command_failed",
|
|
):
|
|
if f"delivery_error={code}" in output:
|
|
return code
|
|
if "prepare failed" in output:
|
|
return "provider_prepare_failed"
|
|
if "context_token" in output or "context token" in output:
|
|
return "provider_context_rejected"
|
|
if "unknown channel" in output or "unsupported channel" in output:
|
|
return "channel_unavailable"
|
|
if "timed out" in output or "timeout" in output:
|
|
return "delivery_timeout"
|
|
if result.returncode != 0:
|
|
return "delivery_command_failed"
|
|
return "delivery_unconfirmed"
|
|
|
|
|
|
def send_official_notification(message: str) -> None:
|
|
config = notification_config()
|
|
if not config["enabled"]:
|
|
raise RuntimeError("health notification is disabled")
|
|
try:
|
|
result = run(
|
|
list(config["command"]),
|
|
timeout=int(config["timeout_seconds"]),
|
|
data=message,
|
|
)
|
|
except subprocess.TimeoutExpired as error:
|
|
raise NotificationDeliveryUncertain("delivery_timeout") from error
|
|
if result.returncode != 0:
|
|
code = notification_failure_code(result)
|
|
if code in {"delivery_timeout", "delivery_unconfirmed"}:
|
|
raise NotificationDeliveryUncertain(code)
|
|
raise RuntimeError(f"official notification delivery failed: {code}")
|
|
|
|
|
|
def labels_for(codes: list[str]) -> str:
|
|
return "、".join(CODE_LABELS.get(code, code) for code in codes)
|
|
|
|
|
|
def incident_message(codes: list[str], start_result: str) -> str:
|
|
lines = [
|
|
"微信自动链路告警",
|
|
"状态：异常",
|
|
f"检查项：{labels_for(codes)}",
|
|
]
|
|
if start_result in {"started", "would_start"}:
|
|
lines.append("处理：已尝试启动微信客户端")
|
|
lines.extend(
|
|
[
|
|
"请打开 noVNC 检查；如显示二维码，请扫码登录。",
|
|
"若未收到后续官方通知，请在此私聊发送 aaa 刷新通道。",
|
|
]
|
|
)
|
|
return "\n".join(lines)
|
|
|
|
|
|
def recovery_message() -> str:
|
|
return "\n".join(
|
|
[
|
|
"微信自动链路已恢复",
|
|
"状态：VNC、微信窗口与 OCR 捕获检查正常。",
|
|
"若未收到后续官方通知，请在此私聊发送 aaa 刷新通道。",
|
|
]
|
|
)
|
|
|
|
|
|
def test_message() -> str:
|
|
return "\n".join(
|
|
[
|
|
"微信自动链路通知测试",
|
|
"状态：投递链路正常。",
|
|
]
|
|
)
|
|
|
|
|
|
def finish_notification(
|
|
state: dict[str, Any],
|
|
phase: str,
|
|
*,
|
|
codes: list[str],
|
|
start_result: str,
|
|
now: float,
|
|
dry_run: bool,
|
|
) -> bool:
|
|
if phase == "incident":
|
|
message = incident_message(codes, start_result)
|
|
elif phase == "recovery":
|
|
message = recovery_message()
|
|
else:
|
|
raise ValueError(f"unknown notification phase: {phase}")
|
|
|
|
if dry_run:
|
|
print(f"notification=would-send:{phase}")
|
|
return True
|
|
|
|
incident = state["wechat_chain"]
|
|
incident[f"{phase}_last_notification_attempt_at"] = now
|
|
try:
|
|
send_official_notification(message)
|
|
except NotificationDeliveryUncertain:
|
|
incident["last_notification_result"] = "uncertain"
|
|
if phase == "incident":
|
|
incident["incident_pending"] = False
|
|
incident["incident_sent"] = True
|
|
else:
|
|
incident["recovery_pending"] = False
|
|
incident["incident_sent"] = False
|
|
print(f"notification=uncertain:{phase}")
|
|
return True
|
|
except (OSError, subprocess.TimeoutExpired, RuntimeError):
|
|
incident["last_notification_result"] = "failed"
|
|
return False
|
|
|
|
incident["last_notification_result"] = "sent"
|
|
if phase == "incident":
|
|
incident["incident_pending"] = False
|
|
incident["incident_sent"] = True
|
|
else:
|
|
incident["recovery_pending"] = False
|
|
incident["incident_sent"] = False
|
|
print(f"notification=sent:{phase}")
|
|
return True
|
|
|
|
|
|
def monitor_once(*, dry_run: bool) -> int:
|
|
now = time.time()
|
|
state = load_json(STATE_FILE)
|
|
codes = health_codes(now)
|
|
start_result = "not_needed"
|
|
if "wechat_process" in codes:
|
|
start_result = start_wechat_once(state, now, dry_run=dry_run)
|
|
|
|
phase = update_incident(state, codes=codes, now=now)
|
|
print(f"health={'unhealthy' if codes else 'healthy'}")
|
|
print(f"codes={','.join(codes) if codes else '-'}")
|
|
print(f"wechat_start={start_result}")
|
|
|
|
if phase:
|
|
success = finish_notification(
|
|
state,
|
|
phase,
|
|
codes=codes,
|
|
start_result=start_result,
|
|
now=now,
|
|
dry_run=dry_run,
|
|
)
|
|
if not dry_run:
|
|
atomic_json(STATE_FILE, state)
|
|
return 0 if success else 1
|
|
|
|
if not dry_run:
|
|
atomic_json(STATE_FILE, state)
|
|
return 0
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument(
|
|
"--dry-run",
|
|
action="store_true",
|
|
help="evaluate health without starting WeChat or sending a notification",
|
|
)
|
|
parser.add_argument(
|
|
"--test-send",
|
|
action="store_true",
|
|
help="send one fixed notification through the configured adapter",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
if args.test_send:
|
|
if args.dry_run:
|
|
parser.error("--test-send cannot be combined with --dry-run")
|
|
try:
|
|
send_official_notification(test_message())
|
|
except NotificationDeliveryUncertain:
|
|
print("notification=uncertain:test")
|
|
return 2
|
|
print("notification=sent:test")
|
|
return 0
|
|
return monitor_once(dry_run=args.dry_run)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
raise SystemExit(main())
|
|
except KeyboardInterrupt:
|
|
raise SystemExit(130)
|