"""UCloud provider — UHost / UDNR / ULB / UCDN / UFile.

UCloud doesn't ship a maintained async Python SDK (the official
``ucloud-sdk-python3`` is sync-only and pulls a heavy pile of deps),
so the provider talks directly to the HTTPS API via httpx with a
hand-rolled signature per the public API spec [1].

Signature rule (v1 API — every UCloud product uses it):

  1. Sort all request parameters (including ``PublicKey``) alphabetically
     by key, but EXCLUDE ``Signature`` itself.
  2. Concatenate into a single string ``<k1><v1><k2><v2>…``.
  3. Append ``<PrivateKey>`` to the end.
  4. SHA-1 hex digest = ``Signature`` parameter.

All UCloud APIs use a single endpoint ``https://api.ucloud.cn`` with
``Action`` telling which product (e.g. ``Action=DescribeUHostInstance``).
Region is a parameter on per-region calls (UHost/ULB), absent on
global calls (UDNR/UCDN/UFile bucket-list).

Capabilities:
  * SERVERS        — ``DescribeUHostInstance`` per region
  * DNS_RECORDS    — ``UdnrDomainDNSQuery`` per zone (requires
                     admin-supplied ``dns_zones`` config; UCloud has
                     no "list all zones" API)
  * LOAD_BALANCERS — ``DescribeULB`` per region
  * CDN            — ``GetUcdnDomainInfoList`` (global)
  * OBJECT_STORAGE — ``DescribeBucket`` (global)

DOMAINS is deliberately absent: UCloud has no public "list all
registered domains" API, so the domain-registration capability
can't be driven programmatically. Historical 域名 rows migrated from
the legacy CMDB remain visible but are never re-synced.

[1] https://docs.ucloud.cn/api/summary/signature
"""
from __future__ import annotations

import asyncio
import hashlib
import logging
from datetime import date, datetime, timedelta
from typing import Any

import httpx

from app.domains.cmdb.providers.base import (
    CloudCdnDomain,
    CloudDnsRecord,
    CloudLoadBalancer,
    CloudProvider,
    CloudServer,
    ProviderCapability,
    ProviderCredentials,
    register_provider,
)
from app.domains.cmdb.providers.exceptions import (
    CapabilityNotSupported,
    ProviderAuthError,
    ProviderRateLimitError,
    ProviderTransientError,
    redact_provider_error,
)

logger = logging.getLogger(__name__)

_API_ENDPOINT = "https://api.ucloud.cn"
_PCDN_ENDPOINT = "https://api.ucloud.cn"  # UCDN shares the same endpoint


# Sentinel for the ``project_id`` kwarg on ``_call`` — distinguishes
# "caller didn't pass anything → use integration default" (``_UNSET``)
# from "caller explicitly passed None → omit ProjectId" (``None``).
# A plain ``None`` default would conflate both cases.
class _UNSET_TYPE:
    pass


_UNSET = _UNSET_TYPE()

# Default UHost regions for UCloud — covers all mainland public regions
# plus Hong Kong and common overseas. Admins narrow via ``regions`` field.
_DEFAULT_REGIONS = (
    "cn-bj2,cn-sh2,cn-gd,cn-qz,cn-north-07,cn-wlcb,hk,"
    "us-ca,us-ws,ge-fra,tw-kh,tw-tp,"
    "sg,kr-seoul,th-bkk,jpn-tky,afr-nigeria,"
    "idn-jakarta,rus-mosc,uae-dubai,vn-sng"
)

# UCloud public-key / private-key auth. The "PublicKey" is the
# plaintext identifier the signer broadcasts; the "PrivateKey" is the
# hash seed — the Signature computation requires both.
UCLOUD_PUBLIC_KEY_FIELD = "public_key"
UCLOUD_PRIVATE_KEY_FIELD = "private_key"

_STATUS_MAP_UHOST = {
    "Running": "running",
    "Starting": "running",
    "Stopped": "stopped",
    "Stopping": "stopped",
    "ResizeFail": "stopped",
    "Install": "running",
    "ImageMaking": "running",
    "BootFailed": "stopped",
    "Rebooting": "running",
    "KeyPending": "stopped",
}


def _sign_params(params: dict[str, Any], private_key: str) -> str:
    """Compute SHA-1 signature per UCloud API v1 spec."""
    sorted_items = sorted(
        (k, v) for k, v in params.items() if k != "Signature"
    )
    concat = "".join(f"{k}{v}" for k, v in sorted_items)
    concat += private_key
    return hashlib.sha1(concat.encode()).hexdigest()  # nosec B324 — vendor-mandated


def _is_region_skip_signal(exc_msg: str) -> bool:
    """Recognize per-region "this tenant doesn't have the product here"
    signals so the per-region sweep can skip and keep going instead of
    failing the whole sync.

    UCloud surfaces this several ways depending on product:
      * ``RetCode=230`` + ``Params [Region] not available`` — UHost
        (account has no UHost quota in that region).
      * ``RetCode=150`` + ``Service unavailable`` — ULB / UFile when
        the product isn't provisioned in the region for this tenant
        (prod 2026-04-24 15:15 — every ULB sync in af/idn/rus/uae/vn
        was failing the whole sweep on this).

    We intentionally match on the message substrings that the
    ``_call`` wrapper embeds — callers don't see the raw RetCode
    after the wrapper raises.
    """
    msg = exc_msg or ""
    if "RetCode=150" in msg and "Service unavailable" in msg:
        return True
    if "Region" in msg and "not available" in msg:
        return True
    return False


# UCloud returns HTTP 200 with a non-zero ``RetCode`` for EVERY error —
# transient server faults included — so a body-level transient never
# reaches the transport / HTTP-5xx retry budget in ``_call``. These
# RetCodes / message shapes indicate a transient server fault that a
# retry usually clears (parity with the HTTP-5xx retry path and with
# Tencent's ``InternalError`` / ``ServiceUnavailable`` retry pool in
# ``_tencent_throttle``). 8498 = "Call third api service timeout,
# please retry or contact technical support" — observed in prod
# 2026-05-27 on the ``jpn-tky`` UHost sweep, where a single transient
# gateway timeout failed the whole ``servers`` capability and paged.
_UCLOUD_TRANSIENT_RETCODES: frozenset[int] = frozenset({8498})

# UCloud's gateway/backend timeouts surface under more than one numeric
# code but consistently carry one of these phrases in ``Message``. Match
# them too so the whole transient-timeout class is covered, not just the
# single code we happened to observe. Bounded by the 3-attempt budget, a
# false-positive costs at most two extra backoff sleeps before the real
# error surfaces — far cheaper than a false-negative (the bug we fix).
_UCLOUD_TRANSIENT_MSG_HINTS: tuple[str, ...] = (
    "timeout",
    "time out",
    "please retry",
    "try again",
)


# RetCodes that must NEVER be retried regardless of ``Message`` text:
#   * 150 / 230 — region-not-provisioned skip (handled by
#     :func:`_is_region_skip_signal`); a fast per-region skip, not a
#     transient — burning three backoff retries per un-provisioned
#     region would slow every sweep for no benefit.
#   * 100 / 231 / 299 — the auth bucket (revoked / invalid key); retrying
#     a bad credential is pointless.
# Excluding them HERE makes the predicate self-contained: the CMDB
# provider's ``_call`` classifies auth BEFORE consulting this predicate,
# but the lifecycle/billing driver's ``_signed_call`` has no body-level
# auth classification (only HTTP 401/403), so without this guard a
# hypothetical auth ``RetCode`` whose ``Message`` happened to carry a
# transient hint ("please retry") would be retried 3× on a read path.
_UCLOUD_NEVER_RETRY_RETCODES: frozenset[int] = frozenset({100, 150, 230, 231, 299})


def _is_retryable_retcode(ret: int, msg: str) -> bool:
    """Whether a non-zero UCloud ``RetCode`` is a transient server fault
    worth retrying in-place. Self-contained — safe to call from either
    the provider ``_call`` or the driver ``_signed_call`` regardless of
    whether the caller has already classified auth / region-skip codes.

    Returns ``False`` for the region-skip + auth buckets
    (:data:`_UCLOUD_NEVER_RETRY_RETCODES`) regardless of message text;
    otherwise ``True`` for the known transient ``RetCode``\\ s or any
    code whose ``Message`` carries a transient-timeout hint.
    """
    if ret in _UCLOUD_NEVER_RETRY_RETCODES:
        return False
    if ret in _UCLOUD_TRANSIENT_RETCODES:
        return True
    low = (msg or "").lower()
    return any(hint in low for hint in _UCLOUD_TRANSIENT_MSG_HINTS)


def _first_present(row: dict[str, Any], keys: tuple[str, ...]) -> Any:
    """First ``row[key]`` that is actually *present* (not None, not ``""``).

    Deliberately NOT truthiness (``row.get(a) or row.get(b) or …``): a
    JSON numeric ``0`` — e.g. a fully-couponed ``AmountReal`` — is falsy,
    so an ``or`` chain silently falls through to the *gross* ``Amount``
    and overcounts the line. Zero is a real value; keep it. Shared by
    the provider monthly path and the driver daily / cost_facts paths
    (imported there) so all three bill parsers agree on the same
    amount-key priority AND the same presence semantics.
    """
    for key in keys:
        val = row.get(key)
        if val is not None and val != "":
            return val
    return None


def _json_safe(obj: Any) -> Any:
    if isinstance(obj, (datetime, date)):
        return obj.isoformat()
    if isinstance(obj, dict):
        return {k: _json_safe(v) for k, v in obj.items()}
    if isinstance(obj, (list, tuple)):
        return [_json_safe(v) for v in obj]
    if isinstance(obj, (bytes, bytearray)):
        return obj.decode("utf-8", errors="replace")
    return obj


@register_provider
class UCloudProvider(CloudProvider):
    """UCloud CMDB data source — live API, all six capabilities."""

    @classmethod
    def provider_type(cls) -> str:
        return "ucloud"

    @classmethod
    def supported_capabilities(cls) -> frozenset[ProviderCapability]:
        return frozenset({
            ProviderCapability.SERVERS,
            ProviderCapability.DNS_RECORDS,
            ProviderCapability.LOAD_BALANCERS,
            ProviderCapability.CDN,
            ProviderCapability.OBJECT_STORAGE,
        })

    def __init__(self, credentials: ProviderCredentials) -> None:
        super().__init__(credentials)
        configured = credentials.fields.get("regions") or _DEFAULT_REGIONS
        self._regions = [r.strip() for r in configured.split(",") if r.strip()]
        # Multi-project support (2026-05-28).
        #
        # UCloud's API takes a per-call ``ProjectId`` query parameter
        # that scopes UHost / ULB / UFile / UCDN list results to ONE
        # project. Pre-fix the driver took a single ``project_id``
        # field, so a tenant with N projects had at most 1/N of their
        # resources visible to Helix — the rest never reached
        # ``cmdb_servers`` etc., and the safe-sweep + sync-history made
        # it look like the integration was healthy.
        #
        # Incident: prod 2026-04-12 → 2026-05-28 ran with
        # ``project_id`` configured on the prod UCloud integration's
        # default project. The user's DLP-test server (project
        # ``org-eneelp``) was never returned by the live sync, every
        # ``GET /cmdb/servers?search=10.9.84.114`` returned 0 hits.
        #
        # New behavior:
        #   * ``project_ids`` (CSV) wins if set — driver iterates each
        #     project for every per-project list call, dedupes by
        #     ``instance_id`` (UCloud cloud IDs are globally unique
        #     even across UCloud's own projects).
        #   * ``project_id`` (singular) is honored when ``project_ids``
        #     is absent — backward-compatible with existing integration
        #     rows. The old + new fields are mutually exclusive in
        #     intent but if both are set, the plural wins (the admin
        #     latest action).
        #   * Both empty → call WITHOUT ``ProjectId`` parameter, which
        #     UCloud interprets as "default project". Same as pre-fix
        #     behavior when neither field was set.
        project_ids_raw = (credentials.fields.get("project_ids") or "").strip()
        legacy_single = (credentials.fields.get("project_id") or "").strip()
        if project_ids_raw:
            self._project_ids: list[str | None] = [
                p.strip() for p in project_ids_raw.split(",") if p.strip()
            ] or [None]
        elif legacy_single:
            self._project_ids = [legacy_single]
        else:
            # ``None`` sentinel — _call omits the ProjectId parameter
            # entirely, deferring to UCloud's default project for the AK.
            self._project_ids = [None]
        # UCloud UDNR has no "list zones" API — admin must declare
        # which zones the sync should iterate. Empty → DNS_RECORDS
        # capability surfaces as CapabilityNotSupported (sync_job marks
        # as "skipped" rather than silently mark-and-sweeping the zones
        # out of cmdb_dns_records).
        dns_zones_raw = credentials.fields.get("dns_zones") or ""
        self._dns_zones = [z.strip() for z in dns_zones_raw.split(",") if z.strip()]
        self._client = httpx.AsyncClient(timeout=30.0, trust_env=False)

    async def close(self) -> None:
        try:
            await self._client.aclose()
        except Exception:  # noqa: BLE001, S110 — best-effort on shutdown
            pass

    # ── Core API call ───────────────────────────────────────

    async def _call(
        self,
        action: str,
        params: dict[str, Any] | None = None,
        *,
        endpoint: str = _API_ENDPOINT,
        project_id: str | None | _UNSET_TYPE = _UNSET,
    ) -> dict[str, Any]:
        """Sign and POST an UCloud API action. Returns the JSON response dict.

        Raises:
          * ``ProviderAuthError`` — HTTP 401/403 or ``RetCode`` in auth bucket
          * ``ProviderRateLimitError`` — HTTP 429
          * ``ProviderTransientError`` — HTTP 5xx / non-zero RetCode (a
            transient ``RetCode`` is retried in-place first; see layer 3
            below)
        """
        public_key = str(
            self._credentials.fields.get(UCLOUD_PUBLIC_KEY_FIELD) or "",
        ).strip()
        private_key = str(
            self._credentials.fields.get(UCLOUD_PRIVATE_KEY_FIELD) or "",
        ).strip()
        if not public_key or not private_key:
            raise ProviderAuthError(
                "UCloud public_key / private_key not configured",
            )
        full_params: dict[str, Any] = {
            "Action": action,
            "PublicKey": public_key,
        }
        # Per-call ``project_id`` override threads multi-project support
        # through the list_* loops without changing every helper's
        # signature: pass ``project_id=<pid>`` to scope this single
        # call, or ``project_id=None`` to omit ProjectId entirely.
        # Default sentinel ``_UNSET`` falls back to the integration's
        # first configured project (preserving single-project behavior
        # for any code path that doesn't iterate explicitly).
        effective_pid: str | None
        if isinstance(project_id, _UNSET_TYPE):
            effective_pid = self._project_ids[0] if self._project_ids else None
        else:
            effective_pid = project_id
        if effective_pid:
            full_params["ProjectId"] = effective_pid
        if params:
            # UCloud wants every scalar serialized to string; nested
            # arrays use ``Key.0=val0&Key.1=val1`` encoding, handled
            # by the caller before passing in.
            for k, v in params.items():
                if v is None:
                    continue
                if isinstance(v, bool):
                    # Python bools stringify to "True"/"False"; UCloud's
                    # server expects lowercase JSON-style booleans.
                    full_params[k] = "true" if v else "false"
                    continue
                if v == "":
                    # Empty strings break signing — both ends have to
                    # agree on whether a ``Key=`` pair participates,
                    # and UCloud's newer APIs drop them. Dropping keeps
                    # us consistent with Aliyun/Tencent behavior.
                    continue
                full_params[k] = str(v)
        full_params["Signature"] = _sign_params(full_params, private_key)
        # Layered retry budget:
        # 1. Transport-layer — pre-fix a single ReadTimeout /
        #    ConnectError immediately raised ProviderTransientError and
        #    the capability was marked failed for the tick. Aliyun's
        #    ``_call`` already retries 3× on the same exceptions; parity
        #    fix so UCloud behaves consistently when ULink / UCloud's
        #    regional API endpoints flap. UCloud's signature is NOT
        #    time-bound (``_sign_params`` only sorts + concatenates, no
        #    Timestamp field), so retrying with the same payload is
        #    safe — no re-sign per attempt.
        # 2. Status-code 429 / 5xx — Round-30 (2026-05-24) parity with
        #    Aliyun's `_call` 429+5xx retry loop. Pre-fix a single 5xx
        #    on a single region's UHost.DescribeUHostInstance dropped
        #    that region's entire instance list (the outer per-region
        #    try/except logged + appended a partial_failure, but the
        #    capability flipped to red and the failure-notifier paged
        #    every cron tick until the transient cleared). UCloud's
        #    API endpoints are global per-action (``api.ucloud.cn``)
        #    so 5xx is RARELY persistent — bounded retry recovers
        #    most transients without admin attention.
        # 3. Body-level transient ``RetCode`` — UCloud returns HTTP 200
        #    + a non-zero ``RetCode`` for transient server faults too
        #    (8498 "Call third api service timeout, please retry" — prod
        #    2026-05-27, jpn-tky UHost), so they NEVER reach layers 1/2.
        #    Pre-fix a single such timeout on one region raised straight
        #    through, the per-region caller recorded a partial_failure +
        #    re-raised, the whole ``servers`` capability flipped red and
        #    the failure-notifier paged — even though the very next call
        #    would have recovered. Classify + retry transient RetCodes
        #    INSIDE the same 3-attempt budget (parity with layer 2 and
        #    with Tencent's InternalError/ServiceUnavailable retry pool);
        #    on the final attempt fall through to raise so a persistent
        #    fault still surfaces. This is why the JSON parse + RetCode
        #    classification now live inside the loop rather than after it.
        resp: httpx.Response | None = None
        last_exc: Exception | None = None
        for attempt in range(3):
            try:
                resp = await self._client.post(endpoint, data=full_params)
            except httpx.HTTPError as exc:
                last_exc = exc
                if attempt < 2:
                    logger.warning(
                        "UCloud HTTP transport error action=%s "
                        "attempt=%d/3: %s",
                        action, attempt + 1,
                        redact_provider_error(exc),
                    )
                    await asyncio.sleep(2 ** attempt)
                    continue
                # Budget exhausted — redact (httpx exceptions can include
                # the failing URL with signed query parameters: PublicKey
                # + Signature) before surfacing to admin-readable
                # surfaces (sync history error column, Feishu failure-
                # notifier cards).
                raise ProviderTransientError(
                    f"UCloud HTTP error: {redact_provider_error(exc)}",
                ) from exc
            # Status-code retry: 429 + 5xx within the same 3-attempt
            # budget. 4xx (other than 429) is deterministic — bail
            # immediately via the 401/403 branch / ``raise_for_status``
            # below so the SignatureError detection downstream still
            # fires.
            if resp.status_code == 429 or resp.status_code >= 500:
                # Round-30 security-review fix (2026-05-24): consume the
                # response body BEFORE the retry sleep so the underlying
                # connection returns to the httpx pool — pre-fix the
                # response sat un-aclosed across the 1s/2s backoff,
                # holding a pool slot. Under a regional 5xx storm with
                # many concurrent UHost regions retrying simultaneously,
                # the pool drained to PoolTimeout (transport-retry
                # bucket) before the API actually recovered. Reading
                # ``.text`` (already needed for the error message)
                # forces httpx to consume + release the connection.
                body_snippet = redact_provider_error(resp.text)
                last_exc = ProviderTransientError(
                    f"UCloud HTTP {resp.status_code} action={action}: "
                    f"{body_snippet}",
                )
                if attempt < 2:
                    logger.warning(
                        "UCloud HTTP %d retry action=%s attempt=%d/3",
                        resp.status_code, action, attempt + 1,
                    )
                    await resp.aclose()
                    await asyncio.sleep(2 ** attempt)
                    continue
                if resp.status_code == 429:
                    raise ProviderRateLimitError()
                raise ProviderTransientError(
                    f"UCloud HTTP {resp.status_code} (after 3 attempts): "
                    f"{body_snippet}"
                )
            if resp.status_code in (401, 403):
                # 401/403 bodies on signed requests echo the failing
                # request URL which carries the AK in query params.
                # Redact before surfacing to admin-readable surfaces.
                raise ProviderAuthError(
                    f"UCloud HTTP {resp.status_code}: "
                    f"{redact_provider_error(resp.text)}"
                )
            resp.raise_for_status()
            try:
                data: dict[str, Any] = resp.json()
            except ValueError as exc:
                raise ProviderTransientError(
                    f"UCloud returned non-JSON: "
                    f"{redact_provider_error(resp.text)}",
                ) from exc
            ret = int(data.get("RetCode") or 0)
            if ret == 0:
                return data
            msg = str(data.get("Message") or data.get("Action") or "UCloud error")
            # RetCode=230 is UCloud's generic "parameter error" bucket —
            # it covers both revoked-key failures ("Invalid PublicKey")
            # AND parameter-value rejections like "Params [Region] not
            # available". The region case is NOT auth: it means the AK
            # is valid but this particular region isn't enabled for the
            # account. Surface as transient so the per-region loops
            # (``list_servers`` / ``list_load_balancers``) skip the
            # region and keep sweeping the rest (prod 2026-04-24 had
            # every UHost sync dying on ``afr-nigeria`` / ``idn-jakarta``
            # regions because the whole sweep was being treated as an
            # auth failure).
            if ret == 230 and "Region" in msg and "not available" in msg:
                raise ProviderTransientError(
                    f"UCloud {action} RetCode={ret}: {msg}",
                )
            # Parentheses are load-bearing — ``A or B and C`` has
            # precedence surprises easy to introduce by mistake.
            if (
                ret in (100, 230, 231, 299)
                or ("Invalid" in msg and "Key" in msg)
            ):
                raise ProviderAuthError(f"UCloud {action} RetCode={ret}: {msg}")
            # Layer 3 — body-level transient RetCode (see the budget
            # comment above). Retry in-place within the same budget; on
            # the final attempt fall through to raise so a persistent
            # fault still surfaces and the per-region caller can record
            # its ``partial_failures`` leg. ``resp.json()`` above already
            # drained the body, so the connection is back in the pool
            # before the backoff sleep.
            if _is_retryable_retcode(ret, msg) and attempt < 2:
                last_exc = ProviderTransientError(
                    f"UCloud {action} RetCode={ret}: {msg}",
                )
                logger.warning(
                    "UCloud %s RetCode=%d transient — retry "
                    "attempt=%d/3: %s",
                    action, ret, attempt + 1, redact_provider_error(msg),
                )
                await asyncio.sleep(2 ** attempt)
                continue
            raise ProviderTransientError(
                f"UCloud {action} RetCode={ret}: {msg}",
            )
        # Defensive — every in-loop path returns or raises; this only
        # fires if the retry budget is somehow exhausted without a
        # verdict (e.g. a future edit drops a terminal branch).
        if last_exc is not None:
            raise last_exc
        raise ProviderTransientError(
            f"UCloud {action}: retry budget exhausted without response",
        )

    # ── Servers (UHost) ───────────────────────────────────────

    async def list_servers(self) -> list[CloudServer]:
        # ``seen`` dedupes across (project × region). UCloud's UHostId
        # is globally unique across an account's projects, so the same
        # instance can technically appear under one project's results
        # only — but the dedupe is cheap and protects against any
        # future UCloud API change that overlaps project scopes.
        servers: list[CloudServer] = []
        seen: set[str] = set()
        for project_id in self._project_ids:
            for region in self._regions:
                offset = 0
                page_size = 100  # UHost DescribeUHostInstance max Limit
                try:
                    while True:
                        data = await self._call(
                            "DescribeUHostInstance",
                            {
                                "Region": region,
                                "Offset": offset,
                                "Limit": page_size,
                            },
                            project_id=project_id,
                        )
                        items = data.get("UHostSet") or []
                        for inst in items:
                            iid = str(inst.get("UHostId") or "")
                            if not iid or iid in seen:
                                continue
                            seen.add(iid)
                            servers.append(_server_from_uhost(inst, region))
                        total = int(data.get("TotalCount") or 0)
                        offset += page_size
                        if offset >= total or not items:
                            break
                except ProviderAuthError:
                    raise
                except ProviderTransientError as exc:
                    # A region the AK/SK isn't enabled in surfaces as a
                    # transient error with one of several shapes:
                    #   * RetCode=230 "Params [Region] not available"
                    #   * RetCode=150 "Service unavailable" when UHost
                    #     isn't provisioned in that (project,region)
                    # Log + skip instead of stalling the whole sweep.
                    leg_label = (
                        f"project:{project_id or 'default'}/region:{region}"
                    )
                    if _is_region_skip_signal(str(exc)):
                        self.skipped_legs.append({
                            "leg": leg_label,
                            "capability": "servers",
                            "reason": "region_not_provisioned",
                        })
                        logger.info(
                            "UCloud UHost %s unavailable: %s",
                            leg_label, exc,
                        )
                        continue
                    self.partial_failures.append({
                        "leg": leg_label,
                        "capability": "servers",
                        "reason": "transient",
                    })
                    raise
        return servers

    # ── Server lifecycle (UHost actions) ─────────────────────
    #
    # All four UHost lifecycle actions share the same parameter shape:
    # ``Region`` + ``UHostId`` (+ optional ``ProjectId`` when the
    # tenant uses multi-project). UCloud's API is itself idempotent —
    # calling ``StopUHostInstance`` on an already-stopped instance
    # returns ``RetCode=174`` ("instance already in target state") or
    # similar; we catch the small set of "already there" RetCodes
    # and treat them as success.
    #
    # Legacy CMDB called these fire-and-forget without polling for
    # job completion (see cmdb-backend-main/utils/ucloud.go) — we keep
    # that contract. The next ``cmdb.servers`` sync (5-min cron)
    # reconciles the actual cloud state into ``cmdb_servers.status``.

    async def _lifecycle_call(
        self,
        action: str,
        *,
        instance_id: str,
        region: str | None,
        project_id: str | None,
        extra: dict[str, Any] | None = None,
        idempotent_retcodes: frozenset[int] = frozenset(),
    ) -> None:
        if not region:
            # UHost actions are region-scoped; reject explicitly so the
            # router surfaces a 422 rather than the cloud returning a
            # less-helpful "Region parameter missing" RetCode.
            raise ValueError(
                f"UCloud {action} requires region — server row has no region",
            )
        # ``project_id`` override threads through the multi-project
        # support already wired in ``_call``: pass the row's stored
        # project_id explicitly so we don't fall back to the
        # integration's first configured project (which may be a
        # different project from where the instance actually lives).
        params: dict[str, Any] = {
            "Region": region,
            "UHostId": instance_id,
        }
        if extra:
            params.update(extra)
        try:
            await self._call(
                action,
                params,
                project_id=project_id if project_id else None,
            )
        except ProviderTransientError as exc:
            # UCloud signals "already in target state" via specific
            # RetCodes — surface as success so the action is idempotent
            # for retried clicks. The transient wrapper carries the
            # RetCode in the message ("UCloud {action} RetCode={n}: ...").
            if idempotent_retcodes:
                import re as _re
                m = _re.search(r"RetCode=(-?\d+)", str(exc))
                if m and int(m.group(1)) in idempotent_retcodes:
                    logger.info(
                        "UCloud %s on %s already in target state (RetCode=%s) — "
                        "treating as idempotent success",
                        action, instance_id, m.group(1),
                    )
                    return
            raise

    async def power_on(
        self,
        *,
        instance_id: str,
        region: str | None = None,
        project_id: str | None = None,
    ) -> None:
        # RetCode 173 = "instance already running" (observed in prod).
        await self._lifecycle_call(
            "StartUHostInstance",
            instance_id=instance_id,
            region=region,
            project_id=project_id,
            idempotent_retcodes=frozenset({173}),
        )

    async def power_off(
        self,
        *,
        instance_id: str,
        region: str | None = None,
        project_id: str | None = None,
        force: bool = False,
    ) -> None:
        # RetCode 174 = "instance already stopped".
        await self._lifecycle_call(
            "StopUHostInstance",
            instance_id=instance_id,
            region=region,
            project_id=project_id,
            idempotent_retcodes=frozenset({174}),
        )

    async def restart_server(
        self,
        *,
        instance_id: str,
        region: str | None = None,
        project_id: str | None = None,
    ) -> None:
        await self._lifecycle_call(
            "RebootUHostInstance",
            instance_id=instance_id,
            region=region,
            project_id=project_id,
        )

    async def destroy_server(
        self,
        *,
        instance_id: str,
        region: str | None = None,
        project_id: str | None = None,
    ) -> None:
        # ReleaseEIP / ReleaseUDisk match legacy CMDB behavior: a true
        # termination should clean up the attached resources, otherwise
        # the customer keeps paying for the EIP + disks of a deleted
        # host. ``Destroy`` returns success on already-destroyed.
        await self._lifecycle_call(
            "TerminateUHostInstance",
            instance_id=instance_id,
            region=region,
            project_id=project_id,
            extra={"ReleaseEIP": "true", "ReleaseUDisk": "true"},
        )

    # ── DNS Records (UDNR) ───────────────────────────────────
    #
    # UCloud's UDNR API only exposes per-zone record queries
    # (``UdnrDomainDNSQuery``) — there is no "list all managed zones"
    # action. Admin must supply the zone list via the ``dns_zones``
    # integration config field. When that field is empty, sync skips
    # DNS records entirely (CapabilityNotSupported → scheduled_job
    # marks the run as ``skipped`` rather than empty-and-sweeping
    # existing rows out of cmdb_dns_records).

    async def list_dns_records(self) -> list[CloudDnsRecord]:
        if not self._dns_zones:
            raise CapabilityNotSupported(
                "UCloud DNS records require the ``dns_zones`` config field "
                "(comma-separated zone names) — UCloud has no list-all-zones API",
            )
        records: list[CloudDnsRecord] = []
        for zone_name in self._dns_zones:
            try:
                # UCloud UDNR (DNS) is account-scoped, not project-scoped
                # — ``UdnrDomainDNSQuery`` resolves the zone owner from
                # the AK regardless of which project the rest of the
                # tenant's resources live in (verified against UCloud
                # docs + the prod tenant's single-zone-per-account
                # observation). Pass ``project_id=None`` explicitly so
                # the per-call override layer doesn't fall back to
                # ``self._project_ids[0]`` and inadvertently scope a
                # global query (forward-defence — if UCloud ever
                # tightens UDNR to project-scoped, the fallback would
                # silently start returning zero zones).
                data = await self._call(
                    "UdnrDomainDNSQuery", {"Dn": zone_name},
                    project_id=None,
                )
            except ProviderTransientError as exc:
                # Zone that isn't registered under this account returns
                # a DB-level error (RetCode=95501) — log and skip
                # instead of failing the whole sweep.
                msg = str(exc)
                if (
                    "95501" in msg
                    or "数据库错误" in msg
                    or "NotFound" in msg
                    or "NoAuth" in msg
                ):
                    logger.info(
                        "UCloud UdnrDomainDNSQuery(zone=%s) not available: %s",
                        zone_name, exc,
                    )
                    continue
                raise
            # UdnrDomainDNSQuery returns an array under ``DataSet`` or
            # ``Records`` (varies by UCloud SDK version — handle both).
            rows = (
                data.get("DataSet")
                or data.get("Records")
                or data.get("Data")
                or []
            )
            for r in rows:
                # Record id is composed from zone + record content —
                # UDNR doesn't expose a stable numeric ID, and the
                # Add/Delete API addresses records by (Dn, RecordName,
                # DnsType, Content) tuple. Build a deterministic key.
                dns_type = str(r.get("DnsType") or r.get("RecordType") or "A")
                name = str(r.get("RecordName") or r.get("Name") or "@")
                value = str(r.get("Content") or r.get("RecordContent") or r.get("Value") or "")
                composite_id = f"{zone_name}::{name}::{dns_type}::{value}"
                records.append(CloudDnsRecord(
                    zone_name=zone_name,
                    record_id=composite_id,
                    name=name,
                    record_type=dns_type,
                    value=value,
                    ttl=int(r.get("TTL") or 600),
                    line=None,  # UDNR doesn't have line-routing
                    status="enable",
                    raw=_json_safe(r),
                ))
        return records

    # ── Load Balancers (ULB) ────────────────────────────────

    async def list_load_balancers(self) -> list[CloudLoadBalancer]:
        lbs: list[CloudLoadBalancer] = []
        # ULBId is globally unique per UCloud account but cheap-to-dedupe
        # across projects in case future API changes overlap the scopes.
        seen: set[str] = set()
        for project_id in self._project_ids:
            for region in self._regions:
                leg_label = f"project:{project_id or 'default'}/region:{region}"
                offset = 0
                limit = 100
                try:
                    while True:
                        data = await self._call(
                            "DescribeULB",
                            {
                                "Region": region,
                                "Offset": offset,
                                "Limit": limit,
                            },
                            project_id=project_id,
                        )
                        items = data.get("DataSet") or []
                        for lb in items:
                            lb_id_str = str(lb.get("ULBId") or "")
                            if not lb_id_str or lb_id_str in seen:
                                continue
                            seen.add(lb_id_str)
                            # UCloud ULB has BGP (public) vs Internal (internal).
                            bandwidth_type = str(lb.get("BandwidthType") or "").lower()
                            ulb_type = str(lb.get("ULBType") or "").lower()
                            network_type = "internal" if "internal" in ulb_type else "public"
                            # Extract first IP
                            ip_list = lb.get("IPSet") or []
                            ip_addr = ""
                            if ip_list and isinstance(ip_list, list):
                                first = ip_list[0]
                                ip_addr = str(
                                    first.get("EIP")
                                    or first.get("IP")
                                    or first.get("IPAddr")
                                    or "",
                                )
                            if not ip_addr:
                                ip_addr = str(lb.get("PrivateIP") or lb.get("IP") or "")
                            lbs.append(CloudLoadBalancer(
                                lb_id=lb_id_str,
                                name=str(lb.get("Name") or lb.get("ULBName") or ""),
                                lb_type="application",  # ULB is L4/L7 unified
                                network_type=network_type,
                                ip_address=ip_addr,
                                region=region,
                                status=(
                                    "active"
                                    if str(lb.get("ExpireTime") or "").lower() != "expired"
                                    else "inactive"
                                ),
                                raw=_json_safe({
                                    "ULBId": lb.get("ULBId"),
                                    "Name": lb.get("Name"),
                                    "ULBType": lb.get("ULBType"),
                                    "BandwidthType": bandwidth_type,
                                    "VPCId": lb.get("VPCId"),
                                    "SubnetId": lb.get("SubnetId"),
                                    "CreateTime": lb.get("CreateTime"),
                                    "ExpireTime": lb.get("ExpireTime"),
                                    "Tag": lb.get("Tag"),
                                }),
                            ))
                        total = int(data.get("TotalCount") or 0)
                        offset += limit
                        if offset >= total or not items:
                            break
                except ProviderAuthError:
                    raise
                except ProviderTransientError as exc:
                    # ULB is an independent product from UHost, so a tenant
                    # with UHost enabled in a region may still lack ULB
                    # there — RetCode=150 "Service unavailable" is the
                    # signal. Also skip "Region not available" (230).
                    exc_msg = str(exc)
                    if _is_region_skip_signal(exc_msg):
                        self.skipped_legs.append({
                            "leg": leg_label,
                            "capability": "load_balancers",
                            "reason": "region_not_provisioned",
                        })
                        logger.info(
                            "UCloud ULB %s unavailable: %s",
                            leg_label, exc,
                        )
                        continue
                    if "Permission" in exc_msg:
                        self.partial_failures.append({
                            "leg": leg_label,
                            "capability": "load_balancers",
                            "reason": "permission_denied",
                        })
                        logger.info(
                            "UCloud ULB %s permission denied: %s",
                            leg_label, exc,
                        )
                        continue
                    self.partial_failures.append({
                        "leg": leg_label,
                        "capability": "load_balancers",
                        "reason": "transient",
                    })
                    raise
        return lbs

    # ── CDN (UCDN) ──────────────────────────────────────────

    async def list_cdn_domains(self) -> list[CloudCdnDomain]:
        """List UCDN accelerated domains via ``GetUcdnDomainInfoList``.

        The older ``GetUcdnDomainConfig`` action requires a pre-known
        ``DomainId.N`` batch and is the wrong call for inventory
        (produces "Action not found" or 0 results on most tenants).
        """
        out: list[CloudCdnDomain] = []
        # UCDN domain_name is globally unique per UCloud account;
        # dedupe across projects for the same defensive reason as
        # list_servers / list_load_balancers.
        seen_cdn: set[str] = set()
        for project_id in self._project_ids:
            offset = 0
            limit = 100
            try:
                while True:
                    data = await self._call(
                        "GetUcdnDomainInfoList",
                        {"Offset": offset, "Limit": limit},
                        endpoint=_PCDN_ENDPOINT,
                        project_id=project_id,
                    )
                    items = (
                        data.get("DomainList")
                        or data.get("DomainInfoList")
                        or data.get("DataSet")
                        or []
                    )
                    for d in items:
                        domain_name = str(d.get("Domain") or d.get("DomainName") or "")
                        if not domain_name or domain_name in seen_cdn:
                            continue
                        seen_cdn.add(domain_name)
                        origin_list = (
                            d.get("OriginConf")
                            or d.get("Origin")
                            or d.get("OriginList")
                            or []
                        )
                        source_type, source_address = "domain", ""
                        if origin_list:
                            first = origin_list[0] if isinstance(origin_list, list) else origin_list
                            if isinstance(first, dict):
                                source_address = str(
                                    first.get("OriginHost")
                                    or first.get("OriginIp")
                                    or first.get("Host")
                                    or "",
                                )
                                ot = str(first.get("OriginType") or "domain").lower()
                                source_type = (
                                    "ip"
                                    if ot == "ip"
                                    else ("oss" if "ufile" in ot or "oss" in ot else "domain")
                                )
                        https_cfg = (
                            d.get("HttpsStatusCn")
                            or d.get("HttpsStatusAbroad")
                            or d.get("HttpsStatus")
                            or ""
                        )
                        is_https = str(https_cfg).lower() == "enable"
                        status_raw = str(d.get("Status") or d.get("DomainStatus") or "").lower()
                        status_map = {
                            "enable": "online",
                            "disable": "offline",
                            "checking": "configuring",
                            "checkfail": "offline",
                            "deleting": "configuring",
                        }
                        out.append(CloudCdnDomain(
                            domain_name=domain_name,
                            cname=str(d.get("Cname") or d.get("CName") or ""),
                            status=status_map.get(status_raw, "offline"),
                            source_type=source_type,
                            source_address=source_address,
                            is_https=is_https,
                            raw=_json_safe(d),
                        ))
                    total = int(data.get("TotalCount") or 0)
                    offset += limit
                    if offset >= total or not items:
                        break
            except ProviderTransientError as exc:
                # Match by numeric RetCode parsed from the message format
                # ``"UCloud {action} RetCode={ret}: {msg}"`` rather than
                # English substring — see UFile branch below for rationale.
                import re as _re
                m = _re.search(r"RetCode=(-?\d+)", str(exc))
                ret = int(m.group(1)) if m else None
                if ret in (100, 270, 271, 16001, 4001):
                    logger.info(
                        "UCloud UCDN not enabled in project=%s (RetCode=%s): %s",
                        project_id or "default", ret, exc,
                    )
                    # UCDN-not-enabled in THIS project is per-project structural —
                    # other projects may still have it. Skip this leg.
                    continue
                raise
        return out

    # ── Object Storage (UFile) ──────────────────────────────

    async def list_object_storage(self) -> list[dict]:
        buckets: list[dict] = []
        seen_buckets: set[str] = set()
        for project_id in self._project_ids:
            offset = 0
            limit = 100
            try:
                while True:
                    data = await self._call(
                        "DescribeBucket",
                        {"Offset": offset, "Limit": limit},
                        project_id=project_id,
                    )
                    items = data.get("DataSet") or data.get("Buckets") or []
                    for b in items:
                        bucket_name = str(b.get("BucketName") or b.get("Name") or "")
                        if not bucket_name or bucket_name in seen_buckets:
                            continue
                        seen_buckets.add(bucket_name)
                        region = str(b.get("Region") or b.get("BucketRegion") or "").lower() or None
                        bucket_type = str(b.get("Type") or b.get("BucketType") or "").lower()
                        access_type = "public" if bucket_type == "public" else "private"
                        buckets.append({
                            "bucket_name": bucket_name,
                            "region": region,
                            "storage_class": str(b.get("Biz") or b.get("Domain", {}).get("Src") or "") or None,
                            "access_type": access_type,
                            "status": "active",
                            "size_bytes": None,  # UFile list doesn't surface size
                            "object_count": None,
                            # UFile is queried once per UCloud project. Keep
                            # that project identity in the persisted payload
                            # so the directory, cost and owner resolvers can
                            # join the bucket to its scope.
                            "raw": _json_safe({**b, "_project_id": project_id}),
                        })
                    total = int(data.get("TotalCount") or 0)
                    offset += limit
                    if offset >= total or not items:
                        break
            except ProviderTransientError as exc:
                # Match by numeric RetCode rather than English substring —
                # see UCDN branch above for rationale.
                import re as _re
                m = _re.search(r"RetCode=(-?\d+)", str(exc))
                ret = int(m.group(1)) if m else None
                if ret in (100, 270, 271, 16001, 4001):
                    logger.info(
                        "UCloud UFile not enabled in project=%s (RetCode=%s): %s",
                        project_id or "default", ret, exc,
                    )
                    # Per-project skip — other projects may have UFile enabled.
                    continue
                raise
        return buckets

    # ── Monthly billing via ListUBillDetail (consumption) ────────
    #
    # UCloud's canonical *consumption* surface is ``ListUBillDetail``
    # keyed by ``BillingCycle`` (``YYYY-MM``) — one row per bill line
    # item with ``AmountReal`` (post-discount CNY). We SUM those and
    # group by ``ItemDetails[0].ProductName``. (The old order API,
    # ``DescribeOrderDetailInfo``, reported purchases/renewals — not
    # usage — and is retired; see the method docstring's history.)
    #
    # The ``BillingApi`` UCloud surfaces (DescribeMonthBalance) is
    # only available to the master account in finance-permission
    # tenants — ``ListUBillDetail`` covers single-account too. All
    # three billing reads (monthly / daily / cost_facts) force
    # ``project_id=None`` and read the WHOLE account's bill: Helix's
    # UCloud integration is wired as ONE integration row per account
    # (a second row for the same account would double-count every
    # billing figure).

    async def get_monthly_billing(self, month: date | None = None) -> dict:
        """Fetch one month's *consumption* cost via ``ListUBillDetail``.

        Aggregates the month's billing-cycle line items — the SAME
        authoritative source as ``list_billing_daily`` /
        ``list_cost_facts`` — so the monthly rollup equals the sum of
        the daily/line-item numbers for the month. UCloud bills in CNY;
        we surface ``currency='CNY'`` + ``total_cost_native`` and leave
        the CNY→USD rebase to the ``billing_sync`` handler.

        History (why not ``DescribeOrderDetailInfo``): the monthly path
        used to call the *order* API (new purchases / renewals), which
        (a) returned its list under ``OrderInfos`` while the parser read
        ``OrderDetails`` → always summed 0, and (b) even parsed
        correctly reports *orders*, not *consumption*: June 2026 had
        47.6 CNY of orders vs 13,167 CNY of actual usage — a ~276x
        undercount. Re-sourced to ``ListUBillDetail`` 2026-07-14.

        Account-wide (``project_id=None``): UCloud billing is
        account-level, so one cycle fetch returns every project's spend
        — no per-project fan-out (that's for resource *discovery*, not
        billing). A calendar month maps to exactly one UCloud
        ``BillingCycle`` (``YYYY-MM``).
        """
        target = (month or (date.today().replace(day=1) - timedelta(days=1))).replace(day=1)
        cycle = target.strftime("%Y-%m")
        if target.month == 12:
            next_month = target.replace(year=target.year + 1, month=1)
        else:
            next_month = target.replace(month=target.month + 1)
        # Bill rows carry a Unix-epoch usage-start; bucket the day-guard
        # in Asia/Shanghai (UCloud's billing TZ) so month-edge rows
        # aren't misattributed. Lazy import — not every provider needs it.
        from zoneinfo import ZoneInfo
        sh = ZoneInfo("Asia/Shanghai")

        cost_by_service: dict[str, float] = {}
        total = 0.0
        last_resp: dict[str, Any] = {}
        offset = 0
        limit = 100  # ListUBillDetail max page size
        max_pages = 500
        hit_cap = True
        for _ in range(max_pages):
            try:
                data = await self._call(
                    "ListUBillDetail",
                    {"BillingCycle": cycle, "Offset": offset, "Limit": limit},
                    project_id=None,  # account-wide — billing is account-level
                )
            except ProviderTransientError as exc:
                msg = str(exc)
                # ``UnknownAction`` / ``NotOpen`` = billing add-on not
                # enabled → fail-open with whatever we have (mirrors the
                # daily/cost_facts paths). Real auth/permission failures
                # re-raise so the handler records the failure.
                if any(x in msg for x in ("UnknownAction", "NotOpen")):
                    logger.info(
                        "UCloud ListUBillDetail not enabled "
                        "(monthly billing path): %s", msg,
                    )
                    last_resp = {"reason": msg[:300], "items": 0}
                    hit_cap = False
                    break
                raise
            last_resp = data
            # Response key is ``Items`` (live API 2026-04-24); tolerate
            # ``UBillDataSet`` / ``DataSet`` in case UCloud renames it.
            items = (
                data.get("Items")
                or data.get("UBillDataSet")
                or data.get("DataSet")
                or []
            )
            for row in items:
                # A cycle == the calendar month, but guard month-edge
                # rows by usage-start day (same key priority as the
                # daily path: StartTime → CreateTime → PayTime).
                # Missing / unparseable timestamps SKIP the row — the
                # daily and cost_facts paths both drop them, and the
                # monthly rollup must stay reconcilable against
                # sum(daily) for the month. Counting the un-dateable
                # rows only here made monthly ≠ Σdaily.
                ts = row.get("StartTime") or row.get("CreateTime") or row.get("PayTime")
                if not ts:
                    continue
                try:
                    if isinstance(ts, (int, float)) or str(ts).isdigit():
                        day = datetime.fromtimestamp(int(ts), tz=sh).date()
                    else:
                        day = datetime.strptime(str(ts)[:10], "%Y-%m-%d").date()
                except (ValueError, OSError):
                    continue
                if day < target or day >= next_month:
                    continue
                # ``AmountReal`` is post-discount (what UCloud actually
                # bills); fall back to gross ``Amount`` / ``TotalPayPrice``.
                # Presence-based (not truthy) — AmountReal=0 stays 0.
                amt_s = _first_present(
                    row, ("AmountReal", "Amount", "TotalPayPrice"),
                )
                try:
                    amt = float(amt_s) if amt_s is not None else 0.0
                except (TypeError, ValueError):
                    continue
                # Service name: prefer human-readable
                # ``ItemDetails[0].ProductName``; fall back to the
                # resource-type code. Same bucketing as the daily path
                # (fallback ``"Other"``) so monthly cost_by_service lines
                # up with the daily/cost_facts service split.
                item_details = row.get("ItemDetails") or []
                svc = ""
                if isinstance(item_details, list) and item_details:
                    first = item_details[0]
                    if isinstance(first, dict):
                        svc = str(first.get("ProductName") or "")
                if not svc:
                    svc = str(
                        row.get("ResourceType")
                        or row.get("ProductName")
                        or row.get("BusinessName")
                        or "Other",
                    )
                cost_by_service[svc] = cost_by_service.get(svc, 0.0) + amt
                total += amt
            if len(items) < limit:
                hit_cap = False
                break
            offset += limit
        if hit_cap:
            # A partial monthly snapshot silently undercounts — raise so
            # the handler records the failure instead of persisting a
            # truncated total (matches the daily/cost_facts cap posture).
            raise ProviderTransientError(
                f"UCloud ListUBillDetail hit {max_pages}-page cap for "
                f"monthly cycle={cycle} — billing data is partial. Raise "
                f"max_pages if the account exceeds {max_pages * limit} line "
                f"items per cycle.",
            )
        return {
            "month": target,
            # UCloud bills in CNY but runs no in-driver FX conversion, so it
            # MUST surface ``total_cost_native`` (+ ``currency='CNY'``) —
            # that's the trigger the billing_sync handler's CNY→USD rebase
            # path keys on (``currency=='CNY' and total_cost_native is not
            # None``). Without it (the bug) the raw CNY amount was persisted
            # into ``total_cost_usd`` as if USD → ~7x overstatement. Same
            # fix shape as Tencent/Aliyun. ``total_cost_usd`` keeps the
            # native value purely as the handler's bounded FX-failure
            # fallback (it then keeps ``currency='CNY'`` so value+label agree).
            "total_cost_usd": total,
            "total_cost_native": total,
            "cost_by_service": cost_by_service,
            "currency": "CNY",
            "raw_response": last_resp,
        }


# ── Helpers ──────────────────────────────────────────────────


def _server_from_uhost(inst: dict[str, Any], region: str) -> CloudServer:
    """Translate a UCloud ``UHostInstance`` dict into a CloudServer."""
    raw_status = str(inst.get("State") or "").strip()
    status = _STATUS_MAP_UHOST.get(raw_status, raw_status.lower() or "unknown")

    public_ips: list[str] = []
    private_ips: list[str] = []
    for nic in inst.get("IPSet") or []:
        ip = str(nic.get("IP") or nic.get("IPAddr") or "")
        if not ip:
            continue
        ip_type = str(nic.get("Type") or "").lower()
        if ip_type == "private" or ip_type == "internal":
            private_ips.append(ip)
        else:
            public_ips.append(ip)

    tags_list = inst.get("Tag") or inst.get("Tags") or []
    tags: dict[str, str] = {}
    if isinstance(tags_list, list):
        for t in tags_list:
            if isinstance(t, dict):
                k = t.get("Key") or t.get("key")
                v = t.get("Value") or t.get("value")
                if k is not None:
                    tags[str(k)] = str(v) if v is not None else ""
    elif isinstance(tags_list, str) and tags_list:
        tags["Tag"] = tags_list

    os_name = str(inst.get("OsName") or inst.get("OSType") or "").lower()
    os_type = "windows" if "windows" in os_name or "win" in os_name else (
        "linux" if os_name else None
    )

    cpu = inst.get("CPU")
    memory = inst.get("Memory")

    raw = _json_safe({
        "UHostId": inst.get("UHostId"),
        "Name": inst.get("Name"),
        "State": inst.get("State"),
        "Zone": inst.get("Zone"),
        "HostType": inst.get("HostType"),
        "OsName": inst.get("OsName"),
        "OSType": inst.get("OSType"),
        "ImageId": inst.get("ImageId"),
        "CPU": cpu,
        "Memory": memory,
        "CreateTime": inst.get("CreateTime"),
        "ExpireTime": inst.get("ExpireTime"),
        "IPSet": inst.get("IPSet"),
        "Tag": tags_list,
    })

    return CloudServer(
        instance_id=str(inst.get("UHostId") or ""),
        name=str(inst.get("Name") or inst.get("UHostId") or ""),
        status=status,
        region=region,
        zone=str(inst.get("Zone") or "") or None,
        vcpu=int(cpu) if cpu is not None and str(cpu).isdigit() else None,
        memory_mb=int(memory) if memory is not None and str(memory).isdigit() else None,
        public_ips=public_ips,
        private_ips=private_ips,
        image_id=str(inst.get("ImageId") or "") or None,
        os_type=os_type,
        tags=tags,
        raw=raw,
    )