"""实例路由：列表、详情、操作（开机/关机/重启/重置密码）。"""
import asyncio
import logging
import random
import string
from dataclasses import asdict, is_dataclass
from datetime import datetime, timedelta, timezone
from decimal import Decimal, InvalidOperation, ROUND_CEILING, ROUND_HALF_UP
from typing import Any, Literal

from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel, Field
from sqlalchemy import func, select
from sqlalchemy.exc import ProgrammingError
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.crypto import decrypt_secret, encrypt_secret
from app.database import get_db
from app.models.cloud_server import CloudServer
from app.models.balance_log import BalanceChangeType, BalanceLog
from app.models.instance import Instance, InstanceStatus, SuspendType
from app.models.instance_disk_addon import InstanceDiskAddon, InstanceDiskAddonStatus
from app.models.order import Order, OrderStatus
from app.models.product import Product, ProductApiType
from app.models.product_group import ProductGroup
from app.models.upstream_action_log import UpstreamActionLog
from app.models.user import User
from app.core.deps import get_current_user
from app.schemas.common import ApiResponse
from app.services.dcim_cloud_client import DcimCloudClient, DcimCloudCredentials, DcimCloudError
from app.services.instance_real_data_service import get_cached_real_data, get_real_data_history
from app.services import agent_service, instance_lifecycle_service, instance_renewal_service, membership_service, notification_service, spend_service, upstream_action_service, zjmf_instance_adapter
from app.services import provision_interface_service, provision_plugin_service
from app.config import settings as app_settings
from mode.provision.base import ProvisionContext, ProvisionPluginError
from app.utils.specs import parse_memory_to_mb
from app.utils.provision_params import build_upstream_param_context, plugin_purchase_fields, sanitize_plugin_snapshot

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

router = APIRouter()

# 魔方云上游状态 → 前端展示标签
UPSTREAM_POWER_MAP: dict[str, str] = {
    "on": "运行中",
    "off": "已关机",
    "init": "创建中",
    "wait_reboot": "等待重启",
    "wait": "等待中",
    "suspend": "已暂停",
    "paused": "已挂起",
    "unknown": "未知",
}


class InstanceNatAclCreateBody(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    ext_port: int | None = Field(None, ge=0, le=65535)
    int_port: int = Field(..., ge=0, le=65535)
    protocol: Literal[1, 2, 3]


class PluginManageActionBody(BaseModel):
    payload: dict[str, Any] | None = None


async def _get_dcim_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 HTTPException(status_code=500, detail="云节点不存在")
    if zjmf_instance_adapter.is_zjmf_server(server):
        raise HTTPException(status_code=400, detail="魔方财务实例暂不支持该 DCIM 功能")
    creds = DcimCloudCredentials(
        base_url=server.base_url,
        username=server.username,
        password=decrypt_secret(server.password_encrypted),
        account_type=server.account_type,
    )
    return DcimCloudClient(creds, server_id=server.id), server


async def _reject_zjmf_unsupported(db: AsyncSession, instance: Instance, message: str | None = None) -> None:
    if await zjmf_instance_adapter.get_zjmf_server_for_instance(db, instance):
        raise HTTPException(
            status_code=400,
            detail=message or zjmf_instance_adapter.UNSUPPORTED_OPERATION_MESSAGE,
        )


def _resolve_os_label(os_image: str | None, group: ProductGroup | None) -> str | None:
    if not os_image:
        return None
    if "|" in os_image:
        return os_image.split("|", 1)[1] or os_image
    if group and isinstance(group.os_options, list):
        for opt in group.os_options:
            if not isinstance(opt, dict):
                continue
            if str(opt.get("value")) == str(os_image):
                label = opt.get("label")
                return str(label) if label else os_image
    return os_image


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


def _product_os_options(product: Product | None) -> list[dict[str, Any]]:
    if not product or not isinstance(product.provision_config, list):
        return []
    result: list[dict[str, Any]] = []
    seen: set[str] = set()
    for field in product.provision_config:
        if not isinstance(field, dict):
            continue
        key = str(field.get("def") or "").split("|", 1)[0].strip()
        if key != "os":
            continue
        candidates: list[Any] = []
        if isinstance(field.get("options"), list):
            candidates.extend(
                option.get("def") if isinstance(option, dict) else option
                for option in field.get("options") or []
            )
        if field.get("value") not in (None, ""):
            candidates.append(field.get("value"))
        for candidate in candidates:
            value, label = _split_option_value(candidate)
            if not value or value in seen:
                continue
            seen.add(value)
            result.append({"value": value, "label": label or value})
    return result


def _instance_os_options(product: Product | None, group: ProductGroup | None) -> list[dict[str, Any]]:
    if group and isinstance(group.os_options, list) and group.os_options:
        return group.os_options
    return _product_os_options(product)


def _find_os_option(product: Product | None, group: ProductGroup | None, os_id: int) -> dict[str, Any] | None:
    for opt in _instance_os_options(product, group):
        if not isinstance(opt, dict):
            continue
        raw_value = opt.get("value")
        if raw_value in (None, ""):
            continue
        try:
            value = int(float(str(raw_value).split("|", 1)[0].strip()))
        except (TypeError, ValueError):
            continue
        if value == os_id:
            return opt
    return None


def _instance_config_value(instance_value: Any, product: Product | None, key: str) -> Any:
    if instance_value not in (None, ""):
        return instance_value
    if product and key == "cpu" and product.vcpu is not None:
        return product.vcpu
    if product and key == "memory":
        memory_mb = parse_memory_to_mb(product.memory)
        if memory_mb is not None:
            return memory_mb
    if not product or not isinstance(product.provision_config, list):
        return None
    for field in product.provision_config:
        if not isinstance(field, dict):
            continue
        field_key = str(field.get("def", "")).split("|", 1)[0].strip()
        if field_key == key and field.get("type", "fixed") == "fixed":
            value = field.get("value")
            return value if value != "" else None
    return None


def _serialize_instance(i: Instance, product: Product | None = None, product_group_name: str | None = None, group: ProductGroup | None = None) -> dict[str, Any]:
    os_label = _resolve_os_label(i.os_image, group)
    provision_plugin = None
    plugin_manage_actions: list[dict[str, Any]] = []
    plugin_links: list[dict[str, Any]] = []
    if product and product.api_type == ProductApiType.PLUGIN.value and product.api_identifier:
        try:
            manifest = provision_plugin_service.get_plugin_manifest(str(product.api_identifier))
            provision_plugin = provision_plugin_service.public_manifest(manifest)
            plugin_manage_actions = [
                item for item in (manifest.get("manage_actions") or []) if isinstance(item, dict)
                and str(item.get("audience") or "both") in {"user", "both"}
            ]
            payload = i.upstream_payload if isinstance(i.upstream_payload, dict) else {}
            plugin_payload = payload.get("provision_plugin") if isinstance(payload.get("provision_plugin"), dict) else {}
            links = plugin_payload.get("links")
            if isinstance(links, list):
                plugin_links = [link for link in links if isinstance(link, dict)]
        except Exception:
            provision_plugin = None
    product_name = product.name if product else f"产品#{i.product_id}"
    return {
        "id": i.id,
        "status": i.status,
        "hostname": i.hostname,
        "main_ip": i.main_ip,
        "ssh_port": i.ssh_port if i.ssh_port and i.ssh_port > 0 else None,
        "os_image": i.os_image,
        "os_name": os_label,
        "expires_at": i.expires_at.isoformat() if i.expires_at else None,
        "created_at": i.created_at.isoformat() if i.created_at else None,
        "product_name": product_name,
        "product_group_name": product_group_name,
        "product_group_area": group.area if group else None,
        "product_group_node": group.node if group else None,
        "product_group_network_type": group.network_type if group else None,
        "os_options": _instance_os_options(product, group),
        "cpu": _instance_config_value(i.cpu, product, "cpu"),
        "memory": _instance_config_value(i.memory, product, "memory"),
        "upstream_id": i.upstream_id,
        "cycle": i.cycle,
        "unit_price": i.unit_price,
        "next_renew_amount": i.next_renew_amount,
        "auto_renew": i.auto_renew,
        "system_disk_size": _instance_config_value(i.system_disk_size, product, "system_disk_size"),
        "data_disk_size": _instance_config_value(i.data_disk_size, product, "data_disk_size"),
        "in_bw": _instance_config_value(i.in_bw, product, "in_bw"),
        "out_bw": _instance_config_value(i.out_bw, product, "out_bw"),
        "internet": i.internet,
        "nat_acl_limit": _instance_config_value(i.nat_acl_limit, product, "nat_acl_limit"),
        "nat_web_limit": _instance_config_value(None, product, "nat_web_limit"),
        "is_locked": i.is_locked,
        "suspend_type": i.suspend_type,
        "api_type": product.api_type if isinstance(product.api_type, str) else (product.api_type.value if product and product.api_type else None),
        "provision_config": product.provision_config if product and isinstance(product.provision_config, list) else [],
        "provision_plugin": provision_plugin,
        "plugin_manage_actions": plugin_manage_actions,
        "plugin_links": plugin_links,
        "security_group_id": None,
        "security_group_name": None,
    }


def _instance_base_renew_amount(instance: Instance) -> int:
    return int(instance.unit_price if instance.unit_price is not None else instance.next_renew_amount or 0)


def _effective_renew_amount(instance: Instance, disk_addon_total: int = 0) -> int:
    return _instance_base_renew_amount(instance) + int(disk_addon_total or 0)


async def _disk_addon_totals(db: AsyncSession, instance_ids: list[int]) -> dict[int, int]:
    if not instance_ids:
        return {}
    try:
        res = await db.execute(
            select(
                InstanceDiskAddon.instance_id,
                func.coalesce(func.sum(InstanceDiskAddon.recurring_amount), 0),
            )
            .where(
                InstanceDiskAddon.instance_id.in_(instance_ids),
                InstanceDiskAddon.status == InstanceDiskAddonStatus.ACTIVE,
            )
            .group_by(InstanceDiskAddon.instance_id)
        )
    except ProgrammingError as exc:
        if "instance_disk_addons" not in str(exc):
            raise
        await db.rollback()
        return {}
    return {int(instance_id): int(total or 0) for instance_id, total in res.all()}


async def _apply_effective_renew_amounts(
    db: AsyncSession,
    items: list[dict[str, Any]],
    instances: list[Instance],
) -> None:
    totals = await _disk_addon_totals(db, [instance.id for instance in instances])
    for item, instance in zip(items, instances):
        item["next_renew_amount"] = _effective_renew_amount(instance, totals.get(instance.id, 0))


async def _active_disk_addons(db: AsyncSession, instance_id: int, for_update: bool = False) -> list[InstanceDiskAddon]:
    stmt = (
        select(InstanceDiskAddon)
        .where(
            InstanceDiskAddon.instance_id == instance_id,
            InstanceDiskAddon.status == InstanceDiskAddonStatus.ACTIVE,
        )
        .order_by(InstanceDiskAddon.id.asc())
    )
    if for_update:
        stmt = stmt.with_for_update()
    res = await db.execute(stmt)
    return list(res.scalars().all())


async def _pending_disk_addons(db: AsyncSession, instance_id: int, for_update: bool = False) -> list[InstanceDiskAddon]:
    stmt = (
        select(InstanceDiskAddon)
        .where(
            InstanceDiskAddon.instance_id == instance_id,
            InstanceDiskAddon.status == InstanceDiskAddonStatus.PENDING,
        )
        .order_by(InstanceDiskAddon.id.asc())
    )
    if for_update:
        stmt = stmt.with_for_update()
    try:
        res = await db.execute(stmt)
    except ProgrammingError as exc:
        if "instance_disk_addons" not in str(exc):
            raise
        await db.rollback()
        return []
    return list(res.scalars().all())


async def _sync_pending_disk_addons_before_renewal(db: AsyncSession, instance: Instance) -> None:
    pending_addons = await _pending_disk_addons(db, instance.id, for_update=True)
    if not pending_addons:
        return

    if not instance.upstream_id or not instance.cloud_server_id:
        raise HTTPException(status_code=400, detail="存在未确认的磁盘任务，请等待任务完成后再续费")

    _, detail = await _fetch_upstream_instance_detail(instance, db, action_label="确认磁盘任务")
    await _sync_pending_disk_addons_from_detail(db, instance, detail)

    pending_after_sync = await _pending_disk_addons(db, instance.id, for_update=True)
    if pending_after_sync:
        raise HTTPException(status_code=400, detail="存在未完成的磁盘任务，请等待任务完成后再续费")


async def _sync_pending_disk_addons_from_detail(
    db: AsyncSession,
    instance: Instance,
    detail: dict[str, Any],
) -> bool:
    disks = detail.get("disk") if isinstance(detail, dict) else []
    if not isinstance(disks, list):
        return False

    by_id = {
        str(disk.get("id")): disk
        for disk in disks
        if isinstance(disk, dict) and disk.get("id") not in (None, "")
    }
    if not by_id:
        return False

    try:
        res = await db.execute(
            select(InstanceDiskAddon).where(
                InstanceDiskAddon.instance_id == instance.id,
                InstanceDiskAddon.status == InstanceDiskAddonStatus.PENDING,
                InstanceDiskAddon.upstream_disk_id.is_not(None),
            )
        )
    except ProgrammingError as exc:
        if "instance_disk_addons" not in str(exc):
            raise
        await db.rollback()
        return False
    changed = False
    now = datetime.now(timezone.utc)
    for addon in res.scalars().all():
        disk = by_id.get(str(addon.upstream_disk_id))
        if not disk:
            continue
        if addon.kind == "resize_disk":
            target_size = addon.target_size_gb or addon.size_gb
            current_size = _to_int_or_none(disk.get("size"))
            if current_size is None or current_size < target_size:
                continue
        addon.status = InstanceDiskAddonStatus.ACTIVE
        addon.activated_at = now
        addon.remark = None
        await membership_service.mark_instance_point_rewards_claimable(
            db,
            instance_id=instance.id,
            order_id=addon.order_id,
        )
        changed = True

    if changed:
        await db.flush()
    return changed


def _gen_renew_order_no() -> str:
    ts = datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S")
    suffix = "".join(random.choices(string.digits, k=6))
    return f"REN{ts}{suffix}"


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


def _cycle_to_timedelta(cycle: str) -> timedelta:
    if cycle.lower() == "hourly":
        return timedelta(hours=1)
    return timedelta(days=max(1, _cycle_to_days(cycle)))


async def _finalize_sales_rewards(
    db: AsyncSession,
    *,
    order: Order,
    user_id: int,
    paid_amount: int,
    instance_id: int,
    source_type: str,
    point_reward_status: str = "claimable",
) -> None:
    """Settle agent/referral rewards and create claimable self-purchase points."""
    settlements = await agent_service.calculate_order_settlement(db, order, user_id, paid_amount)
    for settlement in settlements:
        db.add(settlement)
    await db.flush()

    if settlements:
        await agent_service.settle_order(db, order)

    await membership_service.reward_referrer_for_order(db, order)
    await membership_service.create_self_purchase_point_reward(
        db,
        user_id=user_id,
        order_id=order.id,
        ref_no=order.out_order_no,
        amount_cents=paid_amount,
        source_type=source_type,
        instance_id=instance_id,
        status=point_reward_status,
        payload_snapshot={"instance_id": instance_id},
    )


async def _fetch_upstream_status(
    instance_id: int,
    upstream_id: int,
    cloud_server_id: int,
    db: AsyncSession,
) -> dict[str, Any] | None:
    """获取单个实例的上游实时状态，失败返回 None。"""
    try:
        zjmf_status = await zjmf_instance_adapter.get_upstream_status_for_ids(db, upstream_id, cloud_server_id)
        if zjmf_status is not None:
            return zjmf_status
        client, _ = await _get_dcim_client(cloud_server_id, db)
        code, _, body = await client.request(
            f"/clouds/{upstream_id}", method="GET", timeout=5.0
        )
        if not (200 <= code < 300) or not body:
            return None
        return body.get("data") or body
    except Exception:
        return None


def _upstream_power_label(upstream: dict | None) -> str | None:
    if not upstream:
        return None
    s = str(upstream.get("status", "")).lower()
    return UPSTREAM_POWER_MAP.get(s)


def _upstream_security_id(upstream: dict | None) -> str | None:
    if not upstream:
        return None
    value = upstream.get("security")
    if isinstance(value, dict):
        value = value.get("id")
    if value in (None, "", 0, "0"):
        return None
    return str(value)


def _upstream_security_name(upstream: dict | None) -> str | None:
    if not upstream:
        return None
    value = upstream.get("security_name")
    security = upstream.get("security")
    if not value and isinstance(security, dict):
        value = security.get("name")
    if isinstance(value, str):
        value = value.strip()
    return value or None


@router.get("", response_model=ApiResponse)
async def list_instances(
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    res = await db.execute(
        select(Instance, Product, ProductGroup)
        .outerjoin(Product, Product.id == Instance.product_id)
        .outerjoin(ProductGroup, ProductGroup.id == Product.group_id)
        .where(Instance.user_id == current_user.id)
        .order_by(Instance.id.desc())
    )
    rows = res.all()

    data: list[dict[str, Any]] = []
    # 收集需要上游查询的实例
    upstream_tasks: list[tuple[int, int, int]] = []  # (idx, upstream_id, cloud_server_id)

    for idx, (instance, product, group) in enumerate(rows):
        item = _serialize_instance(instance, product, group.name if group else None, group)
        data.append(item)
        if instance.upstream_id and instance.cloud_server_id:
            upstream_tasks.append((idx, instance.upstream_id, instance.cloud_server_id))

    await _apply_effective_renew_amounts(db, data, [row[0] for row in rows])

    if not upstream_tasks:
        return ApiResponse(data=data)

    # 并行拉取上游状态
    results = await asyncio.gather(
        *[
            _fetch_upstream_status(idx, uid, csid, db)
            for idx, uid, csid in upstream_tasks
        ],
        return_exceptions=True,
    )

    now = datetime.now(timezone.utc)
    updated_instances: list[Instance] = []

    for task_info, result in zip(upstream_tasks, results):
        idx, upstream_id, _ = task_info
        upstream = result if isinstance(result, dict) else None
        if upstream:
            data[idx]["upstream_power"] = _upstream_power_label(upstream)
            data[idx]["security_group_id"] = _upstream_security_id(upstream)
            data[idx]["security_group_name"] = _upstream_security_name(upstream)
            # 同步可能变化的字段回 DB
            instance = rows[idx][0]
            dirty = False
            new_ip = upstream.get("mainip") or ""
            if new_ip and new_ip != instance.main_ip:
                instance.main_ip = new_ip
                data[idx]["main_ip"] = new_ip
                dirty = True
            new_hostname = upstream.get("hostname") or ""
            if new_hostname and new_hostname != instance.hostname:
                instance.hostname = new_hostname
                data[idx]["hostname"] = new_hostname
                dirty = True
            new_port = upstream.get("port")
            if new_port and new_port != instance.ssh_port:
                instance.ssh_port = new_port
                data[idx]["ssh_port"] = new_port
                dirty = True
            if dirty:
                updated_instances.append(instance)
        else:
            data[idx]["upstream_power"] = None

    if updated_instances:
        await db.flush()

    return ApiResponse(data=data)


@router.get("/{instance_id}", response_model=ApiResponse)
async def get_instance(
    instance_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    res = await db.execute(
        select(Instance, Product, ProductGroup)
        .outerjoin(Product, Product.id == Instance.product_id)
        .outerjoin(ProductGroup, ProductGroup.id == Product.group_id)
        .where(Instance.id == instance_id, Instance.user_id == current_user.id)
    )
    row = res.one_or_none()
    if not row:
        raise HTTPException(status_code=404, detail="实例不存在")
    instance, product, group = row

    data = _serialize_instance(instance, product, group.name if group else None, group)
    await _apply_effective_renew_amounts(db, [data], [instance])

    # 插件型产品：解密 root_password 返回给前端
    if instance.root_password:
        try:
            data["root_password"] = decrypt_secret(instance.root_password)
        except Exception:
            pass
    point_rewards = await membership_service.get_unclaimed_instance_point_rewards(
        db,
        user_id=current_user.id,
        instance_id=instance.id,
    )
    data["point_rewards"] = [
        membership_service.point_reward_to_dict(reward)
        for reward in point_rewards
    ]

    # 如果有上游 ID，拉取魔方云实时详情
    if instance.upstream_id and instance.cloud_server_id:
        if await zjmf_instance_adapter.get_zjmf_server_for_instance(db, instance):
            try:
                upstream = await zjmf_instance_adapter.get_instance_detail(db, instance, sync_local=True)
                data["upstream"] = upstream
                data["upstream_power"] = _upstream_power_label(upstream)
                data["security_group_id"] = None
                data["security_group_name"] = None
                data["main_ip"] = upstream.get("mainip") or data["main_ip"]
                data["hostname"] = upstream.get("hostname") or data["hostname"]
                data["ssh_port"] = upstream.get("port") or data["ssh_port"]
                for key in (
                    "cpu",
                    "memory",
                    "system_disk_size",
                    "data_disk_size",
                    "in_bw",
                    "out_bw",
                    "nat_acl_limit",
                    "nat_web_limit",
                ):
                    if upstream.get(key) not in (None, ""):
                        data[key] = upstream.get(key)
                try:
                    reinstall_options = await zjmf_instance_adapter.get_reinstall_options(db, instance)
                    os_options = zjmf_instance_adapter.reinstall_options_to_os_options(reinstall_options)
                    if os_options:
                        data["os_options"] = os_options
                except Exception as option_exc:
                    log.info("get_instance %s fetch zjmf reinstall options skipped: %s", instance_id, option_exc)
                rootpass = upstream.get("rootpassword") or ""
                if rootpass:
                    data["root_password"] = rootpass
            except Exception as e:
                log.warning("get_instance %s fetch zjmf upstream failed: %s", instance_id, e)
                data["upstream_error"] = str(e)
            return ApiResponse(data=data)
        try:
            client, _ = await _get_dcim_client(instance.cloud_server_id, db)
            code, _, body = await client.request(
                f"/clouds/{instance.upstream_id}", method="GET"
            )
            if 200 <= code < 300 and body:
                upstream = body.get("data") or body
                data["upstream"] = upstream
                data["upstream_power"] = _upstream_power_label(upstream)
                data["security_group_id"] = _upstream_security_id(upstream)
                data["security_group_name"] = _upstream_security_name(upstream)
                # 同步部分字段回本地
                data["main_ip"] = upstream.get("mainip") or data["main_ip"]
                data["hostname"] = upstream.get("hostname") or data["hostname"]
                data["ssh_port"] = upstream.get("port") or data["ssh_port"]
                rootpass = upstream.get("rootpassword") or ""
                if rootpass and rootpass != data.get("root_password"):
                    data["root_password"] = rootpass
                if await _sync_pending_disk_addons_from_detail(db, instance, upstream):
                    await _apply_effective_renew_amounts(db, [data], [instance])
        except DcimCloudError as e:
            log.warning("get_instance %s fetch upstream failed: %s", instance_id, e)
            data["upstream_error"] = str(e)

    return ApiResponse(data=data)


@router.get("/{instance_id}/nat-acl", response_model=ApiResponse)
async def list_instance_nat_acl(
    instance_id: int,
    page: int = Query(1, ge=1),
    per_page: int = Query(20, ge=1, le=100),
    orderby: Literal["id", "name", "ext_port", "int_port", "protocol"] = Query("id"),
    sort: Literal["asc", "desc"] = Query("desc"),
    search: str | None = Query(None),
    list_type: Literal["all", "page"] = Query("page"),
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    res = await db.execute(
        select(Instance).where(
            Instance.id == instance_id,
            Instance.user_id == current_user.id,
        )
    )
    instance = res.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")

    data = await instance_lifecycle_service.list_instance_nat_acl(
        db,
        instance,
        page=page,
        per_page=per_page,
        orderby=orderby,
        sort=sort,
        search=search,
        list_type=list_type,
    )
    return ApiResponse(data=data)


@router.post("/{instance_id}/nat-acl", response_model=ApiResponse)
async def create_instance_nat_acl(
    instance_id: int,
    body: InstanceNatAclCreateBody,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    res = await db.execute(
        select(Instance).where(
            Instance.id == instance_id,
            Instance.user_id == current_user.id,
        )
    )
    instance = res.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    data = await instance_lifecycle_service.create_instance_nat_acl(
        db,
        instance,
        name=body.name,
        ext_port=body.ext_port,
        int_port=body.int_port,
        protocol=body.protocol,
    )
    return ApiResponse(data=data, message="NAT 转发已添加")


@router.delete("/{instance_id}/nat-acl/{nat_acl_id}", response_model=ApiResponse)
async def delete_instance_nat_acl(
    instance_id: int,
    nat_acl_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    res = await db.execute(
        select(Instance).where(
            Instance.id == instance_id,
            Instance.user_id == current_user.id,
        )
    )
    instance = res.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    data = await instance_lifecycle_service.delete_instance_nat_acl(
        db,
        instance,
        nat_acl_id=nat_acl_id,
    )
    return ApiResponse(data=data, message="NAT 转发已删除")


@router.post("/{instance_id}/points/claim", response_model=ApiResponse)
async def claim_instance_points(
    instance_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    res = await db.execute(
        select(Instance).where(Instance.id == instance_id, Instance.user_id == current_user.id)
    )
    instance = res.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")

    result = await membership_service.claim_instance_point_rewards(
        db,
        user_id=current_user.id,
        instance_id=instance.id,
    )
    if result["claimed_count"] <= 0:
        if result["pending_count"] > 0:
            raise HTTPException(status_code=400, detail="该服务器积分尚在待开通状态，暂不可领取")
        raise HTTPException(status_code=404, detail="该服务器暂无可领取积分")

    return ApiResponse(data=result, message="积分已领取")


# ── 操作 ──────────────────────────────────────────────────────────────────────


class PasswordResetBody(BaseModel):
    password: str = Field(..., min_length=6, max_length=64)


class ReinstallInstanceBody(BaseModel):
    os: int = Field(..., ge=1)
    password: str | None = Field(None, min_length=6, max_length=64)
    format_data_disk: int = Field(0, ge=0, le=1)


class RenewInstanceBody(BaseModel):
    cycles: int = Field(1, ge=1, le=36)


class AutoRenewBody(BaseModel):
    auto_renew: bool


class SuspendInstanceBody(BaseModel):
    type: Literal["traffic", "due"] = "due"
    free_gpu: int = Field(0, ge=0, le=1)


class DiskCreateBody(BaseModel):
    size: int = Field(..., ge=1, le=1048576)


class DiskResizeBody(BaseModel):
    size: int = Field(..., ge=1, le=1048576)


class SnapshotCreateBody(BaseModel):
    type: Literal["snap", "backup"]
    name: str | None = Field(None, max_length=128)
    remarks: str | None = Field(None, max_length=128)


class CronSnapshotBody(BaseModel):
    type: Literal["snap", "backup"]
    frequency: int = Field(..., ge=1, le=999)
    cycle: Literal["hour", "day", "week", "month"]
    disk: list[int] = Field(..., min_length=1)
    point: str | None = Field(None, pattern=r"^([01]\d|2[0-3]):[0-5]\d$")


DISK_DRIVER_VALUES = {"ide", "virtio", "sata", "scsi"}
DISK_CACHE_VALUES = {"none", "writethrough", "writeback", "directsync"}
DISK_IO_VALUES = {"native", "threads"}


def _provision_config_field(product: Product | None, *keys: str) -> dict[str, Any] | None:
    if not product or not isinstance(product.provision_config, list):
        return None
    key_set = set(keys)
    for field in product.provision_config:
        if not isinstance(field, dict):
            continue
        field_key = str(field.get("def", "")).split("|", 1)[0].strip()
        if field_key in key_set:
            return field
    return None


def _provision_config_value(product: Product | None, *keys: str) -> Any:
    field = _provision_config_field(product, *keys)
    if not field:
        return None
    value = field.get("value")
    return None if value in (None, "") else value


def _to_int_or_none(value: Any) -> int | None:
    if value in (None, ""):
        return None
    if isinstance(value, str):
        value = value.split("|", 1)[0].strip()
    try:
        return int(float(value))
    except (TypeError, ValueError):
        return None


def _to_text_or_none(value: Any) -> str | None:
    if value in (None, ""):
        return None
    return str(value).split("|", 1)[0].strip() or None


def _to_decimal_or_none(value: Any) -> Decimal | None:
    if value in (None, ""):
        return None
    try:
        return Decimal(str(value))
    except (InvalidOperation, TypeError, ValueError):
        return None


def _yuan_decimal_to_cents(value: Decimal) -> int:
    return int((value * Decimal("100")).quantize(Decimal("1"), rounding=ROUND_HALF_UP))


def _aware_datetime(value: datetime | None) -> datetime | None:
    if value is None:
        return None
    if value.tzinfo is None:
        return value.replace(tzinfo=timezone.utc)
    return value


def _system_disk_template(disks: Any) -> dict[str, Any]:
    if not isinstance(disks, list):
        return {}
    for disk in disks:
        if not isinstance(disk, dict):
            continue
        if str(disk.get("type") or "").lower() == "system" or str(disk.get("dev") or "").lower() == "vda":
            return disk
    for disk in disks:
        if isinstance(disk, dict):
            return disk
    return {}


def _resolve_disk_choice(
    configured: Any,
    template_value: Any,
    allowed: set[str],
    default: str,
    label: str,
) -> str:
    configured_text = _to_text_or_none(configured)
    text = configured_text or _to_text_or_none(template_value) or default
    if text in allowed:
        return text
    if configured_text:
        raise HTTPException(status_code=400, detail=f"产品磁盘配置 {label} 无效，请联系管理员")
    return default


def _validate_disk_size(size: int, product: Product | None) -> None:
    field = _provision_config_field(product, "data_disk_size", "disk_size")
    if not field or field.get("type") != "slider":
        return

    min_size = _to_int_or_none(field.get("slider_min")) or 1
    max_size = _to_int_or_none(field.get("slider_max")) or 1048576
    step = _to_int_or_none(field.get("slider_step")) or 1

    if size < min_size or size > max_size:
        raise HTTPException(status_code=400, detail=f"磁盘大小需在 {min_size} - {max_size} GB 范围内")
    if step > 0 and (size - min_size) % step != 0:
        raise HTTPException(status_code=400, detail=f"磁盘大小需按 {step} GB 递增")


def _disk_billing_config(product: Product | None) -> tuple[dict[str, Any], Decimal]:
    field = _provision_config_field(product, "data_disk_size", "disk_size")
    if not field or field.get("type") != "slider":
        raise HTTPException(status_code=400, detail="产品未配置弹性云盘计费规则，请联系管理员")

    unit_price = _to_decimal_or_none(field.get("slider_unit_price"))
    if unit_price is None or unit_price <= 0:
        raise HTTPException(status_code=400, detail="产品未配置弹性云盘单价，请联系管理员")
    return field, unit_price


def _calculate_disk_billing_for_size(
    target_size: int,
    billable_size: int,
    instance: Instance,
    product: Product | None,
    now: datetime | None = None,
) -> dict[str, Any]:
    _validate_disk_size(target_size, product)
    if billable_size <= 0:
        raise HTTPException(status_code=400, detail="磁盘容量增量异常")
    _, unit_price_yuan = _disk_billing_config(product)

    recurring_amount = _yuan_decimal_to_cents(unit_price_yuan * Decimal(billable_size))
    if recurring_amount <= 0:
        raise HTTPException(status_code=400, detail="弹性云盘价格异常，请联系管理员")

    current_time = now or datetime.now(timezone.utc)
    period = _cycle_to_timedelta(instance.cycle)
    period_seconds = Decimal(str(max(period.total_seconds(), 1)))
    expires_at = _aware_datetime(instance.expires_at)

    if expires_at and expires_at > current_time:
        remaining_seconds = Decimal(str((expires_at - current_time).total_seconds()))
        charged_until = expires_at
    else:
        remaining_seconds = period_seconds
        charged_until = current_time + period

    current_charge = int(
        (Decimal(recurring_amount) * remaining_seconds / period_seconds)
        .quantize(Decimal("1"), rounding=ROUND_CEILING)
    )
    current_charge = max(1, current_charge)
    remaining_days = int(
        (remaining_seconds / Decimal("86400")).quantize(Decimal("1"), rounding=ROUND_CEILING)
    )

    return {
        "size": target_size,
        "billable_size": billable_size,
        "cycle": instance.cycle,
        "recurring_amount": recurring_amount,
        "current_charge": current_charge,
        "remaining_days": max(1, remaining_days),
        "charged_until": charged_until,
        "expires_at": expires_at,
        "unit_price_cents": _yuan_decimal_to_cents(unit_price_yuan),
    }


def _calculate_disk_billing(
    body: DiskCreateBody,
    instance: Instance,
    product: Product | None,
    now: datetime | None = None,
) -> dict[str, Any]:
    return _calculate_disk_billing_for_size(body.size, body.size, instance, product, now)


def _disk_billing_snapshot(billing: dict[str, Any]) -> dict[str, Any]:
    return {
        "size": billing["size"],
        "billable_size": billing["billable_size"],
        "cycle": billing["cycle"],
        "recurring_amount": billing["recurring_amount"],
        "current_charge": billing["current_charge"],
        "remaining_days": billing["remaining_days"],
        "charged_until": billing["charged_until"].isoformat() if billing.get("charged_until") else None,
        "expires_at": billing["expires_at"].isoformat() if billing.get("expires_at") else None,
        "unit_price_cents": billing["unit_price_cents"],
    }


def _build_disk_create_payload(body: DiskCreateBody, product: Product | None, detail: dict[str, Any]) -> dict[str, Any]:
    _validate_disk_size(body.size, product)

    disks = detail.get("disk") if isinstance(detail, dict) else []
    system_disk = _system_disk_template(disks)

    store = _to_int_or_none(
        _provision_config_value(product, "data_disk_store", "disk_store", "store", "store_id")
    )
    if store is None:
        store = _to_int_or_none(system_disk.get("store_id") or system_disk.get("store"))
    if store is None:
        raise HTTPException(status_code=400, detail="无法确定磁盘存储，请联系管理员配置")

    return {
        "size": body.size,
        "store": store,
        "driver": _resolve_disk_choice(
            _provision_config_value(product, "data_disk_driver", "disk_driver", "driver"),
            system_disk.get("driver"),
            DISK_DRIVER_VALUES,
            "virtio",
            "driver",
        ),
        "cache": _resolve_disk_choice(
            _provision_config_value(product, "data_disk_cache", "disk_cache", "cache"),
            system_disk.get("cache"),
            DISK_CACHE_VALUES,
            "none",
            "cache",
        ),
        "io": _resolve_disk_choice(
            _provision_config_value(product, "data_disk_io", "disk_io", "io"),
            system_disk.get("io"),
            DISK_IO_VALUES,
            "native",
            "io",
        ),
        "iops_min": _to_int_or_none(
            _provision_config_value(product, "data_disk_iops_min", "disk_iops_min", "iops_min")
        ) or _to_int_or_none(system_disk.get("iops_min")) or 0,
        "iops_max": _to_int_or_none(
            _provision_config_value(product, "data_disk_iops_max", "disk_iops_max", "iops_max")
        ) or _to_int_or_none(system_disk.get("iops_max")) or 0,
    }


DISK_RESIZE_NUMERIC_FIELDS = [
    "read_bytes_sec",
    "write_bytes_sec",
    "read_iops_sec",
    "write_iops_sec",
    "read_bytes_sec_max",
    "write_bytes_sec_max",
    "read_iops_sec_max",
    "write_iops_sec_max",
    "iops_min",
    "iops_max",
]


def _is_system_disk(disk: dict[str, Any]) -> bool:
    return str(disk.get("type") or "").lower() == "system" or str(disk.get("dev") or "").lower() == "vda"


def _find_upstream_disk(detail: dict[str, Any], disk_id: int) -> dict[str, Any]:
    disks = detail.get("disk") if isinstance(detail, dict) else []
    if not isinstance(disks, list):
        raise HTTPException(status_code=400, detail="上游未返回磁盘信息")
    for disk in disks:
        if isinstance(disk, dict) and str(disk.get("id") or "") == str(disk_id):
            return disk
    raise HTTPException(status_code=404, detail="磁盘不存在")


def _disk_current_size(disk: dict[str, Any]) -> int:
    current_size = _to_int_or_none(disk.get("size"))
    if current_size is None or current_size <= 0:
        raise HTTPException(status_code=400, detail="上游磁盘容量异常")
    return current_size


def _validate_resize_disk(body: DiskResizeBody, product: Product | None, disk: dict[str, Any]) -> tuple[int, int]:
    if _is_system_disk(disk):
        raise HTTPException(status_code=400, detail="系统盘不支持扩容")
    current_size = _disk_current_size(disk)
    _validate_disk_size(body.size, product)
    if body.size <= current_size:
        raise HTTPException(status_code=400, detail="目标容量必须大于当前容量")
    return current_size, body.size - current_size


def _disk_status_key(disk: dict[str, Any]) -> str:
    value = disk.get("status")
    return "" if value is None else str(value)


SNAPSHOT_LIMIT_KEYS: dict[str, tuple[str, ...]] = {
    "snap": (
        "snap_limit",
        "snapshot_limit",
        "snap_num_limit",
        "snapshot_num_limit",
        "max_snap",
        "max_snapshot",
        "max_snap_num",
        "max_snapshot_num",
    ),
    "backup": (
        "backup_limit",
        "backup_num_limit",
        "max_backup",
        "max_backup_num",
    ),
}

SNAPSHOT_COUNT_KEYS: dict[str, tuple[str, ...]] = {
    "snap": ("snap_num", "snapshot_num", "snap_count", "snapshot_count"),
    "backup": ("backup_num", "backup_count"),
}


def _snapshot_limit_from_detail(detail: dict[str, Any], snapshot_type: str) -> int | None:
    keys = SNAPSHOT_LIMIT_KEYS.get(snapshot_type, ())
    candidates: list[dict[str, Any]] = [detail]
    for nested_key in ("limits", "limit", "quota", "quotas", "snapshot", "snapshots", "backup"):
        nested = detail.get(nested_key)
        if isinstance(nested, dict):
            candidates.append(nested)

    for candidate in candidates:
        for key in keys:
            value = _to_int_or_none(candidate.get(key))
            if value is not None:
                return value
    return None


def _snapshot_count_from_disk(disk: dict[str, Any], snapshot_type: str) -> int | None:
    for key in SNAPSHOT_COUNT_KEYS.get(snapshot_type, ()):
        value = _to_int_or_none(disk.get(key))
        if value is not None:
            return value
    return None


def _build_disk_resize_payload(body: DiskResizeBody, disk: dict[str, Any]) -> dict[str, Any]:
    payload: dict[str, Any] = {
        "size": body.size,
        "driver": _resolve_disk_choice(None, disk.get("driver"), DISK_DRIVER_VALUES, "virtio", "driver"),
        "cache": _resolve_disk_choice(None, disk.get("cache"), DISK_CACHE_VALUES, "none", "cache"),
        "io": _resolve_disk_choice(None, disk.get("io"), DISK_IO_VALUES, "native", "io"),
    }

    for field in DISK_RESIZE_NUMERIC_FIELDS:
        value = _to_int_or_none(disk.get(field))
        if value is not None:
            payload[field] = value

    remarks = disk.get("remarks")
    if remarks not in (None, ""):
        payload["remarks"] = str(remarks)

    return payload


def _build_disk_mount_payload(
    instance: Instance,
    product: Product | None,
    detail: dict[str, Any],
    disk: dict[str, Any],
) -> dict[str, Any]:
    disks = detail.get("disk") if isinstance(detail, dict) else []
    system_disk = _system_disk_template(disks)
    store_id = (
        _to_int_or_none(disk.get("store_id"))
        or _to_int_or_none(disk.get("store"))
        or _to_int_or_none(_provision_config_value(product, "data_disk_store", "disk_store", "store", "store_id"))
        or _to_int_or_none(system_disk.get("store_id") or system_disk.get("store"))
    )

    payload: dict[str, Any] = {
        "cloud": instance.upstream_id,
        "driver": _resolve_disk_choice(
            _provision_config_value(product, "data_disk_driver", "disk_driver", "driver"),
            disk.get("driver") or system_disk.get("driver"),
            DISK_DRIVER_VALUES,
            "virtio",
            "driver",
        ),
        "cache": _resolve_disk_choice(
            _provision_config_value(product, "data_disk_cache", "disk_cache", "cache"),
            disk.get("cache") or system_disk.get("cache"),
            DISK_CACHE_VALUES,
            "none",
            "cache",
        ),
        "io": _resolve_disk_choice(
            _provision_config_value(product, "data_disk_io", "disk_io", "io"),
            disk.get("io") or system_disk.get("io"),
            DISK_IO_VALUES,
            "native",
            "io",
        ),
    }
    if store_id is not None:
        payload["store_id"] = store_id
    return payload


def _disk_ids_from_upstream(disks: Any) -> set[str]:
    if not isinstance(disks, list):
        return set()
    ids: set[str] = set()
    for disk in disks:
        if isinstance(disk, dict) and disk.get("id") not in (None, ""):
            ids.add(str(disk.get("id")))
    return ids


async def _pending_missing_disk_addon_count(
    db: AsyncSession,
    instance_id: int,
    upstream_disk_ids: set[str],
) -> int:
    res = await db.execute(
        select(InstanceDiskAddon).where(
            InstanceDiskAddon.instance_id == instance_id,
            InstanceDiskAddon.status.in_([
                InstanceDiskAddonStatus.PENDING,
                InstanceDiskAddonStatus.ACTIVE,
            ]),
        )
    )
    count = 0
    for addon in res.scalars().all():
        if addon.upstream_disk_id is None or str(addon.upstream_disk_id) not in upstream_disk_ids:
            count += 1
    return count


def _gen_disk_order_no() -> str:
    ts = datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S")
    suffix = "".join(random.choices(string.digits, k=6))
    return f"DSK{ts}{suffix}"


async def _mark_disk_addons_deleted(db: AsyncSession, instance_id: int, disk_id: int) -> int:
    try:
        res = await db.execute(
            select(InstanceDiskAddon)
            .where(
                InstanceDiskAddon.instance_id == instance_id,
                InstanceDiskAddon.upstream_disk_id == disk_id,
                InstanceDiskAddon.status.in_([
                    InstanceDiskAddonStatus.PENDING,
                    InstanceDiskAddonStatus.ACTIVE,
                ]),
            )
            .with_for_update()
        )
    except ProgrammingError as exc:
        if "instance_disk_addons" not in str(exc):
            raise
        await db.rollback()
        return 0

    now = datetime.now(timezone.utc)
    removed_recurring = 0
    for addon in res.scalars().all():
        if addon.status == InstanceDiskAddonStatus.ACTIVE:
            removed_recurring += int(addon.recurring_amount or 0)
        addon.status = InstanceDiskAddonStatus.DELETED
        addon.deleted_at = now
        addon.remark = "磁盘已删除，不退款"

    await db.flush()
    return removed_recurring


async def _deleted_disk_addon_exists(db: AsyncSession, instance_id: int, disk_id: int) -> bool:
    try:
        res = await db.execute(
            select(InstanceDiskAddon.id)
            .where(
                InstanceDiskAddon.instance_id == instance_id,
                InstanceDiskAddon.upstream_disk_id == disk_id,
                InstanceDiskAddon.status == InstanceDiskAddonStatus.DELETED,
            )
            .limit(1)
        )
    except ProgrammingError as exc:
        if "instance_disk_addons" not in str(exc):
            raise
        await db.rollback()
        return False
    return res.first() is not None


async def _refund_disk_addon(
    db: AsyncSession,
    addon_id: int,
    order_id: int,
    user_id: int,
    amount: int,
    ref_no: str,
    reason: str,
) -> None:
    addon_res = await db.execute(
        select(InstanceDiskAddon)
        .where(InstanceDiskAddon.id == addon_id)
        .with_for_update()
    )
    addon = addon_res.scalar_one_or_none()

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

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

    if addon and addon.status == InstanceDiskAddonStatus.PENDING:
        addon.status = InstanceDiskAddonStatus.REFUNDED
        addon.refunded_at = datetime.now(timezone.utc)
        addon.remark = reason
    if order:
        order.status = OrderStatus.REFUNDED
        order.remark = reason
    if user and amount > 0:
        user.balance += amount
        db.add(BalanceLog(
            user_id=user_id,
            change_type=BalanceChangeType.ORDER_REFUND,
            delta=amount,
            balance_after=user.balance,
            ref_no=ref_no,
            remark=reason,
        ))
        event_type = (
            spend_service.SpendEventType.DISK_RESIZE
            if addon and addon.kind == "resize_disk"
            else spend_service.SpendEventType.DISK_CREATE
        )
        await spend_service.record_spend_event(
            db,
            user_id=user_id,
            event_type=event_type,
            amount_cents=-amount,
            event_key=f"{ref_no}:disk_addon_{addon_id}:refund",
            order_id=order_id,
            ref_no=ref_no,
            remark=reason,
            payload_snapshot={"addon_id": addon_id},
        )
        await membership_service.reverse_order_point_rewards(
            db,
            order_id=order_id,
            reason=reason,
        )
    await db.commit()


async def _operate(
    instance_id: int,
    api_path: str,
    method: str,
    current_user: User,
    db: AsyncSession,
    data: dict[str, Any] | None = None,
) -> ApiResponse:
    """通用操作：鉴权 → 调魔方云 → 返回 taskid。"""
    res = await db.execute(
        select(Instance).where(
            Instance.id == instance_id, Instance.user_id == current_user.id
        )
    )
    instance = res.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    if not instance.upstream_id:
        raise HTTPException(status_code=400, detail="实例尚未开通，无法操作")
    if not instance.cloud_server_id:
        raise HTTPException(status_code=400, detail="未绑定云节点")

    task_data = await instance_lifecycle_service.call_instance_cloud(
        db,
        instance,
        api_path,
        method,
        data=data,
    )
    task_id = task_data.get("taskid") if isinstance(task_data, dict) else None
    response = {"taskid": task_id}
    if isinstance(task_data, dict):
        response.update({key: value for key, value in task_data.items() if key != "taskid"})
    return ApiResponse(data=response)


@router.post("/{instance_id}/power-on", response_model=ApiResponse)
async def power_on(
    instance_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    return await _operate(instance_id, "/on", "POST", current_user, db)


@router.post("/{instance_id}/shutdown", response_model=ApiResponse)
async def shutdown(
    instance_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    return await _operate(instance_id, "/off", "POST", current_user, db)


@router.post("/{instance_id}/hard-shutdown", response_model=ApiResponse)
async def hard_shutdown(
    instance_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    return await _operate(instance_id, "/hardoff", "POST", current_user, db)


@router.post("/{instance_id}/reboot", response_model=ApiResponse)
async def reboot(
    instance_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    return await _operate(instance_id, "/reboot", "POST", current_user, db)


async def build_plugin_manage_context(
    instance: Instance,
    product: Product,
    db: AsyncSession,
    user: User | None = None,
) -> tuple[ProvisionContext, dict[str, Any], Order | None]:
    """构建插件管理动作的 ProvisionContext，供用户端和管理端复用。"""
    order_res = await db.execute(select(Order).where(Order.id == instance.order_id))
    order = order_res.scalar_one_or_none()
    snapshot: dict[str, Any] = {}
    for item in (order.items_snapshot if order and isinstance(order.items_snapshot, list) else []):
        if isinstance(item, dict) and int(item.get("product_id") or 0) == int(instance.product_id):
            snapshot = item
            break

    upstream_context = build_upstream_param_context(instance, snapshot)
    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(snapshot)
    context_snapshot = {
        **sanitize_plugin_snapshot(snapshot),
        "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": product.provision_interface_id,
    }
    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"],
    )
    return context, snapshot, order


@router.post("/{instance_id}/plugin-actions/{action}", response_model=ApiResponse)
async def run_plugin_action(
    instance_id: int,
    action: str,
    body: PluginManageActionBody | None = None,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    res = await db.execute(
        select(Instance).where(
            Instance.id == instance_id,
            Instance.user_id == current_user.id,
        )
    )
    instance = res.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    product_res = await db.execute(select(Product).where(Product.id == instance.product_id))
    product = product_res.scalar_one_or_none()
    if not product or product.api_type != ProductApiType.PLUGIN.value or not product.api_identifier:
        raise HTTPException(status_code=400, detail="该实例不是插件开通产品")
    if instance.status not in (InstanceStatus.ACTIVE, InstanceStatus.SUSPENDED, InstanceStatus.PROVISION_FAILED):
        raise HTTPException(status_code=400, detail="当前实例状态不允许执行插件管理动作")

    plugin_id = str(product.api_identifier)
    manifest = provision_plugin_service.get_plugin_manifest(plugin_id)
    if not provision_plugin_service.plugin_supports_action(manifest, action):
        raise HTTPException(status_code=400, detail=f"插件未声明管理动作：{action}")

    # 拒绝普通用户执行管理员专属动作
    action_audience = "both"
    for item in (manifest.get("manage_actions") or []):
        if isinstance(item, dict) and item.get("key") == action:
            action_audience = str(item.get("audience") or "both")
            break
    if action_audience == "admin":
        raise HTTPException(status_code=400, detail="该操作仅限管理员执行")
    provider = provision_plugin_service.create_plugin_provider(plugin_id)
    manage = getattr(provider, "manage", None)
    if not callable(manage):
        raise HTTPException(status_code=400, detail=f"插件 {plugin_id} 未实现 manage()")

    context, _snapshot, _order = await build_plugin_manage_context(instance, product, db, current_user)
    try:
        result = manage(action, context, (body.payload if body else None) or {})
        if hasattr(result, "__await__"):
            result = await result
        if is_dataclass(result):
            result = asdict(result)
    except ProvisionPluginError as exc:
        raise HTTPException(status_code=502 if exc.retryable else 400, detail=str(exc)) from exc
    except Exception as exc:
        raise HTTPException(status_code=502, detail=f"插件管理动作执行失败：{exc}") from exc

    if not isinstance(result, dict):
        result = {"success": bool(result), "message": str(result)}
    if result.get("success") is False:
        raise HTTPException(status_code=400, detail=str(result.get("message") or "插件管理动作失败"))

    result_payload = result.get("payload") if isinstance(result.get("payload"), dict) else {}
    returned_password = (
        result_payload.get("root_password")
        or result_payload.get("new_password")
        or result_payload.get("password")
    )
    if returned_password:
        instance.root_password = encrypt_secret(str(returned_password))

    status_value = result.get("instance_status") or result_payload.get("instance_status")
    if status_value:
        try:
            next_status = InstanceStatus(str(status_value))
        except ValueError:
            next_status = None
        if next_status:
            instance.status = next_status
            if next_status == InstanceStatus.ACTIVE:
                instance.suspend_type = None
            elif next_status == InstanceStatus.SUSPENDED and not instance.suspend_type:
                instance.suspend_type = SuspendType.MANUAL
            elif next_status == InstanceStatus.TERMINATED and not instance.terminated_at:
                instance.terminated_at = datetime.now(timezone.utc)

    public_result_payload = dict(result_payload)
    for secret_key in ("root_password", "new_password", "password"):
        if secret_key in public_result_payload:
            public_result_payload[secret_key] = "***"
    result = {**result, "payload": public_result_payload}

    payload = instance.upstream_payload if isinstance(instance.upstream_payload, dict) else {}
    action_logs = payload.get("plugin_actions") if isinstance(payload.get("plugin_actions"), list) else []
    action_logs = [
        *action_logs[-19:],
        {
            "action": action,
            "message": result.get("message"),
            "task_id": result.get("task_id"),
            "payload": public_result_payload,
            "created_at": datetime.now(timezone.utc).isoformat(),
        },
    ]
    instance.upstream_payload = {**payload, "plugin_actions": action_logs}
    await db.flush()
    return ApiResponse(data=result)


@router.post("/{instance_id}/suspend", response_model=ApiResponse)
async def suspend_instance(
    instance_id: int,
    body: SuspendInstanceBody,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(Instance)
        .where(Instance.id == instance_id, Instance.user_id == current_user.id)
        .with_for_update()
    )
    instance = result.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    if instance.status != InstanceStatus.ACTIVE:
        raise HTTPException(status_code=400, detail="当前实例状态不支持暂停")

    data = await instance_lifecycle_service.suspend_instance(
        db,
        instance,
        suspend_type=body.type,
        free_gpu=body.free_gpu,
        reason="用户手动暂停实例" if body.type != "due" else "到期暂停",
        source="manual",
        user_id=current_user.id,
    )
    return ApiResponse(data=data)


@router.post("/{instance_id}/unsuspend", response_model=ApiResponse)
async def unsuspend_instance(
    instance_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(Instance)
        .where(Instance.id == instance_id, Instance.user_id == current_user.id)
        .with_for_update()
    )
    instance = result.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    if instance.status != InstanceStatus.SUSPENDED:
        raise HTTPException(status_code=400, detail="当前实例状态不需要解除暂停")

    data = await instance_lifecycle_service.unsuspend_instance(
        db,
        instance,
        reason="用户手动解除暂停",
        source="manual",
        user_id=current_user.id,
    )
    return ApiResponse(data=data)


@router.delete("/{instance_id}", response_model=ApiResponse)
async def delete_instance(
    instance_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(Instance)
        .where(Instance.id == instance_id, Instance.user_id == current_user.id)
        .with_for_update()
    )
    instance = result.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    if instance.status == InstanceStatus.TERMINATED:
        return ApiResponse(data={"taskid": None, "instance_id": instance.id})

    data = await instance_lifecycle_service.force_delete_instance(
        db,
        instance,
        reason="用户删除实例，不退款",
        source="manual",
        user_id=current_user.id,
    )
    return ApiResponse(data=data, message="删除实例任务已下发")


@router.post("/{instance_id}/reset-password", response_model=ApiResponse)
async def reset_password(
    instance_id: int,
    body: PasswordResetBody,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    return await _operate(
        instance_id, "/password", "PUT", current_user, db,
        data={"password": body.password, "password_type": 0},
    )


@router.put("/{instance_id}/reinstall", response_model=ApiResponse)
async def reinstall_instance(
    instance_id: int,
    body: ReinstallInstanceBody,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    res = await db.execute(
        select(Instance, Product, ProductGroup)
        .outerjoin(Product, Product.id == Instance.product_id)
        .outerjoin(ProductGroup, ProductGroup.id == Product.group_id)
        .where(Instance.id == instance_id, Instance.user_id == current_user.id)
    )
    row = res.first()
    if not row:
        raise HTTPException(status_code=404, detail="实例不存在")
    instance, product, group = row
    if instance.status != InstanceStatus.ACTIVE:
        raise HTTPException(status_code=400, detail="实例状态异常，无法重装系统")
    if not instance.upstream_id or not instance.cloud_server_id:
        raise HTTPException(status_code=400, detail="实例尚未开通，无法重装系统")

    is_zjmf_finance = bool(await zjmf_instance_adapter.get_zjmf_server_for_instance(db, instance))
    if is_zjmf_finance:
        os_label = None
        try:
            reinstall_options = await zjmf_instance_adapter.get_reinstall_options(db, instance)
            os_label = zjmf_instance_adapter.find_reinstall_os_label(reinstall_options, body.os)
        except Exception as option_exc:
            log.info("reinstall %s fetch zjmf reinstall options skipped: %s", instance_id, option_exc)
        if not os_label:
            os_option = _find_os_option(product, group, body.os)
            if os_option:
                os_label = str(os_option.get("label") or os_option.get("name") or body.os)
        if not os_label:
            raise HTTPException(status_code=400, detail="selected image is not supported by upstream")
    else:
        os_option = _find_os_option(product, group, body.os)
        if not os_option:
            raise HTTPException(status_code=400, detail="所选镜像不在当前产品组支持范围内")
        os_label = str(os_option.get("label") or os_option.get("name") or body.os)

    password = body.password.strip() if body.password is not None else None
    if password is not None and len(password) < 6:
        raise HTTPException(status_code=400, detail="密码至少需要 6 位")

    if is_zjmf_finance:
        response_payload = await zjmf_instance_adapter.reinstall_instance(
            db,
            instance,
            os_id=body.os,
            password=password,
            port=instance.ssh_port,
            part_type=body.format_data_disk,
        )
        instance.os_image = f"{body.os}|{os_label}"
        await db.flush()
        upstream = response_payload.get("upstream") if isinstance(response_payload, dict) else {}
        if not isinstance(upstream, dict):
            upstream = {}
        returned_password = upstream.get("password") or password
        return ApiResponse(data={
            "taskid": response_payload.get("taskid"),
            "user": upstream.get("user"),
            "password": returned_password,
            "port": upstream.get("port") or instance.ssh_port,
            "system_disk_size": upstream.get("system_disk_size"),
            "provider": "zjmf_finance",
            "upstream": upstream,
        })

    payload: dict[str, Any] = {
        "os": body.os,
        "format_data_disk": body.format_data_disk,
        "password_type": 0,
    }
    if password:
        payload["password"] = password

    client, _ = await _get_dcim_client(instance.cloud_server_id, db)
    try:
        code, text, response_body = await client.request(
            f"/clouds/{instance.upstream_id}/reinstall?image_client_hidden=0",
            data=payload,
            method="PUT",
            timeout=30.0,
        )
    except DcimCloudError as e:
        raise HTTPException(status_code=502, detail=f"上游调用失败：{e}")

    if not (200 <= code < 300):
        raise HTTPException(status_code=502, detail=f"上游返回异常（HTTP {code}）：{text[:200]}")

    response_payload = (response_body or {}).get("data") or response_body or {}
    if not isinstance(response_payload, dict):
        response_payload = {}

    port = _to_int_or_none(response_payload.get("port"))
    system_disk_size = _to_int_or_none(response_payload.get("system_disk_size"))
    returned_password = response_payload.get("password")

    instance.os_image = f"{body.os}|{os_label}"
    if port:
        instance.ssh_port = port
    if system_disk_size and system_disk_size > 0:
        instance.system_disk_size = system_disk_size
    if returned_password:
        instance.root_password = encrypt_secret(str(returned_password))
    elif password:
        instance.root_password = encrypt_secret(password)
    await db.flush()

    return ApiResponse(data={
        "taskid": response_payload.get("taskid"),
        "user": response_payload.get("user"),
        "password": returned_password or password,
        "port": port or response_payload.get("port"),
        "system_disk_size": system_disk_size or response_payload.get("system_disk_size"),
    })


async def _fetch_upstream_instance_detail(
    instance: Instance,
    db: AsyncSession,
    *,
    action_label: str,
) -> tuple[DcimCloudClient, dict[str, Any]]:
    if not instance.upstream_id:
        raise HTTPException(status_code=400, detail=f"实例尚未开通，无法{action_label}")
    if not instance.cloud_server_id:
        raise HTTPException(status_code=400, detail="未绑定云节点")

    client, _ = await _get_dcim_client(instance.cloud_server_id, db)
    try:
        code, text, detail_body = await client.request(
            f"/clouds/{instance.upstream_id}", method="GET", timeout=8.0
        )
    except DcimCloudError as e:
        raise HTTPException(status_code=502, detail=f"上游调用失败：{e}")

    if not (200 <= code < 300):
        raise HTTPException(status_code=502, detail=f"上游返回异常（HTTP {code}）：{text[:200]}")

    detail = (detail_body or {}).get("data") or detail_body or {}
    if not isinstance(detail, dict):
        raise HTTPException(status_code=502, detail="上游返回数据格式异常")
    return client, detail


@router.post("/{instance_id}/disks/quote", response_model=ApiResponse)
async def quote_disk(
    instance_id: int,
    body: DiskCreateBody,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    res = await db.execute(
        select(Instance, Product)
        .outerjoin(Product, Product.id == Instance.product_id)
        .where(Instance.id == instance_id, Instance.user_id == current_user.id)
    )
    row = res.one_or_none()
    if not row:
        raise HTTPException(status_code=404, detail="实例不存在")
    instance, product = row
    await _reject_zjmf_unsupported(db, instance)
    if instance.status != InstanceStatus.ACTIVE:
        raise HTTPException(status_code=400, detail="当前实例状态不支持添加磁盘")
    if not instance.upstream_id:
        raise HTTPException(status_code=400, detail="实例尚未开通，无法添加磁盘")

    billing = _calculate_disk_billing(body, instance, product)
    active_totals = await _disk_addon_totals(db, [instance.id])
    next_renew_amount = _effective_renew_amount(instance, active_totals.get(instance.id, 0)) + billing["recurring_amount"]

    return ApiResponse(data={
        **_disk_billing_snapshot(billing),
        "next_renew_amount": next_renew_amount,
    })


@router.post("/{instance_id}/disks/{disk_id}/resize/quote", response_model=ApiResponse)
async def quote_resize_disk(
    instance_id: int,
    disk_id: int,
    body: DiskResizeBody,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    res = await db.execute(
        select(Instance, Product)
        .outerjoin(Product, Product.id == Instance.product_id)
        .where(Instance.id == instance_id, Instance.user_id == current_user.id)
    )
    row = res.one_or_none()
    if not row:
        raise HTTPException(status_code=404, detail="实例不存在")
    instance, product = row
    await _reject_zjmf_unsupported(db, instance)
    if instance.status != InstanceStatus.ACTIVE:
        raise HTTPException(status_code=400, detail="当前实例状态不支持扩容磁盘")

    _, detail = await _fetch_upstream_instance_detail(instance, db, action_label="扩容磁盘")
    disk = _find_upstream_disk(detail, disk_id)
    current_size, delta_size = _validate_resize_disk(body, product, disk)
    billing = _calculate_disk_billing_for_size(body.size, delta_size, instance, product)
    active_totals = await _disk_addon_totals(db, [instance.id])
    next_renew_amount = _effective_renew_amount(instance, active_totals.get(instance.id, 0)) + billing["recurring_amount"]

    return ApiResponse(data={
        **_disk_billing_snapshot(billing),
        "diskid": disk_id,
        "source_size": current_size,
        "target_size": body.size,
        "delta_size": delta_size,
        "next_renew_amount": next_renew_amount,
    })


@router.post("/{instance_id}/disks", response_model=ApiResponse)
async def create_disk(
    instance_id: int,
    body: DiskCreateBody,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    res = await db.execute(
        select(Instance, Product)
        .outerjoin(Product, Product.id == Instance.product_id)
        .where(Instance.id == instance_id, Instance.user_id == current_user.id)
    )
    row = res.one_or_none()
    if not row:
        raise HTTPException(status_code=404, detail="实例不存在")
    instance, product = row
    await _reject_zjmf_unsupported(db, instance)
    if instance.status != InstanceStatus.ACTIVE:
        raise HTTPException(status_code=400, detail="当前实例状态不支持添加磁盘")
    if not instance.upstream_id:
        raise HTTPException(status_code=400, detail="实例尚未开通，无法添加磁盘")
    if not instance.cloud_server_id:
        raise HTTPException(status_code=400, detail="未绑定云节点")

    client, _ = await _get_dcim_client(instance.cloud_server_id, db)

    try:
        code, text, detail_body = await client.request(
            f"/clouds/{instance.upstream_id}", method="GET", timeout=8.0
        )
    except DcimCloudError as e:
        raise HTTPException(status_code=502, detail=f"上游调用失败：{e}")

    if not (200 <= code < 300):
        raise HTTPException(status_code=502, detail=f"上游返回异常（HTTP {code}）：{text[:200]}")

    detail = (detail_body or {}).get("data") or detail_body or {}
    disks = detail.get("disk") if isinstance(detail, dict) else []
    upstream_disk_count = len(disks) if isinstance(disks, list) else 0
    upstream_disk_ids = _disk_ids_from_upstream(disks)
    pending_missing_count = await _pending_missing_disk_addon_count(db, instance.id, upstream_disk_ids)
    if upstream_disk_count + pending_missing_count >= 3:
        raise HTTPException(status_code=400, detail="单个实例最多允许 3 块磁盘")

    create_payload = _build_disk_create_payload(body, product, detail if isinstance(detail, dict) else {})
    billing = _calculate_disk_billing(body, instance, product)

    locked_res = await db.execute(
        select(Instance)
        .where(Instance.id == instance_id, Instance.user_id == current_user.id)
        .with_for_update()
    )
    locked_instance = locked_res.scalar_one_or_none()
    if not locked_instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    if locked_instance.status != InstanceStatus.ACTIVE:
        raise HTTPException(status_code=400, detail="当前实例状态不支持添加磁盘")
    pending_missing_count = await _pending_missing_disk_addon_count(db, locked_instance.id, upstream_disk_ids)
    if upstream_disk_count + pending_missing_count >= 3:
        raise HTTPException(status_code=400, detail="单个实例最多允许 3 块磁盘")

    user_res = await db.execute(
        select(User).where(User.id == current_user.id).with_for_update()
    )
    locked_user = user_res.scalar_one()
    if locked_user.balance < billing["current_charge"]:
        raise HTTPException(
            status_code=402,
            detail=f"余额不足，需要 ¥{billing['current_charge'] / 100:.2f}，当前余额 ¥{locked_user.balance / 100:.2f}",
        )

    locked_user.balance -= billing["current_charge"]
    out_order_no = _gen_disk_order_no()
    billing_snapshot = _disk_billing_snapshot(billing)
    order = Order(
        user_id=current_user.id,
        out_order_no=out_order_no,
        status=OrderStatus.PAID,
        total_amount=billing["current_charge"],
        discount_amount=0,
        items_snapshot=[{
            "type": "disk_addon",
            "instance_id": locked_instance.id,
            "product_id": locked_instance.product_id,
            "cycle": locked_instance.cycle,
            "size_gb": body.size,
            "current_charge_cents": billing["current_charge"],
            "recurring_amount_cents": billing["recurring_amount"],
            "charged_until": billing_snapshot["charged_until"],
            "request": create_payload,
        }],
        remark=f"实例 #{locked_instance.id} 添加弹性云盘 {body.size} GB",
    )
    db.add(order)
    await db.flush()

    addon = InstanceDiskAddon(
        user_id=current_user.id,
        instance_id=locked_instance.id,
        order_id=order.id,
        kind="create_disk",
        size_gb=body.size,
        source_size_gb=0,
        target_size_gb=body.size,
        delta_size_gb=body.size,
        store_id=create_payload["store"],
        driver=create_payload.get("driver"),
        cache=create_payload.get("cache"),
        io=create_payload.get("io"),
        iops_min=create_payload.get("iops_min") or 0,
        iops_max=create_payload.get("iops_max") or 0,
        cycle=locked_instance.cycle,
        current_charge_amount=billing["current_charge"],
        recurring_amount=billing["recurring_amount"],
        charged_until=billing["charged_until"],
        status=InstanceDiskAddonStatus.PENDING,
        payload_snapshot={
            "request": create_payload,
            "billing": billing_snapshot,
        },
    )
    db.add(addon)
    db.add(BalanceLog(
        user_id=current_user.id,
        change_type=BalanceChangeType.ORDER_PAY,
        delta=-billing["current_charge"],
        balance_after=locked_user.balance,
        ref_no=out_order_no,
        remark=f"实例 #{locked_instance.id} 添加弹性云盘",
    ))
    await spend_service.record_spend_event(
        db,
        user_id=current_user.id,
        event_type=spend_service.SpendEventType.DISK_CREATE,
        amount_cents=billing["current_charge"],
        event_key=f"{out_order_no}:disk_create",
        order_id=order.id,
        ref_no=out_order_no,
        remark=f"实例 #{locked_instance.id} 添加弹性云盘",
        payload_snapshot={
            "instance_id": locked_instance.id,
            "size_gb": body.size,
            "recurring_amount": billing["recurring_amount"],
        },
    )
    await db.flush()
    await db.commit()

    try:
        code, text, create_body = await client.request(
            f"/clouds/{instance.upstream_id}/disks",
            method="POST",
            data=create_payload,
        )
    except DcimCloudError as e:
        await _refund_disk_addon(
            db,
            addon.id,
            order.id,
            current_user.id,
            billing["current_charge"],
            out_order_no,
            f"添加弹性云盘失败，已退款：{e}",
        )
        raise HTTPException(status_code=502, detail=f"上游调用失败：{e}")

    if not (200 <= code < 300):
        await _refund_disk_addon(
            db,
            addon.id,
            order.id,
            current_user.id,
            billing["current_charge"],
            out_order_no,
            f"添加弹性云盘失败，已退款：上游返回 HTTP {code}",
        )
        raise HTTPException(status_code=502, detail=f"上游返回异常（HTTP {code}）：{text[:200]}")

    payload = (create_body or {}).get("data") or create_body or {}
    if not isinstance(payload, dict):
        payload = {}

    addon_res = await db.execute(
        select(InstanceDiskAddon)
        .where(InstanceDiskAddon.id == addon.id)
        .with_for_update()
    )
    saved_addon = addon_res.scalar_one()
    saved_addon.upstream_task_id = _to_int_or_none(payload.get("taskid"))
    saved_addon.upstream_disk_id = _to_int_or_none(payload.get("diskid"))
    if saved_addon.upstream_disk_id is not None:
        saved_addon.status = InstanceDiskAddonStatus.ACTIVE
        saved_addon.activated_at = datetime.now(timezone.utc)
    else:
        saved_addon.status = InstanceDiskAddonStatus.PENDING
        saved_addon.remark = "上游未返回磁盘 ID，等待任务完成后确认"
    saved_addon.payload_snapshot = {
        "request": create_payload,
        "response": payload,
        "billing": billing_snapshot,
    }

    order_res = await db.execute(
        select(Order).where(Order.id == order.id).with_for_update()
    )
    saved_order = order_res.scalar_one()
    saved_order.items_snapshot = [{
        **(saved_order.items_snapshot or [{}])[0],
        "taskid": payload.get("taskid"),
        "diskid": payload.get("diskid"),
    }]

    inst_res = await db.execute(
        select(Instance).where(Instance.id == locked_instance.id).with_for_update()
    )
    final_instance = inst_res.scalar_one()
    await db.flush()
    active_totals = await _disk_addon_totals(db, [final_instance.id])
    final_instance.next_renew_amount = _effective_renew_amount(final_instance, active_totals.get(final_instance.id, 0))
    await _finalize_sales_rewards(
        db,
        order=saved_order,
        user_id=current_user.id,
        paid_amount=billing["current_charge"],
        instance_id=final_instance.id,
        source_type=spend_service.SpendEventType.DISK_CREATE,
        point_reward_status=(
            "claimable"
            if saved_addon.status == InstanceDiskAddonStatus.ACTIVE
            else "pending"
        ),
    )
    await db.commit()

    return ApiResponse(data={
        "taskid": payload.get("taskid"),
        "diskid": payload.get("diskid"),
        "order_id": order.id,
        "order_no": out_order_no,
        "amount": billing["current_charge"],
        "recurring_amount": billing["recurring_amount"],
        "balance": locked_user.balance,
        "next_renew_amount": final_instance.next_renew_amount,
    })


@router.put("/{instance_id}/disks/{disk_id}/resize", response_model=ApiResponse)
async def resize_disk(
    instance_id: int,
    disk_id: int,
    body: DiskResizeBody,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    res = await db.execute(
        select(Instance, Product)
        .outerjoin(Product, Product.id == Instance.product_id)
        .where(Instance.id == instance_id, Instance.user_id == current_user.id)
    )
    row = res.one_or_none()
    if not row:
        raise HTTPException(status_code=404, detail="实例不存在")
    instance, product = row
    await _reject_zjmf_unsupported(db, instance)
    if instance.status != InstanceStatus.ACTIVE:
        raise HTTPException(status_code=400, detail="当前实例状态不支持扩容磁盘")

    client, detail = await _fetch_upstream_instance_detail(instance, db, action_label="扩容磁盘")
    disk = _find_upstream_disk(detail, disk_id)
    current_size, delta_size = _validate_resize_disk(body, product, disk)
    resize_payload = _build_disk_resize_payload(body, disk)
    billing = _calculate_disk_billing_for_size(body.size, delta_size, instance, product)

    locked_res = await db.execute(
        select(Instance)
        .where(Instance.id == instance_id, Instance.user_id == current_user.id)
        .with_for_update()
    )
    locked_instance = locked_res.scalar_one_or_none()
    if not locked_instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    if locked_instance.status != InstanceStatus.ACTIVE:
        raise HTTPException(status_code=400, detail="当前实例状态不支持扩容磁盘")

    user_res = await db.execute(
        select(User).where(User.id == current_user.id).with_for_update()
    )
    locked_user = user_res.scalar_one()
    if locked_user.balance < billing["current_charge"]:
        raise HTTPException(
            status_code=402,
            detail=f"余额不足，需要 ¥{billing['current_charge'] / 100:.2f}，当前余额 ¥{locked_user.balance / 100:.2f}",
        )

    locked_user.balance -= billing["current_charge"]
    out_order_no = _gen_disk_order_no()
    billing_snapshot = _disk_billing_snapshot(billing)
    order = Order(
        user_id=current_user.id,
        out_order_no=out_order_no,
        status=OrderStatus.PAID,
        total_amount=billing["current_charge"],
        discount_amount=0,
        items_snapshot=[{
            "type": "disk_resize",
            "instance_id": locked_instance.id,
            "product_id": locked_instance.product_id,
            "cycle": locked_instance.cycle,
            "diskid": disk_id,
            "source_size_gb": current_size,
            "target_size_gb": body.size,
            "delta_size_gb": delta_size,
            "current_charge_cents": billing["current_charge"],
            "recurring_amount_cents": billing["recurring_amount"],
            "charged_until": billing_snapshot["charged_until"],
            "request": resize_payload,
        }],
        remark=f"实例 #{locked_instance.id} 扩容磁盘 #{disk_id}：{current_size} GB -> {body.size} GB",
    )
    db.add(order)
    await db.flush()

    addon = InstanceDiskAddon(
        user_id=current_user.id,
        instance_id=locked_instance.id,
        order_id=order.id,
        upstream_disk_id=disk_id,
        kind="resize_disk",
        size_gb=body.size,
        source_size_gb=current_size,
        target_size_gb=body.size,
        delta_size_gb=delta_size,
        store_id=_to_int_or_none(disk.get("store_id")) or _to_int_or_none(disk.get("store")) or 0,
        driver=resize_payload.get("driver"),
        cache=resize_payload.get("cache"),
        io=resize_payload.get("io"),
        iops_min=resize_payload.get("iops_min") or 0,
        iops_max=resize_payload.get("iops_max") or 0,
        cycle=locked_instance.cycle,
        current_charge_amount=billing["current_charge"],
        recurring_amount=billing["recurring_amount"],
        charged_until=billing["charged_until"],
        status=InstanceDiskAddonStatus.PENDING,
        payload_snapshot={
            "request": resize_payload,
            "billing": billing_snapshot,
        },
    )
    db.add(addon)
    db.add(BalanceLog(
        user_id=current_user.id,
        change_type=BalanceChangeType.ORDER_PAY,
        delta=-billing["current_charge"],
        balance_after=locked_user.balance,
        ref_no=out_order_no,
        remark=f"实例 #{locked_instance.id} 扩容磁盘 #{disk_id}",
    ))
    await spend_service.record_spend_event(
        db,
        user_id=current_user.id,
        event_type=spend_service.SpendEventType.DISK_RESIZE,
        amount_cents=billing["current_charge"],
        event_key=f"{out_order_no}:disk_resize",
        order_id=order.id,
        ref_no=out_order_no,
        remark=f"实例 #{locked_instance.id} 扩容磁盘 #{disk_id}",
        payload_snapshot={
            "instance_id": locked_instance.id,
            "disk_id": disk_id,
            "source_size_gb": current_size,
            "target_size_gb": body.size,
            "recurring_amount": billing["recurring_amount"],
        },
    )
    await db.flush()
    await db.commit()

    try:
        code, text, resize_body = await client.request(
            f"/disks/{disk_id}",
            method="PUT",
            data=resize_payload,
        )
    except DcimCloudError as e:
        await _refund_disk_addon(
            db,
            addon.id,
            order.id,
            current_user.id,
            billing["current_charge"],
            out_order_no,
            f"扩容磁盘失败，已退款：{e}",
        )
        raise HTTPException(status_code=502, detail=f"上游调用失败：{e}")

    if not (200 <= code < 300):
        await _refund_disk_addon(
            db,
            addon.id,
            order.id,
            current_user.id,
            billing["current_charge"],
            out_order_no,
            f"扩容磁盘失败，已退款：上游返回 HTTP {code}",
        )
        raise HTTPException(status_code=502, detail=f"上游返回异常（HTTP {code}）：{text[:200]}")

    payload = (resize_body or {}).get("data") or resize_body or {}
    if not isinstance(payload, dict):
        payload = {}

    addon_res = await db.execute(
        select(InstanceDiskAddon)
        .where(InstanceDiskAddon.id == addon.id)
        .with_for_update()
    )
    saved_addon = addon_res.scalar_one()
    saved_addon.upstream_task_id = _to_int_or_none(payload.get("taskid"))
    if saved_addon.upstream_task_id is None:
        saved_addon.status = InstanceDiskAddonStatus.ACTIVE
        saved_addon.activated_at = datetime.now(timezone.utc)
    else:
        saved_addon.status = InstanceDiskAddonStatus.PENDING
        saved_addon.remark = "等待上游扩容任务完成后确认"
    saved_addon.payload_snapshot = {
        "request": resize_payload,
        "response": payload,
        "billing": billing_snapshot,
    }

    order_res = await db.execute(
        select(Order).where(Order.id == order.id).with_for_update()
    )
    saved_order = order_res.scalar_one()
    saved_order.items_snapshot = [{
        **(saved_order.items_snapshot or [{}])[0],
        "taskid": payload.get("taskid"),
    }]

    inst_res = await db.execute(
        select(Instance).where(Instance.id == locked_instance.id).with_for_update()
    )
    final_instance = inst_res.scalar_one()
    await db.flush()
    active_totals = await _disk_addon_totals(db, [final_instance.id])
    final_instance.next_renew_amount = _effective_renew_amount(final_instance, active_totals.get(final_instance.id, 0))
    await _finalize_sales_rewards(
        db,
        order=saved_order,
        user_id=current_user.id,
        paid_amount=billing["current_charge"],
        instance_id=final_instance.id,
        source_type=spend_service.SpendEventType.DISK_RESIZE,
        point_reward_status=(
            "claimable"
            if saved_addon.status == InstanceDiskAddonStatus.ACTIVE
            else "pending"
        ),
    )
    await db.commit()

    return ApiResponse(data={
        "taskid": payload.get("taskid"),
        "diskid": disk_id,
        "order_id": order.id,
        "order_no": out_order_no,
        "amount": billing["current_charge"],
        "recurring_amount": billing["recurring_amount"],
        "balance": locked_user.balance,
        "source_size": current_size,
        "target_size": body.size,
        "delta_size": delta_size,
        "next_renew_amount": final_instance.next_renew_amount,
    })


@router.post("/{instance_id}/disks/{disk_id}/mount", response_model=ApiResponse)
async def mount_disk(
    instance_id: int,
    disk_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    res = await db.execute(
        select(Instance, Product)
        .outerjoin(Product, Product.id == Instance.product_id)
        .where(
            Instance.id == instance_id,
            Instance.user_id == current_user.id,
        )
    )
    row = res.one_or_none()
    if not row:
        raise HTTPException(status_code=404, detail="实例不存在")
    instance, product = row
    await _reject_zjmf_unsupported(db, instance)
    if instance.status != InstanceStatus.ACTIVE:
        raise HTTPException(status_code=400, detail="当前实例状态不支持挂载磁盘")

    client, detail = await _fetch_upstream_instance_detail(instance, db, action_label="挂载磁盘")
    disk = _find_upstream_disk(detail, disk_id)
    if _is_system_disk(disk):
        raise HTTPException(status_code=400, detail="系统盘不支持挂载操作")
    if await _deleted_disk_addon_exists(db, instance.id, disk_id):
        raise HTTPException(status_code=400, detail="该磁盘已提交删除，不能重新挂载")

    disk_status = _disk_status_key(disk)
    if disk_status == "1":
        return ApiResponse(data={"taskid": None, "diskid": disk_id, "message": "磁盘已处于挂载状态"})
    if disk_status != "0":
        raise HTTPException(status_code=400, detail="当前磁盘状态不支持挂载")

    mount_payload = _build_disk_mount_payload(instance, product, detail, disk)
    try:
        code, text, body = await client.request(
            f"/disks/{disk_id}/mount",
            method="POST",
            data=mount_payload,
        )
    except DcimCloudError as e:
        raise HTTPException(status_code=502, detail=f"上游调用失败：{e}")

    if not (200 <= code < 300):
        raise HTTPException(status_code=502, detail=f"上游返回异常（HTTP {code}）：{text[:200]}")

    payload = (body or {}).get("data") or body or {}
    if not isinstance(payload, dict):
        payload = {}
    return ApiResponse(data={
        "taskid": payload.get("taskid"),
        "diskid": disk_id,
    })


@router.delete("/{instance_id}/disks/{disk_id}/detach", response_model=ApiResponse)
async def detach_disk(
    instance_id: int,
    disk_id: int,
    force: int = 0,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    res = await db.execute(
        select(Instance).where(
            Instance.id == instance_id,
            Instance.user_id == current_user.id,
        )
    )
    instance = res.scalar_one_or_none()
    if instance:
        await _reject_zjmf_unsupported(db, instance)
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    if instance.status != InstanceStatus.ACTIVE:
        raise HTTPException(status_code=400, detail="当前实例状态不支持卸载磁盘")

    client, detail = await _fetch_upstream_instance_detail(instance, db, action_label="卸载磁盘")
    disk = _find_upstream_disk(detail, disk_id)
    if _is_system_disk(disk):
        raise HTTPException(status_code=400, detail="系统盘不支持卸载")
    if _disk_status_key(disk) == "0":
        return ApiResponse(data={"taskid": None, "diskid": disk_id, "message": "磁盘已处于未挂载状态"})

    try:
        code, text, body = await client.request(
            f"/clouds/{instance.upstream_id}/disks/{disk_id}",
            method="DELETE",
            data={"force": 1 if force else 0},
        )
    except DcimCloudError as e:
        raise HTTPException(status_code=502, detail=f"上游调用失败：{e}")

    if not (200 <= code < 300):
        raise HTTPException(status_code=502, detail=f"上游返回异常（HTTP {code}）：{text[:200]}")

    payload = (body or {}).get("data") or body or {}
    if not isinstance(payload, dict):
        payload = {}
    return ApiResponse(data={
        "taskid": payload.get("taskid"),
        "diskid": disk_id,
    })


@router.delete("/{instance_id}/disks/{disk_id}", response_model=ApiResponse)
async def delete_disk(
    instance_id: int,
    disk_id: int,
    force: int = 0,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    res = await db.execute(
        select(Instance).where(
            Instance.id == instance_id,
            Instance.user_id == current_user.id,
        )
    )
    instance = res.scalar_one_or_none()
    if instance:
        await _reject_zjmf_unsupported(db, instance)
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    if instance.status != InstanceStatus.ACTIVE:
        raise HTTPException(status_code=400, detail="当前实例状态不支持删除磁盘")

    client, detail = await _fetch_upstream_instance_detail(instance, db, action_label="删除磁盘")
    disk = _find_upstream_disk(detail, disk_id)
    if _is_system_disk(disk):
        raise HTTPException(status_code=400, detail="系统盘不支持删除")
    if _disk_status_key(disk) != "0":
        raise HTTPException(status_code=400, detail="请先卸载磁盘，再删除")

    try:
        code, text, body = await client.request(
            f"/disks/{disk_id}",
            method="DELETE",
            data={"force": 1 if force else 0},
        )
    except DcimCloudError as e:
        raise HTTPException(status_code=502, detail=f"上游调用失败：{e}")

    if not (200 <= code < 300):
        raise HTTPException(status_code=502, detail=f"上游返回异常（HTTP {code}）：{text[:200]}")

    payload = (body or {}).get("data") or body or {}
    if not isinstance(payload, dict):
        payload = {}

    locked_res = await db.execute(
        select(Instance)
        .where(Instance.id == instance_id, Instance.user_id == current_user.id)
        .with_for_update()
    )
    locked_instance = locked_res.scalar_one_or_none()
    if not locked_instance:
        raise HTTPException(status_code=404, detail="实例不存在")

    removed_recurring = await _mark_disk_addons_deleted(db, locked_instance.id, disk_id)
    active_totals = await _disk_addon_totals(db, [locked_instance.id])
    locked_instance.next_renew_amount = _effective_renew_amount(locked_instance, active_totals.get(locked_instance.id, 0))
    await db.commit()

    return ApiResponse(data={
        "taskid": payload.get("taskid"),
        "diskid": disk_id,
        "removed_recurring_amount": removed_recurring,
        "next_renew_amount": locked_instance.next_renew_amount,
    })


@router.patch("/{instance_id}/auto-renew", response_model=ApiResponse)
async def update_auto_renew(
    instance_id: int,
    body: AutoRenewBody,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    inst_res = await db.execute(
        select(Instance)
        .where(Instance.id == instance_id, Instance.user_id == current_user.id)
        .with_for_update()
    )
    instance = inst_res.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    if instance.status == InstanceStatus.TERMINATED:
        raise HTTPException(status_code=400, detail="已销毁实例不能修改自动续费")

    if await zjmf_instance_adapter.get_zjmf_server_for_instance(db, instance):
        await zjmf_instance_adapter.set_auto_renew(db, instance, body.auto_renew)
    instance.auto_renew = 1 if body.auto_renew else 0
    await db.flush()
    return ApiResponse(data={"id": instance.id, "auto_renew": instance.auto_renew})


@router.post("/{instance_id}/renew", response_model=ApiResponse)
async def renew_instance(
    instance_id: int,
    body: RenewInstanceBody,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    inst_res = await db.execute(
        select(Instance)
        .where(Instance.id == instance_id, Instance.user_id == current_user.id)
        .with_for_update()
    )
    instance = inst_res.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    if instance.status not in (InstanceStatus.ACTIVE, InstanceStatus.SUSPENDED):
        raise HTTPException(status_code=400, detail="当前实例状态不支持续费")

    try:
        await _sync_pending_disk_addons_before_renewal(db, instance)
        result = await instance_renewal_service.renew_locked_instance(
            db,
            instance,
            cycles=body.cycles,
            source="manual",
        )
    except instance_renewal_service.RenewalError as exc:
        raise HTTPException(status_code=exc.status_code, detail=exc.detail) from exc

    await db.commit()

    # 插件型产品：续费后自动同步到期时间到上游
    product_res = await db.execute(select(Product).where(Product.id == instance.product_id))
    product = product_res.scalar_one_or_none()
    if product and product.api_type == ProductApiType.PLUGIN.value and product.api_identifier:
        try:
            manifest = provision_plugin_service.get_plugin_manifest(str(product.api_identifier))
            if provision_plugin_service.plugin_supports_action(manifest, "renew"):
                context, _snap, _order = await build_plugin_manage_context(instance, product, db)
                provider = provision_plugin_service.create_plugin_provider(str(product.api_identifier))
                manage = getattr(provider, "manage", None)
                if callable(manage):
                    try:
                        renew_result = manage("renew", context)
                        if hasattr(renew_result, "__await__"):
                            renew_result = await renew_result
                        log.info("plugin renew sync instance=%s result=%s", instance_id, renew_result)
                    except Exception as exc:
                        log.warning("plugin renew sync failed instance=%s: %s", instance_id, exc)
        except Exception as exc:
            log.warning("plugin renew setup failed instance=%s: %s", instance_id, exc)

    return ApiResponse(data={
        "order_id": result.order_id,
        "order_no": result.order_no,
        "amount": result.amount,
        "balance": result.balance,
        "expires_at": result.expires_at.isoformat(),
        "unsuspend_taskid": result.unsuspend_taskid,
    })


@router.post("/{instance_id}/vnc", response_model=ApiResponse)
async def open_vnc(
    instance_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """获取 VNC 凭据。代理上游 POST /clouds/{upstream_id}/vnc。"""
    res = await db.execute(
        select(Instance).where(
            Instance.id == instance_id, Instance.user_id == current_user.id
        )
    )
    instance = res.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    if not instance.upstream_id:
        raise HTTPException(status_code=400, detail="实例尚未开通，无法获取 VNC")
    if not instance.cloud_server_id:
        raise HTTPException(status_code=400, detail="未绑定云节点")

    if await zjmf_instance_adapter.get_zjmf_server_for_instance(db, instance):
        payload = await zjmf_instance_adapter.get_vnc_info(db, instance)
        return ApiResponse(data={
            "token": payload.get("token"),
            "vnc_pass": payload.get("vnc_pass"),
            "vnc_url": payload.get("vnc_url"),
            "vnc_url_http": payload.get("vnc_url_http"),
            "vnc_url_https": payload.get("vnc_url_https"),
            "password": payload.get("password"),
            "path": payload.get("path"),
            "direct_url": payload.get("direct_url"),
            "provider": "zjmf_finance",
        })

    client, _ = await _get_dcim_client(instance.cloud_server_id, db)
    try:
        code, text, body = await client.request(
            f"/clouds/{instance.upstream_id}/vnc", method="POST", timeout=10.0
        )
    except DcimCloudError as e:
        raise HTTPException(status_code=502, detail=f"上游调用失败：{e}")
    if not (200 <= code < 300):
        raise HTTPException(status_code=502, detail=f"上游返回异常（HTTP {code}）：{text[:200]}")

    payload = (body or {}).get("data") or body or {}
    if not isinstance(payload, dict):
        raise HTTPException(status_code=502, detail="上游返回数据格式异常")
    return ApiResponse(data={
        "token": payload.get("token"),
        "vnc_pass": payload.get("vnc_pass"),
        "vnc_url": payload.get("vnc_url"),
        "vnc_url_http": payload.get("vnc_url_http"),
        "vnc_url_https": payload.get("vnc_url_https"),
        "password": payload.get("password"),
        "path": payload.get("path"),
    })


@router.get("/{instance_id}/real-data", response_model=ApiResponse)
async def get_instance_real_data(
    instance_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    res = await db.execute(
        select(Instance).where(
            Instance.id == instance_id, Instance.user_id == current_user.id
        )
    )
    instance = res.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")

    cached = await get_cached_real_data(instance.id)
    return ApiResponse(data=cached or {})


@router.get("/{instance_id}/real-data-history", response_model=ApiResponse)
async def get_instance_real_data_history(
    instance_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    res = await db.execute(
        select(Instance).where(
            Instance.id == instance_id, Instance.user_id == current_user.id
        )
    )
    instance = res.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")

    history = await get_real_data_history(instance.id, db)
    return ApiResponse(data=history)


@router.get("/{instance_id}/cron-snapshots", response_model=ApiResponse)
async def list_instance_cron_snapshots(
    instance_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    res = await db.execute(
        select(Instance).where(
            Instance.id == instance_id,
            Instance.user_id == current_user.id,
        )
    )
    instance = res.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    if not instance.upstream_id or not instance.cloud_server_id:
        return ApiResponse(data=[])

    if await zjmf_instance_adapter.get_zjmf_server_for_instance(db, instance):
        return ApiResponse(data=await zjmf_instance_adapter.list_cron_snapshots(db, instance))

    client, _ = await _get_dcim_client(instance.cloud_server_id, db)
    try:
        code, text, body = await client.request(
            f"/clouds/{instance.upstream_id}/cron_snap",
            method="GET",
            timeout=10.0,
        )
    except DcimCloudError as e:
        raise HTTPException(status_code=502, detail=f"上游调用失败：{e}")
    if not (200 <= code < 300):
        raise HTTPException(status_code=502, detail=f"上游返回异常（HTTP {code}）：{text[:200]}")

    payload = body or {}
    items = payload.get("data") if isinstance(payload, dict) else payload
    return ApiResponse(data=items if isinstance(items, list) else [])


@router.post("/{instance_id}/cron-snapshots", response_model=ApiResponse)
async def create_instance_cron_snapshot(
    instance_id: int,
    body: CronSnapshotBody,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    res = await db.execute(
        select(Instance).where(
            Instance.id == instance_id,
            Instance.user_id == current_user.id,
        )
    )
    instance = res.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    if not instance.upstream_id or not instance.cloud_server_id:
        raise HTTPException(status_code=400, detail="实例尚未开通，无法添加定时备份/快照")

    if await zjmf_instance_adapter.get_zjmf_server_for_instance(db, instance):
        return ApiResponse(data=await zjmf_instance_adapter.create_cron_snapshot(
            db,
            instance,
            {
                "type": body.type,
                "frequency": body.frequency,
                "cycle": body.cycle,
                "disk": body.disk,
                "point": (body.point or "").strip(),
            },
        ))

    payload: list[tuple[str, Any]] = [
        ("type", body.type),
        ("frequency", body.frequency),
        ("cycle", body.cycle),
    ]
    point = (body.point or "").strip()
    if point:
        payload.append(("point", point))
    for disk_id in body.disk:
        payload.append(("disk[]", disk_id))

    client, _ = await _get_dcim_client(instance.cloud_server_id, db)
    try:
        code, text, response_body = await client.request(
            f"/clouds/{instance.upstream_id}/cron_snap",
            method="POST",
            data=payload,  # type: ignore[arg-type]
            timeout=15.0,
        )
    except DcimCloudError as e:
        raise HTTPException(status_code=502, detail=f"上游调用失败：{e}")
    if not (200 <= code < 300):
        raise HTTPException(status_code=502, detail=f"上游返回异常（HTTP {code}）：{text[:200]}")

    response_payload = (response_body or {}).get("data") or response_body or {}
    return ApiResponse(data=response_payload if isinstance(response_payload, dict) else {})


@router.delete("/{instance_id}/cron-snapshots/{cron_id}", response_model=ApiResponse)
async def delete_instance_cron_snapshot(
    instance_id: int,
    cron_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    res = await db.execute(
        select(Instance).where(
            Instance.id == instance_id,
            Instance.user_id == current_user.id,
        )
    )
    instance = res.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    if not instance.upstream_id or not instance.cloud_server_id:
        raise HTTPException(status_code=400, detail="实例尚未开通，无法删除定时备份/快照")

    if await zjmf_instance_adapter.get_zjmf_server_for_instance(db, instance):
        return ApiResponse(data=await zjmf_instance_adapter.delete_cron_snapshot(db, instance, cron_id=cron_id))

    client, _ = await _get_dcim_client(instance.cloud_server_id, db)
    try:
        code, text, response_body = await client.request(
            f"/clouds/{instance.upstream_id}/cron_snap/{cron_id}",
            method="DELETE",
            timeout=15.0,
        )
    except DcimCloudError as e:
        raise HTTPException(status_code=502, detail=f"上游调用失败：{e}")
    if not (200 <= code < 300):
        raise HTTPException(status_code=502, detail=f"上游返回异常（HTTP {code}）：{text[:200]}")

    response_payload = (response_body or {}).get("data") or response_body or {}
    return ApiResponse(data=response_payload if isinstance(response_payload, dict) else {"id": cron_id})


@router.get("/{instance_id}/snapshots", response_model=ApiResponse)
async def list_instance_snapshots(
    instance_id: int,
    page: int = Query(1, ge=1),
    per_page: int = Query(20, ge=1, le=100),
    orderby: str = Query("id"),
    sort: str = Query("desc"),
    search: str | None = Query(None),
    snapshot_type: str | None = Query(None, alias="type"),
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """代理上游 /clouds/{id}/snapshots。仅返回当前用户名下实例的备份/快照。"""
    allowed_orderby = {"id", "size", "name", "type"}
    allowed_sort = {"asc", "desc"}
    allowed_type = {"snap", "backup"}
    if orderby not in allowed_orderby:
        raise HTTPException(status_code=400, detail="排序字段无效")
    if sort not in allowed_sort:
        raise HTTPException(status_code=400, detail="排序方向无效")
    if snapshot_type and snapshot_type not in allowed_type:
        raise HTTPException(status_code=400, detail="备份类型无效")

    res = await db.execute(
        select(Instance).where(
            Instance.id == instance_id,
            Instance.user_id == current_user.id,
        )
    )
    instance = res.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    if not instance.upstream_id or not instance.cloud_server_id:
        return ApiResponse(data={
            "list": [],
            "meta": {"total": 0, "total_page": 0, "page": page, "per_page": per_page},
        })

    params: dict[str, Any] = {
        "page": page,
        "per_page": per_page,
        "orderby": orderby,
        "sort": sort,
    }
    if search:
        params["search"] = search.strip()
    if snapshot_type:
        params["type"] = snapshot_type

    if await zjmf_instance_adapter.get_zjmf_server_for_instance(db, instance):
        return ApiResponse(data=await zjmf_instance_adapter.list_snapshots(
            db,
            instance,
            page=page,
            per_page=per_page,
            orderby=orderby,
            sort=sort,
            search=search.strip() if search else None,
            snapshot_type=snapshot_type,
        ))

    client, _ = await _get_dcim_client(instance.cloud_server_id, db)
    try:
        code, text, body = await client.request(
            f"/clouds/{instance.upstream_id}/snapshots",
            method="GET",
            data=params,
            timeout=10.0,
        )
    except DcimCloudError as e:
        raise HTTPException(status_code=502, detail=f"上游调用失败：{e}")
    if not (200 <= code < 300):
        raise HTTPException(status_code=502, detail=f"上游返回异常（HTTP {code}）：{text[:200]}")

    payload = body or {}
    items = payload.get("data") or []
    meta = payload.get("meta") or {}
    if not isinstance(items, list):
        items = []
    if not isinstance(meta, dict):
        meta = {}
    return ApiResponse(data={
        "list": items,
        "meta": {
            "total": _to_int_or_none(meta.get("total")) or 0,
            "total_page": _to_int_or_none(meta.get("total_page")) or 0,
            "page": _to_int_or_none(meta.get("page")) or page,
            "per_page": _to_int_or_none(meta.get("per_page")) or per_page,
        },
    })


@router.post("/{instance_id}/disks/{disk_id}/snapshots", response_model=ApiResponse)
async def create_instance_snapshot(
    instance_id: int,
    disk_id: int,
    body: SnapshotCreateBody,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    res = await db.execute(
        select(Instance).where(
            Instance.id == instance_id,
            Instance.user_id == current_user.id,
        )
    )
    instance = res.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    if instance.status != InstanceStatus.ACTIVE:
        raise HTTPException(status_code=400, detail="当前实例状态不支持创建备份/快照")

    if await zjmf_instance_adapter.get_zjmf_server_for_instance(db, instance):
        return ApiResponse(data=await zjmf_instance_adapter.create_snapshot(
            db,
            instance,
            disk_id=disk_id,
            snapshot_type=body.type,
            name=body.name,
            remarks=body.remarks,
        ))

    client, detail = await _fetch_upstream_instance_detail(instance, db, action_label="创建备份/快照")
    disk = _find_upstream_disk(detail, disk_id)
    if _disk_status_key(disk) == "2":
        raise HTTPException(status_code=400, detail="磁盘正在挂载中，请稍后再创建备份/快照")

    limit = _snapshot_limit_from_detail(detail, body.type)
    current_count = _snapshot_count_from_disk(disk, body.type)
    if limit is not None and current_count is not None and current_count >= limit:
        label = "快照" if body.type == "snap" else "备份"
        raise HTTPException(status_code=400, detail=f"{label}数量已达到上限 {limit}")

    payload: dict[str, Any] = {"type": body.type}
    name = (body.name or "").strip()
    remarks = (body.remarks or "").strip()
    if name:
        payload["name"] = name
    if remarks:
        payload["remarks"] = remarks
        payload.setdefault("name", remarks)

    try:
        code, text, response_body = await client.request(
            f"/disks/{disk_id}/snapshots?hostid={instance.upstream_id}",
            method="POST",
            data=payload,
            timeout=15.0,
        )
    except DcimCloudError as e:
        raise HTTPException(status_code=502, detail=f"上游调用失败：{e}")
    if not (200 <= code < 300):
        raise HTTPException(status_code=502, detail=f"上游返回异常（HTTP {code}）：{text[:200]}")

    response_payload = (response_body or {}).get("data") or response_body or {}
    if not isinstance(response_payload, dict):
        response_payload = {}
    return ApiResponse(data={
        "taskid": response_payload.get("taskid"),
        "diskid": disk_id,
        "type": body.type,
    })


@router.delete("/{instance_id}/snapshots/{snapshot_id}", response_model=ApiResponse)
async def delete_instance_snapshot(
    instance_id: int,
    snapshot_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    res = await db.execute(
        select(Instance).where(
            Instance.id == instance_id,
            Instance.user_id == current_user.id,
        )
    )
    instance = res.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    if not instance.upstream_id or not instance.cloud_server_id:
        raise HTTPException(status_code=400, detail="实例尚未开通，无法删除备份/快照")

    if await zjmf_instance_adapter.get_zjmf_server_for_instance(db, instance):
        return ApiResponse(data=await zjmf_instance_adapter.delete_snapshot(
            db,
            instance,
            snapshot_id=snapshot_id,
        ))

    client, _ = await _get_dcim_client(instance.cloud_server_id, db)
    try:
        code, text, response_body = await client.request(
            f"/snapshots/{snapshot_id}?hostid={instance.upstream_id}",
            method="DELETE",
            timeout=15.0,
        )
    except DcimCloudError as e:
        raise HTTPException(status_code=502, detail=f"上游调用失败：{e}")
    if not (200 <= code < 300):
        raise HTTPException(status_code=502, detail=f"上游返回异常（HTTP {code}）：{text[:200]}")

    response_payload = (response_body or {}).get("data") or response_body or {}
    if not isinstance(response_payload, dict):
        response_payload = {}
    return ApiResponse(data={
        "taskid": response_payload.get("taskid"),
        "snapshot_id": snapshot_id,
    })


@router.post("/{instance_id}/snapshots/{snapshot_id}/restore", response_model=ApiResponse)
async def restore_instance_snapshot(
    instance_id: int,
    snapshot_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    res = await db.execute(
        select(Instance).where(
            Instance.id == instance_id,
            Instance.user_id == current_user.id,
        )
    )
    instance = res.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    if not instance.upstream_id or not instance.cloud_server_id:
        raise HTTPException(status_code=400, detail="实例尚未开通，无法还原备份/快照")

    if await zjmf_instance_adapter.get_zjmf_server_for_instance(db, instance):
        return ApiResponse(data=await zjmf_instance_adapter.restore_snapshot(
            db,
            instance,
            snapshot_id=snapshot_id,
        ))

    client, _ = await _get_dcim_client(instance.cloud_server_id, db)
    try:
        code, text, response_body = await client.request(
            f"/snapshots/{snapshot_id}/restore?hostid={instance.upstream_id}",
            method="POST",
            timeout=30.0,
        )
    except DcimCloudError as e:
        raise HTTPException(status_code=502, detail=f"上游调用失败：{e}")
    if not (200 <= code < 300):
        raise HTTPException(status_code=502, detail=f"上游返回异常（HTTP {code}）：{text[:200]}")

    response_payload = (response_body or {}).get("data") or response_body or {}
    if not isinstance(response_payload, dict):
        response_payload = {}

    os_info = response_payload.get("os")
    if isinstance(os_info, dict):
        os_id = os_info.get("id")
        os_name = os_info.get("name") or os_info.get("filename")
        if os_id is not None and os_name:
            instance.os_image = f"{os_id}|{os_name}"
        elif os_name:
            instance.os_image = str(os_name)
        port = _to_int_or_none(os_info.get("port"))
        if port:
            instance.ssh_port = port
        password = os_info.get("password")
        if password:
            instance.root_password = encrypt_secret(str(password))
        await db.flush()

    return ApiResponse(data={
        "taskid": response_payload.get("taskid"),
        "snapshot_id": snapshot_id,
        "os": os_info if isinstance(os_info, dict) else None,
    })


@router.get("/{instance_id}/tasks", response_model=ApiResponse)
async def list_instance_tasks(
    instance_id: int,
    page: int = 1,
    per_page: int = 20,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """代理上游 /tasks?cloud=<upstream_id>。仅返回当前用户名下实例的任务。"""
    res = await db.execute(
        select(Instance).where(
            Instance.id == instance_id, Instance.user_id == current_user.id
        )
    )
    instance = res.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    if not instance.upstream_id or not instance.cloud_server_id:
        return ApiResponse(data={"list": [], "meta": {"total": 0}})

    if await zjmf_instance_adapter.get_zjmf_server_for_instance(db, instance):
        return ApiResponse(data=await zjmf_instance_adapter.list_tasks(
            db,
            instance,
            page=page,
            per_page=per_page,
        ))

    client, _ = await _get_dcim_client(instance.cloud_server_id, db)
    try:
        code, text, body = await client.request(
            "/tasks",
            method="GET",
            data={
                "cloud": instance.upstream_id,
                "page": page,
                "per_page": per_page,
                "orderby": "id",
                "sort": "desc",
            },
            timeout=8.0,
        )
    except DcimCloudError as e:
        raise HTTPException(status_code=502, detail=f"上游调用失败：{e}")
    if not (200 <= code < 300):
        raise HTTPException(status_code=502, detail=f"上游返回异常（HTTP {code}）：{text[:200]}")

    payload = body or {}
    tasks = payload.get("data") or []
    meta = payload.get("meta") or {}
    return ApiResponse(data={"list": tasks, "meta": meta})


@router.get("/{instance_id}/finance-firewall", response_model=ApiResponse)
async def get_instance_finance_firewall(
    instance_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    res = await db.execute(
        select(Instance).where(
            Instance.id == instance_id,
            Instance.user_id == current_user.id,
        )
    )
    instance = res.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    if not instance.upstream_id or not instance.cloud_server_id:
        return ApiResponse(data={"supported": False, "message": zjmf_instance_adapter.UNSUPPORTED_OPERATION_MESSAGE, "list": []})
    if not await zjmf_instance_adapter.get_zjmf_server_for_instance(db, instance):
        raise HTTPException(status_code=400, detail="当前实例不是魔方财务实例")
    return ApiResponse(data=await zjmf_instance_adapter.get_firewall(db, instance))


@router.get("/{instance_id}/actions", response_model=ApiResponse)
async def list_instance_actions(
    instance_id: int,
    limit: int = Query(50, ge=1, le=200),
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    res = await db.execute(
        select(Instance).where(
            Instance.id == instance_id,
            Instance.user_id == current_user.id,
        )
    )
    instance = res.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")

    result = await db.execute(
        select(UpstreamActionLog)
        .where(UpstreamActionLog.instance_id == instance.id)
        .order_by(UpstreamActionLog.id.desc())
        .limit(limit)
    )
    return ApiResponse(data=[upstream_action_service.serialize_action_log(item) for item in result.scalars().all()])
