"""开通服务：调魔方云 API 创建实例，轮询任务状态，写回 DB。

自动重试策略：
  - 最多尝试 3 次（第 1、2 次立即重试，第 3 次前等待 30 秒）
  - 配置类错误（无节点、无上游 ID 等）直接标记 FAILED 并退款
  - 重试耗尽后标记 PROVISION_FAILED（不退款），发通知邮件
"""
import asyncio
import hashlib
import hmac
import json
import logging
import secrets
import string
from dataclasses import asdict, is_dataclass
from datetime import datetime, timedelta, timezone
from typing import Any

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import settings as app_settings
from app.core.crypto import decrypt_secret, encrypt_secret
from app.models.cloud_server import CloudServer, CloudServerType
from app.models.instance import Instance, InstanceStatus
from app.models.order import Order
from app.models.product import Product, ProductApiType
from app.models.product_group import ProductGroup
from app.models.user import User
from app.services.dcim_cloud_client import DcimCloudClient, DcimCloudCredentials, DcimCloudError
from app.services import stock_compensation_service, system_setting_service, zjmf_instance_adapter
from app.services import provision_interface_service, provision_plugin_service
from app.services.zjmf_finance_client import ZjmfFinanceClient, ZjmfFinanceCredentials, ZjmfFinanceError
from app.utils.provision_params import (
    build_upstream_param_context,
    normalize_upstream_value,
    plugin_purchase_fields,
    sanitize_plugin_snapshot,
)
from mode.provision.base import ProvisionContext, ProvisionPluginError

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

_MAX_ATTEMPTS = 3
_RETRY_DELAYS = [0, 0, 30]  # 第 1、2 次立即重试，第 3 次前等 30 秒
_POLL_MAX_SECONDS = 120
_POLL_INTERVAL = 3
_LOCATION_SELECTOR_KEYS = {
    "node",
    "area",
    "node_group",
    "node_group_id",
    "ip_group",
    "ip_group_id",
    "ip",
    "ip_id",
}
_ZJMF_QUANTITY_OPTION_TYPES = {4, 7, 9, 11, 14, 15, 16, 17, 18, 19}
_ZJMF_CYCLE_MAP = {
    "hourly": "hour",
    "daily": "day",
    "monthly": "monthly",
    "quarterly": "quarterly",
    "semi_annually": "semiannually",
    "annually": "annually",
    "biennially": "biennially",
    "triennially": "triennially",
    "one_time": "onetime",
}
_ZJMF_PRICE_KEYS = {
    "hour": ("hour", "hsetupfee"),
    "day": ("day", "dsetupfee"),
    "ontrial": ("ontrial", "ontrialfee"),
    "monthly": ("monthly", "msetupfee"),
    "quarterly": ("quarterly", "qsetupfee"),
    "semiannually": ("semiannually", "ssetupfee"),
    "annually": ("annually", "asetupfee"),
    "biennially": ("biennially", "bsetupfee"),
    "triennially": ("triennially", "tsetupfee"),
    "fourly": ("fourly", "foursetupfee"),
    "fively": ("fively", "fivesetupfee"),
    "sixly": ("sixly", "sixsetupfee"),
    "sevenly": ("sevenly", "sevensetupfee"),
    "eightly": ("eightly", "eightsetupfee"),
    "ninely": ("ninely", "ninesetupfee"),
    "tenly": ("tenly", "tensetupfee"),
    "onetime": ("onetime", "osetupfee"),
}


class RetryableProvisionError(Exception):
    """可重试的开通异常（网络波动、超时等）。"""


class PendingProvisionError(Exception):
    """Upstream accepted the order but the host is still being provisioned."""


class NonRetryableProvisionError(Exception):
    """Clear upstream rejection that should fail and refund instead of retrying."""


def _is_non_retryable_zjmf_error(message: str) -> bool:
    text = str(message or "").lower()
    return any(
        marker in text
        for marker in (
            "库存不足",
            "stock",
            "产品不存在",
            "product not found",
            "已下架",
            "not available",
        )
    )


def _has_location_selector(params: dict) -> bool:
    return any(params.get(key) not in (None, "", 0, "0") for key in _LOCATION_SELECTOR_KEYS)


def _apply_location_selector_from_snapshot(params: dict, snap: dict) -> None:
    if _has_location_selector(params):
        return
    node = snap.get("product_group_node")
    area = snap.get("product_group_area")
    if node not in (None, "", 0, "0"):
        params["node"] = node
    elif area not in (None, "", 0, "0"):
        params["area"] = area


def _build_provision_params(instance: Instance, snap: dict, server: "CloudServer") -> dict:
    """把 provision_config 字段列表 + 购物车快照 → 魔方云创建参数。"""
    params: dict = {}
    provision_config: list = snap.get("provision_config") or []
    slider_values: dict = snap.get("slider_values") or {}

    for field in provision_config:
        def_raw: str = field.get("def", "")
        key = def_raw.split("|")[0] if "|" in def_raw else def_raw
        ftype = field.get("type", "fixed")

        if ftype == "fixed":
            val = field.get("value", "")
            if val not in (None, ""):
                params[key] = normalize_upstream_value(val)
        elif ftype == "slider":
            val = slider_values.get(key)
            if val not in (None, ""):
                params[key] = normalize_upstream_value(val)
        elif ftype == "input":
            if key == "hostname" and instance.hostname:
                params[key] = instance.hostname
            elif key == "rootpass":
                pass  # 让魔方随机生成，开通后从详情接口取回

    # os 必填：从 os_image 取（存的是"id|label"格式，魔方只需要 ID）
    if instance.os_image:
        os_raw = instance.os_image.split("|")[0]
        try:
            params["os"] = int(os_raw)
        except (ValueError, TypeError):
            params["os"] = os_raw

    # 魔方云客户ID（必填，否则报"请选择客户"）
    if server.client_id:
        params["client"] = server.client_id

    if "client_name" not in params or not params["client_name"]:
        params.pop("client_name", None)

    _apply_location_selector_from_snapshot(params, snap)
    return params


async def _apply_location_selector_from_product_group(params: dict, instance: Instance, db: AsyncSession) -> None:
    if _has_location_selector(params) or not instance.product_id:
        return
    result = await db.execute(
        select(ProductGroup.area, ProductGroup.node)
        .select_from(Product)
        .join(ProductGroup, ProductGroup.id == Product.group_id)
        .where(Product.id == instance.product_id)
    )
    row = result.first()
    if not row:
        return
    area, node = row
    if node not in (None, "", 0, "0"):
        params["node"] = node
    elif area not in (None, "", 0, "0"):
        params["area"] = area


def _enum_value(value: Any) -> str:
    return str(value.value if hasattr(value, "value") else value)


def _split_zjmf_name(value: Any) -> tuple[str, str]:
    text = str(value or "").strip()
    if "|" not in text:
        return text, text
    key, label = text.split("|", 1)
    return key.strip(), label.strip()


def _display_zjmf_label(value: Any) -> str:
    _, label = _split_zjmf_name(value)
    if "^" in label:
        parts = [part.strip() for part in label.split("^") if part.strip()]
        return parts[-1] if parts else label
    return label


def _config_field_key(field: dict[str, Any]) -> str:
    return str(field.get("def") or "").split("|", 1)[0].strip()


def _first_number(value: Any) -> int | None:
    import re

    if value in (None, ""):
        return None
    match = re.search(r"-?\d+(?:\.\d+)?", str(value))
    if not match:
        return None
    try:
        return int(float(match.group(0)))
    except ValueError:
        return None


def _int_or_none(value: Any) -> int | None:
    if value in (None, ""):
        return None
    try:
        return int(float(str(value).strip()))
    except (TypeError, ValueError):
        return None


def _json_object(value: Any) -> dict[str, Any]:
    if isinstance(value, dict):
        return value
    if not isinstance(value, str) or not value.strip():
        return {}
    try:
        parsed = json.loads(value)
    except (TypeError, ValueError):
        return {}
    return parsed if isinstance(parsed, dict) else {}


def _zjmf_product_from_config(raw_config: dict[str, Any]) -> dict[str, Any]:
    if isinstance(raw_config.get("products"), dict):
        return raw_config["products"]
    if isinstance(raw_config.get("product"), dict):
        return raw_config["product"]
    return {}


def _random_from_alphabet(alphabet: str, length: int) -> str:
    if length <= 0:
        return ""
    if not alphabet:
        alphabet = string.ascii_letters + string.digits
    return "".join(secrets.choice(alphabet) for _ in range(length))


def _zjmf_generated_hostname(raw_config: dict[str, Any], instance: Instance) -> str:
    current = str(instance.hostname or "").strip()
    if current:
        return current

    product = _zjmf_product_from_config(raw_config)
    host_rule = _json_object(product.get("host"))
    rule = _json_object(host_rule.get("rule"))

    if not rule:
        return "S" + _random_from_alphabet(string.ascii_letters + string.digits, 12)

    prefix = str(host_rule.get("prefix") or "cloud")
    if str(host_rule.get("show") or "0") in {"0", "", "False", "false"}:
        prefix = "ser"

    alphabet = ""
    if _zjmf_truthy(rule.get("upper")):
        alphabet += string.ascii_uppercase
    if _zjmf_truthy(rule.get("lower")):
        alphabet += string.ascii_lowercase
    if _zjmf_truthy(rule.get("num")):
        alphabet += string.digits
    if not alphabet:
        alphabet = string.ascii_letters + string.digits

    length = _int_or_none(rule.get("len_num")) or 12
    length = max(4, min(20, length))
    return (prefix + _random_from_alphabet(alphabet, length))[:64]


def _zjmf_generated_password(raw_config: dict[str, Any], instance: Instance) -> str:
    if instance.root_password:
        try:
            current = decrypt_secret(instance.root_password)
        except Exception:  # noqa: BLE001
            current = ""
        if current:
            return str(current)

    product = _zjmf_product_from_config(raw_config)
    password_rule = _json_object(product.get("password"))
    rule = _json_object(password_rule.get("rule"))

    length = _int_or_none(rule.get("len_num")) or 12
    length = max(6, min(20, length))
    groups: list[str] = []
    if _zjmf_truthy(rule.get("upper", 1)):
        groups.append(string.ascii_uppercase)
    if _zjmf_truthy(rule.get("lower", 1)):
        groups.append(string.ascii_lowercase)
    if _zjmf_truthy(rule.get("num", 1)):
        groups.append(string.digits)
    if _zjmf_truthy(rule.get("special", 0)):
        groups.append("!@#$%^&*")
    if not groups:
        groups = [string.ascii_uppercase, string.ascii_lowercase, string.digits]

    required = [secrets.choice(group) for group in groups]
    alphabet = "".join(groups)
    while len(required) < length:
        required.append(secrets.choice(alphabet))
    secrets.SystemRandom().shuffle(required)
    return "".join(required)


def _ensure_zjmf_login_fields(instance: Instance, raw_config: dict[str, Any]) -> tuple[str, str]:
    hostname = _zjmf_generated_hostname(raw_config, instance)
    password = _zjmf_generated_password(raw_config, instance)
    instance.hostname = hostname
    if password:
        instance.root_password = encrypt_secret(password)
    return hostname, password


def _norm(value: Any) -> str:
    return str(value or "").strip().lower()


def _is_zjmf_quantity_option(option_type: Any) -> bool:
    option_type_int = _int_or_none(option_type)
    return option_type_int in _ZJMF_QUANTITY_OPTION_TYPES


def _zjmf_option_key(option: dict[str, Any]) -> str:
    key, _ = _split_zjmf_name(option.get("option_name") or option.get("name"))
    return key


def _snapshot_config_values(instance: Instance, snap: dict) -> dict[str, Any]:
    values: dict[str, Any] = {}
    slider_values = snap.get("slider_values") or {}
    if isinstance(slider_values, dict):
        values.update({str(k): normalize_upstream_value(v) for k, v in slider_values.items() if v not in (None, "")})

    for field in snap.get("provision_config") or []:
        if not isinstance(field, dict):
            continue
        key = _config_field_key(field)
        if not key or key in values:
            continue
        field_type = field.get("type", "fixed")
        if field_type == "fixed":
            value = field.get("value")
        elif field_type == "slider":
            value = values.get(key)
        else:
            value = None
        if value not in (None, ""):
            values[key] = normalize_upstream_value(value)

    instance_values = {
        "cpu": instance.cpu,
        "memory": instance.memory,
        "system_disk_size": instance.system_disk_size,
        "data_disk_size": instance.data_disk_size,
        "in_bw": instance.in_bw,
        "out_bw": instance.out_bw,
        "ip_num": instance.ip_num,
        "nat_acl_limit": instance.nat_acl_limit,
        "traffic_quota": instance.traffic_quota,
        "traffic_type": instance.traffic_type,
    }
    for key, value in instance_values.items():
        if key not in values and value not in (None, ""):
            values[key] = value
    if "bw" not in values and instance.out_bw not in (None, ""):
        values["bw"] = instance.out_bw
    return values


def _candidate_config_values(key: str, values: dict[str, Any]) -> list[Any]:
    aliases = {
        "cpu": ("cpu", "vcpu"),
        "vcpu": ("vcpu", "cpu"),
        "memory": ("memory", "mem", "ram"),
        "mem": ("memory", "mem", "ram"),
        "ram": ("memory", "mem", "ram"),
        "bw": ("bw", "out_bw", "bandwidth"),
        "out_bw": ("out_bw", "bw", "bandwidth"),
        "in_bw": ("in_bw",),
        "system_disk": ("system_disk_size", "system_disk"),
        "system_disk_size": ("system_disk_size", "system_disk"),
        "data_disk": ("data_disk_size", "data_disk"),
        "data_disk_size": ("data_disk_size", "data_disk"),
        "ip_num": ("ip_num", "ipv4", "ip"),
        "nat_acl_limit": ("nat_acl_limit", "nat_acl"),
        "os": ("os",),
    }
    keys = aliases.get(key, (key,))
    result: list[Any] = []
    for item in keys:
        value = values.get(item)
        if value not in (None, ""):
            result.append(value)
    return result


def _selected_os(instance: Instance) -> tuple[str | None, str | None]:
    if not instance.os_image:
        return None, None
    if "|" not in instance.os_image:
        return instance.os_image.strip(), instance.os_image.strip()
    value, label = instance.os_image.split("|", 1)
    return value.strip() or None, label.strip() or value.strip() or None


def _option_subs(option: dict[str, Any]) -> list[dict[str, Any]]:
    subs = option.get("sub")
    if isinstance(subs, list):
        return [sub for sub in subs if isinstance(sub, dict)]
    subs = option.get("subs")
    if isinstance(subs, list):
        return [sub for sub in subs if isinstance(sub, dict)]
    return []


def _visible_subs(option: dict[str, Any]) -> list[dict[str, Any]]:
    subs = _option_subs(option)
    visible = [sub for sub in subs if str(sub.get("hidden") or "0") not in {"1", "true", "True"}]
    return visible or subs


def _sub_match_values(sub: dict[str, Any]) -> set[str]:
    raw_name = sub.get("option_name") or sub.get("name")
    key, label = _split_zjmf_name(raw_name)
    values = {
        _norm(sub.get("id")),
        _norm(raw_name),
        _norm(key),
        _norm(label),
        _norm(_display_zjmf_label(raw_name)),
    }
    return {value for value in values if value}


def _match_zjmf_sub_id(option: dict[str, Any], desired: Any) -> tuple[str | None, str | None]:
    subs = _visible_subs(option)
    if not subs:
        return None, None
    desired_text = _norm(desired)
    desired_number = _first_number(desired)

    if desired_text:
        for sub in subs:
            if desired_text in _sub_match_values(sub):
                return str(sub.get("id")), _display_zjmf_label(sub.get("option_name") or sub.get("name"))

    if desired_number is not None:
        for sub in subs:
            key, label = _split_zjmf_name(sub.get("option_name") or sub.get("name"))
            numbers = [
                _first_number(key),
                _first_number(label),
                _first_number(sub.get("qty_minimum")),
                _first_number(sub.get("qty_maximum")),
            ]
            if desired_number in numbers:
                return str(sub.get("id")), _display_zjmf_label(sub.get("option_name") or sub.get("name"))

    first = subs[0]
    return str(first.get("id")), _display_zjmf_label(first.get("option_name") or first.get("name"))


def _quantity_value_for_option(option: dict[str, Any], desired: Any) -> int | None:
    qty = _int_or_none(desired)
    if qty is None:
        qty = _first_number(desired)

    minimum = _int_or_none(option.get("qty_minimum"))
    maximum = _int_or_none(option.get("qty_maximum"))

    if qty is None:
        for sub in _visible_subs(option):
            qty = _int_or_none(sub.get("qty_minimum")) or _first_number(sub.get("option_name"))
            if qty is not None:
                break
    if qty is None:
        qty = minimum
    if qty is None:
        return None
    if minimum is not None and qty < minimum:
        qty = minimum
    if maximum is not None and maximum > 0 and qty > maximum:
        qty = maximum
    return qty


def _zjmf_config_options(raw_config: dict[str, Any], instance: Instance, snap: dict) -> tuple[dict[str, Any], dict[str, str]]:
    values = _snapshot_config_values(instance, snap)
    os_value, os_label = _selected_os(instance)
    configoptions: dict[str, Any] = {}
    os_payload: dict[str, str] = {}

    groups = raw_config.get("config_groups") if isinstance(raw_config.get("config_groups"), list) else []
    for group in groups:
        if not isinstance(group, dict):
            continue
        options = group.get("options") if isinstance(group.get("options"), list) else []
        for option in options:
            if not isinstance(option, dict):
                continue
            config_id = option.get("id")
            if config_id in (None, ""):
                continue
            option_key = _zjmf_option_key(option)
            option_type = option.get("option_type")
            desired: Any = None
            if option_key == "os":
                desired = os_value
            else:
                for candidate in _candidate_config_values(option_key, values):
                    desired = candidate
                    break

            if _is_zjmf_quantity_option(option_type):
                qty = _quantity_value_for_option(option, desired)
                if qty is not None:
                    configoptions[str(config_id)] = qty
                continue

            sub_id, sub_label = _match_zjmf_sub_id(option, desired)
            if not sub_id:
                continue
            configoptions[str(config_id)] = sub_id
            if option_key == "os":
                os_payload[str(sub_id)] = os_label or sub_label or str(sub_id)

    return configoptions, os_payload


def _zjmf_billing_cycle(cycle: str) -> str:
    return _ZJMF_CYCLE_MAP.get(str(cycle or "").lower(), str(cycle or "monthly"))


def _zjmf_cart_data(product: Product, instance: Instance, snap: dict, raw_config: dict[str, Any]) -> dict[str, Any]:
    upstream_pid = str(product.api_identifier or "").strip()
    configoptions, os_payload = _zjmf_config_options(raw_config, instance, snap)
    hostname, password = _ensure_zjmf_login_fields(instance, raw_config)
    return {
        "pid": upstream_pid,
        "billingcycle": _zjmf_billing_cycle(instance.cycle or snap.get("cycle") or "monthly"),
        "serverid": 0,
        "configoptions": configoptions,
        "customfield": {},
        "currencyid": 0,
        "qty": 1,
        "os": os_payload,
        "host": hostname,
        "password": password,
        "hostid": 0,
    }


def _money_or_none(value: Any) -> float | None:
    if value in (None, ""):
        return None
    try:
        number = float(value)
    except (TypeError, ValueError):
        return None
    if number < 0:
        return None
    return number


def _zjmf_truthy(value: Any) -> bool:
    if isinstance(value, bool):
        return value
    if isinstance(value, (int, float)):
        return value != 0
    if isinstance(value, str):
        return value.strip().lower() in {"1", "true", "yes", "on"}
    return False


def _zjmf_stock_state(raw_config: dict[str, Any], requested_qty: int) -> dict[str, Any]:
    product = _zjmf_product_from_config(raw_config)
    stock_qty = _int_or_none(product.get("qty"))
    stock_control = _zjmf_truthy(product.get("stock_control"))
    return {
        "pid": product.get("id"),
        "name": product.get("name"),
        "stock_control": product.get("stock_control"),
        "stock_control_enabled": stock_control,
        "qty": stock_qty,
        "requested_qty": requested_qty,
        "hidden": product.get("hidden"),
        "retired": product.get("retired"),
    }


def _ensure_zjmf_stock_available(raw_config: dict[str, Any], requested_qty: int) -> dict[str, Any]:
    state = _zjmf_stock_state(raw_config, requested_qty)
    stock_qty = state.get("qty")
    if state["stock_control_enabled"] and stock_qty is not None and stock_qty < requested_qty:
        name = state.get("name") or state.get("pid") or "unknown"
        raise NonRetryableProvisionError(
            f"魔方财务库存不足：{name} 上游库存 {stock_qty}，本次需要 {requested_qty}"
        )
    return state


def _zjmf_expected_pricing(raw_config: dict[str, Any], cart_data: dict[str, Any]) -> dict[str, Any]:
    product = _zjmf_product_from_config(raw_config)
    expected_pid = str(cart_data.get("pid") or "").strip()
    upstream_pid = str(product.get("id") or "").strip()
    if upstream_pid and expected_pid and upstream_pid != expected_pid:
        raise RetryableProvisionError(f"魔方财务产品校验失败：返回产品ID {upstream_pid} != 绑定产品ID {expected_pid}")

    billingcycle = str(cart_data.get("billingcycle") or "").strip()
    price_keys = _ZJMF_PRICE_KEYS.get(billingcycle)
    if not price_keys:
        raise RetryableProvisionError(f"魔方财务产品校验失败：不支持的计费周期 {billingcycle}")

    pricing_rows = raw_config.get("product_pricings") if isinstance(raw_config.get("product_pricings"), list) else []
    price_key, setup_key = price_keys
    for row in pricing_rows:
        if not isinstance(row, dict):
            continue
        price = _money_or_none(row.get(price_key))
        setup_fee = _money_or_none(row.get(setup_key))
        if price is None and setup_fee is None:
            continue
        return {
            "pid": expected_pid,
            "billingcycle": billingcycle,
            "price": price or 0,
            "setup_fee": setup_fee or 0,
            "currency": row.get("code"),
        }

    raise RetryableProvisionError(f"魔方财务产品校验失败：上游未返回 {billingcycle} 周期价格")


def _extract_zjmf_invoice_id(body: dict[str, Any] | None) -> int | None:
    if not isinstance(body, dict):
        return None
    data = body.get("data") if isinstance(body.get("data"), dict) else {}
    raw = data.get("invoiceid") or data.get("invoice_id") or body.get("invoiceid") or body.get("invoice_id")
    return _int_or_none(raw)


def _extract_zjmf_host_ids(body: dict[str, Any] | None) -> list[int]:
    if not isinstance(body, dict):
        return []
    data = body.get("data") if isinstance(body.get("data"), dict) else {}
    raw = data.get("hostid") or data.get("host_id") or body.get("hostid") or body.get("host_id")
    if raw in (None, ""):
        return []
    values = raw if isinstance(raw, list) else [raw]
    out: list[int] = []
    for value in values:
        parsed = _int_or_none(value)
        if parsed is not None:
            out.append(parsed)
    return out


def _stable_zjmf_downstream_token(instance: Instance) -> str:
    seed = f"zjmf:{instance.id}:{instance.order_id}:{instance.product_id}"
    return hmac.new(
        app_settings.JWT_SECRET_KEY.encode("utf-8"),
        seed.encode("utf-8"),
        hashlib.sha256,
    ).hexdigest()[:32]


def _normalize_zjmf_downstream_base_url(value: Any) -> str:
    url = str(value or "").strip().rstrip("/")
    if not url.startswith(("http://", "https://")):
        return "http://127.0.0.1"
    for suffix in ("/api/host/sync", "/api"):
        if url.lower().endswith(suffix):
            url = url[: -len(suffix)].rstrip("/")
    return url or "http://127.0.0.1"


async def _zjmf_downstream_payload(instance: Instance, db: AsyncSession) -> dict[str, Any]:
    if not instance.downstream_token:
        instance.downstream_token = _stable_zjmf_downstream_token(instance)
    settings = await system_setting_service.get_business_settings(db)
    downstream_url = _normalize_zjmf_downstream_base_url(
        getattr(settings, "backend_url", "")
        or getattr(app_settings, "PUBLIC_BACKEND_URL", "")
        or getattr(settings, "frontend_url", "")
        or ""
    )
    return {
        "is_api": 1,
        "downstream_url": downstream_url,
        "downstream_token": instance.downstream_token,
        "downstream_id": instance.id,
    }


def _merge_upstream_payload(instance: Instance, key: str, value: Any) -> None:
    payload = instance.upstream_payload if isinstance(instance.upstream_payload, dict) else {}
    payload[key] = value
    instance.upstream_payload = payload


async def _safe_send_instance_active_email(
    *,
    db: AsyncSession,
    user: User | None,
    instance: Instance,
    product_name: str,
) -> None:
    if not user or not user.email:
        return
    try:
        from app.services import email_service

        business_settings = await system_setting_service.get_business_settings(db)
        expires_str = instance.expires_at.strftime("%Y-%m-%d") if instance.expires_at else None
        await email_service.send_instance_active(
            to=user.email,
            username=user.username,
            product_name=product_name,
            main_ip=instance.main_ip or "",
            ssh_port=instance.ssh_port,
            hostname=instance.hostname,
            expires_at=expires_str,
            frontend_url=business_settings.frontend_url,
        )
    except Exception as exc:  # noqa: BLE001
        log.warning("instance active email skipped: instance=%s user=%s error=%s", instance.id, getattr(user, "id", None), exc)


async def _safe_send_instance_failed_email(
    *,
    db: AsyncSession,
    user: User | None,
    order: Order | None,
    product_name: str,
    reason: str,
) -> None:
    if not user or not user.email or not order:
        return
    try:
        from app.services import email_service

        business_settings = await system_setting_service.get_business_settings(db)
        await email_service.send_instance_failed(
            to=user.email,
            username=user.username,
            product_name=product_name,
            out_order_no=order.out_order_no,
            reason=reason,
            frontend_url=business_settings.frontend_url,
        )
    except Exception as exc:  # noqa: BLE001
        log.warning("instance failed email skipped: order=%s user=%s error=%s", getattr(order, "id", None), getattr(user, "id", None), exc)


def _timestamp_to_datetime(value: Any) -> datetime | None:
    raw = _int_or_none(value)
    if not raw or raw <= 0:
        return None
    try:
        return datetime.fromtimestamp(raw, tz=timezone.utc)
    except (OverflowError, OSError, ValueError):
        return None


def _apply_zjmf_host_detail(instance: Instance, detail: dict[str, Any] | None) -> None:
    if not isinstance(detail, dict):
        return
    host_data = detail.get("host_data") if isinstance(detail.get("host_data"), dict) else detail
    if not isinstance(host_data, dict):
        return

    main_ip = (
        host_data.get("dedicatedip")
        or host_data.get("mainip")
        or host_data.get("main_ip")
        or host_data.get("ip")
        or ""
    )
    assigned = host_data.get("assignedips")
    if not main_ip and isinstance(assigned, list) and assigned:
        main_ip = assigned[0]
    if main_ip:
        instance.main_ip = str(main_ip)

    port = _int_or_none(host_data.get("port"))
    if port is not None and 0 < port <= 65535:
        instance.ssh_port = port

    hostname = host_data.get("domain") or host_data.get("hostname")
    if hostname:
        instance.hostname = str(hostname)

    raw_password = host_data.get("password") or host_data.get("rootpassword")
    if raw_password:
        instance.root_password = encrypt_secret(str(raw_password))

    expires_at = _timestamp_to_datetime(host_data.get("nextduedate") or host_data.get("next_due_date"))
    if expires_at:
        instance.expires_at = expires_at

    zjmf_instance_adapter.sync_instance_from_host_detail(instance, detail)


def _zjmf_host_ready(detail: dict[str, Any] | None) -> bool:
    if not isinstance(detail, dict) or not detail:
        return False
    host_data = detail.get("host_data") if isinstance(detail.get("host_data"), dict) else detail
    status = str(
        detail.get("domainstatus")
        or detail.get("status")
        or host_data.get("domainstatus")
        or host_data.get("status")
        or ""
    ).strip().lower()
    if status == "active":
        return True
    return any(
        value not in (None, "", 0, "0")
        for value in (
            detail.get("mainip"),
            detail.get("main_ip"),
            host_data.get("dedicatedip"),
            host_data.get("mainip"),
            host_data.get("main_ip"),
            host_data.get("ip"),
        )
    )


async def _poll_zjmf_host_detail(client: ZjmfFinanceClient, host_id: int) -> dict[str, Any] | None:
    deadline = asyncio.get_event_loop().time() + _POLL_MAX_SECONDS
    last_detail: dict[str, Any] | None = None
    while asyncio.get_event_loop().time() < deadline:
        try:
            detail = await client.get_host_header(host_id)
        except ZjmfFinanceError as exc:
            log.warning("zjmf finance host header %s error: %s", host_id, exc.message)
            await asyncio.sleep(_POLL_INTERVAL)
            continue
        last_detail = detail
        host_data = detail.get("host_data") if isinstance(detail.get("host_data"), dict) else detail
        if isinstance(host_data, dict):
            status = str(host_data.get("domainstatus") or "").lower()
            if host_data.get("dedicatedip") or status == "active":
                try:
                    openapi_detail = await client.get_openapi_host(host_id)
                except ZjmfFinanceError:
                    openapi_detail = None
                return zjmf_instance_adapter.normalize_host_detail(detail, openapi_detail, None, None)
        await asyncio.sleep(_POLL_INTERVAL)
    if last_detail:
        try:
            openapi_detail = await client.get_openapi_host(host_id)
        except ZjmfFinanceError:
            openapi_detail = None
        return zjmf_instance_adapter.normalize_host_detail(last_detail, openapi_detail, None, None)
    return last_detail


async def _validate_zjmf_invoice_before_payment(
    client: ZjmfFinanceClient,
    invoice_id: int,
    instance: Instance,
) -> dict[str, Any]:
    try:
        invoice = await client.get_invoice_page(invoice_id)
    except ZjmfFinanceError as exc:
        raise RetryableProvisionError(f"魔方财务账单校验失败：{exc.message}") from exc

    item_data = invoice.get("item_data")
    if not isinstance(item_data, list) or not item_data:
        raise RetryableProvisionError("魔方财务账单校验失败：账单没有明细")

    invoice_amount = _money_or_none(invoice.get("invoice_total"))
    if invoice_amount is None:
        invoice_amount = _money_or_none(invoice.get("invoice_subtotal"))
    if invoice_amount is None:
        raise RetryableProvisionError("魔方财务账单校验失败：账单金额为空")

    local_amount = (instance.unit_price or 0) / 100
    if invoice_amount - local_amount > 0.01:
        raise RetryableProvisionError(
            f"魔方财务账单校验失败：上游账单 {invoice_amount:.2f} 大于本地实收 {local_amount:.2f}"
        )

    return invoice


async def _get_zjmf_client(server: CloudServer) -> ZjmfFinanceClient:
    creds = ZjmfFinanceCredentials(
        base_url=server.base_url,
        username=server.username,
        password=decrypt_secret(server.password_encrypted),
        account_type=server.account_type,
    )
    return ZjmfFinanceClient(creds, server_id=server.id, timeout=15.0)


async def _get_client(cloud_server_id: int, db: AsyncSession) -> tuple[DcimCloudClient, CloudServer]:
    res = await db.execute(select(CloudServer).where(CloudServer.id == cloud_server_id))
    server = res.scalar_one_or_none()
    if not server:
        raise RuntimeError(f"cloud_server {cloud_server_id} 不存在")
    creds = DcimCloudCredentials(
        base_url=server.base_url,
        username=server.username,
        password=decrypt_secret(server.password_encrypted),
        account_type=server.account_type,
    )
    client = DcimCloudClient(creds, server_id=server.id)
    return client, server


async def _poll_task(client: DcimCloudClient, task_id: int) -> dict | None:
    """轮询魔方任务状态，返回任务详情 dict 或 None（超时/失败）。

    魔方云任务 status 为数字：2=成功，progress=100 也表示完成。
    """
    deadline = asyncio.get_event_loop().time() + _POLL_MAX_SECONDS
    while asyncio.get_event_loop().time() < deadline:
        await asyncio.sleep(_POLL_INTERVAL)
        try:
            code, _, body = await client.request(f"/tasks/{task_id}", method="GET")
        except DcimCloudError as e:
            log.warning("poll task %s error: %s", task_id, e)
            continue
        if not (200 <= code < 300) or not body:
            continue
        task = body.get("data") or body
        status = task.get("status")
        progress = task.get("progress", 0)
        if isinstance(status, str):
            if status == "success":
                return task
            if status in ("failed", "error"):
                return None
        elif isinstance(status, int):
            if status == 2 or progress == 100:
                return task
            if status < 0:
                return None
    return None


async def _do_provision_attempt(
    instance: Instance,
    snap: dict,
    server: CloudServer,
    client: DcimCloudClient,
    instance_id: int,
    user: User | None,
    product_name: str,
    db: AsyncSession,
    order: Order | None = None,
) -> None:
    """执行单次开通尝试（写回 instance 字段，不 commit）。

    成功：设置 status=ACTIVE，写回 IP/密码等字段。
    失败：抛出 RetryableProvisionError。
    """
    from app.services import email_service, notification_service, spend_service

    params = _build_provision_params(instance, snap, server)
    await _apply_location_selector_from_product_group(params, instance, db)
    log.info("provision: instance %s params=%s", instance_id, params)

    # 调魔方创建接口
    try:
        code, text, body = await client.request("/clouds", data=params, method="POST")
    except DcimCloudError as e:
        raise RetryableProvisionError(f"调用魔方云失败：{e.message}") from e

    if not (200 <= code < 300) or not body:
        raise RetryableProvisionError(f"魔方云返回异常（HTTP {code}）：{text[:200]}")

    data = body.get("data") or body
    upstream_id = data.get("id")
    task_id = data.get("taskid")

    if not upstream_id:
        raise RetryableProvisionError(f"魔方云未返回实例 ID：{text[:200]}")

    instance.upstream_id = int(upstream_id)
    instance.upstream_task_id = int(task_id) if task_id else None
    await db.flush()

    # 轮询任务等待创建完成
    if task_id:
        task_data = await _poll_task(client, int(task_id))
        if task_data is None:
            raise RetryableProvisionError("魔方云实例创建超时或失败")

    # 拉取实例详情
    try:
        code, _, detail_body = await client.request(f"/clouds/{upstream_id}", method="GET")
    except DcimCloudError as e:
        raise RetryableProvisionError(f"获取实例详情失败：{e.message}") from e

    if not (200 <= code < 300) or not detail_body:
        raise RetryableProvisionError(f"获取实例详情异常（HTTP {code}）")

    detail = detail_body.get("data") or detail_body

    # 写回实例字段
    main_ip = detail.get("mainip") or ""
    if not main_ip:
        ip_list = detail.get("ip") or []
        if ip_list and isinstance(ip_list, list):
            main_ip = ip_list[0].get("ip", "") if isinstance(ip_list[0], dict) else str(ip_list[0])
    instance.main_ip = main_ip
    instance.ssh_port = detail.get("port")
    instance.hostname = detail.get("hostname") or instance.hostname

    raw_pass = detail.get("rootpassword") or ""
    if raw_pass:
        instance.root_password = encrypt_secret(raw_pass)

    cycle_days = _cycle_to_days(instance.cycle)
    instance.expires_at = datetime.now(timezone.utc) + timedelta(days=cycle_days)

    instance.status = InstanceStatus.ACTIVE
    instance.provision_log = f"开通成功 upstream_id={upstream_id}"

    log.info(
        "provision: instance %s active upstream_id=%s ip=%s",
        instance_id, upstream_id, instance.main_ip,
    )

    # ── 执行代理/会员结算 ──
    if order and order.settlement_status == "pending":
        try:
            from app.services import agent_service as _agent_svc
            await _agent_svc.settle_order(db, order)
            log.info("provision: settlement completed for order %s", order.id)
        except Exception as settle_err:
            log.error("provision: settlement failed for order %s: %s", order.id, settle_err)

    if order:
        try:
            from app.services import membership_service as _membership_svc
            await _membership_svc.mark_instance_point_rewards_claimable(
                db,
                instance_id=instance.id,
                order_id=order.id,
            )
            await _membership_svc.reward_referrer_for_order(db, order)
        except Exception as referral_err:
            log.error("provision: point/referral reward failed for order %s: %s", order.id, referral_err)

    # 发送开通成功邮件
    await _safe_send_instance_active_email(
        db=db,
        user=user,
        instance=instance,
        product_name=product_name,
    )

    # 站内信：开通成功（不依赖邮箱）
    await notification_service.safe_notify(
        db,
        user_id=instance.user_id,
        category="instance",
        title="服务器开通成功",
        content=f"您购买的「{product_name}」已开通成功，IP：{instance.main_ip or '-'}。",
        link=f"/control/server/{instance.id}",
    )


async def _do_zjmf_finance_provision_attempt(
    instance: Instance,
    snap: dict,
    server: CloudServer,
    client: ZjmfFinanceClient,
    instance_id: int,
    user: User | None,
    product: Product | None,
    product_name: str,
    db: AsyncSession,
    order: Order | None = None,
) -> None:
    from app.services import email_service, notification_service

    if not product:
        raise RetryableProvisionError("魔方财务开通失败：产品不存在")
    upstream_pid = str(product.api_identifier or "").strip()
    if not upstream_pid:
        raise RetryableProvisionError("魔方财务开通失败：产品未绑定上游产品 ID")
    if _enum_value(server.server_type) != CloudServerType.ZJMF_FINANCE.value:
        raise RetryableProvisionError("魔方财务开通失败：绑定的云接口不是魔方财务")

    try:
        raw_config = await client.get_product_config(upstream_pid)
    except ZjmfFinanceError as exc:
        raise RetryableProvisionError(f"魔方财务获取产品配置失败：{exc.message}") from exc

    cart_data = _zjmf_cart_data(product, instance, snap, raw_config)
    requested_qty = _int_or_none(cart_data.get("qty")) or 1
    stock_state = _zjmf_stock_state(raw_config, requested_qty)
    _merge_upstream_payload(instance, "zjmf_upstream_stock", stock_state)
    expected_pricing = _zjmf_expected_pricing(raw_config, cart_data)
    _ensure_zjmf_stock_available(raw_config, requested_qty)
    downstream = await _zjmf_downstream_payload(instance, db)
    _merge_upstream_payload(
        instance,
        "zjmf_cart",
        {
            "pid": cart_data.get("pid"),
            "billingcycle": cart_data.get("billingcycle"),
            "configoptions": cart_data.get("configoptions"),
            "os": cart_data.get("os"),
            "host": cart_data.get("host"),
            "password_provided": bool(cart_data.get("password")),
            "expected_pricing": expected_pricing,
            "upstream_stock": stock_state,
        },
    )
    log.info(
        "zjmf finance provision: instance=%s pid=%s cycle=%s host=%s password_provided=%s configoptions=%s",
        instance_id,
        cart_data.get("pid"),
        cart_data.get("billingcycle"),
        cart_data.get("host"),
        bool(cart_data.get("password")),
        cart_data.get("configoptions"),
    )

    payload = instance.upstream_payload if isinstance(instance.upstream_payload, dict) else {}
    credit_body = payload.get("zjmf_credit") if isinstance(payload.get("zjmf_credit"), dict) else None
    paid = int((credit_body or {}).get("status") or 0) == 1001
    host_ids = [int(instance.upstream_id)] if instance.upstream_id else []
    invoice_id = instance.upstream_invoice_id

    if not invoice_id and not host_ids:
        try:
            check_body = await client.check_downstream_order(downstream)
        except ZjmfFinanceError as exc:
            raise RetryableProvisionError(f"魔方财务检查重复订单失败：{exc.message}") from exc

        check_invoice_id = _extract_zjmf_invoice_id(check_body)
        check_host_ids = _extract_zjmf_host_ids(check_body)
        check_status = int(check_body.get("status") or 0)
        if check_status not in (200, 400):
            raise RetryableProvisionError(f"魔方财务检查重复订单异常：{check_body.get('msg') or check_status}")
        if check_status == 400 and not check_host_ids:
            raise RetryableProvisionError(f"魔方财务检查重复订单失败：{check_body.get('msg') or check_status}")
        if check_invoice_id or check_host_ids:
            _merge_upstream_payload(instance, "zjmf_downstream_check", check_body)
            invoice_id = check_invoice_id
            host_ids = check_host_ids
            if invoice_id:
                instance.upstream_invoice_id = invoice_id
            if host_ids:
                instance.upstream_id = host_ids[0]
            paid = check_status == 400 and bool(host_ids) and not invoice_id
            await db.flush()

    if not invoice_id and not host_ids:
        try:
            settle_body = await client.settle_cart(cart_data, downstream=downstream)
        except ZjmfFinanceError as exc:
            if _is_non_retryable_zjmf_error(exc.message):
                raise NonRetryableProvisionError(f"魔方财务创建订单失败：{exc.message}") from exc
            raise RetryableProvisionError(f"魔方财务创建订单失败：{exc.message}") from exc

        _merge_upstream_payload(instance, "zjmf_settle", settle_body)
        invoice_id = _extract_zjmf_invoice_id(settle_body)
        host_ids = _extract_zjmf_host_ids(settle_body)
        if invoice_id:
            instance.upstream_invoice_id = invoice_id
        if host_ids:
            instance.upstream_id = host_ids[0]
        paid = int(settle_body.get("status") or 0) == 1001
        await db.flush()

    if invoice_id and not paid:
        invoice_detail = await _validate_zjmf_invoice_before_payment(client, invoice_id, instance)
        _merge_upstream_payload(instance, "zjmf_invoice", invoice_detail)
        await db.flush()

        try:
            credit_body = await client.apply_credit(invoice_id, downstream=downstream)
        except ZjmfFinanceError as exc:
            raise RetryableProvisionError(f"魔方财务余额支付失败：{exc.message}") from exc

        _merge_upstream_payload(instance, "zjmf_credit", credit_body)
        host_ids = _extract_zjmf_host_ids(credit_body) or host_ids
        if host_ids:
            instance.upstream_id = host_ids[0]
        paid = int(credit_body.get("status") or 0) == 1001
        await db.flush()

    if not paid:
        raise RetryableProvisionError("魔方财务订单已创建，但上游账单未完成支付")

    if not host_ids and instance.upstream_id:
        host_ids = [int(instance.upstream_id)]
    if not host_ids:
        raise RetryableProvisionError("魔方财务支付成功，但未返回上游产品 ID")

    host_id = host_ids[0]
    instance.upstream_id = host_id
    detail = await _poll_zjmf_host_detail(client, host_id)
    _apply_zjmf_host_detail(instance, detail)
    if detail:
        host_data = detail.get("host_data") if isinstance(detail.get("host_data"), dict) else detail
        if isinstance(host_data, dict):
            sanitized = {k: v for k, v in host_data.items() if k not in {"password", "rootpassword"}}
            _merge_upstream_payload(instance, "zjmf_host", sanitized)

    if not _zjmf_host_ready(detail):
        instance.provision_log = f"魔方财务主机仍在开通中 zjmf_host_id={host_id}"
        await db.flush()
        raise PendingProvisionError(f"魔方财务主机仍在开通中：host_id={host_id}")

    if not instance.expires_at:
        cycle_days = _cycle_to_days(instance.cycle)
        instance.expires_at = datetime.now(timezone.utc) + timedelta(days=cycle_days)

    instance.status = InstanceStatus.ACTIVE
    instance.provision_log = f"开通成功 zjmf_host_id={host_id}"

    log.info(
        "zjmf finance provision: instance %s active host_id=%s ip=%s",
        instance_id,
        host_id,
        instance.main_ip,
    )

    if order and order.settlement_status == "pending":
        try:
            from app.services import agent_service as _agent_svc
            await _agent_svc.settle_order(db, order)
            log.info("provision: settlement completed for order %s", order.id)
        except Exception as settle_err:
            log.error("provision: settlement failed for order %s: %s", order.id, settle_err)

    if order:
        try:
            from app.services import membership_service as _membership_svc
            await _membership_svc.mark_instance_point_rewards_claimable(
                db,
                instance_id=instance.id,
                order_id=order.id,
            )
            await _membership_svc.reward_referrer_for_order(db, order)
        except Exception as referral_err:
            log.error("provision: point/referral reward failed for order %s: %s", order.id, referral_err)

    await _safe_send_instance_active_email(
        db=db,
        user=user,
        instance=instance,
        product_name=product_name,
    )

    await notification_service.safe_notify(
        db,
        user_id=instance.user_id,
        category="instance",
        title="服务器开通成功",
        content=f"您购买的「{product_name}」已开通成功，IP：{instance.main_ip or '-'}。",
        link=f"/control/server/{instance.id}",
    )


def _plugin_result_dict(result: Any) -> dict[str, Any]:
    if is_dataclass(result):
        result = asdict(result)
    if not isinstance(result, dict):
        raise RetryableProvisionError("插件返回值必须是 dict 或 ProvisionResult")
    return result


def _parse_plugin_datetime(value: Any) -> datetime | None:
    if isinstance(value, datetime):
        return value
    if isinstance(value, str) and value.strip():
        text = value.strip().replace("Z", "+00:00")
        try:
            return datetime.fromisoformat(text)
        except ValueError:
            return None
    return None


async def _do_plugin_provision_attempt(
    instance: Instance,
    snap: dict,
    instance_id: int,
    user: User | None,
    product: Product | None,
    product_name: str,
    db: AsyncSession,
    order: Order | None = None,
) -> None:
    from app.services import email_service, notification_service

    plugin_id = str((product.api_identifier if product else None) or snap.get("api_identifier") or "").strip()
    if not plugin_id:
        raise NonRetryableProvisionError("插件开通失败：产品未绑定开通插件")

    try:
        provider = provision_plugin_service.create_plugin_provider(plugin_id)
    except Exception as exc:
        raise NonRetryableProvisionError(f"插件开通失败：加载插件 {plugin_id} 失败：{exc}") from exc

    provision = getattr(provider, "provision", None)
    if not callable(provision):
        raise NonRetryableProvisionError(f"插件开通失败：插件 {plugin_id} 未实现 provision()")

    upstream_context = build_upstream_param_context(instance, snap)
    interface_config = await provision_interface_service.runtime_config_for_product(db, product)
    upstream_params = dict(upstream_context["params"])
    upstream_labels = dict(upstream_context["labels"])
    upstream_sources = dict(upstream_context["sources"])
    plugin_fields = plugin_purchase_fields(snap)
    context_snapshot = {
        **sanitize_plugin_snapshot(snap),
        "upstream_params": upstream_params,
        "upstream_param_labels": upstream_labels,
        "upstream_param_sources": upstream_sources,
        "interface_config": interface_config,
        "product_config_values": upstream_context["fields"],
        "plugin_fields": plugin_fields,
        "provision_interface_id": getattr(product, "provision_interface_id", None) if product else None,
    }

    context = ProvisionContext(
        instance=instance,
        snapshot=context_snapshot,
        product=product,
        order=order,
        user=user,
        db=db,
        settings=app_settings,
        upstream_params=upstream_params,
        interface_config=interface_config,
        plugin_fields=plugin_fields,
        product_config_values=upstream_context["fields"],
    )

    try:
        result = provision(context)
        if hasattr(result, "__await__"):
            result = await result
        data = _plugin_result_dict(result)
    except ProvisionPluginError as exc:
        if exc.retryable:
            raise RetryableProvisionError(str(exc)) from exc
        raise NonRetryableProvisionError(str(exc)) from exc
    except Exception as exc:
        raise RetryableProvisionError(f"插件开通调用失败：{exc}") from exc

    if data.get("success") is False:
        message = str(data.get("message") or "插件返回开通失败")
        if data.get("retryable") is False:
            raise NonRetryableProvisionError(message)
        raise RetryableProvisionError(message)

    raw_upstream_id = data.get("upstream_id")
    numeric_upstream_id = _int_or_none(raw_upstream_id)
    if numeric_upstream_id is not None:
        instance.upstream_id = numeric_upstream_id
    task_id = _int_or_none(data.get("task_id"))
    if task_id is not None:
        instance.upstream_task_id = task_id
    invoice_id = _int_or_none(data.get("invoice_id"))
    if invoice_id is not None:
        instance.upstream_invoice_id = invoice_id

    if data.get("hostname"):
        instance.hostname = str(data.get("hostname"))
    if data.get("main_ip"):
        instance.main_ip = str(data.get("main_ip"))
    ssh_port = _int_or_none(data.get("ssh_port"))
    if ssh_port is not None:
        instance.ssh_port = ssh_port
    if data.get("root_password"):
        instance.root_password = encrypt_secret(str(data.get("root_password")))

    expires_at = _parse_plugin_datetime(data.get("expires_at"))
    if expires_at:
        instance.expires_at = expires_at
    elif not instance.expires_at:
        instance.expires_at = datetime.now(timezone.utc) + timedelta(days=_cycle_to_days(instance.cycle))

    payload = data.get("payload") if isinstance(data.get("payload"), dict) else {}
    _merge_upstream_payload(
        instance,
        "provision_plugin",
        {
            "plugin_id": plugin_id,
            "upstream_id": raw_upstream_id,
            "task_id": data.get("task_id"),
            "invoice_id": data.get("invoice_id"),
            "links": data.get("links") if isinstance(data.get("links"), list) else payload.get("links"),
            "upstream_params": upstream_context["params"],
            "upstream_param_labels": upstream_context["labels"],
            "upstream_param_sources": upstream_context["sources"],
            "provision_interface_id": getattr(product, "provision_interface_id", None) if product else None,
            "provision_interface_keys": sorted(interface_config.keys()),
            "plugin_fields": plugin_fields,
            "payload": payload,
        },
    )
    instance.status = InstanceStatus.ACTIVE
    instance.provision_log = str(data.get("message") or f"插件 {plugin_id} 开通成功")

    if order and order.settlement_status == "pending":
        try:
            from app.services import agent_service as _agent_svc
            await _agent_svc.settle_order(db, order)
            log.info("plugin provision: settlement completed for order %s", order.id)
        except Exception as settle_err:
            log.error("plugin provision: settlement failed for order %s: %s", order.id, settle_err)

    if order:
        try:
            from app.services import membership_service as _membership_svc
            await _membership_svc.mark_instance_point_rewards_claimable(
                db,
                instance_id=instance.id,
                order_id=order.id,
            )
            await _membership_svc.reward_referrer_for_order(db, order)
        except Exception as referral_err:
            log.error("plugin provision: point/referral reward failed for order %s: %s", order.id, referral_err)

    await _safe_send_instance_active_email(
        db=db,
        user=user,
        instance=instance,
        product_name=product_name,
    )

    await notification_service.safe_notify(
        db,
        user_id=instance.user_id,
        category="instance",
        title="服务开通成功",
        content=f"您购买的「{product_name}」已通过插件开通成功。",
        link=f"/control/server/{instance.id}",
    )


async def provision_instance(
    instance_id: int,
    snap: dict,
    db: AsyncSession,
) -> None:
    """
    开通单个实例，带自动重试。
    snap 是 order.items_snapshot 中对应这个实例的那一条。
    """
    from app.models.balance_log import BalanceChangeType, BalanceLog
    from app.services import email_service, notification_service

    # 重新加载实例
    res = await db.execute(select(Instance).where(Instance.id == instance_id))
    instance = res.scalar_one_or_none()
    if not instance:
        log.error("provision: instance %s not found", instance_id)
        return

    if instance.status != InstanceStatus.PROVISIONING:
        log.info("provision: instance %s already %s, skip", instance_id, instance.status)
        return

    # 加载关联数据
    prod_res = await db.execute(select(Product).where(Product.id == instance.product_id))
    product = prod_res.scalar_one_or_none()

    order_res = await db.execute(select(Order).where(Order.id == instance.order_id))
    order = order_res.scalar_one_or_none()

    user_res = await db.execute(select(User).where(User.id == instance.user_id))
    user = user_res.scalar_one_or_none()

    product_name = product.name if product else f"产品#{instance.product_id}"

    instance.provision_attempts += 1
    instance.last_provision_at = datetime.now(timezone.utc)

    # auto_provision=false → 人工模式，直接停在 provisioning
    if product and not product.auto_provision:
        log.info("provision: instance %s product auto_provision=false, manual mode", instance_id)
        await db.flush()
        return

    # ── 非重试性错误：直接 _fail（退款） ──────────────────────────────────
    api_type = _enum_value(product.api_type if product else snap.get("api_type"))
    is_zjmf_finance = api_type == ProductApiType.ZJMF_FINANCE.value
    is_plugin = api_type == ProductApiType.PLUGIN.value

    if not is_plugin and not instance.cloud_server_id:
        await _fail(instance, order, user, product_name, db, "未绑定魔方云节点")
        return

    try:
        if is_plugin:
            client = None
            server = None
        elif is_zjmf_finance:
            server_res = await db.execute(select(CloudServer).where(CloudServer.id == instance.cloud_server_id))
            server = server_res.scalar_one_or_none()
            if not server:
                raise RuntimeError(f"cloud_server {instance.cloud_server_id} 不存在")
            client = await _get_zjmf_client(server)
        else:
            client, server = await _get_client(instance.cloud_server_id, db)
    except Exception as e:
        await _fail(instance, order, user, product_name, db, f"获取云节点失败：{e}")
        return

    # ── 重试循环 ──────────────────────────────────────────────────────────
    last_error = ""
    for attempt in range(_MAX_ATTEMPTS):
        if attempt > 0:
            delay = _RETRY_DELAYS[attempt] if attempt < len(_RETRY_DELAYS) else 0
            if delay > 0:
                log.info("provision: instance %s waiting %ss before retry %s/%s", instance_id, delay, attempt + 1, _MAX_ATTEMPTS)
                await asyncio.sleep(delay)
            log.info("provision: retrying instance %s attempt %s/%s", instance_id, attempt + 1, _MAX_ATTEMPTS)

        try:
            if is_plugin:
                await _do_plugin_provision_attempt(
                    instance,
                    snap,
                    instance_id,
                    user,
                    product,
                    product_name,
                    db,
                    order,
                )
            elif is_zjmf_finance:
                await _do_zjmf_finance_provision_attempt(
                    instance,
                    snap,
                    server,
                    client,
                    instance_id,
                    user,
                    product,
                    product_name,
                    db,
                    order,
                )
            else:
                await _do_provision_attempt(instance, snap, server, client, instance_id, user, product_name, db, order)
            # 成功 → 直接返回
            return
        except NonRetryableProvisionError as e:
            await _fail(instance, order, user, product_name, db, str(e))
            return
        except PendingProvisionError as e:
            log.info("provision: instance %s pending upstream completion: %s", instance_id, e)
            return
        except RetryableProvisionError as e:
            last_error = str(e)
            log.warning("provision: instance %s attempt %s/%s failed: %s", instance_id, attempt + 1, _MAX_ATTEMPTS, last_error)
            # 继续下一次重试

    # ── 所有重试耗尽：标记 PROVISION_FAILED（不退款） ──────────────────────
    log.error("provision: instance %s all %s attempts exhausted: %s", instance_id, _MAX_ATTEMPTS, last_error)
    instance.status = InstanceStatus.PROVISION_FAILED
    instance.provision_log = f"自动开通失败（已重试{_MAX_ATTEMPTS}次）：{last_error}"
    await db.flush()

    # 发开通失败通知邮件（不退款）
    await _safe_send_instance_failed_email(
        db=db,
        user=user,
        order=order,
        product_name=product_name,
        reason=last_error,
    )

    # 站内信：开通失败
    await notification_service.safe_notify(
        db,
        user_id=instance.user_id,
        category="instance",
        title="服务器开通失败",
        content=f"「{product_name}」自动开通失败：{last_error}。我们的客服会尽快介入处理。",
        link=f"/control/server/{instance.id}",
    )


async def _fail(
    instance: Instance,
    order: Order | None,
    user: User | None,
    product_name: str,
    db: AsyncSession,
    reason: str,
) -> None:
    """标记实例失败，退款，发邮件。"""
    from app.models.balance_log import BalanceChangeType, BalanceLog
    from app.services import email_service, spend_service

    log.error("provision failed: instance %s reason=%s", instance.id, reason)
    instance.status = InstanceStatus.FAILED
    instance.provision_log = reason
    await stock_compensation_service.restock_released_instances(
        db,
        [instance],
        source="provision_failed",
        reason=reason,
    )

    # 退款
    if user and instance.unit_price > 0:
        user_res = await db.execute(
            select(User).where(User.id == instance.user_id).with_for_update()
        )
        locked_user = user_res.scalar_one_or_none()
        if locked_user:
            refund_amount = instance.unit_price
            locked_user.balance += refund_amount
            refund_log = BalanceLog(
                user_id=locked_user.id,
                change_type=BalanceChangeType.ORDER_REFUND,
                delta=refund_amount,
                balance_after=locked_user.balance,
                ref_no=order.out_order_no if order else None,
                remark=f"开通失败退款：{reason[:100]}",
            )
            db.add(refund_log)
            await spend_service.record_spend_event(
                db,
                user_id=locked_user.id,
                event_type=spend_service.SpendEventType.SERVER_PURCHASE,
                amount_cents=-refund_amount,
                event_key=(
                    f"{order.out_order_no if order else 'no_order'}:"
                    f"instance_{instance.id}:server_purchase_refund"
                ),
                order_id=order.id if order else None,
                ref_no=order.out_order_no if order else None,
                remark=f"开通失败扣回累计消费：{reason[:100]}",
                payload_snapshot={"instance_id": instance.id},
            )
            from app.services import membership_service as _membership_svc
            await _membership_svc.reverse_instance_point_rewards(
                db,
                instance_id=instance.id,
                reason=f"开通失败退款：{reason[:100]}",
            )

    await db.flush()

    # 发失败邮件
    await _safe_send_instance_failed_email(
        db=db,
        user=user,
        order=order,
        product_name=product_name,
        reason=reason,
    )


def _cycle_to_days(cycle: str) -> int:
    mapping = {
        "hourly": 0,
        "daily": 1,
        "weekly": 7,
        "monthly": 30,
        "quarterly": 90,
        "semi_annually": 180,
        "annually": 365,
        "biennially": 730,
        "triennially": 1095,
    }
    return mapping.get(cycle.lower(), 30)
