from __future__ import annotations

import asyncio
import logging
import re
from datetime import datetime, timezone
from html.parser import HTMLParser
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, encrypt_secret
from app.models.cloud_server import CloudServer, CloudServerType
from app.models.instance import Instance, InstanceStatus
from app.services.zjmf_finance_client import ZjmfFinanceClient, ZjmfFinanceCredentials, ZjmfFinanceError

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

SUCCESS_STATUSES = {200, 1001}
UNSUPPORTED_OPERATION_MESSAGE = "对接产品不支持该操作"
FINANCE_TICKET_MESSAGE = "该对接产品需要通过财务工单处理，请在实例页面使用财务工单入口提交。"
UPGRADE_UNSUPPORTED_MESSAGE = "对接产品不支持升降配"
POWER_ACTIONS = {
    "/on": "on",
    "/off": "off",
    "/reboot": "reboot",
    "/hardoff": "hard_off",
    "/hard_off": "hard_off",
    "/hard-reboot": "hard_reboot",
    "/hard_reboot": "hard_reboot",
}
POWER_STATUS_MAP = {
    "active": "on",
    "running": "on",
    "run": "on",
    "on": "on",
    "online": "on",
    "off": "off",
    "offline": "off",
    "stopped": "off",
    "shutdown": "off",
    "suspended": "suspend",
    "suspend": "suspend",
    "paused": "paused",
    "pending": "init",
    "provisioning": "init",
    "creating": "init",
}


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


def is_zjmf_server(server: CloudServer | None) -> bool:
    return bool(server and enum_value(server.server_type) == CloudServerType.ZJMF_FINANCE.value)


async def get_cloud_server(db: AsyncSession, cloud_server_id: int | None) -> CloudServer | None:
    if not cloud_server_id:
        return None
    result = await db.execute(select(CloudServer).where(CloudServer.id == cloud_server_id))
    return result.scalar_one_or_none()


async def get_zjmf_server_for_instance(db: AsyncSession, instance: Instance) -> CloudServer | None:
    server = await get_cloud_server(db, instance.cloud_server_id)
    return server if is_zjmf_server(server) else None


async def require_zjmf_server_for_instance(db: AsyncSession, instance: Instance) -> CloudServer:
    if not instance.cloud_server_id:
        raise HTTPException(status_code=400, detail="实例未绑定上游节点")
    server = await get_cloud_server(db, instance.cloud_server_id)
    if not server:
        raise HTTPException(status_code=500, detail="上游节点不存在")
    if not is_zjmf_server(server):
        raise HTTPException(status_code=400, detail="当前实例不是魔方财务实例")
    return server


def client_for_server(server: CloudServer, *, timeout: float = 20.0) -> ZjmfFinanceClient:
    creds = ZjmfFinanceCredentials(
        base_url=server.base_url,
        username=server.username,
        password=decrypt_secret(server.password_encrypted),
        account_type=server.account_type,
    )
    return ZjmfFinanceClient(creds, server_id=server.id, timeout=timeout)


def _host_id(instance: Instance) -> str:
    if not instance.upstream_id:
        raise HTTPException(status_code=400, detail="实例尚未开通，无法操作")
    return str(instance.upstream_id)


def _int_or_none(value: Any) -> int | None:
    if value in (None, ""):
        return None
    if isinstance(value, str):
        value = value.split("|", 1)[0].strip()
        match = re.search(r"-?\d+(?:\.\d+)?", value)
        if match:
            value = match.group(0)
    try:
        return int(float(value))
    except (TypeError, ValueError):
        return None


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


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


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


def _business_status(body: dict[str, Any] | None) -> int | str | None:
    if not isinstance(body, dict) or "status" not in body:
        return None
    status = body.get("status")
    if isinstance(status, str):
        stripped = status.strip()
        if stripped.isdigit():
            return int(stripped)
        return stripped.lower()
    return status


def _success_body(body: dict[str, Any] | None) -> bool:
    status = _business_status(body)
    if status is None:
        return True
    if isinstance(status, int):
        return status in SUCCESS_STATUSES
    return status in {"success", "ok"}


def _upstream_message(body: dict[str, Any] | None, text: str = "") -> str:
    if isinstance(body, dict):
        for key in ("msg", "message", "error", "info"):
            value = body.get(key)
            if value not in (None, ""):
                return str(value)
    return text[:200] or "上游返回异常"


def _looks_like_html(text: str | None) -> bool:
    if not text:
        return False
    stripped = text.lstrip().lower()
    return stripped.startswith("<!doctype") or stripped.startswith("<html") or "<body" in stripped[:300]


def _clean_endpoint_error(code: int, text: str, body: dict[str, Any] | None, label: str) -> str | None:
    if code in {404, 405} or _looks_like_html(text):
        return None
    if isinstance(body, dict):
        message = _upstream_message(body, "")
        if message:
            return f"HTTP {code}: {message}"
    short = text.strip()
    if not short:
        return f"{label} HTTP {code}"
    return f"HTTP {code}: {short[:160]}"


def _is_unsupported_message(message: str | None) -> bool:
    if not message:
        return False
    lowered = message.lower()
    keywords = (
        "not support",
        "unsupported",
        "not allowed",
        "forbidden",
        "disabled",
        "permission",
        "不支持",
        "不允许",
        "禁止",
        "无权限",
        "未开启",
        "不可用",
        "不能执行",
    )
    return any(keyword in lowered or keyword in message for keyword in keywords)


def _extract_invoice_id(payload: dict[str, Any]) -> int | None:
    candidates: list[Any] = []
    candidates.extend(payload.get(key) for key in ("invoiceid", "invoice_id", "invoice", "id"))
    data = payload.get("data")
    if isinstance(data, dict):
        candidates.extend(data.get(key) for key in ("invoiceid", "invoice_id", "invoice", "id"))
    for value in candidates:
        if value in (None, ""):
            continue
        try:
            return int(value)
        except (TypeError, ValueError):
            continue
    return None


def _list_from_payload(payload: Any) -> tuple[list[Any], dict[str, Any]]:
    if isinstance(payload, list):
        return payload, {}
    if not isinstance(payload, dict):
        return [], {}

    data = payload.get("data")
    meta = payload.get("meta") if isinstance(payload.get("meta"), dict) else {}
    if isinstance(data, list):
        return data, meta
    if isinstance(data, dict):
        nested_meta = data.get("meta") if isinstance(data.get("meta"), dict) else meta
        for key in ("list", "items", "rows", "data", "snap", "backup", "rules", "rule", "security_groups"):
            value = data.get(key)
            if isinstance(value, list):
                return value, nested_meta
        return [], nested_meta

    for key in ("list", "items", "rows", "data", "snap", "backup", "rules", "rule", "security_groups"):
        value = payload.get(key)
        if isinstance(value, list):
            return value, meta
    vars_payload = payload.get("vars")
    if isinstance(vars_payload, dict):
        vars_meta = vars_payload.get("meta") if isinstance(vars_payload.get("meta"), dict) else meta
        for key in ("list", "items", "rows", "data", "rules", "rule"):
            value = vars_payload.get(key)
            if isinstance(value, list):
                return value, vars_meta
    return [], meta


def _normal_meta(meta: dict[str, Any], *, page: int = 1, per_page: int = 20, total: int | None = None) -> dict[str, int]:
    resolved_total = _int_or_none(meta.get("total")) if isinstance(meta, dict) else None
    if resolved_total is None:
        resolved_total = total or 0
    resolved_per_page = _int_or_none(meta.get("per_page")) if isinstance(meta, dict) else None
    resolved_per_page = resolved_per_page or per_page
    total_page = _int_or_none(meta.get("total_page")) if isinstance(meta, dict) else None
    if total_page is None:
        total_page = (resolved_total + resolved_per_page - 1) // resolved_per_page if resolved_total else 0
    return {
        "total": resolved_total,
        "total_page": total_page,
        "page": (_int_or_none(meta.get("page")) if isinstance(meta, dict) else None) or page,
        "per_page": resolved_per_page,
    }


class _NatAclTableParser(HTMLParser):
    def __init__(self) -> None:
        super().__init__()
        self.rows: list[dict[str, Any]] = []
        self._in_tr = False
        self._in_td = False
        self._td_text: list[str] = []
        self._cells: list[str] = []
        self._row_id: int | None = None

    def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
        if tag == "tr":
            self._in_tr = True
            self._cells = []
            self._row_id = None
        elif self._in_tr and tag == "td":
            self._in_td = True
            self._td_text = []
        elif self._in_tr and tag in {"button", "a"}:
            attr = dict(attrs)
            self._row_id = _int_or_none(attr.get("data-id")) or self._row_id

    def handle_endtag(self, tag: str) -> None:
        if tag == "td" and self._in_td:
            self._cells.append(" ".join("".join(self._td_text).split()))
            self._td_text = []
            self._in_td = False
        elif tag == "tr" and self._in_tr:
            if len(self._cells) >= 4:
                external = self._cells[1]
                nat_host_ip = external.rsplit(":", 1)[0] if ":" in external else ""
                ext_port = external.rsplit(":", 1)[1] if ":" in external else external
                protocol_text = self._cells[3].lower().replace("+", ",")
                protocol = 3 if "tcp" in protocol_text and "udp" in protocol_text else 2 if "udp" in protocol_text else 1 if "tcp" in protocol_text else None
                self.rows.append({
                    "id": self._row_id,
                    "name": self._cells[0],
                    "description": "",
                    "ext_port": _int_or_none(ext_port),
                    "int_port": _int_or_none(self._cells[2]),
                    "int_ip": "",
                    "protocol": protocol,
                    "nat_host_ip": nat_host_ip or None,
                })
            self._in_tr = False

    def handle_data(self, data: str) -> None:
        if self._in_td:
            self._td_text.append(data)


def _parse_nat_acl_html(html: str) -> dict[str, Any]:
    parser = _NatAclTableParser()
    parser.feed(html)
    nat_host_ip = next((item.get("nat_host_ip") for item in parser.rows if item.get("nat_host_ip")), None)
    for item in parser.rows:
        item.pop("nat_host_ip", None)
    return {
        "list": parser.rows,
        "nat_host_ip": nat_host_ip,
    }


def _nat_acl_payload_from_header(header: dict[str, Any] | None) -> dict[str, Any] | None:
    if not isinstance(header, dict):
        return None
    dcimcloud = header.get("dcimcloud")
    if not isinstance(dcimcloud, dict):
        return None
    raw = dcimcloud.get("nat_acl")
    if raw in (None, "", [], {}):
        return None
    if isinstance(raw, str):
        text = raw.strip()
        if _looks_like_html(text):
            parsed = _parse_nat_acl_html(text)
            parsed["raw"] = text
            return parsed
        return {"list": [], "nat_host_ip": text}
    if isinstance(raw, list):
        return {"list": raw, "nat_host_ip": dcimcloud.get("nat_host_ip")}
    if isinstance(raw, dict):
        items, _ = _list_from_payload(raw)
        if items:
            return {"list": items, "nat_host_ip": raw.get("nat_host_ip") or dcimcloud.get("nat_host_ip")}
        return {"list": [], "nat_host_ip": raw.get("nat_host_ip") or dcimcloud.get("nat_host_ip"), "raw": raw}
    return None


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


def _host_data(detail: dict[str, Any] | None) -> dict[str, Any]:
    if not isinstance(detail, dict):
        return {}
    for key in ("host_data", "host"):
        value = detail.get(key)
        if isinstance(value, dict):
            host = dict(value)
            for extra_key in (
                "config_options",
                "config_option",
                "configoptions",
                "custom_field_data",
                "dcimcloud",
                "module_button",
                "module_client_area",
                "module_client_main_area",
                "module_power_status",
                "module_chart",
                "reinstall_random_port",
                "reinstall_format_data_disk",
            ):
                extra_value = detail.get(extra_key)
                if extra_value not in (None, "") and extra_key not in host:
                    host[extra_key] = extra_value
            return host
    return dict(detail)


def _assigned_ips(value: Any) -> list[str]:
    if isinstance(value, list):
        return [str(item) for item in value if item not in (None, "")]
    if isinstance(value, str):
        return [part.strip() for part in value.replace("\r", "\n").replace(",", "\n").split("\n") if part.strip()]
    return []


def _merge_host_data(*details: dict[str, Any] | None) -> dict[str, Any]:
    merged: dict[str, Any] = {}
    for detail in details:
        host = _host_data(detail)
        if host:
            merged.update(host)
    return merged


_CONFIG_KEY_ALIASES = {
    "cpu": "cpu",
    "vcpu": "cpu",
    "core": "cpu",
    "cores": "cpu",
    "processor": "cpu",
    "处理器核心": "cpu",
    "内核": "cpu",
    "memory": "memory",
    "mem": "memory",
    "ram": "memory",
    "内存": "memory",
    "system_disk_size": "system_disk_size",
    "system_disk": "system_disk_size",
    "sys_disk": "system_disk_size",
    "os_disk": "system_disk_size",
    "系统盘": "system_disk_size",
    "data_disk_size": "data_disk_size",
    "data_disk": "data_disk_size",
    "disk_space": "data_disk_size",
    "disk": "data_disk_size",
    "数据盘": "data_disk_size",
    "磁盘": "data_disk_size",
    "bw": "out_bw",
    "bandwidth": "out_bw",
    "bandwidth_out": "out_bw",
    "outbound_bw": "out_bw",
    "network_speed": "out_bw",
    "带宽": "out_bw",
    "网络速度": "out_bw",
    "in_bw": "in_bw",
    "bandwidth_in": "in_bw",
    "inbound_bw": "in_bw",
    "流入带宽": "in_bw",
    "ip": "ip_num",
    "ip_num": "ip_num",
    "ip_count": "ip_num",
    "extra_ip_address": "ip_num",
    "ip数量": "ip_num",
    "额外ip数量": "ip_num",
    "nat_acl": "nat_acl_limit",
    "nat_acl_limit": "nat_acl_limit",
    "nat转发": "nat_acl_limit",
    "nat_web": "nat_web_limit",
    "nat_web_limit": "nat_web_limit",
    "共享建站": "nat_web_limit",
}


def _normalize_config_key(key: Any) -> str:
    raw = str(key or "").split("|", 1)[0].strip()
    lowered = raw.lower().replace(" ", "_").replace("-", "_")
    return _CONFIG_KEY_ALIASES.get(lowered) or _CONFIG_KEY_ALIASES.get(raw) or lowered


def _config_option_map(host: dict[str, Any]) -> dict[str, Any]:
    result: dict[str, Any] = {}
    options = host.get("config_option") or host.get("config_options") or host.get("configoptions")
    if isinstance(options, dict):
        for key, value in options.items():
            if key not in (None, ""):
                result[_normalize_config_key(key)] = value
        return result
    if not isinstance(options, list):
        return result
    for option in options:
        if not isinstance(option, dict):
            continue
        key = option.get("key") or option.get("name_k") or option.get("option_key") or option.get("option_name") or option.get("name")
        if key in (None, ""):
            continue
        key = _normalize_config_key(key)
        value = option.get("value")
        if value in (None, ""):
            value = (
                option.get("sub_name")
                or option.get("suboption_name")
                or option.get("sub_option")
                or option.get("option_names")
                or option.get("qty")
            )
        result[key] = value
    return result


def _pick_value(host: dict[str, Any], options: dict[str, Any], *keys: str) -> Any:
    for key in keys:
        value = host.get(key)
        if value not in (None, ""):
            return value
    for key in keys:
        value = options.get(_normalize_config_key(key))
        if value not in (None, ""):
            return value
    return None


def _memory_mb(value: Any) -> int | None:
    if value in (None, ""):
        return None
    text = str(value).split("|", 1)[0].strip().lower()
    number = _int_or_none(text)
    if number is None:
        return None
    if "g" in text and "m" not in text:
        return number * 1024
    if "m" in text:
        return number
    return number * 1024 if 0 < number <= 128 else number


def _disk_list(host: dict[str, Any]) -> list[dict[str, Any]]:
    raw = host.get("disk") or host.get("disks")
    if not isinstance(raw, list):
        return []
    result: list[dict[str, Any]] = []
    for item in raw:
        if not isinstance(item, dict):
            continue
        size = _int_or_none(item.get("size") or item.get("capacity"))
        disk_type = item.get("type") or item.get("disk_type") or item.get("name")
        normalized = dict(item)
        if size is not None:
            normalized["size"] = size
        if disk_type not in (None, ""):
            normalized["type"] = disk_type
        result.append(normalized)
    return result


def _data_disk_size_from_disks(disks: list[dict[str, Any]]) -> int | None:
    total = 0
    found = False
    for disk in disks:
        size = _int_or_none(disk.get("size"))
        if size is None:
            continue
        disk_type = str(disk.get("type") or disk.get("disk_type") or "").lower()
        if any(token in disk_type for token in ("data", "数据", "鏁版嵁")):
            total += size
            found = True
    return total if found else None


def _system_disk_size_from_disks(disks: list[dict[str, Any]]) -> int | None:
    for disk in disks:
        size = _int_or_none(disk.get("size"))
        if size is None:
            continue
        disk_type = str(disk.get("type") or disk.get("disk_type") or "").lower()
        if any(token in disk_type for token in ("system", "os", "系统", "绯荤粺")):
            return size
    return None


def _specs_from_host(host: dict[str, Any]) -> dict[str, Any]:
    options = _config_option_map(host)
    default_bw = host.get("default_bw_group") if isinstance(host.get("default_bw_group"), dict) else {}
    disks = _disk_list(host)
    specs: dict[str, Any] = {}
    field_map = {
        "cpu": ("cpu", "vcpu", "cores", "core"),
        "system_disk_size": ("system_disk_size", "system_disk", "sys_disk", "os_disk"),
        "data_disk_size": ("data_disk_size", "data_disk"),
        "in_bw": ("in_bw", "inbound_bw", "bandwidth_in"),
        "out_bw": ("out_bw", "outbound_bw", "bandwidth_out", "bw"),
        "ip_num": ("ip_num", "ip", "ip_count"),
        "nat_acl_limit": ("nat_acl_limit", "nat_acl"),
        "nat_web_limit": ("nat_web_limit", "nat_web"),
        "traffic_quota": ("traffic_quota", "traffic"),
        "traffic_type": ("traffic_type",),
        "support_nat": ("support_nat",),
        "nat_acl_used": ("nat_acl_used",),
        "nat_web_used": ("nat_web_used",),
    }
    for field, keys in field_map.items():
        value = _pick_value(host, options, *keys)
        if value not in (None, ""):
            parsed = _int_or_none(value)
            specs[field] = parsed if parsed is not None else value
    memory = _memory_mb(_pick_value(host, options, "memory", "mem", "ram"))
    if memory is not None:
        specs["memory"] = memory
    if "in_bw" not in specs and isinstance(default_bw, dict):
        specs["in_bw"] = _int_or_none(default_bw.get("in_bw"))
    if "out_bw" not in specs and isinstance(default_bw, dict):
        specs["out_bw"] = _int_or_none(default_bw.get("out_bw"))
    if disks:
        specs["disk"] = disks
        specs.setdefault("system_disk_size", _system_disk_size_from_disks(disks))
        specs.setdefault("data_disk_size", _data_disk_size_from_disks(disks))
    return {key: value for key, value in specs.items() if value not in (None, "")}


def _apply_instance_value(instance: Instance, attr: str, value: Any) -> bool:
    if value in (None, "") or not hasattr(instance, attr):
        return False
    parsed = _int_or_none(value)
    if parsed is None:
        return False
    if getattr(instance, attr) == parsed:
        return False
    setattr(instance, attr, parsed)
    return True


def _power_status_from(value: Any) -> str | None:
    if value in (None, ""):
        return None
    raw = str(value).strip().lower()
    return POWER_STATUS_MAP.get(raw, raw or None)


def _normalize_power_status(status_payload: dict[str, Any] | None, host_data: dict[str, Any]) -> tuple[str | None, str | None]:
    data = _extract_payload(status_payload)
    status = (
        _power_status_from(data.get("status"))
        or _power_status_from(data.get("power"))
        or _power_status_from(data.get("state"))
        or _power_status_from(host_data.get("power_status"))
        or _power_status_from(host_data.get("domainstatus"))
    )
    label = data.get("des") or data.get("msg") or data.get("message") or data.get("text") or None
    return status, str(label) if label not in (None, "") else None


def _redact(value: Any) -> Any:
    if isinstance(value, dict):
        out: dict[str, Any] = {}
        for key, item in value.items():
            if str(key).lower() in {"password", "rootpassword", "root_password"}:
                out[key] = "***"
            else:
                out[key] = _redact(item)
        return out
    if isinstance(value, list):
        return [_redact(item) for item in value]
    return value


def _valid_port(value: Any) -> int | None:
    port = _int_or_none(value)
    if port is None or port <= 0 or port > 65535:
        return None
    return port


def _contains_module_key(value: Any, target: str) -> bool:
    target = target.lower()
    if isinstance(value, dict):
        for key in ("key", "name", "func"):
            raw = value.get(key)
            text = str(raw or "").strip().lower()
            if text and (text == target or target in text):
                return True
        return any(_contains_module_key(item, target) for item in value.values())
    if isinstance(value, list):
        return any(_contains_module_key(item, target) for item in value)
    if isinstance(value, str):
        text = value.strip().lower()
        return text == target or target in text
    return False


def _supports_nat_acl(module_client_area: Any, modules: Any, specs: dict[str, Any]) -> bool:
    if _contains_module_key(module_client_area, "nat_acl") or _contains_module_key(modules, "nat_acl"):
        return True
    limit = _int_or_none(specs.get("nat_acl_limit"))
    if limit is not None:
        return limit >= 0
    support = specs.get("support_nat")
    if isinstance(support, bool):
        return support
    if isinstance(support, (int, float)):
        return support != 0
    if isinstance(support, str):
        return support.strip().lower() in {"1", "true", "yes", "on"}
    dcimcloud = specs.get("dcimcloud")
    if isinstance(dcimcloud, dict):
        nat_acl = dcimcloud.get("nat_acl")
        if nat_acl not in (None, "", [], {}):
            return True
    return False


def sync_instance_from_host_detail(instance: Instance, detail: dict[str, Any] | None) -> bool:
    host = _host_data(detail)
    if not host:
        return False

    dirty = False
    domainstatus = str(host.get("domainstatus") or "").strip().lower()
    mapped_status = {
        "active": InstanceStatus.ACTIVE,
        "suspended": InstanceStatus.SUSPENDED,
        "deleted": InstanceStatus.TERMINATED,
        "cancelled": InstanceStatus.TERMINATED,
        "fraud": InstanceStatus.PROVISION_FAILED,
    }.get(domainstatus)
    if mapped_status is not None and instance.status != mapped_status:
        instance.status = mapped_status
        dirty = True

    assigned = _assigned_ips(host.get("assignedips"))
    main_ip = host.get("dedicatedip") or host.get("mainip") or host.get("main_ip") or host.get("ip")
    if not main_ip and assigned:
        main_ip = assigned[0]
    if main_ip and str(main_ip) != (instance.main_ip or ""):
        instance.main_ip = str(main_ip)
        dirty = True

    hostname = host.get("domain") or host.get("hostname")
    if hostname and str(hostname) != (instance.hostname or ""):
        instance.hostname = str(hostname)
        dirty = True

    port = _valid_port(host.get("port"))
    if port is not None and port != instance.ssh_port:
        instance.ssh_port = port
        dirty = True

    os_image = host.get("os") or host.get("os_name") or host.get("os_osname")
    if os_image and str(os_image) != (instance.os_image or ""):
        instance.os_image = str(os_image)
        dirty = True

    specs = _specs_from_host(host)
    for attr in (
        "cpu",
        "memory",
        "system_disk_size",
        "data_disk_size",
        "in_bw",
        "out_bw",
        "ip_num",
        "nat_acl_limit",
        "nat_web_limit",
        "traffic_quota",
        "traffic_type",
    ):
        dirty = _apply_instance_value(instance, attr, specs.get(attr)) or dirty

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

    expires_at = _timestamp_to_datetime(host.get("nextduedate") or host.get("next_due_date"))
    if expires_at and expires_at != instance.expires_at:
        instance.expires_at = expires_at
        dirty = True

    return dirty


def normalize_host_detail(
    header_detail: dict[str, Any] | None = None,
    openapi_detail: dict[str, Any] | None = None,
    status_payload: dict[str, Any] | None = None,
    modules: dict[str, Any] | None = None,
) -> dict[str, Any]:
    host = _merge_host_data(openapi_detail, header_detail)
    assigned = _assigned_ips(host.get("assignedips"))
    main_ip = host.get("dedicatedip") or host.get("mainip") or host.get("main_ip") or host.get("ip")
    if not main_ip and assigned:
        main_ip = assigned[0]
    power_status, power_label = _normalize_power_status(status_payload, host)
    specs = _specs_from_host(host)
    dcimcloud = host.get("dcimcloud") if isinstance(host.get("dcimcloud"), dict) else {}
    if dcimcloud and "dcimcloud" not in specs:
        specs["dcimcloud"] = dcimcloud
    module_client_area = (header_detail or {}).get("module_client_area") if isinstance(header_detail, dict) else None
    support_nat = _supports_nat_acl(module_client_area, modules or {}, specs)

    normalized: dict[str, Any] = {
        "provider": "zjmf_finance",
        "status": power_status,
        "power_status": power_status,
        "power_label": power_label,
        "mainip": main_ip,
        "hostname": host.get("domain") or host.get("hostname"),
        "port": _valid_port(host.get("port")),
        "rootpassword": host.get("password") or host.get("rootpassword") or host.get("root_password"),
        "domainstatus": host.get("domainstatus"),
        "assignedips": assigned,
        "os": host.get("os") or host.get("os_name") or host.get("os_osname"),
        "bwusage": host.get("bwusage"),
        "bwlimit": host.get("bwlimit"),
        "module_button": (header_detail or {}).get("module_button") if isinstance(header_detail, dict) else None,
        "module_power_status": (header_detail or {}).get("module_power_status") if isinstance(header_detail, dict) else None,
        "module_client_area": module_client_area,
        "module_chart": (header_detail or {}).get("module_chart") if isinstance(header_detail, dict) else None,
        "modules": modules or {},
        "support_nat": support_nat,
        "dcimcloud": dcimcloud,
    }
    normalized.update(specs)
    if "disk" not in normalized and specs.get("disk"):
        normalized["disk"] = specs["disk"]
    normalized["zjmf"] = {
        "host_header": _redact(header_detail or {}),
        "host_detail": _redact(openapi_detail or {}),
        "status": _redact(status_payload or {}),
    }
    return normalized


async def _request_module_payload(
    client: ZjmfFinanceClient,
    host_id: str,
    action: str,
    data: dict[str, Any] | None = None,
    *,
    method: str = "PUT",
    timeout: float = 30.0,
    label: str,
) -> dict[str, Any]:
    try:
        code, text, body = await client.request_host_module(host_id, action, data or {}, method=method, timeout=timeout)
    except ZjmfFinanceError as exc:
        raise HTTPException(status_code=502, detail=f"魔方财务{label}调用失败：{exc.message}") from exc

    if not (200 <= code < 300):
        raise HTTPException(status_code=502, detail=f"魔方财务{label}返回异常（HTTP {code}）：{text[:200]}")
    if not _success_body(body):
        status = _business_status(body)
        code_for_client = 400 if status in {400, 406, "error", "failed"} else 502
        raise HTTPException(status_code=code_for_client, detail=f"魔方财务{label}失败：{_upstream_message(body, text)}")
    return _extract_payload(body)


async def _request_default_action_payload(
    client: ZjmfFinanceClient,
    host_id: str,
    action: str,
    data: dict[str, Any] | None = None,
    *,
    timeout: float = 30.0,
    label: str,
) -> dict[str, Any]:
    try:
        code, text, body = await client.request_provision_default(host_id, action, data or {}, timeout=timeout)
    except ZjmfFinanceError as exc:
        raise HTTPException(status_code=502, detail=f"zjmf {label} call failed: {exc.message}") from exc

    if not (200 <= code < 300):
        raise HTTPException(status_code=502, detail=f"zjmf {label} returned HTTP {code}: {text[:200]}")
    if not _success_body(body):
        status = _business_status(body)
        code_for_client = 400 if status in {400, 406, "error", "failed"} else 502
        raise HTTPException(status_code=code_for_client, detail=f"zjmf {label} failed: {_upstream_message(body, text)}")
    return _extract_payload(body)


async def _request_power_payload(
    client: ZjmfFinanceClient,
    host_id: str,
    action: str,
    *,
    timeout: float = 30.0,
) -> dict[str, Any]:
    default_error: HTTPException | None = None
    try:
        return await _request_default_action_payload(
            client,
            host_id,
            action,
            {},
            timeout=timeout,
            label=action,
        )
    except HTTPException as exc:
        if exc.status_code not in {400, 404, 405, 502}:
            raise
        default_error = exc
        log.info("zjmf power provision/default %s fallback to module: host_id=%s detail=%s", action, host_id, exc.detail)

    try:
        return await _request_module_payload(
            client,
            host_id,
            action,
            {},
            method="PUT",
            timeout=timeout,
            label=action,
        )
    except HTTPException as module_error:
        default_detail = str(default_error.detail if default_error is not None else "")
        default_missing = "HTTP 404" in default_detail or _looks_like_html(default_detail)
        if default_error is not None and not default_missing:
            raise default_error
        raise module_error


async def _optional_endpoint_payload(
    client: ZjmfFinanceClient,
    host_id: str,
    attempts: list[dict[str, Any]],
    *,
    label: str,
) -> tuple[dict[str, Any] | None, str | None]:
    errors: list[str] = []
    for attempt in attempts:
        kind = str(attempt.get("kind") or "module")
        name = str(attempt.get("name") or "")
        data = attempt.get("data") if isinstance(attempt.get("data"), dict) else {}
        method = str(attempt.get("method") or "GET")
        timeout = float(attempt.get("timeout") or 30.0)
        try:
            if kind == "custom":
                code, text, body = await client.request_host_custom(
                    host_id,
                    name,
                    data,
                    method=method,
                    timeout=timeout,
                )
            elif kind == "func":
                code, text, body = await client.request_host_custom_func(
                    host_id,
                    name,
                    data,
                    timeout=timeout,
                )
            else:
                code, text, body = await client.request_host_module(
                    host_id,
                    name,
                    data,
                    method=method,
                    timeout=timeout,
                )
        except ZjmfFinanceError as exc:
            errors.append(exc.message)
            continue

        if not (200 <= code < 300):
            error = _clean_endpoint_error(code, text, body, label)
            if error:
                errors.append(error)
            continue
        if body is None:
            stripped = text.strip()
            if stripped:
                return {"html": stripped, "provider": "zjmf_finance"}, None
            errors.append(f"{label} returned empty response")
            continue
        if _success_body(body):
            payload = _extract_payload(body)
            if not payload and isinstance(body.get("data"), list):
                payload = {"list": body.get("data")}
            return payload, None

        errors.append(_upstream_message(body, text))

    message = next((item for item in errors if item), None)
    return None, message


def _unsupported_list(page: int = 1, per_page: int = 20, message: str | None = None) -> dict[str, Any]:
    return {
        "list": [],
        "meta": _normal_meta({}, page=page, per_page=per_page, total=0),
        "supported": False,
        "message": message or UNSUPPORTED_OPERATION_MESSAGE,
    }


def _raise_optional_failure(message: str | None) -> None:
    if not message or _is_unsupported_message(message):
        raise HTTPException(status_code=400, detail=UNSUPPORTED_OPERATION_MESSAGE)
    raise HTTPException(status_code=502, detail=message)


async def _get_status_payload(client: ZjmfFinanceClient, host_id: str) -> dict[str, Any]:
    try:
        code, text, body = await client.request_provision_default(host_id, "status", {}, timeout=15.0)
        if code == 404 or code == 405 or (isinstance(body, dict) and _business_status(body) in {404, 405}):
            code, text, body = await client.get_host_module_status(host_id, status_type="host")
        if code == 404 or code == 405 or (isinstance(body, dict) and _business_status(body) in {404, 405}):
            code, text, body = await client.request_host_module(
                host_id,
                "status",
                {"type": "host"},
                method="PUT",
                timeout=15.0,
            )
    except ZjmfFinanceError as exc:
        log.info("zjmf host status failed: host_id=%s error=%s", host_id, exc.message)
        return {}
    if not (200 <= code < 300) or not isinstance(body, dict) or not _success_body(body):
        message = _upstream_message(body if isinstance(body, dict) else None, text)
        if _is_unsupported_message(message) or _business_status(body if isinstance(body, dict) else None) in {400, 406}:
            return {}
        log.info("zjmf host status failed: host_id=%s http=%s body=%s text=%s", host_id, code, body, text[:120])
        return {}
    return body


async def get_upstream_status_for_ids(
    db: AsyncSession,
    upstream_id: int,
    cloud_server_id: int,
) -> dict[str, Any] | None:
    server = await get_cloud_server(db, cloud_server_id)
    if not is_zjmf_server(server):
        return None
    client = client_for_server(server)  # type: ignore[arg-type]
    header: dict[str, Any] | None = None
    try:
        header = await client.get_host_header(upstream_id)
    except ZjmfFinanceError as exc:
        log.info("zjmf host header failed: host_id=%s error=%s", upstream_id, exc.message)
    status_payload = await _get_status_payload(client, str(upstream_id))
    if not header and not status_payload:
        return None
    return normalize_host_detail(header, None, status_payload, None)


def _chart_last_values(payload: dict[str, Any] | None) -> list[float | None]:
    """从 charts 接口返回的时序数据里取每条序列的最后一个有效值。"""
    data = _extract_payload(payload)
    series_list = data.get("list")
    if not isinstance(series_list, list):
        return []
    result: list[float | None] = []
    for series in series_list:
        if not isinstance(series, list):
            result.append(None)
            continue
        last_value: float | None = None
        for point in series:
            if isinstance(point, dict):
                candidate = _float_or_none(point.get("value"))
            else:
                candidate = _float_or_none(point)
            if candidate is not None:
                last_value = candidate
        result.append(last_value)
    return result


def _unit_scale_to_base(unit: str | None, *, bit_based: bool) -> float:
    """把 charts 自动缩放后的单位换算回基础单位（bps 或 B/s）。"""
    if not unit:
        return 1.0
    lowered = unit.strip().lower()
    prefix = lowered[0] if lowered else ""
    factors = {"k": 1024.0, "m": 1024.0 ** 2, "g": 1024.0 ** 3, "t": 1024.0 ** 4}
    return factors.get(prefix, 1.0)


def _format_bps(value: float | None) -> str | None:
    if value is None:
        return None
    return f"{value / (1024.0 ** 2):.2f}Mbps"


async def _collect_zjmf_chart(
    client: ZjmfFinanceClient,
    host_id: str,
    chart_type: str,
    *,
    select: str | None = None,
) -> dict[str, Any] | None:
    try:
        code, _text, body = await client.get_host_chart(host_id, chart_type, select=select)
    except ZjmfFinanceError as exc:
        log.info("zjmf chart skipped: host_id=%s type=%s error=%s", host_id, chart_type, exc.message)
        return None
    if not (200 <= code < 300) or not isinstance(body, dict) or not _success_body(body):
        return None
    return body


async def collect_real_data_for_instance(
    db: AsyncSession,
    instance: Instance,
) -> dict[str, Any] | None:
    """采集魔方财务实例的实时监控数据，输出与魔方云 real_data 一致的字段结构。"""
    server = await get_zjmf_server_for_instance(db, instance)
    if not server or not instance.upstream_id:
        return None
    client = client_for_server(server, timeout=20.0)
    host_id = str(instance.upstream_id)

    cpu_body = await _collect_zjmf_chart(client, host_id, "cpu")
    mem_body = await _collect_zjmf_chart(client, host_id, "memory")
    flow_body = await _collect_zjmf_chart(client, host_id, "flow")
    disk_body = await _collect_zjmf_chart(client, host_id, "disk")
    if not any((cpu_body, mem_body, flow_body, disk_body)):
        return None

    result: dict[str, Any] = {"id": instance.upstream_id, "provider": "zjmf_finance"}

    cpu_values = _chart_last_values(cpu_body)
    if cpu_values and cpu_values[0] is not None:
        result["cpu_usage"] = cpu_values[0]

    mem_values = _chart_last_values(mem_body)
    if mem_values:
        total = mem_values[0] if len(mem_values) > 0 else None
        used = mem_values[1] if len(mem_values) > 1 else None
        if total and total > 0 and used is not None:
            result["memory_total"] = str(total)
            result["memory_usable"] = str(max(0.0, total - used))
            result["memory_usage"] = round(used / total * 100, 2)
        elif used is not None and "cpu_usage" not in result:
            result["memory_usage"] = used

    if flow_body:
        flow_data = _extract_payload(flow_body)
        scale = _unit_scale_to_base(flow_data.get("unit"), bit_based=True)
        flow_values = _chart_last_values(flow_body)
        in_value = flow_values[0] if len(flow_values) > 0 else None
        out_value = flow_values[1] if len(flow_values) > 1 else None
        if in_value is not None:
            result["current_in_bw"] = _format_bps(in_value * scale)
        if out_value is not None:
            result["current_out_bw"] = _format_bps(out_value * scale)

    if disk_body:
        disk_data = _extract_payload(disk_body)
        scale = _unit_scale_to_base(disk_data.get("unit"), bit_based=False)
        disk_values = _chart_last_values(disk_body)
        read_value = disk_values[0] if len(disk_values) > 0 else None
        write_value = disk_values[1] if len(disk_values) > 1 else None
        if read_value is not None:
            result["current_read_byte"] = round(read_value * scale / (1024.0 ** 2), 4)
        if write_value is not None:
            result["current_write_byte"] = round(write_value * scale / (1024.0 ** 2), 4)

    has_metric = any(
        key in result
        for key in ("cpu_usage", "memory_usage", "current_in_bw", "current_out_bw", "current_read_byte", "current_write_byte")
    )
    return result if has_metric else None


async def get_instance_detail(
    db: AsyncSession,
    instance: Instance,
    *,
    sync_local: bool = True,
) -> dict[str, Any]:
    server = await require_zjmf_server_for_instance(db, instance)
    host_id = _host_id(instance)
    client = client_for_server(server)

    header: dict[str, Any] | None = None
    openapi_detail: dict[str, Any] | None = None
    modules: dict[str, Any] | None = None
    errors: dict[str, str] = {}

    header_result, detail_result, modules_result, status_payload = await asyncio.gather(
        client.get_host_header(host_id),
        client.get_openapi_host(host_id),
        client.get_host_modules(host_id),
        _get_status_payload(client, host_id),
        return_exceptions=True,
    )
    if isinstance(header_result, ZjmfFinanceError):
        errors["host_header"] = header_result.message
    elif isinstance(header_result, dict):
        header = header_result
    if isinstance(detail_result, ZjmfFinanceError):
        errors["host_detail"] = detail_result.message
    elif isinstance(detail_result, dict):
        openapi_detail = detail_result
    if isinstance(modules_result, ZjmfFinanceError):
        errors["modules"] = modules_result.message
    elif isinstance(modules_result, dict):
        modules = modules_result
    if not isinstance(status_payload, dict):
        status_payload = {}
    if not header and not openapi_detail and not status_payload:
        raise HTTPException(status_code=502, detail=f"魔方财务实例详情获取失败：{'; '.join(errors.values()) or '无可用响应'}")

    detail = normalize_host_detail(header, openapi_detail, status_payload, modules)
    if errors:
        detail["upstream_warnings"] = errors
    if sync_local:
        dirty = sync_instance_from_host_detail(instance, detail)
        payload = instance.upstream_payload if isinstance(instance.upstream_payload, dict) else {}
        payload["zjmf_instance_detail"] = _redact(detail)
        instance.upstream_payload = payload
        if dirty:
            await db.flush()
    return detail


async def operate_instance(
    db: AsyncSession,
    instance: Instance,
    api_path: str,
    method: str = "POST",
    *,
    data: dict[str, Any] | None = None,
    timeout: float | None = None,
) -> dict[str, Any]:
    server = await require_zjmf_server_for_instance(db, instance)
    host_id = _host_id(instance)
    client = client_for_server(server, timeout=timeout or 30.0)
    path = api_path.split("?", 1)[0].rstrip("/")

    if path in POWER_ACTIONS:
        action = POWER_ACTIONS[path]
        payload = await _request_power_payload(
            client,
            host_id,
            action,
            timeout=timeout or 30.0,
        )
        return {
            "taskid": _extract_task_id(payload),
            "instance_id": instance.id,
            "provider": "zjmf_finance",
            "action": action,
            "upstream": payload,
        }

    if path in {"/password", "/repassword", "/reset-password"}:
        password = str((data or {}).get("password") or "").strip()
        if not password:
            raise HTTPException(status_code=400, detail="缺少新密码")
        try:
            payload = await _request_default_action_payload(
                client,
                host_id,
                "crack_pass",
                {"password": password},
                timeout=timeout or 30.0,
                label="reset password",
            )
        except HTTPException as exc:
            if exc.status_code not in {400, 404, 405, 502}:
                raise
            log.info("zjmf reset password provision/default fallback to module: host_id=%s detail=%s", host_id, exc.detail)
            payload = await _request_module_payload(
                client,
                host_id,
                "repassword",
                {"password": password},
                method="PUT",
                timeout=timeout or 30.0,
                label="reset password",
            )
        instance.root_password = encrypt_secret(password)
        await db.flush()
        return {
            "taskid": _extract_task_id(payload),
            "instance_id": instance.id,
            "provider": "zjmf_finance",
            "action": "repassword",
            "upstream": payload,
        }

    raise HTTPException(status_code=400, detail=f"魔方财务实例暂不支持该操作：{api_path}")


async def get_reinstall_options(db: AsyncSession, instance: Instance) -> dict[str, Any]:
    server = await require_zjmf_server_for_instance(db, instance)
    client = client_for_server(server)
    try:
        return await client.get_host_reinstall_options(_host_id(instance))
    except ZjmfFinanceError as exc:
        raise HTTPException(status_code=502, detail=f"魔方财务重装选项获取失败：{exc.message}") from exc


def reinstall_options_to_os_options(payload: dict[str, Any] | None) -> list[dict[str, Any]]:
    if not isinstance(payload, dict):
        return []
    raw = payload.get("os")
    if not isinstance(raw, list):
        raw = payload.get("list") if isinstance(payload.get("list"), list) else []
    result: list[dict[str, Any]] = []
    seen: set[str] = set()
    for item in raw:
        if not isinstance(item, dict):
            continue
        value = item.get("os_id") or item.get("id") or item.get("value")
        if value in (None, ""):
            continue
        key = str(value)
        if key in seen:
            continue
        seen.add(key)
        label = item.get("name") or item.get("label") or value
        result.append({
            "value": value,
            "label": str(label),
            "group_name": item.get("group_name") or "",
        })
    return result


def find_reinstall_os_label(payload: dict[str, Any] | None, os_id: int | str) -> str | None:
    target = str(os_id)
    for option in reinstall_options_to_os_options(payload):
        if str(option.get("value")) == target:
            return str(option.get("label") or os_id)
    return None


async def reinstall_instance(
    db: AsyncSession,
    instance: Instance,
    *,
    os_id: int | str,
    password: str | None = None,
    port: int | None = None,
    part_type: int | None = None,
) -> dict[str, Any]:
    server = await require_zjmf_server_for_instance(db, instance)
    host_id = _host_id(instance)
    client = client_for_server(server, timeout=45.0)

    payload: dict[str, Any] = {"os_id": str(os_id)}
    if port:
        payload["port"] = port
    if password:
        payload["password"] = password
        payload["dcim"] = {
            "password": password,
            "port": port or instance.ssh_port or 0,
            "part_type": part_type or 0,
        }

    try:
        upstream = await _request_module_payload(
            client,
            host_id,
            "reinstall",
            payload,
            method="PUT",
            timeout=45.0,
            label="重装系统",
        )
    except HTTPException as exc:
        if exc.status_code not in {400, 404, 405, 502}:
            raise
        default_payload: dict[str, Any] = {"os": str(os_id)}
        if port:
            default_payload["port"] = port
        if password:
            default_payload["password"] = password
        if part_type is not None:
            default_payload["format_data_disk"] = 1 if int(part_type) else 0
            default_payload["part_type"] = int(part_type)
        log.info("zjmf reinstall module fallback to provision/default: host_id=%s detail=%s", host_id, exc.detail)
        upstream = await _request_default_action_payload(
            client,
            host_id,
            "reinstall",
            default_payload,
            timeout=45.0,
            label="reinstall",
        )
    if password:
        instance.root_password = encrypt_secret(password)
    if port:
        instance.ssh_port = port
    await db.flush()
    return {
        "taskid": _extract_task_id(upstream),
        "instance_id": instance.id,
        "provider": "zjmf_finance",
        "action": "reinstall",
        "upstream": upstream,
    }


def _normalize_vnc_payload(payload: dict[str, Any]) -> dict[str, Any]:
    data = _extract_payload(payload)
    url = (
        data.get("vnc_url")
        or data.get("url")
        or data.get("vnc_url_https")
        or data.get("vnc_url_http")
        or data.get("host")
    )
    if not url:
        raise HTTPException(status_code=502, detail="魔方财务未返回 VNC 地址")
    vnc_pass = data.get("vnc_pass") or data.get("pass") or data.get("password") or ""
    return {
        "token": data.get("token"),
        "vnc_pass": vnc_pass,
        "vnc_url": url,
        "vnc_url_http": data.get("vnc_url_http") or (url if str(url).startswith("http://") else None),
        "vnc_url_https": data.get("vnc_url_https") or (url if str(url).startswith("https://") else None),
        "password": data.get("password"),
        "path": data.get("path"),
        "direct_url": url if str(url).startswith(("http://", "https://")) else None,
        "provider": "zjmf_finance",
        "upstream": _redact(data),
    }


async def get_vnc_info(db: AsyncSession, instance: Instance) -> dict[str, Any]:
    server = await require_zjmf_server_for_instance(db, instance)
    host_id = _host_id(instance)
    client = client_for_server(server, timeout=30.0)
    try:
        code, text, body = await client.request_host_module(host_id, "vnc", {}, method="PUT", timeout=30.0)
        if code in {404, 405} or (isinstance(body, dict) and _business_status(body) in {404, 405}):
            code, text, body = await client.request_host_module(host_id, "vnc", {}, method="GET", timeout=30.0)
    except ZjmfFinanceError as exc:
        raise HTTPException(status_code=502, detail=f"魔方财务 VNC 调用失败：{exc.message}") from exc
    if not (200 <= code < 300):
        raise HTTPException(status_code=502, detail=f"魔方财务 VNC 返回异常（HTTP {code}）：{text[:200]}")
    if not _success_body(body):
        raise HTTPException(status_code=400, detail=f"魔方财务 VNC 失败：{_upstream_message(body, text)}")
    return _normalize_vnc_payload(_extract_payload(body))


async def renew_instance(
    db: AsyncSession,
    instance: Instance,
    *,
    cycles: int = 1,
) -> dict[str, Any]:
    if cycles != 1:
        raise HTTPException(status_code=400, detail="魔方财务续费暂只支持单周期续费")
    server = await require_zjmf_server_for_instance(db, instance)
    host_id = _host_id(instance)
    client = client_for_server(server, timeout=45.0)
    billingcycle = str(instance.cycle or "monthly")

    renew_page: dict[str, Any] = {}
    try:
        renew_page = await client.get_host_renew_page(host_id, billingcycle=billingcycle)
    except ZjmfFinanceError as exc:
        log.info("zjmf renew page skipped: host_id=%s error=%s", host_id, exc.message)

    try:
        renew_body = await client.renew_host(host_id, billingcycle)
        invoice_id = _extract_invoice_id(renew_body)
        credit_body: dict[str, Any] | None = None
        if invoice_id:
            credit_body = await client.apply_credit(invoice_id)
    except ZjmfFinanceError as exc:
        raise HTTPException(status_code=502, detail=f"魔方财务续费失败：{exc.message}") from exc

    detail = await get_instance_detail(db, instance, sync_local=True)
    return {
        "provider": "zjmf_finance",
        "billingcycle": billingcycle,
        "cycles": cycles,
        "invoice_id": _extract_invoice_id(renew_body),
        "renew_page": _redact(renew_page),
        "renew": _redact(renew_body),
        "credit": _redact(credit_body or {}),
        "detail": _redact(detail),
        "expires_at": instance.expires_at.isoformat() if instance.expires_at else None,
    }


async def set_auto_renew(db: AsyncSession, instance: Instance, enabled: bool) -> dict[str, Any]:
    server = await require_zjmf_server_for_instance(db, instance)
    client = client_for_server(server, timeout=30.0)
    try:
        body = await client.set_host_auto_renew(_host_id(instance), enabled)
    except ZjmfFinanceError as exc:
        raise HTTPException(status_code=502, detail=f"魔方财务自动续费设置失败：{exc.message}") from exc
    return {"provider": "zjmf_finance", "auto_renew": 1 if enabled else 0, "upstream": _redact(body)}


def _task_item(
    *,
    task_id: int,
    task_type: str,
    status: int,
    progress: int,
    msg: str,
    host_id: str,
    hostname: str | None,
    now: datetime,
) -> dict[str, Any]:
    stamp = now.strftime("%Y-%m-%d %H:%M:%S")
    return {
        "id": task_id,
        "type": task_type,
        "status": status,
        "create_time": stamp,
        "start_time": stamp,
        "end_time": stamp if status in {2, 3, 4, 5} else "",
        "progress": progress,
        "msg": msg,
        "hostid": _int_or_none(host_id) or 0,
        "hostname": hostname or "",
        "rel_type": "zjmf_finance",
        "provider": "zjmf_finance",
    }


async def list_tasks(
    db: AsyncSession,
    instance: Instance,
    *,
    page: int = 1,
    per_page: int = 20,
) -> dict[str, Any]:
    server = await require_zjmf_server_for_instance(db, instance)
    host_id = _host_id(instance)
    client = client_for_server(server, timeout=20.0)
    now = datetime.now(timezone.utc)
    tasks: list[dict[str, Any]] = []

    status_payload = await _get_status_payload(client, host_id)
    if status_payload:
        status, label = _normalize_power_status(status_payload, {})
        tasks.append(_task_item(
            task_id=1,
            task_type="status",
            status=2,
            progress=100,
            msg=label or status or "魔方财务状态已同步",
            host_id=host_id,
            hostname=instance.hostname,
            now=now,
        ))

    try:
        code, text, body = await client.get_host_module_status(host_id, status_type="reinstall")
        if 200 <= code < 300 and isinstance(body, dict) and _success_body(body):
            payload = _extract_payload(body)
            progress = _int_or_none(payload.get("progress")) or 100
            status_value = 2 if progress >= 100 else 1
            tasks.append(_task_item(
                task_id=2,
                task_type="reinstall",
                status=status_value,
                progress=min(100, max(0, progress)),
                msg=str(payload.get("msg") or payload.get("message") or payload.get("des") or "魔方财务重装状态已同步"),
                host_id=host_id,
                hostname=instance.hostname,
                now=now,
            ))
        elif code not in {404, 405} and _business_status(body if isinstance(body, dict) else None) not in {400, 406}:
            log.info("zjmf reinstall status failed: host_id=%s http=%s text=%s", host_id, code, text[:120])
    except ZjmfFinanceError as exc:
        log.info("zjmf reinstall status skipped: host_id=%s error=%s", host_id, exc.message)

    start = max(0, (page - 1) * per_page)
    sliced = tasks[start:start + per_page]
    return {
        "list": sliced,
        "meta": _normal_meta({}, page=page, per_page=per_page, total=len(tasks)),
        "provider": "zjmf_finance",
    }


async def list_snapshots(
    db: AsyncSession,
    instance: Instance,
    *,
    page: int = 1,
    per_page: int = 20,
    orderby: str = "id",
    sort: str = "desc",
    search: str | None = None,
    snapshot_type: str | None = None,
) -> dict[str, Any]:
    server = await require_zjmf_server_for_instance(db, instance)
    host_id = _host_id(instance)
    client = client_for_server(server, timeout=30.0)
    params: dict[str, Any] = {"page": page, "per_page": per_page, "orderby": orderby, "sort": sort}
    if search:
        params["search"] = search
    if snapshot_type:
        params["type"] = snapshot_type
    payload, message = await _optional_endpoint_payload(
        client,
        host_id,
        [
            {"kind": "func", "name": "listSnapBackup", "data": params, "timeout": 30.0},
            {"kind": "module", "name": "snapshot", "data": params, "method": "GET", "timeout": 30.0},
            {"kind": "custom", "name": "snapshot", "data": params, "method": "GET", "timeout": 30.0},
            {"kind": "module", "name": "snapshots", "data": params, "method": "GET", "timeout": 30.0},
        ],
        label="snapshot",
    )
    if not payload or payload.get("html"):
        return _unsupported_list(page, per_page, message)
    items, meta = _list_from_payload(payload)
    if snapshot_type:
        items = [item for item in items if not isinstance(item, dict) or str(item.get("type") or "") == snapshot_type]
    if search:
        keyword = search.lower()
        items = [
            item for item in items
            if not isinstance(item, dict)
            or keyword in str(item.get("name") or item.get("remarks") or item.get("disk_name") or "").lower()
        ]
    return {
        "list": items,
        "meta": _normal_meta(meta, page=page, per_page=per_page, total=len(items)),
        "supported": True,
        "provider": "zjmf_finance",
    }


async def create_snapshot(
    db: AsyncSession,
    instance: Instance,
    *,
    disk_id: int,
    snapshot_type: str,
    name: str | None = None,
    remarks: str | None = None,
) -> dict[str, Any]:
    server = await require_zjmf_server_for_instance(db, instance)
    host_id = _host_id(instance)
    client = client_for_server(server, timeout=30.0)
    display_name = (remarks or name or f"{snapshot_type}-{datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')}").strip()
    func = "createBackup" if snapshot_type == "backup" else "createSnap"
    payload, message = await _optional_endpoint_payload(
        client,
        host_id,
        [{"kind": "func", "name": func, "data": {"id": disk_id, "name": display_name}, "timeout": 30.0}],
        label=func,
    )
    if not payload:
        _raise_optional_failure(message)
    return {
        "taskid": _extract_task_id(payload or {}),
        "diskid": disk_id,
        "type": snapshot_type,
        "provider": "zjmf_finance",
        "upstream": _redact(payload or {}),
    }


async def delete_snapshot(db: AsyncSession, instance: Instance, *, snapshot_id: int) -> dict[str, Any]:
    server = await require_zjmf_server_for_instance(db, instance)
    host_id = _host_id(instance)
    client = client_for_server(server, timeout=30.0)
    payload, message = await _optional_endpoint_payload(
        client,
        host_id,
        [
            {"kind": "func", "name": "deleteSnap", "data": {"id": snapshot_id}, "timeout": 30.0},
            {"kind": "func", "name": "deleteBackup", "data": {"id": snapshot_id}, "timeout": 30.0},
        ],
        label="delete snapshot",
    )
    if not payload:
        _raise_optional_failure(message)
    return {
        "taskid": _extract_task_id(payload or {}),
        "snapshot_id": snapshot_id,
        "provider": "zjmf_finance",
        "upstream": _redact(payload or {}),
    }


async def restore_snapshot(db: AsyncSession, instance: Instance, *, snapshot_id: int) -> dict[str, Any]:
    server = await require_zjmf_server_for_instance(db, instance)
    host_id = _host_id(instance)
    client = client_for_server(server, timeout=45.0)
    payload, message = await _optional_endpoint_payload(
        client,
        host_id,
        [
            {"kind": "func", "name": "restoreSnap", "data": {"id": snapshot_id}, "timeout": 45.0},
            {"kind": "func", "name": "restoreBackup", "data": {"id": snapshot_id}, "timeout": 45.0},
        ],
        label="restore snapshot",
    )
    if not payload:
        _raise_optional_failure(message)
    return {
        "taskid": _extract_task_id(payload or {}),
        "snapshot_id": snapshot_id,
        "provider": "zjmf_finance",
        "upstream": _redact(payload or {}),
    }


async def list_cron_snapshots(db: AsyncSession, instance: Instance) -> dict[str, Any]:
    server = await require_zjmf_server_for_instance(db, instance)
    host_id = _host_id(instance)
    client = client_for_server(server, timeout=30.0)
    payload, message = await _optional_endpoint_payload(
        client,
        host_id,
        [
            {"kind": "module", "name": "cron_snap", "method": "GET", "timeout": 30.0},
            {"kind": "module", "name": "cron_snapshot", "method": "GET", "timeout": 30.0},
            {"kind": "custom", "name": "cron_snap", "method": "GET", "timeout": 30.0},
            {"kind": "custom", "name": "cron_snapshot", "method": "GET", "timeout": 30.0},
        ],
        label="cron snapshot",
    )
    if not payload or payload.get("html"):
        return {"list": [], "supported": False, "message": message or UNSUPPORTED_OPERATION_MESSAGE}
    items, _ = _list_from_payload(payload)
    return {"list": items, "supported": True, "provider": "zjmf_finance", "upstream": _redact(payload)}


async def create_cron_snapshot(db: AsyncSession, instance: Instance, payload: dict[str, Any]) -> dict[str, Any]:
    server = await require_zjmf_server_for_instance(db, instance)
    host_id = _host_id(instance)
    client = client_for_server(server, timeout=30.0)
    upstream, message = await _optional_endpoint_payload(
        client,
        host_id,
        [
            {"kind": "module", "name": "cron_snap", "data": payload, "method": "POST", "timeout": 30.0},
            {"kind": "module", "name": "cron_snapshot", "data": payload, "method": "POST", "timeout": 30.0},
        ],
        label="create cron snapshot",
    )
    if not upstream:
        _raise_optional_failure(message)
    return {"provider": "zjmf_finance", "upstream": _redact(upstream or {})}


async def delete_cron_snapshot(db: AsyncSession, instance: Instance, *, cron_id: int) -> dict[str, Any]:
    server = await require_zjmf_server_for_instance(db, instance)
    host_id = _host_id(instance)
    client = client_for_server(server, timeout=30.0)
    payload, message = await _optional_endpoint_payload(
        client,
        host_id,
        [
            {"kind": "module", "name": f"cron_snap/{cron_id}", "method": "DELETE", "timeout": 30.0},
            {"kind": "module", "name": f"cron_snapshot/{cron_id}", "method": "DELETE", "timeout": 30.0},
        ],
        label="delete cron snapshot",
    )
    if not payload:
        _raise_optional_failure(message)
    return {"id": cron_id, "provider": "zjmf_finance", "upstream": _redact(payload or {})}


async def list_nat_acl(
    db: AsyncSession,
    instance: Instance,
    *,
    page: int = 1,
    per_page: int = 20,
    orderby: str = "id",
    sort: str = "desc",
    search: str | None = None,
    list_type: str = "page",
) -> dict[str, Any]:
    server = await require_zjmf_server_for_instance(db, instance)
    host_id = _host_id(instance)
    client = client_for_server(server, timeout=30.0)
    try:
        header = await client.get_host_header(host_id, nat=True)
        header_payload = _nat_acl_payload_from_header(header)
    except ZjmfFinanceError as exc:
        log.info("zjmf nat header skipped: host_id=%s error=%s", host_id, exc.message)
        header_payload = None
    if header_payload is not None:
        items = header_payload.get("list") if isinstance(header_payload.get("list"), list) else []
        if search:
            needle = search.strip().lower()
            items = [item for item in items if needle in str(item).lower()]
        reverse = sort == "desc"
        if orderby in {"id", "name", "ext_port", "int_port", "protocol"}:
            items = sorted(items, key=lambda item: (item.get(orderby) is None, item.get(orderby) or 0), reverse=reverse)
        total = len(items)
        if list_type != "all":
            start = max(0, (page - 1) * per_page)
            items = items[start:start + per_page]
        return {
            "list": items,
            "meta": _normal_meta({}, page=page, per_page=per_page, total=total),
            "nat_host_ip": header_payload.get("nat_host_ip"),
            "supported": True,
            "provider": "zjmf_finance",
            "upstream": _redact(header_payload),
        }

    payload, message = await _optional_endpoint_payload(
        client,
        host_id,
        [
            {"kind": "custom", "name": "nat_acl", "method": "GET", "timeout": 30.0},
            {"kind": "func", "name": "listNatAcl", "method": "POST", "timeout": 30.0},
            {"kind": "module", "name": "nat_acl", "method": "GET", "timeout": 30.0},
        ],
        label="nat_acl",
    )
    if not payload:
        return {
            "list": [],
            "meta": _normal_meta({}, page=page, per_page=per_page, total=0),
            "nat_host_ip": None,
            "supported": False,
            "message": message or UNSUPPORTED_OPERATION_MESSAGE,
            "provider": "zjmf_finance",
        }
    if payload.get("html"):
        parsed = _parse_nat_acl_html(str(payload.get("html") or ""))
        items = parsed["list"]
    else:
        items, _ = _list_from_payload(payload)
        parsed = {"nat_host_ip": payload.get("nat_host_ip")}
    if search:
        needle = search.strip().lower()
        items = [item for item in items if needle in str(item).lower()]
    reverse = sort == "desc"
    if orderby in {"id", "name", "ext_port", "int_port", "protocol"}:
        items = sorted(items, key=lambda item: (item.get(orderby) is None, item.get(orderby) or 0), reverse=reverse)
    total = len(items)
    if list_type != "all":
        start = max(0, (page - 1) * per_page)
        items = items[start:start + per_page]
    return {
        "list": items,
        "meta": _normal_meta({}, page=page, per_page=per_page, total=total),
        "nat_host_ip": parsed.get("nat_host_ip"),
        "supported": True,
        "provider": "zjmf_finance",
        "upstream": _redact(payload),
    }


async def create_nat_acl(
    db: AsyncSession,
    instance: Instance,
    *,
    name: str,
    int_port: int,
    protocol: int,
    ext_port: int | None = None,
) -> dict[str, Any]:
    server = await require_zjmf_server_for_instance(db, instance)
    host_id = _host_id(instance)
    client = client_for_server(server, timeout=30.0)
    data: dict[str, Any] = {
        "name": name,
        "int_port": int_port,
        "protocol": protocol,
        "select-protocol": protocol,
    }
    if ext_port is not None:
        data["ext_port"] = ext_port
    payload, message = await _optional_endpoint_payload(
        client,
        host_id,
        [{"kind": "func", "name": "addNatAcl", "data": data, "timeout": 30.0}],
        label="addNatAcl",
    )
    if not payload:
        _raise_optional_failure(message)
    refreshed = await list_nat_acl(db, instance, page=1, per_page=20, orderby="id", sort="desc", list_type="all")
    created = next((
        item for item in refreshed.get("list", [])
        if item.get("name") == name
        and item.get("int_port") == int_port
        and item.get("protocol") == protocol
        and (ext_port is None or item.get("ext_port") == ext_port)
    ), None)
    if isinstance(created, dict):
        created = dict(created)
        created["provider"] = "zjmf_finance"
        created["upstream"] = _redact(payload or {})
        return created
    return {
        "id": _int_or_none((payload or {}).get("id")),
        "name": name,
        "description": "",
        "ext_port": _int_or_none((payload or {}).get("ext_port")) or ext_port,
        "int_port": int_port,
        "int_ip": "",
        "protocol": protocol,
        "provider": "zjmf_finance",
        "upstream": _redact(payload or {}),
    }


async def delete_nat_acl(db: AsyncSession, instance: Instance, *, nat_acl_id: int) -> dict[str, Any]:
    server = await require_zjmf_server_for_instance(db, instance)
    host_id = _host_id(instance)
    client = client_for_server(server, timeout=30.0)
    payload, message = await _optional_endpoint_payload(
        client,
        host_id,
        [{"kind": "func", "name": "delNatAcl", "data": {"id": nat_acl_id}, "timeout": 30.0}],
        label="delNatAcl",
    )
    if not payload:
        _raise_optional_failure(message)
    return {"id": nat_acl_id, "provider": "zjmf_finance", "upstream": _redact(payload or {})}


async def get_firewall(db: AsyncSession, instance: Instance) -> dict[str, Any]:
    server = await require_zjmf_server_for_instance(db, instance)
    host_id = _host_id(instance)
    client = client_for_server(server, timeout=30.0)
    payload, message = await _optional_endpoint_payload(
        client,
        host_id,
        [
            {"kind": "module", "name": "security_groups", "method": "GET", "timeout": 30.0},
            {"kind": "custom", "name": "security_groups", "method": "GET", "timeout": 30.0},
            {"kind": "module", "name": "firewall", "method": "GET", "timeout": 30.0},
            {"kind": "custom", "name": "firewall", "method": "GET", "timeout": 30.0},
        ],
        label="firewall",
    )
    if not payload:
        return {"supported": False, "message": message or UNSUPPORTED_OPERATION_MESSAGE, "list": []}
    if payload.get("html"):
        return {
            "supported": False,
            "message": UNSUPPORTED_OPERATION_MESSAGE,
            "html_available": True,
            "provider": "zjmf_finance",
        }
    items, meta = _list_from_payload(payload)
    return {
        "supported": True,
        "list": items,
        "meta": _normal_meta(meta, page=1, per_page=max(len(items), 1), total=len(items)),
        "provider": "zjmf_finance",
        "upstream": _redact(payload),
    }


def _cloud_suffix(instance: Instance, path: str) -> str | None:
    upstream_id = str(instance.upstream_id or "")
    if not upstream_id:
        return None
    clean_path = path.split("?", 1)[0].rstrip("/")
    prefix = f"/clouds/{upstream_id}"
    if clean_path == prefix:
        return ""
    if clean_path.startswith(prefix + "/"):
        return clean_path[len(prefix):]
    return None


async def call_path(
    db: AsyncSession,
    instance: Instance,
    path: str,
    method: str,
    *,
    data: dict[str, Any] | list[Any] | None = None,
    timeout: float | None = None,
) -> dict[str, Any]:
    suffix = _cloud_suffix(instance, path)
    method_upper = method.upper()
    if suffix == "" and method_upper == "GET":
        return await get_instance_detail(db, instance, sync_local=True)
    if suffix in POWER_ACTIONS and method_upper in {"POST", "PUT"}:
        return await operate_instance(
            db,
            instance,
            suffix,
            method_upper,
            data=data if isinstance(data, dict) else None,
            timeout=timeout,
        )
    if suffix in {"/password", "/repassword", "/reset-password"} and method_upper in {"POST", "PUT"}:
        return await operate_instance(
            db,
            instance,
            suffix,
            method_upper,
            data=data if isinstance(data, dict) else None,
            timeout=timeout,
        )
    if suffix == "/vnc" and method_upper in {"POST", "PUT", "GET"}:
        return await get_vnc_info(db, instance)
    if suffix == "/reinstall" and method_upper == "PUT":
        payload = data if isinstance(data, dict) else {}
        os_id = payload.get("os") or payload.get("os_id")
        if os_id in (None, ""):
            raise HTTPException(status_code=400, detail="缺少重装系统 ID")
        return await reinstall_instance(
            db,
            instance,
            os_id=os_id,
            password=payload.get("password"),
            port=_int_or_none(payload.get("port")),
            part_type=_int_or_none(payload.get("part_type")),
        )
    if suffix == "/suspend" and method_upper in {"POST", "PUT"}:
        return await operate_instance(db, instance, "/off", "PUT", timeout=timeout)
    if suffix == "/unsuspend" and method_upper in {"POST", "PUT"}:
        return await operate_instance(db, instance, "/on", "PUT", timeout=timeout)
    if suffix == "" and method_upper in {"DELETE", "PUT", "PATCH"}:
        raise HTTPException(status_code=400, detail="魔方财务实例不支持通过本站直接销毁或改配，请使用魔方财务取消/升降级流程")
    if suffix and (suffix.startswith("/disks") or suffix.startswith("/bw")):
        raise HTTPException(status_code=400, detail=FINANCE_TICKET_MESSAGE)
    raise HTTPException(status_code=400, detail=f"魔方财务实例暂不支持该上游接口：{path}")
