from __future__ import annotations

import asyncio
import logging
import time
from dataclasses import dataclass
from typing import Any
from urllib.parse import urljoin, urlparse

import aiohttp

from app.redis_client import k, redis_client

log = logging.getLogger("qycw.zjmf_finance")

TOKEN_TTL_SECONDS = 5400
_REDIRECT_CODES = {301, 302, 303, 307, 308}


def _flatten_form_data(data: dict[str, Any] | None) -> list[tuple[str, str]] | None:
    if data is None:
        return None

    pairs: list[tuple[str, str]] = []

    def add(prefix: str, value: Any) -> None:
        if isinstance(value, dict):
            for key, nested in value.items():
                add(f"{prefix}[{key}]", nested)
            return
        if isinstance(value, (list, tuple)):
            for nested in value:
                add(f"{prefix}[]", nested)
            return
        if isinstance(value, bool):
            pairs.append((prefix, "1" if value else "0"))
            return
        pairs.append((prefix, "" if value is None else str(value)))

    for key, value in data.items():
        add(str(key), value)
    return pairs


def token_cache_key(server_id: int) -> str:
    return k("zjmf_finance", "jwt", str(server_id))


@dataclass
class ZjmfFinanceCredentials:
    base_url: str
    username: str
    password: str
    account_type: str = "client"


class ZjmfFinanceError(Exception):
    def __init__(self, message: str, *, http_code: int | None = None, payload: Any = None):
        super().__init__(message)
        self.message = message
        self.http_code = http_code
        self.payload = payload


class ZjmfFinanceClient:
    def __init__(
        self,
        creds: ZjmfFinanceCredentials,
        *,
        server_id: int | None = None,
        timeout: float = 10.0,
    ):
        self.creds = creds
        self.server_id = server_id
        self.timeout = aiohttp.ClientTimeout(total=timeout)

    @property
    def is_resource(self) -> bool:
        return self.creds.account_type == "resource"

    def _build_url(self, path: str) -> str:
        path = path if path.startswith("/") else "/" + path
        return f"{self.creds.base_url.rstrip('/')}{path}"

    @staticmethod
    def _is_safe_redirect(current_url: str, next_url: str) -> bool:
        current = urlparse(current_url)
        target = urlparse(next_url)
        if not target.scheme or not target.hostname:
            return False
        if target.scheme not in {"http", "https"}:
            return False
        return current.hostname == target.hostname

    async def _request_once(
        self,
        session: aiohttp.ClientSession,
        url: str,
        method: str,
        data: dict[str, Any] | None,
        headers: dict[str, str],
    ) -> tuple[int, str, dict[str, str]]:
        if method.upper() == "GET":
            async with session.get(url, params=data, headers=headers, allow_redirects=False) as resp:
                return resp.status, await resp.text(), dict(resp.headers)
        async with session.request(
            method.upper(),
            url,
            data=_flatten_form_data(data),
            headers=headers,
            allow_redirects=False,
        ) as resp:
            return resp.status, await resp.text(), dict(resp.headers)

    async def _basecurl(
        self,
        session: aiohttp.ClientSession,
        path: str,
        data: dict[str, Any] | None = None,
        method: str = "POST",
        headers: dict[str, str] | None = None,
    ) -> tuple[int, str, dict[str, Any] | None]:
        url = self._build_url(path)
        h = {"User-Agent": "QYCW-ZJMF-Finance/1.0"}
        if headers:
            h.update(headers)

        try:
            text = ""
            code = 0
            for _ in range(3):
                code, text, response_headers = await self._request_once(session, url, method, data, h)
                if code not in _REDIRECT_CODES:
                    break
                location = response_headers.get("Location") or response_headers.get("location")
                if not location:
                    break
                next_url = urljoin(url, location)
                if not self._is_safe_redirect(url, next_url):
                    raise ZjmfFinanceError(f"Unsafe redirect blocked: {url} -> {next_url}", http_code=code)
                url = next_url
        except aiohttp.ClientConnectorError as exc:
            raise ZjmfFinanceError(f"Cannot connect to {self.creds.base_url}: {exc}") from exc
        except aiohttp.ServerTimeoutError as exc:
            raise ZjmfFinanceError(f"Connection to {self.creds.base_url} timed out") from exc
        except asyncio.TimeoutError as exc:
            raise ZjmfFinanceError(f"Connection to {self.creds.base_url} timed out") from exc
        except aiohttp.ClientError as exc:
            raise ZjmfFinanceError(f"HTTP error: {exc}") from exc

        body: dict[str, Any] | None = None
        stripped = text.strip()
        if stripped.startswith("{"):
            try:
                import json as _json

                parsed = _json.loads(stripped)
                if isinstance(parsed, dict):
                    body = parsed
            except ValueError:
                body = None
        return code, text, body

    async def login(self, *, force: bool = False) -> str:
        if self.server_id is not None and not force:
            cached = await redis_client.get(token_cache_key(self.server_id))
            if cached:
                return cached

        data: dict[str, Any] = {
            "username": self.creds.username,
            "password": self.creds.password,
        }
        login_paths = ["/zjmf_api_login"]
        if self.is_resource:
            data["type"] = "agent"
            login_paths = ["/resource_login", "/resource_login_supplier"]

        last_error = ""
        async with aiohttp.ClientSession(timeout=self.timeout) as session:
            for login_path in login_paths:
                code, text, body = await self._basecurl(session, login_path, data=data, method="POST")
                if code == 404 and login_path != login_paths[-1]:
                    last_error = f"Login endpoint {login_path} returned 404"
                    continue
                if not (200 <= code < 300) or not body:
                    msg = (body or {}).get("msg") or text[:200] or f"HTTP {code}"
                    raise ZjmfFinanceError(f"Login failed: {msg}", http_code=code, payload=body)
                if int(body.get("status") or 0) != 200:
                    msg = body.get("msg") or text[:200] or f"status={body.get('status')}"
                    raise ZjmfFinanceError(f"Login failed: {msg}", http_code=code, payload=body)
                jwt = str(body.get("jwt") or "").strip()
                if not jwt:
                    raise ZjmfFinanceError("Login failed: response does not include jwt", http_code=code, payload=body)
                if self.server_id is not None:
                    await redis_client.set(token_cache_key(self.server_id), jwt, ex=TOKEN_TTL_SECONDS)
                return jwt

        raise ZjmfFinanceError(last_error or "Login failed")

    async def invalidate_token(self) -> None:
        if self.server_id is not None:
            await redis_client.delete(token_cache_key(self.server_id))

    async def request(
        self,
        path: str,
        data: dict[str, Any] | None = None,
        *,
        method: str = "POST",
        timeout: float | None = None,
    ) -> tuple[int, str, dict[str, Any] | None]:
        cs_timeout = aiohttp.ClientTimeout(total=timeout) if timeout else self.timeout

        async def _do(jwt: str) -> tuple[int, str, dict[str, Any] | None]:
            async with aiohttp.ClientSession(timeout=cs_timeout) as session:
                return await self._basecurl(
                    session,
                    path,
                    data=data,
                    method=method,
                    headers={"Authorization": f"Bearer {jwt}"},
                )

        jwt = await self.login()
        code, text, body = await _do(jwt)
        if code == 401 or (isinstance(body, dict) and int(body.get("status") or 0) == 405):
            log.info("zjmf_finance %s refreshing jwt for %s", self.server_id, path)
            await self.invalidate_token()
            jwt = await self.login(force=True)
            code, text, body = await _do(jwt)
        return code, text, body

    async def get_user_info(self) -> dict[str, Any]:
        code, text, body = await self.request("/user_info", {}, method="GET")
        if not (200 <= code < 300) or not body:
            raise ZjmfFinanceError(f"User info failed: HTTP {code} {text[:200]}", http_code=code, payload=body)
        if int(body.get("status") or 0) != 200:
            raise ZjmfFinanceError(
                f"User info failed: {body.get('msg') or body.get('status')}",
                http_code=code,
                payload=body,
            )
        return body

    async def get_products(self) -> dict[str, Any]:
        path = "/resource/products" if self.is_resource else "/cart/all"
        code, text, body = await self.request(path, {}, method="GET", timeout=30.0)
        if not (200 <= code < 300) or not body:
            raise ZjmfFinanceError(f"Product list failed: HTTP {code} {text[:200]}", http_code=code, payload=body)
        if int(body.get("status") or 0) != 200:
            raise ZjmfFinanceError(
                f"Product list failed: {body.get('msg') or body.get('status')}",
                http_code=code,
                payload=body,
            )
        data = body.get("data")
        return data if isinstance(data, dict) else {}

    async def get_product_config(self, pid: int | str) -> dict[str, Any]:
        code, text, body = await self.request(
            "/cart/get_product_config",
            {"pid": str(pid)},
            method="GET",
            timeout=30.0,
        )
        if not (200 <= code < 300) or not body:
            raise ZjmfFinanceError(f"Product config failed: HTTP {code} {text[:200]}", http_code=code, payload=body)
        if int(body.get("status") or 0) != 200:
            raise ZjmfFinanceError(
                f"Product config failed: {body.get('msg') or body.get('status')}",
                http_code=code,
                payload=body,
            )
        data = body.get("data")
        return data if isinstance(data, dict) else {}

    async def settle_cart(
        self,
        cart_data: dict[str, Any],
        *,
        payment: str = "",
        downstream: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        payload: dict[str, Any] = {
            "payment": payment,
            "checkout": 1,
            "cart_data": cart_data,
        }
        if downstream:
            payload.update(downstream)
        code, text, body = await self.request("/cart/settle", payload, method="POST", timeout=45.0)
        if not (200 <= code < 300) or not body:
            raise ZjmfFinanceError(f"Settle failed: HTTP {code} {text[:200]}", http_code=code, payload=body)
        status = int(body.get("status") or 0)
        if status not in (200, 1001):
            raise ZjmfFinanceError(
                f"Settle failed: {body.get('msg') or body.get('status')}",
                http_code=code,
                payload=body,
            )
        return body

    async def apply_credit(
        self,
        invoice_id: int | str,
        *,
        downstream: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        payload: dict[str, Any] = {
            "invoiceid": str(invoice_id),
            "use_credit": 1,
            "enough": 1,
            "is_api": 1,
        }
        if downstream:
            payload.update(downstream)
        code, text, body = await self.request("/apply_credit", payload, method="POST", timeout=45.0)
        if not (200 <= code < 300) or not body:
            raise ZjmfFinanceError(f"Apply credit failed: HTTP {code} {text[:200]}", http_code=code, payload=body)
        status = int(body.get("status") or 0)
        if status not in (200, 1001):
            raise ZjmfFinanceError(
                f"Apply credit failed: {body.get('msg') or body.get('status')}",
                http_code=code,
                payload=body,
            )
        return body

    async def get_invoice_page(self, invoice_id: int | str) -> dict[str, Any]:
        code, text, body = await self.request(
            "/invoice_page",
            {"invoiceid": str(invoice_id)},
            method="POST",
            timeout=30.0,
        )
        if not (200 <= code < 300) or not body:
            raise ZjmfFinanceError(f"Invoice page failed: HTTP {code} {text[:200]}", http_code=code, payload=body)
        if int(body.get("status") or 0) != 200:
            raise ZjmfFinanceError(
                f"Invoice page failed: {body.get('msg') or body.get('status')}",
                http_code=code,
                payload=body,
            )
        data = body.get("data")
        return data if isinstance(data, dict) else {}

    async def check_downstream_order(self, downstream: dict[str, Any]) -> dict[str, Any]:
        payload: dict[str, Any] = {"is_api": 1}
        payload.update(downstream)
        code, text, body = await self.request("/cart/clear", payload, method="POST", timeout=30.0)
        if not (200 <= code < 300) or not body:
            raise ZjmfFinanceError(
                f"Downstream order check failed: HTTP {code} {text[:200]}",
                http_code=code,
                payload=body,
            )
        return body

    async def get_host_header(self, host_id: int | str, *, nat: bool = False) -> dict[str, Any]:
        payload: dict[str, Any] = {"host_id": str(host_id)}
        if nat:
            payload["nat"] = 1
        code, text, body = await self.request("/host/header", payload, method="GET", timeout=30.0)
        if not (200 <= code < 300) or not body:
            raise ZjmfFinanceError(f"Host header failed: HTTP {code} {text[:200]}", http_code=code, payload=body)
        if int(body.get("status") or 0) != 200:
            raise ZjmfFinanceError(
                f"Host header failed: {body.get('msg') or body.get('status')}",
                http_code=code,
                payload=body,
            )
        data = body.get("data")
        return data if isinstance(data, dict) else {}

    async def get_openapi_host(self, host_id: int | str) -> dict[str, Any]:
        code, text, body = await self.request(f"/v1/hosts/{host_id}", {}, method="GET", timeout=30.0)
        if not (200 <= code < 300) or not body:
            raise ZjmfFinanceError(f"Host detail failed: HTTP {code} {text[:200]}", http_code=code, payload=body)
        if int(body.get("status") or 0) != 200:
            raise ZjmfFinanceError(
                f"Host detail failed: {body.get('msg') or body.get('status')}",
                http_code=code,
                payload=body,
            )
        data = body.get("data")
        return data if isinstance(data, dict) else {}

    async def get_host_modules(self, host_id: int | str) -> dict[str, Any]:
        code, text, body = await self.request(f"/v1/hosts/{host_id}/module", {}, method="GET", timeout=30.0)
        if not (200 <= code < 300) or not body:
            raise ZjmfFinanceError(f"Host module failed: HTTP {code} {text[:200]}", http_code=code, payload=body)
        if int(body.get("status") or 0) != 200:
            raise ZjmfFinanceError(
                f"Host module failed: {body.get('msg') or body.get('status')}",
                http_code=code,
                payload=body,
            )
        data = body.get("data")
        return data if isinstance(data, dict) else {}

    async def request_host_module(
        self,
        host_id: int | str,
        action: str,
        data: dict[str, Any] | None = None,
        *,
        method: str = "PUT",
        timeout: float = 30.0,
    ) -> tuple[int, str, dict[str, Any] | None]:
        return await self.request(
            f"/v1/hosts/{host_id}/module/{action}",
            data or {},
            method=method,
            timeout=timeout,
        )

    async def request_host_custom(
        self,
        host_id: int | str,
        key: str,
        data: dict[str, Any] | None = None,
        *,
        method: str = "GET",
        timeout: float = 30.0,
    ) -> tuple[int, str, dict[str, Any] | None]:
        payload: dict[str, Any] = {"id": str(host_id), "key": key}
        if data:
            payload.update(data)
        return await self.request(
            f"/v1/hosts/{host_id}/module/custom",
            payload,
            method=method,
            timeout=timeout,
        )

    async def request_host_custom_func(
        self,
        host_id: int | str,
        func: str,
        data: dict[str, Any] | None = None,
        *,
        timeout: float = 30.0,
    ) -> tuple[int, str, dict[str, Any] | None]:
        payload: dict[str, Any] = {"func": func}
        if data:
            payload.update(data)
        return await self.request(
            f"/provision/custom/{host_id}",
            payload,
            method="POST",
            timeout=timeout,
        )

    async def request_provision_default(
        self,
        host_id: int | str,
        func: str,
        data: dict[str, Any] | None = None,
        *,
        timeout: float = 30.0,
    ) -> tuple[int, str, dict[str, Any] | None]:
        payload: dict[str, Any] = {
            "id": str(host_id),
            "func": func,
            "is_api": 1,
        }
        if data:
            payload.update(data)
        return await self.request(
            "/provision/default",
            payload,
            method="POST",
            timeout=timeout,
        )

    async def get_host_module_status(
        self,
        host_id: int | str,
        *,
        status_type: str = "host",
    ) -> tuple[int, str, dict[str, Any] | None]:
        return await self.request_host_module(
            host_id,
            "status",
            {"type": status_type},
            method="GET",
            timeout=15.0,
        )

    async def get_host_chart(
        self,
        host_id: int | str,
        chart_type: str,
        *,
        select: str | None = None,
        start: int | None = None,
        end: int | None = None,
        timeout: float = 20.0,
    ) -> tuple[int, str, dict[str, Any] | None]:
        params: dict[str, Any] = {"type": chart_type}
        if select:
            params["select"] = select
        if start is not None:
            params["start"] = start
        if end is not None:
            params["end"] = end
        return await self.request_host_module(
            host_id,
            "charts",
            params,
            method="GET",
            timeout=timeout,
        )

    async def get_host_reinstall_options(self, host_id: int | str) -> dict[str, Any]:
        code, text, body = await self.request_host_module(
            host_id,
            "reinstall",
            {},
            method="GET",
            timeout=30.0,
        )
        if not (200 <= code < 300) or not body:
            raise ZjmfFinanceError(f"Host reinstall options failed: HTTP {code} {text[:200]}", http_code=code, payload=body)
        if int(body.get("status") or 0) != 200:
            raise ZjmfFinanceError(
                f"Host reinstall options failed: {body.get('msg') or body.get('status')}",
                http_code=code,
                payload=body,
            )
        data = body.get("data")
        return data if isinstance(data, dict) else {}

    async def get_host_renew_page(
        self,
        host_id: int | str,
        *,
        billingcycle: str | None = None,
    ) -> dict[str, Any]:
        payload: dict[str, Any] = {}
        if billingcycle:
            payload["billingcycle"] = billingcycle
        code, text, body = await self.request(f"/v1/hosts/{host_id}/renew", payload, method="GET", timeout=30.0)
        if not (200 <= code < 300) or not body:
            raise ZjmfFinanceError(f"Host renew page failed: HTTP {code} {text[:200]}", http_code=code, payload=body)
        if int(body.get("status") or 0) != 200:
            raise ZjmfFinanceError(
                f"Host renew page failed: {body.get('msg') or body.get('status')}",
                http_code=code,
                payload=body,
            )
        data = body.get("data")
        return data if isinstance(data, dict) else {}

    async def renew_host(
        self,
        host_id: int | str,
        billingcycle: str,
    ) -> dict[str, Any]:
        code, text, body = await self.request(
            f"/v1/hosts/{host_id}/renew",
            {"billingcycle": billingcycle},
            method="POST",
            timeout=45.0,
        )
        if not (200 <= code < 300) or not body:
            raise ZjmfFinanceError(f"Host renew failed: HTTP {code} {text[:200]}", http_code=code, payload=body)
        status = int(body.get("status") or 0)
        if status not in (200, 1001):
            raise ZjmfFinanceError(
                f"Host renew failed: {body.get('msg') or body.get('status')}",
                http_code=code,
                payload=body,
            )
        return body

    async def set_host_auto_renew(
        self,
        host_id: int | str,
        enabled: bool,
    ) -> dict[str, Any]:
        code, text, body = await self.request(
            f"/v1/hosts/{host_id}/renew",
            {"initiative_renew": 1 if enabled else 0},
            method="PUT",
            timeout=30.0,
        )
        if not (200 <= code < 300) or not body:
            raise ZjmfFinanceError(f"Host auto renew failed: HTTP {code} {text[:200]}", http_code=code, payload=body)
        status = int(body.get("status") or 0)
        if status not in (200, 1001):
            raise ZjmfFinanceError(
                f"Host auto renew failed: {body.get('msg') or body.get('status')}",
                http_code=code,
                payload=body,
            )
        return body


async def test_connection(
    creds: ZjmfFinanceCredentials,
    *,
    server_id: int | None = None,
    force: bool = True,
) -> tuple[bool, str, int]:
    started = time.perf_counter()
    client = ZjmfFinanceClient(creds, server_id=server_id, timeout=8.0)
    try:
        if force:
            await client.invalidate_token()
        body = await client.get_user_info()
    except ZjmfFinanceError as exc:
        latency = int((time.perf_counter() - started) * 1000)
        if server_id is not None:
            await client.invalidate_token()
        return False, exc.message, latency
    except Exception as exc:  # noqa: BLE001
        latency = int((time.perf_counter() - started) * 1000)
        log.exception("zjmf_finance test_connection unexpected error")
        if server_id is not None:
            await client.invalidate_token()
        return False, f"Unknown error: {exc}", latency

    latency = int((time.perf_counter() - started) * 1000)
    user = body.get("user") if isinstance(body.get("user"), dict) else {}
    username = user.get("username") or user.get("email") or creds.username
    return True, f"Connection successful: {username}, {latency} ms", latency
