"""Admin instance management."""
import asyncio
from dataclasses import asdict, is_dataclass
from datetime import datetime, timedelta, timezone
from typing import Any, Literal

from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel, Field
from sqlalchemy import cast, func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.types import String

from app.config import settings as app_settings
from app.core.crypto import encrypt_secret, decrypt_secret
from app.core.deps import require_role
from app.database import async_session, get_db
from app.models.cloud_server import CloudServer
from app.models.instance import Instance, InstanceStatus, SuspendType
from app.models.order import Order
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.routers.instances import build_plugin_manage_context, PluginManageActionBody
from app.schemas.common import ApiResponse
from app.services import (
    instance_lifecycle_service,
    provision_plugin_service,
    provision_service,
    upstream_action_service,
)
from mode.provision.base import ProvisionPluginError

router = APIRouter()
_admin = require_role("admin", "super_admin")


POWER_ACTION_PATHS = {
    "power-on": "/on",
    "shutdown": "/off",
    "hard-shutdown": "/hardoff",
    "reboot": "/reboot",
}


class AdminInstanceSuspendBody(BaseModel):
    type: Literal["traffic", "due", "manual"] = "manual"
    free_gpu: int = Field(0, ge=0, le=1)
    reason: str | None = Field(None, max_length=255)


class AdminInstanceAutoRenewBody(BaseModel):
    auto_renew: bool


class AdminInstanceBillingBody(BaseModel):
    cycle: str | None = Field(None, min_length=1, max_length=16)
    unit_price: int | None = Field(None, ge=0)
    next_renew_amount: int | None = Field(None, ge=0)
    expires_at: datetime | None = None
    auto_renew: bool | None = None


class AdminInstanceBatchBody(BaseModel):
    ids: list[int] = Field(..., min_length=1, max_length=100)
    type: Literal["traffic", "due", "manual"] = "manual"
    free_gpu: int = Field(0, ge=0, le=1)
    reason: str | None = Field(None, max_length=255)


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


def _upstream_console_url(instance: Instance, cloud_server: CloudServer | None) -> str | None:
    if not cloud_server or not instance.upstream_id:
        return None
    return f"{cloud_server.base_url.rstrip('/')}/#/cloudsHome?id={instance.upstream_id}"


def _user_label(user: User | None) -> str | None:
    if not user:
        return None
    return user.nickname or user.username or user.email or user.phone or f"用户 #{user.id}"


def _serialize_instance(
    instance: Instance,
    product: Product | None,
    group: ProductGroup | None,
    cloud_server: CloudServer | None,
    user: User | None,
) -> dict[str, Any]:
    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.admin_manifest(manifest)
            plugin_manage_actions = [
                item for item in (manifest.get("manage_actions") or []) if isinstance(item, dict)
            ]
            payload = instance.upstream_payload if isinstance(instance.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
    return {
        "id": instance.id,
        "user_id": instance.user_id,
        "order_id": instance.order_id,
        "product_id": instance.product_id,
        "product_name": product.name if product else f"产品 #{instance.product_id}",
        "product_group_name": group.name if group else None,
        "api_type": product.api_type if isinstance(product.api_type, str) else (product.api_type.value if product and product.api_type else None),
        "status": _enum_value(instance.status),
        "suspend_type": _enum_value(instance.suspend_type),
        "hostname": instance.hostname,
        "alias": instance.alias,
        "main_ip": instance.main_ip,
        "ssh_port": instance.ssh_port,
        "os_image": instance.os_image,
        "cpu": instance.cpu,
        "memory": instance.memory,
        "system_disk_size": instance.system_disk_size,
        "data_disk_size": instance.data_disk_size,
        "in_bw": instance.in_bw,
        "out_bw": instance.out_bw,
        "ip_num": instance.ip_num,
        "nat_acl_limit": instance.nat_acl_limit,
        "traffic_quota": instance.traffic_quota,
        "traffic_type": instance.traffic_type,
        "internet": instance.internet,
        "is_locked": instance.is_locked,
        "cycle": instance.cycle,
        "unit_price": instance.unit_price,
        "next_renew_amount": instance.next_renew_amount,
        "auto_renew": instance.auto_renew,
        "expires_at": instance.expires_at.isoformat() if instance.expires_at else None,
        "created_at": instance.created_at.isoformat() if instance.created_at else None,
        "updated_at": instance.updated_at.isoformat() if instance.updated_at else None,
        "terminated_at": instance.terminated_at.isoformat() if instance.terminated_at else None,
        "upstream_id": instance.upstream_id,
        "upstream_task_id": instance.upstream_task_id,
        "upstream_invoice_id": instance.upstream_invoice_id,
        "cloud_server_id": instance.cloud_server_id,
        "cloud_server_name": cloud_server.name if cloud_server else None,
        "cloud_server_base_url": cloud_server.base_url if cloud_server else None,
        "upstream_console_url": _upstream_console_url(instance, cloud_server),
        "provision_attempts": instance.provision_attempts,
        "last_provision_at": instance.last_provision_at.isoformat() if instance.last_provision_at else None,
        "provision_log": instance.provision_log,
        "provision_plugin": provision_plugin,
        "plugin_manage_actions": plugin_manage_actions,
        "plugin_links": plugin_links,
        "user": {
            "id": user.id if user else instance.user_id,
            "username": user.username if user else None,
            "nickname": user.nickname if user else None,
            "email": user.email if user else None,
            "phone": user.phone if user else None,
            "balance": user.balance if user else 0,
            "label": _user_label(user),
        },
    }


def _instance_snapshot_for_provision(order: Order | None, instance: Instance) -> dict:
    if not order or not isinstance(order.items_snapshot, list):
        raise HTTPException(status_code=400, detail="订单快照不存在，无法重新开通")
    for item in order.items_snapshot:
        if not isinstance(item, dict):
            continue
        if int(item.get("product_id") or 0) == int(instance.product_id):
            return item
    raise HTTPException(status_code=400, detail="订单快照未匹配该实例，无法重新开通")


async def _provision_with_new_session(instance_id: int, snap: dict) -> None:
    try:
        async with async_session() as db:
            try:
                await provision_service.provision_instance(instance_id, snap, db)
                await db.commit()
            except Exception:
                await db.rollback()
                raise
    except Exception:
        import logging

        logging.getLogger("qycw.admin.instances").exception(
            "admin provision retry failed: instance=%s", instance_id
        )


async def _get_instance(db: AsyncSession, instance_id: int, *, for_update: bool = False) -> Instance:
    stmt = select(Instance).where(Instance.id == instance_id)
    if for_update:
        stmt = stmt.with_for_update()
    result = await db.execute(stmt)
    instance = result.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    return instance


def _status_counts(rows: list[tuple[Any, int]]) -> dict[str, int]:
    return {_enum_value(status) or "unknown": int(count or 0) for status, count in rows}


def _suspend_type(value: str) -> str:
    if value == "traffic":
        return SuspendType.TRAFFIC.value
    if value == "due":
        return SuspendType.DUE.value
    return SuspendType.MANUAL.value


async def _serialize_rows(db: AsyncSession, rows: list[tuple[Instance, Product | None, ProductGroup | None, CloudServer | None, User | None]]) -> list[dict[str, Any]]:
    return [
        _serialize_instance(instance, product, group, cloud_server, user)
        for instance, product, group, cloud_server, user in rows
    ]


@router.get("", response_model=ApiResponse)
async def list_instances(
    keyword: str | None = Query(None),
    status: str | None = Query(None),
    user_id: int | None = Query(None),
    product_id: int | None = Query(None),
    group_id: int | None = Query(None),
    cloud_server_id: int | None = Query(None),
    has_upstream: bool | None = Query(None),
    auto_renew: bool | None = Query(None),
    expired: bool | None = Query(None),
    due_days: int | None = Query(None, ge=0, le=365),
    order_by: Literal["id", "created_at", "updated_at", "expires_at", "user_id"] = Query("id"),
    sort: Literal["asc", "desc"] = Query("desc"),
    limit: int = Query(50, ge=1, le=300),
    offset: int = Query(0, ge=0),
    db: AsyncSession = Depends(get_db),
    _: User = Depends(_admin),
):
    conditions = []
    now = datetime.now(timezone.utc)
    if status:
        conditions.append(Instance.status == status)
    if user_id:
        conditions.append(Instance.user_id == user_id)
    if product_id:
        conditions.append(Instance.product_id == product_id)
    if group_id:
        conditions.append(Product.group_id == group_id)
    if cloud_server_id:
        conditions.append(Instance.cloud_server_id == cloud_server_id)
    if has_upstream is not None:
        conditions.append(Instance.upstream_id.is_not(None) if has_upstream else Instance.upstream_id.is_(None))
    if auto_renew is not None:
        conditions.append(Instance.auto_renew == (1 if auto_renew else 0))
    if expired is not None:
        conditions.append(Instance.expires_at <= now if expired else or_(Instance.expires_at.is_(None), Instance.expires_at > now))
    if due_days is not None:
        conditions.extend([
            Instance.expires_at.is_not(None),
            Instance.expires_at > now,
            Instance.expires_at <= now + timedelta(days=due_days),
        ])
    if keyword:
        like = f"%{keyword.strip()}%"
        conditions.append(or_(
            cast(Instance.id, String).like(like),
            cast(Instance.upstream_id, String).like(like),
            Instance.hostname.like(like),
            Instance.alias.like(like),
            Instance.main_ip.like(like),
            Product.name.like(like),
            Product.sku.like(like),
            User.username.like(like),
            User.nickname.like(like),
            User.email.like(like),
            User.phone.like(like),
        ))

    order_column = {
        "id": Instance.id,
        "created_at": Instance.created_at,
        "updated_at": Instance.updated_at,
        "expires_at": Instance.expires_at,
        "user_id": Instance.user_id,
    }[order_by]
    order_expr = order_column.desc() if sort == "desc" else order_column.asc()

    base_from = (
        select(Instance, Product, ProductGroup, CloudServer, User)
        .outerjoin(Product, Product.id == Instance.product_id)
        .outerjoin(ProductGroup, ProductGroup.id == Product.group_id)
        .outerjoin(CloudServer, CloudServer.id == Instance.cloud_server_id)
        .outerjoin(User, User.id == Instance.user_id)
        .where(*conditions)
    )
    rows_result = await db.execute(base_from.order_by(order_expr).limit(limit).offset(offset))
    rows = list(rows_result.all())

    total_result = await db.execute(
        select(func.count(Instance.id))
        .outerjoin(Product, Product.id == Instance.product_id)
        .outerjoin(ProductGroup, ProductGroup.id == Product.group_id)
        .outerjoin(CloudServer, CloudServer.id == Instance.cloud_server_id)
        .outerjoin(User, User.id == Instance.user_id)
        .where(*conditions)
    )
    total = int(total_result.scalar() or 0)

    summary_result = await db.execute(
        select(Instance.status, func.count(Instance.id))
        .outerjoin(Product, Product.id == Instance.product_id)
        .outerjoin(ProductGroup, ProductGroup.id == Product.group_id)
        .outerjoin(CloudServer, CloudServer.id == Instance.cloud_server_id)
        .outerjoin(User, User.id == Instance.user_id)
        .where(*conditions)
        .group_by(Instance.status)
    )
    summary = _status_counts(list(summary_result.all()))

    return ApiResponse(data={
        "items": await _serialize_rows(db, rows),
        "total": total,
        "limit": limit,
        "offset": offset,
        "summary": summary,
    })


@router.post("/batch/{action}", response_model=ApiResponse)
async def batch_operate_instances(
    action: Literal["power-on", "shutdown", "hard-shutdown", "reboot", "suspend", "unsuspend", "delete", "provision"],
    body: AdminInstanceBatchBody,
    db: AsyncSession = Depends(get_db),
    current: User = Depends(_admin),
):
    ids = list(dict.fromkeys(int(item) for item in body.ids))
    results: list[dict[str, Any]] = []

    for instance_id in ids:
        try:
            instance = await _get_instance(db, instance_id, for_update=True)
            data: dict[str, Any]
            if action in POWER_ACTION_PATHS:
                payload = await instance_lifecycle_service.call_instance_cloud(
                    db,
                    instance,
                    POWER_ACTION_PATHS[action],
                    "POST",
                )
                data = {"taskid": payload.get("taskid"), "instance_id": instance.id}
            elif action == "suspend":
                if instance.status != InstanceStatus.ACTIVE:
                    raise HTTPException(status_code=400, detail="当前实例状态不支持暂停")
                data = await instance_lifecycle_service.suspend_instance(
                    db,
                    instance,
                    suspend_type=_suspend_type(body.type),
                    free_gpu=body.free_gpu,
                    reason=body.reason or "管理员批量暂停实例",
                    source="admin",
                    user_id=current.id,
                )
            elif action == "unsuspend":
                if instance.status != InstanceStatus.SUSPENDED:
                    raise HTTPException(status_code=400, detail="当前实例状态不需要解除暂停")
                data = await instance_lifecycle_service.unsuspend_instance(
                    db,
                    instance,
                    reason=body.reason or "管理员批量解除暂停",
                    source="admin",
                    user_id=current.id,
                )
            elif action == "delete":
                if instance.status == InstanceStatus.TERMINATED:
                    data = {"taskid": None, "instance_id": instance.id}
                else:
                    data = await instance_lifecycle_service.force_delete_instance(
                        db,
                        instance,
                        reason=body.reason or "管理员批量删除实例，不退款",
                        source="admin",
                        user_id=current.id,
                    )
            else:
                if instance.status not in (InstanceStatus.PROVISIONING, InstanceStatus.PROVISION_FAILED):
                    raise HTTPException(status_code=400, detail="仅开通中或开通失败的实例可触发开通")
                order = await db.get(Order, instance.order_id)
                snap = _instance_snapshot_for_provision(order, instance)
                instance.status = InstanceStatus.PROVISIONING
                instance.provision_log = body.reason or "管理员批量触发重新开通"
                await db.flush()
                data = {"instance_id": instance.id}
                await db.commit()
                asyncio.create_task(_provision_with_new_session(instance.id, snap))
                results.append({"id": instance.id, "success": True, "data": data})
                continue

            await db.commit()
            results.append({"id": instance.id, "success": True, "data": data})
        except HTTPException as exc:
            await db.rollback()
            results.append({"id": instance_id, "success": False, "error": exc.detail})
        except Exception as exc:
            await db.rollback()
            results.append({"id": instance_id, "success": False, "error": str(exc)})

    success = sum(1 for item in results if item["success"])
    failed = len(results) - success
    return ApiResponse(data={"success": success, "failed": failed, "results": results})


@router.get("/{instance_id}", response_model=ApiResponse)
async def get_instance(
    instance_id: int,
    db: AsyncSession = Depends(get_db),
    _: User = Depends(_admin),
):
    row = await db.execute(
        select(Instance, Product, ProductGroup, CloudServer, User)
        .outerjoin(Product, Product.id == Instance.product_id)
        .outerjoin(ProductGroup, ProductGroup.id == Product.group_id)
        .outerjoin(CloudServer, CloudServer.id == Instance.cloud_server_id)
        .outerjoin(User, User.id == Instance.user_id)
        .where(Instance.id == instance_id)
    )
    result = row.first()
    if not result:
        raise HTTPException(status_code=404, detail="实例不存在")
    instance, product, group, cloud_server, user = result
    data = _serialize_instance(instance, product, group, cloud_server, user)

    # 插件型产品：解密 root_password 返回
    if instance.root_password:
        try:
            data["root_password"] = decrypt_secret(instance.root_password)
        except Exception:
            pass

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


@router.post("/{instance_id}/power/{action}", response_model=ApiResponse)
async def operate_instance_power(
    instance_id: int,
    action: Literal["power-on", "shutdown", "hard-shutdown", "reboot"],
    db: AsyncSession = Depends(get_db),
    _: User = Depends(_admin),
):
    instance = await _get_instance(db, instance_id)
    payload = await instance_lifecycle_service.call_instance_cloud(db, instance, POWER_ACTION_PATHS[action], "POST")
    return ApiResponse(data={"taskid": payload.get("taskid"), "instance_id": instance.id}, message="指令已下发")


@router.post("/{instance_id}/plugin-actions/{action}", response_model=ApiResponse)
async def admin_run_plugin_action(
    instance_id: int,
    action: str,
    body: PluginManageActionBody | None = None,
    db: AsyncSession = Depends(get_db),
    _current: User = Depends(_admin),
):
    instance = await _get_instance(db, instance_id)
    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}")
    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)
    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: AdminInstanceSuspendBody,
    db: AsyncSession = Depends(get_db),
    current: User = Depends(_admin),
):
    instance = await _get_instance(db, instance_id, for_update=True)
    if instance.status != InstanceStatus.ACTIVE:
        raise HTTPException(status_code=400, detail="当前实例状态不支持暂停")
    data = await instance_lifecycle_service.suspend_instance(
        db,
        instance,
        suspend_type=_suspend_type(body.type),
        free_gpu=body.free_gpu,
        reason=body.reason or "管理员手动暂停实例",
        source="admin",
        user_id=current.id,
    )
    return ApiResponse(data=data, message="暂停任务已下发")


@router.post("/{instance_id}/unsuspend", response_model=ApiResponse)
async def unsuspend_instance(
    instance_id: int,
    db: AsyncSession = Depends(get_db),
    current: User = Depends(_admin),
):
    instance = await _get_instance(db, instance_id, for_update=True)
    if instance.status != InstanceStatus.SUSPENDED:
        raise HTTPException(status_code=400, detail="当前实例状态不需要解除暂停")
    data = await instance_lifecycle_service.unsuspend_instance(
        db,
        instance,
        reason="管理员手动解除暂停",
        source="admin",
        user_id=current.id,
    )
    return ApiResponse(data=data, message="解除暂停任务已下发")


@router.delete("/{instance_id}", response_model=ApiResponse)
async def delete_instance(
    instance_id: int,
    db: AsyncSession = Depends(get_db),
    current: User = Depends(_admin),
):
    instance = await _get_instance(db, instance_id, for_update=True)
    if instance.status == InstanceStatus.TERMINATED:
        return ApiResponse(data={"taskid": None, "instance_id": instance.id}, message="实例已销毁")
    data = await instance_lifecycle_service.force_delete_instance(
        db,
        instance,
        reason="管理员删除实例，不退款",
        source="admin",
        user_id=current.id,
    )
    return ApiResponse(data=data, message="删除实例任务已下发")


@router.patch("/{instance_id}/auto-renew", response_model=ApiResponse)
async def update_instance_auto_renew(
    instance_id: int,
    body: AdminInstanceAutoRenewBody,
    db: AsyncSession = Depends(get_db),
    _: User = Depends(_admin),
):
    instance = await _get_instance(db, instance_id, for_update=True)
    if instance.status == InstanceStatus.TERMINATED:
        raise HTTPException(status_code=400, detail="已销毁实例不能修改自动续费")
    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.patch("/{instance_id}/billing", response_model=ApiResponse)
async def update_instance_billing(
    instance_id: int,
    body: AdminInstanceBillingBody,
    db: AsyncSession = Depends(get_db),
    _: User = Depends(_admin),
):
    instance = await _get_instance(db, instance_id, for_update=True)
    if instance.status == InstanceStatus.TERMINATED:
        raise HTTPException(status_code=400, detail="已销毁实例不能修改计费信息")

    update_data = body.model_dump(exclude_unset=True)
    if "cycle" in update_data and update_data["cycle"] is not None:
        cycle = str(update_data["cycle"]).strip()
        if not cycle:
            raise HTTPException(status_code=400, detail="计费周期不能为空")
        instance.cycle = cycle
    if "unit_price" in update_data and update_data["unit_price"] is not None:
        instance.unit_price = int(update_data["unit_price"])
    if "next_renew_amount" in update_data and update_data["next_renew_amount"] is not None:
        instance.next_renew_amount = int(update_data["next_renew_amount"])
    if "expires_at" in update_data:
        instance.expires_at = update_data["expires_at"]
    if "auto_renew" in update_data and update_data["auto_renew"] is not None:
        instance.auto_renew = 1 if bool(update_data["auto_renew"]) else 0

    await db.flush()
    return ApiResponse(data={
        "id": instance.id,
        "cycle": instance.cycle,
        "unit_price": instance.unit_price,
        "next_renew_amount": instance.next_renew_amount,
        "expires_at": instance.expires_at.isoformat() if instance.expires_at else None,
        "auto_renew": instance.auto_renew,
    })


@router.post("/{instance_id}/provision", response_model=ApiResponse)
async def provision_instance(
    instance_id: int,
    db: AsyncSession = Depends(get_db),
    _: User = Depends(_admin),
):
    instance = await _get_instance(db, instance_id, for_update=True)
    if instance.status not in (InstanceStatus.PROVISIONING, InstanceStatus.PROVISION_FAILED):
        raise HTTPException(status_code=400, detail="仅开通中或开通失败的实例可触发开通")
    order = await db.get(Order, instance.order_id)
    snap = _instance_snapshot_for_provision(order, instance)
    instance.status = InstanceStatus.PROVISIONING
    instance.provision_log = "管理员触发重新开通"
    await db.flush()
    await db.commit()
    asyncio.create_task(_provision_with_new_session(instance.id, snap))
    return ApiResponse(data={"instance_id": instance.id}, message="已触发开通任务")
