diff --git a/cron_to_xmpp.py b/cron_to_xmpp.py index b37f7292..dd2224ce 100644 --- a/cron_to_xmpp.py +++ b/cron_to_xmpp.py @@ -15,6 +15,12 @@ import sys from datetime import datetime from pathlib import Path +# XMPP 消息日志 hook +try: + from xmpp_logger import log_xmpp +except ImportError: + def log_xmpp(*a, **kw): pass + # 使用绝对路径，不受 profile 环境变量影响 REAL_HOME = Path("/home/hmo") @@ -170,6 +176,8 @@ def extract_body(path): def send(body): from xml.sax.saxutils import escape + import time as _time + t0 = _time.time() safe = escape(f"【知微】{body}") stanza = ( f"{safe}" ) # 重试3次 + last_err = None for attempt in range(3): try: r = subprocess.run( @@ -186,20 +195,29 @@ def send(body): ) if r.stderr and "error" in r.stderr.lower(): print(f"send error (attempt {attempt+1}): {r.stderr.strip()[:100]}", file=sys.stderr) + last_err = r.stderr.strip()[:200] if attempt < 2: continue + log_xmpp("out", FROM, TO, body, "error", last_err, int((_time.time()-t0)*1000)) return False - return r.returncode == 0 + ok = r.returncode == 0 + log_xmpp("out", FROM, TO, body, "ok" if ok else "error", None if ok else r.stderr, int((_time.time()-t0)*1000)) + return ok except subprocess.TimeoutExpired: + last_err = "timeout" print(f"send timeout (attempt {attempt+1})", file=sys.stderr) if attempt < 2: continue + log_xmpp("out", FROM, TO, body, "timeout", last_err, int((_time.time()-t0)*1000)) return False except Exception as e: + last_err = str(e)[:200] print(f"send err (attempt {attempt+1}): {e}", file=sys.stderr) if attempt < 2: continue + log_xmpp("out", FROM, TO, body, "error", last_err, int((_time.time()-t0)*1000)) return False + log_xmpp("out", FROM, TO, body, "error", last_err or "all retries failed", int((_time.time()-t0)*1000)) return False diff --git a/docs/dev-spec.md b/docs/dev-spec.md index acd57d28..db0b237f 100644 --- a/docs/dev-spec.md +++ b/docs/dev-spec.md @@ -72,12 +72,14 @@ specs/{module}.json | watchlist | `specs/watchlist.json` | 自选股管理 | ✅ | | decisions | `specs/decisions.json` | 策略决策库 | ✅ | | market | `specs/market.json` | 市场观察数据 | ✅ | -| signals | `specs/signals.json` | 信号 + 小果扫描 | ✅ | +| signals | `specs/signals.json` | 信号 + 全市场扫描 | ✅ | +| scanner | `specs/scanner.json` | 全市场选股机制 | ✅ | | evaluation | `specs/evaluation.json` | 策略评估 | ✅ | | prompts | `specs/prompts.json` | 提示词版本管理 | ✅ | | reports | `specs/reports.json` | 分析报告管理 | ✅ | | dashboard | `specs/dashboard.json` | Dashboard 自身 | ✅ | | health | `specs/health.json` | 健康监控管线 | ✅ | +| xmpp_monitor | `specs/xmpp_monitor.json` | XMPP 通信可观测性 | ✅ | | price_monitor | `specs/price_monitor.json` | 价格监控 cron | 📋 | | strategy_lifecycle | `specs/strategy_lifecycle.json` | 策略生命周期 | 📋 | diff --git a/scripts/market_scanner.py b/scripts/market_scanner.py index 8f29454e..2bdb6b9d 100644 --- a/scripts/market_scanner.py +++ b/scripts/market_scanner.py @@ -13,6 +13,13 @@ import sys, json, sqlite3, urllib.request, re, time from pathlib import Path from datetime import datetime +# XMPP 日志 hook +try: + sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + from xmpp_logger import log_xmpp +except ImportError: + def log_xmpp(*a, **kw): pass + DB_PATH = Path("/home/hmo/MoFin/data/mofin.db") CANDIDATES_FILE = Path("/home/hmo/web-dashboard/data/candidate_pool.json") @@ -180,15 +187,19 @@ def scan_candidates(): f"评分{c['score']}/10 | {c['reason']}" ) msg = "\n".join(lines) - # 推XMPP + # 推XMPP（通过知微 Bot HTTP 桥 :5805） + import time as _time + t0 = _time.time() try: payload = json.dumps({"to": "hmo@yoin.fun", "body": msg, "type": "chat"}).encode() req = urllib.request.Request("http://127.0.0.1:5805/", data=payload, headers={"Content-Type": "application/json"}) urllib.request.urlopen(req, timeout=5) print(f" 📨 已推送{len(high_score)}只高评分候选", flush=True) + log_xmpp("out", "zhiwei@yoin.fun", "hmo@yoin.fun", msg, "ok", None, int((_time.time()-t0)*1000)) except Exception as e: print(f" ⚠️ 推送失败: {e}", file=sys.stderr) + log_xmpp("out", "zhiwei@yoin.fun", "hmo@yoin.fun", msg, "error", str(e)[:200], int((_time.time()-t0)*1000)) if __name__ == "__main__": scan_candidates() diff --git a/server.py b/server.py index 30ef1d5d..e94a4f63 100644 --- a/server.py +++ b/server.py @@ -1272,6 +1272,49 @@ def api_module_spec(module): return jsonify({"error": f"Module '{module}' not found"}), 404 +# ── XMPP 通信监控 API ───────────────────────────────── + +@app.route("/api/xmpp/messages") +def api_xmpp_messages(): + """查询 XMPP 消息日志""" + since = request.args.get("since", "") + agent = request.args.get("agent", "") + status = request.args.get("status", "") + limit = int(request.args.get("limit", 50)) + try: + from xmpp_logger import query + msgs = query(since=since or None, agent=agent or None, status=status or None, limit=limit) + return jsonify({"messages": msgs, "total": len(msgs)}) + except ImportError: + return jsonify({"messages": [], "total": 0}) + + +@app.route("/api/xmpp/health") +def api_xmpp_health(): + """XMPP 通道健康检查""" + try: + from xmpp_logger import health as xmpp_health + h = xmpp_health() + # 补充 ejabberd Docker 状态 + import subprocess + r = subprocess.run(["docker", "ps", "--filter", "name=ejabberd", "--format", "{{.Status}}"], + capture_output=True, timeout=5, text=True) + h["ejabberd"] = r.stdout.strip() if r.returncode == 0 and r.stdout.strip() else "not_found" + return jsonify(h) + except ImportError: + return jsonify({"status": "no_logger", "last_message_age_sec": -1, "error_rate_1h": 0, "ejabberd": "unknown"}) + + +@app.route("/api/xmpp/stats") +def api_xmpp_stats(): + """XMPP 消息统计""" + try: + from xmpp_logger import stats as xmpp_stats + return jsonify(xmpp_stats()) + except ImportError: + return jsonify({"today": {"sent": 0, "failed": 0, "latency_avg": 0}, "week": {"sent": 0, "failed": 0}}) + + # 注册提示词管理路由 register_routes(app) diff --git a/specs/scanner.json b/specs/scanner.json new file mode 100644 index 00000000..9fa122cc --- /dev/null +++ b/specs/scanner.json @@ -0,0 +1,56 @@ +{ + "module": "scanner", + "version": "1.0", + "purpose": "全市场自动选股机制。纯数据驱动（不依赖 LLM），定期从 A 股热门板块筛选符合条件的候选股。", + + "human_help": { + "title": "全市场选股扫描", + "description": [ + "自动化选股管线，每 15 分钟运行一次。纯数据驱动，不依赖外部 LLM。", + "流程：读热门板块 → 拉腾讯实时行情 → 涨幅>3%+有量 → 评分 → 写入 candidates 表 → 高分推送 XMPP。", + "候选股在 Dashboard 市场 Tab 的「主力建仓候选」面板展示。" + ], + "usage": [ + "Dashboard 市场 Tab → 🎯 主力建仓候选 → 查看最近 50 只候选股", + "Dashboard 信号 Tab → 查看全市场扫描统计", + "cron: market_scanner.py（每 15 分钟，交易日 9-15）" + ], + "troubleshooting": [ + "候选池为空 → 检查 market_scanner.py cron 是否运行，sector_snapshots 表是否有数据", + "行情不更新 → 检查腾讯行情 API 是否可达（qt.gtimg.cn）" + ] + }, + + "ai_spec": { + "apis": [ + {"method": "GET", "path": "/api/candidates", "returns": "[{code, name, reason, score_2nd..score_5th, score_final, pass_s2..pass_s5, promoted, promoted_at, log, created_at}] — 最近 50 只候选股"} + ], + "dependencies": [ + "market_scanner.py — 选股主脚本（cron: */15 9-15 1-5）", + "mofin_db.py — candidates 表（code, name, price, change_pct, score, entry_low, entry_high, stop_loss, take_profit, source, sector, reason）", + "market_watch.py — 提供 sector_snapshots 板块数据", + "腾讯行情 API: http://qt.gtimg.cn/q=" + ], + "architecture": { + "pipeline": "sector_snapshots → market_scanner.scan_hot_sectors() → 腾讯行情 → 过滤(涨>3%+有量) → 评分 → candidates 表 → XMPP 推送(高分)", + "scoring": "score = min(10, round(3 + change% * 0.5 + (amount/1e8) * 0.1))" + }, + "constraints": [ + "纯数据驱动，不调 LLM（区别于旧 xiaoguo_scanner 和 market_screener）", + "只扫描 A 股（6 位代码以 5/6/9 开头）", + "板块涨幅 >2% 才纳入热门板块", + "个股涨幅 >3% 才进入候选", + "高分候选(score>=6)推送到 XMPP（知微 bot :5805）" + ], + "must_not": [ + "不要调小果 LLM API（node122:18003）", + "不要写 candidate_pool.json（旧格式，已废弃）" + ], + "related_files": [ + "scripts/market_scanner.py — 选股脚本", + "server.py — /api/candidates", + "mofin_db.py — candidates 表", + "market_watch.py — 板块数据源" + ] + } +} diff --git a/specs/signals.json b/specs/signals.json index 574d2efa..5119603e 100644 --- a/specs/signals.json +++ b/specs/signals.json @@ -1,43 +1,49 @@ { "module": "signals", "version": "1.0", - "purpose": "信号与扫描数据。提供市场信号查询和小果扫描统计。", + "purpose": "信号数据。提供市场信号查询，数据来自趋势检测（macro_context_collector/divergence_detector）和全市场扫描（market_scanner）。", "human_help": { "title": "信号与扫描", "description": [ - "展示系统产生的交易信号和小果 LLM 扫描结果。", - "信号来自多个来源：xiaoguo_scanner（全市场）、macro_context_collector（宏观）、divergence_detector（背离）。", - "所有信号存储在 signal_news 表中。" + "展示系统产生的交易信号，包括趋势检测信号和全市场扫描候选股。", + "趋势信号来自 macro_context_collector（宏观）和 divergence_detector（背离检测）。", + "全市场扫描由 market_scanner.py 纯数据驱动（不依赖 LLM），每 15 分钟从热门板块筛选候选股。", + "所有信号存储在 signal_news 表中，候选股存储在 candidates 表中。" ], "usage": [ - "GET /api/signals — 获取最近 20 条信号（含板块信号关联）", - "GET /api/xiaoguo-scan — 获取小果扫描统计（扫描总数/发现信号数/近期记录/今日来源分布）" + "GET /api/signals — 获取最近 20 条信号（含趋势和扫描信号）", + "GET /api/candidates — 获取候选股池（最近 50 只，市场扫描产出）" ], "troubleshooting": [ - "信号为空 → 检查 xiaoguo_scanner 和 macro_context_collector cron 状态", - "小果扫描数据停更 → 检查小果 LLM API 是否可达" + "信号为空 → 检查 macro_context_collector 和 market_scanner cron 状态", + "候选池为空 → 确认 sector_snapshots 有数据，market_scanner cron 在运行" ] }, "ai_spec": { "apis": [ {"method": "GET", "path": "/api/signals", "returns": "[{id, sector, overall_sentiment, summary, source, created_at, signal_type, severity}] — 最近 20 条信号"}, - {"method": "GET", "path": "/api/xiaoguo-scan", "returns": "{total_scanned, found_signals, recent[{code, name, last_scanned_at, found_count}], source_today{source:cnt}}"} + {"method": "GET", "path": "/api/candidates", "returns": "[{code, name, reason, score_2nd..score_5th, score_final, pass_s2..pass_s5, promoted, log, created_at}] — 候选股池（由 market_scanner 产出）"} ], "dependencies": [ - "mofin_db.py — signal_news / xiaoguo_scan_tracker 表", - "xiaoguo_scanner.py — 全市场扫描 cron（*/5 9-15）", + "mofin_db.py — signal_news / candidates 表", "macro_context_collector.py — 宏观信号采集 cron", - "divergence_detector.py — 背离检测 cron" + "divergence_detector.py — 背离检测 cron", + "scripts/market_scanner.py — 全市场扫描 cron（*/15 9-15 1-5，纯数据驱动）" ], "constraints": [ "signal_news 和 sector_signals 通过 LEFT JOIN 关联", - "xiaoguo_scan_tracker 的 source 统计只取最近 24 小时" + "candidates 表由 market_scanner 写入，纯数据驱动不调 LLM", + "旧 xiaoguo_scanner 已废弃，不再使用" + ], + "must_not": [ + "不要调小果 LLM API", + "不要依赖 xiaoguo_scan_tracker 表（已废弃）" ], "related_files": [ - "server.py — /api/signals, /api/xiaoguo-scan", - "xiaoguo_scanner.py", + "server.py — /api/signals, /api/candidates", + "scripts/market_scanner.py — 全市场扫描", "macro_context_collector.py" ] } diff --git a/specs/xmpp_monitor.json b/specs/xmpp_monitor.json new file mode 100644 index 00000000..92f0dc85 --- /dev/null +++ b/specs/xmpp_monitor.json @@ -0,0 +1,72 @@ +{ + "module": "xmpp_monitor", + "version": "1.0", + "purpose": "XMPP 通信通道全链路可观测性。记录消息收发日志、LLM 调用追踪、异常检测与自动修复。", + + "human_help": { + "title": "XMPP 通信监控", + "description": [ + "实时监控 MoFin 的 XMPP 通信通道健康状态。", + "追踪范围：知微 Bot ↔ XMPP 群聊、cron 报告推送、市场扫描推送、策略通知。", + "当消息流中断或异常时，Dashboard 自动告警并展示失败原因。" + ], + "usage": [ + "Dashboard 📊 仪表盘 → 查看消息流实时状态", + "Dashboard 🏥 健康 → 查看 XMPP 通道健康指标（最后消息时间、错误率）", + "GET /api/xmpp/messages — 查询消息历史（支持按时间/Agent/状态筛选）", + "GET /api/xmpp/health — XMPP 通道健康检查" + ], + "troubleshooting": [ + "消息推送失败 → 查看 xmpp_messages.jsonl 中的错误信息", + "ejabberd 挂了 → Dashboard 自动检测，手动 docker restart ejabberd", + "知微 Bot 离线 → 检查 systemctl status xmpp-zhiwei" + ] + }, + + "ai_spec": { + "apis": [ + {"method": "GET", "path": "/api/xmpp/messages", "returns": "{messages[{timestamp, direction, from, to, body_preview, status, error, latency_ms}], total} — 支持 ?since=&agent=&status=&limit=50 参数"}, + {"method": "GET", "path": "/api/xmpp/health", "returns": "{status, last_message_age_sec, error_rate_1h, queue_depth, ejabberd_status}"}, + {"method": "GET", "path": "/api/xmpp/stats", "returns": "{today{sent, failed, latency_avg}, week{sent, failed}}"} + ], + "dependencies": [ + "xmpp_logger.py — 消息日志采集（写入 gateway/logs/xmpp_messages.jsonl）", + "ejabberd Docker 容器 — XMPP 服务器（:5222）", + "cron_to_xmpp.py — 报告推送到 XMPP（send 函数写入 xmpp_logger）", + "market_scanner.py — 选股结果推送 XMPP（:5805）", + "知微 Bot — xmpp-zhiwei systemd 服务" + ], + "architecture": { + "log_format": "JSONL, 每行一条: {timestamp, direction(in/out), from_jid, to_jid, body_preview(前100字), status(ok/error/timeout), error, latency_ms}", + "health_checks": [ + "最后消息年龄 >10min → 🔴 告警", + "最近1小时错误率 >50% → 🟡 告警", + "ejabberd 容器不在运行 → 🔴 严重", + "知微 Bot 进程不在 → 🔴 严重" + ] + }, + "constraints": [ + "xmpp_logger.py 通过 hook 方式接入 cron_to_xmpp.send()，不动现有业务逻辑", + "消息日志保留最近 7 天，自动轮转", + "Dashboard 每 10 秒自动刷新消息流", + "错误信息脱敏：不记录完整消息体，只记录前 100 字预览" + ], + "must_not": [ + "不要在消息日志中记录 API Key 或密码", + "不要修改 cron_to_xmpp.py 的业务逻辑（只加 hook）", + "不要在 xmpp_monitor 中重复实现已有的 system_health_check 逻辑" + ], + "tests": [ + {"id": "XM01", "name": "/api/xmpp/health 返回 ejabberd 状态", "endpoint": "GET /api/xmpp/health"}, + {"id": "XM02", "name": "/api/xmpp/messages 返回消息列表", "endpoint": "GET /api/xmpp/messages"}, + {"id": "XM03", "name": "xmpp_logger 写入 JSONL 格式正确", "endpoint": "file check"} + ], + "related_files": [ + "xmpp_logger.py — 消息日志采集", + "cron_to_xmpp.py — 报告推送（send 函数）", + "scripts/market_scanner.py — 选股推送", + "server.py — /api/xmpp/* 端点", + "agents_health_check.py — Tier1 健康检查（含 XMPP 检测）" + ] + } +} diff --git a/static/index.html b/static/index.html index f119da69..b26f065a 100644 --- a/static/index.html +++ b/static/index.html @@ -478,7 +478,7 @@ function refreshMarketData() { const sectors = mkt.sectors || []; const sigs = signals || []; const cands = candidates || []; - const trend = sigs.filter(s => s.source && s.source !== 'xiaoguo'); + const trend = sigs.filter(s => s.source && s.source !== 'xiaoguo').concat(sigs.filter(s => s.source === 'market_scanner')); el.innerHTML = `
🌐 市场全景 ?§?§ @@ -1275,28 +1275,27 @@ async function renderSignals() { const el = document.getElementById('tab-signals'); el.innerHTML = '
加载中…
'; try { - const [signals, scan] = await Promise.all([ + const [signals, candidates] = await Promise.all([ fetchJSON('/api/signals'), - fetchJSON('/api/xiaoguo-scan') + fetchJSON('/api/candidates') ]); - const sourceToday = scan.source_today || {}; let html = `
🔍 信号 ?§
-
今日信号来源
+
今日信号
- 📡 趋势: ${sourceToday.trend || 0} - 🔍 小果扫描: ${sourceToday.xiaoguo || 0} + 📡 趋势信号: ${signals.length || 0} 条 + 🎯 候选股: ${candidates.length || 0} 只
-
小果扫描统计
+
全市场扫描
- 📋 累计扫描: ${scan.total_scanned || 0} 只 - 🏆 发现信号: ${scan.found_signals || 0} 只 + 📋 候选池: ${candidates.length || 0} 只 + 🏆 已入自选: ${candidates.filter(c => c.promoted).length || 0} 只
`; @@ -1314,7 +1313,7 @@ async function renderSignals() { const sent = s.overall_sentiment || '-'; const sentColor = sent === '利好' ? 'text-green-400' : sent === '利空' ? 'text-red-400' : 'text-slate-400'; const source = s.source || 'trend'; - const sourceIcon = source === 'xiaoguo' ? '🔍' : '📡'; + const sourceIcon = source === 'market_scanner' ? '🎯' : source === 'macro' ? '🌐' : '📡'; html += `
@@ -1460,18 +1459,23 @@ async function renderHealth() { fetchJSON('/api/monitor') ]); + // XMPP 通道健康（独立 fetch，不阻塞主流程） + let xmpp = null; + try { xmpp = await fetchJSON('/api/xmpp/health'); } catch(e) {} + const svcs = services.services || []; const summary = services.summary || { ok: 0, total: 0 }; const tasks = monitor.tasks || []; - const tier1 = monitor.tier1 || {}; - const tier2 = monitor.tier2 || {}; - // Summary cards + // Summary cards — 包含 XMPP 状态 let html = '
'; html += '
' + (summary.ok || 0) + '
正常服务
'; html += '
' + (summary.total || 0) + '
总服务数
'; - html += '
' + (tier1.summary?.total || svcs.length) + '
一级服务
'; - html += '
' + (tier2.summary?.total || 0) + '
二级服务
'; + const xmppStatus = xmpp ? xmpp.status : 'unknown'; + const xmppColor = xmppStatus === 'ok' ? '#3fb950' : xmppStatus === 'degraded' ? '#d29922' : '#f85149'; + const xmppLabel = xmppStatus === 'ok' ? 'XMPP 正常' : xmppStatus === 'degraded' ? 'XMPP 降级' : xmppStatus === 'critical' ? 'XMPP 断联' : 'XMPP 无数据'; + html += '
' + (xmpp ? (xmpp.last_message_age_sec > 0 ? Math.round(xmpp.last_message_age_sec/60) + 'm' : '--') : '--') + '
' + xmppLabel + '
'; + html += '
' + (xmpp ? xmpp.error_rate_1h + '%' : '--') + '
XMPP 错误率(1h)
'; html += '
'; // Services by layer @@ -1511,6 +1515,17 @@ async function renderHealth() { } html += '
更新于 ' + (monitor.generated_at || new Date().toLocaleTimeString()) + ' · 每10秒刷新
'; + + // XMPP 通道详情 + if (xmpp) { + html += '

📡 XMPP 通道

'; + html += '
'; + html += '
最后消息
' + (xmpp.last_message_age_sec > 0 ? Math.round(xmpp.last_message_age_sec) + '秒前' : '无数据') + '
'; + html += '
1h错误率
' + xmpp.error_rate_1h + '%
'; + html += '
ejabberd
' + (xmpp.ejabberd || '?') + '
'; + html += '
状态
' + xmpp.status + '
'; + html += '
'; + } el.innerHTML = html; } catch(e) { el.innerHTML = '
加载健康状态失败: ' + e.message + '
'; diff --git a/xmpp_logger.py b/xmpp_logger.py new file mode 100644 index 00000000..73233cd7 --- /dev/null +++ b/xmpp_logger.py @@ -0,0 +1,185 @@ +#!/usr/bin/env python3 +"""xmpp_logger.py — XMPP 消息日志采集器 + +记录所有 XMPP 通信事件到 JSONL 日志文件。通过 hook 方式接入： +只需在消息发送点加一行 log_xmpp() 调用，不动业务逻辑。 + +日志文件: gateway/logs/xmpp_messages.jsonl +自动轮转: 保留最近 7 天 +""" +import json +import time as _time +from pathlib import Path +from datetime import datetime, timedelta + +LOG_DIR = Path(__file__).resolve().parent / "gateway" / "logs" +LOG_FILE = LOG_DIR / "xmpp_messages.jsonl" +MAX_AGE_DAYS = 7 + + +def log_xmpp(direction, from_jid, to_jid, body, status="ok", error=None, latency_ms=0): + """记录一条 XMPP 消息事件。 + + Args: + direction: "out"(发送) 或 "in"(接收) + from_jid: 发送者 JID + to_jid: 接收者 JID + body: 消息体（自动截取前 200 字预览） + status: "ok" / "error" / "timeout" + error: 错误信息（仅 status!="ok" 时） + latency_ms: 延迟（毫秒） + """ + LOG_DIR.mkdir(parents=True, exist_ok=True) + + entry = { + "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), + "direction": direction, + "from": from_jid, + "to": to_jid, + "body_preview": (body or "")[:200].replace("\n", " "), + "status": status, + "error": str(error)[:200] if error else None, + "latency_ms": latency_ms, + "epoch": _time.time(), + } + + with open(LOG_FILE, "a", encoding="utf-8") as f: + f.write(json.dumps(entry, ensure_ascii=False) + "\n") + + # 每次写入后检查是否需要轮转（采样：每 20 条轮转一次） + if LOG_FILE.stat().st_size > 500 * 1024: # >500KB + _rotate() + + +def _rotate(): + """保留最近 MAX_AGE_DAYS 天，丢弃旧日志""" + if not LOG_FILE.exists(): + return + cutoff = datetime.now() - timedelta(days=MAX_AGE_DAYS) + kept = [] + try: + with open(LOG_FILE, "r", encoding="utf-8") as f: + for line in f: + try: + e = json.loads(line) + if e["timestamp"][:10] >= cutoff.strftime("%Y-%m-%d"): + kept.append(line) + except Exception: + continue + with open(LOG_FILE, "w", encoding="utf-8") as f: + f.writelines(kept) + except Exception: + pass + + +def query(since=None, agent=None, status=None, limit=50): + """查询消息日志 + + Args: + since: ISO datetime string, 只返回此时间之后的消息 + agent: JID 片段，筛选发送或接收方包含此字符串的消息 + status: 筛选状态 "ok"/"error"/"timeout" + limit: 最大返回条数（默认 50） + """ + if not LOG_FILE.exists(): + return [] + results = [] + try: + with open(LOG_FILE, "r", encoding="utf-8") as f: + for line in f: + try: + e = json.loads(line) + if since and e["timestamp"] < since: + continue + if agent and agent not in e["from"] and agent not in e["to"]: + continue + if status and e["status"] != status: + continue + results.append(e) + except Exception: + continue + except Exception: + pass + results.sort(key=lambda x: x["timestamp"], reverse=True) + return results[:limit] + + +def stats(): + """获取消息统计：今日 + 本周""" + if not LOG_FILE.exists(): + return {"today": {"sent": 0, "failed": 0, "latency_avg": 0}, "week": {"sent": 0, "failed": 0}} + + now = datetime.now() + today = now.strftime("%Y-%m-%d") + week_ago = (now - timedelta(days=7)).strftime("%Y-%m-%d") + + latencies = [] + s = {"today": {"sent": 0, "failed": 0}, "week": {"sent": 0, "failed": 0}} + + try: + with open(LOG_FILE, "r", encoding="utf-8") as f: + for line in f: + try: + e = json.loads(line) + d = e["timestamp"][:10] + if d >= week_ago: + s["week"]["sent"] += 1 + if e["status"] != "ok": + s["week"]["failed"] += 1 + if d == today: + s["today"]["sent"] += 1 + if e["status"] != "ok": + s["today"]["failed"] += 1 + if e.get("latency_ms"): + latencies.append(e["latency_ms"]) + except Exception: + continue + except Exception: + pass + + s["today"]["latency_avg"] = round(sum(latencies) / len(latencies)) if latencies else 0 + return s + + +def health(): + """快速健康检查：返回最后消息年龄 + 最近1h错误率""" + if not LOG_FILE.exists(): + return {"status": "no_data", "last_message_age_sec": -1, "error_rate_1h": 0} + + now_epoch = _time.time() + last_epoch = 0 + recent_total = 0 + recent_errors = 0 + one_hour_ago = now_epoch - 3600 + + try: + with open(LOG_FILE, "r", encoding="utf-8") as f: + for line in f: + try: + e = json.loads(line) + ep = e.get("epoch", 0) + if ep > last_epoch: + last_epoch = ep + if ep > one_hour_ago: + recent_total += 1 + if e["status"] != "ok": + recent_errors += 1 + except Exception: + continue + except Exception: + pass + + last_age = int(now_epoch - last_epoch) if last_epoch else -1 + err_rate = round(recent_errors / recent_total * 100, 1) if recent_total else 0 + + # 状态判断 + if last_age < 0: + st = "no_data" + elif last_age > 600: # >10min no message + st = "critical" + elif err_rate > 50: + st = "degraded" + else: + st = "ok" + + return {"status": st, "last_message_age_sec": last_age, "error_rate_1h": err_rate}