93 lines
3.1 KiB
Python
Executable File
93 lines
3.1 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
NewAPI 周度评估报告 — 汇总观测数据，输出评估结论
|
||
每周日 18:00 由 cron 触发，输出可读报告（非静默，始终输出）
|
||
评估维度：可用性 / 延迟 / 路由一致性
|
||
替代 omniroute-weekly-report.py（OmniRoute 已于 2026-09-02 关停）
|
||
"""
|
||
import json
|
||
import os
|
||
import statistics
|
||
from datetime import datetime, timezone, timedelta
|
||
|
||
STATE_DIR = os.path.expanduser("~/.hermes/newapi-observe")
|
||
LOG = os.path.join(STATE_DIR, "observations.jsonl")
|
||
|
||
# 评估阈值
|
||
LATENCY_WARN_MS = 8000 # 平均延迟超 8s 警告
|
||
SUCCESS_RATE_MIN = 0.90 # 成功率低于 90% 警告
|
||
|
||
|
||
def load_observations(days=7):
|
||
if not os.path.exists(LOG):
|
||
return []
|
||
cutoff = datetime.now(timezone.utc) - timedelta(days=days)
|
||
obs = []
|
||
with open(LOG) as f:
|
||
for line in f:
|
||
try:
|
||
d = json.loads(line)
|
||
ts = datetime.fromisoformat(d["ts"])
|
||
if ts >= cutoff:
|
||
obs.append(d)
|
||
except Exception:
|
||
continue
|
||
return obs
|
||
|
||
|
||
def main():
|
||
obs = load_observations(7)
|
||
n = len(obs)
|
||
|
||
print(f"=== NewAPI 周度评估（最近 7 天）===")
|
||
print(f"观测次数: {n}")
|
||
print(f"生成时间: {datetime.now(timezone.utc).isoformat()}")
|
||
print()
|
||
|
||
if n == 0:
|
||
print("⚠️ 无观测数据（cron 可能在观测脚本切换期间未运行）")
|
||
return
|
||
|
||
# 1. 服务可用性
|
||
svc_ok = sum(1 for o in obs if o.get("service") == "ok")
|
||
svc_rate = svc_ok / n
|
||
print(f"【服务可用性】 {svc_ok}/{n} = {svc_rate*100:.1f}%")
|
||
|
||
# 2. 按模型统计
|
||
by_model = {}
|
||
for o in obs:
|
||
for t in o.get("tests", []):
|
||
m = t["model"]
|
||
if m not in by_model:
|
||
by_model[m] = {"ok": 0, "total": 0, "latencies": []}
|
||
by_model[m]["total"] += 1
|
||
if t["ok"]:
|
||
by_model[m]["ok"] += 1
|
||
by_model[m]["latencies"].append(t["latency_ms"])
|
||
|
||
print(f"\n【按模型表现】")
|
||
for m, d in sorted(by_model.items()):
|
||
rate = d["ok"] / d["total"] * 100 if d["total"] else 0
|
||
if d["latencies"]:
|
||
avg = statistics.mean(d["latencies"])
|
||
p95 = sorted(d["latencies"])[int(len(d["latencies"])*0.95)] if len(d["latencies"]) > 1 else d["latencies"][0]
|
||
else:
|
||
avg = p95 = 0
|
||
status = "✅" if rate >= SUCCESS_RATE_MIN*100 and avg < LATENCY_WARN_MS else "⚠️"
|
||
print(f" {status} {m}: {d['ok']}/{d['total']} = {rate:.0f}%, 平均 {avg:.0f}ms, P95 {p95:.0f}ms")
|
||
|
||
# 3. 总体评估
|
||
total_tests = sum(d["total"] for d in by_model.values())
|
||
total_ok = sum(d["ok"] for d in by_model.values())
|
||
overall_rate = total_ok / total_tests * 100 if total_tests else 0
|
||
print(f"\n【总体】 {total_ok}/{total_tests} = {overall_rate:.1f}%")
|
||
|
||
if overall_rate < SUCCESS_RATE_MIN*100:
|
||
print("⚠️ 建议: 检查 NewAPI 上游 channel 健康度")
|
||
else:
|
||
print("✅ 评估: NewAPI 表现健康，可作为本地主网关")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|