from __future__ import annotations

import hashlib
import json
import random
import re
import string
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Any
from urllib.parse import urlencode

import aiohttp

from mode.provision.base import ManageResult, ProvisionContext, ProvisionPluginError, ProvisionResult


_CYCLE_DAYS = {
    "hourly": 1,
    "daily": 1,
    "monthly": 30,
    "quarterly": 90,
    "semi_annually": 180,
    "annually": 365,
    "biennially": 730,
    "triennially": 1095,
    "one_time": 3650,
}
_SECRET_KEYS = {"api_key", "mn_key", "bt_secret", "mn_keye", "password", "root_password", "new_password"}
_BLOCKED_PRODUCT_GROUP_KEYS = {"product_group_node", "product_group_area", "node", "area"}
# config_fields 中仅用于连接/测试的键，不应作为 upstream_param 传给开通
_CONFIG_ONLY_KEYS = {"username", "timeout", "api_timeout"}
_BLOCKED_UPSTREAM_KEYS = _BLOCKED_PRODUCT_GROUP_KEYS | _CONFIG_ONLY_KEYS


@dataclass(slots=True)
class MnbtConfig:
    base_url: str
    api_key: str
    bt_code: str
    bt_secret: str
    mn_vs: int = 15
    timeout: float = 20.0

    def endpoint(self, action: str) -> str:
        base = self.base_url.strip().rstrip("/")
        lower = base.lower()
        if lower.endswith("/api/api.php"):
            endpoint = base
        elif lower.endswith("/api"):
            endpoint = f"{base}/api.php"
        else:
            endpoint = f"{base}/api/api.php"
        return f"{endpoint}?{urlencode({'gn': action})}"

    def auth_payload(self, username: str) -> dict[str, str]:
        return {
            "mn_bh": self.bt_code,
            "mn_key": self.api_key,
            "mn_keye": self.bt_secret,
            "mn_vs": str(max(int(self.mn_vs or 15), 15)),
            "username": username,
        }


class MnbtClient:
    def __init__(self, config: MnbtConfig):
        self.config = config

    async def request(self, action: str, data: dict[str, Any]) -> dict[str, Any]:
        timeout = aiohttp.ClientTimeout(total=self.config.timeout)
        url = self.config.endpoint(action)
        try:
            async with aiohttp.ClientSession(timeout=timeout) as session:
                async with session.post(url, data=data, ssl=False) as response:
                    text = await response.text()
                    status = response.status
        except (TimeoutError, aiohttp.ClientError) as exc:
            raise ProvisionPluginError(f"梦奈宝塔 API 请求失败：{exc}", retryable=True) from exc

        try:
            payload = json.loads(text)
        except json.JSONDecodeError as exc:
            snippet = text[:200].replace("\n", " ").replace("\r", " ")
            raise ProvisionPluginError(f"梦奈宝塔 API 返回非 JSON：HTTP {status} {snippet}", retryable=status >= 500) from exc

        if not isinstance(payload, dict):
            raise ProvisionPluginError("梦奈宝塔 API 返回格式异常：根节点不是对象", retryable=status >= 500)
        if status >= 500:
            raise ProvisionPluginError(f"梦奈宝塔 API 服务端错误：HTTP {status}", retryable=True)
        if status >= 400:
            message = str(payload.get("msg") or payload.get("message") or f"HTTP {status}")
            raise ProvisionPluginError(f"梦奈宝塔 API 拒绝请求：{message}", retryable=False)

        try:
            code = int(payload.get("code", 0))
        except (TypeError, ValueError):
            code = 0
        if code != 200:
            message = str(payload.get("msg") or payload.get("message") or "上游返回失败")
            raise ProvisionPluginError(message, retryable=False)
        return payload


class MengnaiBtProvider:
    def __init__(self, config: Any):
        self.settings = config

    async def test(self, config: dict[str, Any] | None = None) -> dict[str, Any]:
        values = self._values_from_test_config(config or {})
        try:
            mnbt_config = self._mnbt_config(values)
            username = _normalize_username(_first(values, "username", default="qycwtest"), "qycw")
            payload = mnbt_config.auth_payload(username)
            response = await MnbtClient(mnbt_config).request("cfif", payload)
        except ProvisionPluginError as exc:
            return {
                "ok": False,
                "message": str(exc),
                "retryable": exc.retryable,
                "sent": _masked({"username": values.get("username"), **values}),
            }

        return {
            "ok": True,
            "message": str(response.get("msg") or "梦奈宝塔连接验证成功"),
            "endpoint": mnbt_config.endpoint("cfif"),
            "response": response,
            "sent": _masked(payload),
        }

    async def provision(self, context: ProvisionContext) -> ProvisionResult:
        values = self._values_from_context(context)
        mnbt_config = self._mnbt_config(values)
        client = MnbtClient(mnbt_config)

        username = self._new_username(context, values)
        password = str(_first(values, "password", "root_password", default="") or "").strip() or _random_password()
        if len(password) < 6:
            password = f"{password}{_random_password(6)}"[:12]
        dqtime, expires_at = self._expiry_value(context, values)
        product_type = str(_first(values, "type", "product_type", "mnbt_type", default="2")).strip() or "2"
        product_payload = {
            "type": product_type,
            "sizemax": _quota_value(values, ("sizemax", "traffic", "traffic_quota", "flow"), default="0"),
            "dqtime": dqtime,
            "webdx": _quota_value(values, ("webdx", "web_size", "websize", "space", "disk", "system_disk_size"), default="0"),
            "sqldx": _quota_value(values, ("sqldx", "sql_size", "sqlsize", "database_size"), default="0"),
            "ymbds": _quota_value(values, ("ymbds", "domain_limit", "domain_count"), default="1"),
        }
        request_payload = {
            **mnbt_config.auth_payload(username),
            "password": password,
            **product_payload,
        }

        response = await client.request("kt", request_payload)
        base_url = str(_first(values, "panel_url", "client_url", "login_url", default=mnbt_config.base_url) or mnbt_config.base_url).strip()
        panel_url = f"{base_url.rstrip('/')}/user/idcdl.php?gn=logine&username={username}&password={password}"
        links = [{"label": "本站实例", "url": f"/control/server/{context.instance.id}"}]
        if panel_url:
            links.insert(0, {"label": "梦奈宝塔面板", "url": panel_url})

        return ProvisionResult(
            success=True,
            upstream_id=username,
            hostname=username,
            root_password=password,
            expires_at=expires_at,
            links=links,
            payload={
                "plugin": "mengnai_bt",
                "mnbt_username": username,
                "mnbt_product_type": "CDN" if product_type == "1" else "主机",
                "mnbt_panel_url": panel_url,
                "mnbt_response": response,
                "request": _masked(request_payload),
                "product_params": product_payload,
                "plugin_fields": dict(context.plugin_fields or {}),
                "links": links,
            },
            message=str(response.get("msg") or "梦奈宝塔产品开通成功"),
            retryable=False,
        )

    async def manage(
        self,
        action: str,
        context: ProvisionContext,
        payload: dict[str, Any] | None = None,
    ) -> ManageResult:
        runtime = payload or {}
        values = self._values_from_context(context)
        mnbt_config = self._mnbt_config({**values, **_config_only(runtime)})
        username = self._existing_username(context, values, runtime)
        client = MnbtClient(mnbt_config)

        if action == "open-panel":
            # 一键登录链接：{mnbt_url}/user/idcdl.php?gn=logine&username=x&password=x
            base_url = str(_first(values, "panel_url", "client_url", "login_url", default=mnbt_config.base_url) or mnbt_config.base_url).strip()
            login_url = base_url
            raw_password = getattr(context.instance, "root_password", None)
            if raw_password:
                try:
                    from app.core.crypto import decrypt_secret
                    plain_password = decrypt_secret(str(raw_password))
                except Exception:
                    plain_password = None
                if plain_password:
                    login_url = f"{base_url.rstrip('/')}/user/idcdl.php?gn=logine&username={username}&password={plain_password}"
            return ManageResult(
                message="已生成梦奈宝塔面板链接",
                payload={"url": login_url, "mnbt_username": username},
                retryable=False,
            )

        auth = mnbt_config.auth_payload(username)
        if action == "sync":
            response = await client.request("cfif", auth)
            return ManageResult(
                message=str(response.get("msg") or "梦奈宝塔连接正常"),
                payload={"mnbt_username": username, "mnbt_response": response, "api_alive": True},
                retryable=False,
            )

        if action == "suspend":
            response = await client.request("zt", auth)
            return ManageResult(
                message=str(response.get("msg") or "梦奈宝塔主机暂停成功"),
                payload={"mnbt_username": username, "mnbt_response": response, "instance_status": "suspended"},
                retryable=False,
            )

        if action == "unsuspend":
            response = await client.request("jc", auth)
            return ManageResult(
                message=str(response.get("msg") or "梦奈宝塔主机恢复成功"),
                payload={"mnbt_username": username, "mnbt_response": response, "instance_status": "active"},
                retryable=False,
            )

        if action == "renew":
            setdate = str(_first(runtime, "setdate", default="") or "").strip()
            if not setdate:
                setdate, _ = self._expiry_value(context, values)
            response = await client.request("xf", {**auth, "setdate": setdate})
            return ManageResult(
                message=str(response.get("msg") or "梦奈宝塔到期时间同步成功"),
                payload={"mnbt_username": username, "mnbt_response": response, "setdate": setdate},
                retryable=False,
            )

        if action == "reset-password":
            new_password = str(_first(runtime, "password", "new_password", default="") or "").strip() or _random_password()
            if len(new_password) < 6:
                raise ProvisionPluginError("新密码长度不能小于 6 位", retryable=False)
            response = await client.request("czmm", {**auth, "password": new_password})
            return ManageResult(
                message=str(response.get("msg") or "梦奈宝塔密码重置成功"),
                payload={
                    "mnbt_username": username,
                    "mnbt_response": response,
                    "password_changed": True,
                    "root_password": new_password,
                },
                retryable=False,
            )

        if action == "resize":
            resize_payload = {
                "websize": str(_first(runtime, "websize", default="") or _quota_value(values, ("websize", "webdx", "web_size", "system_disk_size"), default="0")),
                "sqlsize": str(_first(runtime, "sqlsize", default="") or _quota_value(values, ("sqlsize", "sqldx", "sql_size"), default="0")),
                "ll": str(_first(runtime, "ll", default="") or _quota_value(values, ("ll", "sizemax", "traffic", "traffic_quota"), default="0")),
            }
            response = await client.request("zjmode", {**auth, **resize_payload})
            return ManageResult(
                message=str(response.get("msg") or "梦奈宝塔配额调整成功"),
                payload={"mnbt_username": username, "mnbt_response": response, **resize_payload},
                retryable=False,
            )

        if action == "terminate":
            response = await client.request("tz", auth)
            return ManageResult(
                message=str(response.get("msg") or "梦奈宝塔主机删除成功"),
                payload={"mnbt_username": username, "mnbt_response": response, "instance_status": "terminated"},
                retryable=False,
            )

        return ManageResult(success=False, message=f"未知梦奈宝塔管理动作：{action}", retryable=False)

    def _mnbt_config(self, values: dict[str, Any]) -> MnbtConfig:
        merged = {**self._settings_defaults(), **values}
        base_url = str(_first(merged, "mnbt_url", "api_url", "base_url", "url", default="") or "").strip()
        api_key = str(_first(merged, "api_key", "mn_key", default="") or "").strip()
        bt_code = str(_first(merged, "bt_code", "mn_bh", "btdh", default="") or "").strip()
        bt_secret = str(_first(merged, "bt_secret", "mn_keye", "bt_call_key", default="") or "").strip()
        if not bt_secret and _present(merged.get("ktmy")) and _present(merged.get("qmk")):
            bt_secret = hashlib.md5(f"{merged.get('ktmy')}{merged.get('qmk')}".encode("utf-8")).hexdigest()
        missing = []
        if not base_url:
            missing.append("mnbt_url")
        if not api_key:
            missing.append("api_key")
        if not bt_code:
            missing.append("bt_code")
        if not bt_secret:
            missing.append("bt_secret")
        if missing:
            raise ProvisionPluginError(f"缺少梦奈宝塔连接参数：{', '.join(missing)}", retryable=False)
        return MnbtConfig(
            base_url=base_url,
            api_key=api_key,
            bt_code=bt_code,
            bt_secret=bt_secret,
            mn_vs=max(_int(_first(merged, "mn_vs", default=15), 15), 15),
            timeout=float(_first(merged, "timeout", "api_timeout", default=20) or 20),
        )

    def _values_from_context(self, context: ProvisionContext) -> dict[str, Any]:
        values: dict[str, Any] = {}
        stored = _stored_plugin_record(context)
        if isinstance(stored.get("upstream_params"), dict):
            values.update(stored["upstream_params"])
        if isinstance(context.snapshot.get("upstream_params"), dict):
            values.update(context.snapshot["upstream_params"])
        values.update(context.upstream_params or {})
        if isinstance(stored.get("interface_config"), dict):
            values.update(stored["interface_config"])
        if isinstance(context.snapshot.get("interface_config"), dict):
            values.update(context.snapshot["interface_config"])
        values.update(context.interface_config or {})
        return {str(k): v for k, v in values.items() if str(k) not in _BLOCKED_UPSTREAM_KEYS}

    def _values_from_test_config(self, config: dict[str, Any]) -> dict[str, Any]:
        values: dict[str, Any] = {}
        for nested_key in ("config", "upstream_params", "params"):
            nested = config.get(nested_key)
            if isinstance(nested, dict):
                values.update(nested)
        values.update({k: v for k, v in config.items() if not isinstance(v, (dict, list))})
        return {str(k): v for k, v in values.items() if str(k) not in _BLOCKED_PRODUCT_GROUP_KEYS}

    def _settings_defaults(self) -> dict[str, Any]:
        defaults: dict[str, Any] = {}
        mapping = {
            "mnbt_url": ("MNBT_URL", "mnbt_url"),
            "api_key": ("MNBT_API_KEY", "mnbt_api_key"),
            "bt_code": ("MNBT_BT_CODE", "mnbt_bt_code"),
            "bt_secret": ("MNBT_BT_SECRET", "mnbt_bt_secret"),
            "mn_vs": ("MNBT_MN_VS", "mnbt_mn_vs"),
        }
        for key, attrs in mapping.items():
            for attr in attrs:
                value = getattr(self.settings, attr, None)
                if _present(value):
                    defaults[key] = value
                    break
        return defaults

    def _new_username(self, context: ProvisionContext, values: dict[str, Any]) -> str:
        fields = context.plugin_fields or {}
        prefix = str(_first(values, "username_prefix", "prefix", default="qy") or "qy")
        candidate = _first(fields, "username", "preferred_username", default=None)
        if not _present(candidate):
            candidate = _first(values, "username", default=None)
        if _present(candidate):
            return _normalize_username(candidate, prefix)

        order_no = str(getattr(context.order, "out_order_no", "") or "")
        user_id = str(getattr(context.user, "id", "") or "")
        seed = f"{prefix}{getattr(context.instance, 'id', '')}{order_no[-6:]}{user_id}"
        return _normalize_username(seed, prefix)

    def _existing_username(
        self,
        context: ProvisionContext,
        values: dict[str, Any],
        payload: dict[str, Any],
    ) -> str:
        stored = _stored_plugin_record(context)
        stored_payload = stored.get("payload") if isinstance(stored.get("payload"), dict) else {}
        prefix = str(_first(values, "username_prefix", "prefix", default="qy") or "qy")
        candidate = (
            _first(payload, "username", default=None)
            or _first(stored_payload, "mnbt_username", "username", default=None)
            or stored.get("upstream_id")
            or getattr(context.instance, "hostname", None)
            or _first(values, "username", default=None)
        )
        if not _present(candidate):
            candidate = self._new_username(context, values)
        return _normalize_username(candidate, prefix)

    def _expiry_value(self, context: ProvisionContext, values: dict[str, Any]) -> tuple[str, datetime | None]:
        configured = _first(values, "dqtime", "expire_date", "expires_at", default=None)
        if _present(configured):
            text = str(configured).strip()
            if text == "0":
                return "0", None
            parsed = _parse_date(text)
            return (parsed.strftime("%Y-%m-%d"), parsed) if parsed else (text, None)

        cycle = str(getattr(context.instance, "cycle", "") or "")
        if cycle == "one_time":
            return "0", None
        current = getattr(context.instance, "expires_at", None)
        if isinstance(current, datetime):
            return current.strftime("%Y-%m-%d"), current
        expires_at = datetime.now(timezone.utc) + timedelta(days=_CYCLE_DAYS.get(cycle, 30))
        return expires_at.strftime("%Y-%m-%d"), expires_at

    def _panel_url(self, values: dict[str, Any], config: MnbtConfig) -> str:
        return str(_first(values, "panel_url", "client_url", "login_url", default=config.base_url) or "").strip()


def _stored_plugin_record(context: ProvisionContext) -> dict[str, Any]:
    payload = getattr(context.instance, "upstream_payload", None)
    if not isinstance(payload, dict):
        return {}
    record = payload.get("provision_plugin")
    return record if isinstance(record, dict) else {}


def _config_only(values: dict[str, Any]) -> dict[str, Any]:
    allowed = {
        "mnbt_url",
        "api_url",
        "base_url",
        "url",
        "api_key",
        "mn_key",
        "bt_code",
        "mn_bh",
        "btdh",
        "bt_secret",
        "mn_keye",
        "bt_call_key",
        "mn_vs",
        "timeout",
        "api_timeout",
    }
    return {str(k): v for k, v in values.items() if str(k) in allowed}


def _first(values: dict[str, Any], *keys: str, default: Any = None) -> Any:
    for key in keys:
        value = values.get(key)
        if _present(value):
            return value
    return default


def _present(value: Any) -> bool:
    return value is not None and value != ""


def _int(value: Any, default: int) -> int:
    try:
        return int(float(str(value).strip()))
    except (TypeError, ValueError):
        return default


def _quota_value(values: dict[str, Any], keys: tuple[str, ...], default: str) -> str:
    value = _first(values, *keys, default=default)
    match = re.search(r"-?\d+(?:\.\d+)?", str(value))
    if not match:
        return str(default)
    number = float(match.group(0))
    return str(int(number)) if number.is_integer() else str(number)


def _parse_date(value: str) -> datetime | None:
    text = value.strip().replace("Z", "+00:00")
    for fmt in ("%Y-%m-%d", "%Y/%m/%d"):
        try:
            parsed = datetime.strptime(text, fmt)
            return parsed.replace(tzinfo=timezone.utc)
        except ValueError:
            pass
    try:
        parsed = datetime.fromisoformat(text)
    except ValueError:
        return None
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=timezone.utc)
    return parsed


def _normalize_username(value: Any, prefix: str) -> str:
    text = re.sub(r"[^a-z0-9]", "", str(value or "").lower())
    prefix_clean = re.sub(r"[^a-z0-9]", "", prefix.lower()) or "qy"
    if not text:
        text = f"{prefix_clean}{random.randint(100000, 999999)}"
    if not text[0].isalpha():
        text = f"{prefix_clean}{text}"
    if len(text) < 6:
        text = f"{text}{random.randint(100000, 999999)}"
    return text[:32]


def _random_password(length: int = 14) -> str:
    alphabet = string.ascii_letters + string.digits
    while True:
        value = "".join(random.choice(alphabet) for _ in range(length))
        if any(ch.islower() for ch in value) and any(ch.isupper() for ch in value) and any(ch.isdigit() for ch in value):
            return value


def _masked(values: dict[str, Any]) -> dict[str, Any]:
    masked: dict[str, Any] = {}
    for key, value in values.items():
        key_text = str(key)
        if key_text in _SECRET_KEYS or "secret" in key_text.lower() or "key" in key_text.lower():
            masked[key_text] = "***" if _present(value) else value
        else:
            masked[key_text] = value
    return masked


def create_provider(config: Any) -> MengnaiBtProvider:
    return MengnaiBtProvider(config)
