joy-cli/scripts/manhuang/refresh-chain-breaks.py

#!/usr/bin/env python3
"""本地缓存刷新 chain_v2 breaks（不 SSH scout）。

读 net-path / relays / quotes / channels catalog → breaks.json
供 actions-to-board / 面板旁路使用。

用法:
  PYTHONPATH=src python3 scripts/manhuang/refresh-chain-breaks.py
  PYTHONPATH=src python3 scripts/manhuang/refresh-chain-breaks.py --json
"""
from __future__ import annotations

import argparse
import json
import sys
import time
from pathlib import Path

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT / "src"))

from joy.dashboard.manhuang import (  # noqa: E402
    _build_chain_status,
    _compute_path_id,
    _load_local_codex_summary,
    _load_net_path,
    _load_relays_status,
    _pay_rail_meta,
    _record_pool_quota,
    _sms_source_meta,
    _write_chain_breaks_cache,
    _QUOTES_AI,
)


def _load_local_summary() -> dict:
    """从本机缓存拼最小 summary（无 scout）。"""
    summary: dict = {
        "codex_total": 0,
        "codex_quota": 0,
        "codex_exhausted": 0,
        "codex_dead": 0,
        "live_probes_ok": 0,
        "live_probes_fail": 0,
        "free_endpoints": 0,
        "tor_socks": False,
        "tor_active": 0,
        "guishi_channels": 0,
        "proxies_alive": 0,
        "miling_need_register": 0,
    }
    cx = _load_local_codex_summary()
    if cx.get("ok") or cx.get("path"):
        summary["codex_total"] = int(cx.get("codex_total") or 0)
        summary["codex_quota"] = int(cx.get("codex_quota") or 0)
        summary["codex_exhausted"] = int(cx.get("codex_exhausted") or 0)
        summary["codex_dead"] = int(cx.get("codex_dead") or 0)
        summary["codex_health_path"] = cx.get("path")
    # health 文件存在但 total=0 也算已同步空仓
    hs = Path.home() / "data" / "codex-tokens" / "health-status.json"
    if hs.is_file() and not summary.get("codex_health_path"):
        summary["codex_health_path"] = str(hs)
    # tor
    try:
        import socket
        s = socket.socket()
        s.settimeout(0.4)
        s.connect(("127.0.0.1", 9050))
        s.close()
        summary["tor_socks"] = True
        summary["tor_active"] = 1
    except OSError:
        pass
    return summary


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--json", action="store_true")
    args = ap.parse_args()

    np = _load_net_path()
    if isinstance(np, dict) and not np.get("path_id"):
        pid = _compute_path_id(np)
        if pid:
            np["path_id"] = pid
    # 实时 scutil 覆盖缓存（避免 status.json 陈旧导致假 path_sys_proxy_off）
    try:
        import importlib.util
        pp = Path(__file__).resolve().parent / "probe-proxy-path.py"
        spec = importlib.util.spec_from_file_location("_mh_pp", pp)
        if spec and spec.loader:
            mod = importlib.util.module_from_spec(spec)
            spec.loader.exec_module(mod)
            if hasattr(mod, "probe_system_proxy"):
                sp = mod.probe_system_proxy()
                if isinstance(np, dict) and isinstance(sp, dict) and sp.get("ok"):
                    np["system_proxy"] = sp
    except Exception:
        pass
    rel = _load_relays_status()
    sms = _sms_source_meta()
    summary = _load_local_summary()
    pmeta = _record_pool_quota(
        int(summary.get("codex_quota") or 0),
        int(summary.get("codex_total") or 0),
    )
    q_skus = None
    if _QUOTES_AI.is_file():
        try:
            qd = json.loads(_QUOTES_AI.read_text(encoding="utf-8"))
            if isinstance(qd, dict):
                q_skus = qd.get("skus")
        except (OSError, json.JSONDecodeError):
            q_skus = None
    pay = _pay_rail_meta(quotes_skus=q_skus if isinstance(q_skus, list) else None)

    # 轻量 e2e：仅当缓存缺失/过旧时跑 E0–E3；不覆盖近期 --call 结果
    try:
        e2e_path = Path.home() / ".cache" / "manhuang" / "e2e" / "status.json"
        age = 1e9
        if e2e_path.is_file():
            age = max(0, time.time() - e2e_path.stat().st_mtime)
        had_call = False
        if e2e_path.is_file() and age < 3600:
            try:
                prev = json.loads(e2e_path.read_text(encoding="utf-8"))
                had_call = bool((prev or {}).get("call"))
            except (OSError, json.JSONDecodeError):
                had_call = False
        if age > 1800 and not had_call:
            import importlib.util
            ep = Path(__file__).resolve().parent / "probe-e2e-capability.py"
            if ep.is_file():
                spec = importlib.util.spec_from_file_location("_mh_e2e", ep)
                if spec and spec.loader:
                    mod = importlib.util.module_from_spec(spec)
                    spec.loader.exec_module(mod)
                    if hasattr(mod, "run"):
                        mod.run(do_call=False)
    except Exception:
        pass

    # free endpoints from free list if present
    free_p = Path.home() / ".cache" / "manhuang" / "quotes" / "free-llm-lists.json"
    if free_p.is_file():
        try:
            fl = json.loads(free_p.read_text(encoding="utf-8"))
            n = fl.get("n") or len(fl.get("providers") or fl.get("items") or [])
            summary["free_endpoints"] = int(n or 0)
        except (OSError, json.JSONDecodeError, TypeError, ValueError):
            pass

    chain = _build_chain_status(
        summary=summary,
        miling={"summary": {}},
        blackmarket={"skus": q_skus or [], "demand": [], "channels": []},
        pool={},
        net_path=np,
        relays=rel,
        pool_meta=pmeta,
        sms_meta=sms,
        pay_meta=pay,
    )
    path = _write_chain_breaks_cache(chain)
    if args.json:
        print(json.dumps(chain, ensure_ascii=False, indent=2))
    else:
        print(f"[{chain.get('mode')}] {chain.get('headline')}")
        print(f"  wrote {path}")
        print(f"  path_id={chain.get('path_id')} breaks={len(chain.get('breaks') or [])}")
        top = chain.get("top_break") or {}
        if top:
            print(f"  top: {top.get('reason')} · {top.get('title')}")
        for b in (chain.get("breaks") or [])[:6]:
            print(f"  - [{b.get('level')}] {b.get('reason')}: {b.get('title')}")
        pay_d = chain.get("pay") or {}
        print(f"  pay: market={pay_d.get('market_n')} msrp={pay_d.get('msrp_n')} sku={pay_d.get('sku_n')}")
    return 0 if chain.get("ok") else 1


if __name__ == "__main__":
    raise SystemExit(main())