#!/usr/bin/env python3 from __future__ import annotations import json import re from datetime import datetime from pathlib import Path from typing import Any SOURCE_DIR = Path('/Users/bot1/.hermes/profiles/it/cron/output/a49833a87c60') PROJECT_DIR = Path(__file__).resolve().parents[1] RAW_REPORT_DIR = PROJECT_DIR / 'work' / 'raw_reports' OUT_DIR = PROJECT_DIR / 'deliverables' DATA_PATH = OUT_DIR / 'profile_audit_dashboard_data.json' HTML_PATH = OUT_DIR / 'index.html' SECRET_PAT = re.compile(r'(?i)(secret|token|key|password|access_token|refresh_token|app_secret|verification_token|encrypt_key)(\s*[:=]\s*)([^\s,;`]+)') def redact(s: str) -> str: return SECRET_PAT.sub(lambda m: m.group(1) + m.group(2) + '[REDACTED]', s) def split_md_row(line: str) -> list[str]: return [c.strip() for c in line.strip().strip('|').split('|')] def parse_mem_cell(cell: str) -> dict[str, Any]: raw = cell.strip() if re.search(r'(?i)missing|read_error|不存在', raw): return {'chars': 0, 'pct': 0.0, 'status': 'missing', 'raw': raw} # Supported report formats: # 1950字/88.6%/高 # 1950 / 88.6% # 1950/88.6%/高 m = re.search(r'(\d+)\s*(?:字)?\s*/\s*([\d.]+)%\s*(?:/\s*([^/\s]+))?', raw) if not m: return {'chars': 0, 'pct': 0.0, 'status': '未知', 'raw': raw} pct = float(m.group(2)) status = m.group(3) or ('临界' if pct >= 95 else '高' if pct >= 80 else '正常') return {'chars': int(m.group(1)), 'pct': pct, 'status': status, 'raw': raw} def parse_int(cell: str) -> int: try: return int(re.sub(r'\D+', '', cell) or 0) except Exception: return 0 def parse_report(path: Path) -> dict[str, Any]: raw_text = redact(path.read_text(encoding='utf-8', errors='replace')) # Cron output includes the original prompt before "## Response"; parse only the actual delivered report. text = raw_text.split('## Response', 1)[1] if '## Response' in raw_text else raw_text date_match = re.search(r'每日 Profile 记忆与异常巡检报告 - (\d{4}-\d{2}-\d{2})', text) report_date = date_match.group(1) if date_match else path.name[:10] run_match = re.search(r'\*\*Run Time:\*\*\s*([^\n]+)', text) run_time = run_match.group(1).strip() if run_match else '' overview_match = re.search(r'总览：([^\n]+)', text) overview = overview_match.group(1).strip() if overview_match else '' rows: list[dict[str, Any]] = [] for line in text.splitlines(): if not line.startswith('|') or '---' in line or '机器' in line: continue cols = split_md_row(line) if len(cols) < 8: continue machine, profile, gateway, user_mem, memory_mem, timeout, errors, rating = cols[:8] if machine not in {'bot1', 'bot2'}: continue rows.append({ 'machine': machine, 'profile': profile, 'gateway': gateway, 'user_mem': parse_mem_cell(user_mem), 'memory_mem': parse_mem_cell(memory_mem), 'timeouts_24h': parse_int(timeout), 'errors_24h': parse_int(errors), 'rating': rating, }) red = sum(1 for r in rows if r['rating'] == '红') yellow = sum(1 for r in rows if r['rating'] == '黄') green = sum(1 for r in rows if r['rating'] == '绿') high_mem = sum(1 for r in rows if r['user_mem']['pct'] >= 80 or r['memory_mem']['pct'] >= 80) critical_mem = sum(1 for r in rows if r['user_mem']['pct'] >= 95 or r['memory_mem']['pct'] >= 95) gateway_bad = sum(1 for r in rows if re.search(r'停止|异常|stale|not running', r['gateway'], re.I)) total_timeout = sum(r['timeouts_24h'] for r in rows) total_errors = sum(r['errors_24h'] for r in rows) def block_after(title: str, until_titles: list[str]) -> list[str]: lines = text.splitlines() start = None for i, line in enumerate(lines): if line.strip().startswith(title): start = i + 1 break if start is None: return [] out: list[str] = [] for line in lines[start:]: stripped = line.strip() if any(stripped.startswith(t) for t in until_titles): break if stripped and not stripped.startswith('|'): out.append(stripped) return out def clean_anomaly_lines(lines: list[str]) -> list[str]: cleaned = [] for line in lines: # Markdown bullets from the source report become real list bullets in the dashboard, # so strip the source bullet marker to avoid "• - ..." visual noise. cleaned.append(re.sub(r'^[-*]\s*', '', line).strip()) return cleaned return { 'source_file': str(path), 'report_date': report_date, 'run_time': run_time, 'overview': overview, 'rows': rows, 'summary': { 'profiles': len(rows), 'red': red, 'yellow': yellow, 'green': green, 'high_mem': high_mem, 'critical_mem': critical_mem, 'gateway_bad': gateway_bad, 'timeouts_24h': total_timeout, 'errors_24h': total_errors, 'bot2_reachable': 'bot2 可达' in text, }, 'anomalies': clean_anomaly_lines(block_after('重点异常：', ['建议动作：'])), 'suggestions': block_after('建议动作：', ['##', '# Cron Job:']), } def severity_rank(row: dict[str, Any]) -> tuple: rating_rank = {'红': 0, '黄': 1, '绿': 2}.get(row['rating'], 3) max_mem = max(row['user_mem']['pct'], row['memory_mem']['pct']) return (rating_rank, -row['errors_24h'], -row['timeouts_24h'], -max_mem, row['machine'], row['profile']) def build_data() -> dict[str, Any]: report_paths: list[Path] = [] report_paths.extend(sorted(SOURCE_DIR.glob('*.md'))) if RAW_REPORT_DIR.exists(): report_paths.extend(sorted(RAW_REPORT_DIR.glob('*.md'))) # De-dupe by report date and prefer the explicit raw_reports file generated by the dashboard cron. by_date: dict[str, dict[str, Any]] = {} for path in report_paths: parsed = parse_report(path) if parsed['rows']: by_date[parsed['report_date']] = parsed reports = sorted(by_date.values(), key=lambda r: r['report_date']) latest = reports[-1] if reports else {'rows': [], 'summary': {}, 'anomalies': [], 'suggestions': []} latest['rows'] = sorted(latest['rows'], key=severity_rank) trends = [{'date': r['report_date'], **r['summary']} for r in reports] return { 'generated_at': datetime.now().strftime('%Y-%m-%d %H:%M:%S'), 'source_dir': str(SOURCE_DIR), 'report_count': len(reports), 'latest': latest, 'trends': trends, 'top_memory': sorted(latest['rows'], key=lambda r: max(r['user_mem']['pct'], r['memory_mem']['pct']), reverse=True)[:10], 'top_errors': sorted(latest['rows'], key=lambda r: (r['errors_24h'], r['timeouts_24h']), reverse=True)[:10], 'machines': sorted({r['machine'] for r in latest['rows']}), } HTML = r''' 每日 Profile 记忆与异常巡检看板
Hermes Profiles · Daily Audit

每日 Profile 记忆与异常巡检看板

把早上 7 点巡检报告转成可视化视图：一眼看出哪些 profile 记忆接近上限、哪些网关异常、哪些 profile 最近 24 小时错误/超时偏高。当前页面为静态 HTML，数据来自 cron 历史报告。

今日需要关注

只把会明显影响使用体验的高优先级项放在这里：记忆临界、重连/错误集中、gateway 未运行。普通项折叠在下方。
展开普通项 / 低优先级明细

    近日报告趋势

    红/黄项、临界记忆、错误量的变化。

    记忆使用 TOP 10

    取 USER 与 MEMORY 两项中的最大使用率。

    24h 错误 TOP 10

    用于优先定位模型/API/网关/工具异常。

    重点异常与建议

    保留报告里的红/黄项摘要，敏感字段已脱敏。

    Profile 明细

    默认按风险排序：红项、错误量、超时量、记忆使用率。
    Profile评级GatewayUSER 记忆MEMORY 记忆24h 超时24h 错误
    数据源：
    说明：这是只读巡检结果的可视化，不会修改 profile、不会重启 gateway、不会清理 memory。要刷新页面，请运行 python3 work/generate_dashboard.py。
    ''' def json_for_html(data: dict[str, Any]) -> str: return json.dumps(data, ensure_ascii=False, separators=(',', ':')).replace(' None: OUT_DIR.mkdir(parents=True, exist_ok=True) data = build_data() DATA_PATH.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding='utf-8') HTML_PATH.write_text(HTML.replace('__DATA__', json_for_html(data)), encoding='utf-8') print(json.dumps({ 'ok': True, 'html': str(HTML_PATH), 'data': str(DATA_PATH), 'report_count': data['report_count'], 'latest_date': data.get('latest', {}).get('report_date'), 'profiles': data.get('latest', {}).get('summary', {}).get('profiles'), 'red': data.get('latest', {}).get('summary', {}).get('red'), 'critical_mem': data.get('latest', {}).get('summary', {}).get('critical_mem'), }, ensure_ascii=False)) if __name__ == '__main__': main()