"""后台用户管理路由。"""
import asyncio
from datetime import datetime, timezone
from typing import Any, Literal

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

from app.config import settings
from app.core import sensitive
from app.core.deps import require_role
from app.core.security import create_token
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.agent_tier import AgentTier
from app.models.member_level import MemberLevel
from app.models.order import Order
from app.models.product import Product
from app.models.product_group import ProductGroup
from app.models.user import User, UserRole, UserStatus
from app.models.user_agent import AgentStatus, UserAgent
from app.models.user_invite_code import UserInviteCode
from app.models.user_invite_code_log import UserInviteCodeLog
from app.models.user_refund_quota_log import UserRefundQuotaLog
from app.models.ticket import Ticket, TicketStatus
from app.schemas.admin_user import (
    AdminBalanceAdjust,
    AdminInviteCodeQuotaAdjust,
    AdminInviteCodeQuotaReset,
    AdminRefundQuotaAdjust,
    AdminRefundQuotaReset,
    AdminUserDetail,
    AdminUserListItem,
    AdminUserPasswordReset,
    AdminUserUpdate,
    AdminVerificationQuotaAdjust,
    AdminVerificationQuotaReset,
    ImpersonateResponse,
)
from app.schemas.common import ApiResponse
from app.services import (
    auth_service,
    instance_lifecycle_service,
    membership_service,
    provision_service,
    session_service,
    system_setting_service,
    agent_service,
)
from app.models.balance_log import BalanceChangeType, BalanceLog
from app.models.user_verification_quota import UserVerificationQuota
from app.models.user_verification_record import UserVerificationRecord
from app.services import verification_service

router = APIRouter()
_admin = require_role("admin", "super_admin")
_super = require_role("super_admin")
MEMBERSHIP_UPDATE_FIELDS = {"member_level_id", "member_level_locked", "agent_tier"}


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


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 AdminInstanceNatAclCreateBody(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]


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


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.user").exception("admin provision retry failed: instance=%s", instance_id)


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


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 _serialize_admin_instance(
    instance: Instance,
    product: Product | None,
    group: ProductGroup | None,
    cloud_server: CloudServer | None = None,
) -> dict:
    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,
        "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,
        "upstream_id": instance.upstream_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),
        "nat_acl_limit": instance.nat_acl_limit,
        "ip_num": instance.ip_num,
        "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,
        "terminated_at": instance.terminated_at.isoformat() if instance.terminated_at else None,
        "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,
    }


async def _serialize_admin_user_detail(db: AsyncSession, user: User) -> AdminUserDetail:
    from app.routers import orders as order_helpers

    membership = await membership_service.get_or_create_membership(db, user.id)
    data = AdminUserDetail.model_validate(user).model_dump()
    data["real_name"] = sensitive.decrypt_optional(user.real_name)
    data["real_name_full"] = sensitive.decrypt_optional(user.real_name)
    data["id_card_no"] = sensitive.decrypt_optional(user.id_card_no)
    data["id_card_no_full"] = sensitive.decrypt_optional(user.id_card_no)
    data["id_card_front_image"] = user.id_card_front_image
    data["id_card_back_image"] = user.id_card_back_image
    data["id_card_face_image"] = user.id_card_face_image
    data["id_card_verify_remark"] = user.id_card_verify_remark
    data["membership"] = membership
    settings = await system_setting_service.get_business_settings(db)
    extra = max(0, int(getattr(user, "full_refund_quota_extra", 0) or 0))
    limit = int(settings.full_refund_limit) + extra
    reset_at = getattr(user, "full_refund_quota_reset_at", None)
    used = await order_helpers._full_refund_count(db, user.id, reset_at)
    unlimited = await order_helpers._is_active_agent(db, user.id)
    data["refund_quota"] = {
        "base_limit": int(settings.full_refund_limit),
        "extra": extra,
        "limit": limit,
        "used": used,
        "remaining": None if unlimited else max(0, limit - used),
        "unlimited": unlimited,
        "reset_at": reset_at,
    }
    invite_extra = max(0, int(getattr(user, "invite_code_change_extra", 0) or 0))
    invite_used = int(getattr(user, "invite_code_change_count", 0) or 0)
    invite_limit = agent_service.invite_code_change_limit(user)
    data["invite_code_quota"] = {
        "current_code": user.invite_code,
        "base_limit": agent_service.INVITE_CODE_CHANGE_LIMIT,
        "extra": invite_extra,
        "limit": invite_limit,
        "used": invite_used,
        "remaining": max(0, invite_limit - invite_used),
    }
    code_rows = await db.execute(
        select(UserInviteCode)
        .where(UserInviteCode.user_id == user.id)
        .order_by(UserInviteCode.is_current.desc(), UserInviteCode.id.desc())
    )
    data["invite_codes"] = list(code_rows.scalars().all())
    log_rows = await db.execute(
        select(UserInviteCodeLog)
        .where(UserInviteCodeLog.user_id == user.id)
        .order_by(UserInviteCodeLog.id.desc())
        .limit(50)
    )
    data["invite_code_logs"] = list(log_rows.scalars().all())
    verification_config = await system_setting_service.get_verification_provider_config(db)
    data["verification_quota"] = await verification_service.get_verification_quota(db, user, verification_config)
    verification_rows = await db.execute(
        select(UserVerificationRecord)
        .where(UserVerificationRecord.user_id == user.id)
        .order_by(UserVerificationRecord.id.desc())
        .limit(20)
    )
    data["verification_records"] = [
        {
            "id": record.id,
            "provider": record.provider,
            "mode": record.mode,
            "status": record.status,
            "provider_task_id": record.provider_task_id,
            "real_name": sensitive.decrypt_optional(record.real_name),
            "id_card_no": sensitive.decrypt_optional(record.id_card_no),
            "id_card_front_image": record.id_card_front_image,
            "id_card_back_image": record.id_card_back_image,
            "id_card_face_image": record.id_card_face_image,
            "reject_reason": record.reject_reason,
            "provider_response": record.provider_response,
            "material_expires_at": record.material_expires_at,
            "material_cleaned": record.material_cleaned,
            "verified_at": record.verified_at,
            "rejected_at": record.rejected_at,
            "created_at": record.created_at,
        }
        for record in verification_rows.scalars().all()
    ]
    instance_total = int(await db.scalar(select(func.count()).select_from(Instance).where(Instance.user_id == user.id)) or 0)
    instance_active = int(
        await db.scalar(
            select(func.count()).select_from(Instance).where(Instance.user_id == user.id, Instance.status == InstanceStatus.ACTIVE)
        )
        or 0
    )
    instance_suspended = int(
        await db.scalar(
            select(func.count()).select_from(Instance).where(Instance.user_id == user.id, Instance.status == InstanceStatus.SUSPENDED)
        )
        or 0
    )
    order_total = int(await db.scalar(select(func.count()).select_from(Order).where(Order.user_id == user.id)) or 0)
    order_paid_amount = int(
        await db.scalar(
            select(func.coalesce(func.sum(Order.total_amount - Order.discount_amount), 0)).where(Order.user_id == user.id)
        )
        or 0
    )
    balance_logs_total = int(await db.scalar(select(func.count()).select_from(BalanceLog).where(BalanceLog.user_id == user.id)) or 0)
    verification_records_total = int(
        await db.scalar(select(func.count()).select_from(UserVerificationRecord).where(UserVerificationRecord.user_id == user.id))
        or 0
    )
    ticket_rows = await db.execute(
        select(Ticket)
        .where(Ticket.user_id == user.id)
        .order_by(Ticket.last_message_at.desc(), Ticket.id.desc())
        .limit(10)
    )
    tickets = list(ticket_rows.scalars().all())
    data["tickets"] = [
        {
            "id": ticket.id,
            "ticket_no": ticket.ticket_no,
            "subject": ticket.subject,
            "status": _enum_value(ticket.status) or "",
            "priority": _enum_value(ticket.priority) or "",
            "department_name": ticket.department_name,
            "instance_id": ticket.instance_id,
            "assignee_id": ticket.assignee_id,
            "admin_unread_count": ticket.admin_unread_count,
            "user_unread_count": ticket.user_unread_count,
            "satisfaction_rating": ticket.satisfaction_rating,
            "last_message_at": ticket.last_message_at,
            "closed_at": ticket.closed_at,
            "created_at": ticket.created_at,
        }
        for ticket in tickets
    ]
    tickets_total = int(await db.scalar(select(func.count()).select_from(Ticket).where(Ticket.user_id == user.id)) or 0)
    tickets_open = int(
        await db.scalar(
            select(func.count())
            .select_from(Ticket)
            .where(
                Ticket.user_id == user.id,
                Ticket.status.in_(
                    [
                        TicketStatus.OPEN,
                        TicketStatus.PENDING,
                        TicketStatus.PROCESSING,
                        TicketStatus.WAITING_USER,
                        TicketStatus.ANSWERED,
                    ]
                ),
            )
        )
        or 0
    )
    tickets_unread = int(
        await db.scalar(select(func.coalesce(func.sum(Ticket.admin_unread_count), 0)).where(Ticket.user_id == user.id))
        or 0
    )
    data["stats"] = {
        "instances_total": instance_total,
        "instances_active": instance_active,
        "instances_suspended": instance_suspended,
        "orders_total": order_total,
        "orders_paid_amount": order_paid_amount,
        "balance_logs_total": balance_logs_total,
        "verification_records_total": verification_records_total,
        "tickets_total": tickets_total,
        "tickets_open": tickets_open,
        "tickets_unread": tickets_unread,
    }
    return AdminUserDetail.model_validate(data)


def _required_reason(reason: str) -> str:
    text = str(reason or "").strip()
    if not text:
        raise HTTPException(status_code=400, detail="请填写调整原因")
    return text[:200]


def _add_refund_quota_log(
    db: AsyncSession,
    *,
    user: User,
    operator: User | None,
    action: str,
    count: int,
    before_extra: int,
    after_extra: int,
    before_reset_at: datetime | None,
    after_reset_at: datetime | None,
    reason: str,
) -> None:
    db.add(UserRefundQuotaLog(
        user_id=user.id,
        operator_id=operator.id if operator else None,
        action=action,
        count=count,
        before_extra=before_extra,
        after_extra=after_extra,
        before_reset_at=before_reset_at,
        after_reset_at=after_reset_at,
        reason=reason,
    ))


def _add_invite_code_log(
    db: AsyncSession,
    *,
    user: User,
    operator: User | None,
    action: str,
    count: int,
    before_used: int,
    after_used: int,
    before_extra: int,
    after_extra: int,
    reason: str,
) -> None:
    db.add(UserInviteCodeLog(
        user_id=user.id,
        operator_id=operator.id if operator else None,
        action=action,
        count=count,
        before_used=before_used,
        after_used=after_used,
        before_extra=before_extra,
        after_extra=after_extra,
        reason=reason,
    ))


async def _get_or_create_verification_quota(db: AsyncSession, user_id: int) -> UserVerificationQuota:
    result = await db.execute(
        select(UserVerificationQuota)
        .where(UserVerificationQuota.user_id == user_id)
        .with_for_update()
    )
    quota = result.scalar_one_or_none()
    if quota:
        return quota
    quota = UserVerificationQuota(user_id=user_id)
    db.add(quota)
    await db.flush()
    return quota


@router.get("", response_model=ApiResponse)
async def list_users(
    keyword: str | None = Query(None, description="按用户名/邮箱/手机号模糊搜索"),
    role: str | None = Query(None),
    status: str | None = Query(None),
    limit: int = Query(100, ge=1, le=500),
    offset: int = Query(0, ge=0),
    db: AsyncSession = Depends(get_db),
    _: User = Depends(_admin),
):
    stmt = select(User).order_by(User.id.desc()).limit(limit).offset(offset)
    if keyword:
        like = f"%{keyword}%"
        clauses = [User.username.like(like), User.email.like(like), User.phone.like(like)]
        id_hash = sensitive.deterministic_hash(keyword)
        if id_hash:
            clauses.append(User.id_card_no_hash == id_hash)
            clauses.append(User.id_card_no == keyword.strip().upper())
        stmt = stmt.where(or_(*clauses))
    if role:
        stmt = stmt.where(User.role == role)
    if status:
        stmt = stmt.where(User.status == status)
    res = await db.execute(stmt)
    users = res.scalars().all()
    items = []
    for item in users:
        data = AdminUserListItem.model_validate(item).model_dump()
        data["real_name"] = sensitive.mask_name(sensitive.decrypt_optional(item.real_name))
        items.append(data)
    return ApiResponse(data=items)


@router.get("/{user_id}", response_model=ApiResponse)
async def get_user(
    user_id: int,
    db: AsyncSession = Depends(get_db),
    _: User = Depends(_admin),
):
    res = await db.execute(select(User).where(User.id == user_id))
    user = res.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    return ApiResponse(data=await _serialize_admin_user_detail(db, user))


@router.get("/{user_id}/instances", response_model=ApiResponse)
async def list_user_instances(
    user_id: int,
    db: AsyncSession = Depends(get_db),
    _: User = Depends(_admin),
):
    rows = await db.execute(
        select(Instance, Product, ProductGroup, CloudServer)
        .outerjoin(Product, Product.id == Instance.product_id)
        .outerjoin(ProductGroup, ProductGroup.id == Product.group_id)
        .outerjoin(CloudServer, CloudServer.id == Instance.cloud_server_id)
        .where(Instance.user_id == user_id)
        .order_by(Instance.id.desc())
    )
    return ApiResponse(data=[
        _serialize_admin_instance(instance, product, group, cloud_server)
        for instance, product, group, cloud_server in rows.all()
    ])


@router.get("/{user_id}/instances/{instance_id}", response_model=ApiResponse)
async def get_user_instance(
    user_id: int,
    instance_id: int,
    db: AsyncSession = Depends(get_db),
    _: User = Depends(_admin),
):
    row = await db.execute(
        select(Instance, Product, ProductGroup, CloudServer)
        .outerjoin(Product, Product.id == Instance.product_id)
        .outerjoin(ProductGroup, ProductGroup.id == Product.group_id)
        .outerjoin(CloudServer, CloudServer.id == Instance.cloud_server_id)
        .where(Instance.id == instance_id, Instance.user_id == user_id)
    )
    result = row.first()
    if not result:
        raise HTTPException(status_code=404, detail="实例不存在")
    instance, product, group, cloud_server = result
    user = await db.get(User, user_id)
    data = _serialize_admin_instance(instance, product, group, cloud_server)
    data["user"] = {
        "id": user.id if user else 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,
    }
    return ApiResponse(data=data)


@router.get("/{user_id}/instances/{instance_id}/nat-acl", response_model=ApiResponse)
async def list_user_instance_nat_acl(
    user_id: int,
    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"),
    db: AsyncSession = Depends(get_db),
    _: User = Depends(_admin),
):
    instance = await _get_user_instance(db, user_id, instance_id)
    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("/{user_id}/instances/{instance_id}/nat-acl", response_model=ApiResponse)
async def create_user_instance_nat_acl(
    user_id: int,
    instance_id: int,
    body: AdminInstanceNatAclCreateBody,
    db: AsyncSession = Depends(get_db),
    _: User = Depends(_admin),
):
    instance = await _get_user_instance(db, user_id, instance_id)
    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("/{user_id}/instances/{instance_id}/nat-acl/{nat_acl_id}", response_model=ApiResponse)
async def delete_user_instance_nat_acl(
    user_id: int,
    instance_id: int,
    nat_acl_id: int,
    db: AsyncSession = Depends(get_db),
    _: User = Depends(_admin),
):
    instance = await _get_user_instance(db, user_id, instance_id)
    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("/{user_id}/instances/{instance_id}/power/{action}", response_model=ApiResponse)
async def operate_user_instance_power(
    user_id: int,
    instance_id: int,
    action: Literal["power-on", "shutdown", "hard-shutdown", "reboot"],
    db: AsyncSession = Depends(get_db),
    _: User = Depends(_admin),
):
    instance = await _get_user_instance(db, user_id, instance_id)
    path_map = {
        "power-on": "/on",
        "shutdown": "/off",
        "hard-shutdown": "/hardoff",
        "reboot": "/reboot",
    }
    payload = await instance_lifecycle_service.call_instance_cloud(db, instance, path_map[action], "POST")
    return ApiResponse(data={"taskid": payload.get("taskid"), "instance_id": instance.id})


@router.post("/{user_id}/instances/{instance_id}/suspend", response_model=ApiResponse)
async def admin_suspend_user_instance(
    user_id: int,
    instance_id: int,
    body: AdminInstanceSuspendBody,
    db: AsyncSession = Depends(get_db),
    current: User = Depends(_admin),
):
    instance = await _get_user_instance(db, user_id, instance_id, for_update=True)
    if instance.status != InstanceStatus.ACTIVE:
        raise HTTPException(status_code=400, detail="当前实例状态不支持暂停")
    suspend_type = SuspendType.DUE.value if body.type == "due" else (
        SuspendType.TRAFFIC.value if body.type == "traffic" else SuspendType.MANUAL.value
    )
    data = await instance_lifecycle_service.suspend_instance(
        db,
        instance,
        suspend_type=suspend_type,
        free_gpu=body.free_gpu,
        reason="管理员手动暂停实例",
        source="admin",
        user_id=current.id,
    )
    return ApiResponse(data=data)


@router.post("/{user_id}/instances/{instance_id}/unsuspend", response_model=ApiResponse)
async def admin_unsuspend_user_instance(
    user_id: int,
    instance_id: int,
    db: AsyncSession = Depends(get_db),
    current: User = Depends(_admin),
):
    instance = await _get_user_instance(db, user_id, 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)


@router.delete("/{user_id}/instances/{instance_id}", response_model=ApiResponse)
async def admin_delete_user_instance(
    user_id: int,
    instance_id: int,
    db: AsyncSession = Depends(get_db),
    current: User = Depends(_admin),
):
    instance = await _get_user_instance(db, user_id, instance_id, for_update=True)
    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="admin",
        user_id=current.id,
    )
    return ApiResponse(data=data, message="删除实例任务已下发")


@router.patch("/{user_id}/instances/{instance_id}/auto-renew", response_model=ApiResponse)
async def admin_update_instance_auto_renew(
    user_id: int,
    instance_id: int,
    body: AdminInstanceAutoRenewBody,
    db: AsyncSession = Depends(get_db),
    _: User = Depends(_admin),
):
    instance = await _get_user_instance(db, user_id, 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("/{user_id}/instances/{instance_id}/billing", response_model=ApiResponse)
async def admin_update_instance_billing(
    user_id: int,
    instance_id: int,
    body: AdminInstanceBillingBody,
    db: AsyncSession = Depends(get_db),
    _: User = Depends(_admin),
):
    instance = await _get_user_instance(db, user_id, 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("/{user_id}/instances/{instance_id}/provision", response_model=ApiResponse)
async def admin_provision_user_instance(
    user_id: int,
    instance_id: int,
    db: AsyncSession = Depends(get_db),
    _: User = Depends(_admin),
):
    instance = await _get_user_instance(db, user_id, 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="已触发开通任务")


@router.patch("/{user_id}", response_model=ApiResponse)
async def update_user(
    user_id: int,
    data: AdminUserUpdate,
    db: AsyncSession = Depends(get_db),
    current: User = Depends(_admin),
):
    res = await db.execute(select(User).where(User.id == user_id))
    user = res.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")

    update_data = data.model_dump(exclude_unset=True)
    membership_update = {
        key: update_data.pop(key)
        for key in list(update_data.keys())
        if key in MEMBERSHIP_UPDATE_FIELDS
    }
    # 角色变更需要 super_admin
    if "role" in update_data and current.role != UserRole.SUPER_ADMIN:
        raise HTTPException(status_code=403, detail="仅超级管理员可修改用户角色")
    # 禁止操作自己
    if user.id == current.id and ("role" in update_data or "status" in update_data):
        raise HTTPException(status_code=400, detail="不能修改自己的角色或状态")

    if "real_name" in update_data and sensitive.is_masked(update_data.get("real_name")):
        update_data.pop("real_name")

    if "id_card_no" in update_data and sensitive.is_masked(update_data.get("id_card_no")):
        update_data.pop("id_card_no")

    if update_data.get("id_card_no"):
        plain_id_card = str(update_data["id_card_no"]).strip().upper()
        id_hash = sensitive.deterministic_hash(plain_id_card)
        exists = await db.scalar(
            select(User.id).where(
                User.id != user.id,
                or_(
                    User.id_card_no_hash == id_hash,
                    User.id_card_no == plain_id_card,
                ),
            )
        )
        if exists:
            raise HTTPException(status_code=400, detail="身份证号已被其他用户使用")
        update_data["id_card_no"] = sensitive.encrypt_optional(plain_id_card)
        update_data["id_card_no_hash"] = id_hash
    elif "id_card_no" in update_data:
        update_data["id_card_no_hash"] = None

    if update_data.get("real_name"):
        update_data["real_name"] = sensitive.encrypt_optional(update_data["real_name"])

    for k_, v in update_data.items():
        setattr(user, k_, v)

    # 当管理员直接设置实名认证为已通过时，自动记录时间戳
    if update_data.get("id_card_verified") == "verified" and not user.id_card_verified_at:
        user.id_card_verified_at = datetime.now(timezone.utc)
    elif update_data.get("id_card_verified") == "unverified":
        user.id_card_verified_at = None
        user.id_card_verify_remark = None
        user.real_name = None
        user.id_card_no = None
        user.id_card_no_hash = None

    if membership_update:
        membership = await membership_service.get_or_create_membership(db, user.id)
        if "member_level_id" in membership_update:
            member_level_id = membership_update["member_level_id"]
            if member_level_id is not None:
                level = await db.scalar(
                    select(MemberLevel).where(MemberLevel.id == member_level_id, MemberLevel.enabled == True)  # noqa: E712
                )
                if not level:
                    raise HTTPException(status_code=404, detail="会员等级不存在或已停用")
            membership.member_level_id = member_level_id
            membership.member_level_locked = True
            membership.upgraded_at = None if member_level_id is None else datetime.now()
        if "member_level_locked" in membership_update:
            membership.member_level_locked = bool(membership_update["member_level_locked"])
        if "agent_tier" in membership_update:
            membership.agent_tier = membership_update["agent_tier"]
            await _sync_user_agent_tier(db, user.id, membership_update["agent_tier"], current.id)

    await db.flush()
    await db.refresh(user)
    return ApiResponse(data=await _serialize_admin_user_detail(db, user))


async def _sync_user_agent_tier(db: AsyncSession, user_id: int, tier_code: str | None, admin_id: int) -> None:
    """同步用户代理身份：设置档位则激活/更新代理，清空则终止代理。"""
    agent_result = await db.execute(select(UserAgent).where(UserAgent.user_id == user_id))
    agent = agent_result.scalar_one_or_none()

    if not tier_code:
        if agent:
            agent.status = AgentStatus.TERMINATED
            agent.agent_tier_id = None
        return

    tier = await db.scalar(select(AgentTier).where(AgentTier.code == tier_code))
    if not tier:
        raise HTTPException(status_code=400, detail=f"代理档位 {tier_code} 不存在")

    if not agent:
        agent = UserAgent(user_id=user_id)
        db.add(agent)
    agent.status = AgentStatus.ACTIVE
    agent.agent_tier_id = tier.id
    agent.apply_tier = tier.code
    if not agent.approved_at:
        agent.approved_at = datetime.now()
        agent.approved_by = admin_id


@router.post("/{user_id}/reset-password", response_model=ApiResponse)
async def reset_user_password(
    user_id: int,
    data: AdminUserPasswordReset,
    db: AsyncSession = Depends(get_db),
    _: User = Depends(_admin),
):
    res = await db.execute(select(User).where(User.id == user_id))
    user = res.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    user.hashed_password = auth_service.hash_password(data.new_password)
    await db.flush()
    # 重置密码后，强制踢下线所有会话
    await session_service.revoke_all_refresh(user.id)
    return ApiResponse(data=None, message="密码已重置")


@router.post("/{user_id}/adjust-balance", response_model=ApiResponse)
async def adjust_user_balance(
    user_id: int,
    data: AdminBalanceAdjust,
    db: AsyncSession = Depends(get_db),
    current: User = Depends(_admin),
):
    res = await db.execute(select(User).where(User.id == user_id).with_for_update())
    user = res.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    new_balance = user.balance + data.delta
    if new_balance < 0:
        raise HTTPException(status_code=400, detail="余额不足，无法扣减")
    user.balance = new_balance
    db.add(BalanceLog(
        user_id=user.id,
        change_type=BalanceChangeType.ADMIN_ADJUST,
        delta=data.delta,
        balance_after=new_balance,
        operator_id=current.id,
        remark=data.reason or ("管理员充值" if data.delta > 0 else "管理员扣款"),
    ))
    await db.flush()
    await db.refresh(user)
    return ApiResponse(data=await _serialize_admin_user_detail(db, user))


@router.post("/{user_id}/refund-quota/add", response_model=ApiResponse)
async def add_user_refund_quota(
    user_id: int,
    data: AdminRefundQuotaAdjust,
    db: AsyncSession = Depends(get_db),
    current: User = Depends(_admin),
):
    res = await db.execute(select(User).where(User.id == user_id).with_for_update())
    user = res.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    reason = _required_reason(data.reason)
    before_extra = max(0, int(user.full_refund_quota_extra or 0))
    before_reset_at = user.full_refund_quota_reset_at
    user.full_refund_quota_extra = before_extra + data.count
    _add_refund_quota_log(
        db,
        user=user,
        operator=current,
        action="add",
        count=data.count,
        before_extra=before_extra,
        after_extra=user.full_refund_quota_extra,
        before_reset_at=before_reset_at,
        after_reset_at=user.full_refund_quota_reset_at,
        reason=reason,
    )
    await db.flush()
    await db.refresh(user)
    return ApiResponse(data=await _serialize_admin_user_detail(db, user), message="全额退订名额已增加")


@router.post("/{user_id}/refund-quota/reduce", response_model=ApiResponse)
async def reduce_user_refund_quota(
    user_id: int,
    data: AdminRefundQuotaAdjust,
    db: AsyncSession = Depends(get_db),
    current: User = Depends(_admin),
):
    res = await db.execute(select(User).where(User.id == user_id).with_for_update())
    user = res.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    reason = _required_reason(data.reason)
    before_extra = max(0, int(user.full_refund_quota_extra or 0))
    before_reset_at = user.full_refund_quota_reset_at
    user.full_refund_quota_extra = max(0, before_extra - data.count)
    _add_refund_quota_log(
        db,
        user=user,
        operator=current,
        action="reduce",
        count=data.count,
        before_extra=before_extra,
        after_extra=user.full_refund_quota_extra,
        before_reset_at=before_reset_at,
        after_reset_at=user.full_refund_quota_reset_at,
        reason=reason,
    )
    await db.flush()
    await db.refresh(user)
    return ApiResponse(data=await _serialize_admin_user_detail(db, user), message="全额退订额外名额已减少")


@router.post("/{user_id}/refund-quota/reset", response_model=ApiResponse)
async def reset_user_refund_quota(
    user_id: int,
    data: AdminRefundQuotaReset,
    db: AsyncSession = Depends(get_db),
    current: User = Depends(_admin),
):
    res = await db.execute(select(User).where(User.id == user_id).with_for_update())
    user = res.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    reason = _required_reason(data.reason)
    before_extra = max(0, int(user.full_refund_quota_extra or 0))
    before_reset_at = user.full_refund_quota_reset_at
    user.full_refund_quota_reset_at = datetime.now(timezone.utc)
    _add_refund_quota_log(
        db,
        user=user,
        operator=current,
        action="reset",
        count=0,
        before_extra=before_extra,
        after_extra=before_extra,
        before_reset_at=before_reset_at,
        after_reset_at=user.full_refund_quota_reset_at,
        reason=reason,
    )
    await db.flush()
    await db.refresh(user)
    return ApiResponse(data=await _serialize_admin_user_detail(db, user), message="全额退订次数已重置")


@router.post("/{user_id}/invite-code-quota/add", response_model=ApiResponse)
async def add_user_invite_code_quota(
    user_id: int,
    data: AdminInviteCodeQuotaAdjust,
    db: AsyncSession = Depends(get_db),
    current: User = Depends(_admin),
):
    res = await db.execute(select(User).where(User.id == user_id).with_for_update())
    user = res.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    reason = _required_reason(data.reason)
    before_used = int(user.invite_code_change_count or 0)
    before_extra = max(0, int(getattr(user, "invite_code_change_extra", 0) or 0))
    user.invite_code_change_extra = before_extra + data.count
    _add_invite_code_log(
        db,
        user=user,
        operator=current,
        action="add",
        count=data.count,
        before_used=before_used,
        after_used=before_used,
        before_extra=before_extra,
        after_extra=user.invite_code_change_extra,
        reason=reason,
    )
    await db.flush()
    await db.refresh(user)
    return ApiResponse(data=await _serialize_admin_user_detail(db, user), message="邀请码额外修改次数已增加")


@router.post("/{user_id}/invite-code-quota/reset", response_model=ApiResponse)
async def reset_user_invite_code_quota(
    user_id: int,
    data: AdminInviteCodeQuotaReset,
    db: AsyncSession = Depends(get_db),
    current: User = Depends(_admin),
):
    res = await db.execute(select(User).where(User.id == user_id).with_for_update())
    user = res.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    reason = _required_reason(data.reason)
    before_used = int(user.invite_code_change_count or 0)
    before_extra = max(0, int(getattr(user, "invite_code_change_extra", 0) or 0))
    user.invite_code_change_count = 0
    _add_invite_code_log(
        db,
        user=user,
        operator=current,
        action="reset",
        count=0,
        before_used=before_used,
        after_used=0,
        before_extra=before_extra,
        after_extra=before_extra,
        reason=reason,
    )
    await db.flush()
    await db.refresh(user)
    return ApiResponse(data=await _serialize_admin_user_detail(db, user), message="邀请码修改次数已重置")


@router.post("/{user_id}/verification-quota/add", response_model=ApiResponse)
async def add_user_verification_quota(
    user_id: int,
    data: AdminVerificationQuotaAdjust,
    db: AsyncSession = Depends(get_db),
    _: User = Depends(_admin),
):
    res = await db.execute(select(User).where(User.id == user_id).with_for_update())
    user = res.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    _required_reason(data.reason)
    quota = await _get_or_create_verification_quota(db, user.id)
    quota.paid_remaining = int(quota.paid_remaining or 0) + data.count
    quota.paid_purchased = int(quota.paid_purchased or 0) + data.count
    await db.flush()
    await db.refresh(user)
    return ApiResponse(data=await _serialize_admin_user_detail(db, user), message="实名认证次数已增加")


@router.post("/{user_id}/verification-quota/reduce", response_model=ApiResponse)
async def reduce_user_verification_quota(
    user_id: int,
    data: AdminVerificationQuotaAdjust,
    db: AsyncSession = Depends(get_db),
    _: User = Depends(_admin),
):
    res = await db.execute(select(User).where(User.id == user_id).with_for_update())
    user = res.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    _required_reason(data.reason)
    quota = await _get_or_create_verification_quota(db, user.id)
    quota.paid_remaining = max(0, int(quota.paid_remaining or 0) - data.count)
    await db.flush()
    await db.refresh(user)
    return ApiResponse(data=await _serialize_admin_user_detail(db, user), message="实名认证次数已减少")


@router.post("/{user_id}/verification-quota/reset-free", response_model=ApiResponse)
async def reset_user_verification_free_quota(
    user_id: int,
    data: AdminVerificationQuotaReset,
    db: AsyncSession = Depends(get_db),
    _: User = Depends(_admin),
):
    res = await db.execute(select(User).where(User.id == user_id).with_for_update())
    user = res.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    _required_reason(data.reason)
    quota = await _get_or_create_verification_quota(db, user.id)
    quota.free_used = 0
    await db.flush()
    await db.refresh(user)
    return ApiResponse(data=await _serialize_admin_user_detail(db, user), message="实名认证免费已用次数已重置")


@router.post("/{user_id}/ban", response_model=ApiResponse)
async def ban_user(
    user_id: int,
    db: AsyncSession = Depends(get_db),
    current: User = Depends(_admin),
):
    if user_id == current.id:
        raise HTTPException(status_code=400, detail="不能封禁自己")
    res = await db.execute(select(User).where(User.id == user_id))
    user = res.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    user.status = UserStatus.BANNED
    await db.flush()
    await session_service.revoke_all_refresh(user.id)
    await db.refresh(user)
    return ApiResponse(data=await _serialize_admin_user_detail(db, user))


@router.post("/{user_id}/unban", response_model=ApiResponse)
async def unban_user(
    user_id: int,
    db: AsyncSession = Depends(get_db),
    _: User = Depends(_admin),
):
    res = await db.execute(select(User).where(User.id == user_id))
    user = res.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    user.status = UserStatus.ACTIVE
    await db.flush()
    await db.refresh(user)
    return ApiResponse(data=await _serialize_admin_user_detail(db, user))


@router.post("/{user_id}/impersonate", response_model=ApiResponse)
async def impersonate_user(
    user_id: int,
    db: AsyncSession = Depends(get_db),
    _: User = Depends(_admin),
):
    """一键登录目标用户：签发该用户的 access/refresh token。

    注意：返回的 token 是真实用户身份令牌，前端拿到后写入普通用户 storage 即可登录。
    """
    res = await db.execute(select(User).where(User.id == user_id))
    user = res.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    if user.status == UserStatus.BANNED:
        raise HTTPException(status_code=400, detail="被封禁的用户无法一键登录")

    access_token, _jti_a = create_token(str(user.id), user.role.value, "access")
    refresh_token, refresh_jti = create_token(str(user.id), user.role.value, "refresh")
    await session_service.add_refresh(user.id, refresh_jti)
    return ApiResponse(
        data=ImpersonateResponse(
            access_token=access_token,
            refresh_token=refresh_token,
            user_id=user.id,
            username=user.username,
        )
    )
