#!/usr/bin/env python3 """ easyclaw 可视化仪表盘生成器 =========================== 读取 easyclaw 日志文件，生成一个自包含的 HTML 可视化仪表盘。 用法： python easyclaw_dashboard.py <日志文件路径> [--output dashboard.html] [--since YYYY-MM-DD] """ import argparse import json import os import re import sys from collections import Counter, defaultdict from datetime import datetime, timedelta LOG_PATTERN = re.compile( r"^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[(\w+)\]\s+(.*)$" ) BYTES_PATTERN = re.compile(r"(\d+)\s*bytes") URL_PATTERN = re.compile(r"https?://\S+") def parse_log_file(path, since=None): entries = [] with open(path, "r", encoding="utf-8", errors="replace") as f: for line in f: line = line.strip() if not line: continue m = LOG_PATTERN.match(line) if not m: continue ts_str, level, message = m.groups() try: ts = datetime.strptime(ts_str, "%Y-%m-%d %H:%M:%S") except ValueError: continue if since and ts < since: continue entries.append({"timestamp": ts, "level": level.upper(), "message": message, "raw": line}) return entries def build_dashboard_data(entries, log_path): if not entries: return {"empty": True} level_counts = Counter(e["level"] for e in entries) total_tasks = level_counts.get("SUCCESS", 0) + level_counts.get("ERROR", 0) success_rate = round(level_counts["SUCCESS"] / total_tasks * 100, 1) if total_tasks > 0 else 0 total_bytes = 0 url_counts = Counter() hourly = defaultdict(lambda: {"total": 0, "success": 0, "error": 0}) daily = defaultdict(lambda: {"total": 0, "success": 0, "error": 0, "bytes": 0}) errors = [] recent_logs = [] for e in entries: bm = BYTES_PATTERN.search(e["message"]) if bm: total_bytes += int(bm.group(1)) um = URL_PATTERN.search(e["message"]) if um: url_counts[um.group()] += 1 hour_key = e["timestamp"].strftime("%Y-%m-%d %H:00") day_key = e["timestamp"].strftime("%Y-%m-%d") hourly[hour_key]["total"] += 1 daily[day_key]["total"] += 1 if e["level"] == "SUCCESS": hourly[hour_key]["success"] += 1 daily[day_key]["success"] += 1 if bm: daily[day_key]["bytes"] += int(bm.group(1)) elif e["level"] == "ERROR": hourly[hour_key]["error"] += 1 daily[day_key]["error"] += 1 errors.append({"ts": e["timestamp"].strftime("%Y-%m-%d %H:%M:%S"), "msg": e["message"]}) for e in entries[-200:]: lc = {"INFO": "info", "SUCCESS": "success", "ERROR": "error", "WARNING": "warning"}.get(e["level"], "info") recent_logs.append({ "ts": e["timestamp"].strftime("%Y-%m-%d %H:%M:%S"), "level": e["level"], "lc": lc, "msg": e["message"], }) first_ts = entries[0]["timestamp"] last_ts = entries[-1]["timestamp"] age_seconds = (datetime.now() - last_ts).total_seconds() if age_seconds < 300: status = "running" elif age_seconds < 3600: status = "idle" else: status = "offline" # hourly sorted sorted_hourly = sorted(hourly.items()) # daily sorted sorted_daily = sorted(daily.items()) # top URLs top_urls = url_counts.most_common(15) def fmt_bytes(n): for unit in ("B", "KB", "MB", "GB"): if n < 1024: return f"{n:.1f} {unit}" n /= 1024 return f"{n:.1f} TB" return { "empty": False, "log_path": os.path.basename(log_path), "total_entries": len(entries), "first_ts": first_ts.strftime("%Y-%m-%d %H:%M:%S"), "last_ts": last_ts.strftime("%Y-%m-%d %H:%M:%S"), "status": status, "level_counts": dict(level_counts), "success_rate": success_rate, "total_tasks": total_tasks, "total_bytes": total_bytes, "total_bytes_fmt": fmt_bytes(total_bytes), "unique_urls": len(url_counts), "hourly_labels": [h[0] for h in sorted_hourly], "hourly_success": [h[1]["success"] for h in sorted_hourly], "hourly_error": [h[1]["error"] for h in sorted_hourly], "hourly_total": [h[1]["total"] for h in sorted_hourly], "daily_labels": [d[0] for d in sorted_daily], "daily_success": [d[1]["success"] for d in sorted_daily], "daily_error": [d[1]["error"] for d in sorted_daily], "daily_bytes": [d[1]["bytes"] for d in sorted_daily], "top_urls": [{"url": u, "count": c} for u, c in top_urls], "errors": errors[-50:], "recent_logs": recent_logs, "generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), } HTML_TEMPLATE = r''' easyclaw Dashboard

easyclaw Monitor

__LOG_PATH__ · Generated __GENERATED_AT__

Total Entries
Success Rate
Data Fetched
Unique URLs

Hourly Activity

24h
3d
All

Level Distribution

Daily Data Volume

Top URLs

Recent Errors

Live Logs

All
Success
Error
Warn
easyclaw monitor · dashboard auto-generated
''' def generate_html(data: dict) -> str: """将数据嵌入 HTML 模板。""" html = HTML_TEMPLATE html = html.replace("__DATA_JSON__", json.dumps(data, ensure_ascii=False)) html = html.replace("__LOG_PATH__", data.get("log_path", "")) html = html.replace("__GENERATED_AT__", data.get("generated_at", "")) html = html.replace("__STATUS__", data.get("status", "offline")) return html def main(): parser = argparse.ArgumentParser(description="easyclaw 可视化仪表盘生成器") parser.add_argument("logfile", help="easyclaw 日志文件路径") parser.add_argument("--output", "-o", default="easyclaw_dashboard.html", help="输出 HTML 文件路径 (默认 easyclaw_dashboard.html)") parser.add_argument("--since", help="只分析此时间之后的日志 (YYYY-MM-DD 或 YYYY-MM-DD HH:MM:SS)") args = parser.parse_args() since = None if args.since: for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d"): try: since = datetime.strptime(args.since, fmt) break except ValueError: continue if since is None: print("错误: --since 格式不正确") sys.exit(1) if not os.path.isfile(args.logfile): print(f"错误: 找不到日志文件 '{args.logfile}'") sys.exit(1) print(f"正在解析日志: {args.logfile} ...") entries = parse_log_file(args.logfile, since=since) if not entries: print("日志为空或无匹配记录。") sys.exit(1) print(f"解析到 {len(entries)} 条记录，正在生成仪表盘 ...") data = build_dashboard_data(entries, args.logfile) html = generate_html(data) with open(args.output, "w", encoding="utf-8") as f: f.write(html) print(f"仪表盘已生成: {args.output}") if __name__ == "__main__": main()