50f5089459
sync-codex-health 拉 health/alive/endpoints（key 文件 600）； e2e --from-endpoints 真调中转；余额不足单独 break；refresh 不覆盖 call 结果。
299 lines
10 KiB
Python
299 lines
10 KiB
Python
#!/usr/bin/env python3
|
||
"""端到端能力探针（最小）：出口 path + 中转 + 可选真实 completion。
|
||
|
||
不代替号池 health；只回答「今天链路是否可调用」。
|
||
|
||
检查：
|
||
E0 system proxy ON（scutil）
|
||
E1 Clash controller + mode=rule
|
||
E2 生产 DIRECT 覆盖（串线 risk != crit）
|
||
E3 具名中转至少 1 ok
|
||
E4 可选：JOY_E2E_BASE_URL + JOY_E2E_API_KEY 发 /v1/chat/completions
|
||
|
||
写入：~/.cache/manhuang/e2e/status.json
|
||
|
||
用法:
|
||
python3 scripts/manhuang/probe-e2e-capability.py
|
||
JOY_E2E_BASE_URL=https://... JOY_E2E_API_KEY=sk-... python3 ... --call
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import time
|
||
import urllib.error
|
||
import urllib.request
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
ROOT = Path(__file__).resolve().parents[2]
|
||
OUT = Path.home() / ".cache" / "manhuang" / "e2e" / "status.json"
|
||
NET = Path.home() / ".cache" / "manhuang" / "net-path" / "status.json"
|
||
RELAYS = Path.home() / ".cache" / "manhuang" / "relays" / "status.json"
|
||
|
||
|
||
def _now() -> str:
|
||
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||
|
||
|
||
def _load(p: Path) -> dict[str, Any]:
|
||
if not p.is_file():
|
||
return {}
|
||
try:
|
||
d = json.loads(p.read_text(encoding="utf-8"))
|
||
return d if isinstance(d, dict) else {}
|
||
except (OSError, json.JSONDecodeError):
|
||
return {}
|
||
|
||
|
||
def _sys_proxy_on() -> tuple[bool, dict[str, Any]]:
|
||
try:
|
||
import importlib.util
|
||
pp = ROOT / "scripts/manhuang/probe-proxy-path.py"
|
||
spec = importlib.util.spec_from_file_location("_pp", pp)
|
||
if not spec or not spec.loader:
|
||
return False, {"error": "no_probe"}
|
||
mod = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(mod)
|
||
sp = mod.probe_system_proxy()
|
||
return bool(sp.get("enabled")), sp
|
||
except Exception as e:
|
||
return False, {"error": str(e)[:120]}
|
||
|
||
|
||
def _chat_probe(base: str, key: str, model: str, timeout: float = 25.0) -> dict[str, Any]:
|
||
url = base.rstrip("/") + "/v1/chat/completions"
|
||
body = json.dumps({
|
||
"model": model,
|
||
"messages": [{"role": "user", "content": "ping"}],
|
||
"max_tokens": 4,
|
||
}).encode()
|
||
req = urllib.request.Request(
|
||
url,
|
||
data=body,
|
||
method="POST",
|
||
headers={
|
||
"Content-Type": "application/json",
|
||
"Authorization": f"Bearer {key}",
|
||
"User-Agent": "joy-manhuang-e2e/1",
|
||
},
|
||
)
|
||
t0 = time.time()
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=timeout) as r:
|
||
raw = r.read()
|
||
ms = int((time.time() - t0) * 1000)
|
||
try:
|
||
data = json.loads(raw.decode())
|
||
except json.JSONDecodeError:
|
||
return {"ok": False, "http": r.status, "ms": ms, "error": "bad_json"}
|
||
# 有 choices 或 error 字段
|
||
if data.get("choices") or data.get("id"):
|
||
return {"ok": True, "http": r.status, "ms": ms, "model": model}
|
||
if data.get("error"):
|
||
return {"ok": False, "http": r.status, "ms": ms, "error": str(data.get("error"))[:160]}
|
||
return {"ok": True, "http": r.status, "ms": ms, "note": "no_choices"}
|
||
except urllib.error.HTTPError as e:
|
||
ms = int((time.time() - t0) * 1000)
|
||
try:
|
||
err = e.read().decode()[:160]
|
||
except Exception:
|
||
err = str(e)
|
||
return {"ok": False, "http": e.code, "ms": ms, "error": err}
|
||
except Exception as e:
|
||
return {"ok": False, "http": 0, "ms": int((time.time() - t0) * 1000), "error": str(e)[:160]}
|
||
|
||
|
||
def run(*, do_call: bool) -> dict[str, Any]:
|
||
checks: list[dict[str, Any]] = []
|
||
np = _load(NET)
|
||
rel = _load(RELAYS)
|
||
|
||
# E0
|
||
sp_on, sp = _sys_proxy_on()
|
||
checks.append({
|
||
"id": "E0_sys_proxy",
|
||
"ok": sp_on,
|
||
"detail": "ON" if sp_on else "OFF — 本机可能上不了外网",
|
||
})
|
||
|
||
# E1 controller
|
||
prim = np.get("primary") if isinstance(np.get("primary"), dict) else {}
|
||
ctrl_ok = bool(prim.get("ok"))
|
||
mode = str(prim.get("mode") or "")
|
||
checks.append({
|
||
"id": "E1_controller",
|
||
"ok": ctrl_ok and mode.lower() == "rule",
|
||
"detail": f"ok={ctrl_ok} mode={mode or '—'}",
|
||
})
|
||
|
||
# E2 serial
|
||
risk = str((np.get("n4_n5_isolation") or {}).get("risk") or "")
|
||
checks.append({
|
||
"id": "E2_serial",
|
||
"ok": risk != "crit",
|
||
"detail": f"serial_risk={risk or '—'}",
|
||
})
|
||
|
||
# E3 relays
|
||
rc = rel.get("counts") if isinstance(rel.get("counts"), dict) else {}
|
||
rel_ok_n = int(rc.get("ok") or 0)
|
||
checks.append({
|
||
"id": "E3_relays",
|
||
"ok": rel_ok_n >= 1,
|
||
"detail": f"relays_ok={rel_ok_n}/{rc.get('n') or 0}",
|
||
})
|
||
|
||
# E4 optional call
|
||
call: dict[str, Any] | None = None
|
||
base = os.environ.get("JOY_E2E_BASE_URL") or os.environ.get("OPENAI_BASE_URL") or ""
|
||
key = os.environ.get("JOY_E2E_API_KEY") or os.environ.get("OPENAI_API_KEY") or ""
|
||
model = os.environ.get("JOY_E2E_MODEL") or "gpt-4o-mini"
|
||
if do_call:
|
||
if base and key:
|
||
call = _chat_probe(base, key, model)
|
||
call = {k: v for k, v in call.items() if k != "key"}
|
||
call["base_host"] = base.split("/")[2] if "://" in base else base[:40]
|
||
err_s = str(call.get("error") or "")
|
||
bal_fail = "INSUFFICIENT" in err_s.upper() or "balance" in err_s.lower() or "余额" in err_s
|
||
call["insufficient_balance"] = bal_fail
|
||
if bal_fail:
|
||
detail = f"http={call.get('http')} 账户余额不足 host={call.get('base_host')}"
|
||
else:
|
||
detail = f"http={call.get('http')} ms={call.get('ms')} err={err_s or '—'}"
|
||
checks.append({
|
||
"id": "E4_completion",
|
||
"ok": bool(call.get("ok")),
|
||
"detail": detail,
|
||
"insufficient_balance": bal_fail,
|
||
})
|
||
else:
|
||
checks.append({
|
||
"id": "E4_completion",
|
||
"ok": False,
|
||
"detail": "缺 JOY_E2E_BASE_URL / JOY_E2E_API_KEY 或 --from-endpoints",
|
||
})
|
||
else:
|
||
checks.append({
|
||
"id": "E4_completion",
|
||
"ok": True,
|
||
"detail": "skipped（未 --call）",
|
||
"skipped": True,
|
||
})
|
||
|
||
hard = [c for c in checks if not c.get("skipped")]
|
||
fail = [c for c in hard if not c.get("ok")]
|
||
bal_only = (
|
||
len(fail) == 1
|
||
and fail[0].get("id") == "E4_completion"
|
||
and fail[0].get("insufficient_balance")
|
||
)
|
||
if any(c["id"] in ("E1_controller", "E2_serial") and not c["ok"] for c in hard):
|
||
level = "crit"
|
||
elif fail:
|
||
level = "warn"
|
||
else:
|
||
level = "ok"
|
||
|
||
# 路径通但中转没钱：整体 ok=False，但 headline 说清楚
|
||
ok = len(fail) == 0
|
||
if ok:
|
||
headline = "端到端 OK"
|
||
next_s = "正常"
|
||
elif bal_only:
|
||
headline = "路径通 · 中转余额不足（INSUFFICIENT_BALANCE）"
|
||
next_s = "给 tokenx24/中转充值，或换 JOY_E2E_* 有额端点；号池仍看 codex health"
|
||
else:
|
||
headline = f"端到端失败 ×{len(fail)}: " + ",".join(c["id"] for c in fail)
|
||
next_s = "clashctl ensure-direct / 查中转 / clashctl e2e --call --from-endpoints"
|
||
|
||
status = {
|
||
"ts": _now(),
|
||
"ok": ok,
|
||
"level": level,
|
||
"headline": headline,
|
||
"next": next_s,
|
||
"path_id": np.get("path_id"),
|
||
"checks": checks,
|
||
"call": call,
|
||
"selected": prim.get("selected"),
|
||
"mode": mode,
|
||
}
|
||
OUT.parent.mkdir(parents=True, exist_ok=True)
|
||
tmp = OUT.with_suffix(".tmp")
|
||
tmp.write_text(json.dumps(status, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
tmp.rename(OUT)
|
||
return status
|
||
|
||
|
||
def _load_endpoint_creds() -> tuple[str, str, str]:
|
||
"""从 ~/data/codex-tokens/api-endpoints.json 取第一个 alive 端点（不打印 key）。"""
|
||
p = Path.home() / "data" / "codex-tokens" / "api-endpoints.json"
|
||
if not p.is_file():
|
||
return "", "", ""
|
||
try:
|
||
data = json.loads(p.read_text(encoding="utf-8"))
|
||
except (OSError, json.JSONDecodeError):
|
||
return "", "", ""
|
||
eps = data.get("endpoints") if isinstance(data, dict) else data
|
||
if not isinstance(eps, list):
|
||
return "", "", ""
|
||
# 优先 alive，再 degraded
|
||
ordered = sorted(
|
||
[x for x in eps if isinstance(x, dict)],
|
||
key=lambda x: 0 if str(x.get("status") or "") == "alive" else 1,
|
||
)
|
||
for item in ordered:
|
||
base = str(item.get("base_url") or item.get("url") or "").rstrip("/")
|
||
key = str(item.get("api_key") or item.get("key") or item.get("token") or "")
|
||
if not base or not key:
|
||
continue
|
||
# base 可能已带 /v1
|
||
if base.endswith("/v1"):
|
||
base = base[: -len("/v1")]
|
||
model = str(item.get("model") or item.get("default_model") or "") or (
|
||
(item.get("models") or [None])[0] if isinstance(item.get("models"), list) else ""
|
||
) or "gpt-4o-mini"
|
||
return base, key, str(model)
|
||
return "", "", ""
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--call", action="store_true", help="发真实 chat completion（需 env key）")
|
||
ap.add_argument(
|
||
"--from-endpoints",
|
||
action="store_true",
|
||
help="--call 时从 ~/data/codex-tokens/api-endpoints.json 取 base+key（mode 600）",
|
||
)
|
||
ap.add_argument("--json", action="store_true")
|
||
args = ap.parse_args()
|
||
if args.call and args.from_endpoints:
|
||
base, key, model = _load_endpoint_creds()
|
||
if base and key:
|
||
os.environ.setdefault("JOY_E2E_BASE_URL", base)
|
||
os.environ.setdefault("JOY_E2E_API_KEY", key)
|
||
if model:
|
||
os.environ.setdefault("JOY_E2E_MODEL", model)
|
||
# 绝不 print key
|
||
host = base.split("/")[2] if "://" in base else base[:40]
|
||
print(f"[e2e] using endpoint host={host} model={model or 'default'}")
|
||
else:
|
||
print("[e2e] --from-endpoints: no alive endpoint with key", file=__import__("sys").stderr)
|
||
st = run(do_call=args.call)
|
||
if args.json:
|
||
print(json.dumps(st, ensure_ascii=False, indent=2))
|
||
else:
|
||
print(f"[{st.get('level')}] {st.get('headline')}")
|
||
print(f" wrote {OUT}")
|
||
for c in st.get("checks") or []:
|
||
mark = "ok" if c.get("ok") else "FAIL"
|
||
skip = " (skip)" if c.get("skipped") else ""
|
||
print(f" [{mark}] {c.get('id')}: {c.get('detail')}{skip}")
|
||
return 0 if st.get("ok") else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|
