from __future__ import annotations

import hashlib
import json
import os
import py_compile
import re
import shutil
import ssl
import subprocess
import sys
import urllib.error
import urllib.parse
import urllib.request
import webbrowser
import zipfile
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Callable

import tkinter as tk

from common.ui_helpers import ask_app_confirm_dialog, show_app_message_dialog


VERSION_API_URL = "https://www.wawaapi.com/api/version"
STATIC_MANIFEST_URL = "https://www.wawaapi.com/YunTools/manifest.json"
PATCH_MANIFEST_URL = "https://www.wawaapi.com/YunTools/updates/manifest.json"
DOWNLOAD_PAGE_URL = "https://www.wawaapi.com/YunTools/"
HUAWEI_DOWNLOAD_PAGE_URL = "https://www.wawaapi.com/YunTools/Huawei/"
HUAWEI_STATIC_MANIFEST_URL = "https://www.wawaapi.com/YunTools/Huawei/manifest.json"
HUAWEI_PATCH_MANIFEST_URL = "https://www.wawaapi.com/YunTools/Huawei/updates/manifest.json"
CTYUN_DOWNLOAD_PAGE_URL = "https://www.wawaapi.com/YunTools/CTYUN/"
CTYUN_STATIC_MANIFEST_URL = "https://www.wawaapi.com/YunTools/CTYUN/manifest.json"
CTYUN_PATCH_MANIFEST_URL = "https://www.wawaapi.com/YunTools/CTYUN/updates/manifest.json"
VOLCENGINE_DOWNLOAD_PAGE_URL = "https://www.wawaapi.com/YunTools/Volcengine/"
VOLCENGINE_STATIC_MANIFEST_URL = "https://www.wawaapi.com/YunTools/Volcengine/manifest.json"
VOLCENGINE_PATCH_MANIFEST_URL = "https://www.wawaapi.com/YunTools/Volcengine/updates/manifest.json"
CHECK_TIMEOUT_SECONDS = 5
UPDATE_DOWNLOAD_TIMEOUT_SECONDS = 90
FULL_UPDATE_DOWNLOAD_TIMEOUT_SECONDS = 10 * 60
UPDATE_COUNTDOWN_SECONDS = 5
UPDATE_FAILED_CONTACT_MESSAGE = "更新失败，请联系客户经理处理。"
FULL_UPDATE_HEALTH_MARKER_ENV = "WAWA_FULL_UPDATE_HEALTH_MARKER"
DEFAULT_UPDATE_PACKAGE = "desktop"
HUAWEI_UPDATE_PACKAGE = "huawei"
CTYUN_UPDATE_PACKAGE = "ctyun"
VOLCENGINE_UPDATE_PACKAGE = "volcengine"
UPDATE_CHANNELS = {
    DEFAULT_UPDATE_PACKAGE: {
        "client": "desktop",
        "version_api_url": VERSION_API_URL,
        "static_manifest_url": STATIC_MANIFEST_URL,
        "patch_manifest_url": PATCH_MANIFEST_URL,
        "download_page_url": DOWNLOAD_PAGE_URL,
        "user_agent": "WawaCloudDesktop",
    },
    HUAWEI_UPDATE_PACKAGE: {
        "client": "huawei",
        "version_api_url": VERSION_API_URL,
        "static_manifest_url": HUAWEI_STATIC_MANIFEST_URL,
        "patch_manifest_url": HUAWEI_PATCH_MANIFEST_URL,
        "download_page_url": HUAWEI_DOWNLOAD_PAGE_URL,
        "user_agent": "WawaCloudHuawei",
    },
    CTYUN_UPDATE_PACKAGE: {
        "client": "ctyun",
        "version_api_url": VERSION_API_URL,
        "static_manifest_url": CTYUN_STATIC_MANIFEST_URL,
        "patch_manifest_url": CTYUN_PATCH_MANIFEST_URL,
        "download_page_url": CTYUN_DOWNLOAD_PAGE_URL,
        "user_agent": "WawaCloudCTYUN",
    },
    VOLCENGINE_UPDATE_PACKAGE: {
        "client": "volcengine",
        "version_api_url": VERSION_API_URL,
        "static_manifest_url": VOLCENGINE_STATIC_MANIFEST_URL,
        "patch_manifest_url": VOLCENGINE_PATCH_MANIFEST_URL,
        "download_page_url": VOLCENGINE_DOWNLOAD_PAGE_URL,
        "user_agent": "WawaCloudVolcengine",
    },
}
ALLOWED_PATCH_PREFIXES = (
    "source/",
    "assets/",
    "data/CTYUN/config/",
)
ALLOWED_PATCH_FILES = (
    "更新说明.txt",
)
HUAWEI_ALLOWED_PATCH_PREFIXES = (
    "source/main_huawei.py",
    "source/main_huawei_shell.py",
    "source/launcher/",
    "source/common/",
    "source/HUAWEI/",
    "source/assets/",
    "assets/",
    "data/HUAWEI/",
)
CTYUN_ALLOWED_PATCH_PREFIXES = (
    "source/main_ctyun.py",
    "source/launcher/",
    "source/common/",
    "source/CTYUN/",
    "source/assets/",
    "assets/",
    "data/CTYUN/",
)
VOLCENGINE_ALLOWED_PATCH_PREFIXES = (
    "source/main_volcengine.py",
    "source/launcher/",
    "source/common/",
    "source/Volcengine/",
    "source/assets/",
    "assets/",
    "data/Volcengine/",
)
STARTUP_UPDATE_FEATURE_TITLE = "新增队列任务功能"
STARTUP_UPDATE_FEATURE_DETAIL = "任务执行中可继续添加待办队列，当前任务完成后自动衔接执行，减少等待和重复操作。"
STARTUP_UPDATE_OPTIMIZATION_TITLE = "软件优化"
STARTUP_NOTICE_SEPARATOR = "────────────────────────"
UI_FONT_FAMILY = "SimSun"
STARTUP_NOTICE_CONTENT_FONT = (UI_FONT_FAMILY, 11)
STARTUP_NOTICE_TITLE_FONT = (UI_FONT_FAMILY, 12, "bold")


class UpdateCheckError(RuntimeError):
    """更新服务器不可用或返回内容无法识别时抛出。"""

    def __init__(self, message: str, *, display_reason: str = "") -> None:
        super().__init__(message)
        self.display_reason = display_reason


@dataclass
class UpdateCheckResult:
    allowed: bool
    needs_update: bool
    current_version: str
    latest_version: str
    download_url: str
    message: str
    error: str = ""
    auto_update_available: bool = False
    patches: list[dict[str, Any]] = field(default_factory=list)
    restart_required: bool = True
    package_key: str = DEFAULT_UPDATE_PACKAGE
    download_page_url: str = DOWNLOAD_PAGE_URL
    disabled: bool = False
    announcement: dict[str, Any] = field(default_factory=dict)
    auto_migration_available: bool = False
    migration: dict[str, Any] = field(default_factory=dict)
    full_package: dict[str, Any] = field(default_factory=dict)
    update_retry_blocked: bool = False


@dataclass
class AutoUpdateResult:
    ok: bool
    message: str
    restart_required: bool = True
    error: str = ""
    external_restart: bool = False
    retry_blocked: bool = False


ProgressCallback = Callable[[str], None]


def _app_root() -> Path:
    return Path(__file__).resolve().parents[2]


def _data_dir() -> Path:
    path = _app_root() / "data"
    path.mkdir(parents=True, exist_ok=True)
    return path


def _normalize_package_key(package_key: str | None) -> str:
    value = str(package_key or DEFAULT_UPDATE_PACKAGE).strip().lower()
    if value in {"huawei", "huaweicloud", "huawei_cloud", "huawei-standalone"}:
        return HUAWEI_UPDATE_PACKAGE
    if value in {"ctyun", "ct", "tianyi", "ctyun-standalone"}:
        return CTYUN_UPDATE_PACKAGE
    if value in {"volc", "volcengine", "volc_engine", "huoshan", "volcengine-standalone"}:
        return VOLCENGINE_UPDATE_PACKAGE
    return DEFAULT_UPDATE_PACKAGE


def _channel(package_key: str | None) -> dict[str, str]:
    return UPDATE_CHANNELS[_normalize_package_key(package_key)]


def _compact_notice_lines(text: str) -> list[str]:
    return [line.strip() for line in str(text or "").splitlines() if line.strip()]


def _startup_update_items(
    has_new_feature: bool = True,
    has_optimization: bool = True,
) -> list[tuple[int, str, str]]:
    items: list[tuple[int, str, str]] = []
    if has_new_feature:
        items.append((1, STARTUP_UPDATE_FEATURE_TITLE, STARTUP_UPDATE_FEATURE_DETAIL))
    if has_optimization:
        items.append((2, STARTUP_UPDATE_OPTIMIZATION_TITLE, ""))
    return items


def build_startup_notice_text(
    disclaimer_body: str,
    *,
    has_new_feature: bool = True,
    has_optimization: bool = True,
) -> str:
    lines = [
        "免责声明",
        *_compact_notice_lines(disclaimer_body),
    ]
    update_items = _startup_update_items(has_new_feature, has_optimization)
    if update_items:
        lines.extend([STARTUP_NOTICE_SEPARATOR, "更新说明"])
        for number, title, detail in update_items:
            lines.append(f"{number}. {title}")
            if detail:
                lines.append(f"   {detail}")
    return "\n".join(lines).strip()


def _insert_notice_line(widget: tk.Text, text: str, tag: str) -> None:
    if widget.index("end-1c") != "1.0":
        widget.insert(tk.END, "\n")
    widget.insert(tk.END, text, (tag,))


def insert_startup_notice_text(
    widget: tk.Text,
    disclaimer_body: str,
    *,
    has_new_feature: bool = True,
    has_optimization: bool = True,
) -> None:
    widget.configure(font=STARTUP_NOTICE_CONTENT_FONT, wrap="word")
    widget.tag_configure(
        "notice_title",
        justify=tk.CENTER,
        font=STARTUP_NOTICE_TITLE_FONT,
        spacing1=0,
        spacing3=6,
    )
    widget.tag_configure(
        "notice_body",
        justify=tk.LEFT,
        font=STARTUP_NOTICE_CONTENT_FONT,
        lmargin1=18,
        lmargin2=18,
        rmargin=18,
        spacing3=5,
    )
    widget.tag_configure(
        "notice_separator",
        justify=tk.CENTER,
        font=STARTUP_NOTICE_CONTENT_FONT,
        spacing1=6,
        spacing3=6,
    )
    widget.tag_configure(
        "notice_update_item",
        justify=tk.LEFT,
        font=(UI_FONT_FAMILY, 11, "bold"),
        lmargin1=18,
        lmargin2=36,
        rmargin=18,
        spacing3=3,
    )
    widget.tag_configure(
        "notice_update_detail",
        justify=tk.LEFT,
        font=STARTUP_NOTICE_CONTENT_FONT,
        lmargin1=36,
        lmargin2=36,
        rmargin=18,
        spacing3=6,
    )

    widget.delete("1.0", tk.END)
    _insert_notice_line(widget, "免责声明", "notice_title")
    for line in _compact_notice_lines(disclaimer_body):
        _insert_notice_line(widget, line, "notice_body")

    update_items = _startup_update_items(has_new_feature, has_optimization)
    if not update_items:
        return

    _insert_notice_line(widget, STARTUP_NOTICE_SEPARATOR, "notice_separator")
    _insert_notice_line(widget, "更新说明", "notice_title")
    for number, title, detail in update_items:
        _insert_notice_line(widget, f"{number}. {title}", "notice_update_item")
        if detail:
            _insert_notice_line(widget, detail, "notice_update_detail")


def _state_path(package_key: str | None = DEFAULT_UPDATE_PACKAGE) -> Path:
    key = _normalize_package_key(package_key)
    if key == DEFAULT_UPDATE_PACKAGE:
        return _data_dir() / "update_state.json"
    return _data_dir() / f"update_state_{key}.json"


def _version_key(value: str) -> tuple[int, ...]:
    parts = [int(item) for item in re.findall(r"\d+", str(value or ""))]
    return tuple(parts[:6])


def _version_lt(left: str, right: str) -> bool:
    left_key = _version_key(left)
    right_key = _version_key(right)
    width = max(len(left_key), len(right_key), 1)
    left_key = left_key + (0,) * (width - len(left_key))
    right_key = right_key + (0,) * (width - len(right_key))
    return left_key < right_key


def _trusted_ssl_context() -> ssl.SSLContext:
    try:
        import certifi

        return ssl.create_default_context(cafile=certifi.where())
    except Exception:
        return ssl.create_default_context()


def urlopen_with_fallback(
    request: urllib.request.Request | str,
    *,
    timeout: float,
):
    """使用内置 CA 请求；系统代理失败时再尝试不走代理直连。"""
    context = _trusted_ssl_context()
    direct_request: urllib.request.Request | str = request
    if isinstance(request, urllib.request.Request):
        # ProxyHandler mutates Request in place, so the direct retry needs a
        # clean copy made before the first attempt.
        direct_request = urllib.request.Request(
            request.full_url,
            data=request.data,
            headers=dict(request.header_items()),
            origin_req_host=request.origin_req_host,
            unverifiable=request.unverifiable,
            method=request.get_method(),
        )
    try:
        return urllib.request.urlopen(request, timeout=timeout, context=context)
    except Exception as proxy_error:
        direct_opener = urllib.request.build_opener(
            urllib.request.ProxyHandler({}),
            urllib.request.HTTPSHandler(context=context),
        )
        try:
            return direct_opener.open(direct_request, timeout=timeout)
        except Exception as direct_error:
            raise direct_error from proxy_error


def _friendly_connection_error(exc: BaseException) -> str:
    display_reason = str(getattr(exc, "display_reason", "") or "").strip()
    if display_reason:
        return display_reason
    if isinstance(exc, urllib.error.HTTPError):
        return f"服务器返回 HTTP {exc.code}"

    parts: list[str] = []
    current: BaseException | None = exc
    seen: set[int] = set()
    while current is not None and id(current) not in seen:
        seen.add(id(current))
        parts.append(f"{type(current).__name__}: {current}")
        current = current.__cause__ or current.__context__
    text = " | ".join(parts).lower()
    if "certificate verify failed" in text or "sslcertverificationerror" in text:
        return "证书校验失败"
    if "getaddrinfo failed" in text or "name or service not known" in text or "11001" in text:
        return "域名解析失败"
    if "timed out" in text or "timeout" in text or "10060" in text:
        return "连接超时"
    if "tunnel connection failed" in text or "proxy" in text:
        return "系统代理连接失败"
    if "connection refused" in text or "10061" in text:
        return "连接被拒绝"
    if "connection reset" in text or "remote end closed" in text or "10054" in text:
        return "连接被中断"
    if "不是 json" in text or "格式不正确" in text:
        return "服务器返回内容异常"
    return f"网络连接失败（{type(exc).__name__}）"


def _fetch_json(url: str, current_version: str, package_key: str | None = DEFAULT_UPDATE_PACKAGE) -> dict[str, Any]:
    key = _normalize_package_key(package_key)
    channel = _channel(key)
    query = urllib.parse.urlencode(
        {
            "current": current_version,
            "client": channel["client"],
            "package": key,
        }
    )
    target = f"{url}?{query}" if "?" not in url else f"{url}&{query}"
    request = urllib.request.Request(
        target,
        headers={
            "Accept": "application/json",
            "Cache-Control": "no-cache",
            "User-Agent": f"{channel['user_agent']}/{current_version}",
        },
    )
    with urlopen_with_fallback(request, timeout=CHECK_TIMEOUT_SECONDS) as response:
        raw = response.read(512 * 1024)
    try:
        data = json.loads(raw.decode("utf-8-sig"))
    except Exception as exc:
        raise UpdateCheckError(f"更新接口返回内容不是 JSON: {exc}") from exc
    if not isinstance(data, dict):
        raise UpdateCheckError("更新接口返回格式不正确")
    return data


def _load_manifest(current_version: str, package_key: str | None = DEFAULT_UPDATE_PACKAGE) -> dict[str, Any]:
    errors: list[tuple[str, BaseException]] = []
    channel = _channel(package_key)
    for url in (channel["version_api_url"], channel["static_manifest_url"]):
        try:
            return _fetch_json(url, current_version, package_key)
        except Exception as exc:
            errors.append((url, exc))
    reasons = list(dict.fromkeys(_friendly_connection_error(exc) for _url, exc in errors))
    details = "; ".join(f"{url}: {type(exc).__name__}: {exc}" for url, exc in errors)
    raise UpdateCheckError(details, display_reason="、".join(reasons)) from errors[-1][1]


def _read_update_state(package_key: str | None = DEFAULT_UPDATE_PACKAGE) -> dict[str, Any]:
    path = _state_path(package_key)
    if not path.exists():
        return {}
    try:
        data = json.loads(path.read_text(encoding="utf-8-sig"))
    except Exception:
        return {}
    return data if isinstance(data, dict) else {}


def read_installed_update_metadata(package_key: str | None = DEFAULT_UPDATE_PACKAGE) -> dict[str, str]:
    key = _normalize_package_key(package_key)
    state = _read_update_state(key)
    return {
        "package_key": key,
        "base_version": str(state.get("base_version") or "").strip(),
        "patch_version": str(state.get("patch_version") or "").strip(),
    }


def _write_update_state(data: dict[str, Any], package_key: str | None = DEFAULT_UPDATE_PACKAGE) -> None:
    _state_path(package_key).write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")


def _update_attempt_record(state: dict[str, Any], state_key: str, kind: str) -> dict[str, str]:
    raw = state.get(state_key)
    if not isinstance(raw, dict) or str(raw.get("kind") or "").strip() != kind:
        return {}
    version = str(raw.get("version") or "").strip()
    if not version:
        return {}
    return {
        "kind": kind,
        "version": version,
        "error": str(raw.get("error") or "").strip(),
    }


def _mark_pending_update(state: dict[str, Any], kind: str, version: str) -> None:
    retry = _update_attempt_record(state, "interrupted_retry", kind)
    if retry.get("version") != version:
        state.pop("interrupted_retry", None)
    state["pending_update"] = {"kind": kind, "version": version}
    state.pop("failed_update", None)


def _prepare_interrupted_update_retry(package_key: str, kind: str) -> tuple[str, bool]:
    state = _read_update_state(package_key)
    failed = _update_attempt_record(state, "failed_update", kind)
    if failed:
        return failed["version"], False

    pending = _update_attempt_record(state, "pending_update", kind)
    if not pending:
        return "", False

    version = pending["version"]
    retry = _update_attempt_record(state, "interrupted_retry", kind)
    if retry.get("version") == version:
        _record_update_failure(package_key, kind, version, "update interrupted twice")
        return version, False

    state.pop("pending_update", None)
    state["interrupted_retry"] = {"kind": kind, "version": version}
    try:
        _write_update_state(state, package_key)
    except OSError:
        return version, False
    return "", True


def _set_pending_update(package_key: str, kind: str, version: str) -> None:
    state = _read_update_state(package_key)
    _mark_pending_update(state, kind, version)
    _write_update_state(state, package_key)


def _record_update_failure(package_key: str, kind: str, version: str, error: BaseException | str) -> None:
    state = _read_update_state(package_key)
    state.pop("pending_update", None)
    state.pop("interrupted_retry", None)
    state["failed_update"] = {
        "kind": kind,
        "version": version,
        "error": str(error or "")[:500],
    }
    try:
        _write_update_state(state, package_key)
    except OSError:
        pass


def _clear_completed_update_attempt(current_version: str, package_key: str) -> None:
    state = _read_update_state(package_key)
    local_patch = str(state.get("patch_version") or "").strip()
    changed = False
    for state_key in ("pending_update", "failed_update", "interrupted_retry"):
        raw = state.get(state_key)
        if not isinstance(raw, dict):
            continue
        kind = str(raw.get("kind") or "").strip()
        target_version = str(raw.get("version") or "").strip()
        completed = (
            kind == "full" and target_version and not _version_lt(current_version, target_version)
        ) or (
            kind == "patch" and target_version and local_patch and not _version_lt(local_patch, target_version)
        )
        if completed:
            state.pop(state_key, None)
            changed = True
    if changed:
        try:
            _write_update_state(state, package_key)
        except OSError:
            pass


def _manager_auth_store_path() -> Path:
    local_app_data = os.environ.get("LOCALAPPDATA", "").strip()
    root = Path(local_app_data) if local_app_data else Path.home() / "AppData" / "Local"
    return root / "WaWaCloud" / "manager_auth_codes.json"


def _read_manager_auth_store() -> dict[str, Any]:
    path = _manager_auth_store_path()
    if not path.exists():
        return {}
    try:
        data = json.loads(path.read_text(encoding="utf-8-sig"))
    except Exception:
        return {}
    return data if isinstance(data, dict) else {}


def _write_manager_auth_store(data: dict[str, Any]) -> None:
    path = _manager_auth_store_path()
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary_path = path.with_suffix(f"{path.suffix}.tmp")
    temporary_path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
    temporary_path.replace(path)


def read_saved_manager_auth_code(package_key: str | None = DEFAULT_UPDATE_PACKAGE) -> str:
    key = _normalize_package_key(package_key)
    store = _read_manager_auth_store()
    normalized = re.sub(r"\s+", "", str(store.get(key) or "")).upper()
    if normalized:
        return normalized

    legacy_state = _read_update_state(key)
    normalized = re.sub(r"\s+", "", str(legacy_state.get("manager_auth_code") or "")).upper()
    if normalized:
        save_manager_auth_code(normalized, key)
    return normalized


def save_manager_auth_code(code: str, package_key: str | None = DEFAULT_UPDATE_PACKAGE) -> None:
    key = _normalize_package_key(package_key)
    normalized = re.sub(r"\s+", "", str(code or "")).upper()
    store = _read_manager_auth_store()
    if normalized:
        store[key] = normalized
    else:
        store.pop(key, None)
    try:
        _write_manager_auth_store(store)
    except OSError:
        pass

    # 继续同步旧状态文件，兼容尚未安装本修复的旧启动代码。
    state = _read_update_state(key)
    if normalized:
        state["manager_auth_code"] = normalized
    else:
        state.pop("manager_auth_code", None)
    try:
        _write_update_state(state, key)
    except OSError:
        # 便携包位于只读目录时不应因“记住授权码”阻止正常进入软件。
        return


def _needs_update(data: dict[str, Any], current_version: str) -> bool:
    if "needs_update" in data:
        return bool(data.get("needs_update"))
    latest = str(data.get("latest_version") or "").strip()
    minimum = str(data.get("min_supported_version") or latest).strip()
    force_update = bool(data.get("force_update", True))
    if minimum and _version_lt(current_version, minimum):
        return True
    if force_update and latest and _version_lt(current_version, latest):
        return True
    return False


def _is_disabled(data: dict[str, Any]) -> bool:
    if "enabled" in data:
        return not bool(data.get("enabled"))
    if "disabled" in data:
        return bool(data.get("disabled"))
    return False


def _disabled_message(data: dict[str, Any], current_version: str) -> str:
    message = str(data.get("disabled_message") or data.get("message") or "").strip()
    if message:
        return message
    return f"当前版本已停用，暂时无法继续使用。\n当前版本：{current_version}"


def _announcement_from_manifest(data: dict[str, Any]) -> dict[str, Any]:
    raw = data.get("announcement")
    if not isinstance(raw, dict) or not bool(raw.get("enabled")):
        return {}
    title = str(raw.get("title") or "\U0001f4e2 \u8f6f\u4ef6\u516c\u544a").strip()
    message = str(raw.get("message") or raw.get("body") or "").strip()
    if not message:
        return {}
    announcement_id = str(raw.get("id") or "").strip()
    if not announcement_id:
        announcement_id = hashlib.sha256(f"{title}\n{message}".encode("utf-8")).hexdigest()[:16]
    return {
        "id": announcement_id,
        "title": title,
        "message": message,
        "show_once": bool(raw.get("show_once", True)),
    }


def _migration_from_manifest(data: dict[str, Any], package_key: str | None = DEFAULT_UPDATE_PACKAGE) -> dict[str, Any]:
    key = _normalize_package_key(package_key)
    if key == DEFAULT_UPDATE_PACKAGE:
        return {}
    raw = data.get("migration")
    if not isinstance(raw, dict) or not bool(raw.get("enabled")):
        raw = data.get("migration_v2")
    if not isinstance(raw, dict) or not bool(raw.get("enabled")):
        return {}
    target_package = _normalize_package_key(str(raw.get("target_package") or DEFAULT_UPDATE_PACKAGE))
    if target_package != DEFAULT_UPDATE_PACKAGE:
        return {}
    package_url = str(raw.get("package_url") or raw.get("url") or "").strip()
    target_version = str(raw.get("target_version") or data.get("latest_version") or "").strip()
    if not package_url or not target_version:
        return {}
    title = str(raw.get("title") or "分包升级").strip()
    message = str(raw.get("message") or "当前分包将自动升级为多云管理。").strip()
    return {
        "enabled": True,
        "target_package": DEFAULT_UPDATE_PACKAGE,
        "target_version": target_version,
        "package_url": package_url,
        "sha256": str(raw.get("sha256") or "").strip().lower(),
        "size": int(raw.get("size") or 0),
        "title": title,
        "message": message,
    }


def _full_package_from_manifest(data: dict[str, Any], package_key: str) -> dict[str, Any]:
    if _normalize_package_key(package_key) != DEFAULT_UPDATE_PACKAGE:
        return {}
    version = str(data.get("latest_version") or "").strip()
    url = str(data.get("download_url") or "").strip()
    expected_hash = str(data.get("sha256") or "").strip().lower()
    parsed_url = urllib.parse.urlparse(url)
    if (
        not version
        or parsed_url.scheme.lower() != "https"
        or not parsed_url.path.lower().endswith(".zip")
        or not re.fullmatch(r"[0-9a-f]{64}", expected_hash)
    ):
        return {}
    try:
        size = max(0, int(data.get("size") or 0))
    except (TypeError, ValueError):
        size = 0
    return {
        "version": version,
        "url": url,
        "sha256": expected_hash,
        "size": size,
        "file_name": str(data.get("file_name") or Path(parsed_url.path).name or "YunTools.zip").strip(),
    }


def show_release_announcement_if_needed(parent: tk.Misc, result: UpdateCheckResult | None) -> bool:
    if result is None or not result.allowed:
        return False
    announcement = getattr(result, "announcement", {}) or {}
    if not isinstance(announcement, dict):
        return False
    message = str(announcement.get("message") or "").strip()
    if not message:
        return False
    announcement_id = str(announcement.get("id") or "").strip()
    show_once = bool(announcement.get("show_once", True))
    package_key = _normalize_package_key(getattr(result, "package_key", DEFAULT_UPDATE_PACKAGE))
    state = _read_update_state(package_key)
    seen_ids = state.get("announcement_seen_ids")
    if not isinstance(seen_ids, list):
        seen_ids = []
    seen_ids = [str(item) for item in seen_ids if str(item).strip()]
    if show_once and announcement_id and announcement_id in seen_ids:
        return False
    title = str(announcement.get("title") or "\U0001f4e2 \u8f6f\u4ef6\u516c\u544a").strip()
    show_app_message_dialog(parent, title, message, modal=True)
    if show_once and announcement_id:
        seen_ids.append(announcement_id)
        state["announcement_seen_ids"] = seen_ids[-20:]
        _write_update_state(state, package_key)
    return True


def _patches_for_current_version(current_version: str, package_key: str | None = DEFAULT_UPDATE_PACKAGE) -> list[dict[str, Any]]:
    channel = _channel(package_key)
    try:
        data = _fetch_json(channel["patch_manifest_url"], current_version, package_key)
    except Exception:
        return []
    if str(data.get("base_version") or "").strip() != current_version:
        return []
    state = _read_update_state(package_key)
    local_patch = str(state.get("patch_version") or "").strip()
    patches = data.get("patches")
    if not isinstance(patches, list):
        return []
    needed: list[dict[str, Any]] = []
    for patch in patches:
        if not isinstance(patch, dict):
            continue
        version = str(patch.get("version") or "").strip()
        url = str(patch.get("url") or patch.get("patch_url") or "").strip()
        if not version or not url:
            continue
        if _version_lt(local_patch, version):
            needed.append(patch)
            continue
    needed.sort(key=lambda item: _version_key(str(item.get("version") or "")))
    return needed


def check_latest_version(current_version: str, package_key: str | None = DEFAULT_UPDATE_PACKAGE) -> UpdateCheckResult:
    key = _normalize_package_key(package_key)
    channel = _channel(key)
    download_page_url = channel["download_page_url"]
    _clear_completed_update_attempt(current_version, key)
    try:
        data = _load_manifest(current_version, key)
    except Exception as exc:
        display_reason = _friendly_connection_error(exc)
        return UpdateCheckResult(
            allowed=False,
            needs_update=True,
            current_version=current_version,
            latest_version="未知",
            download_url=download_page_url,
            message=(
                f"更新服务连接失败：{display_reason}，请点击“重新检测”。\n"
                "软件暂时无法继续使用。\n"
                f"当前版本：{current_version}\n"
                f"下载地址：{download_page_url}\n"
                f"错误信息：{exc}"
            ),
            error=str(exc),
            package_key=key,
            download_page_url=download_page_url,
        )

    latest = str(data.get("latest_version") or "最新版本")
    download_page_url = str(data.get("download_page_url") or data.get("download_page") or download_page_url).strip() or download_page_url
    download_url = str(data.get("download_url") or download_page_url)
    announcement = _announcement_from_manifest(data)
    migration = _migration_from_manifest(data, key)
    if _is_disabled(data):
        return UpdateCheckResult(
            allowed=False,
            needs_update=False,
            current_version=current_version,
            latest_version=latest,
            download_url=download_url,
            message=_disabled_message(data, current_version),
            package_key=key,
            download_page_url=download_page_url,
            disabled=True,
            announcement=announcement,
        )

    if migration:
        target_version = str(migration.get("target_version") or latest)
        return UpdateCheckResult(
            allowed=False,
            needs_update=True,
            current_version=current_version,
            latest_version=target_version,
            download_url=download_url,
            message=str(migration.get("message") or "天翼云管理分包将自动升级为多云管理总包拆分版。"),
            package_key=key,
            download_page_url=download_page_url,
            announcement=announcement,
            auto_migration_available=True,
            migration=migration,
        )

    if _needs_update(data, current_version):
        blocked_version, interrupted_retry = _prepare_interrupted_update_retry(key, "full")
        if blocked_version and not _version_lt(blocked_version, latest):
            return UpdateCheckResult(
                allowed=False,
                needs_update=True,
                current_version=current_version,
                latest_version=latest,
                download_url=download_url,
                message=UPDATE_FAILED_CONTACT_MESSAGE,
                package_key=key,
                download_page_url=download_page_url,
                announcement=announcement,
                update_retry_blocked=True,
            )
        full_package = _full_package_from_manifest(data, key)
        if full_package:
            return UpdateCheckResult(
                allowed=False,
                needs_update=True,
                current_version=current_version,
                latest_version=latest,
                download_url=download_url,
                message=(
                    f"检测到上次完整版本更新被中断，将自动重新下载版本 {latest} 一次。"
                    if interrupted_retry
                    else f"检测到完整版本 {latest}，即将自动更新。"
                ),
                auto_update_available=True,
                restart_required=True,
                package_key=key,
                download_page_url=download_page_url,
                announcement=announcement,
                full_package=full_package,
            )
        return UpdateCheckResult(
            allowed=False,
            needs_update=True,
            current_version=current_version,
            latest_version=latest,
            download_url=download_url,
            message="完整包更新配置异常，请联系客户经理处理。",
            package_key=key,
            download_page_url=download_page_url,
            announcement=announcement,
            update_retry_blocked=True,
        )

    patches = _patches_for_current_version(current_version, key)
    if patches:
        blocked_version, interrupted_retry = _prepare_interrupted_update_retry(key, "patch")
        if blocked_version:
            patches = [
                patch
                for patch in patches
                if _version_lt(blocked_version, str(patch.get("version") or "").strip())
            ]
            if not patches:
                return UpdateCheckResult(
                    allowed=False,
                    needs_update=True,
                    current_version=current_version,
                    latest_version=latest,
                    download_url=download_url,
                    message=UPDATE_FAILED_CONTACT_MESSAGE,
                    package_key=key,
                    download_page_url=download_page_url,
                    announcement=announcement,
                    update_retry_blocked=True,
                )
        last_patch = str(patches[-1].get("version") or "")
        return UpdateCheckResult(
            allowed=False,
            needs_update=True,
            current_version=current_version,
            latest_version=latest,
            download_url=download_url,
            message=(
                f"检测到上次更新被中断，将自动重新下载功能优化更新 {last_patch} 一次。"
                if interrupted_retry
                else f"检测到功能优化更新 {last_patch}，即将自动更新。"
            ),
            auto_update_available=True,
            patches=patches,
            restart_required=any(bool(patch.get("restart_required", True)) for patch in patches),
            package_key=key,
            download_page_url=download_page_url,
            announcement=announcement,
        )

    return UpdateCheckResult(
        allowed=True,
        needs_update=False,
        current_version=current_version,
        latest_version=latest,
        download_url=download_url,
        message="版本检测完成：最新版，欢迎使用!",
        package_key=key,
        download_page_url=download_page_url,
        announcement=announcement,
    )


def _format_download_size(size: int) -> str:
    if size >= 1024 * 1024:
        return f"{size / 1024 / 1024:.1f}MB"
    if size >= 1024:
        return f"{size / 1024:.1f}KB"
    return f"{size}B"


def _download_file(
    url: str,
    target: Path,
    progress: ProgressCallback | None = None,
    *,
    timeout_seconds: int = UPDATE_DOWNLOAD_TIMEOUT_SECONDS,
    progress_label: str = "更新包",
    progress_stage_lines: tuple[str, str] = (
        "2/3 下载完成后校验文件",
        "3/3 校验通过后自动安装",
    ),
) -> None:
    target.parent.mkdir(parents=True, exist_ok=True)
    request = urllib.request.Request(
        url,
        headers={
            "Cache-Control": "no-cache",
            "User-Agent": "WawaCloudDesktopUpdater/1.0",
        },
    )
    def report_download(downloaded: int, total: int | None, done: bool = False) -> None:
        if not progress:
            return
        if total and total > 0:
            percent = min(100, int(downloaded * 100 / total))
            progress(
                f"1/3 正在下载{progress_label}：{percent}%"
                f"（{_format_download_size(downloaded)} / {_format_download_size(total)}）\n"
                f"{progress_stage_lines[0]}\n"
                f"{progress_stage_lines[1]}"
            )
            return
        status = "下载完成" if done else "正在下载"
        progress(
            f"1/3 {status}{progress_label}：{_format_download_size(downloaded)}\n"
            f"{progress_stage_lines[0]}\n"
            f"{progress_stage_lines[1]}"
        )

    with urlopen_with_fallback(request, timeout=timeout_seconds) as response:
        raw_total = response.headers.get("Content-Length", "")
        try:
            total_size = int(raw_total) if raw_total else 0
        except (TypeError, ValueError):
            total_size = 0
        downloaded_size = 0
        last_reported_percent = -1
        last_reported_mb = -1
        with target.open("wb") as out:
            while True:
                chunk = response.read(1024 * 256)
                if not chunk:
                    break
                out.write(chunk)
                downloaded_size += len(chunk)
                if total_size > 0:
                    percent = min(100, int(downloaded_size * 100 / total_size))
                    if last_reported_percent < 0 or percent == 100 or percent - last_reported_percent >= 2:
                        report_download(downloaded_size, total_size)
                        last_reported_percent = percent
                else:
                    current_mb = downloaded_size // (1024 * 1024)
                    if current_mb != last_reported_mb:
                        report_download(downloaded_size, None)
                        last_reported_mb = current_mb
    if progress:
        report_download(downloaded_size, total_size, done=True)


def _sha256(path: Path) -> str:
    hasher = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            hasher.update(chunk)
    return hasher.hexdigest()


def _installed_patch_integrity_error(
    state: dict[str, Any],
    package_key: str | None = DEFAULT_UPDATE_PACKAGE,
) -> str:
    expected = state.get("patch_file_hashes")
    if not isinstance(expected, dict) or not expected:
        return ""
    root = _app_root()
    for raw_name, raw_hash in expected.items():
        name = str(raw_name or "").strip()
        expected_hash = str(raw_hash or "").strip().lower()
        if not name or not expected_hash:
            continue
        try:
            rel = _safe_patch_path(name, package_key)
        except Exception as exc:
            return f"{name}: invalid recorded path: {exc}"
        target = root / rel
        if not target.exists():
            return f"{name}: missing"
        try:
            actual_hash = _sha256(target).lower()
        except Exception as exc:
            return f"{name}: cannot read: {exc}"
        if actual_hash != expected_hash:
            return f"{name}: hash mismatch"
    return ""


def _allowed_patch_prefixes(package_key: str | None = DEFAULT_UPDATE_PACKAGE) -> tuple[str, ...]:
    key = _normalize_package_key(package_key)
    if key == HUAWEI_UPDATE_PACKAGE:
        return HUAWEI_ALLOWED_PATCH_PREFIXES
    if key == CTYUN_UPDATE_PACKAGE:
        return CTYUN_ALLOWED_PATCH_PREFIXES
    if key == VOLCENGINE_UPDATE_PACKAGE:
        return VOLCENGINE_ALLOWED_PATCH_PREFIXES
    return ALLOWED_PATCH_PREFIXES


def _safe_patch_path(name: str, package_key: str | None = DEFAULT_UPDATE_PACKAGE) -> Path:
    normalized = name.replace("\\", "/").lstrip("/")
    path = Path(normalized)
    if not normalized or path.is_absolute() or ".." in path.parts:
        raise UpdateCheckError(f"更新路径不安全：{name}")
    if "__pycache__" in path.parts or normalized.endswith(".pyc"):
        raise UpdateCheckError(f"更新不允许包含缓存文件：{name}")
    if (
        _normalize_package_key(package_key) != HUAWEI_UPDATE_PACKAGE
        and normalized in ALLOWED_PATCH_FILES
    ):
        return path
    if not any(normalized.startswith(prefix) for prefix in _allowed_patch_prefixes(package_key)):
        raise UpdateCheckError(f"更新路径不在允许范围内：{name}")
    return path


def _apply_patch_zip(
    zip_path: Path,
    patch_version: str,
    progress: ProgressCallback | None = None,
    *,
    package_key: str | None = DEFAULT_UPDATE_PACKAGE,
) -> dict[str, str]:
    root = _app_root()
    backup_dir = _data_dir() / "backups" / f"update-{patch_version}"
    backup_dir.mkdir(parents=True, exist_ok=True)
    touched_py: list[Path] = []
    created_files: list[Path] = []
    backed_up_files: list[tuple[Path, Path]] = []
    applied_files: list[Path] = []

    try:
        with zipfile.ZipFile(zip_path) as archive:
            for info in archive.infolist():
                if info.is_dir():
                    continue
                rel = _safe_patch_path(info.filename, package_key)
                target = root / rel
                backup = backup_dir / rel
                if target.exists():
                    backup.parent.mkdir(parents=True, exist_ok=True)
                    shutil.copy2(target, backup)
                    backed_up_files.append((target, backup))
                else:
                    created_files.append(target)
                target.parent.mkdir(parents=True, exist_ok=True)
                with archive.open(info) as src, target.open("wb") as dst:
                    shutil.copyfileobj(src, dst)
                applied_files.append(rel)
                if target.suffix == ".py":
                    touched_py.append(target)
                if progress:
                    progress(f"已更新：{rel.as_posix()}")

        for py_file in touched_py:
            py_compile.compile(str(py_file), doraise=True)
        return {
            rel.as_posix(): _sha256(root / rel)
            for rel in applied_files
            if (root / rel).is_file()
        }
    except Exception:
        for target, backup in backed_up_files:
            if backup.exists():
                target.parent.mkdir(parents=True, exist_ok=True)
                shutil.copy2(backup, target)
        for target in created_files:
            if target.exists():
                target.unlink()
        raise


def _validate_full_package_zip(zip_path: Path) -> str:
    executable_name = "多云管理.exe"
    normalized_names: set[str] = set()
    total_uncompressed = 0
    with zipfile.ZipFile(zip_path) as archive:
        infos = [info for info in archive.infolist() if not info.is_dir()]
        if not infos or len(infos) > 20_000:
            raise UpdateCheckError("完整包文件数量异常")
        for info in infos:
            normalized = info.filename.replace("\\", "/")
            path = Path(normalized)
            if (
                not normalized
                or normalized.startswith("/")
                or re.match(r"^[A-Za-z]:", normalized)
                or path.is_absolute()
                or ".." in path.parts
            ):
                raise UpdateCheckError(f"完整包路径不安全：{info.filename}")
            total_uncompressed += max(0, int(info.file_size))
            if total_uncompressed > 2 * 1024 * 1024 * 1024:
                raise UpdateCheckError("完整包解压大小异常")
            normalized_names.add(normalized)
        damaged_name = archive.testzip()
        if damaged_name:
            raise UpdateCheckError(f"完整包文件损坏：{damaged_name}")

    roots = {
        name[: -len(executable_name)]
        for name in normalized_names
        if name.endswith(executable_name)
    }
    valid_roots = [
        root
        for root in roots
        if f"{root}source/launcher/app.py" in normalized_names
        and any(name.startswith(f"{root}_internal/") for name in normalized_names)
    ]
    if len(valid_roots) != 1:
        raise UpdateCheckError("完整包目录结构不正确")
    return valid_roots[0].rstrip("/")


def _write_full_update_script(script_path: Path) -> None:
    script = r'''
param(
    [Parameter(Mandatory = $true)]
    [string]$TargetDir,
    [Parameter(Mandatory = $true)]
    [string]$PayloadZip,
    [Parameter(Mandatory = $true)]
    [string]$ExpectedSha256,
    [Parameter(Mandatory = $true)]
    [string]$CurrentExeName,
    [Parameter(Mandatory = $true)]
    [string]$TargetExeName,
    [Parameter(Mandatory = $true)]
    [string]$TargetVersion,
    [Parameter(Mandatory = $true)]
    [string]$HealthMarker,
    [int]$WaitPid = 0
)

$ErrorActionPreference = "Stop"
Set-StrictMode -Version 2

function Get-Sha256 {
    param([string]$Path)
    return (Get-FileHash -Algorithm SHA256 -LiteralPath $Path).Hash.ToLower()
}

function Copy-ProgramTree {
    param([string]$SourceDir, [string]$DestinationDir)
    New-Item -ItemType Directory -Force -Path $DestinationDir | Out-Null
    Get-ChildItem -LiteralPath $SourceDir -Force | Where-Object { $_.Name -ne "data" } | ForEach-Object {
        $destination = Join-Path $DestinationDir $_.Name
        if ($_.PSIsContainer) {
            Copy-Item -LiteralPath $_.FullName -Destination $destination -Recurse -Force
        } else {
            Copy-Item -LiteralPath $_.FullName -Destination $destination -Force
        }
    }
}

function Remove-ProgramTree {
    param([string]$RootDir)
    Get-ChildItem -LiteralPath $RootDir -Force | Where-Object { $_.Name -ne "data" } | ForEach-Object {
        Remove-Item -LiteralPath $_.FullName -Recurse -Force
    }
}

function Write-UpdateLog {
    param([string]$LogPath, [string]$Result, [string]$Detail)
    @(
        "time=$(Get-Date -Format 'yyyy-MM-dd HH:mm:ss')",
        "version=$TargetVersion",
        "result=$Result",
        "detail=$Detail"
    ) | Set-Content -LiteralPath $LogPath -Encoding UTF8
}

if ([IO.Path]::GetFileName($CurrentExeName) -ne $CurrentExeName -or [IO.Path]::GetFileName($TargetExeName) -ne $TargetExeName) {
    throw "Executable name is invalid."
}

$targetRoot = [IO.Path]::GetFullPath($TargetDir).TrimEnd("\")
$payloadFullPath = [IO.Path]::GetFullPath($PayloadZip)
$dataDir = [IO.Path]::GetFullPath((Join-Path $targetRoot "data"))
$healthMarkerFullPath = [IO.Path]::GetFullPath($HealthMarker)
$dataPrefix = $dataDir.TrimEnd("\") + "\"
if (-not $healthMarkerFullPath.StartsWith($dataPrefix, [StringComparison]::OrdinalIgnoreCase)) {
    throw "Health marker is outside the data directory."
}
if (-not (Test-Path -LiteralPath $targetRoot -PathType Container)) {
    throw "Target directory not found."
}
if (-not (Test-Path -LiteralPath (Join-Path $targetRoot $CurrentExeName) -PathType Leaf)) {
    throw "Current executable not found."
}
if (-not (Test-Path -LiteralPath $payloadFullPath -PathType Leaf)) {
    throw "Payload zip not found."
}
if ((Get-Sha256 -Path $payloadFullPath) -ne $ExpectedSha256.ToLower()) {
    throw "Payload SHA256 mismatch."
}

$safeVersion = ($TargetVersion -replace '[^0-9A-Za-z._-]+', '_')
$stamp = Get-Date -Format "yyyyMMdd-HHmmss"
$workRoot = Join-Path $env:TEMP ("yuntools-full-update-" + [guid]::NewGuid().ToString("N"))
$backupRoot = Join-Path $dataDir "full-update-backups"
$backupDir = Join-Path $backupRoot ($safeVersion + "-" + $stamp + "-" + [guid]::NewGuid().ToString("N"))
$logDir = Join-Path $dataDir "full-update-logs"
$logPath = Join-Path $logDir ("full-update-" + $safeVersion + "-" + $stamp + ".log")
New-Item -ItemType Directory -Force -Path $workRoot | Out-Null
New-Item -ItemType Directory -Force -Path $backupRoot | Out-Null
New-Item -ItemType Directory -Force -Path $logDir | Out-Null

$backupReady = $false
$replacementStarted = $false
$newProcess = $null
$success = $false

try {
    if ($WaitPid -gt 0) {
        try {
            Wait-Process -Id $WaitPid -Timeout 90 -ErrorAction SilentlyContinue
        } catch {}
        Start-Sleep -Seconds 2
        if (Get-Process -Id $WaitPid -ErrorAction SilentlyContinue) {
            throw "Current application did not exit."
        }
    }

    Add-Type -AssemblyName System.IO.Compression.FileSystem
    [System.IO.Compression.ZipFile]::ExtractToDirectory($payloadFullPath, $workRoot)
    $payloadRoot = Get-ChildItem -LiteralPath $workRoot -Directory -ErrorAction SilentlyContinue |
        Where-Object { Test-Path -LiteralPath (Join-Path $_.FullName $TargetExeName) -PathType Leaf } |
        Select-Object -First 1
    if ($payloadRoot) {
        $payloadPath = $payloadRoot.FullName
    } elseif (Test-Path -LiteralPath (Join-Path $workRoot $TargetExeName) -PathType Leaf) {
        $payloadPath = $workRoot
    } else {
        throw "Payload format is invalid."
    }
    if (-not (Test-Path -LiteralPath (Join-Path $payloadPath "source\launcher\app.py") -PathType Leaf)) {
        throw "Payload launcher is missing."
    }
    if (-not (Test-Path -LiteralPath (Join-Path $payloadPath "_internal") -PathType Container)) {
        throw "Payload runtime is missing."
    }

    Copy-ProgramTree -SourceDir $targetRoot -DestinationDir $backupDir
    if (-not (Test-Path -LiteralPath (Join-Path $backupDir $CurrentExeName) -PathType Leaf)) {
        throw "Backup verification failed."
    }
    $backupReady = $true

    $replacementStarted = $true
    Remove-ProgramTree -RootDir $targetRoot
    Copy-ProgramTree -SourceDir $payloadPath -DestinationDir $targetRoot
    $newExe = Join-Path $targetRoot $TargetExeName
    if (-not (Test-Path -LiteralPath $newExe -PathType Leaf)) {
        throw "Updated executable not found."
    }

    Remove-Item -LiteralPath $healthMarkerFullPath -Force -ErrorAction SilentlyContinue
    $env:WAWA_FULL_UPDATE_HEALTH_MARKER = $healthMarkerFullPath
    $newProcess = Start-Process -FilePath $newExe -WorkingDirectory $targetRoot -PassThru
    $deadline = [DateTime]::UtcNow.AddSeconds(60)
    $healthy = $false
    while ([DateTime]::UtcNow -lt $deadline) {
        if (Test-Path -LiteralPath $healthMarkerFullPath -PathType Leaf) {
            $healthy = $true
            break
        }
        if ($newProcess.HasExited) {
            break
        }
        Start-Sleep -Milliseconds 500
    }
    if (-not $healthy) {
        throw "Updated application health check failed."
    }

    $success = $true
    Write-UpdateLog -LogPath $logPath -Result "ok" -Detail "updated application opened successfully"
    Remove-Item -LiteralPath $healthMarkerFullPath -Force -ErrorAction SilentlyContinue
    Remove-Item -LiteralPath $backupDir -Recurse -Force -ErrorAction SilentlyContinue
}
catch {
    $failure = $_.Exception.Message
    if ($newProcess -and -not $newProcess.HasExited) {
        Stop-Process -Id $newProcess.Id -Force -ErrorAction SilentlyContinue
        Start-Sleep -Seconds 1
    }
    Remove-Item -LiteralPath $healthMarkerFullPath -Force -ErrorAction SilentlyContinue
    $rollbackError = ""
    if ($backupReady -and $replacementStarted) {
        try {
            Remove-ProgramTree -RootDir $targetRoot
            Copy-ProgramTree -SourceDir $backupDir -DestinationDir $targetRoot
        } catch {
            $rollbackError = $_.Exception.Message
        }
    }
    $detail = $failure
    if ($rollbackError) {
        $detail = $detail + "; rollback=" + $rollbackError
    }
    Write-UpdateLog -LogPath $logPath -Result "failed" -Detail $detail
    Remove-Item -LiteralPath Env:WAWA_FULL_UPDATE_HEALTH_MARKER -ErrorAction SilentlyContinue
    $fallbackExe = Join-Path $targetRoot $CurrentExeName
    $oldProcessStillRunning = $WaitPid -gt 0 -and [bool](Get-Process -Id $WaitPid -ErrorAction SilentlyContinue)
    if (-not $oldProcessStillRunning -and (Test-Path -LiteralPath $fallbackExe -PathType Leaf)) {
        Start-Process -FilePath $fallbackExe -WorkingDirectory $targetRoot
    }
    exit 1
}
finally {
    Remove-Item -LiteralPath Env:WAWA_FULL_UPDATE_HEALTH_MARKER -ErrorAction SilentlyContinue
    if (Test-Path -LiteralPath $workRoot) {
        Remove-Item -LiteralPath $workRoot -Recurse -Force -ErrorAction SilentlyContinue
    }
    if (-not $success -and -not $backupReady -and (Test-Path -LiteralPath $backupDir)) {
        Remove-Item -LiteralPath $backupDir -Recurse -Force -ErrorAction SilentlyContinue
    }
}
'''
    script_path.parent.mkdir(parents=True, exist_ok=True)
    script_path.write_text(script, encoding="utf-8-sig")


def _perform_full_package_update(result: UpdateCheckResult, progress: ProgressCallback | None = None) -> AutoUpdateResult:
    package_key = _normalize_package_key(result.package_key)
    package = result.full_package or {}
    target_version = str(package.get("version") or result.latest_version or "").strip()
    try:
        if package_key != DEFAULT_UPDATE_PACKAGE:
            raise UpdateCheckError("完整包自动更新仅支持多云管理总包")
        if not getattr(sys, "frozen", False):
            raise UpdateCheckError("完整包自动更新仅支持打包后的软件")
        url = str(package.get("url") or "").strip()
        expected_hash = str(package.get("sha256") or "").strip().lower()
        if not target_version or not url or not re.fullmatch(r"[0-9a-f]{64}", expected_hash):
            raise UpdateCheckError("完整包更新配置不完整")

        root = _app_root().resolve()
        executable = Path(sys.executable).resolve()
        if executable.parent != root or not executable.is_file():
            raise UpdateCheckError("软件运行目录与更新目录不一致")

        downloads_dir = _data_dir() / "downloads" / package_key
        downloads_dir.mkdir(parents=True, exist_ok=True)
        safe_version = re.sub(r"[^0-9A-Za-z._-]+", "_", target_version) or "latest"
        payload_path = downloads_dir / f"full-{safe_version}.zip"
        if payload_path.exists() and _sha256(payload_path).lower() == expected_hash:
            if progress:
                progress(
                    "1/3 已找到完整包下载文件\n"
                    "2/3 正在校验完整包\n"
                    "3/3 校验通过后自动安装"
                )
        else:
            if progress:
                progress(
                    f"1/3 正在下载完整版本 {target_version}：准备中\n"
                    "2/3 下载完成后校验完整包\n"
                    "3/3 校验通过后自动安装"
                )
            _download_file(
                url,
                payload_path,
                progress,
                timeout_seconds=FULL_UPDATE_DOWNLOAD_TIMEOUT_SECONDS,
                progress_label=f"完整版本 {target_version}",
                progress_stage_lines=(
                    "2/3 下载完成后校验完整包",
                    "3/3 校验通过后自动安装",
                ),
            )
        if _sha256(payload_path).lower() != expected_hash:
            raise UpdateCheckError("完整包校验失败")
        expected_size = int(package.get("size") or 0)
        if expected_size > 0 and payload_path.stat().st_size != expected_size:
            raise UpdateCheckError("完整包大小校验失败")
        _validate_full_package_zip(payload_path)

        if progress:
            progress(
                f"1/3 完整版本 {target_version} 已下载\n"
                "2/3 完整包校验通过\n"
                "3/3 正在启动安全安装程序..."
            )
        script_path = downloads_dir / f"run-full-update-{safe_version}.ps1"
        _write_full_update_script(script_path)
        health_token = os.urandom(16).hex()
        health_marker = downloads_dir / f"full-update-health-{health_token}.ok"
        _set_pending_update(package_key, "full", target_version)
        command = [
            "powershell.exe",
            "-NoProfile",
            "-ExecutionPolicy",
            "Bypass",
            "-File",
            str(script_path),
            "-TargetDir",
            str(root),
            "-PayloadZip",
            str(payload_path),
            "-ExpectedSha256",
            expected_hash,
            "-CurrentExeName",
            executable.name,
            "-TargetExeName",
            "多云管理.exe",
            "-TargetVersion",
            target_version,
            "-HealthMarker",
            str(health_marker),
            "-WaitPid",
            str(os.getpid()),
        ]
        creationflags = subprocess.CREATE_NO_WINDOW if hasattr(subprocess, "CREATE_NO_WINDOW") else 0
        subprocess.Popen(command, cwd=str(root), close_fds=True, creationflags=creationflags)
        return AutoUpdateResult(
            True,
            "完整包已准备完成，软件将自动关闭、安装并重新打开。",
            restart_required=True,
            external_restart=True,
        )
    except Exception as exc:
        if target_version:
            _record_update_failure(package_key, "full", target_version, exc)
        return AutoUpdateResult(
            False,
            UPDATE_FAILED_CONTACT_MESSAGE,
            error=str(exc),
            retry_blocked=True,
        )


def perform_auto_update(result: UpdateCheckResult, progress: ProgressCallback | None = None) -> AutoUpdateResult:
    if result.auto_update_available and result.full_package:
        return _perform_full_package_update(result, progress)
    if not result.auto_update_available or not result.patches:
        return AutoUpdateResult(False, UPDATE_FAILED_CONTACT_MESSAGE, error="no_update_payload", retry_blocked=True)

    package_key = _normalize_package_key(result.package_key)
    downloads_dir = _data_dir() / "downloads" / package_key
    state = _read_update_state(package_key)
    state["base_version"] = result.current_version
    state["package_key"] = package_key
    target_version = str(result.patches[-1].get("version") or "").strip()
    _mark_pending_update(state, "patch", target_version)
    try:
        _write_update_state(state, package_key)
        for patch in result.patches:
            version = str(patch.get("version") or "").strip()
            url = str(patch.get("url") or patch.get("patch_url") or "").strip()
            expected_hash = str(patch.get("sha256") or "").strip().lower()
            if not version or not url:
                raise UpdateCheckError("更新清单缺少版本号或下载地址")
            if progress:
                progress(
                    f"1/3 正在下载功能优化更新 {version}：准备中\n"
                    "2/3 下载完成后校验更新\n"
                    "3/3 校验通过后自动安装更新"
                )
            target = downloads_dir / f"patch-{version}.zip"
            _download_file(
                url,
                target,
                progress,
                progress_label=f"功能优化更新 {version}",
                progress_stage_lines=(
                    "2/3 下载完成后校验更新",
                    "3/3 校验通过后自动安装更新",
                ),
            )
            if expected_hash:
                actual_hash = _sha256(target)
                if actual_hash.lower() != expected_hash:
                    raise UpdateCheckError(f"更新校验失败：{version}")
            if progress:
                progress(
                    f"1/3 功能优化更新 {version} 已下载\n"
                    "2/3 更新校验通过\n"
                    "3/3 正在安装更新..."
                )
            patch_file_hashes = _apply_patch_zip(target, version, progress, package_key=package_key)
            state["patch_version"] = version
            state["patch_file_hashes"] = patch_file_hashes
            state["patch_integrity_enabled"] = True
            _write_update_state(state, package_key)
        state.pop("pending_update", None)
        state.pop("failed_update", None)
        state.pop("interrupted_retry", None)
        _write_update_state(state, package_key)
        success_message = "自动更新完成，请重新打开软件。" if result.restart_required else "更新完毕，欢迎使用。"
        return AutoUpdateResult(True, success_message, restart_required=result.restart_required)
    except Exception as exc:
        _record_update_failure(package_key, "patch", target_version, exc)
        return AutoUpdateResult(
            False,
            UPDATE_FAILED_CONTACT_MESSAGE,
            error=str(exc),
            retry_blocked=True,
        )


def _write_package_migration_script(script_path: Path) -> None:
    script = r'''
param(
    [Parameter(Mandatory = $true)]
    [string]$TargetDir,
    [Parameter(Mandatory = $true)]
    [string]$PayloadZip,
    [string]$ExpectedSha256 = "",
    [Parameter(Mandatory = $true)]
    [string]$CurrentExeName,
    [int]$WaitPid = 0
)

$ErrorActionPreference = "Stop"

function U {
    param([int[]]$CodePoints)
    return -join ($CodePoints | ForEach-Object { [char]$_ })
}

function Get-Sha256 {
    param([string]$Path)
    return (Get-FileHash -Algorithm SHA256 -LiteralPath $Path).Hash.ToLower()
}

function Copy-Tree {
    param([string]$SourceDir, [string]$TargetDir)
    $sourceRoot = (Get-Item -LiteralPath $SourceDir).FullName.TrimEnd("\") + "\"
    Get-ChildItem -LiteralPath $SourceDir -Force -Recurse | ForEach-Object {
        $rel = $_.FullName.Substring($sourceRoot.Length)
        if ($_.PSIsContainer) {
            New-Item -ItemType Directory -Force -Path (Join-Path $TargetDir $rel) | Out-Null
            return
        }
        $dest = Join-Path $TargetDir $rel
        New-Item -ItemType Directory -Force -Path (Split-Path -Parent $dest) | Out-Null
        Copy-Item -LiteralPath $_.FullName -Destination $dest -Force
    }
}

function Copy-MergeTree {
    param([string]$SourceDir, [string]$TargetDir)
    $sourceRoot = (Get-Item -LiteralPath $SourceDir).FullName.TrimEnd("\") + "\"
    Get-ChildItem -LiteralPath $SourceDir -Force -Recurse | ForEach-Object {
        $rel = $_.FullName.Substring($sourceRoot.Length)
        $dest = Join-Path $TargetDir $rel
        if ($_.PSIsContainer) {
            New-Item -ItemType Directory -Force -Path $dest | Out-Null
            return
        }
        New-Item -ItemType Directory -Force -Path (Split-Path -Parent $dest) | Out-Null
        $isExistingData = $rel.StartsWith("data\", [StringComparison]::OrdinalIgnoreCase) -and
            (Test-Path -LiteralPath $dest -PathType Leaf)
        if (-not $isExistingData) {
            Copy-Item -LiteralPath $_.FullName -Destination $dest -Force
        }
    }
}

$multiName = U @(0x591a, 0x4e91, 0x7ba1, 0x7406)
$huaweiName = U @(0x534e, 0x4e3a, 0x4e91, 0x7ba1, 0x7406)
$ctyunName = U @(0x5929, 0x7ffc, 0x4e91, 0x7ba1, 0x7406)
$volcengineName = U @(0x706b, 0x5c71, 0x5f15, 0x64ce, 0x7ba1, 0x7406)
$multiExeName = "$multiName.exe"
$oldExeNames = @("$huaweiName.exe", "$ctyunName.exe", "$volcengineName.exe")

$targetRoot = [IO.Path]::GetFullPath($TargetDir).TrimEnd("\")
$targetDriveRoot = [IO.Path]::GetPathRoot($targetRoot).TrimEnd("\")
if (-not $targetRoot -or $targetRoot -eq $targetDriveRoot) {
    throw "Target directory is unsafe."
}
if ([IO.Path]::GetFileName($CurrentExeName) -ne $CurrentExeName -or $oldExeNames -notcontains $CurrentExeName) {
    throw "Current executable name is invalid."
}

if ($WaitPid -gt 0) {
    try {
        Wait-Process -Id $WaitPid -Timeout 90 -ErrorAction SilentlyContinue
    } catch {}
    Start-Sleep -Seconds 2
    if (Get-Process -Id $WaitPid -ErrorAction SilentlyContinue) {
        throw "Current application did not exit."
    }
}

if (-not (Test-Path -LiteralPath $targetRoot -PathType Container)) {
    throw "Target directory not found: $targetRoot"
}
if (-not (Test-Path -LiteralPath $PayloadZip -PathType Leaf)) {
    throw "Payload zip not found: $PayloadZip"
}
if ($ExpectedSha256) {
    $actual = Get-Sha256 -Path $PayloadZip
    if ($actual -ne $ExpectedSha256.ToLower()) {
        throw "Payload SHA256 mismatch."
    }
}

Add-Type -AssemblyName System.IO.Compression.FileSystem
$stamp = Get-Date -Format "yyyyMMdd-HHmmss"
$workRoot = Join-Path $env:TEMP ("yuntools-package-migration-" + [guid]::NewGuid().ToString("N"))
$backupRoot = Join-Path (Split-Path -Parent $targetRoot) "_yuntools_upgrade_backups"
$backupDir = Join-Path $backupRoot ("subpackage-to-desktop-" + $stamp)
$logDir = Join-Path $targetRoot "_upgrade_logs"
New-Item -ItemType Directory -Force -Path $backupRoot | Out-Null
New-Item -ItemType Directory -Force -Path $logDir | Out-Null
New-Item -ItemType Directory -Force -Path $workRoot | Out-Null

$backupReady = $false
$mergeStarted = $false
$newProcess = $null

try {
    [System.IO.Compression.ZipFile]::ExtractToDirectory($PayloadZip, $workRoot)
    $payloadRoot = Get-ChildItem -LiteralPath $workRoot -Directory -ErrorAction SilentlyContinue |
        Where-Object { Test-Path -LiteralPath (Join-Path $_.FullName $multiExeName) -PathType Leaf } |
        Select-Object -First 1
    if ($payloadRoot) {
        $payloadPath = $payloadRoot.FullName
    } elseif (Test-Path -LiteralPath (Join-Path $workRoot $multiExeName) -PathType Leaf) {
        $payloadPath = $workRoot
    } else {
        throw "Payload format is invalid."
    }
    if (-not (Test-Path -LiteralPath (Join-Path $payloadPath "source\launcher\app.py") -PathType Leaf)) {
        throw "Payload launcher is missing."
    }
    if (-not (Test-Path -LiteralPath (Join-Path $payloadPath "_internal") -PathType Container)) {
        throw "Payload runtime is missing."
    }

    Copy-Tree -SourceDir $targetRoot -TargetDir $backupDir
    $backupReady = $true

    $existingOldExeNames = @()
    foreach ($oldExeName in $oldExeNames) {
        $oldExe = Join-Path $targetRoot $oldExeName
        if (Test-Path -LiteralPath $oldExe -PathType Leaf) {
            Copy-Item -LiteralPath $oldExe -Destination (Join-Path $targetRoot ($oldExeName + ".old-" + $stamp)) -Force
            $existingOldExeNames += $oldExeName
        }
    }

    $mergeStarted = $true
    Copy-MergeTree -SourceDir $payloadPath -TargetDir $targetRoot

    $newExe = Join-Path $targetRoot $multiExeName
    if (-not (Test-Path -LiteralPath $newExe -PathType Leaf)) {
        throw "Migration finished but multicloud exe was not found."
    }
    $newExeHash = Get-Sha256 -Path $newExe
    foreach ($oldExeName in $existingOldExeNames) {
        $oldExe = Join-Path $targetRoot $oldExeName
        Copy-Item -LiteralPath $newExe -Destination $oldExe -Force
        if ((Get-Sha256 -Path $oldExe) -ne $newExeHash) {
            throw "Legacy executable alias verification failed."
        }
    }

    @(
        "time=$(Get-Date -Format 'yyyy-MM-dd HH:mm:ss')",
        "target=$targetRoot",
        "backup=$backupDir",
        "payload=$PayloadZip",
        "legacy_aliases=$($existingOldExeNames -join ',')",
        "result=ok"
    ) | Set-Content -LiteralPath (Join-Path $logDir ("subpackage-to-desktop-" + $stamp + ".log")) -Encoding UTF8

    foreach ($envName in @(
        "WAWA_SINGLE_VENDOR",
        "WAWA_HUAWEI_STANDALONE_LIMITED",
        "WAWA_CTYUN_STANDALONE_LIMITED",
        "WAWA_VOLCENGINE_STANDALONE_LIMITED",
        "WAWA_LIMITED_VENDOR_FEATURES",
        "WAWA_ALLOW_HUAWEI_PLAIN_SOURCE",
        "WAWA_ALLOW_CTYUN_PLAIN_SOURCE",
        "WAWA_ALLOW_VOLCENGINE_PLAIN_SOURCE"
    )) {
        Remove-Item -LiteralPath ("Env:" + $envName) -ErrorAction SilentlyContinue
    }

    $newProcess = Start-Process -FilePath $newExe -WorkingDirectory $targetRoot -PassThru
    Start-Sleep -Seconds 2
    if ($newProcess.HasExited) {
        throw "Updated application exited during startup."
    }
}
catch {
    $failure = $_.Exception.Message
    if ($newProcess -and -not $newProcess.HasExited) {
        Stop-Process -Id $newProcess.Id -Force -ErrorAction SilentlyContinue
        Start-Sleep -Seconds 1
    }
    if ($backupReady -and $mergeStarted) {
        Get-ChildItem -LiteralPath $targetRoot -Force | Remove-Item -Recurse -Force
        Copy-Tree -SourceDir $backupDir -TargetDir $targetRoot
    }
    New-Item -ItemType Directory -Force -Path $logDir | Out-Null
    @(
        "time=$(Get-Date -Format 'yyyy-MM-dd HH:mm:ss')",
        "target=$targetRoot",
        "backup=$backupDir",
        "payload=$PayloadZip",
        "result=failed",
        "detail=$failure"
    ) | Set-Content -LiteralPath (Join-Path $logDir ("subpackage-to-desktop-" + $stamp + ".log")) -Encoding UTF8
    throw
}
finally {
    try {
        if (Test-Path -LiteralPath $workRoot) {
            Remove-Item -LiteralPath $workRoot -Recurse -Force
        }
    } catch {}
}
'''
    script_path.parent.mkdir(parents=True, exist_ok=True)
    script_path.write_text(script, encoding="utf-8-sig")


def perform_package_migration(result: UpdateCheckResult, progress: ProgressCallback | None = None) -> AutoUpdateResult:
    migration = result.migration or {}
    if not result.auto_migration_available or not migration:
        return AutoUpdateResult(False, "当前没有可执行的分包升级任务。", error="no_migration")

    package_key = _normalize_package_key(result.package_key)
    if package_key == DEFAULT_UPDATE_PACKAGE:
        return AutoUpdateResult(False, "当前升级任务仅支持云管理分包。", error="unsupported_package")

    url = str(migration.get("package_url") or "").strip()
    expected_hash = str(migration.get("sha256") or "").strip().lower()
    target_version = str(migration.get("target_version") or result.latest_version or "latest").strip()
    if not url or not re.fullmatch(r"[0-9a-f]{64}", expected_hash):
        return AutoUpdateResult(False, "分包升级配置不完整。", error="invalid_migration_payload")

    try:
        if not getattr(sys, "frozen", False):
            raise UpdateCheckError("分包升级仅支持打包后的软件")
        root = _app_root().resolve()
        executable = Path(sys.executable).resolve()
        if executable.parent != root or not executable.is_file():
            raise UpdateCheckError("软件运行目录与升级目录不一致")
        downloads_dir = _data_dir() / "downloads" / package_key
        downloads_dir.mkdir(parents=True, exist_ok=True)
        safe_version = re.sub(r"[^0-9A-Za-z._-]+", "_", target_version) or "latest"
        payload_path = downloads_dir / f"subpackage-to-desktop-{safe_version}.zip"
        if payload_path.exists() and expected_hash and _sha256(payload_path).lower() == expected_hash:
            if progress:
                progress(
                    "1/3 已找到本地升级文件\n"
                    "2/3 正在校验文件\n"
                    "3/3 校验通过后自动合并本地文件"
                )
        else:
            if progress:
                progress(
                    "1/3 正在下载分包升级文件：准备中\n"
                    "2/3 下载完成后校验文件\n"
                    "3/3 校验通过后自动合并本地文件"
                )
            _download_file(
                url,
                payload_path,
                progress,
                progress_label="分包升级文件",
                timeout_seconds=FULL_UPDATE_DOWNLOAD_TIMEOUT_SECONDS,
                progress_stage_lines=(
                    "2/3 下载完成后校验文件",
                    "3/3 校验通过后自动合并本地文件",
                ),
            )
            if expected_hash and _sha256(payload_path).lower() != expected_hash:
                raise UpdateCheckError("分包升级包校验失败")
        expected_size = int(migration.get("size") or 0)
        if expected_size > 0 and payload_path.stat().st_size != expected_size:
            raise UpdateCheckError("分包升级包大小校验失败")
        _validate_full_package_zip(payload_path)
        if progress:
            progress(
                "1/3 升级文件已准备完成\n"
                "2/3 文件校验通过\n"
                "3/3 正在启动本地合并程序..."
            )

        script_path = downloads_dir / f"run-subpackage-to-desktop-{safe_version}.ps1"
        _write_package_migration_script(script_path)
        command = [
            "powershell.exe",
            "-NoProfile",
            "-ExecutionPolicy",
            "Bypass",
            "-File",
            str(script_path),
            "-TargetDir",
            str(root),
            "-PayloadZip",
            str(payload_path),
            "-ExpectedSha256",
            expected_hash,
            "-CurrentExeName",
            executable.name,
            "-WaitPid",
            str(os.getpid()),
        ]
        creationflags = subprocess.CREATE_NO_WINDOW if hasattr(subprocess, "CREATE_NO_WINDOW") else 0
        subprocess.Popen(command, cwd=str(root), close_fds=True, creationflags=creationflags)
        return AutoUpdateResult(
            True,
            "升级程序已启动，软件将自动关闭并合并本地文件。升级完成后会自动打开多云管理，原快捷方式可继续使用。",
            restart_required=True,
            external_restart=True,
        )
    except Exception as exc:
        return AutoUpdateResult(False, f"分包升级启动失败：{exc}", error=str(exc))


def _show_message(
    title: str,
    message: str,
    *,
    open_download: bool = False,
    download_url: str = DOWNLOAD_PAGE_URL,
) -> None:
    root = tk.Tk()
    root.withdraw()
    try:
        root.attributes("-topmost", True)
    except tk.TclError:
        pass
    try:
        if open_download:
            if ask_app_confirm_dialog(root, title, f"{message}\n\n是否打开下载页面？"):
                webbrowser.open(download_url or DOWNLOAD_PAGE_URL)
        else:
            show_app_message_dialog(root, title, message, modal=True)
    finally:
        root.destroy()


def ensure_latest_version_or_exit(current_version: str, package_key: str | None = DEFAULT_UPDATE_PACKAGE) -> bool:
    """保留给旧入口使用；新启动窗口会直接调用 check_latest_version。"""
    result = check_latest_version(current_version, package_key)
    if result.allowed:
        return True
    if result.auto_update_available:
        update_result = perform_auto_update(result)
        if update_result.ok:
            if not update_result.external_restart:
                _show_message("自动更新完成", update_result.message)
            return False
        _show_message("自动更新失败", UPDATE_FAILED_CONTACT_MESSAGE)
        return False
    if result.auto_migration_available:
        migration_result = perform_package_migration(result)
        _show_message("分包升级", migration_result.message)
        return False
    _show_message("必须更新", UPDATE_FAILED_CONTACT_MESSAGE)
    return False
