MoltBot 状态监控

"""MoltBot 状态监控网页""" import json import os import subprocess import time from datetime import datetime, timedelta import psutil from flask import Flask, Response app = Flask(__name__) START_TIME = time.time() def get_gateway_status(): """检查 OpenClaw Gateway 是否在运行""" for proc in psutil.process_iter(["pid", "name", "cmdline"]): try: cmdline = " ".join(proc.info.get("cmdline") or []) if "openclaw" in cmdline and "gateway" in cmdline: return {"running": True, "pid": proc.info["pid"]} except (psutil.NoSuchProcess, psutil.AccessDenied): continue return {"running": False, "pid": None} def get_feishu_status(): """检查飞书连接状态（通过日志）""" log_file = f"/tmp/openclaw/openclaw-{datetime.now().strftime('%Y-%m-%d')}.log" if not os.path.exists(log_file): return "unknown" try: result = subprocess.run( ["tail", "-100", log_file], capture_output=True, text=True, timeout=5 ) lines = result.stdout if "[feishu]" in lines: if "ws client ready" in lines or "Received from" in lines: return "connected" elif "auto-restart" in lines: return "reconnecting" return "initializing" except Exception: return "unknown" def get_message_count(): """统计今日消息数""" log_file = f"/tmp/openclaw/openclaw-{datetime.now().strftime('%Y-%m-%d')}.log" if not os.path.exists(log_file): return 0 try: result = subprocess.run( ["grep", "-c", "Received from", log_file], capture_output=True, text=True, timeout=5 ) return int(result.stdout.strip() or 0) except Exception: return 0 def get_system_info(): """获取系统资源信息""" uptime = timedelta(seconds=int(time.time() - START_TIME)) return { "cpu_percent": psutil.cpu_percent(interval=0.5), "memory": psutil.virtual_memory()._asdict(), "uptime": str(uptime), } def format_bytes(b): """格式化字节""" for unit in ["B", "KB", "MB", "GB"]: if b < 1024: return f"{b:.1f} {unit}" b /= 1024 return f"{b:.1f} TB" STATUS_COLORS = { "connected": "#00e676", "reconnecting": "#ffc107", "initializing": "#2196f3", "unknown": "#9e9e9e", } STATUS_TEXT = { "connected": "✅ 已连接", "reconnecting": "🔄 重连中", "initializing": "⏳ 初始化中", "unknown": "❓ 未知", } @app.route("/") def index(): gw = get_gateway_status() feishu = get_feishu_status() msgs = get_message_count() sys_info = get_system_info() mem = sys_info["memory"] feishu_color = STATUS_COLORS.get(feishu, "#9e9e9e") feishu_text = STATUS_TEXT.get(feishu, "❓ 未知") gw_color = "#00e676" if gw["running"] else "#f44336" gw_text = f"✅ 运行中 (PID {gw['pid']})" if gw["running"] else "❌ 已停止" html = f""" MoltBot 状态监控

OpenClaw 网关

{gw_text}

飞书连接

{feishu_text}

系统概览

{sys_info['uptime']}

运行时长

{msgs}

今日消息

{os.environ.get('MODEL_NAME', 'N/A')}

当前模型

CPU 使用率

{sys_info['cpu_percent']}%

内存使用

{format_bytes(mem['used'])} / {format_bytes(mem['total'])}

""" return Response(html, content_type="text/html; charset=utf-8") @app.route("/health") def health(): gw = get_gateway_status() return json.dumps({ "status": "ok" if gw["running"] else "error", "gateway": gw, "feishu": get_feishu_status(), "uptime": time.time() - START_TIME, }), 200 if gw["running"] else 503 if __name__ == "__main__": app.run(host="0.0.0.0", port=7860)

🤖 MoltBot AI