"""拉取各 crypto_okx 客户端只读快照.""" from __future__ import annotations import os import time from concurrent.futures import ThreadPoolExecutor, as_completed from typing import Any import requests from lib.instances_lib import list_instances def _timeout() -> float: try: return max(2.0, float(os.getenv("HUB_PULL_TIMEOUT_SEC") or 5)) except (TypeError, ValueError): return 5.0 def fetch_one_snapshot(inst: dict[str, Any]) -> dict[str, Any]: base = (inst.get("base_url") or "").rstrip("/") token = (inst.get("bridge_token") or "").strip() out: dict[str, Any] = { "id": inst.get("id"), "name": inst.get("name"), "base_url": base, "embed_path": inst.get("embed_path") or "/embed", "tags": inst.get("tags") or "", "online": False, "error": "", "snapshot": None, "pulled_at": time.strftime("%Y-%m-%d %H:%M:%S"), } if not base or not token: out["error"] = "缺少 URL 或 Token" return out url = f"{base}/api/hub/options/snapshot" try: r = requests.get( url, headers={"X-Hub-Token": token, "Accept": "application/json"}, timeout=_timeout(), ) data = {} try: data = r.json() except Exception: data = {} if r.status_code != 200 or not isinstance(data, dict) or not data.get("ok"): out["error"] = (data.get("msg") if isinstance(data, dict) else None) or f"HTTP {r.status_code}" out["snapshot"] = data if isinstance(data, dict) else None return out out["online"] = True out["snapshot"] = data return out except requests.Timeout: out["error"] = "超时" return out except Exception as e: out["error"] = str(e) or "拉取失败" return out def pull_all_enabled() -> list[dict[str, Any]]: instances = list_instances(enabled_only=True) if not instances: return [] results: list[dict[str, Any]] = [] with ThreadPoolExecutor(max_workers=min(8, len(instances))) as pool: futs = {pool.submit(fetch_one_snapshot, inst): inst["id"] for inst in instances} by_id: dict[Any, dict[str, Any]] = {} for fut in as_completed(futs): row = fut.result() by_id[row.get("id")] = row for inst in instances: results.append(by_id.get(inst["id"]) or fetch_one_snapshot(inst)) return results