"""RoboHub project-backend reference client (standard library only)."""

from __future__ import annotations

import hashlib
import hmac
import ipaddress
import json
import time
import urllib.error
import urllib.parse
import urllib.request
import uuid
from typing import Any


def masked_proxy_endpoint(credential: dict[str, Any]) -> str:
    host = str(credential.get("host") or "")
    try:
        address = ipaddress.ip_address(host)
        if address.version == 4:
            parts = host.split(".")
            masked = f"{parts[0]}.{parts[1]}.*.*"
        else:
            masked = ":".join(address.exploded.split(":")[:2]) + ":*"
    except ValueError:
        parts = host.split(".")
        masked = "*." + ".".join(parts[1:]) if len(parts) > 1 else "***"
    return f"{credential.get('scheme') or 'http'}://{masked}:{int(credential.get('port') or 0)}"


def proxy_result_metadata(credential: dict[str, Any], observed_exit_ip: str | None = None) -> dict[str, Any]:
    try:
        exit_ip = str(ipaddress.ip_address(observed_exit_ip)) if observed_exit_ip else None
    except ValueError:
        exit_ip = None
    return {
        "proxy_egress_id": str(credential["egress_id"]),
        "proxy_node_hash": str(credential["node_hash"]),
        "proxy_endpoint_masked": masked_proxy_endpoint(credential),
        "proxy_exit_ip": exit_ip,
    }


def distinct_egress_families(values: list[dict[str, Any]], *, require_observed_identity: bool = False) -> list[dict[str, Any]]:
    """Keep one entry per observed egress family without inventing identity."""
    selected: list[dict[str, Any]] = []
    seen: set[str] = set()
    for value in values or []:
        family = str(value.get("egress_family_key") or "").strip()
        observed = str(value.get("identity_status") or "unknown") == "passed" and bool(family)
        if require_observed_identity and not observed:
            continue
        key = family if observed else "unknown"
        if not key or key in seen:
            continue
        seen.add(key)
        selected.append(value)
    return selected


class PoolCredentialAllocator:
    def __init__(self, credentials: list[dict[str, Any]]):
        if not credentials:
            raise ValueError("RoboHub pool credentials are required")
        egress_ids = [str(item.get("egress_id") or "") for item in credentials]
        if any(not item.get("node_hash") or not item.get("host") or not int(item.get("port") or 0) for item in credentials):
            raise ValueError("RoboHub pool credential is invalid")
        if any(not value for value in egress_ids) or len(egress_ids) != len(set(egress_ids)):
            raise ValueError("RoboHub pool egress_id must be unique")
        self.credentials = list(credentials)

    def assign(self, worker_index: int, attempt: int = 0) -> tuple[dict[str, Any], dict[str, Any]]:
        if worker_index < 0 or attempt < 0:
            raise ValueError("worker_index and attempt must be non-negative")
        credential = self.credentials[(worker_index + attempt) % len(self.credentials)]
        return credential, proxy_result_metadata(credential)


def observe_proxy_exit_ip(credential: dict[str, Any], proxy_request, echo_url: str) -> str | None:
    """Observe the public IP through the supplied credential-aware requester."""
    if not callable(proxy_request):
        raise ValueError("A proxy-aware request function is required")
    value = proxy_request(url=echo_url, credential=credential)
    if isinstance(value, dict):
        value = value.get("ip") or value.get("origin") or ""
    candidate = str(value or "").split(",", 1)[0].strip()
    try:
        return str(ipaddress.ip_address(candidate))
    except ValueError:
        return None


def summarize_by_run_and_egress(results: list[dict[str, Any]]) -> list[dict[str, Any]]:
    summary: dict[tuple[str, str], dict[str, Any]] = {}
    for result in results:
        run_id = str(result.get("run_id") or "")
        egress_id = str(result.get("proxy_egress_id") or "")
        if not run_id or not egress_id:
            continue
        row = summary.setdefault((run_id, egress_id), {"run_id": run_id, "proxy_egress_id": egress_id, "total": 0, "completed": 0, "failed": 0})
        row["total"] += 1
        if str(result.get("status") or "").lower() in {"completed", "success", "succeeded"}:
            row["completed"] += 1
        else:
            row["failed"] += 1
    return [summary[key] for key in sorted(summary)]


class RoboHubApiError(RuntimeError):
    def __init__(self, status: int, body: dict[str, Any] | None):
        body = body or {}
        super().__init__(str(body.get("message") or f"RoboHub request failed: HTTP {status}"))
        self.status = status
        self.code = str(body.get("code") or "RESPONSE_INVALID")
        self.request_id = body.get("request_id")


class RoboHubClient:
    def __init__(self, base_url: str, project_id: str, hmac_key: str, timeout_seconds: float = 240):
        if not base_url or not project_id or not hmac_key:
            raise ValueError("base_url, project_id, and hmac_key are required")
        self.base_url = base_url.rstrip("/")
        self.project_id = project_id
        self.hmac_key = hmac_key.encode("utf-8")
        self.timeout_seconds = timeout_seconds

    def request(
        self,
        project_user_id: str,
        method: str,
        path: str,
        body: dict[str, Any] | None = None,
        *,
        idempotency_key: str | None = None,
    ) -> dict[str, Any]:
        method = method.upper()
        raw = b"" if body is None else json.dumps(body, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
        parsed = urllib.parse.urlsplit(path)
        timestamp = str(int(time.time()))
        nonce = uuid.uuid4().hex
        body_hash = hashlib.sha256(raw).hexdigest()
        canonical = "\n".join((method, parsed.path, body_hash, timestamp, nonce, self.project_id))
        headers = {
            "X-RoboHub-Project": self.project_id,
            "X-RoboHub-User": project_user_id,
            "X-RoboHub-Timestamp": timestamp,
            "X-RoboHub-Nonce": nonce,
            "X-RoboHub-Signature": hmac.new(self.hmac_key, canonical.encode("utf-8"), hashlib.sha256).hexdigest(),
            "X-Request-Id": uuid.uuid4().hex,
        }
        if body is not None:
            headers["Content-Type"] = "application/json"
        if idempotency_key:
            headers["Idempotency-Key"] = idempotency_key
        request = urllib.request.Request(self.base_url + path, data=raw or None, headers=headers, method=method)
        try:
            with urllib.request.urlopen(request, timeout=self.timeout_seconds) as response:
                return json.loads(response.read())
        except urllib.error.HTTPError as error:
            try:
                payload = json.loads(error.read())
            except (TypeError, ValueError, json.JSONDecodeError):
                payload = None
            raise RoboHubApiError(error.code, payload) from error

    def bind_account(
        self, user: str, *, idempotency_key: str, account_ref: str | None = None,
        account_email: str | None = None, account_password: str | None = None,
    ) -> dict[str, Any]:
        body = {"mode": "account", "account_ref": account_ref} if account_ref else {
            "mode": "account", "account_email": account_email, "account_password": account_password,
        }
        return self.request(user, "POST", "/v1/bindings/login", body, idempotency_key=idempotency_key)

    def list_services(self, user: str) -> dict[str, Any]:
        return self.request(user, "GET", "/v1/services")

    def get_account(self, user: str) -> dict[str, Any]:
        return self.request(user, "GET", "/v1/account")

    def list_nodes(self, user: str, service_id: str, *, cursor: str | None = None, limit: int = 50) -> dict[str, Any]:
        query = urllib.parse.urlencode({key: value for key, value in {"cursor": cursor, "limit": limit}.items() if value is not None})
        return self.request(user, "GET", f"/v1/services/{urllib.parse.quote(service_id, safe='')}/nodes?{query}")

    def list_active_dynamic_lines(self, user: str, service_id: str) -> dict[str, Any]:
        return self.request(user, "GET", f"/v1/services/{urllib.parse.quote(service_id, safe='')}/active-dynamic/lines")

    def rotate_active_dynamic_lines(self, user: str, service_id: str, line_ids: list[str], *, idempotency_key: str) -> dict[str, Any]:
        return self.request(
            user,
            "POST",
            f"/v1/services/{urllib.parse.quote(service_id, safe='')}/active-dynamic/rotate",
            {"line_ids": line_ids},
            idempotency_key=idempotency_key,
        )

    def start_node_test(self, user: str, service_id: str, node_hashes: list[str], *, idempotency_key: str) -> dict[str, Any]:
        return self.request(user, "POST", "/v1/node-tests", {
            "service_id": service_id, "node_hashes": node_hashes, "probe_target_class": "all",
        }, idempotency_key=idempotency_key)

    def wait_node_test(
        self, user: str, job: dict[str, Any], *, timeout_seconds: float = 180,
        poll_seconds: float = 1, require_all_passed: bool = True,
    ) -> dict[str, Any]:
        deadline = time.monotonic() + timeout_seconds
        terminal = {"completed", "failed", "cancelled", "expired"}
        current = job
        while current.get("status") not in terminal:
            if time.monotonic() >= deadline:
                raise TimeoutError("RoboHub node test polling timed out")
            time.sleep(poll_seconds)
            current = self.request(user, "GET", f"/v1/node-tests/{urllib.parse.quote(str(current['job_id']), safe='')}")
        if require_all_passed and not (current.get("status") == "completed" and current.get("passed") == current.get("total")):
            raise RoboHubApiError(409, {"code": "NODE_TEST_REQUIRED", "message": "selected nodes did not pass all required probes"})
        return current

    def cancel_node_test(self, user: str, job_id: str, *, idempotency_key: str) -> dict[str, Any]:
        encoded = urllib.parse.quote(job_id, safe="")
        return self.request(user, "DELETE", f"/v1/node-tests/{encoded}", idempotency_key=idempotency_key)

    def ensure_runtime(self, user: str, service_id: str, node_hashes: list[str], *, idempotency_key: str) -> dict[str, Any]:
        return self.request(user, "POST", "/v1/runtime/ensure", {
            "service_id": service_id, "node_hashes": node_hashes,
        }, idempotency_key=idempotency_key)

    def get_runtime_status(self, user: str) -> dict[str, Any]:
        return self.request(user, "GET", "/v1/runtime/status")

    @staticmethod
    def _lease_body(runtime: dict[str, Any]) -> dict[str, Any]:
        return {
            "lease_id": runtime["lease_id"],
            "fencing_token": runtime["fencing_token"],
            "expected_config_revision": runtime["config_revision"],
        }

    def issue_credentials(
        self, user: str, runtime: dict[str, Any], task_ref: str, *, idempotency_key: str, ttl_seconds: int = 120,
    ) -> dict[str, Any]:
        body = {**self._lease_body(runtime), "task_ref": task_ref, "credential_ttl_seconds": ttl_seconds}
        return self.request(user, "POST", "/v1/runtime/credentials/issue", body, idempotency_key=idempotency_key)

    def renew_credentials(
        self, user: str, runtime: dict[str, Any], task_ref: str, *, idempotency_key: str, ttl_seconds: int = 300,
    ) -> dict[str, Any]:
        body = {**self._lease_body(runtime), "task_ref": task_ref, "credential_ttl_seconds": ttl_seconds}
        return self.request(user, "POST", "/v1/runtime/credentials/renew", body, idempotency_key=idempotency_key)

    def heartbeat(self, user: str, runtime: dict[str, Any], *, idempotency_key: str) -> dict[str, Any]:
        return self.request(user, "POST", "/v1/runtime/heartbeat", self._lease_body(runtime), idempotency_key=idempotency_key)

    def probe_runtime(self, user: str, runtime: dict[str, Any], target_class: str) -> dict[str, Any]:
        return self.request(user, "POST", "/v1/runtime/probe", {
            **self._lease_body(runtime), "target_class": target_class,
        })

    def select_service(
        self, user: str, service_id: str, node_hashes: list[str], *, idempotency_key: str,
        runtime: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        body = {"service_id": service_id, "node_hashes": node_hashes}
        if runtime:
            body.update(self._lease_body(runtime))
        return self.request(user, "POST", "/v1/services/select", body, idempotency_key=idempotency_key)

    def release_runtime(self, user: str, runtime: dict[str, Any], *, idempotency_key: str) -> dict[str, Any]:
        return self.request(user, "DELETE", "/v1/runtime", self._lease_body(runtime), idempotency_key=idempotency_key)

    def ensure_pool(
        self, user: str, service_id: str, node_hashes: list[str], max_egresses: int, *,
        idempotency_key: str, egress_policy: str = "line_usable",
    ) -> dict[str, Any]:
        return self.request(user, "POST", "/v1/runtime/pool/ensure", {
            "service_id": service_id, "node_hashes": node_hashes, "max_egresses": max_egresses,
            "egress_policy": egress_policy,
        }, idempotency_key=idempotency_key)

    def get_pool_status(self, user: str) -> dict[str, Any]:
        return self.request(user, "GET", "/v1/runtime/pool/status")

    def refresh_pool_identity(self, user: str, pool: dict[str, Any], *, idempotency_key: str) -> dict[str, Any]:
        return self.request(user, "POST", "/v1/runtime/pool/identity/refresh", self._lease_body(pool), idempotency_key=idempotency_key)

    def issue_pool_credentials(
        self, user: str, pool: dict[str, Any], task_ref: str, *, idempotency_key: str, ttl_seconds: int = 120,
    ) -> dict[str, Any]:
        body = {**self._lease_body(pool), "task_ref": task_ref, "credential_ttl_seconds": ttl_seconds}
        return self.request(user, "POST", "/v1/runtime/pool/credentials/issue", body, idempotency_key=idempotency_key)

    def pool_heartbeat(self, user: str, pool: dict[str, Any], *, idempotency_key: str) -> dict[str, Any]:
        return self.request(user, "POST", "/v1/runtime/pool/heartbeat", self._lease_body(pool), idempotency_key=idempotency_key)

    def release_pool(self, user: str, pool: dict[str, Any], *, idempotency_key: str) -> dict[str, Any]:
        return self.request(user, "DELETE", "/v1/runtime/pool", self._lease_body(pool), idempotency_key=idempotency_key)

    def logout(self, user: str, *, idempotency_key: str, runtime: dict[str, Any] | None = None) -> dict[str, Any]:
        body = self._lease_body(runtime) if runtime else {}
        return self.request(user, "POST", "/v1/bindings/logout", body, idempotency_key=idempotency_key)


class RoboHubManagedPool:
    """Small backend-only adapter for the standard RoboHub registration flow.

    It deliberately persists nothing.  The caller may persist only the opaque
    ``account_ref`` in its own encrypted settings; the active pool's fencing
    token and proxy credentials stay in this process until ``close``.
    """

    def __init__(self, client: RoboHubClient, project_user_id: str, *, idempotency_prefix: str | None = None):
        if not project_user_id:
            raise ValueError("project_user_id is required")
        self.client = client
        self.project_user_id = project_user_id
        self.idempotency_prefix = idempotency_prefix or f"robohub-{uuid.uuid4().hex}"
        self.account_ref = ""
        self.pool: dict[str, Any] | None = None
        self.credentials: list[dict[str, Any]] = []

    def bind(
        self, *, account_ref: str | None = None,
        account_email: str | None = None, account_password: str | None = None,
    ) -> dict[str, Any]:
        if account_ref:
            binding = self.client.bind_account(
                self.project_user_id,
                account_ref=account_ref,
                idempotency_key=self._key("bind"),
            )
        elif account_email and account_password:
            binding = self.client.bind_account(
                self.project_user_id,
                account_email=account_email,
                account_password=account_password,
                idempotency_key=self._key("bind"),
            )
        else:
            raise ValueError("account_ref or account_email/account_password is required")
        self.account_ref = str(binding.get("account_ref") or account_ref or "")
        return binding

    def prepare(
        self, *, task_ref: str, max_workers: int,
        preferred_kind: str = "home-payg", candidate_limit: int = 50,
        credential_ttl_seconds: int = 120, egress_policy: str = "line_usable",
        node_test_timeout_seconds: float = 180,
    ) -> dict[str, Any]:
        """Select one eligible service, test nodes, then create a real pool.

        The helper never treats the upstream directory's ``available`` flag as
        final proof.  It submits the required three-target test and only sends
        nodes that later report ``runnable=true`` (and project admission when
        that field exists) into the pool request.
        """
        if not task_ref or not 1 <= max_workers <= 50 or not 1 <= candidate_limit <= 50:
            raise ValueError("task_ref, max_workers, or candidate_limit is invalid")
        service = self._select_service(self.client.list_services(self.project_user_id), preferred_kind)
        service_id = str(service["service_id"])
        catalog = self.client.list_nodes(self.project_user_id, service_id, limit=candidate_limit)
        candidates = [
            str(item.get("node_hash") or "")
            for item in catalog.get("nodes", [])
            if isinstance(item, dict) and item.get("available") is True and item.get("node_hash")
        ]
        if not candidates:
            raise RoboHubApiError(503, {"code": "NO_AVAILABLE_NODE", "message": "套餐没有可测速节点"})
        job = self.client.start_node_test(
            self.project_user_id,
            service_id,
            candidates,
            idempotency_key=self._key("node-test"),
        )
        node_test = self.client.wait_node_test(
            self.project_user_id,
            job,
            timeout_seconds=node_test_timeout_seconds,
            require_all_passed=False,
        )
        refreshed = self.client.list_nodes(self.project_user_id, service_id, limit=candidate_limit)
        selected = self._runnable_node_hashes(refreshed, candidates)
        if not selected:
            raise RoboHubApiError(409, {"code": "NODE_TEST_REQUIRED", "message": "没有节点通过当前项目的真实测速门禁"})
        self.pool = self.client.ensure_pool(
            self.project_user_id,
            service_id,
            selected,
            min(max_workers, len(selected)),
            idempotency_key=self._key("pool-ensure"),
            egress_policy=egress_policy,
        )
        issued = self.client.issue_pool_credentials(
            self.project_user_id,
            self.pool,
            task_ref,
            idempotency_key=self._key("pool-credentials"),
            ttl_seconds=credential_ttl_seconds,
        )
        self.credentials = [dict(item) for item in issued.get("credentials", []) if isinstance(item, dict)]
        if not self.credentials:
            raise RoboHubApiError(503, {"code": "EGRESS_UNAVAILABLE", "message": "RoboHub 未返回可用任务凭据"})
        return {
            "account_ref": self.account_ref or None,
            "service": service,
            "node_test": node_test,
            "selected_node_hashes": selected,
            "pool": self.pool,
            "credentials": self.credentials,
        }

    def heartbeat(self) -> dict[str, Any]:
        return self.client.pool_heartbeat(self.project_user_id, self._require_pool(), idempotency_key=self._key("pool-heartbeat"))

    def renew_credentials(self, task_ref: str, *, credential_ttl_seconds: int = 120) -> list[dict[str, Any]]:
        issued = self.client.issue_pool_credentials(
            self.project_user_id,
            self._require_pool(),
            task_ref,
            idempotency_key=self._key("pool-credentials-renew"),
            ttl_seconds=credential_ttl_seconds,
        )
        self.credentials = [dict(item) for item in issued.get("credentials", []) if isinstance(item, dict)]
        if not self.credentials:
            raise RoboHubApiError(503, {"code": "EGRESS_UNAVAILABLE", "message": "RoboHub 未返回续期凭据"})
        return list(self.credentials)

    def close(self, *, logout: bool = False) -> None:
        pool, self.pool = self.pool, None
        self.credentials = []
        if pool:
            self.client.release_pool(self.project_user_id, pool, idempotency_key=self._key("pool-release"))
        if logout:
            self.client.logout(self.project_user_id, idempotency_key=self._key("binding-logout"))

    def _key(self, operation: str) -> str:
        return f"{self.idempotency_prefix}-{operation}-{uuid.uuid4().hex[:12]}"

    @staticmethod
    def _select_service(payload: dict[str, Any], preferred_kind: str) -> dict[str, Any]:
        services = [
            dict(item) for item in payload.get("services", [])
            if isinstance(item, dict)
            and item.get("status") == "active"
            and item.get("subscription_ready") is not False
            and int(item.get("node_count", 0) or 0) > 0
            and item.get("service_id")
        ]
        if not services:
            raise RoboHubApiError(503, {"code": "UPSTREAM_SERVICE_NOT_READY", "message": "没有已就绪套餐"})
        return min(
            services,
            key=lambda item: (
                str(item.get("kind") or "").lower() != preferred_kind.lower(),
                -int(item.get("node_count", 0) or 0),
                str(item.get("service_id")),
            ),
        )

    @staticmethod
    def _runnable_node_hashes(payload: dict[str, Any], requested: list[str]) -> list[str]:
        requested_set = set(requested)
        selected = []
        for item in payload.get("nodes", []):
            if not isinstance(item, dict):
                continue
            node_hash = str(item.get("node_hash") or "")
            if node_hash not in requested_set or item.get("runnable") is not True:
                continue
            if item.get("project_runnable") is False:
                continue
            selected.append(node_hash)
        return selected

    def _require_pool(self) -> dict[str, Any]:
        if not self.pool:
            raise RuntimeError("RoboHub pool is not active")
        return self.pool