import logging
from datetime import datetime, timezone
from typing import Any

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

from app.core.crypto import decrypt_secret
from app.database import async_session
from app.models.cloud_server import CloudServer
from app.models.instance import Instance, InstanceStatus, SuspendType
from app.models.order_refund import OrderRefund
from app.models.upstream_action_log import UpstreamActionLog
from app.services.dcim_cloud_client import DcimCloudClient, DcimCloudCredentials, DcimCloudError
from app.services import stock_compensation_service, zjmf_instance_adapter

log = logging.getLogger("qycw.upstream.actions")

PENDING_STATUSES = ("submitted", "running")
TASK_SUCCESS_STATUSES = {2}
TASK_FAILED_STATUSES = {3, 4, 5, -1}
FINAL_STATUSES = ("success", "failed", "no_task", "manual_handled", "ignored", "superseded")
RESOURCE_RUNNING_STATUSES = {"submitted", "running"}
RESOURCE_FINISHED_STATUSES = {"success", "no_task"}
RESOURCE_INACTIVE_STATUSES = {"superseded"}
RESOURCE_STATUS_LABELS = {
    "none": "无资源动作",
    "running": "资源处理中",
    "failed": "资源异常",
    "finished": "资源已处理",
    "manual_handled": "人工已处理",
    "ignored": "无需处理",
}


def _jsonable(value: Any) -> Any:
    if value is None or isinstance(value, (dict, list, str, int, float, bool)):
        return value
    if isinstance(value, tuple):
        return [_jsonable(item) for item in value]
    return str(value)


def _extract_payload(body: Any) -> dict[str, Any]:
    payload = (body or {}).get("data") if isinstance(body, dict) else None
    payload = payload or body or {}
    return payload if isinstance(payload, dict) else {}


def _extract_task_id(payload: dict[str, Any]) -> int | None:
    for key in ("taskid", "task_id", "id"):
        value = payload.get(key)
        if value in (None, ""):
            continue
        try:
            return int(value)
        except (TypeError, ValueError):
            continue
    return None


async def _get_cloud_server(db: AsyncSession, cloud_server_id: int) -> CloudServer:
    result = await db.execute(select(CloudServer).where(CloudServer.id == cloud_server_id))
    server = result.scalar_one_or_none()
    if not server:
        raise HTTPException(status_code=500, detail="上游节点不存在")
    return server


async def _get_dcim_client(db: AsyncSession, cloud_server_id: int) -> DcimCloudClient:
    result = await db.execute(select(CloudServer).where(CloudServer.id == cloud_server_id))
    server = result.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)


async def _write_action_log(**values: Any) -> int:
    async with async_session() as session:
        item = UpstreamActionLog(**values)
        session.add(item)
        await session.commit()
        return int(item.id)


async def _add_action_log(db: AsyncSession, **values: Any) -> UpstreamActionLog:
    item = UpstreamActionLog(**values)
    db.add(item)
    await db.flush()
    return item


async def call_path_with_audit(
    db: AsyncSession,
    instance: Instance,
    path: str,
    method: str,
    *,
    action_type: str,
    resource_type: str,
    source: str,
    data: dict[str, Any] | list[Any] | None = None,
    timeout: float | None = None,
    user_id: int | None = None,
    order_id: int | None = None,
    resource_id: str | int | None = None,
    remark: str | None = None,
) -> tuple[dict[str, Any], int]:
    if not instance.cloud_server_id:
        raise HTTPException(status_code=400, detail="实例未绑定云节点")

    server = await _get_cloud_server(db, instance.cloud_server_id)
    request_payload = _jsonable(data)
    base_values = {
        "user_id": user_id if user_id is not None else instance.user_id,
        "order_id": order_id,
        "instance_id": instance.id,
        "cloud_server_id": instance.cloud_server_id,
        "upstream_id": instance.upstream_id,
        "action_type": action_type,
        "resource_type": resource_type,
        "resource_id": str(resource_id) if resource_id is not None else None,
        "source": source,
        "request_method": method.upper(),
        "request_path": path,
        "request_payload": request_payload,
        "remark": remark,
    }

    if zjmf_instance_adapter.is_zjmf_server(server):
        try:
            payload = await zjmf_instance_adapter.call_path(
                db,
                instance,
                path,
                method,
                data=data,
                timeout=timeout,
            )
        except HTTPException as exc:
            await _write_action_log(
                **base_values,
                status="failed",
                error_message=str(exc.detail),
                completed_at=datetime.now(timezone.utc),
            )
            raise

        task_id = _extract_task_id(payload)
        status = "submitted" if task_id else "no_task"
        log_id = await _write_action_log(
            **base_values,
            status=status,
            upstream_task_id=task_id,
            response_payload=_jsonable(payload),
            completed_at=None if task_id else datetime.now(timezone.utc),
        )
        return payload, log_id

    client = await _get_dcim_client(db, instance.cloud_server_id)

    try:
        kwargs: dict[str, Any] = {"method": method, "data": data}
        if timeout is not None:
            kwargs["timeout"] = timeout
        code, text, body = await client.request(path, **kwargs)
    except DcimCloudError as exc:
        await _write_action_log(
            **base_values,
            status="failed",
            error_message=f"上游调用失败：{exc}",
            completed_at=datetime.now(timezone.utc),
        )
        raise HTTPException(status_code=502, detail=f"上游调用失败：{exc}") from exc

    payload = _extract_payload(body)
    task_id = _extract_task_id(payload)
    if not (200 <= code < 300):
        await _write_action_log(
            **base_values,
            status="failed",
            response_payload=_jsonable(body),
            error_message=f"上游返回异常（HTTP {code}）：{text[:200]}",
            completed_at=datetime.now(timezone.utc),
        )
        raise HTTPException(status_code=502, detail=f"上游返回异常（HTTP {code}）：{text[:200]}")

    status = "submitted" if task_id else "no_task"
    log_id = await _write_action_log(
        **base_values,
        status=status,
        upstream_task_id=task_id,
        response_payload=_jsonable(body),
        completed_at=None if task_id else datetime.now(timezone.utc),
    )
    return payload, log_id


def serialize_action_log(item: UpstreamActionLog) -> dict[str, Any]:
    return {
        "id": item.id,
        "user_id": item.user_id,
        "order_id": item.order_id,
        "order_refund_id": item.order_refund_id,
        "instance_id": item.instance_id,
        "cloud_server_id": item.cloud_server_id,
        "upstream_id": item.upstream_id,
        "action_type": item.action_type,
        "resource_type": item.resource_type,
        "resource_id": item.resource_id,
        "source": item.source,
        "status": item.status,
        "upstream_task_id": item.upstream_task_id,
        "request_method": item.request_method,
        "request_path": item.request_path,
        "request_payload": item.request_payload,
        "response_payload": item.response_payload,
        "task_payload": item.task_payload,
        "error_message": item.error_message,
        "remark": item.remark,
        "attempt_count": item.attempt_count,
        "last_checked_at": item.last_checked_at.isoformat() if item.last_checked_at else None,
        "completed_at": item.completed_at.isoformat() if item.completed_at else None,
        "created_at": item.created_at.isoformat() if item.created_at else None,
        "updated_at": item.updated_at.isoformat() if item.updated_at else None,
    }


def summarize_resource_actions(actions: list[UpstreamActionLog]) -> dict[str, Any]:
    active_actions = [item for item in actions if item.status not in RESOURCE_INACTIVE_STATUSES]
    failed = [item for item in active_actions if item.status == "failed"]
    running = [item for item in active_actions if item.status in RESOURCE_RUNNING_STATUSES]
    if failed:
        status = "failed"
    elif running:
        status = "running"
    elif any(item.status == "manual_handled" for item in active_actions):
        status = "manual_handled"
    elif any(item.status == "ignored" for item in active_actions):
        status = "ignored"
    elif any(item.status in RESOURCE_FINISHED_STATUSES for item in active_actions):
        status = "finished"
    else:
        status = "none"
    return {
        "status": status,
        "label": RESOURCE_STATUS_LABELS.get(status, status),
        "failed_count": len(failed),
        "running_count": len(running),
    }


def _latest_failed_error(actions: list[UpstreamActionLog]) -> str | None:
    for item in sorted(actions, key=lambda action: action.id or 0, reverse=True):
        if item.status == "failed" and item.status not in RESOURCE_INACTIVE_STATUSES:
            return item.error_message
    return None


async def refresh_refund_resource_status(db: AsyncSession, refund_id: int | None) -> dict[str, Any] | None:
    if not refund_id:
        return None

    refund_result = await db.execute(select(OrderRefund).where(OrderRefund.id == refund_id))
    refund = refund_result.scalar_one_or_none()
    if not refund:
        return None

    action_result = await db.execute(
        select(UpstreamActionLog)
        .where(UpstreamActionLog.order_refund_id == refund_id)
        .order_by(UpstreamActionLog.id.desc())
    )
    actions = list(action_result.scalars().all())
    summary = summarize_resource_actions(actions)
    status = str(summary["status"])
    refund.resource_status = status
    refund.resource_error = _latest_failed_error(actions) if status == "failed" else None
    refund.resource_completed_at = None if status == "running" else datetime.now(timezone.utc)
    await db.flush()
    return summary


def _resolve_task_status(task: dict[str, Any]) -> str:
    status = task.get("status")
    progress = task.get("progress", 0)
    if isinstance(status, str):
        normalized = status.lower()
        if normalized in ("success", "finished", "completed", "done"):
            return "success"
        if normalized in ("failed", "failure", "error", "cancelled", "canceled"):
            return "failed"
        return "running"
    try:
        numeric_status = int(status)
    except (TypeError, ValueError):
        numeric_status = None
    if numeric_status in TASK_SUCCESS_STATUSES:
        return "success"
    if numeric_status in TASK_FAILED_STATUSES:
        return "failed"
    try:
        if int(progress or 0) >= 100:
            return "success"
    except (TypeError, ValueError):
        pass
    return "running"


async def _apply_local_task_result(db: AsyncSession, item: UpstreamActionLog, status: str, task: dict[str, Any]) -> None:
    if not item.instance_id:
        return
    result = await db.execute(select(Instance).where(Instance.id == item.instance_id))
    instance = result.scalar_one_or_none()
    if not instance:
        return

    if status == "failed":
        msg = task.get("msg") or task.get("message") or item.error_message or "上游任务失败，待人工处理"
        instance.provision_log = f"上游任务失败，待人工处理：{msg}"
        return

    if status != "success":
        return

    if item.action_type == "instance_delete":
        instance.status = InstanceStatus.TERMINATED
        instance.terminated_at = instance.terminated_at or datetime.now(timezone.utc)
        instance.suspend_type = None
        await stock_compensation_service.restock_released_instances(
            db,
            [instance],
            source=item.source or "upstream_action",
            reason=item.remark or "instance delete completed",
        )
    elif item.action_type == "instance_suspend":
        instance.status = InstanceStatus.SUSPENDED
        if not instance.suspend_type:
            instance.suspend_type = SuspendType.DUE
    elif item.action_type == "instance_unsuspend":
        instance.status = InstanceStatus.ACTIVE
        instance.suspend_type = None


async def _sync_zjmf_action(db: AsyncSession, item: UpstreamActionLog, *, now: datetime) -> str:
    instance: Instance | None = None
    if item.instance_id:
        result = await db.execute(select(Instance).where(Instance.id == item.instance_id))
        instance = result.scalar_one_or_none()

    if not instance:
        item.status = "success"
        item.completed_at = now
        item.error_message = None
        item.task_payload = {"provider": "zjmf_finance", "note": "missing local instance, treated as completed"}
        return "success"

    payload = await zjmf_instance_adapter.list_tasks(db, instance, page=1, per_page=50)
    tasks = payload.get("list") if isinstance(payload, dict) else []
    if not isinstance(tasks, list):
        tasks = []
    task = next(
        (
            task_item
            for task_item in tasks
            if str(task_item.get("id")) == str(item.upstream_task_id)
        ),
        None,
    )

    if task is None:
        item.status = "success"
        item.completed_at = now
        item.error_message = None
        item.task_payload = {
            "provider": "zjmf_finance",
            "note": "finance task was not returned by upstream status, treated as completed",
            "tasks": _jsonable(tasks[:5]),
        }
        await _apply_local_task_result(db, item, "success", {})
        return "success"

    item.task_payload = {"provider": "zjmf_finance", "task": _jsonable(task)}
    resolved = _resolve_task_status(task)
    item.status = resolved
    if resolved in ("success", "failed"):
        item.completed_at = now
    if resolved == "failed":
        item.error_message = str(task.get("msg") or task.get("message") or "魔方财务任务失败")
    else:
        item.error_message = None
    await _apply_local_task_result(db, item, resolved, task)
    return resolved


async def _sync_one_action(db: AsyncSession, item: UpstreamActionLog, *, now: datetime | None = None) -> str:
    now = now or datetime.now(timezone.utc)
    item.attempt_count += 1
    item.last_checked_at = now

    if not item.cloud_server_id or not item.upstream_task_id:
        item.status = "failed"
        item.error_message = "缺少云节点或上游任务 ID"
        item.completed_at = now
        return "failed"

    server = await _get_cloud_server(db, item.cloud_server_id)
    if zjmf_instance_adapter.is_zjmf_server(server):
        return await _sync_zjmf_action(db, item, now=now)

    client = await _get_dcim_client(db, item.cloud_server_id)
    code, text, body = await client.request(f"/tasks/{item.upstream_task_id}", method="GET", timeout=8.0)
    if code == 404:
        # 上游任务记录已不存在：磁盘卸载/删除等短任务完成后会被上游清理，
        # 此时查询 /tasks/{id} 返回 404。视为已完成，避免误判为同步失败。
        item.status = "success"
        item.completed_at = now
        item.error_message = None
        item.task_payload = {"note": "上游任务记录已不存在，推断为已完成", "http_status": 404}
        await _apply_local_task_result(db, item, "success", {})
        return "success"
    if not (200 <= code < 300):
        raise DcimCloudError(f"HTTP {code}: {text[:200]}")

    task = _extract_payload(body)
    item.task_payload = _jsonable(body)
    resolved = _resolve_task_status(task)
    item.status = resolved
    if resolved in ("success", "failed"):
        item.completed_at = now
    if resolved == "failed":
        item.error_message = str(task.get("msg") or task.get("message") or "上游任务失败")
    await _apply_local_task_result(db, item, resolved, task)
    return resolved


async def sync_action(db: AsyncSession, item: UpstreamActionLog) -> str:
    if not item.upstream_task_id:
        item.status = "no_task"
        item.completed_at = item.completed_at or datetime.now(timezone.utc)
        await refresh_refund_resource_status(db, item.order_refund_id)
        await db.flush()
        return item.status
    try:
        status = await _sync_one_action(db, item)
    except Exception as exc:
        item.status = "failed" if item.attempt_count >= 5 else "running"
        item.error_message = f"任务同步失败：{exc}"
        if item.status == "failed":
            item.completed_at = datetime.now(timezone.utc)
        log.warning("sync upstream action failed: action=%s task=%s error=%s", item.id, item.upstream_task_id, exc)
        status = item.status
    await refresh_refund_resource_status(db, item.order_refund_id)
    await db.flush()
    return status


async def retry_action(
    db: AsyncSession,
    item: UpstreamActionLog,
    *,
    admin_user_id: int,
    remark: str | None = None,
) -> dict[str, Any]:
    if not item.cloud_server_id:
        raise HTTPException(status_code=400, detail="缺少云节点，无法重试")
    if not item.request_path or not item.request_method:
        raise HTTPException(status_code=400, detail="缺少原始请求信息，无法重试")
    if item.status not in ("failed", "manual_handled", "ignored"):
        raise HTTPException(status_code=400, detail="仅失败或已人工处理的动作允许重试")

    client = await _get_dcim_client(db, item.cloud_server_id)
    base_values = {
        "user_id": item.user_id,
        "order_id": item.order_id,
        "order_refund_id": item.order_refund_id,
        "instance_id": item.instance_id,
        "cloud_server_id": item.cloud_server_id,
        "upstream_id": item.upstream_id,
        "action_type": item.action_type,
        "resource_type": item.resource_type,
        "resource_id": item.resource_id,
        "source": "retry",
        "request_method": item.request_method,
        "request_path": item.request_path,
        "request_payload": item.request_payload,
        "remark": remark or f"管理员 #{admin_user_id} 重试动作 #{item.id}",
    }

    try:
        code, text, body = await client.request(
            item.request_path,
            method=item.request_method,
            data=item.request_payload,  # type: ignore[arg-type]
            timeout=30.0,
        )
    except DcimCloudError as exc:
        new_item = await _add_action_log(
            db,
            **base_values,
            status="failed",
            error_message=f"上游调用失败：{exc}",
            completed_at=datetime.now(timezone.utc),
        )
        log_id = int(new_item.id)
        item.status = "superseded"
        item.completed_at = datetime.now(timezone.utc)
        item.remark = f"已重试，重试日志 #{log_id} 调用失败"
        await refresh_refund_resource_status(db, item.order_refund_id)
        await db.flush()
        return {"new_action_log_id": log_id, "status": "failed", "taskid": None}

    payload = _extract_payload(body)
    task_id = _extract_task_id(payload)
    if not (200 <= code < 300):
        new_item = await _add_action_log(
            db,
            **base_values,
            status="failed",
            response_payload=_jsonable(body),
            error_message=f"上游返回异常（HTTP {code}）：{text[:200]}",
            completed_at=datetime.now(timezone.utc),
        )
        log_id = int(new_item.id)
        item.status = "superseded"
        item.completed_at = datetime.now(timezone.utc)
        item.remark = f"已重试，重试日志 #{log_id} 返回异常"
        await refresh_refund_resource_status(db, item.order_refund_id)
        await db.flush()
        return {"new_action_log_id": log_id, "status": "failed", "taskid": None}

    status = "submitted" if task_id else "no_task"
    new_item = await _add_action_log(
        db,
        **base_values,
        status=status,
        upstream_task_id=task_id,
        response_payload=_jsonable(body),
        completed_at=None if task_id else datetime.now(timezone.utc),
    )
    log_id = int(new_item.id)
    item.status = "superseded"
    item.completed_at = datetime.now(timezone.utc)
    item.remark = f"已重试，重试日志 #{log_id}"
    await refresh_refund_resource_status(db, item.order_refund_id)
    await db.flush()
    return {"new_action_log_id": log_id, "status": status, "taskid": task_id}


async def mark_action_manual_handled(
    db: AsyncSession,
    item: UpstreamActionLog,
    *,
    admin_user_id: int,
    remark: str | None,
) -> None:
    item.status = "manual_handled"
    item.completed_at = datetime.now(timezone.utc)
    item.error_message = None
    item.remark = remark or f"管理员 #{admin_user_id} 标记人工已处理"
    await _apply_local_task_result(db, item, "success", {})
    await refresh_refund_resource_status(db, item.order_refund_id)
    await db.flush()


async def mark_action_ignored(
    db: AsyncSession,
    item: UpstreamActionLog,
    *,
    admin_user_id: int,
    remark: str | None,
) -> None:
    item.status = "ignored"
    item.completed_at = datetime.now(timezone.utc)
    item.error_message = None
    item.remark = remark or f"管理员 #{admin_user_id} 标记无需处理"
    await refresh_refund_resource_status(db, item.order_refund_id)
    await db.flush()


async def sync_pending_actions(db: AsyncSession, *, limit: int = 100) -> dict[str, int]:
    result = await db.execute(
        select(UpstreamActionLog)
        .where(
            UpstreamActionLog.status.in_(PENDING_STATUSES),
            UpstreamActionLog.upstream_task_id.is_not(None),
        )
        .order_by(
            UpstreamActionLog.last_checked_at.is_not(None).asc(),
            UpstreamActionLog.last_checked_at.asc(),
            UpstreamActionLog.id.asc(),
        )
        .limit(limit)
    )
    items = list(result.scalars().all())

    checked = 0
    success = 0
    failed = 0
    running = 0
    errored = 0
    now = datetime.now(timezone.utc)
    refund_ids: set[int] = set()

    for item in items:
        checked += 1
        try:
            resolved = await _sync_one_action(db, item, now=now)
            if item.order_refund_id:
                refund_ids.add(int(item.order_refund_id))
            if resolved == "success":
                success += 1
            elif resolved == "failed":
                failed += 1
            else:
                running += 1
        except Exception as exc:
            errored += 1
            item.status = "failed" if item.attempt_count >= 5 else "running"
            item.error_message = f"任务同步失败：{exc}"
            if item.status == "failed":
                item.completed_at = now
                failed += 1
            if item.order_refund_id:
                refund_ids.add(int(item.order_refund_id))
            log.warning("sync upstream action failed: action=%s task=%s error=%s", item.id, item.upstream_task_id, exc)

    for refund_id in refund_ids:
        await refresh_refund_resource_status(db, refund_id)

    if items:
        await db.flush()
    return {"checked": checked, "success": success, "failed": failed, "running": running, "errored": errored}
