from __future__ import annotations

import threading
import time
import socket
import urllib.error
import urllib.parse
import urllib.request
from decimal import Decimal, InvalidOperation

from common import shared as _shared
from common.debug_billing import is_on_demand_debug_enabled
from common.legacy_namespace import import_legacy_names
from HUAWEI.signing import json_dumps_bytes, sign_headers

# 过渡期保留旧单文件全局命名空间，避免大拆时影响华为云旧功能。
import_legacy_names(globals(), _shared)

from common.models import CloudApiError, ProjectPair, RegionPlan
from HUAWEI.flavor_helpers import (
    flavor_has_usable_size,
    flavor_identity,
    flavor_is_probably_available,
    flavor_lowest_score,
    flavor_matches_low_config_size,
    flavor_matches_target_size,
    flavor_preference_score,
    low_config_flavor_ids,
)

HUAWEI_SECURITY_GROUP_INGRESS_RULES = (
    {"protocol": "tcp", "port_range_min": 1, "port_range_max": 52},
    {"protocol": "tcp", "port_range_min": 54, "port_range_max": 65535},
    {"protocol": "udp", "port_range_min": 1, "port_range_max": 52},
    {"protocol": "udp", "port_range_min": 54, "port_range_max": 65535},
    {"protocol": "47", "port_range_min": None, "port_range_max": None},
    {"protocol": "icmp", "port_range_min": None, "port_range_max": None},
)


def _security_group_rule_port(value: Any) -> Optional[int]:
    try:
        return int(value)
    except (TypeError, ValueError):
        return None


def _security_group_rule_protocol(value: Any) -> Optional[str]:
    protocol = str(value or "").strip().lower()
    if protocol == "gre":
        return "47"
    return protocol or None


class HuaweiCloudClient:
    def __init__(
        self,
        ak: str,
        sk: str,
        domain_id: str = "",
        *,
        network_diagnostic: Optional[Callable[[str], None]] = None,
    ):
        self.ak = ak.strip()
        self.sk = sk.strip()
        self.domain_id = domain_id.strip()
        self._network_diagnostic = network_diagnostic
        self._network_diagnostic_seen: Set[str] = set()
        self._network_diagnostic_lock = threading.Lock()
        self._direct_hosts: Set[str] = set()
        self._request_pace_lock = threading.Lock()
        self._request_pace_interval = 0.0
        self._request_pace_next_at = 0.0
        self._request_rate_limit_cooldown = 65.0
        self._request_cooldown_until = 0.0

    def _emit_network_diagnostic(self, message: str) -> None:
        text = str(message or "").strip()
        if not text:
            return
        with self._network_diagnostic_lock:
            if text in self._network_diagnostic_seen:
                return
            self._network_diagnostic_seen.add(text)
        callback = self._network_diagnostic
        if callback is None:
            return
        try:
            callback(text)
        except Exception:
            pass

    @staticmethod
    def _network_error_summary(exc: BaseException) -> str:
        reason = getattr(exc, "reason", exc)
        text = " ".join(str(reason or exc).split())
        text = re.sub(r"(?i)(https?://)[^\s/@]+(?::[^\s/@]*)?@", r"\1***@", text)
        return text[:240] or type(exc).__name__

    @staticmethod
    def _proxy_route_configured(url: str) -> bool:
        parsed = urllib.parse.urlsplit(url)
        host = str(parsed.hostname or "").strip()
        if not host:
            return False
        try:
            if urllib.request.proxy_bypass(host):
                return False
        except Exception:
            pass
        try:
            proxies = {
                str(key or "").strip().lower(): str(value or "").strip()
                for key, value in urllib.request.getproxies().items()
            }
        except Exception:
            return False
        scheme = str(parsed.scheme or "https").strip().lower()
        return bool(proxies.get(scheme) or proxies.get("all"))

    @staticmethod
    def _is_proxy_connection_failure(exc: BaseException) -> bool:
        pending: List[BaseException] = [exc]
        seen: Set[int] = set()
        messages: List[str] = []
        while pending:
            current = pending.pop()
            marker = id(current)
            if marker in seen:
                continue
            seen.add(marker)
            messages.append(str(current or ""))
            if isinstance(current, (ConnectionRefusedError, socket.gaierror)):
                return True
            if getattr(current, "winerror", None) in {10061, 11001}:
                return True
            if getattr(current, "errno", None) in {61, 111}:
                return True
            for nested in (
                getattr(current, "reason", None),
                getattr(current, "__cause__", None),
                getattr(current, "__context__", None),
            ):
                if isinstance(nested, BaseException):
                    pending.append(nested)
        combined = " ".join(messages).casefold()
        return any(
            token in combined
            for token in (
                "connection refused",
                "actively refused",
                "积极拒绝",
                "tunnel connection failed",
                "proxy connection",
                "proxyerror",
            )
        )

    def configure_request_pacing(
        self,
        interval_seconds: float = 0.0,
        *,
        rate_limit_cooldown_seconds: float = 65.0,
    ) -> None:
        interval = max(0.0, float(interval_seconds or 0.0))
        cooldown = max(0.0, float(rate_limit_cooldown_seconds or 0.0))
        with self._request_pace_lock:
            now = time.monotonic()
            self._request_pace_interval = interval
            self._request_rate_limit_cooldown = cooldown
            self._request_pace_next_at = now
            self._request_cooldown_until = now

    def _wait_request_slot(self) -> None:
        while True:
            with self._request_pace_lock:
                interval = self._request_pace_interval
                if interval <= 0:
                    return
                now = time.monotonic()
                slot_at = max(self._request_pace_next_at, self._request_cooldown_until)
                if now >= slot_at:
                    self._request_pace_next_at = now + interval
                    return
                wait_seconds = slot_at - now
            if wait_seconds > 0:
                time.sleep(wait_seconds)

    def _defer_requests_after_rate_limit(self) -> None:
        with self._request_pace_lock:
            if self._request_pace_interval <= 0:
                return
            cooldown_until = time.monotonic() + self._request_rate_limit_cooldown
            self._request_cooldown_until = max(self._request_cooldown_until, cooldown_until)
            self._request_pace_next_at = max(self._request_pace_next_at, self._request_cooldown_until)

    def request(
        self,
        method: str,
        url: str,
        body: Optional[Any] = None,
        *,
        content_type: str = "application/json",
        use_domain_header: bool = False,
        timeout: int = 60,
        retry_attempts: Optional[int] = None,
    ) -> Tuple[int, Any, str]:
        payload = b""
        if body is not None:
            payload = json_dumps_bytes(body)
        headers = {
            "Content-Type": content_type,
            "X-Sdk-Date": time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()),
        }
        if use_domain_header and self.domain_id:
            headers["X-Domain-Id"] = self.domain_id
        signed_headers = sign_headers(method, url, headers, payload, self.ak, self.sk)
        method_upper = method.upper()
        retryable_methods = {"GET", "PUT", "DELETE"}
        attempts = retry_attempts if retry_attempts is not None else REQUEST_RETRY_ATTEMPTS
        if method_upper not in retryable_methods and retry_attempts is None:
            attempts = 1
        last_error: Optional[Exception] = None
        parsed_url = urllib.parse.urlsplit(url)
        request_host = str(parsed_url.hostname or parsed_url.netloc or "未知端点").strip()
        proxy_configured = self._proxy_route_configured(url)
        for attempt in range(1, attempts + 1):
            self._wait_request_slot()
            with self._network_diagnostic_lock:
                use_direct = request_host in self._direct_hosts
            while True:
                if use_direct and proxy_configured:
                    route = "直连兜底"
                elif use_direct or not proxy_configured:
                    route = "直连"
                else:
                    route = "代理"
                req = urllib.request.Request(
                    url,
                    data=payload if body is not None else None,
                    headers=signed_headers,
                    method=method_upper,
                )
                try:
                    if use_direct:
                        opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
                        response = opener.open(req, timeout=timeout)
                    else:
                        response = urllib.request.urlopen(req, timeout=timeout)
                    with response as resp:
                        raw = resp.read()
                        text = raw.decode("utf-8", errors="replace") if raw else ""
                        self._emit_network_diagnostic(f"目标={request_host}；方式={route}；结果=成功")
                        return resp.status, self._parse_json(text), text
                except urllib.error.HTTPError as exc:
                    self._emit_network_diagnostic(
                        f"目标={request_host}；方式={route}；结果=已连接，HTTP={exc.code}"
                    )
                    raw = exc.read()
                    text = raw.decode("utf-8", errors="replace") if raw else ""
                    detail = text or exc.reason or "HTTP error"
                    if exc.code == 429:
                        self._defer_requests_after_rate_limit()
                    if (
                        (method_upper in retryable_methods or retry_attempts is not None)
                        and exc.code in {429, 500, 502, 503, 504}
                        and attempt < attempts
                    ):
                        last_error = exc
                        break
                    if exc.code == 409:
                        detail = (
                            "HTTP 409 表示云平台资源冲突、配额冲突、重复规则/重复资源，或并发操作还未完成。"
                            f"原始返回: {detail}"
                        )
                    raise CloudApiError(f"HTTP {exc.code}: {detail}", status=exc.code, body=text) from exc
                except urllib.error.URLError as exc:
                    summary = self._network_error_summary(exc)
                    if proxy_configured and not use_direct and self._is_proxy_connection_failure(exc):
                        self._emit_network_diagnostic(
                            f"目标={request_host}；方式=代理；结果=失败，改用直连；原因={summary}"
                        )
                        with self._network_diagnostic_lock:
                            self._direct_hosts.add(request_host)
                        use_direct = True
                        continue
                    self._emit_network_diagnostic(
                        f"目标={request_host}；方式={route}；结果=失败；原因={summary}"
                    )
                    last_error = exc
                    if attempt < attempts:
                        break
                    raise CloudApiError(
                        f"网络连接失败: 连接华为云接口超时或本机网络不稳定，已重试 {attempts} 次；原因: {exc.reason}"
                    ) from exc
            time.sleep(1.5 * attempt)
        raise CloudApiError(f"网络连接失败: 连接华为云接口超时或本机网络不稳定；原因: {last_error}")

    @staticmethod
    def _parse_json(text: str) -> Any:
        if not text:
            return None
        try:
            return json.loads(text)
        except json.JSONDecodeError:
            return text

    def list_auth_projects(self) -> List[Dict[str, Any]]:
        _, data, _ = self.request(
            "GET",
            f"{IAM_ENDPOINT}/v3/auth/projects",
            content_type="application/json;charset=utf8",
            use_domain_header=bool(self.domain_id),
        )
        if isinstance(data, dict):
            return list(data.get("projects") or [])
        return []

    def list_auth_domains(self) -> List[Dict[str, Any]]:
        _, data, _ = self.request(
            "GET",
            f"{IAM_ENDPOINT}/v3/auth/domains",
            content_type="application/json;charset=utf8",
        )
        if isinstance(data, dict):
            return list(data.get("domains") or [])
        return []

    def list_projects_by_name(self, region: str) -> List[Dict[str, Any]]:
        region_name = urllib.parse.quote(str(region or "").strip(), safe="")
        if not region_name:
            return []
        _, data, _ = self.request(
            "GET",
            f"{IAM_ENDPOINT}/v3/projects?name={region_name}",
            content_type="application/json;charset=utf8",
            use_domain_header=bool(self.domain_id),
            timeout=25,
            retry_attempts=3,
        )
        if isinstance(data, dict):
            return list(data.get("projects") or [])
        return []

    def list_projects(self) -> List[Dict[str, Any]]:
        _, data, _ = self.request(
            "GET",
            f"{IAM_ENDPOINT}/v3/projects",
            content_type="application/json;charset=utf8",
            use_domain_header=bool(self.domain_id),
            timeout=25,
            retry_attempts=3,
        )
        if isinstance(data, dict):
            return list(data.get("projects") or [])
        return []

    def _remember_project_pair(
        self,
        found: Dict[str, ProjectPair],
        wanted: set[str],
        project: Dict[str, Any],
    ) -> None:
        region = (project.get("name") or "").strip()
        project_id = (project.get("id") or "").strip()
        if not region or not project_id or project.get("is_domain"):
            return
        if region in wanted:
            found[region] = ProjectPair(region=region, project_id=project_id)
        domain_id = str(project.get("domain_id") or "").strip()
        if domain_id and not self.domain_id:
            self.domain_id = domain_id

    def selected_project_pairs(self, plans: Sequence[RegionPlan]) -> Dict[str, ProjectPair]:
        ordered_regions = list(dict.fromkeys(plan.region for plan in plans if plan.region))
        wanted = set(ordered_regions)
        found: Dict[str, ProjectPair] = {}
        def remember_from_projects(projects: Sequence[Dict[str, Any]]) -> None:
            for project in projects:
                self._remember_project_pair(found, wanted, project)

        try:
            self.discover_domain_id()
        except Exception:
            pass
        try:
            remember_from_projects(self.list_auth_projects())
        except Exception:
            pass
        # 授权项目列表偶尔不会返回某些已初始化地区；继续按 regionId 查询项目，
        # 让主界面和自动购买子流程使用同一套 IAM 项目 ID 兜底逻辑。
        missing = [region for region in ordered_regions if region not in found]
        for region in missing:
            try:
                remember_from_projects(self.list_projects_by_name(region))
            except Exception:
                continue
        missing = [region for region in ordered_regions if region not in found]
        if missing:
            try:
                remember_from_projects(self.list_projects())
            except Exception:
                pass
        missing = [region for region in ordered_regions if region not in found]
        for region in missing:
            remember_from_projects(self.warmup_region(region))
        missing = [region for region in ordered_regions if region not in found]
        if missing:
            try:
                remember_from_projects(self.list_auth_projects())
            except Exception:
                pass
        missing = [region for region in ordered_regions if region not in found]
        for region in missing:
            try:
                remember_from_projects(self.list_projects_by_name(region))
            except Exception:
                continue
        missing = [region for region in ordered_regions if region not in found]
        if missing:
            try:
                remember_from_projects(self.list_projects())
            except Exception:
                pass
        return found

    def warmup_region(self, region: str) -> List[Dict[str, Any]]:
        region = str(region or "").strip()
        if not region:
            return []
        iam_endpoint = f"https://iam.{region}.myhuaweicloud.com"
        try:
            self.request(
                "GET",
                f"{iam_endpoint}/v3/auth/domains",
                content_type="application/json;charset=utf8",
                timeout=25,
                retry_attempts=2,
            )
        except Exception:
            pass
        try:
            _, data, _ = self.request(
                "GET",
                f"{iam_endpoint}/v3/auth/projects",
                content_type="application/json;charset=utf8",
                use_domain_header=bool(self.domain_id),
                timeout=25,
                retry_attempts=2,
            )
        except Exception:
            return []
        if isinstance(data, dict):
            return list(data.get("projects") or [])
        return []

    def discover_domain_id(self) -> str:
        if self.domain_id:
            return self.domain_id
        try:
            for domain in self.list_auth_domains():
                domain_id = str(domain.get("id") or "").strip()
                if domain_id:
                    self.domain_id = domain_id
                    return domain_id
        except Exception:
            pass
        for project in self.list_auth_projects():
            domain_id = str(project.get("domain_id") or "").strip()
            if domain_id:
                self.domain_id = domain_id
                return domain_id
        return ""

    @staticmethod
    def _linux_flavor_spec(flavor_id: str) -> str:
        value = str(flavor_id or FLAVOR_ID).strip()
        if not value:
            value = FLAVOR_ID
        if value.lower().endswith((".linux", ".win")):
            return value
        return f"{value}.linux"

    def list_subscription_resources(
        self,
        only_main_resource: bool,
        log: Optional[Callable[[str], None]] = None,
    ) -> List[Dict[str, Any]]:
        try:
            self.discover_domain_id()
        except Exception as exc:
            if log:
                log(f"自动识别账号ID失败，继续尝试查询包年包月资源: {exc}")

        last_error: Optional[CloudApiError] = None
        for path in SUBSCRIPTION_QUERY_PATHS:
            try:
                return self._list_subscription_resources_with_path(path, only_main_resource)
            except CloudApiError as exc:
                last_error = exc
                if exc.status == 404:
                    continue
                raise
        if last_error:
            raise last_error
        return []

    def _list_subscription_resources_with_path(
        self,
        path: str,
        only_main_resource: bool,
    ) -> List[Dict[str, Any]]:
        resources: List[Dict[str, Any]] = []
        offset = 0
        limit = 500
        while True:
            body = {
                "only_main_resource": 1 if only_main_resource else 0,
                "offset": offset,
                "limit": limit,
            }
            _, data, _ = self.request(
                "POST",
                f"{BSS_ENDPOINT}{path}",
                body,
                use_domain_header=bool(self.domain_id),
                timeout=25,
                retry_attempts=4,
            )
            if not isinstance(data, dict):
                raise CloudApiError(f"包年包月资源查询返回格式异常: {data}")
            page = list(data.get("data") or [])
            resources.extend(page)
            total = int(data.get("total_count") or len(resources))
            if len(resources) >= total or not page:
                return resources
            offset += limit

    def get_available_balance(self) -> Decimal:
        _, data, _ = self.request(
            "GET",
            f"{BSS_ENDPOINT}/v2/accounts/customer-accounts/balances",
            use_domain_header=bool(self.domain_id),
            timeout=20,
            retry_attempts=2,
        )
        if not isinstance(data, dict):
            raise CloudApiError("账户余额返回格式异常")
        balances = data.get("account_balances")
        if not isinstance(balances, list):
            raise CloudApiError("账户余额字段缺失")

        def amount_from(item: Dict[str, Any]) -> Optional[Decimal]:
            raw = item.get("amount")
            if raw in (None, ""):
                return None
            try:
                amount = Decimal(str(raw).replace(",", ""))
            except (InvalidOperation, ValueError):
                return None
            measure_id = item.get("measure_id") or item.get("measure_unit") or data.get("measure_id")
            if str(measure_id) == "2":
                amount *= Decimal("0.1")
            elif str(measure_id) == "3":
                amount *= Decimal("0.01")
            return amount

        cash_amounts: List[Decimal] = []
        fallback_amounts: List[Decimal] = []
        for item in balances:
            if not isinstance(item, dict):
                continue
            amount = amount_from(item)
            if amount is None:
                continue
            fallback_amounts.append(amount)
            if str(item.get("account_type") or "").strip() == "1":
                cash_amounts.append(amount)
        values = cash_amounts or fallback_amounts
        if not values:
            raise CloudApiError("账户余额为空")
        return sum(values, Decimal("0"))

    def unsubscribe_subscription_resources(
        self,
        resource_ids: Sequence[str],
        unsubscribe_type: int,
        reason: str,
    ) -> List[Dict[str, Any]]:
        try:
            self.discover_domain_id()
        except Exception:
            pass
        results: List[Dict[str, Any]] = []
        for group in chunks([rid for rid in resource_ids if rid], 10):
            body = {
                "resource_ids": list(group),
                "unsubscribe_type": unsubscribe_type,
                "unsubscribe_reason_type": 5,
                "unsubscribe_reason": reason or "不再使用",
            }
            _, data, text = self.request(
                "POST",
                f"{BSS_ENDPOINT}{SUBSCRIPTION_UNSUBSCRIBE_PATH}",
                body,
                use_domain_header=bool(self.domain_id),
                timeout=60,
            )
            results.append(data if isinstance(data, dict) else {"raw": text})
        return results

    @staticmethod
    def vpc_endpoint(region: str) -> str:
        return f"https://vpc.{region}.myhuaweicloud.com"

    @staticmethod
    def ecs_endpoint(region: str) -> str:
        return f"https://ecs.{region}.myhuaweicloud.com"

    @staticmethod
    def ims_endpoint(region: str) -> str:
        return f"https://ims.{region}.myhuaweicloud.com"

    @staticmethod
    def eip_endpoint(region: str) -> str:
        return f"https://eip.{region}.myhuaweicloud.com"

    def build_public_ip_list_proof(self, pair: ProjectPair) -> Dict[str, str]:
        """Build a short-lived signature for the fixed read-only EIP list request."""
        url = f"{self.vpc_endpoint(pair.region)}/v1/{pair.project_id}/publicips?limit=2000"
        headers = {
            "Content-Type": "application/json",
            "X-Sdk-Date": time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()),
        }
        signed_headers = sign_headers("GET", url, headers, b"", self.ak, self.sk)
        return {
            "region_id": pair.region,
            "project_id": pair.project_id,
            "authorization": signed_headers["Authorization"],
            "x_sdk_date": signed_headers["X-Sdk-Date"],
        }

    def list_vpcs(self, pair: ProjectPair, *, timeout: int = 60, retry_attempts: Optional[int] = None) -> List[Dict[str, Any]]:
        url = f"{self.vpc_endpoint(pair.region)}/v1/{pair.project_id}/vpcs?limit=2000"
        _, data, _ = self.request("GET", url, timeout=timeout, retry_attempts=retry_attempts)
        if not isinstance(data, dict):
            raise CloudApiError(f"VPC列表返回格式异常: {data}")
        vpcs = list(data.get("vpcs") or [])
        for item in vpcs:
            item["_region"] = pair.region
            item["_project_id"] = pair.project_id
        return vpcs

    def create_vpc(self, pair: ProjectPair, name: str) -> Dict[str, Any]:
        body = {"vpc": {"name": name, "cidr": VPC_CIDR, "description": "wawacloud wtdiy v2"}}
        url = f"{self.vpc_endpoint(pair.region)}/v1/{pair.project_id}/vpcs"
        _, data, _ = self.request("POST", url, body)
        if not isinstance(data, dict) or not isinstance(data.get("vpc"), dict):
            raise CloudApiError(f"创建VPC返回格式异常: {data}")
        return data["vpc"]

    def delete_vpc(self, pair: ProjectPair, vpc_id: str) -> None:
        url = f"{self.vpc_endpoint(pair.region)}/v1/{pair.project_id}/vpcs/{vpc_id}"
        self.request("DELETE", url, timeout=60)

    def wait_vpc_deleted(
        self,
        pair: ProjectPair,
        vpc_id: str,
        emit: Callable[[str], None],
        *,
        timeout: int = 120,
        interval: int = 4,
    ) -> bool:
        deadline = time.time() + timeout
        while time.time() < deadline:
            try:
                current = self.list_vpcs(pair, timeout=20, retry_attempts=2)
            except Exception as exc:
                emit(f"VPC {vpc_id} 删除等待查询提示: {exc}")
                time.sleep(interval)
                continue
            if not any(str(item.get("id") or "") == vpc_id for item in current):
                return True
            time.sleep(interval)
        emit(f"VPC {vpc_id} 删除等待超时，稍后刷新资源库确认。")
        return False

    def list_subnets(self, pair: ProjectPair, *, timeout: int = 60, retry_attempts: Optional[int] = None) -> List[Dict[str, Any]]:
        url = f"{self.vpc_endpoint(pair.region)}/v1/{pair.project_id}/subnets?limit=2000"
        _, data, _ = self.request("GET", url, timeout=timeout, retry_attempts=retry_attempts)
        if not isinstance(data, dict):
            raise CloudApiError(f"子网列表返回格式异常: {data}")
        subnets = list(data.get("subnets") or [])
        for item in subnets:
            item["_region"] = pair.region
            item["_project_id"] = pair.project_id
        return subnets

    def create_subnet(self, pair: ProjectPair, vpc_id: str, name: str, availability_zone: str = "") -> Dict[str, Any]:
        subnet_body = {
            "name": name,
            "description": "wawacloud wtdiy v2",
            "cidr": SUBNET_CIDR,
            "gateway_ip": SUBNET_GATEWAY,
            "dhcp_enable": True,
            "primary_dns": "100.125.1.250",
            "secondary_dns": "114.114.114.114",
            "dnsList": ["100.125.1.250", "114.114.114.114"],
            "vpc_id": vpc_id,
        }
        if availability_zone:
            subnet_body["availability_zone"] = availability_zone
        body = {"subnet": subnet_body}
        url = f"{self.vpc_endpoint(pair.region)}/v1/{pair.project_id}/subnets"
        _, data, _ = self.request("POST", url, body)
        if not isinstance(data, dict) or not isinstance(data.get("subnet"), dict):
            raise CloudApiError(f"创建子网返回格式异常: {data}")
        return data["subnet"]

    def delete_subnet(self, pair: ProjectPair, subnet_id: str) -> None:
        url = f"{self.vpc_endpoint(pair.region)}/v1/{pair.project_id}/subnets/{subnet_id}"
        self.request("DELETE", url, timeout=60)

    def delete_port(self, pair: ProjectPair, port_id: str) -> None:
        url = f"{self.vpc_endpoint(pair.region)}/v1/{pair.project_id}/ports/{port_id}"
        self.request("DELETE", url, timeout=60)

    def update_port(self, pair: ProjectPair, port_id: str, body: Dict[str, Any]) -> Dict[str, Any]:
        url = f"{self.vpc_endpoint(pair.region)}/v1/{pair.project_id}/ports/{port_id}"
        _, data, text = self.request("PUT", url, body, timeout=60)
        return data if isinstance(data, dict) else {"raw": text}

    def create_virtual_ip_port(
        self,
        pair: ProjectPair,
        subnet_network_id: str,
        private_ip: str,
        name: str,
        security_group_id: str = "",
    ) -> Dict[str, Any]:
        # 华为云虚拟 IP 本质是 device_owner=neutron:VIP_PORT 的端口，后续再通过 allowed_address_pairs 绑定到 ECS 网卡。
        port_body: Dict[str, Any] = {
            "network_id": subnet_network_id,
            "fixed_ips": [{"ip_address": private_ip}],
            "device_owner": "neutron:VIP_PORT",
            "name": name,
            "admin_state_up": True,
        }
        if security_group_id:
            port_body["security_groups"] = [security_group_id]
        body = {"port": port_body}
        urls = [
            f"{self.vpc_endpoint(pair.region)}/v1/{pair.project_id}/ports",
            f"{self.vpc_endpoint(pair.region)}/v2.0/ports",
        ]
        last_error: Optional[CloudApiError] = None
        for url in urls:
            try:
                _, data, text = self.request("POST", url, body, timeout=60, retry_attempts=3)
            except CloudApiError as exc:
                last_error = exc
                if exc.status in {404, 405}:
                    continue
                raise
            if isinstance(data, dict) and isinstance(data.get("port"), dict):
                return data["port"]
            return {"raw": text}
        if last_error:
            raise last_error
        raise CloudApiError("创建虚拟IP端口失败：没有可用的 VPC port API。")

    def show_subnet(self, pair: ProjectPair, subnet_id: str) -> Dict[str, Any]:
        url = f"{self.vpc_endpoint(pair.region)}/v1/{pair.project_id}/subnets/{subnet_id}"
        _, data, _ = self.request("GET", url)
        if not isinstance(data, dict) or not isinstance(data.get("subnet"), dict):
            raise CloudApiError(f"查询子网返回格式异常: {data}")
        return data["subnet"]

    def wait_subnet_deleted(
        self,
        pair: ProjectPair,
        subnet_id: str,
        emit: Callable[[str], None],
        *,
        timeout: int = 90,
        interval: int = 3,
    ) -> bool:
        deadline = time.time() + timeout
        while time.time() < deadline:
            try:
                self.show_subnet(pair, subnet_id)
            except CloudApiError as exc:
                if exc.status == 404:
                    return True
                raise
            time.sleep(interval)
        emit(f"子网 {subnet_id} 删除等待超时，继续尝试下一步。")
        return False

    def wait_subnet_active(
        self,
        pair: ProjectPair,
        subnet_id: str,
        emit: Callable[[str], None],
        *,
        timeout: int = 180,
    ) -> Dict[str, Any]:
        deadline = time.time() + timeout
        last: Dict[str, Any] = {}
        while time.time() < deadline:
            last = self.show_subnet(pair, subnet_id)
            status = str(last.get("status") or "").upper()
            emit(f"步骤: 子网状态 {status or 'UNKNOWN'}")
            if status in {"ACTIVE", "NORMAL", ""}:
                return last
            if status in {"ERROR", "DOWN"}:
                raise CloudApiError(f"子网状态异常: {json.dumps(last, ensure_ascii=False)}")
            time.sleep(5)
        raise TimeoutError(f"子网等待ACTIVE超时: {json.dumps(last, ensure_ascii=False)}")

    def list_security_groups(self, pair: ProjectPair, *, timeout: int = 60, retry_attempts: Optional[int] = None) -> List[Dict[str, Any]]:
        url = f"{self.vpc_endpoint(pair.region)}/v1/{pair.project_id}/security-groups?limit=2000"
        _, data, _ = self.request("GET", url, timeout=timeout, retry_attempts=retry_attempts)
        if not isinstance(data, dict):
            raise CloudApiError(f"安全组列表返回格式异常: {data}")
        groups = list(data.get("security_groups") or [])
        for item in groups:
            item["_region"] = pair.region
            item["_project_id"] = pair.project_id
        return groups

    def delete_security_group(self, pair: ProjectPair, security_group_id: str) -> None:
        v3_url = f"{self.vpc_endpoint(pair.region)}/v3/{pair.project_id}/vpc/security-groups/{security_group_id}"
        try:
            self.request("DELETE", v3_url, timeout=60)
            return
        except CloudApiError as v3_error:
            v1_url = f"{self.vpc_endpoint(pair.region)}/v1/{pair.project_id}/security-groups/{security_group_id}"
            try:
                self.request("DELETE", v1_url, timeout=60)
                return
            except CloudApiError as v1_error:
                raise CloudApiError(f"V3删除失败: {v3_error}; V1删除失败: {v1_error}") from v1_error

    def delete_security_group_rule(self, pair: ProjectPair, rule_id: str) -> None:
        url = f"{self.vpc_endpoint(pair.region)}/v1/{pair.project_id}/security-group-rules/{rule_id}"
        self.request("DELETE", url, timeout=60)

    def list_security_group_rules(self, pair: ProjectPair, security_group_id: str) -> List[Dict[str, Any]]:
        query = urllib.parse.urlencode({"security_group_id": security_group_id, "limit": 2000})
        url = f"{self.vpc_endpoint(pair.region)}/v1/{pair.project_id}/security-group-rules?{query}"
        _, data, _ = self.request("GET", url, timeout=60)
        if not isinstance(data, dict):
            return []
        return list(data.get("security_group_rules") or [])

    def wait_security_group_deleted(
        self,
        pair: ProjectPair,
        security_group_id: str,
        emit: Callable[[str], None],
        *,
        timeout: int = 90,
        interval: int = 3,
    ) -> bool:
        deadline = time.time() + timeout
        while time.time() < deadline:
            try:
                current = self.list_security_groups(pair, timeout=20, retry_attempts=2)
            except Exception as exc:
                emit(f"安全组 {security_group_id} 删除等待查询提示: {exc}")
                time.sleep(interval)
                continue
            if not any(str(item.get("id") or "") == security_group_id for item in current):
                return True
            time.sleep(interval)
        emit(f"安全组 {security_group_id} 删除等待超时，稍后刷新资源库确认。")
        return False

    def create_security_group(self, pair: ProjectPair, vpc_id: str, name: str) -> Dict[str, Any]:
        body = {"security_group": {"name": name, "vpc_id": vpc_id, "description": "wawacloud wtdiy v2"}}
        url = f"{self.vpc_endpoint(pair.region)}/v1/{pair.project_id}/security-groups"
        _, data, _ = self.request("POST", url, body)
        if not isinstance(data, dict) or not isinstance(data.get("security_group"), dict):
            raise CloudApiError(f"创建安全组返回格式异常: {data}")
        return data["security_group"]

    def create_security_group_rule(self, pair: ProjectPair, rule: Dict[str, Any]) -> Dict[str, Any]:
        url = f"{self.vpc_endpoint(pair.region)}/v1/{pair.project_id}/security-group-rules"
        _, data, _ = self.request("POST", url, {"security_group_rule": rule})
        if isinstance(data, dict):
            return data.get("security_group_rule") or data
        return {"raw": data}

    def open_security_group(self, pair: ProjectPair, security_group_id: str, emit: Callable[[str], None]) -> None:
        rules = []
        for item in HUAWEI_SECURITY_GROUP_INGRESS_RULES:
            rule = {
                **item,
                "direction": "ingress",
                "ethertype": "IPv4",
                "remote_ip_prefix": "0.0.0.0/0",
                "security_group_id": security_group_id,
            }
            rules.append({key: value for key, value in rule.items() if value is not None})
        desired_keys = {
            (rule["direction"], _security_group_rule_protocol(rule["protocol"]), rule.get("port_range_min"), rule.get("port_range_max"), rule["ethertype"], rule["remote_ip_prefix"])
            for rule in rules
        }
        try:
            existing_security_group_rules = self.list_security_group_rules(pair, security_group_id)
        except CloudApiError as exc:
            emit(f"安全组旧入方向规则查询失败，继续补齐新规则: {exc}")
            existing_security_group_rules = []
        for existing in existing_security_group_rules:
            if str(existing.get("direction") or "").lower() != "ingress":
                continue
            key = (
                str(existing.get("direction") or "").lower(),
                _security_group_rule_protocol(existing.get("protocol")),
                _security_group_rule_port(existing.get("port_range_min")),
                _security_group_rule_port(existing.get("port_range_max")),
                existing.get("ethertype") or "IPv4",
                existing.get("remote_ip_prefix") or "",
            )
            if key in desired_keys:
                continue
            rule_id = str(existing.get("id") or "")
            if not rule_id:
                continue
            try:
                self.delete_security_group_rule(pair, rule_id)
                emit("安全组旧入方向规则已移除。")
            except CloudApiError as exc:
                emit(f"安全组旧入方向规则移除失败，继续补齐新规则: {exc}")
        for rule in rules:
            port_min = rule.get("port_range_min")
            port_max = rule.get("port_range_max")
            port_label = "全部" if port_min is None or port_max is None else f"{port_min}-{port_max}"
            label = f"{rule['direction']} IPv4 {rule['protocol'].upper()} {port_label} 0.0.0.0/0"
            for attempt in range(1, REQUEST_RETRY_ATTEMPTS + 1):
                try:
                    self.create_security_group_rule(pair, rule)
                    break
                except CloudApiError as exc:
                    text = f"{exc} {exc.body}".lower()
                    if exc.status == 409 or "already" in text or "exist" in text or "duplicate" in text:
                        emit(f"安全组规则已存在: {label}")
                        break
                    if attempt < REQUEST_RETRY_ATTEMPTS:
                        emit(f"安全组规则创建重试 {attempt}/{REQUEST_RETRY_ATTEMPTS}: {label} {exc}")
                        time.sleep(2 * attempt)
                        continue
                    raise CloudApiError(f"安全组规则创建失败: {label} {exc}", status=exc.status, body=exc.body) from exc

    def list_availability_zones(self, pair: ProjectPair, emit: Optional[Callable[[str], None]] = None) -> List[str]:
        urls = [
            f"{self.ecs_endpoint(pair.region)}/v1/{pair.project_id}/availability-zones",
            f"{self.ecs_endpoint(pair.region)}/v2.1/{pair.project_id}/os-availability-zone",
            f"{self.ecs_endpoint(pair.region)}/v2/{pair.project_id}/os-availability-zone",
        ]
        errors = []
        for url in urls:
            try:
                _, data, _ = self.request("GET", url)
                zones = self._parse_availability_zones(data)
                if zones:
                    return zones
            except Exception as exc:
                errors.append(str(exc))
                if emit:
                    emit(f"可用区接口不可用，尝试下一个: {exc}")
        if emit and errors:
            emit("可用区查询不可用，将不传 availability_zone，由华为云自动选择。")
        return []

    @staticmethod
    def _parse_availability_zones(data: Any) -> List[str]:
        if not isinstance(data, dict):
            return []
        raw_zones = data.get("availability_zones") or data.get("availabilityZoneInfo") or data.get("availability_zones_info") or []
        zones = []
        for item in raw_zones:
            if not isinstance(item, dict):
                continue
            zone_name = item.get("availability_zone_id") or item.get("zoneName") or item.get("name") or item.get("zone_name")
            zone_state = item.get("zoneState") or item.get("zone_state") or {}
            available = item.get("available")
            zone_type = str(item.get("type") or "").lower()
            if isinstance(zone_state, dict) and zone_state.get("available") is False:
                continue
            if available is False:
                continue
            if zone_type and zone_type != "center":
                continue
            if zone_name:
                zones.append(str(zone_name))
        return zones

    def find_image_id(self, pair: ProjectPair) -> Tuple[str, str]:
        queries = [
            "__imagetype=gold&__platform=Debian&status=active&limit=100",
            "__imagetype=gold&name=Debian&status=active&limit=100",
            "__imagetype=gold&__platform=Ubuntu&status=active&limit=100",
            "__imagetype=gold&name=Ubuntu&status=active&limit=100",
            "__imagetype=gold&status=active&limit=500",
        ]
        images: List[Dict[str, Any]] = []
        seen_ids = set()
        for query in queries:
            url = f"{self.ims_endpoint(pair.region)}/v2/cloudimages?{query}"
            try:
                _, data, _ = self.request("GET", url)
            except CloudApiError:
                continue
            if isinstance(data, dict):
                for item in data.get("images") or []:
                    if not isinstance(item, dict):
                        continue
                    image_id = item.get("id")
                    if image_id and image_id in seen_ids:
                        continue
                    if image_id:
                        seen_ids.add(image_id)
                    images.append(item)
        apt_images = [
            img
            for img in images
            if any(
                os_name in f"{img.get('name') or ''} {img.get('__os_version') or ''} {img.get('__platform') or ''}".lower()
                for os_name in ("debian", "ubuntu")
            )
        ]
        if not apt_images:
            raise CloudApiError("未查询到 Debian/Ubuntu 公共镜像，无法保证 huawei3.sh 可正常 apt 安装。")

        def score(img: Dict[str, Any]) -> Tuple[int, str]:
            text = f"{img.get('name') or ''} {img.get('__os_version') or ''}".lower()
            value = 0
            if "debian 12" in text or "debian-12" in text:
                value += 100
            if "debian 11" in text or "debian-11" in text:
                value += 80
            if "debian" in text:
                value += 60
            if "ubuntu 22" in text or "ubuntu-22" in text:
                value += 40
            if "ubuntu 20" in text or "ubuntu-20" in text:
                value += 30
            if "ubuntu" in text:
                value += 20
            if "64" in text:
                value += 10
            return value, str(img.get("name") or "")

        apt_images.sort(key=score, reverse=True)
        image = apt_images[0]
        image_id = image.get("id") or ""
        image_name = image.get("name") or image.get("__os_version") or image_id
        if not image_id:
            raise CloudApiError(f"镜像返回缺少ID: {image}")
        return str(image_id), str(image_name)

    def list_flavors(self, pair: ProjectPair, availability_zone: str) -> List[Dict[str, Any]]:
        query_parts: Dict[str, str] = {}
        if availability_zone:
            query_parts["availability_zone"] = availability_zone
        query = urllib.parse.urlencode(query_parts)
        suffix = f"?{query}" if query else ""
        url = f"{self.ecs_endpoint(pair.region)}/v1/{pair.project_id}/cloudservers/flavors{suffix}"
        _, data, _ = self.request("GET", url)
        if not isinstance(data, dict):
            return []
        return list(data.get("flavors") or [])

    def choose_flavor_and_zone(
        self,
        pair: ProjectPair,
        zones: Sequence[str],
        emit: Callable[[str], None],
        flavor_mode: str = FLAVOR_MODE_FIXED,
    ) -> Tuple[str, str]:
        candidates = self.zone_flavor_candidates(pair, zones, emit, flavor_mode)
        if candidates:
            return candidates[0]
        return "", FLAVOR_ID

    def _lowest_flavors(self, flavors: Sequence[Dict[str, Any]]) -> List[Dict[str, Any]]:
        usable = [
            item
            for item in flavors
            if flavor_has_usable_size(item) and flavor_is_probably_available(item) and flavor_identity(item)
        ]
        return sorted(usable, key=flavor_lowest_score)

    def _fixed_zone_flavor_candidates(
        self,
        pair: ProjectPair,
        zones: Sequence[str],
        emit: Callable[[str], None],
    ) -> List[Tuple[str, str]]:
        candidates: List[Tuple[str, str]] = []
        seen = set()

        def add(zone: str, flavor_id: str) -> None:
            key = (zone, flavor_id)
            if flavor_id and key not in seen:
                seen.add(key)
                candidates.append(key)

        search_zones = list(zones) or [""]
        for zone in search_zones:
            try:
                flavors = self.list_flavors(pair, zone)
            except Exception as exc:
                emit(f"规格列表查询失败 {zone or '自动AZ'}: {exc}")
                continue
            exact = [
                item
                for item in flavors
                if flavor_identity(item) == FLAVOR_ID and flavor_is_probably_available(item)
            ]
            if exact:
                add(zone, FLAVOR_ID)
                continue
            for item in flavors:
                if flavor_identity(item) == FLAVOR_ID:
                    emit(f"{zone or '自动AZ'} 发现 {FLAVOR_ID}，但状态可能不可用，继续尝试其他可用区。")

        add("", FLAVOR_ID)
        if not candidates:
            emit(f"未在规格列表中找到可用 {FLAVOR_ID}，仍按 {FLAVOR_ID} 直接尝试创建；不会改买其他更贵规格。")
        return candidates

    def _wawacloud_low_zone_flavor_candidates(
        self,
        pair: ProjectPair,
        zones: Sequence[str],
        emit: Callable[[str], None],
    ) -> List[Tuple[str, str]]:
        preferred_flavors = low_config_flavor_ids(pair.region)
        candidates: List[Tuple[str, str]] = []
        seen = set()

        def add(zone: str, flavor_id: str) -> None:
            key = (zone, flavor_id)
            if flavor_id and key not in seen:
                seen.add(key)
                candidates.append(key)

        search_zones = list(dict.fromkeys(list(zones) or [""]))
        for zone in search_zones:
            try:
                flavors = self.list_flavors(pair, zone)
            except Exception as exc:
                emit(f"1vCPUs | 1GiB 规格列表查询失败 {zone or '自动AZ'}: {exc}")
                continue
            by_id = {flavor_identity(item): item for item in flavors if flavor_identity(item)}
            for flavor_id in preferred_flavors:
                item = by_id.get(flavor_id)
                if item and flavor_is_probably_available(item):
                    add(zone, flavor_id)
                elif item:
                    emit(f"{zone or '自动AZ'} 发现 1vCPUs | 1GiB {flavor_id}，但状态可能不可用，继续尝试其他可用区。")

        for flavor_id in preferred_flavors:
            add("", flavor_id)
        preview = ", ".join(f"{zone or '自动AZ'}/{flavor}" for zone, flavor in candidates[:6])
        if len(candidates) > 6:
            preview += f", ... 共{len(candidates)}组"
        emit(f"1vCPUs | 1GiB 候选 {preview}")
        return candidates

    def _wawacloud_twenty_zone_flavor_candidates(
        self,
        pair: ProjectPair,
        zones: Sequence[str],
        emit: Callable[[str], None],
    ) -> List[Tuple[str, str]]:
        candidates: List[Tuple[str, str]] = []
        seen = set()

        def add(zone: str, flavor_id: str) -> None:
            key = (zone, flavor_id)
            if flavor_id and key not in seen:
                seen.add(key)
                candidates.append(key)

        search_zones = list(dict.fromkeys(list(zones) or [""]))
        preferred_flavors = list(TWENTY_CONFIG_FLAVORS)
        for zone in search_zones:
            try:
                flavors = self.list_flavors(pair, zone)
            except Exception as exc:
                emit(f"20规格查询可用区规格失败 {zone or '自动AZ'}: {exc}")
                continue
            by_id = {flavor_identity(item): item for item in flavors if flavor_identity(item)}
            for flavor_id in preferred_flavors:
                item = by_id.get(flavor_id)
                if item and flavor_is_probably_available(item):
                    add(zone, flavor_id)
                elif item:
                    emit(f"{zone or '自动AZ'} 发现20规格候选 {flavor_id}，但状态可能不可用，继续尝试其他可用区。")

        for flavor_id in preferred_flavors:
            add("", flavor_id)
        preview = ", ".join(f"{zone or '自动AZ'}/{flavor}" for zone, flavor in candidates[:6])
        if len(candidates) > 6:
            preview += f", ... 共{len(candidates)}组"
        emit(f"20规格候选 {preview}")
        return candidates

    def zone_flavor_candidates(
        self,
        pair: ProjectPair,
        zones: Sequence[str],
        emit: Callable[[str], None],
        flavor_mode: str = FLAVOR_MODE_FIXED,
    ) -> List[Tuple[str, str]]:
        if flavor_mode == FLAVOR_MODE_WAWACLOUD_LOW:
            return self._wawacloud_low_zone_flavor_candidates(pair, zones, emit)
        if flavor_mode == FLAVOR_MODE_WAWACLOUD_20:
            return self._wawacloud_twenty_zone_flavor_candidates(pair, zones, emit)
        if flavor_mode != FLAVOR_MODE_LOWEST:
            return self._fixed_zone_flavor_candidates(pair, zones, emit)

        candidates: List[Tuple[str, str]] = []
        seen = set()

        def add(zone: str, flavor_id: str) -> None:
            key = (zone, flavor_id)
            if flavor_id and key not in seen:
                seen.add(key)
                candidates.append(key)

        search_zones = list(dict.fromkeys(list(zones) or [""]))
        scored: List[Tuple[Tuple[int, int, int, str], str, str]] = []
        for zone in search_zones:
            try:
                flavors = self.list_flavors(pair, zone)
            except Exception as exc:
                emit(f"规格列表查询失败 {zone or '自动AZ'}: {exc}")
                continue
            for item in self._lowest_flavors(flavors)[:8]:
                scored.append((flavor_lowest_score(item), zone, flavor_identity(item)))
        scored.sort(key=lambda item: item[0])
        for _score, zone, flavor_id in scored:
            add(zone, flavor_id)
        if not candidates:
            emit("未在规格列表中找到可用最低配规格，将回退按 x1.2u.2g 直接尝试创建。")
            add("", FLAVOR_ID)
        else:
            preview = ", ".join(f"{zone or '自动AZ'}/{flavor}" for zone, flavor in candidates[:6])
            if len(candidates) > 6:
                preview += f", ... 共{len(candidates)}组"
            emit(f"最低配候选 {preview}")
        return candidates

    def choose_fallback_flavor(
        self,
        pair: ProjectPair,
        zones: Sequence[str],
        current_flavor_id: str,
        emit: Callable[[str], None],
    ) -> Tuple[str, str]:
        candidates: List[Tuple[Tuple[int, str], str, Dict[str, Any]]] = []
        for zone in list(zones) or [""]:
            try:
                flavors = self.list_flavors(pair, zone)
            except Exception as exc:
                emit(f"备用规格查询失败 {zone or '自动AZ'}: {exc}")
                continue
            for item in flavors:
                item_id = flavor_identity(item)
                if item_id == current_flavor_id:
                    continue
                if flavor_matches_target_size(item) and flavor_is_probably_available(item):
                    candidates.append((flavor_preference_score(item), zone, item))
        if not candidates:
            return "", ""
        candidates.sort(reverse=True)
        _score, zone, flavor = candidates[0]
        return zone, flavor_identity(flavor)

    def create_server(
        self,
        pair: ProjectPair,
        name: str,
        availability_zone: str,
        flavor_id: str,
        root_volume_type: str,
        image_id: str,
        vpc_id: str,
        subnet_network_id: str,
        security_group_id: str,
        password: str,
        nic_private_ips: Optional[Sequence[str]] = None,
        on_demand_billing: Optional[bool] = None,
    ) -> Dict[str, Any]:
        nics = [
            {"subnet_id": subnet_network_id, "ip_address": private_ip}
            for private_ip in (list(nic_private_ips or []) or [PRIMARY_PRIVATE_IP])
        ]
        use_on_demand = is_on_demand_debug_enabled("huawei") if on_demand_billing is None else bool(on_demand_billing)
        charging_mode = "postPaid" if use_on_demand else BILLING_MODE
        extendparam: Dict[str, Any] = {"chargingMode": charging_mode}
        if charging_mode != "postPaid":
            extendparam.update(
                {
                    "periodType": BILLING_PERIOD_TYPE,
                    "periodNum": BILLING_PERIOD_NUM,
                    "isAutoRenew": str(BILLING_AUTO_RENEW).lower(),
                    "isAutoPay": str(BILLING_AUTO_PAY).lower(),
                }
            )
        server_body: Dict[str, Any] = {
            "name": name,
            "imageRef": image_id,
            "flavorRef": flavor_id,
            "vpcid": vpc_id,
            "nics": nics,
            "security_groups": [{"id": security_group_id}],
            "root_volume": {"volumetype": root_volume_type, "size": ROOT_VOLUME_SIZE},
            "adminPass": password,
            "count": 1,
            "extendparam": extendparam,
        }
        if availability_zone:
            server_body["availability_zone"] = availability_zone
        else:
            server_body["batch_create_in_multi_az"] = False
        body = {"server": server_body}
        url = f"{self.ecs_endpoint(pair.region)}/v1.1/{pair.project_id}/cloudservers"
        _, data, text = self.request("POST", url, body, timeout=90)
        if isinstance(data, dict):
            return data
        return {"raw": text}

    def create_server_interface(
        self,
        pair: ProjectPair,
        server_id: str,
        subnet_network_id: str,
        security_group_id: str,
        private_ip: str,
    ) -> Dict[str, Any]:
        body = {
            "interfaceAttachment": {
                "net_id": subnet_network_id,
                "fixed_ips": [{"ip_address": private_ip}],
                "security_groups": [{"id": security_group_id}],
            }
        }
        url = f"{self.ecs_endpoint(pair.region)}/v1/{pair.project_id}/cloudservers/{server_id}/os-interface"
        _, data, text = self.request("POST", url, body, timeout=90)
        if isinstance(data, dict):
            raw = data.get("interfaceAttachment") or data.get("interface_attachment") or data
            if isinstance(raw, dict):
                return raw
        return {"raw": text}

    def show_job(self, pair: ProjectPair, job_id: str) -> Dict[str, Any]:
        url = f"{self.ecs_endpoint(pair.region)}/v1/{pair.project_id}/jobs/{job_id}"
        _, data, text = self.request("GET", url)
        if isinstance(data, dict):
            return data
        return {"raw": text}

    def wait_job(
        self,
        pair: ProjectPair,
        job_id: str,
        emit: Callable[[str], None],
        *,
        timeout: int = 900,
        interval: int = 10,
    ) -> Dict[str, Any]:
        deadline = time.time() + timeout
        last_job: Dict[str, Any] = {}
        attempt = 0
        while time.time() < deadline:
            attempt += 1
            job = self.show_job(pair, job_id)
            last_job = job
            status = str(job.get("status") or (job.get("job") or {}).get("status") or "UNKNOWN").upper()
            progress = job.get("progress") or (job.get("job") or {}).get("progress") or ""
            message = f"Job {job_id} 状态 {attempt}: {status}"
            if progress != "":
                message += f"，进度 {progress}"
            emit(message)
            if status == "SUCCESS":
                return job
            if status == "PENDING_PAYMENT" and attempt >= 6:
                raise CloudApiError(
                    "订单处于待支付状态，自动支付未完成。通常是账户余额不足、自动支付失败，或该包年包月订单需要手动支付；已停止等待，避免一直卡住。",
                    body=json.dumps(job, ensure_ascii=False),
                )
            if status in {"FAIL", "FAILED", "ERROR"}:
                reason = job.get("fail_reason") or job.get("error_code") or json.dumps(job, ensure_ascii=False)
                raise CloudApiError(f"Job失败: {reason}")
            time.sleep(interval)
        raise TimeoutError(f"Job等待超时: {job_id}; last={json.dumps(last_job, ensure_ascii=False)}")

    @staticmethod
    def job_id_from_create_response(response: Dict[str, Any]) -> str:
        for key in ("job_id", "jobId", "jobID", "id"):
            value = response.get(key)
            if value:
                return str(value)
        job = response.get("job") or {}
        if isinstance(job, dict):
            for key in ("id", "job_id", "jobId", "jobID"):
                value = job.get(key)
                if value:
                    return str(value)
        return ""

    @staticmethod
    def server_id_from_create_response(response: Dict[str, Any], job: Dict[str, Any]) -> str:
        server_ids = response.get("serverIds") or response.get("server_ids") or []
        if isinstance(server_ids, list) and server_ids:
            return str(server_ids[0])
        server = response.get("server") or {}
        if isinstance(server, dict) and server.get("id"):
            return str(server["id"])

        entities = job.get("entities") or {}
        if isinstance(entities, dict):
            direct_id = entities.get("server_id") or entities.get("serverId")
            if direct_id:
                return str(direct_id)
            for sub_job in entities.get("sub_jobs") or []:
                if not isinstance(sub_job, dict):
                    continue
                sub_entities = sub_job.get("entities") or {}
                if isinstance(sub_entities, dict):
                    server_id = sub_entities.get("server_id") or sub_entities.get("serverId")
                    if server_id:
                        return str(server_id)
        return ""

    def show_server(self, pair: ProjectPair, server_id: str) -> Dict[str, Any]:
        url = f"{self.ecs_endpoint(pair.region)}/v1/{pair.project_id}/cloudservers/{server_id}"
        _, data, _ = self.request("GET", url)
        if not isinstance(data, dict) or not isinstance(data.get("server"), dict):
            raise CloudApiError(f"ECS详情返回格式异常: {data}")
        return data["server"]

    def list_server_interfaces(self, pair: ProjectPair, server_id: str) -> List[Dict[str, Any]]:
        url = f"{self.ecs_endpoint(pair.region)}/v1/{pair.project_id}/cloudservers/{server_id}/os-interface"
        _, data, _ = self.request("GET", url)
        if not isinstance(data, dict):
            return []
        raw = data.get("interfaceAttachments") or data.get("interface_attachments") or data.get("ports") or []
        return [item for item in raw if isinstance(item, dict)]

    def list_ports(self, pair: ProjectPair, *, timeout: int = 60, retry_attempts: Optional[int] = None) -> List[Dict[str, Any]]:
        ports: List[Dict[str, Any]] = []
        marker = ""
        limit = 2000
        while True:
            query = f"limit={limit}"
            if marker:
                query += f"&marker={urllib.parse.quote(marker, safe='')}"
            url = f"{self.vpc_endpoint(pair.region)}/v1/{pair.project_id}/ports?{query}"
            _, data, _ = self.request("GET", url, timeout=timeout, retry_attempts=retry_attempts)
            if not isinstance(data, dict):
                raise CloudApiError(f"端口列表返回格式异常: {data}")
            page = list(data.get("ports") or [])
            for item in page:
                item["_region"] = pair.region
                item["_project_id"] = pair.project_id
            ports.extend(page)
            if len(page) < limit:
                return ports
            marker = str(page[-1].get("id") or "")
            if not marker:
                return ports

    @staticmethod
    def interface_port_id(interface: Dict[str, Any]) -> str:
        return str(interface.get("port_id") or interface.get("id") or interface.get("portId") or "")

    @staticmethod
    def interface_private_ip(interface: Dict[str, Any]) -> str:
        fixed_ips = interface.get("fixed_ips") or interface.get("fixedIps") or []
        if isinstance(fixed_ips, list):
            for fixed_ip in fixed_ips:
                if isinstance(fixed_ip, dict):
                    ip_text = str(fixed_ip.get("ip_address") or fixed_ip.get("ipAddress") or "")
                    if ip_text and not is_public_ip(ip_text):
                        return ip_text
        return str(
            interface.get("fixed_ip")
            or interface.get("private_ip")
            or interface.get("private_ip_address")
            or interface.get("ip_address")
            or ""
        )

    @staticmethod
    def primary_port_id_from_server(server: Dict[str, Any], interfaces: Sequence[Dict[str, Any]]) -> str:
        addresses = server.get("addresses") or {}
        for items in addresses.values():
            if not isinstance(items, list):
                continue
            for address in items:
                if not isinstance(address, dict):
                    continue
                if address.get("OS-EXT-IPS:type") != "fixed":
                    continue
                if address.get("primary") is False:
                    continue
                port_id = address.get("OS-EXT-IPS:port_id") or address.get("port_id")
                if port_id:
                    return str(port_id)
        for item in interfaces:
            port_id = HuaweiCloudClient.interface_port_id(item)
            if port_id:
                return str(port_id)
        return ""

    @staticmethod
    def public_ip_from_server(server: Dict[str, Any]) -> str:
        addresses = server.get("addresses") or {}
        for items in addresses.values():
            if not isinstance(items, list):
                continue
            for address in items:
                if not isinstance(address, dict):
                    continue
                ip_text = address.get("addr") or ""
                if address.get("OS-EXT-IPS:type") == "floating" and ip_text:
                    return ip_text
                if ip_text and is_public_ip(ip_text):
                    return ip_text
        return ""

    @staticmethod
    def server_flavor_id(server: Dict[str, Any]) -> str:
        flavor = server.get("flavor") or {}
        if isinstance(flavor, dict):
            for key in ("id", "name", "original_name", "flavor_id", "flavor_name"):
                if flavor.get(key):
                    return str(flavor[key])
        for key in ("flavorRef", "flavor_ref", "flavor_id", "flavor_name"):
            if server.get(key):
                return str(server[key])
        return str(flavor or "")

    @staticmethod
    def server_mentions_flavor(server: Dict[str, Any], flavor_id: str = FLAVOR_ID) -> bool:
        return flavor_id.lower() in json.dumps(server, ensure_ascii=False).lower()

    @staticmethod
    def server_matches_2c2g(server: Dict[str, Any]) -> bool:
        flavor = server.get("flavor") or {}
        if isinstance(flavor, dict) and flavor_matches_target_size(flavor):
            return True
        vcpus = int_or_zero(server.get("vcpus") or server.get("cpu") or server.get("cpu_count"))
        ram = int_or_zero(server.get("ram") or server.get("memory") or server.get("memory_mb"))
        if vcpus == TARGET_VCPUS and ram == TARGET_RAM_MB:
            return True
        text = json.dumps(server, ensure_ascii=False).lower()
        has_cpu = bool(re.search(r"2\s*(?:vcpus|vcpu|核|c)", text))
        has_ram = bool(re.search(r"(?:2\s*(?:gib|gb|g)|2048\s*(?:mib|mb|m))", text))
        return has_cpu and has_ram

    @staticmethod
    def server_matches_target_size(server: Dict[str, Any]) -> bool:
        return HuaweiCloudClient.server_mentions_flavor(server) or HuaweiCloudClient.server_matches_2c2g(server)

    @staticmethod
    def server_matches_low_config(server: Dict[str, Any], region: str) -> bool:
        flavor_id = HuaweiCloudClient.server_flavor_id(server)
        preferred = {item.lower() for item in low_config_flavor_ids(region)}
        if flavor_id and flavor_id.lower() in preferred:
            return True
        flavor = server.get("flavor") or {}
        if isinstance(flavor, dict) and flavor_matches_low_config_size(flavor):
            text = json.dumps(server, ensure_ascii=False).lower()
            return any(item in text for item in preferred)
        return False

    @staticmethod
    def server_matches_purchase_config(server: Dict[str, Any], flavor_mode: str, region: str = "") -> bool:
        flavor = server.get("flavor") or {}
        if flavor_mode == FLAVOR_MODE_WAWACLOUD_LOW:
            return HuaweiCloudClient.server_matches_low_config(server, region)
        if flavor_mode == FLAVOR_MODE_WAWACLOUD_20:
            flavor_id = HuaweiCloudClient.server_flavor_id(server).lower()
            return flavor_id in {item.lower() for item in TWENTY_CONFIG_FLAVORS}
        if flavor_mode == FLAVOR_MODE_LOWEST:
            return bool(HuaweiCloudClient.server_flavor_id(server)) or HuaweiCloudClient.server_matches_2c2g(server)
        return HuaweiCloudClient.server_mentions_flavor(server) or HuaweiCloudClient.server_matches_2c2g(server)

    def create_sub_network_interface(
        self,
        pair: ProjectPair,
        subnet_id: str,
        parent_port_id: str,
        security_group_id: str,
        private_ip: str,
    ) -> Dict[str, Any]:
        body = {
            "sub_network_interface": {
                "virsubnet_id": subnet_id,
                "parent_id": parent_port_id,
                "security_groups": [security_group_id],
                "private_ip_address": private_ip,
            }
        }
        url = f"{self.vpc_endpoint(pair.region)}/v3/{pair.project_id}/vpc/sub-network-interfaces"
        _, data, _ = self.request("POST", url, body)
        if not isinstance(data, dict) or not isinstance(data.get("sub_network_interface"), dict):
            raise CloudApiError(f"创建辅助弹性网卡返回格式异常: {data}")
        return data["sub_network_interface"]

    def delete_sub_network_interface(self, pair: ProjectPair, interface_id: str) -> None:
        url = f"{self.vpc_endpoint(pair.region)}/v3/{pair.project_id}/vpc/sub-network-interfaces/{interface_id}"
        self.request("DELETE", url, timeout=60)

    def list_sub_network_interfaces(self, pair: ProjectPair, *, timeout: int = 60, retry_attempts: Optional[int] = None) -> List[Dict[str, Any]]:
        interfaces: List[Dict[str, Any]] = []
        marker = ""
        while True:
            query = "limit=2000"
            if marker:
                query += f"&marker={urllib.parse.quote(marker, safe='')}"
            url = f"{self.vpc_endpoint(pair.region)}/v3/{pair.project_id}/vpc/sub-network-interfaces?{query}"
            _, data, _ = self.request("GET", url, timeout=timeout, retry_attempts=retry_attempts)
            if not isinstance(data, dict):
                raise CloudApiError(f"辅助弹性网卡列表返回格式异常: {data}")
            page = list(data.get("sub_network_interfaces") or [])
            for item in page:
                item["_region"] = pair.region
                item["_project_id"] = pair.project_id
            interfaces.extend(page)
            page_info = data.get("page_info") or {}
            marker = page_info.get("next_marker") or ""
            if not marker or not page:
                break
        return interfaces

    def create_public_ip(
        self,
        pair: ProjectPair,
        name: str,
        eip_type: str = EIP_TYPE,
        *,
        ip_address: str = "",
        port_id: str = "",
        timeout: int = 90,
    ) -> Dict[str, Any]:
        publicip_body = {"type": eip_type or EIP_TYPE, "ip_version": 4}
        ip_address = str(ip_address or "").strip()
        port_id = str(port_id or "").strip()
        if ip_address:
            publicip_body["ip_address"] = ip_address
        if port_id:
            publicip_body["port_id"] = port_id
        body = {
            "publicip": publicip_body,
            "bandwidth": {
                "name": name,
                "size": BANDWIDTH_SIZE,
                "share_type": BANDWIDTH_SHARE_TYPE,
                "charge_mode": BANDWIDTH_CHARGE_MODE,
            },
        }
        url = f"{self.vpc_endpoint(pair.region)}/v1/{pair.project_id}/publicips"
        _, data, _ = self.request("POST", url, body, timeout=timeout)
        if not isinstance(data, dict) or not isinstance(data.get("publicip"), dict):
            raise CloudApiError(f"申请EIP返回格式异常: {data}")
        return data["publicip"]

    def bind_public_ip(self, pair: ProjectPair, publicip_id: str, port_id: str) -> Dict[str, Any]:
        body = {"publicip": {"port_id": port_id}}
        url = f"{self.vpc_endpoint(pair.region)}/v1/{pair.project_id}/publicips/{publicip_id}"
        _, data, text = self.request("PUT", url, body, timeout=90)
        if isinstance(data, dict):
            return data.get("publicip") or data
        return {"raw": text}

    def unbind_public_ip(self, pair: ProjectPair, publicip_id: str) -> Dict[str, Any]:
        body = {"publicip": {"port_id": ""}}
        url = f"{self.vpc_endpoint(pair.region)}/v1/{pair.project_id}/publicips/{publicip_id}"
        _, data, text = self.request("PUT", url, body, timeout=90)
        if isinstance(data, dict):
            return data.get("publicip") or data
        return {"raw": text}

    def list_public_ips(self, pair: ProjectPair, *, timeout: int = 60, retry_attempts: Optional[int] = None) -> List[Dict[str, Any]]:
        url = f"{self.vpc_endpoint(pair.region)}/v1/{pair.project_id}/publicips?limit=2000"
        _, data, _ = self.request("GET", url, timeout=timeout, retry_attempts=retry_attempts)
        if not isinstance(data, dict):
            raise CloudApiError(f"EIP列表返回格式异常: {data}")
        publicips = list(data.get("publicips") or [])
        for item in publicips:
            item["_region"] = pair.region
            item["_project_id"] = pair.project_id
        return publicips

    def delete_public_ip(self, pair: ProjectPair, publicip_id: str) -> None:
        url = f"{self.vpc_endpoint(pair.region)}/v1/{pair.project_id}/publicips/{publicip_id}"
        self.request("DELETE", url, timeout=60)

    def wait_public_ip_unbound(self, pair: ProjectPair, publicip_id: str, *, timeout: int = 90, interval: float = 3.0) -> None:
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            for item in self.list_public_ips(pair, timeout=30, retry_attempts=2):
                if str(item.get("id") or item.get("publicip_id") or "") != publicip_id:
                    continue
                if not str(item.get("port_id") or ""):
                    return
                break
            time.sleep(interval)
        raise CloudApiError(f"EIP解绑后等待释放超时: {publicip_id}")

    def wait_public_ip_bound(
        self,
        pair: ProjectPair,
        publicip_id: str,
        port_id: str,
        *,
        timeout: int = 45,
        interval: float = 2.0,
    ) -> Dict[str, Any]:
        deadline = time.monotonic() + timeout
        last_port_id = ""
        while time.monotonic() < deadline:
            for item in self.list_public_ips(pair, timeout=30, retry_attempts=2):
                if str(item.get("id") or item.get("publicip_id") or "") != publicip_id:
                    continue
                last_port_id = str(item.get("port_id") or "").strip()
                if last_port_id == port_id:
                    return item
                break
            time.sleep(interval)
        raise CloudApiError(
            f"EIP绑定状态确认超时: {publicip_id}，目标端口={port_id}，当前端口={last_port_id or '未绑定'}"
        )

    def wait_public_ip_deleted(
        self,
        pair: ProjectPair,
        publicip_id: str,
        emit: Callable[[str], None],
        *,
        timeout: int = 120,
        interval: float = 4.0,
    ) -> bool:
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            try:
                current = self.list_public_ips(pair, timeout=30, retry_attempts=2)
            except Exception as exc:
                emit(f"EIP {publicip_id} 释放等待查询提示: {exc}")
                time.sleep(interval)
                continue
            if not any(str(item.get("id") or item.get("publicip_id") or "") == publicip_id for item in current):
                return True
            time.sleep(interval)
        emit(f"EIP {publicip_id} 释放等待超时，稍后刷新资源库确认。")
        return False

    def public_ips_for_ports(self, pair: ProjectPair, ordered_port_ids: Sequence[str]) -> List[str]:
        port_order = [str(port_id) for port_id in ordered_port_ids if port_id]
        by_port = self.public_ips_by_port(pair)
        ips: List[str] = []
        for port_id in port_order:
            for ip_addr in by_port.get(port_id, []):
                if ip_addr not in ips:
                    ips.append(ip_addr)
        return ips

    def public_ips_by_port(self, pair: ProjectPair) -> Dict[str, List[str]]:
        by_port: Dict[str, List[str]] = {}
        for item in self.list_public_ips(pair):
            port_id = str(item.get("port_id") or "")
            ip_addr = str(item.get("public_ip_address") or item.get("public_ip") or "")
            if port_id and is_public_ip(ip_addr):
                by_port.setdefault(port_id, [])
                if ip_addr not in by_port[port_id]:
                    by_port[port_id].append(ip_addr)
        return by_port

    def list_cloud_servers(self, pair: ProjectPair, *, timeout: int = 60, retry_attempts: Optional[int] = None) -> List[Dict[str, Any]]:
        servers: List[Dict[str, Any]] = []
        offset = 0
        limit = 100
        while True:
            url = f"{self.ecs_endpoint(pair.region)}/v1/{pair.project_id}/cloudservers/detail?offset={offset}&limit={limit}"
            _, data, _ = self.request("GET", url, timeout=timeout, retry_attempts=retry_attempts)
            if not isinstance(data, dict):
                raise CloudApiError(f"ECS列表返回格式异常: {data}")
            page = list(data.get("servers") or [])
            for item in page:
                item["_region"] = pair.region
                item["_project_id"] = pair.project_id
            servers.extend(page)
            total = int(data.get("count") or len(servers))
            if len(servers) >= total or not page:
                return servers
            offset += limit

    def delete_cloud_servers(
        self,
        pair: ProjectPair,
        server_ids: Sequence[str],
        *,
        delete_publicip: bool = False,
        delete_volume: bool = True,
    ) -> Dict[str, Any]:
        body = {
            "servers": [{"id": server_id} for server_id in server_ids if server_id],
            "delete_publicip": delete_publicip,
            "delete_volume": delete_volume,
        }
        url = f"{self.ecs_endpoint(pair.region)}/v1/{pair.project_id}/cloudservers/delete"
        _, data, text = self.request("POST", url, body, timeout=60)
        return data if isinstance(data, dict) else {"raw": text}

    def wait_cloud_servers_deleted(
        self,
        pair: ProjectPair,
        server_ids: Sequence[str],
        emit: Callable[[str], None],
        *,
        timeout: int = 300,
        interval: int = 8,
    ) -> bool:
        wanted = {str(server_id).casefold() for server_id in server_ids if server_id}
        if not wanted:
            return True
        deadline = time.time() + timeout
        while time.time() < deadline:
            try:
                current = self.list_cloud_servers(pair, timeout=30, retry_attempts=2)
            except Exception as exc:
                emit(f"ECS删除等待查询提示: {exc}")
                time.sleep(interval)
                continue
            existing = {
                str(server.get("id") or "").casefold()
                for server in current
                if str(server.get("id") or "").casefold() in wanted
            }
            if not existing:
                return True
            emit(f"ECS删除等待中，仍存在 {len(existing)} 台。")
            time.sleep(interval)
        emit("ECS删除等待超时，稍后刷新资源库确认。")
        return False

    def reinstall_server_with_cloud_init(
        self,
        pair: ProjectPair,
        server_id: str,
        password: str,
    ) -> Dict[str, Any]:
        body = {
            "os-reinstall": {
                "adminpass": password,
                "mode": "withStopServer",
            }
        }
        url = f"{self.ecs_endpoint(pair.region)}/v2/{pair.project_id}/cloudservers/{server_id}/reinstallos"
        # 中文注释：重装接口偶发被远端断开(WinError 10054)，这里允许重试，避免误报未提交。
        _, data, text = self.request("POST", url, body, timeout=60, retry_attempts=4)
        return data if isinstance(data, dict) else {"raw": text}

# 旧的 common.shared 里还有少量函数通过全局名查找 HuaweiCloudClient。
# 这里回填一次，保证过渡期旧调用链不被拆分影响。
_shared.HuaweiCloudClient = HuaweiCloudClient
