"""拉取各 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 get_instance, 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 _close_timeout() -> float:
    try:
        return max(8.0, float(os.getenv("HUB_CLOSE_TIMEOUT_SEC") or 45))
    except (TypeError, ValueError):
        return 45.0


def _headers(token: str) -> dict[str, str]:
    return {
        "X-Hub-Token": token,
        "Accept": "application/json",
        "Content-Type": "application/json",
    }


def _response_json(r: requests.Response) -> tuple[dict[str, Any] | None, str]:
    """安全解析 JSON;失败返回 (None, 可读原因)."""
    raw = (r.content or b"").strip()
    if not raw:
        return None, f"HTTP {r.status_code} 空响应"
    try:
        data = r.json()
    except ValueError:
        snippet = raw[:120].decode("utf-8", errors="replace").replace("\n", " ")
        hint = ""
        if r.status_code in (401, 403):
            hint = ";检查 Bridge Token"
        elif r.status_code == 404:
            hint = ";客户端可能未部署 /api/hub/version(请先手动更新实例)"
        elif snippet.lstrip().startswith("<"):
            hint = ";收到 HTML(可能被登录页/反代拦截)"
        return None, f"HTTP {r.status_code} 非 JSON{hint}: {snippet}"
    if not isinstance(data, dict):
        return None, f"HTTP {r.status_code} 响应不是对象"
    return data, ""


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 post_close_full(inst: dict[str, Any], *, inst_id: str) -> dict[str, Any]:
    """向客户端下发买一全平指令;下单在客户端本地执行."""
    base = (inst.get("base_url") or "").rstrip("/")
    token = (inst.get("bridge_token") or "").strip()
    iid = (inst_id or "").strip()
    if not base or not token:
        return {"ok": False, "msg": "缺少 URL 或 Token"}
    if not iid:
        return {"ok": False, "msg": "缺少 inst_id"}
    url = f"{base}/api/hub/options/close"
    try:
        r = requests.post(
            url,
            headers=_headers(token),
            json={"inst_id": iid},
            timeout=_close_timeout(),
        )
        try:
            data = r.json()
        except Exception:
            data = {}
        if isinstance(data, dict) and data:
            if r.status_code >= 400 and "ok" not in data:
                data = dict(data)
                data["ok"] = False
                data.setdefault("msg", f"HTTP {r.status_code}")
            return data
        return {"ok": False, "msg": f"HTTP {r.status_code}"}
    except requests.Timeout:
        return {"ok": False, "msg": "客户端超时(下单可能仍在进行,请刷新确认)"}
    except Exception as e:
        return {"ok": False, "msg": str(e) or "下发失败"}


def post_cond_order(inst: dict[str, Any], payload: dict[str, Any]) -> dict[str, Any]:
    base = (inst.get("base_url") or "").rstrip("/")
    token = (inst.get("bridge_token") or "").strip()
    if not base or not token:
        return {"ok": False, "msg": "缺少 URL 或 Token", "instance_id": inst.get("id")}
    url = f"{base}/api/hub/options/cond_orders"
    try:
        r = requests.post(url, headers=_headers(token), json=payload, timeout=_close_timeout())
        data = r.json() if r.content else {}
        if not isinstance(data, dict):
            data = {"ok": False, "msg": f"HTTP {r.status_code}"}
        data["instance_id"] = inst.get("id")
        data["instance_name"] = inst.get("name")
        return data
    except requests.Timeout:
        return {
            "ok": False,
            "msg": "客户端超时",
            "instance_id": inst.get("id"),
            "instance_name": inst.get("name"),
        }
    except Exception as e:
        return {
            "ok": False,
            "msg": str(e) or "下发失败",
            "instance_id": inst.get("id"),
            "instance_name": inst.get("name"),
        }


def post_cond_cancel(inst: dict[str, Any], hub_order_id: str) -> dict[str, Any]:
    base = (inst.get("base_url") or "").rstrip("/")
    token = (inst.get("bridge_token") or "").strip()
    oid = (hub_order_id or "").strip()
    if not base or not token or not oid:
        return {"ok": False, "msg": "参数不完整"}
    url = f"{base}/api/hub/options/cond_orders/{oid}/cancel"
    try:
        r = requests.post(url, headers=_headers(token), json={}, timeout=_timeout())
        data = r.json() if r.content else {}
        return data if isinstance(data, dict) else {"ok": False, "msg": f"HTTP {r.status_code}"}
    except Exception as e:
        return {"ok": False, "msg": str(e) or "取消失败"}


def post_cond_exit(inst: dict[str, Any], hub_order_id: str, payload: dict[str, Any]) -> dict[str, Any]:
    base = (inst.get("base_url") or "").rstrip("/")
    token = (inst.get("bridge_token") or "").strip()
    oid = (hub_order_id or "").strip()
    if not base or not token or not oid:
        return {"ok": False, "msg": "参数不完整"}
    url = f"{base}/api/hub/options/cond_orders/{oid}/exit"
    try:
        r = requests.post(url, headers=_headers(token), json=payload or {}, timeout=_timeout())
        data = r.json() if r.content else {}
        return data if isinstance(data, dict) else {"ok": False, "msg": f"HTTP {r.status_code}"}
    except Exception as e:
        return {"ok": False, "msg": str(e) or "更新失败"}


def fetch_cond_orders(inst: dict[str, Any]) -> dict[str, Any]:
    base = (inst.get("base_url") or "").rstrip("/")
    token = (inst.get("bridge_token") or "").strip()
    if not base or not token:
        return {"ok": False, "msg": "缺少 URL 或 Token", "orders": []}
    url = f"{base}/api/hub/options/cond_orders"
    try:
        r = requests.get(
            url,
            headers={"X-Hub-Token": token, "Accept": "application/json"},
            timeout=_timeout(),
        )
        data = r.json() if r.content else {}
        if isinstance(data, dict):
            data["instance_id"] = inst.get("id")
            data["instance_name"] = inst.get("name")
            return data
        return {"ok": False, "msg": f"HTTP {r.status_code}", "orders": []}
    except Exception as e:
        return {"ok": False, "msg": str(e) or "拉取失败", "orders": [], "instance_id": inst.get("id")}


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


def fetch_one_stats(inst: dict[str, Any], *, window: str = "7d") -> 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"),
        "online": False,
        "error": "",
        "stats": None,
    }
    if not base or not token:
        out["error"] = "缺少 URL 或 Token"
        return out
    url = f"{base}/api/hub/options/stats"
    try:
        r = requests.get(
            url,
            headers={"X-Hub-Token": token, "Accept": "application/json"},
            params={"window": window},
            timeout=max(_timeout(), 15.0),
        )
        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["stats"] = data if isinstance(data, dict) else None
            return out
        out["online"] = True
        out["stats"] = data
        return out
    except requests.Timeout:
        out["error"] = "超时"
        return out
    except Exception as e:
        out["error"] = str(e) or "拉取失败"
        return out


def pull_all_stats(window: str = "7d") -> 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_stats, inst, window=window): 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_stats(inst, window=window))
    return results


def fetch_client_version(inst: dict[str, Any]) -> dict[str, Any]:
    """探测客户端 git 版本 / 在线."""
    base = (inst.get("base_url") or "").rstrip("/")
    token = (inst.get("bridge_token") or "").strip()
    out: dict[str, Any] = {
        "ok": False,
        "online": False,
        "sha": "",
        "short_sha": "",
        "branch": "",
        "dirty": False,
        "mode": "",
        "mode_label": "",
        "msg": "",
        "instance_id": inst.get("id"),
        "instance_name": inst.get("name"),
    }
    if not base or not token:
        out["msg"] = "缺少 URL 或 Token"
        return out
    url = f"{base}/api/hub/version"
    try:
        r = requests.get(
            url,
            headers={"X-Hub-Token": token, "Accept": "application/json"},
            timeout=_timeout(),
        )
        data, err = _response_json(r)
        if data is None:
            out["msg"] = err or f"HTTP {r.status_code}"
            return out
        for key in ("sha", "short_sha", "branch", "dirty", "msg", "repo", "mode", "mode_label"):
            if key in data:
                out[key] = data.get(key)
        out["ok"] = bool(data.get("ok")) and r.status_code == 200
        out["online"] = out["ok"]
        if not out["ok"]:
            out["msg"] = data.get("msg") or err or f"HTTP {r.status_code}"
        return out
    except requests.Timeout:
        out["msg"] = "超时"
        return out
    except Exception as e:
        out["msg"] = str(e) or "探测失败"
        return out


def post_client_update(inst: dict[str, Any]) -> dict[str, Any]:
    """触发客户端快速更新(git pull + pm2 restart)."""
    base = (inst.get("base_url") or "").rstrip("/")
    token = (inst.get("bridge_token") or "").strip()
    if not base or not token:
        return {
            "ok": False,
            "msg": "缺少 URL 或 Token",
            "instance_id": inst.get("id"),
            "instance_name": inst.get("name"),
        }
    url = f"{base}/api/hub/update"
    try:
        r = requests.post(
            url,
            headers=_headers(token),
            json={},
            timeout=_timeout(),
        )
        data, err = _response_json(r)
        if data is None:
            data = {"ok": False, "msg": err or f"HTTP {r.status_code}"}
        data["instance_id"] = inst.get("id")
        data["instance_name"] = inst.get("name")
        if r.status_code >= 400 and not data.get("ok"):
            data["ok"] = False
            data.setdefault("msg", err or f"HTTP {r.status_code}")
        return data
    except requests.Timeout:
        return {
            "ok": False,
            "msg": "客户端超时(更新可能仍在进行,请刷新版本确认)",
            "instance_id": inst.get("id"),
            "instance_name": inst.get("name"),
        }
    except Exception as e:
        return {
            "ok": False,
            "msg": str(e) or "下发失败",
            "instance_id": inst.get("id"),
            "instance_name": inst.get("name"),
        }


def fetch_backup_list(inst: dict[str, Any]) -> dict[str, Any]:
    base = (inst.get("base_url") or "").rstrip("/")
    token = (inst.get("bridge_token") or "").strip()
    if not base or not token:
        return {"ok": False, "msg": "缺少 URL 或 Token", "items": []}
    url = f"{base}/api/hub/backup/list"
    try:
        r = requests.get(
            url,
            headers={"X-Hub-Token": token, "Accept": "application/json"},
            timeout=_timeout(),
        )
        data, err = _response_json(r)
        if data is None:
            return {
                "ok": False,
                "msg": err or f"HTTP {r.status_code}",
                "items": [],
                "instance_id": inst.get("id"),
                "instance_name": inst.get("name"),
            }
        data.setdefault("instance_id", inst.get("id"))
        data.setdefault("instance_name", inst.get("name"))
        return data
    except Exception as e:
        return {
            "ok": False,
            "msg": str(e) or "拉取失败",
            "items": [],
            "instance_id": inst.get("id"),
            "instance_name": inst.get("name"),
        }


def fetch_backup_file(inst: dict[str, Any], name: str) -> tuple[bool, bytes | None, str]:
    """返回 (ok, content, msg_or_filename)."""
    base = (inst.get("base_url") or "").rstrip("/")
    token = (inst.get("bridge_token") or "").strip()
    safe = (name or "").strip()
    if not base or not token or not safe:
        return False, None, "参数不完整"
    url = f"{base}/api/hub/backup/download/{safe}"
    try:
        r = requests.get(
            url,
            headers={"X-Hub-Token": token, "Accept": "application/gzip, application/json"},
            timeout=max(_timeout(), 60.0),
            stream=True,
        )
        ctype = (r.headers.get("content-type") or "").lower()
        if r.status_code != 200:
            try:
                data = r.json()
                return False, None, (data.get("msg") if isinstance(data, dict) else None) or f"HTTP {r.status_code}"
            except Exception:
                return False, None, f"HTTP {r.status_code}"
        if "json" in ctype:
            try:
                data = r.json()
                return False, None, (data.get("msg") if isinstance(data, dict) else None) or "下载失败"
            except Exception:
                return False, None, "下载失败"
        return True, r.content, os.path.basename(safe)
    except requests.Timeout:
        return False, None, "下载超时"
    except Exception as e:
        return False, None, str(e) or "下载失败"


def dispatch_cond_order_to_ids(instance_ids: list[int], payload: dict[str, Any]) -> list[dict[str, Any]]:
    results = []
    for iid in instance_ids:
        inst = get_instance(int(iid))
        if not inst:
            results.append({"ok": False, "msg": "实例不存在", "instance_id": iid})
            continue
        if not inst.get("enabled"):
            results.append(
                {
                    "ok": False,
                    "msg": "实例未启用",
                    "instance_id": iid,
                    "instance_name": inst.get("name"),
                }
            )
            continue
        results.append(post_cond_order(inst, payload))
    return results