import importlib
import importlib.util
import ctypes
import decimal
import hashlib
import json
import os
import re
import shutil
import threading
import time
import subprocess
import sys
import tkinter.font as tkfont
import urllib.error
import urllib.request
import uuid
import webbrowser
from ctypes import wintypes
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple

os.environ.setdefault("WAWA_DIALOG_ICON_FILENAME", "app_dialog_icon.ico")
os.environ.setdefault("WAWA_DIALOG_LOGO_FILENAME", "logo_app_68.png")

from common import shared as _shared
from common.credential_web_login import (
    CLOUD_CREDENTIAL_PAGE_NAMES,
    CLOUD_CREDENTIAL_PAGE_URLS,
    WEBVIEW_IMPORT_MISSING_MESSAGE,
)
from common.legacy_namespace import import_legacy_names
from common.portable_runtime import ensure_portable_desktop_shortcut
from common.runtime_paths import portable_root_dir
from common.process_utils import cleanup_stale_pyinstaller_dirs
from common.task_manager import TaskProcessManager
from common.task_runtime import cleanup_stale_task_locks
from common.update_guard import (
    DEFAULT_UPDATE_PACKAGE,
    DOWNLOAD_PAGE_URL,
    FULL_UPDATE_HEALTH_MARKER_ENV,
    UPDATE_COUNTDOWN_SECONDS,
    UPDATE_FAILED_CONTACT_MESSAGE,
    check_latest_version,
    perform_auto_update,
    perform_package_migration,
    read_saved_manager_auth_code,
    save_manager_auth_code,
    show_release_announcement_if_needed,
    urlopen_with_fallback,
)
from common.credential_parser import extract_mark
from common.credential_mark_cache import cached_mark_for_credentials, credential_history_for_cloud, delete_credential_history_record, remember_mark_for_credentials
from common.realtime_log_filter import set_manager_auth_code_for_log_filter
from common.user_messages import friendly_popup_text
from common.ui_helpers import UI_FONT_FAMILY, UI_LARGE_FONT_FAMILY, bind_canvas_mousewheel, build_manager_auth_row, show_app_message_dialog

import_legacy_names(globals(), _shared)

SINGLE_VENDOR_ENV = "WAWA_SINGLE_VENDOR"


def _acknowledge_full_update_health_check() -> None:
    raw_path = os.environ.pop(FULL_UPDATE_HEALTH_MARKER_ENV, "").strip()
    if not raw_path:
        return
    try:
        marker = Path(raw_path).resolve()
        allowed_dir = (Path(__file__).resolve().parents[2] / "data" / "downloads" / DEFAULT_UPDATE_PACKAGE).resolve()
        if marker.parent != allowed_dir or not re.fullmatch(r"full-update-health-[0-9a-f]{32}\.ok", marker.name):
            return
        marker.parent.mkdir(parents=True, exist_ok=True)
        marker.write_text("welcome-ready\n", encoding="ascii")
    except OSError:
        pass


def _normalize_single_vendor_key(value: str | None) -> str:
    normalized = str(value or "").strip().lower()
    if normalized in {"huawei", "huaweicloud", "huawei_cloud"}:
        return "huawei"
    if normalized in {"ctyun", "ct", "tianyi"}:
        return "ctyun"
    if normalized in {"volc", "volcengine", "volc_engine", "huoshan"}:
        return "volc"
    return ""


_SINGLE_VENDOR_KEY = _normalize_single_vendor_key(os.environ.get(SINGLE_VENDOR_ENV))
_huawei_app = None
_huawei_cloud_client = None
_volc_app = None
ctyun_workflow_gui = None
if _SINGLE_VENDOR_KEY in {"", "huawei"}:
    from HUAWEI import app as _huawei_app
    from HUAWEI import cloud_client as _huawei_cloud_client
if _SINGLE_VENDOR_KEY in {"", "volc"}:
    from Volcengine import app as _volc_app
if _SINGLE_VENDOR_KEY in {"", "ctyun"}:
    from CTYUN import app as ctyun_workflow_gui

_legacy_modules = [module for module in (_huawei_cloud_client, _huawei_app, _volc_app) if module is not None]
if _legacy_modules:
    import_legacy_names(globals(), *_legacy_modules)
APP_TITLE = _shared.APP_TITLE
MULTI_APP_ICON_FILENAME = "app_icon_v85.ico"
NOTICE_PANEL_WIDTH = 760
NOTICE_CARD_BG = "#ffffff"
NOTICE_META_BG = "#f9fafb"
NOTICE_CARD_PAD = 28
NOTICE_CARD_GAP = 14
NOTICE_TITLE_CONTENT_GAP = 6
NOTICE_ITEM_GAP = 5
NOTICE_MAIN_TITLE_FONT = (UI_LARGE_FONT_FAMILY, 20)
NOTICE_SUBTITLE_FONT = (UI_FONT_FAMILY, 11)
NOTICE_SECTION_TITLE_FONT = (UI_FONT_FAMILY, 12, "bold")
NOTICE_CONTENT_FONT = (UI_FONT_FAMILY, 11)
DEFAULT_RELEASE_NOTES = (
    "新增功能：软件内可直接登录各云账号并自动提取访问密钥AK/SK。",
    "新增功能：华为云、火山云余额显示（请备足服务器费用和流量费，余额不足将自动停机断网！）",
    "优化软件：优化启动检测、欢迎提示、窗口缩放显示和整体稳定性，界面信息更清晰。",
    "推荐最大化全屏使用软件，便于资源管理。",
)
STARTUP_DISCLAIMER_BODY = (
    "本软件仅用于本人或已获授权账号的云资源管理、购买搭建、资源查询、重装系统、退订/删除资源等合法合规用途。\n\n"
    "严禁用于翻墙、诈骗、攻击、滥用云资源、规避监管或其他违法违规行为。"
    "因使用者操作造成的费用、数据丢失、资源释放、业务中断或法律责任，由使用者自行承担。\n\n"
    "下载并使用本软件，表示你已阅读并同意以上内容。"
)
CUSTOMER_MANAGER_BINDING_LINKS: dict[str, dict[str, str]] = {}
CUSTOMER_MANAGER_VENDOR_NAMES = {
    "huawei": "华为云",
    "ctyun": "天翼云",
    "volc": "火山引擎",
}
CUSTOMER_MANAGER_VENDOR_ORDER = ("huawei", "ctyun", "volc")
CUSTOMER_MANAGER_VENDOR_SUFFIXES = {
    "HW": "huawei",
    "CT": "ctyun",
    "HS": "volc",
}
PUBLIC_LEGACY_MANAGER_AUTH_CODES = {"KHJLND", "KHJLWW", "KHJLTT"}
CUSTOMER_MANAGER_BINDING_CODE_ALIASES = {
    "KHJLND": "ND",
    "KHJLTT": "WT",
    "KHJLWW": "WW",
}
CREDENTIAL_HISTORY_PLACEHOLDER = "选择历史账号(点击输入搜索)"
LOCAL_WEB_LOGIN_SOURCE_PYTHON_FLAG = "local_web_login_python.flag"
CUSTOMER_MANAGER_CONTACTS: dict[str, tuple[tuple[str, tuple[str, ...]], ...]] = {}
CUSTOMER_MANAGER_ALLOWED_VENDORS: dict[str, tuple[str, ...]] = {}
CUSTOMER_MANAGER_LIMITED_FEATURE_CODES: set[str] = set()
LIMITED_VENDOR_FEATURES_ENV = "WAWA_LIMITED_VENDOR_FEATURES"
AK_SK_TUTORIAL_URLS = {
    "huawei": "https://www.wawaapi.com/YunTools/Huawei/AKSK",
    "ctyun": "https://www.wawaapi.com/YunTools/CTYUN/AKSK",
    "volc": "https://www.wawaapi.com/YunTools/Volcengine/AKSK",
}
CUSTOMER_MANAGER_AUTH_API_URL = "https://www.wawaapi.com/api/software/manager-auth"
CUSTOMER_MANAGER_AUTH_USAGE_API_URL = "https://www.wawaapi.com/api/software/manager-auth-usage"
CUSTOMER_MANAGER_AUTH_TIMEOUT_SECONDS = 8
CUSTOMER_MANAGER_AUTH_USAGE_TIMEOUT_SECONDS = 2
CUSTOMER_MANAGER_AUTH_USAGE_HEARTBEAT_INTERVAL_SECONDS = 6 * 60
CURRENT_MANAGER_AUTH_CODE = ""
RUNTIME_UPDATE_POLL_INTERVAL_MS = 18 * 60 * 1000
INVALID_MANAGER_AUTH_MESSAGE = "校验授权码失败，请联系软件获取渠道重新获取授权码。"
MANAGER_AUTH_SERVICE_UNAVAILABLE_MESSAGE = "授权码校验服务暂不可用，请检查网络后重试。"
_manager_auth_usage_sessions: set[tuple[str, str]] = set()
LOCAL_INTERNAL_MANAGER_AUTH_CODES: dict[str, dict[str, object]] = {
    "WW": {
        "links": {
            "huawei": "https://account.huaweicloud.com/obmgr/invitation1/invitation1.html?id=cadb48ee-e215-4ebb-a7af-382232de23b8#/",
            "ctyun": "https://www.ctyun.cn/h5/auth/invitation/23144053",
            "volc": "https://partner.volcengine.com/partners/auth/confirm?inviteToken=Y47ECGP6Z3SN71HHKQMXOMIQM5UBBWJXYALZI91LXQH7FYR9MQRSSIC79VFJ4122&identityType=12",
        },
        "contacts": (
            ("华为云QQ交流群", ("916383784",)),
            ("天翼云QQ交流群", ("159500025",)),
            ("火山引擎QQ交流群", ("1051985214",)),
            ("微信", ("Lywwykf", "lywwyjs")),
            ("QQ", ("2697603393", "2855784767")),
        ),
    },
    "WT": {
        "links": {
            "huawei": "https://account.huaweicloud.com/obmgr/invitation/invitation.html?id=75d03aff-b5b4-4218-ac47-6381a46a2550",
            "ctyun": "https://www.ctyun.cn/h5/auth/invitation/23144053?staffUserId=23532720",
            "volc": "https://partner.volcengine.com/partner/invite/RCQQ4IQLANU",
        },
        "contacts": (
            ("微信", ("Wucloud8",)),
            ("QQ", ("1980299167",)),
        ),
    },
    "ND": {
        "links": {
            "huawei": "https://account.huaweicloud.com/obmgr/invitation/invitation.html?id=6d612c30-f205-4078-86e2-2182130a8917",
            "ctyun": "https://www.ctyun.cn/h5/auth/invitation/23144053?staffUserId=23698360",
            "volc": "https://partner.volcengine.com/partners/auth/confirm?inviteToken=Y47ECGP6Z3SN71HHKQMXOMIQM5UBBWJXYALZI91LXQH7FYR9MQRSSIC79VFJ4122&partnerType=101&partnerName=%E6%B4%9B%E9%98%B3%E5%96%B5%E5%91%9C%E4%BA%91%E8%AE%A1%E7%AE%97%E6%9C%89%E9%99%90%E5%85%AC%E5%8F%B8&identityType=12&PartnerEmployeeId=62763498",
        },
        "contacts": (
            ("微信", ("L613857857", "Wuciloud9")),
            ("QQ", ("191216996", "78602932")),
        ),
    },
}


def normalize_manager_auth_code(code: str) -> str:
    return re.sub(r"\s+", "", str(code or "")).upper()


def _manager_auth_code_vendor_scope(normalized_code: str) -> tuple[str, tuple[str, ...], bool]:
    """Return the base authorization code and cloud scope encoded by HW/HS/CT suffixes."""
    value = normalize_manager_auth_code(normalized_code)
    suffix_vendors: set[str] = set()
    while len(value) >= 2:
        token = value[-2:]
        vendor = CUSTOMER_MANAGER_VENDOR_SUFFIXES.get(token)
        if not vendor:
            break
        suffix_vendors.add(vendor)
        value = value[:-2]
    if not suffix_vendors:
        return normalize_manager_auth_code(normalized_code), (), False
    ordered_vendors = tuple(vendor for vendor in CUSTOMER_MANAGER_VENDOR_ORDER if vendor in suffix_vendors)
    return value, ordered_vendors, True


def _manager_auth_public_binding_links(code: str) -> dict[str, str]:
    base_code, _vendors, _scoped = _manager_auth_code_vendor_scope(code)
    source_code = CUSTOMER_MANAGER_BINDING_CODE_ALIASES.get(base_code, base_code)
    record = LOCAL_INTERNAL_MANAGER_AUTH_CODES.get(source_code)
    raw_links = record.get("links") if isinstance(record, dict) else None
    if not isinstance(raw_links, dict):
        return {}
    return {
        vendor: str(raw_links.get(vendor) or "").strip()
        for vendor in CUSTOMER_MANAGER_VENDOR_ORDER
        if str(raw_links.get(vendor) or "").strip().startswith("https://")
    }


def _remote_allowed_vendors(value: object) -> tuple[str, ...]:
    if not isinstance(value, list):
        return ()
    aliases = {
        "huawei": "huawei",
        "huaweicloud": "huawei",
        "hw": "huawei",
        "ctyun": "ctyun",
        "ct": "ctyun",
        "tianyi": "ctyun",
        "volc": "volc",
        "volcengine": "volc",
        "hs": "volc",
        "huoshan": "volc",
    }
    vendors: set[str] = set()
    for item in value:
        vendor = aliases.get(str(item or "").strip().lower())
        if vendor:
            vendors.add(vendor)
    return tuple(vendor for vendor in CUSTOMER_MANAGER_VENDOR_ORDER if vendor in vendors)


def _remember_manager_auth_payload(
    normalized_code: str,
    links: dict[str, str],
    contacts: tuple[tuple[str, tuple[str, ...]], ...],
    allowed_vendors: tuple[str, ...] = (),
    *,
    limited_features: bool = False,
) -> None:
    CUSTOMER_MANAGER_BINDING_LINKS[normalized_code] = links
    CUSTOMER_MANAGER_CONTACTS[normalized_code] = contacts
    CUSTOMER_MANAGER_ALLOWED_VENDORS[normalized_code] = allowed_vendors or CUSTOMER_MANAGER_VENDOR_ORDER
    if limited_features:
        CUSTOMER_MANAGER_LIMITED_FEATURE_CODES.add(normalized_code)
    else:
        CUSTOMER_MANAGER_LIMITED_FEATURE_CODES.discard(normalized_code)


def manager_auth_allowed_vendors(code: str | None = None) -> tuple[str, ...]:
    normalized_code = normalize_manager_auth_code(code or CURRENT_MANAGER_AUTH_CODE)
    vendors = CUSTOMER_MANAGER_ALLOWED_VENDORS.get(normalized_code)
    return vendors or CUSTOMER_MANAGER_VENDOR_ORDER


def manager_auth_limited_features(code: str | None = None) -> bool:
    normalized_code = normalize_manager_auth_code(code or CURRENT_MANAGER_AUTH_CODE)
    return normalized_code in CUSTOMER_MANAGER_LIMITED_FEATURE_CODES


def _manager_auth_public_upgrade_options(normalized_code: str) -> tuple[tuple[str, str], ...]:
    if normalized_code not in PUBLIC_LEGACY_MANAGER_AUTH_CODES:
        return ()
    return (
        ("华为云", f"{normalized_code}HW"),
        ("火山引擎", f"{normalized_code}HS"),
        ("华为云+火山引擎", f"{normalized_code}HWHS"),
    )


def set_current_manager_auth_code(code: str) -> str:
    global CURRENT_MANAGER_AUTH_CODE
    normalized_code = normalize_manager_auth_code(code)
    CURRENT_MANAGER_AUTH_CODE = normalized_code
    set_manager_auth_code_for_log_filter(normalized_code)
    return normalized_code


def _authorize_local_internal_manager_auth_code(normalized_code: str) -> bool:
    record = LOCAL_INTERNAL_MANAGER_AUTH_CODES.get(normalized_code)
    if not isinstance(record, dict):
        return False
    raw_links = record.get("links")
    if not isinstance(raw_links, dict):
        return False
    links = {
        vendor: str(raw_links.get(vendor) or "").strip()
        for vendor in ("huawei", "ctyun", "volc")
    }
    if any(not url.startswith("https://") for url in links.values()):
        return False
    contacts = record.get("contacts")
    if not isinstance(contacts, tuple) or not contacts:
        return False
    _remember_manager_auth_payload(normalized_code, links, contacts, limited_features=False)  # type: ignore[arg-type]
    return True


def _remote_manager_contacts(value: object) -> tuple[tuple[str, tuple[str, ...]], ...]:
    contacts: list[tuple[str, tuple[str, ...]]] = []
    if not isinstance(value, list):
        return ()
    for item in value:
        if not isinstance(item, dict):
            continue
        label = str(item.get("label") or "").strip()
        raw_values = item.get("values")
        if not label or not isinstance(raw_values, list):
            continue
        values = tuple(str(entry or "").strip() for entry in raw_values if str(entry or "").strip())
        if values:
            contacts.append((label, values))
    return tuple(contacts)


def _fetch_manager_auth_payload(
    code: str,
    package_key: str,
    requested_code: str,
    requested_vendors: tuple[str, ...],
) -> tuple[dict[str, object] | None, str]:
    body = json.dumps(
        {
            "code": code,
            "package": str(package_key or DEFAULT_UPDATE_PACKAGE),
            "app_version": APP_VERSION,
            "requested_code": requested_code,
            "requested_vendors": list(requested_vendors),
        },
        ensure_ascii=False,
    ).encode("utf-8")
    request = urllib.request.Request(
        CUSTOMER_MANAGER_AUTH_API_URL,
        data=body,
        method="POST",
        headers={
            "Content-Type": "application/json; charset=utf-8",
            "Accept": "application/json",
            "User-Agent": "WawaCloudDesktopAuth/1.0",
        },
    )
    try:
        with urlopen_with_fallback(request, timeout=CUSTOMER_MANAGER_AUTH_TIMEOUT_SECONDS) as response:
            payload = json.loads(response.read().decode("utf-8-sig"))
    except (OSError, ValueError, urllib.error.URLError, urllib.error.HTTPError):
        return None, MANAGER_AUTH_SERVICE_UNAVAILABLE_MESSAGE
    if not isinstance(payload, dict):
        return None, MANAGER_AUTH_SERVICE_UNAVAILABLE_MESSAGE
    return payload, ""


def authorize_manager_auth_code(code: str, package_key: str = DEFAULT_UPDATE_PACKAGE) -> tuple[bool, str]:
    normalized_code = normalize_manager_auth_code(code)
    if not normalized_code:
        return False, INVALID_MANAGER_AUTH_MESSAGE
    auth_code, suffix_vendors, has_vendor_suffix = _manager_auth_code_vendor_scope(normalized_code)
    if not auth_code:
        return False, INVALID_MANAGER_AUTH_MESSAGE
    if _authorize_local_internal_manager_auth_code(auth_code):
        links = CUSTOMER_MANAGER_BINDING_LINKS.get(auth_code, {})
        contacts = CUSTOMER_MANAGER_CONTACTS.get(auth_code, ())
        _remember_manager_auth_payload(
            normalized_code,
            links,
            contacts,
            suffix_vendors or CUSTOMER_MANAGER_VENDOR_ORDER,
            limited_features=False,
        )
        return True, ""

    payload, fetch_error = _fetch_manager_auth_payload(normalized_code, package_key, normalized_code, suffix_vendors)
    if fetch_error:
        return False, fetch_error
    if (
        has_vendor_suffix
        and auth_code != normalized_code
        and isinstance(payload, dict)
        and payload.get("ok")
        and not payload.get("valid")
    ):
        fallback_payload, fallback_error = _fetch_manager_auth_payload(auth_code, package_key, normalized_code, suffix_vendors)
        if fallback_error:
            return False, fallback_error
        if isinstance(fallback_payload, dict) and fallback_payload.get("ok") and fallback_payload.get("valid"):
            payload = fallback_payload

    if not isinstance(payload, dict) or not payload.get("ok"):
        return False, str(payload.get("message") or MANAGER_AUTH_SERVICE_UNAVAILABLE_MESSAGE) if isinstance(payload, dict) else MANAGER_AUTH_SERVICE_UNAVAILABLE_MESSAGE
    if not payload.get("valid"):
        return False, str(payload.get("message") or INVALID_MANAGER_AUTH_MESSAGE)

    raw_links = payload.get("links")
    if not isinstance(raw_links, dict):
        return False, MANAGER_AUTH_SERVICE_UNAVAILABLE_MESSAGE
    links = {
        vendor: str(raw_links.get(vendor) or "").strip()
        for vendor in ("huawei", "ctyun", "volc")
    }
    if any(not url.startswith("https://") for url in links.values()):
        return False, MANAGER_AUTH_SERVICE_UNAVAILABLE_MESSAGE
    contacts = _remote_manager_contacts(payload.get("contacts"))
    if not contacts:
        return False, MANAGER_AUTH_SERVICE_UNAVAILABLE_MESSAGE

    allowed_vendors = suffix_vendors or _remote_allowed_vendors(payload.get("allowed_vendors")) or CUSTOMER_MANAGER_VENDOR_ORDER
    _remember_manager_auth_payload(normalized_code, links, contacts, allowed_vendors, limited_features=True)
    return True, ""


def _manager_auth_device_id() -> str:
    try:
        data_dir = portable_root_dir() / "data"
        data_dir.mkdir(parents=True, exist_ok=True)
    except Exception:
        data_dir = Path.home() / "Documents" / "多云管理"
        data_dir.mkdir(parents=True, exist_ok=True)
    path = data_dir / "manager_auth_device_id.txt"
    try:
        existing = re.sub(r"[^0-9a-fA-F-]", "", path.read_text(encoding="utf-8").strip())[:64]
    except Exception:
        existing = ""
    if existing:
        return existing
    device_id = uuid.uuid4().hex
    try:
        path.write_text(device_id + "\n", encoding="utf-8")
    except Exception:
        pass
    return device_id


def _post_manager_auth_usage(code: str, package_key: str, app_version: str, event: str) -> None:
    normalized_code = normalize_manager_auth_code(code)
    if not normalized_code:
        return
    body = json.dumps(
        {
            "code": normalized_code,
            "package": str(package_key or DEFAULT_UPDATE_PACKAGE),
            "app_version": str(app_version or APP_VERSION),
            "event": str(event or "auth"),
            "device_id": _manager_auth_device_id(),
        },
        ensure_ascii=False,
    ).encode("utf-8")
    request = urllib.request.Request(
        CUSTOMER_MANAGER_AUTH_USAGE_API_URL,
        data=body,
        method="POST",
        headers={
            "Content-Type": "application/json; charset=utf-8",
            "Accept": "application/json",
            "User-Agent": f"WawaCloudDesktopAuthUsage/{app_version or APP_VERSION}",
        },
    )
    try:
        with urlopen_with_fallback(request, timeout=CUSTOMER_MANAGER_AUTH_USAGE_TIMEOUT_SECONDS):
            pass
    except Exception:
        pass


def start_manager_auth_usage_reporting(code: str, package_key: str = DEFAULT_UPDATE_PACKAGE, app_version: str = APP_VERSION) -> None:
    normalized_code = normalize_manager_auth_code(code)
    package_value = str(package_key or DEFAULT_UPDATE_PACKAGE)
    if not normalized_code:
        return
    key = (normalized_code, package_value)
    if key in _manager_auth_usage_sessions:
        return
    _manager_auth_usage_sessions.add(key)

    def worker() -> None:
        _post_manager_auth_usage(normalized_code, package_value, app_version, "auth")
        while True:
            time.sleep(CUSTOMER_MANAGER_AUTH_USAGE_HEARTBEAT_INTERVAL_SECONDS)
            _post_manager_auth_usage(normalized_code, package_value, app_version, "heartbeat")

    threading.Thread(target=worker, daemon=True).start()


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


def _candidate_update_note_paths() -> list[Path]:
    paths: list[Path] = []
    try:
        paths.append(Path(resource_path("更新说明.txt")))
    except Exception:
        pass
    try:
        paths.append(portable_root_dir() / "更新说明.txt")
    except Exception:
        pass
    try:
        paths.append(Path(__file__).resolve().parents[2] / "更新说明.txt")
    except Exception:
        pass
    unique: list[Path] = []
    seen: set[str] = set()
    for path in paths:
        key = str(path)
        if key not in seen:
            seen.add(key)
            unique.append(path)
    return unique


def read_release_note_lines() -> list[str]:
    for path in _candidate_update_note_paths():
        if not path.exists():
            continue
        try:
            lines = path.read_text(encoding="utf-8-sig", errors="ignore").splitlines()
        except Exception:
            continue
        in_update_block = False
        items: list[str] = []
        for line in lines:
            trimmed = line.strip()
            if not trimmed:
                continue
            if trimmed == "更新内容：":
                in_update_block = True
                continue
            if trimmed == "使用方法：":
                break
            if not in_update_block:
                continue
            match = re.match(r"^\d+[\.\．]\s*(.+)$", trimmed)
            items.append((match.group(1) if match else trimmed).strip())
        if items:
            return list(dict.fromkeys(items))
    return list(DEFAULT_RELEASE_NOTES)


def apply_multi_window_icon(window: tk.Misc) -> None:
    icon_path = resource_path(MULTI_APP_ICON_FILENAME)
    if os.path.exists(icon_path):
        try:
            window.iconbitmap(icon_path)  # type: ignore[attr-defined]
        except tk.TclError:
            pass
    logo_path = resource_path("logo_app_68.png")
    if os.path.exists(logo_path):
        try:
            photo = tk.PhotoImage(file=logo_path)
            window.iconphoto(True, photo)  # type: ignore[attr-defined]
            setattr(window, "_multi_app_icon_photo_ref", photo)
        except tk.TclError:
            pass


def _notice_label(
    parent: tk.Misc,
    text: str,
    font: tuple,
    *,
    justify: str = tk.LEFT,
    anchor: str = "w",
    pady: tuple[int, int] = (0, 0),
    padx: tuple[int, int] = (0, 0),
    bg: str = APP_BG,
    fg: str = TEXT_FG,
    wraplength: int | None = None,
) -> tk.Label:
    label = tk.Label(
        parent,
        text=text,
        bg=bg,
        fg=fg,
        font=font,
        justify=justify,
        anchor=anchor,
        wraplength=wraplength if wraplength is not None else NOTICE_PANEL_WIDTH,
    )
    label.pack(fill=tk.X, padx=padx, pady=pady)
    return label


def _notice_card(parent: tk.Misc, title: str, *, version_text: str = "") -> tk.Frame:
    outer = tk.Frame(parent, bg=BORDER_COLOR)
    outer.pack(fill=tk.X, pady=(0, NOTICE_CARD_GAP))
    panel = tk.Frame(outer, bg=NOTICE_CARD_BG)
    panel.pack(fill=tk.X, padx=1, pady=1)
    content = tk.Frame(panel, bg=NOTICE_CARD_BG)
    content.pack(fill=tk.X, padx=NOTICE_CARD_PAD, pady=NOTICE_CARD_PAD)
    header = tk.Frame(content, bg=NOTICE_CARD_BG)
    header.pack(fill=tk.X, pady=(0, NOTICE_TITLE_CONTENT_GAP))
    _notice_label(
        header,
        title,
        NOTICE_SECTION_TITLE_FONT,
        justify=tk.CENTER,
        anchor="center",
        bg=NOTICE_CARD_BG,
        wraplength=NOTICE_PANEL_WIDTH - (NOTICE_CARD_PAD * 2),
    )
    if version_text:
        _notice_label(
            header,
            version_text,
            (UI_FONT_FAMILY, 10, "bold"),
            justify=tk.CENTER,
            anchor="center",
            pady=(4, 0),
            bg=NOTICE_CARD_BG,
            fg=MUTED_FG,
        )
    return content


def _notice_card_line(parent: tk.Misc, text: str, *, number: int | None = None) -> None:
    row = tk.Frame(parent, bg=NOTICE_CARD_BG)
    row.pack(fill=tk.X, pady=(0, NOTICE_ITEM_GAP))
    text_wrap = NOTICE_PANEL_WIDTH - (NOTICE_CARD_PAD * 2)
    if number is not None:
        tk.Label(
            row,
            text=f"{number}.",
            bg=NOTICE_CARD_BG,
            fg=TEXT_FG,
            font=NOTICE_CONTENT_FONT,
            anchor="ne",
            width=3,
        ).pack(side=tk.LEFT, anchor="n")
        text_wrap -= 34
    tk.Label(
        row,
        text=text,
        bg=NOTICE_CARD_BG,
        fg=TEXT_FG,
        font=NOTICE_CONTENT_FONT,
        justify=tk.LEFT,
        anchor="w",
        wraplength=text_wrap,
    ).pack(side=tk.LEFT, fill=tk.X, expand=True)


def _notice_logo(parent: tk.Misc, filename: str) -> None:
    logo_path = resource_path(filename)
    if not os.path.exists(logo_path):
        return
    try:
        photo = tk.PhotoImage(file=logo_path)
    except tk.TclError:
        return
    label = tk.Label(
        parent,
        image=photo,
        bg=NOTICE_CARD_BG,
        bd=0,
        highlightthickness=0,
        width=78,
        height=78,
    )
    label.pack(side=tk.LEFT, padx=(0, 16))
    setattr(parent, f"_notice_logo_{filename}", photo)


def _notice_meta(parent: tk.Misc, items: list[tuple[str, str]]) -> None:
    row = tk.Frame(parent, bg=NOTICE_CARD_BG)
    row.pack(anchor="center", pady=(14, 0))
    for index, (label_text, value_text) in enumerate(items):
        outer = tk.Frame(row, bg=BORDER_COLOR)
        outer.pack(side=tk.LEFT, padx=(0 if index == 0 else 10, 0))
        box = tk.Frame(outer, bg=NOTICE_META_BG)
        box.pack(padx=1, pady=1)
        tk.Label(box, text=f"{label_text}：", bg=NOTICE_META_BG, fg=MUTED_FG, font=(UI_FONT_FAMILY, 10)).pack(side=tk.LEFT, padx=(12, 0), pady=9)
        tk.Label(box, text=value_text, bg=NOTICE_META_BG, fg=TEXT_FG, font=(UI_FONT_FAMILY, 10, "bold")).pack(side=tk.LEFT, padx=(0, 12), pady=9)


def _notice_welcome_card(parent: tk.Misc, app_title: str, logo_filename: str) -> None:
    outer = tk.Frame(parent, bg=BORDER_COLOR)
    outer.pack(fill=tk.X, pady=(0, NOTICE_CARD_GAP))
    panel = tk.Frame(outer, bg=NOTICE_CARD_BG)
    panel.pack(fill=tk.X, padx=1, pady=1)
    content = tk.Frame(panel, bg=NOTICE_CARD_BG)
    content.pack(fill=tk.X, padx=NOTICE_CARD_PAD, pady=NOTICE_CARD_PAD)
    row = tk.Frame(content, bg=NOTICE_CARD_BG)
    row.pack(anchor="center")
    _notice_logo(row, logo_filename)
    tk.Label(
        row,
        text=f"欢迎使用 {app_title}",
        bg=NOTICE_CARD_BG,
        fg=TEXT_FG,
        font=(UI_LARGE_FONT_FAMILY, 18, "bold"),
        anchor="center",
    ).pack(side=tk.LEFT)


def _outer_window_rect(window: tk.Toplevel | tk.Tk) -> tuple[int, int, int, int] | None:
    if os.name != "nt":
        return None
    handles: list[int] = []
    try:
        frame_id = str(window.tk.call("wm", "frame", window._w))
        if frame_id:
            handles.append(int(frame_id, 0))
    except Exception:
        pass
    try:
        handles.append(int(window.winfo_id()))
    except Exception:
        pass
    for hwnd in handles:
        try:
            rect = wintypes.RECT()
            if ctypes.windll.user32.GetWindowRect(wintypes.HWND(hwnd), ctypes.byref(rect)):  # type: ignore[attr-defined]
                width = rect.right - rect.left
                height = rect.bottom - rect.top
                if width > 0 and height > 0:
                    return rect.left, rect.top, rect.right, rect.bottom
        except Exception:
            continue
    return None


def _window_work_area(window: tk.Toplevel | tk.Tk) -> tuple[int, int, int, int]:
    screen_width = window.winfo_screenwidth()
    screen_height = window.winfo_screenheight()
    if os.name == "nt":
        try:
            rect = wintypes.RECT()
            if ctypes.windll.user32.SystemParametersInfoW(0x0030, 0, ctypes.byref(rect), 0):  # type: ignore[attr-defined]
                width = rect.right - rect.left
                height = rect.bottom - rect.top
                if width > 0 and height > 0:
                    return rect.left, rect.top, width, height
        except Exception:
            pass
    return 0, 0, screen_width, screen_height


def _center_window_on_screen(window: tk.Toplevel | tk.Tk) -> None:
    window.update_idletasks()
    width = max(window.winfo_width(), window.winfo_reqwidth())
    height = max(window.winfo_height(), window.winfo_reqheight())
    screen_width = window.winfo_screenwidth()
    screen_height = window.winfo_screenheight()
    x = max(0, (screen_width - width) // 2)
    y = max(0, (screen_height - height) // 2)
    window.geometry(f"+{x}+{y}")
    window.update_idletasks()
    rect = _outer_window_rect(window)
    if not rect:
        return
    left, top, right, bottom = rect
    outer_width = right - left
    outer_height = bottom - top
    target_left = max(0, (screen_width - outer_width) // 2)
    target_top = max(0, (screen_height - outer_height) // 2)
    adjusted_x = max(0, x + target_left - left)
    adjusted_y = max(0, y + target_top - top)
    window.geometry(f"+{adjusted_x}+{adjusted_y}")


def _startup_notice_content_frame(window: tk.Misc) -> ttk.Frame:
    container = ttk.Frame(window)
    container.pack(fill=tk.BOTH, expand=True)
    container.columnconfigure(0, weight=1)
    container.rowconfigure(0, weight=1)

    canvas = tk.Canvas(container, bg=APP_BG, bd=0, highlightthickness=0)
    v_scrollbar = ttk.Scrollbar(container, orient=tk.VERTICAL, command=canvas.yview)
    h_scrollbar = ttk.Scrollbar(container, orient=tk.HORIZONTAL, command=canvas.xview)
    canvas.grid(row=0, column=0, sticky="nsew")
    v_scrollbar.grid(row=0, column=1, sticky="ns")
    h_scrollbar.grid(row=1, column=0, sticky="ew")

    def yview_changed(first: str, last: str) -> None:
        v_scrollbar.set(first, last)
        if float(first) <= 0.0 and float(last) >= 1.0:
            v_scrollbar.grid_remove()
        else:
            v_scrollbar.grid(row=0, column=1, sticky="ns")

    def xview_changed(first: str, last: str) -> None:
        h_scrollbar.set(first, last)
        if float(first) <= 0.0 and float(last) >= 1.0:
            h_scrollbar.grid_remove()
        else:
            h_scrollbar.grid(row=1, column=0, sticky="ew")

    canvas.configure(yscrollcommand=yview_changed, xscrollcommand=xview_changed)
    frame = ttk.Frame(canvas, padding=20)
    window_id = canvas.create_window((0, 0), window=frame, anchor="nw")

    def refresh_scroll_region(_event: tk.Event | None = None) -> None:
        canvas.configure(scrollregion=canvas.bbox("all"))

    def resize_content(event: tk.Event) -> None:
        target_width = max(frame.winfo_reqwidth(), NOTICE_PANEL_WIDTH + 40)
        canvas.itemconfigure(window_id, width=max(event.width, target_width))
        refresh_scroll_region()

    def on_mousewheel(event: tk.Event) -> str:
        if getattr(event, "state", 0) & 0x0001:
            canvas.xview_scroll(int(-1 * (event.delta / 120)), "units")
        else:
            canvas.yview_scroll(int(-1 * (event.delta / 120)), "units")
        return "break"

    frame.bind("<Configure>", refresh_scroll_region)
    canvas.bind("<Configure>", resize_content)
    container.bind("<Enter>", lambda _event: canvas.bind_all("<MouseWheel>", on_mousewheel))
    container.bind("<Leave>", lambda _event: canvas.unbind_all("<MouseWheel>"))
    container.bind("<Destroy>", lambda _event: canvas.unbind_all("<MouseWheel>"), add="+")
    return frame


def _fit_startup_notice_window(window: tk.Toplevel | tk.Tk, content_frame: tk.Misc | None = None) -> None:
    window.update_idletasks()
    work_x, work_y, work_width, work_height = _window_work_area(window)
    margin = 14
    max_width = max(360, work_width - margin * 2)
    max_height = max(320, work_height - margin * 2)
    content_width = content_frame.winfo_reqwidth() if content_frame is not None else 0
    content_height = content_frame.winfo_reqheight() if content_frame is not None else 0
    width = min(max(NOTICE_PANEL_WIDTH + 84, window.winfo_reqwidth(), content_width + 8), max_width)
    height = min(max(520, window.winfo_reqheight(), content_height + 8), max_height)

    for _ in range(2):
        x = work_x + max(margin, (work_width - width) // 2)
        y = work_y + max(margin, (work_height - height) // 2)
        window.geometry(f"{int(width)}x{int(height)}+{int(x)}+{int(y)}")
        window.update_idletasks()
        rect = _outer_window_rect(window)
        if not rect:
            break
        left, top, right, bottom = rect
        overflow_width = max(0, right - (work_x + work_width - margin), (work_x + margin) - left)
        overflow_height = max(0, bottom - (work_y + work_height - margin), (work_y + margin) - top)
        if overflow_width <= 0 and overflow_height <= 0:
            break
        width = max(360, width - overflow_width)
        height = max(320, height - overflow_height)

    window.minsize(min(640, int(width)), min(420, int(height)))


def pack_startup_notice(
    parent: tk.Misc,
    disclaimer_body: str,
    *,
    app_title: str = APP_TITLE,
    subtitle: str = "正式完整包，包含华为云、天翼云、火山引擎管理功能。",
    version: str = APP_VERSION,
    package_label: str = "正式完整包",
    platform_label: str = "华为云 / 天翼云 / 火山引擎",
    logo_filename: str = "logo_app_68.png",
    release_notes: list[str] | None = None,
) -> None:
    panel = tk.Frame(parent, bg=APP_BG, width=NOTICE_PANEL_WIDTH)
    panel.pack(fill=tk.X)

    notice_card = _notice_card(panel, "免责声明")
    for index, line in enumerate(_compact_notice_lines(disclaimer_body), start=1):
        _notice_card_line(notice_card, line, number=index)

    _notice_welcome_card(panel, app_title, logo_filename)

    updates_card = _notice_card(panel, "更新说明", version_text=f"版本 {version}")
    for index, line in enumerate(release_notes or read_release_note_lines(), start=1):
        _notice_card_line(updates_card, line, number=index)

class MultiCloudManagerApp(tk.Tk):
    def __init__(self, *, single_vendor: str | None = None, app_title: str | None = None) -> None:
        super().__init__()
        self.withdraw()
        self._initial_show_pending = True
        self.single_vendor = _normalize_single_vendor_key(single_vendor or os.environ.get(SINGLE_VENDOR_ENV))
        auth_vendors = tuple(
            vendor
            for vendor in manager_auth_allowed_vendors(CURRENT_MANAGER_AUTH_CODE)
            if vendor in CUSTOMER_MANAGER_VENDOR_ORDER
        )
        self.available_vendors = [self.single_vendor] if self.single_vendor else list(auth_vendors or CUSTOMER_MANAGER_VENDOR_ORDER)
        if manager_auth_limited_features(CURRENT_MANAGER_AUTH_CODE):
            os.environ[LIMITED_VENDOR_FEATURES_ENV] = ",".join(self.available_vendors)
        else:
            os.environ.pop(LIMITED_VENDOR_FEATURES_ENV, None)
        self.app_title = app_title or APP_TITLE
        self.title(self.app_title)
        self.geometry("1380x900")
        self.minsize(1024, 640)
        self.configure(bg=APP_BG)
        self.vendor_pages: Dict[str, ttk.Frame] = {}
        self.vendor_buttons: Dict[str, tk.Button] = {}
        self.vendor_button_images: Dict[str, tk.PhotoImage] = {}
        self.vendor_progressbars: Dict[str, tk.Canvas] = {}
        self.vendor_progress_active: Dict[str, bool] = {}
        self.vendor_progress_jobs: Dict[str, str] = {}
        self.vendor_progress_phase: Dict[str, int] = {}
        self.vendor_credentials: Dict[str, Tuple[str, str]] = {"huawei": ("", ""), "ctyun": ("", ""), "volc": ("", "")}
        self.vendor_marks: Dict[str, str] = {"huawei": "", "ctyun": "", "volc": ""}
        self.vendor_child_windows: Dict[str, List[Dict[str, Any]]] = {"huawei": [], "ctyun": [], "volc": []}
        self.credential_history_controls: List[Dict[str, Any]] = []
        self._credential_history_popup: Optional[tk.Toplevel] = None
        self._credential_history_outside_bind: Optional[tuple[tk.Misc, str]] = None
        self._mark_lookup_jobs: Dict[str, str] = {}
        self._mark_lookup_tokens: Dict[str, int] = {}
        self._balance_lookup_jobs: Dict[str, str] = {}
        self._balance_lookup_tokens: Dict[str, int] = {}
        self._balance_lookup_cache: Dict[str, str] = {}
        self._auto_resource_read_jobs: Dict[str, str] = {}
        self._auto_resource_read_tokens: Dict[str, int] = {}
        self._web_login_sessions: Dict[str, Dict[str, Any]] = {}
        cleanup_stale_task_locks()
        self.subprocess_task_manager = TaskProcessManager()
        self.images: Dict[str, tk.PhotoImage] = {}
        self.current_vendor = ""
        self.customer_manager_auth_code = normalize_manager_auth_code(CURRENT_MANAGER_AUTH_CODE)
        self.ctyun_agreed = False
        self.shell_ak_var = tk.StringVar()
        self.shell_sk_var = tk.StringVar()
        self.shell_mark_var = tk.StringVar()
        self.clone_count_var = tk.StringVar(value="1")
        self.main_balance_var = tk.StringVar(value="余额: --")
        self.global_status_var = tk.StringVar(value="就绪")
        self._syncing_credentials = False
        self._closing = False
        self._announcement_poll_job: str | None = None
        self._runtime_update_wait_job: str | None = None
        self._runtime_update_pending_result: Any = None
        self._runtime_update_dialog: tk.Toplevel | None = None
        self._window_recovery_job: str | None = None
        self.vendor_loading: Dict[str, bool] = {}
        self._interface_update_mtimes: Dict[str, float] = {}
        self._scale_fonts()
        self._configure_style()
        self._set_icon()
        self._build_shell()
        self._maximize_main_window()
        self.protocol("WM_DELETE_WINDOW", self.force_close)
        self.bind("<Map>", self._on_shell_window_mapped, add="+")
        self.bind("<FocusIn>", self._on_shell_window_focus, add="+")
        self.shell_ak_var.trace_add("write", self._on_shell_credentials_changed)
        self.shell_sk_var.trace_add("write", self._on_shell_credentials_changed)
        self.shell_mark_var.trace_add("write", lambda *_args: (self._sync_active_credentials(), self._remember_active_shell_mark()))
        self.show_vendor(self.available_vendors[0])
        self._interface_update_mtimes = self._collect_interface_update_mtimes()
        self._schedule_status_poll()
        self._schedule_window_recovery_watchdog()
        self._schedule_release_announcement_poll()

    def _maximize_main_window(self) -> None:
        """首次打开主窗口时先在隐藏状态最大化，避免窗口化首帧闪现。"""
        self.after_idle(self._show_initial_window_maximized)

    def _show_initial_window_maximized(self) -> None:
        if not getattr(self, "_initial_show_pending", False) or self._closing:
            return
        self._initial_show_pending = False
        try:
            self.update_idletasks()
            self._set_system_maximized()
            self.update_idletasks()
            self.deiconify()
            self._set_system_maximized()
            self.lift()
            self.after_idle(self._force_shell_repaint)
        except tk.TclError:
            pass

    def _set_system_maximized(self) -> None:
        for method_name in ("state", "wm_state"):
            method = getattr(self, method_name, None)
            if not callable(method):
                continue
            try:
                method("zoomed")
                return
            except tk.TclError:
                continue

    def _apply_stable_maximize(self) -> None:
        try:
            self._set_system_maximized()
        except tk.TclError:
            pass
        self._force_shell_repaint()

    def _force_shell_repaint(self) -> None:
        try:
            self.configure(bg=APP_BG)
            if hasattr(self, "main_shell"):
                self.main_shell.configure(style="Content.TFrame")
            self.update_idletasks()
            if hasattr(self, "page_host"):
                self.page_host.configure(style="Content.TFrame")
                self.page_host.update_idletasks()
            current_page = self.vendor_pages.get(self.current_vendor)
            if current_page is not None:
                current_page.update_idletasks()
                current_page.tkraise()
            self.update_idletasks()
        except tk.TclError:
            pass

    def _on_shell_window_mapped(self, event: tk.Event) -> None:
        if getattr(event, "widget", None) is self:
            self.after(80, self._recover_visible_shell_window)

    def _on_shell_window_focus(self, event: tk.Event) -> None:
        if getattr(event, "widget", None) is self:
            self.after_idle(self._release_invisible_tk_grab)

    def _release_invisible_tk_grab(self) -> None:
        try:
            grabbed = self.grab_current()
        except tk.TclError:
            return
        if grabbed is None:
            return
        try:
            if not grabbed.winfo_exists() or not grabbed.winfo_viewable():
                grabbed.grab_release()
        except tk.TclError:
            pass

    def _recover_visible_shell_window(self) -> None:
        if self._closing:
            return
        self._release_invisible_tk_grab()
        try:
            if str(self.state()) == "iconic":
                return
            self.update_idletasks()
            x = self.winfo_x()
            y = self.winfo_y()
            width = self.winfo_width()
            height = self.winfo_height()
            offscreen_or_tiny = x < -10000 or y < -10000 or width < 400 or height < 300
            if offscreen_or_tiny:
                self.deiconify()
                self._set_system_maximized()
            self.lift()
            self._force_shell_repaint()
        except tk.TclError:
            pass

    def _schedule_window_recovery_watchdog(self) -> None:
        if self._closing:
            return
        self._release_invisible_tk_grab()
        try:
            self._window_recovery_job = self.after(2500, self._schedule_window_recovery_watchdog)
        except tk.TclError:
            self._window_recovery_job = None

    def _scale_fonts(self) -> None:
        for name in (
            "TkDefaultFont",
            "TkTextFont",
            "TkFixedFont",
            "TkMenuFont",
            "TkHeadingFont",
            "TkCaptionFont",
            "TkSmallCaptionFont",
            "TkIconFont",
            "TkTooltipFont",
        ):
            try:
                font = tkfont.nametofont(name)
            except Exception:
                continue
            try:
                base_size = int(font.cget("size"))
                font.configure(family=UI_FONT_FAMILY, size=max(11, int(base_size * 1.18)))
            except Exception:
                pass

    def _configure_style(self) -> None:
        style = ttk.Style()
        try:
            if "vista" in style.theme_names():
                style.theme_use("vista")
        except Exception:
            pass
        style.configure("TFrame", background=CONTENT_BG)
        style.configure("Content.TFrame", background=CONTENT_BG)
        style.configure("TLabel", background=CONTENT_BG, foreground=TEXT_FG)
        style.configure("Header.TLabel", background=CONTENT_BG, foreground=TEXT_FG, font=(UI_LARGE_FONT_FAMILY, 18, "bold"))
        style.configure("HeaderSubtitle.TLabel", background=CONTENT_BG, foreground=MUTED_FG, font=(UI_FONT_FAMILY, 11))
        style.configure("Subheader.TLabel", background=CONTENT_BG, foreground=MUTED_FG, font=(UI_FONT_FAMILY, 10))
        style.configure("TLabelframe", background=CONTENT_BG)
        style.configure("TLabelframe.Label", background=CONTENT_BG, foreground="#344054", font=(UI_FONT_FAMILY, 10, "bold"))
        style.configure("TButton", padding=(8, 5), font=(UI_FONT_FAMILY, 10))
        style.configure("LinkAction.TButton", padding=(8, 4), font=(UI_FONT_FAMILY, 10))
        style.configure("Counter.TButton", padding=(0, 0), font=(UI_LARGE_FONT_FAMILY, 13, "bold"))
        style.configure("CloneWindow.TButton", padding=(0, 0))
        style.configure("TRadiobutton", background=CONTENT_BG, foreground=TEXT_FG)
        style.map("TRadiobutton", background=[("active", CONTENT_BG), ("selected", CONTENT_BG)], foreground=[("active", TEXT_FG), ("selected", TEXT_FG)])
        style.configure("TCheckbutton", background=CONTENT_BG, foreground=TEXT_FG)
        style.map("TCheckbutton", background=[("active", CONTENT_BG), ("selected", CONTENT_BG)], foreground=[("active", TEXT_FG), ("selected", TEXT_FG)])
        style.configure("TNotebook", background=CONTENT_BG, tabmargins=(0, 0, 0, 0))
        style.configure(
            "TNotebook.Tab",
            padding=(37, 18),
            font=(UI_LARGE_FONT_FAMILY, 14, "bold"),
            background=CONTENT_BG,
            foreground=TEXT_FG,
        )
        style.map("TNotebook.Tab", background=[("selected", CONTENT_BG), ("active", BUTTON_ACTIVE_BG)], foreground=[("selected", TEXT_FG), ("active", TEXT_FG)])
        style.configure("Treeview", rowheight=28, font=(UI_FONT_FAMILY, 10), background=TEXT_BG, fieldbackground=TEXT_BG, foreground=TEXT_FG)
        style.configure("Treeview.Heading", font=(UI_FONT_FAMILY, 10), background=TABLE_HEADER_BG, foreground=TEXT_FG, bordercolor=TABLE_BORDER_COLOR)
        for name in ("Vertical.TScrollbar", "Horizontal.TScrollbar", "App.Vertical.TScrollbar", "App.Horizontal.TScrollbar"):
            style.configure(
                name,
                background=SCROLL_THUMB,
                darkcolor=SCROLL_THUMB,
                lightcolor=SCROLL_THUMB,
                troughcolor=SCROLL_TROUGH,
                bordercolor=SCROLL_TROUGH,
                arrowcolor=MUTED_FG,
                relief="flat",
                width=12,
            )
            style.map(
                name,
                background=[("active", SCROLL_ACTIVE), ("pressed", SCROLL_ACTIVE)],
                arrowcolor=[("active", TEXT_FG)],
            )
        style.configure(
            "Grid.Treeview",
            rowheight=28,
            font=(UI_FONT_FAMILY, 10),
            background=TEXT_BG,
            fieldbackground=TEXT_BG,
            foreground=TEXT_FG,
            borderwidth=1,
            relief="solid",
        )
        style.configure(
            "Grid.Treeview.Heading",
            font=(UI_FONT_FAMILY, 10),
            background=TABLE_HEADER_BG,
            foreground=TEXT_FG,
            borderwidth=1,
            bordercolor=TABLE_BORDER_COLOR,
            relief="solid",
        )
        style.map("Grid.Treeview", background=[("selected", "#1570ef")], foreground=[("selected", "#ffffff")])
        style.configure(
            "Vendor.Horizontal.TProgressbar",
            troughcolor=PANEL_BG,
            background="#1570ef",
            darkcolor="#1570ef",
            lightcolor="#53b1fd",
            bordercolor=PANEL_BG,
            relief="flat",
            thickness=15,
        )

    def _set_icon(self) -> None:
        apply_multi_window_icon(self)

    def _load_photo(self, filename: str) -> Optional[tk.PhotoImage]:
        path = resource_path(filename)
        if not os.path.exists(path):
            return None
        try:
            image = tk.PhotoImage(file=path)
            self.images[filename] = image
            return image
        except tk.TclError:
            return None

    def _build_shell(self) -> None:
        self.columnconfigure(1, weight=1)
        self.rowconfigure(0, weight=1)
        self.rowconfigure(1, weight=0)

        sidebar = tk.Frame(self, bg=PANEL_BG, width=100, highlightthickness=1, highlightbackground=BORDER_COLOR)
        sidebar.grid(row=0, column=0, sticky="ns")
        sidebar.grid_propagate(False)
        sidebar.columnconfigure(0, weight=1)
        sidebar.rowconfigure(0, weight=1)
        sidebar.rowconfigure(2, weight=0)

        vendor_area = tk.Frame(sidebar, bg=PANEL_BG)
        vendor_area.grid(row=0, column=0, sticky="nsew", pady=(18, 0))
        vendor_area.columnconfigure(0, weight=1)
        vendor_area.rowconfigure(0, weight=1)
        vendor_area.rowconfigure(5, weight=1)
        vendor_defs = (
            ("huawei", "华为云管理", "", "logo_huawei_58.png"),
            ("ctyun", "天翼云管理", "", "logo_ctyun_58.png"),
            ("volc", "火山引擎管理", "", "logo_volc_58.png"),
        )
        row_index = 1
        for vendor_key, label, subtitle, logo in vendor_defs:
            if vendor_key not in self.available_vendors:
                continue
            self._add_vendor_button(vendor_area, vendor_key, label, subtitle, logo, row_index)
            row_index += 1
        self._add_sidebar_action_button(vendor_area, "联系客户经理", "logo_contact_manager_58.png", row_index, self._show_customer_manager_contact_dialog)

        menu_area = tk.Frame(sidebar, bg=PANEL_BG)
        menu_area.grid(row=2, column=0, sticky="ew", pady=(0, 18))
        menu_area.columnconfigure(0, weight=1)
        self.sidebar_menu_button = tk.Canvas(
            menu_area,
            width=38,
            height=30,
            bg=PANEL_BG,
            highlightthickness=0,
            bd=0,
            cursor="hand2",
        )
        self.sidebar_menu_button.grid(row=0, column=0)
        self._draw_sidebar_menu_icon(PANEL_BG)
        self.sidebar_menu_button.bind("<Button-1>", lambda _event: self._show_sidebar_menu())
        self.sidebar_menu_button.bind("<Enter>", lambda _event: self._draw_sidebar_menu_icon(ACCENT_BG))
        self.sidebar_menu_button.bind("<Leave>", lambda _event: self._draw_sidebar_menu_icon(PANEL_BG))

        main = ttk.Frame(self, style="Content.TFrame", padding=(14, 12, 14, 14))
        self.main_shell = main
        main.grid(row=0, column=1, sticky="nsew")
        main.columnconfigure(0, weight=1)
        main.rowconfigure(1, weight=1)

        header = ttk.Frame(main, style="Content.TFrame")
        header.grid(row=0, column=0, sticky="ew", pady=(0, 10))
        header.columnconfigure(0, weight=1)
        header.columnconfigure(1, weight=0)
        header.columnconfigure(2, weight=0)
        header.columnconfigure(3, weight=0)
        header.columnconfigure(4, weight=0)
        self.page_title_var = tk.StringVar()
        self.page_subtitle_var = tk.StringVar()
        title_block = ttk.Frame(header, style="Content.TFrame", padding=(0, 18, 0, 0))
        title_block.grid(row=0, column=0, rowspan=2, sticky="nw")
        ttk.Label(title_block, textvariable=self.page_title_var, style="Header.TLabel").grid(row=0, column=0, sticky="w")
        ttk.Label(title_block, textvariable=self.page_subtitle_var, style="HeaderSubtitle.TLabel").grid(row=1, column=0, sticky="w", pady=(4, 0))

        clone_box = tk.Frame(header, width=100, height=118, bg=CONTENT_BG)
        clone_box.grid(row=0, column=1, rowspan=3, sticky="se", padx=(0, 8))
        clone_box.grid_propagate(False)
        tk.Label(
            clone_box,
            text="新增数量",
            bg=CONTENT_BG,
            fg=TEXT_FG,
            font=(UI_FONT_FAMILY, 10),
        ).place(x=0, y=23, width=100, height=24)
        ttk.Button(
            clone_box,
            text="-",
            command=lambda: self._adjust_clone_count(self.clone_count_var, -1),
            style="Counter.TButton",
        ).place(x=5, y=50, width=30, height=30)
        ttk.Entry(
            clone_box,
            textvariable=self.clone_count_var,
            justify="center",
        ).place(x=35, y=50, width=30, height=30)
        ttk.Button(
            clone_box,
            text="+",
            command=lambda: self._adjust_clone_count(self.clone_count_var, 1),
            style="Counter.TButton",
        ).place(x=65, y=50, width=30, height=30)
        ttk.Button(
            clone_box,
            text="新增窗口",
            command=lambda: self._open_vendor_subwindows(self.current_vendor, self.clone_count_var),
            style="LinkAction.TButton",
        ).place(x=0, y=88, width=100, height=30)

        login_box = self._build_web_login_box(header, command=self.open_current_web_login)
        login_box.grid(row=0, column=2, rowspan=2, sticky="se", padx=(0, 5))

        cred = ttk.Frame(header, style="Content.TFrame")
        cred.grid(row=0, column=3, rowspan=2, sticky="e")
        cred.columnconfigure(1, minsize=488)
        ttk.Label(cred, text="访问密钥:AK", style="Subheader.TLabel").grid(row=0, column=0, sticky="w", padx=(0, 6))
        ttk.Entry(cred, textvariable=self.shell_ak_var, width=68).grid(row=0, column=1, sticky="ew", padx=(0, 10), pady=(0, 4))
        ttk.Label(cred, text="访问密钥:SK", style="Subheader.TLabel").grid(row=1, column=0, sticky="w", padx=(0, 6))
        ttk.Entry(cred, textvariable=self.shell_sk_var, width=68).grid(row=1, column=1, sticky="ew", padx=(0, 10), pady=(0, 4))
        ttk.Label(cred, text="用户名(可编辑):", style="Subheader.TLabel").grid(row=2, column=0, sticky="w", padx=(0, 6))
        self.main_balance_box = self._build_credential_history_row(
            cred,
            vendor=None,
            app=None,
            parent_window=self,
            ak_var=self.shell_ak_var,
            sk_var=self.shell_sk_var,
            mark_var=self.shell_mark_var,
            balance_var=self.main_balance_var,
        )
        paste_box = tk.Frame(cred, width=75, height=88, bg=CONTENT_BG)
        paste_box.grid(row=0, column=2, rowspan=3, sticky="ns")
        paste_box.grid_propagate(False)
        ttk.Button(
            paste_box,
            text="识别\n粘贴",
            command=self.paste_current_credentials,
        ).place(x=0, y=0, width=75, height=88)
        self.binding_link_button, self.binding_tutorial_button = self._build_binding_action_row(
            header,
            vendor=None,
            row=2,
            column=2,
            columnspan=2,
        )

        self.page_host = ttk.Frame(main, style="Content.TFrame")
        self.page_host.grid(row=1, column=0, sticky="nsew")
        self.page_host.columnconfigure(0, weight=1)
        self.page_host.rowconfigure(0, weight=1)
        all_vendor_factories: Dict[str, Callable[[ttk.Frame], None]] = {
            "huawei": self._create_huawei_page,
            "ctyun": self._create_ctyun_page,
            "volc": self._create_volc_page,
        }
        self.vendor_factories = {key: all_vendor_factories[key] for key in self.available_vendors}

        status_bar = tk.Frame(self, bg=PANEL_BG, height=28, highlightthickness=1, highlightbackground=BORDER_COLOR)
        status_bar.grid(row=1, column=0, columnspan=2, sticky="ew")
        status_bar.grid_propagate(False)
        status_bar.columnconfigure(0, weight=1)
        tk.Label(
            status_bar,
            textvariable=self.global_status_var,
            bg=PANEL_BG,
            fg=TEXT_FG,
            anchor="e",
            font=(UI_FONT_FAMILY, 10),
        ).grid(row=0, column=0, sticky="e", padx=16, pady=4)

    def _create_huawei_page(self, huawei_page: ttk.Frame) -> None:
        huawei_page.columnconfigure(0, weight=1)
        huawei_page.rowconfigure(0, weight=1)
        self.huawei_app = HuaweiAutoBuyApp(huawei_page, embedded=True)
        self.huawei_app.subprocess_task_manager = self.subprocess_task_manager
        self._set_app_mark(self.huawei_app, self.vendor_marks.get("huawei", ""))
        self.huawei_app.grid(row=0, column=0, sticky="nsew")
        self.huawei_app.status_var.trace_add("write", lambda *_args: self._update_global_status())

    def _create_ctyun_page(self, ctyun_page: ttk.Frame) -> None:
        ctyun_page.columnconfigure(0, weight=1)
        ctyun_page.rowconfigure(0, weight=1)
        self.ctyun_gui = ctyun_workflow_gui.WorkflowGui(self, parent=ctyun_page, embedded=True)
        self.ctyun_gui.subprocess_task_manager = self.subprocess_task_manager
        self._set_app_mark(self.ctyun_gui, self.vendor_marks.get("ctyun", ""))
        self.ctyun_gui.status_var.trace_add("write", lambda *_args: self._update_global_status())

    def _create_volc_page(self, volc_page: ttk.Frame) -> None:
        volc_page.columnconfigure(0, weight=1)
        volc_page.rowconfigure(0, weight=1)
        self.volc_app = VolcEngineManagerApp(volc_page, embedded=True)
        self.volc_app.subprocess_task_manager = self.subprocess_task_manager
        self._set_app_mark(self.volc_app, self.vendor_marks.get("volc", ""))
        self.volc_app.grid(row=0, column=0, sticky="nsew")
        self.volc_app.status_var.trace_add("write", lambda *_args: self._update_global_status())

    def _ensure_vendor_page(self, key: str, *, immediate: bool = False) -> ttk.Frame:
        page = self.vendor_pages.get(key)
        if page is None:
            page = ttk.Frame(self.page_host, style="Content.TFrame")
            page.grid(row=0, column=0, sticky="nsew")
            page.columnconfigure(0, weight=1)
            page.rowconfigure(0, weight=1)
            self.vendor_pages[key] = page
            ttk.Label(page, text="正在加载...", style="Subheader.TLabel").grid(row=0, column=0, sticky="n", pady=30)
            self.vendor_loading[key] = True
            if immediate:
                self._load_vendor_page(key)
            else:
                self.after(20, lambda item_key=key: self._load_vendor_page(item_key))
        return page

    def _load_vendor_page(self, key: str) -> None:
        if self._closing or not self.vendor_loading.get(key):
            return
        page = self.vendor_pages.get(key)
        if page is None:
            return
        for child in page.winfo_children():
            child.destroy()
        try:
            self.vendor_factories[key](page)
            self.vendor_loading[key] = False
            saved_ak, saved_sk = self.vendor_credentials.get(key, ("", ""))
            saved_mark = self.vendor_marks.get(key, "")
            app = self._active_vendor_app(key)
            if app is not None:
                self._set_app_credentials(app, saved_ak, saved_sk, saved_mark, silent=True)
            if key == self.current_vendor:
                self._load_shell_credentials(key)
                self._update_global_status()
            self._refresh_vendor_layout(page)
        except Exception as exc:
            self.vendor_loading[key] = False
            trace = traceback.format_exc()
            ttk.Label(page, text=f"加载失败：{exc}", foreground="#b42318", style="Subheader.TLabel").grid(row=0, column=0, sticky="n", pady=30)
            self._show_message(f"{key} 页面加载失败：{exc}")
            print(trace)

    def _show_message(self, message: str, title: str | None = None) -> None:
        message = friendly_popup_text(message)
        show_app_message_dialog(self, title or getattr(self, "app_title", APP_TITLE), message, modal=True)

    def _build_web_login_box(self, parent: tk.Misc, *, command: Callable[[], None]) -> tk.Frame:
        box = tk.Frame(parent, width=75, height=88, bg=CONTENT_BG)
        box.grid_propagate(False)
        ttk.Button(
            box,
            text="登录账号\n\n自动获取",
            command=command,
        ).place(x=0, y=0, width=75, height=88)
        return box

    def _vendor_binding_display_name(self, vendor: Optional[str] = None) -> str:
        return CUSTOMER_MANAGER_VENDOR_NAMES.get(vendor or self.current_vendor or "huawei", "云")

    def open_current_web_login(self) -> None:
        app = self._active_vendor_app(self.current_vendor)
        if app is None:
            self._show_message("当前云页面尚未加载完成。")
            return
        self._open_web_login_for_credentials(
            vendor=self.current_vendor,
            app=app,
            ak_var=self.shell_ak_var,
            sk_var=self.shell_sk_var,
            mark_var=self.shell_mark_var,
            parent=self,
        )

    def _build_binding_action_row(
        self,
        parent: tk.Misc,
        *,
        vendor: Optional[str],
        row: int,
        column: int,
        columnspan: int,
        parent_window: Optional[tk.Misc] = None,
    ) -> Tuple[ttk.Button, ttk.Button]:
        action_bar = ttk.Frame(parent, style="Content.TFrame")
        action_bar.grid(row=row, column=column, columnspan=columnspan, sticky="ew", pady=(8, 0))
        action_bar.columnconfigure(0, weight=1, uniform="binding_action")
        action_bar.columnconfigure(1, weight=1, uniform="binding_action")
        action_bar.rowconfigure(0, minsize=30)

        def target_vendor() -> str:
            return vendor or self.current_vendor or "huawei"

        cloud_name = self._vendor_binding_display_name(target_vendor())
        link_button = ttk.Button(
            action_bar,
            text=f"进入{cloud_name}注册绑定链接",
            command=lambda: self._open_vendor_binding_link(target_vendor(), parent_window),
            style="LinkAction.TButton",
        )
        tutorial_button = ttk.Button(
            action_bar,
            text=f"查看{cloud_name}获取Ak/Sk教程",
            command=lambda: self._show_vendor_aksk_tutorial(target_vendor(), parent_window),
            style="LinkAction.TButton",
        )
        tutorial_button.grid(row=0, column=0, sticky="nsew", padx=(0, 8))
        link_button.grid(row=0, column=1, sticky="nsew", padx=(8, 0))
        return link_button, tutorial_button

    def _refresh_binding_action_labels(self, vendor: Optional[str] = None) -> None:
        if not hasattr(self, "binding_link_button") or not hasattr(self, "binding_tutorial_button"):
            return
        cloud_name = self._vendor_binding_display_name(vendor)
        self.binding_link_button.configure(text=f"进入{cloud_name}注册绑定链接")
        self.binding_tutorial_button.configure(text=f"查看{cloud_name}获取Ak/Sk教程")

    def _current_manager_auth_code(self) -> str:
        code = normalize_manager_auth_code(getattr(self, "customer_manager_auth_code", "") or CURRENT_MANAGER_AUTH_CODE)
        return code if code in CUSTOMER_MANAGER_BINDING_LINKS else ""

    def _vendor_binding_url(self, vendor: str) -> str:
        code = normalize_manager_auth_code(
            getattr(self, "customer_manager_auth_code", "") or CURRENT_MANAGER_AUTH_CODE
        )
        cached_url = str(CUSTOMER_MANAGER_BINDING_LINKS.get(code, {}).get(vendor, "")).strip()
        if cached_url.startswith("https://"):
            return cached_url
        return _manager_auth_public_binding_links(code).get(vendor, "")

    def _open_vendor_binding_link(self, vendor: str, parent: Optional[tk.Misc] = None) -> None:
        url = self._vendor_binding_url(vendor)
        if not url:
            cloud_name = self._vendor_binding_display_name(vendor)
            show_app_message_dialog(
                parent or self,
                f"{cloud_name}注册链接",
                f"{cloud_name}注册链接暂不可用，请联系软件分享方。",
                modal=True,
            )
            return
        webbrowser.open(url)

    def _show_vendor_aksk_tutorial(self, vendor: str, parent: Optional[tk.Misc] = None) -> None:
        cloud_name = self._vendor_binding_display_name(vendor)
        url = AK_SK_TUTORIAL_URLS.get(vendor, "")
        if not url:
            show_app_message_dialog(parent or self, f"{cloud_name}获取AK/SK教程", f"请联系软件获取渠道查看{cloud_name}获取 AK/SK 教程。", modal=True)
            return
        webbrowser.open(url)

    @staticmethod
    def _credential_web_login_available() -> bool:
        util = getattr(importlib, "util", None)
        if util is None:
            return False
        try:
            return util.find_spec("webview") is not None
        except Exception:
            return False

    @staticmethod
    def _credential_web_login_command(vendor: str, *, source_python: bool = False) -> tuple[list[str], str | None]:
        source_root = Path(__file__).resolve().parents[1]
        force_source_python = source_python
        try:
            portable_root = portable_root_dir()
            force_source_python = force_source_python or (portable_root / "data" / LOCAL_WEB_LOGIN_SOURCE_PYTHON_FLAG).exists()
        except Exception:
            pass
        if force_source_python:
            python_exe = shutil.which("python")
            if python_exe:
                return [python_exe, "-B", str(source_root / "main.py"), "--credential-web-login", vendor, "-"], str(source_root)
            return [], str(source_root)
        if getattr(sys, "frozen", False):
            return [sys.executable, "--credential-web-login", vendor, "-"], str(Path(sys.executable).resolve().parent)
        return [sys.executable, "-B", str(source_root / "main.py"), "--credential-web-login", vendor, "-"], str(source_root)

    @staticmethod
    def _start_web_login_process(command: list[str], cwd: str | None) -> subprocess.Popen:
        if not command:
            raise RuntimeError("未找到可用的 Python 运行环境。")
        creationflags = subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0
        env = os.environ.copy()
        env["PYTHONUTF8"] = "1"
        env["PYTHONIOENCODING"] = "utf-8"
        return subprocess.Popen(
            command,
            cwd=cwd,
            close_fds=True,
            creationflags=creationflags,
            env=env,
            stdout=subprocess.PIPE,
            stderr=subprocess.DEVNULL,
            text=True,
            encoding="utf-8",
            errors="replace",
        )

    def _web_login_target_key(self, vendor: str, parent: tk.Misc) -> str:
        scope = "main" if parent is self else f"child-{id(parent)}"
        return f"{scope}:{vendor}:web-login"

    def _cleanup_web_login_session(self, target_key: str, *, terminate_process: bool = False) -> None:
        session = self._web_login_sessions.pop(target_key, None)
        if not isinstance(session, dict):
            return
        job = str(session.get("job") or "")
        if job:
            try:
                self.after_cancel(job)
            except tk.TclError:
                pass
        process = session.get("process")
        if terminate_process and process is not None:
            try:
                if process.poll() is None:
                    process.terminate()
            except Exception:
                pass

    @staticmethod
    def _start_web_login_stdout_reader(process: Any) -> list[str]:
        lines: list[str] = []
        stream = getattr(process, "stdout", None)
        if stream is None:
            return lines

        def read_stdout() -> None:
            try:
                for line in stream:
                    lines.append(line)
            except Exception:
                pass

        threading.Thread(target=read_stdout, daemon=True).start()
        return lines

    def _open_web_login_for_credentials(
        self,
        *,
        vendor: str,
        app: Any,
        ak_var: tk.StringVar,
        sk_var: tk.StringVar,
        mark_var: tk.StringVar,
        parent: tk.Misc,
        title_var: Optional[tk.StringVar] = None,
    ) -> None:
        vendor = (vendor or "").strip()
        cloud_name = CLOUD_CREDENTIAL_PAGE_NAMES.get(vendor, self._vendor_binding_display_name(vendor))
        if vendor not in CLOUD_CREDENTIAL_PAGE_URLS:
            self._show_credential_message(parent, "未知云厂商，无法打开 AK/SK 登录页。")
            return
        if app is None:
            self._show_credential_message(parent, "当前云页面尚未加载完成。")
            return
        if not self._credential_web_login_available():
            show_app_message_dialog(parent, f"{cloud_name}登录获取AK/SK", WEBVIEW_IMPORT_MISSING_MESSAGE, modal=True)
            return

        target_key = self._web_login_target_key(vendor, parent)
        self._cleanup_web_login_session(target_key, terminate_process=True)
        command, cwd = self._credential_web_login_command(vendor)
        try:
            process = self._start_web_login_process(command, cwd)
        except Exception as exc:
            show_app_message_dialog(parent, f"{cloud_name}登录获取AK/SK", f"打开网页登录窗口失败：{exc}", modal=True)
            return

        self._web_login_sessions[target_key] = {
            "process": process,
            "source_python": False,
            "stdout_lines": self._start_web_login_stdout_reader(process),
            "result_applied": False,
        }
        self._note_mark_lookup_result(app, f"已打开{cloud_name}网页登录窗口，登录后将自动提取 AK/SK。")
        self._poll_web_login_result(
            target_key=target_key,
            vendor=vendor,
            app=app,
            ak_var=ak_var,
            sk_var=sk_var,
            mark_var=mark_var,
            parent=parent,
            title_var=title_var,
        )

    @staticmethod
    def _read_web_login_stdout_result(lines: Any) -> Dict[str, Any]:
        if not isinstance(lines, list):
            return {}
        for line in reversed(lines):
            text = line.strip()
            if not text:
                continue
            try:
                data = json.loads(text)
            except Exception:
                continue
            if isinstance(data, dict):
                return data
        return {}

    def _poll_web_login_result(
        self,
        *,
        target_key: str,
        vendor: str,
        app: Any,
        ak_var: tk.StringVar,
        sk_var: tk.StringVar,
        mark_var: tk.StringVar,
        parent: tk.Misc,
        title_var: Optional[tk.StringVar] = None,
    ) -> None:
        session = self._web_login_sessions.get(target_key)
        if not isinstance(session, dict):
            return
        try:
            if isinstance(parent, tk.Toplevel) and not parent.winfo_exists():
                self._cleanup_web_login_session(target_key, terminate_process=True)
                return
        except tk.TclError:
            self._cleanup_web_login_session(target_key, terminate_process=True)
            return
        process = session.get("process")
        try:
            finished = process is not None and process.poll() is not None
        except Exception:
            finished = True
        result_applied = bool(session.get("result_applied"))
        if not result_applied:
            data = self._read_web_login_stdout_result(session.get("stdout_lines"))
            if data:
                session["result_applied"] = True
                self._apply_web_login_result(
                    vendor=vendor,
                    app=app,
                    ak_var=ak_var,
                    sk_var=sk_var,
                    mark_var=mark_var,
                    parent=parent,
                    title_var=title_var,
                    data=data,
                )
                result_applied = True
        if finished:
            data = {} if result_applied else self._read_web_login_stdout_result(session.get("stdout_lines"))
            if not data and not bool(session.get("source_python")) and not result_applied:
                try:
                    fallback_command, fallback_cwd = self._credential_web_login_command(vendor, source_python=True)
                    fallback_process = self._start_web_login_process(fallback_command, fallback_cwd)
                    session["process"] = fallback_process
                    session["source_python"] = True
                    session["stdout_lines"] = self._start_web_login_stdout_reader(fallback_process)
                    job = self.after(
                        600,
                        lambda: self._poll_web_login_result(
                            target_key=target_key,
                            vendor=vendor,
                            app=app,
                            ak_var=ak_var,
                            sk_var=sk_var,
                            mark_var=mark_var,
                            parent=parent,
                            title_var=title_var,
                        ),
                    )
                    session["job"] = str(job)
                    return
                except Exception:
                    data = {}
            self._cleanup_web_login_session(target_key)
            if result_applied:
                return
            if not data:
                data = {"ok": False, "message": "网页登录窗口已关闭，未提取到完整 AK/SK。"}
            self._apply_web_login_result(
                vendor=vendor,
                app=app,
                ak_var=ak_var,
                sk_var=sk_var,
                mark_var=mark_var,
                parent=parent,
                title_var=title_var,
                data=data,
            )
            return
        try:
            job = self.after(
                600,
                lambda: self._poll_web_login_result(
                    target_key=target_key,
                    vendor=vendor,
                    app=app,
                    ak_var=ak_var,
                    sk_var=sk_var,
                    mark_var=mark_var,
                    parent=parent,
                    title_var=title_var,
                ),
            )
            session["job"] = str(job)
        except tk.TclError:
            self._cleanup_web_login_session(target_key, terminate_process=True)

    def _apply_web_login_result(
        self,
        *,
        vendor: str,
        app: Any,
        ak_var: tk.StringVar,
        sk_var: tk.StringVar,
        mark_var: tk.StringVar,
        parent: tk.Misc,
        title_var: Optional[tk.StringVar],
        data: Dict[str, Any],
    ) -> None:
        cloud_name = CLOUD_CREDENTIAL_PAGE_NAMES.get(vendor, self._vendor_binding_display_name(vendor))
        if not data.get("ok"):
            message = str(data.get("message") or "未从网页登录页提取到完整 AK/SK，请按页面提示创建或下载访问密钥后重试。")
            show_app_message_dialog(parent, f"{cloud_name}登录获取AK/SK", message, modal=True)
            return
        ak = str(data.get("ak") or "").strip()
        sk = str(data.get("sk") or "").strip()
        mark = self._clean_resolved_mark(data.get("mark") or "")
        if not ak or not sk:
            show_app_message_dialog(parent, f"{cloud_name}登录获取AK/SK", "未从网页登录页提取到完整 AK/SK，请按页面提示创建或下载访问密钥后重试。", modal=True)
            return
        self._apply_resolved_credentials_into_app(
            vendor=vendor,
            app=app,
            ak_var=ak_var,
            sk_var=sk_var,
            mark_var=mark_var,
            parent=parent,
            title_var=title_var,
            ak=ak,
            sk=sk,
            mark=mark,
            success_note=f"已从{cloud_name}网页登录页提取 AK/SK。",
        )

    def _copy_contact_value(self, value: str, status_var: tk.StringVar) -> None:
        text = str(value or "").strip()
        if not text:
            return
        try:
            self.clipboard_clear()
            self.clipboard_append(text)
            self.update_idletasks()
            status_var.set(f"已复制：{text}")
        except tk.TclError:
            status_var.set("复制失败，请手动选择复制。")

    def _show_customer_manager_contact_dialog(self) -> None:
        code = self._current_manager_auth_code()
        contacts = CUSTOMER_MANAGER_CONTACTS.get(code)
        if not contacts:
            show_app_message_dialog(self, "联系客户经理", INVALID_MANAGER_AUTH_MESSAGE, modal=True)
            return

        dialog = tk.Toplevel(self)
        dialog.title("联系客户经理")
        dialog.resizable(False, False)
        dialog.transient(self)
        apply_multi_window_icon(dialog)

        frame = ttk.Frame(dialog, padding=(24, 20, 24, 18))
        frame.pack(fill=tk.BOTH, expand=True)
        tk.Label(
            frame,
            text="联系客户经理",
            bg=APP_BG,
            fg=TEXT_FG,
            font=(UI_FONT_FAMILY, 12, "bold"),
            anchor="w",
        ).pack(fill=tk.X)
        tk.Label(
            frame,
            text="如需获取授权码、绑定合作伙伴、开通折扣、疑难解决，请联系客户经理。",
            bg=APP_BG,
            fg=TEXT_FG,
            font=(UI_FONT_FAMILY, 10),
            justify=tk.LEFT,
            wraplength=520,
            anchor="w",
        ).pack(fill=tk.X, pady=(10, 12))

        body = tk.Frame(frame, bg=APP_BG)
        body.pack(fill=tk.X)
        status_var = tk.StringVar(value="点击蓝色字体复制")

        def add_copy_link(parent: tk.Widget, value: str) -> None:
            link = tk.Label(
                parent,
                text=value,
                bg=APP_BG,
                fg="#0969da",
                activeforeground="#074ea3",
                cursor="hand2",
                font=(UI_FONT_FAMILY, 10, "underline"),
            )
            link.pack(side=tk.LEFT)
            link.bind("<Button-1>", lambda _event, item=value: self._copy_contact_value(item, status_var))

        tk.Label(
            body,
            text="当前授权码：",
            bg=APP_BG,
            fg=TEXT_FG,
            font=(UI_FONT_FAMILY, 10),
            anchor="e",
        ).grid(row=0, column=0, sticky="e", pady=(0, 8))
        auth_frame = tk.Frame(body, bg=APP_BG)
        auth_frame.grid(row=0, column=1, sticky="w", pady=(0, 8))
        add_copy_link(auth_frame, code)
        tk.Label(
            auth_frame,
            text="（如若分享软件，请一并分享授权码，否则无法进入软件）",
            bg=APP_BG,
            fg=TEXT_FG,
            font=(UI_FONT_FAMILY, 10),
        ).pack(side=tk.LEFT)

        row_index = 1
        for label_text, values in contacts:
            tk.Label(
                body,
                text=f"{label_text}：",
                bg=APP_BG,
                fg=TEXT_FG,
                font=(UI_FONT_FAMILY, 10),
                anchor="e",
            ).grid(row=row_index, column=0, sticky="e", pady=3)
            value_frame = tk.Frame(body, bg=APP_BG)
            value_frame.grid(row=row_index, column=1, sticky="w", pady=3)
            for index, value in enumerate(values):
                if index > 0:
                    separator = "丨" if label_text == "QQ" else " | "
                    tk.Label(value_frame, text=separator, bg=APP_BG, fg=TEXT_FG, font=(UI_FONT_FAMILY, 10)).pack(side=tk.LEFT)
                add_copy_link(value_frame, value)
            row_index += 1

        tk.Label(
            frame,
            textvariable=status_var,
            bg=APP_BG,
            fg="#6b7280",
            font=(UI_FONT_FAMILY, 9),
            anchor="w",
        ).pack(fill=tk.X, pady=(14, 12))
        ttk.Button(frame, text="关闭", command=dialog.destroy, width=14).pack(anchor="center")

        dialog.protocol("WM_DELETE_WINDOW", dialog.destroy)
        dialog.update_idletasks()
        width = max(420, dialog.winfo_reqwidth())
        height = max(250, dialog.winfo_reqheight())
        try:
            x = self.winfo_rootx() + max(0, (self.winfo_width() - width) // 2)
            y = self.winfo_rooty() + max(0, (self.winfo_height() - height) // 2)
        except tk.TclError:
            x = y = 120
        dialog.geometry(f"{width}x{height}+{x}+{y}")
        dialog.grab_set()
        dialog.focus_force()
        self.wait_window(dialog)

    def _show_sidebar_menu(self) -> None:
        menu = tk.Menu(self, tearoff=0)
        menu.add_command(label="界面更新", command=self._refresh_interface_from_menu)
        menu.add_separator()
        menu.add_command(label=f"当前版本：{APP_VERSION}", state="disabled")
        try:
            x = self.sidebar_menu_button.winfo_rootx()
            y = self.sidebar_menu_button.winfo_rooty() + self.sidebar_menu_button.winfo_height()
            menu.tk_popup(x, y)
        finally:
            menu.grab_release()

    def _draw_sidebar_menu_icon(self, background: str) -> None:
        """用画布绘制三横菜单，避免打包后字体缺失导致按钮看不见。"""
        try:
            self.sidebar_menu_button.delete("all")
            self.sidebar_menu_button.configure(bg=background)
            for y in (9, 15, 21):
                self.sidebar_menu_button.create_line(10, y, 28, y, fill=TEXT_FG, width=2, capstyle=tk.ROUND)
        except tk.TclError:
            pass

    def _active_task_running(self) -> bool:
        huawei_busy = bool(getattr(getattr(self, "huawei_app", None), "busy", False))
        ctyun_busy = bool(getattr(getattr(self, "ctyun_gui", None), "running", False))
        volc_busy = bool(getattr(getattr(self, "volc_app", None), "busy", False))
        subprocess_busy = self.subprocess_task_manager.has_running_tasks()
        return huawei_busy or ctyun_busy or volc_busy or subprocess_busy

    def _module_source_mtime(self, module: Any) -> Optional[float]:
        filename = str(getattr(module, "__file__", "") or "")
        if not filename:
            return None
        if filename.endswith((".pyc", ".pyo")):
            try:
                filename = importlib.util.source_from_cache(filename)
            except Exception:
                return None
        if not filename.endswith(".py") or not os.path.exists(filename):
            return None
        try:
            return os.path.getmtime(filename)
        except OSError:
            return None

    def _collect_interface_update_mtimes(self) -> Dict[str, float]:
        prefixes = ("launcher", "common", "HUAWEI", "CTYUN", "Volcengine")
        tracked: Dict[str, float] = {}
        for name, module in list(sys.modules.items()):
            if not module:
                continue
            filename = str(getattr(module, "__file__", "") or "")
            should_track = (
                name == "main"
                or (name == "__main__" and os.path.basename(filename).lower() == "main.py")
                or any(name == prefix or name.startswith(prefix + ".") for prefix in prefixes)
            )
            if not should_track:
                continue
            mtime = self._module_source_mtime(module)
            if mtime is not None:
                tracked[name] = mtime
        return tracked

    def _module_change_requires_restart(self, name: str) -> bool:
        return name in {"main", "__main__"} or name == "launcher" or name.startswith("launcher.")

    def _changed_restart_modules(self) -> List[str]:
        current = self._collect_interface_update_mtimes()
        changed: List[str] = []
        for name, new_mtime in current.items():
            old_mtime = self._interface_update_mtimes.get(name)
            if old_mtime is None:
                continue
            if self._module_change_requires_restart(name) and abs(new_mtime - old_mtime) > 0.0001:
                changed.append(name)
        return sorted(set(changed))

    def _restart_self_for_interface_update(self, changed_modules: Sequence[str]) -> None:
        # 主窗口和启动器代码已经在当前进程中运行，检测到变化时直接重启最稳。
        args = [sys.executable] if getattr(sys, "frozen", False) else [sys.executable, *sys.argv]
        try:
            subprocess.Popen(args, cwd=os.getcwd(), close_fds=True)
        except Exception as exc:
            self._show_message(f"自动重启失败：{exc}")
            return
        detail = "、".join(changed_modules[:3]) if changed_modules else "主窗口"
        self._show_message(f"检测到 {detail} 更新，软件已自动重启。")
        self.force_close()

    def _snapshot_tree_items(self, tree: ttk.Treeview, parent: str = "") -> List[Dict[str, Any]]:
        items: List[Dict[str, Any]] = []
        for iid in tree.get_children(parent):
            item = tree.item(iid)
            image = item.get("image", "")
            if isinstance(image, (list, tuple)):
                image = image[0] if image else ""
            items.append(
                {
                    "iid": iid,
                    "text": item.get("text", ""),
                    "image": str(image or ""),
                    "values": list(item.get("values") or []),
                    "open": bool(item.get("open")),
                    "tags": list(item.get("tags") or []),
                    "children": self._snapshot_tree_items(tree, iid),
                }
            )
        return items

    def _tree_items_have_images(self, tree: ttk.Treeview, parent: str = "") -> bool:
        for iid in tree.get_children(parent):
            image = tree.item(iid, "image")
            if isinstance(image, (list, tuple)):
                image = image[0] if image else ""
            if str(image or ""):
                return True
            if self._tree_items_have_images(tree, iid):
                return True
        return False

    def _snapshot_items_have_images(self, items: Sequence[Dict[str, Any]]) -> bool:
        for item in items:
            if str(item.get("image") or ""):
                return True
            if self._snapshot_items_have_images(item.get("children") or []):
                return True
        return False

    def _restore_tree_items(self, tree: ttk.Treeview, items: Sequence[Dict[str, Any]], parent: str = "") -> None:
        existing_images = set(str(name) for name in tree.tk.splitlist(tree.tk.call("image", "names")))
        for item in items:
            kwargs = {
                "text": item.get("text", ""),
                "values": item.get("values") or [],
                "open": bool(item.get("open")),
                "tags": item.get("tags") or [],
            }
            image = str(item.get("image") or "")
            if image and image in existing_images:
                kwargs["image"] = image
            iid = str(item.get("iid") or "")
            try:
                tree.insert(parent, tk.END, iid=iid, **kwargs)
                child_parent = iid
            except Exception:
                child_parent = tree.insert(parent, tk.END, **kwargs)
            self._restore_tree_items(tree, item.get("children") or [], child_parent)

    def _snapshot_one_widget(self, widget: tk.Widget) -> Dict[str, Any]:
        data: Dict[str, Any] = {"class": widget.winfo_class()}
        for option in ("textvariable", "variable"):
            try:
                var_name = str(widget.cget(option) or "")
            except Exception:
                var_name = ""
            if var_name:
                try:
                    data[f"{option}_value"] = self.globalgetvar(var_name)
                except Exception:
                    pass
        widget_class = widget.winfo_class()
        try:
            if isinstance(widget, tk.Text):
                data["text"] = widget.get("1.0", "end-1c")
            elif isinstance(widget, tk.Listbox):
                data["items"] = list(widget.get(0, tk.END))
                data["selection"] = list(widget.curselection())
            elif isinstance(widget, ttk.Treeview):
                data["tree_columns"] = list(widget["columns"])
                data["tree_displaycolumns"] = list(widget.tk.splitlist(widget.cget("displaycolumns")))
                data["tree_show"] = str(widget["show"])
                data["tree_items"] = self._snapshot_tree_items(widget)
                data["tree_selection"] = list(widget.selection())
            elif widget_class in {"Entry", "TEntry", "Combobox", "TCombobox", "Spinbox", "TSpinbox"} and hasattr(widget, "get"):
                data["value"] = widget.get()
        except Exception:
            pass
        return data

    def _snapshot_widget_tree(self, root: Optional[tk.Widget]) -> Dict[str, Any]:
        if root is None:
            return {}
        snapshot: Dict[str, Any] = {}

        def walk(widget: tk.Widget, key: str) -> None:
            snapshot[key] = self._snapshot_one_widget(widget)
            for index, child in enumerate(widget.winfo_children()):
                walk(child, f"{key}/{index}:{child.winfo_class()}")

        walk(root, "root")
        return snapshot

    def _widget_by_snapshot_path(self, root: tk.Widget, key: str) -> Optional[tk.Widget]:
        widget: tk.Widget = root
        for part in key.split("/")[1:]:
            try:
                index_text, expected_class = part.split(":", 1)
                child = widget.winfo_children()[int(index_text)]
            except Exception:
                return None
            if child.winfo_class() != expected_class:
                return None
            widget = child
        return widget

    def _restore_one_widget(self, widget: tk.Widget, data: Dict[str, Any]) -> None:
        for option in ("textvariable", "variable"):
            value_key = f"{option}_value"
            if value_key not in data:
                continue
            try:
                var_name = str(widget.cget(option) or "")
                if var_name:
                    self.globalsetvar(var_name, data[value_key])
            except Exception:
                pass
        widget_class = widget.winfo_class()
        try:
            if isinstance(widget, tk.Text) and "text" in data:
                state = str(widget.cget("state") or "")
                if state == tk.DISABLED:
                    widget.configure(state=tk.NORMAL)
                widget.delete("1.0", tk.END)
                widget.insert("1.0", data.get("text", ""))
                if state == tk.DISABLED:
                    widget.configure(state=tk.DISABLED)
            elif isinstance(widget, tk.Listbox) and "items" in data:
                widget.delete(0, tk.END)
                for item in data.get("items") or []:
                    widget.insert(tk.END, item)
                for index in data.get("selection") or []:
                    widget.selection_set(index)
            elif isinstance(widget, ttk.Treeview) and "tree_items" in data:
                if "tree_columns" not in data:
                    return
                if list(widget["columns"]) != list(data.get("tree_columns") or []):
                    return
                if list(widget.tk.splitlist(widget.cget("displaycolumns"))) != list(data.get("tree_displaycolumns") or []):
                    return
                if str(widget["show"]) != str(data.get("tree_show") or ""):
                    return
                tree_items = data.get("tree_items") or []
                if self._tree_items_have_images(widget) and not self._snapshot_items_have_images(tree_items):
                    return
                widget.delete(*widget.get_children(""))
                self._restore_tree_items(widget, tree_items)
                for iid in data.get("tree_selection") or []:
                    try:
                        widget.selection_add(iid)
                    except Exception:
                        pass
            elif widget_class in {"Entry", "TEntry", "Combobox", "TCombobox", "Spinbox", "TSpinbox"} and "value" in data:
                if hasattr(widget, "set"):
                    widget.set(data.get("value", ""))
                else:
                    widget.delete(0, tk.END)
                    widget.insert(0, data.get("value", ""))
        except Exception:
            pass

    def _restore_widget_tree(self, root: Optional[tk.Widget], snapshot: Dict[str, Any]) -> None:
        if root is None or not snapshot:
            return
        for key, data in snapshot.items():
            widget = self._widget_by_snapshot_path(root, key)
            if widget is None:
                continue
            self._restore_one_widget(widget, data)

    def _active_vendor_app(self, key: str) -> Any:
        if key == "huawei":
            return getattr(self, "huawei_app", None)
        if key == "ctyun":
            return getattr(self, "ctyun_gui", None)
        if key == "volc":
            return getattr(self, "volc_app", None)
        return None

    def _snapshot_app_internal_state(self, key: str) -> Dict[str, Any]:
        app = self._active_vendor_app(key)
        if app is None:
            return {}
        attrs_by_vendor = {
            "huawei": ("last_results", "manage_items", "tree_sort_reverse", "results_credential_fingerprint", "manage_last_clicked_iid"),
            "ctyun": ("cleanup_items", "selected_region_order", "current_workflow_keys", "current_workflow_regions", "current_workflow_mode", "active_log_name", "log_action_names"),
            "volc": ("manage_items", "volc_cached_resource_items", "volc_region_server_ips", "volc_region_socks", "_last_socks_lines", "manage_last_clicked_iid", "current_log_context", "volc_log_action_names"),
        }
        state: Dict[str, Any] = {}
        for attr in attrs_by_vendor.get(key, ()):
            if not hasattr(app, attr):
                continue
            value = getattr(app, attr)
            if isinstance(value, dict):
                state[attr] = dict(value)
            elif isinstance(value, list):
                state[attr] = list(value)
            elif isinstance(value, set):
                state[attr] = set(value)
            elif isinstance(value, tuple):
                state[attr] = tuple(value)
            elif isinstance(value, (str, int, float, bool)) or value is None:
                state[attr] = value
        return state

    def _restore_app_internal_state(self, key: str, state: Dict[str, Any]) -> None:
        app = self._active_vendor_app(key)
        if app is None:
            return
        for attr, value in state.items():
            if hasattr(app, attr):
                try:
                    setattr(app, attr, value)
                except Exception:
                    pass

    def _snapshot_current_vendor_state(self, key: str) -> Dict[str, Any]:
        page = self.vendor_pages.get(key)
        return {
            "credentials": self.vendor_credentials.get(key, ("", "")),
            "mark": self.vendor_marks.get(key, ""),
            "shell_credentials": (self.shell_ak_var.get(), self.shell_sk_var.get()),
            "shell_mark": self.shell_mark_var.get(),
            "app_state": self._snapshot_app_internal_state(key),
            "widgets": self._snapshot_widget_tree(page),
        }

    def _restore_current_vendor_state(self, key: str, snapshot: Dict[str, Any]) -> None:
        credentials = snapshot.get("credentials")
        if isinstance(credentials, tuple) and len(credentials) == 2:
            self.vendor_credentials[key] = credentials
        elif isinstance(credentials, list) and len(credentials) == 2:
            self.vendor_credentials[key] = (str(credentials[0]), str(credentials[1]))
        if "mark" in snapshot:
            self.vendor_marks[key] = str(snapshot.get("mark") or "")
        self._restore_app_internal_state(key, snapshot.get("app_state") or {})
        app = self._active_vendor_app(key)
        old_suspended = getattr(app, "suspend_customer_guard_trace", False) if app is not None else False
        if app is not None and hasattr(app, "suspend_customer_guard_trace"):
            try:
                setattr(app, "suspend_customer_guard_trace", True)
            except Exception:
                pass
        try:
            self._restore_widget_tree(self.vendor_pages.get(key), snapshot.get("widgets") or {})
        finally:
            if app is not None and hasattr(app, "suspend_customer_guard_trace"):
                try:
                    setattr(app, "suspend_customer_guard_trace", old_suspended)
                except Exception:
                    pass
        if "shell_mark" in snapshot:
            self.shell_mark_var.set(str(snapshot.get("shell_mark") or ""))
        self._save_shell_credentials(key, silent_app_update=True)
        self._load_shell_credentials(key)

    def _reload_vendor_modules_if_source(self) -> None:
        """重载磁盘上的 common 和三家云模块；外置源码版 exe 也能热更新。"""
        global _shared, _huawei_app, _huawei_cloud_client, _volc_app, ctyun_workflow_gui
        prefixes = ("common", "HUAWEI", "CTYUN", "Volcengine")
        modules: List[Tuple[int, str, Any]] = []
        for name, module in list(sys.modules.items()):
            if not module or not any(name == prefix or name.startswith(prefix + ".") for prefix in prefixes):
                continue
            # 只重载真实存在的 .py 文件；纯打包内置模块没有源码路径，会自动跳过。
            if self._module_source_mtime(module) is None:
                continue
            modules.append((prefixes.index(name.split(".", 1)[0]), name, module))
        for _prefix_index, _name, module in sorted(modules, key=lambda item: (item[0], item[1].count("."), self._module_reload_priority(item[1]), item[1])):
            importlib.reload(module)
        _shared = sys.modules.get("common.shared", _shared)
        _huawei_cloud_client = sys.modules.get("HUAWEI.cloud_client", _huawei_cloud_client)
        _huawei_app = sys.modules.get("HUAWEI.app", _huawei_app)
        _volc_app = sys.modules.get("Volcengine.app", _volc_app)
        ctyun_workflow_gui = sys.modules.get("CTYUN.app", ctyun_workflow_gui)
        import_legacy_names(globals(), _shared)
        _modules = [module for module in (_huawei_cloud_client, _huawei_app, _volc_app) if module is not None]
        if _modules:
            import_legacy_names(globals(), *_modules)
        globals()["APP_TITLE"] = _shared.APP_TITLE
        if self._shell_bindings_need_rebuild():
            self._reset_shell_for_interface_update()

    @staticmethod
    def _module_reload_priority(name: str) -> int:
        # 中文注释：页面模块通常会从同云的工具模块导入常量/函数，热更新时必须先重载依赖再重载 app。
        priorities = {
            "HUAWEI.cloud_client": -20,
            "HUAWEI.ros_builder": -20,
            "HUAWEI.app": 50,
            "CTYUN.app": 50,
            "Volcengine.app": 50,
        }
        return priorities.get(name, 0)

    def _shell_bindings_need_rebuild(self) -> bool:
        """判断顶部栏/侧边栏是否仍绑定旧类方法。"""
        factories = getattr(self, "vendor_factories", None)
        if not isinstance(factories, dict):
            return True
        factory_names = {
            "huawei": "_create_huawei_page",
            "ctyun": "_create_ctyun_page",
            "volc": "_create_volc_page",
        }
        check_key = self.available_vendors[0] if self.available_vendors else "huawei"
        current_factory = factories.get(check_key)
        expected_factory = getattr(self.__class__, factory_names.get(check_key, "_create_huawei_page"), None)
        return getattr(current_factory, "__func__", None) is not expected_factory

    def _reset_shell_for_interface_update(self) -> None:
        """重建主窗口壳体，让顶部凭证栏、左侧菜单这类界面改动立即生效。"""
        for child in list(self.winfo_children()):
            if isinstance(child, tk.Toplevel):
                continue
            try:
                child.destroy()
            except tk.TclError:
                pass
        for attr in ("huawei_app", "ctyun_gui", "volc_app"):
            if hasattr(self, attr):
                try:
                    delattr(self, attr)
                except Exception:
                    pass
        self.vendor_pages = {}
        self.vendor_buttons = {}
        self.vendor_button_images = {}
        self.vendor_progressbars = {}
        self.vendor_progress_active = {}
        self.vendor_progress_jobs = {}
        self.vendor_progress_phase = {}
        self.vendor_loading = {}
        self.current_vendor = ""
        self._configure_style()
        self._build_shell()
        self._shell_rebuilt_during_update = True

    def _reload_launcher_module_if_source(self) -> bool:
        """界面更新时重载主窗口模块，并让当前窗口切到新类方法。"""
        module_name = self.__class__.__module__
        module = sys.modules.get(module_name)
        if module is None or self._module_source_mtime(module) is None:
            return False
        refreshed = importlib.reload(module)
        refreshed_class = getattr(refreshed, "MultiCloudManagerApp", None)
        if isinstance(refreshed_class, type) and refreshed_class is not self.__class__:
            self.__class__ = refreshed_class
            return True
        return False

    def _refresh_interface_from_menu(self) -> None:
        try:
            cleanup_stale_task_locks()
            current = self.current_vendor or "huawei"
            try:
                self._save_shell_credentials(current, silent_app_update=True)
            except Exception:
                pass
            if self._active_task_running():
                self._refresh_vendor_layout()
                self._update_global_status()
                self._show_message("界面已刷新。当前有云任务运行中，为避免影响任务，任务结束后再点“界面更新”可重新载入页面逻辑。")
                return
            self._shell_rebuilt_during_update = False
            snapshot = self._snapshot_current_vendor_state(current)
            launcher_reloaded = self._reload_launcher_module_if_source()
            self._reload_vendor_modules_if_source()
            if not getattr(self, "_shell_rebuilt_during_update", False):
                page = self.vendor_pages.pop(current, None)
                if page is not None:
                    try:
                        page.destroy()
                    except tk.TclError:
                        pass
                self.vendor_loading.pop(current, None)
            attr_by_vendor = {"huawei": "huawei_app", "ctyun": "ctyun_gui", "volc": "volc_app"}
            attr_name = attr_by_vendor.get(current)
            if attr_name:
                if hasattr(self, attr_name):
                    try:
                        delattr(self, attr_name)
                    except Exception:
                        pass
            self.current_vendor = ""
            self.show_vendor(current)
            self._restore_current_vendor_state(current, snapshot)
            self._refresh_vendor_layout()
            self._update_global_status()
            self._interface_update_mtimes = self._collect_interface_update_mtimes()
            mode = "界面模块已热加载" if launcher_reloaded else "界面已重新加载"
            self._show_message(f"{mode}，当前版本：{APP_VERSION}")
        except Exception as exc:
            self._show_message(f"界面更新失败：{exc}", title="界面更新")

    def _refresh_vendor_layout(self, page: Optional[ttk.Frame] = None) -> None:
        try:
            target = page or self.vendor_pages.get(self.current_vendor)
            if target is not None:
                target.grid(row=0, column=0, sticky="nsew")
                target.tkraise()
            self.page_host.update_idletasks()
            self.update_idletasks()
        except tk.TclError:
            pass

    def _add_vendor_button(self, sidebar: tk.Frame, key: str, title: str, subtitle: str, logo_name: str, row: int) -> None:
        image = self._load_photo(logo_name)
        box = tk.Frame(sidebar, bg=PANEL_BG)
        box.grid(row=row, column=0, sticky="ew", padx=8, pady=(0, 18))
        box.columnconfigure(0, weight=1)
        button = tk.Button(
            box,
            text=title if not subtitle else f"{title}\n{subtitle}",
            image=image,
            compound=tk.TOP,
            anchor="center",
            justify="center",
            bg=CONTENT_BG,
            fg=TEXT_FG,
            activebackground=ACCENT_BG,
            activeforeground=TEXT_FG,
            relief="flat",
            bd=0,
            highlightthickness=1,
            highlightbackground=BORDER_COLOR,
            padx=2,
            pady=4,
            font=(UI_FONT_FAMILY, 10),
            command=lambda: self.show_vendor(key),
        )
        button.grid(row=0, column=0, sticky="ew")
        progress_slot = tk.Frame(box, bg=PANEL_BG, height=13)
        progress_slot.grid(row=1, column=0, sticky="ew", padx=5, pady=(0, 0))
        progress_slot.grid_propagate(False)
        progress = tk.Canvas(
            progress_slot,
            bg=PANEL_BG,
            highlightthickness=0,
            bd=0,
            width=74,
            height=13,
        )
        self.vendor_buttons[key] = button
        self.vendor_progressbars[key] = progress
        if image is not None:
            self.vendor_button_images[key] = image

    def _add_sidebar_action_button(self, sidebar: tk.Frame, title: str, logo_name: str, row: int, command: Callable[[], None]) -> None:
        image = self._load_photo(logo_name)
        box = tk.Frame(sidebar, bg=PANEL_BG)
        box.grid(row=row, column=0, sticky="ew", padx=8, pady=(0, 18))
        box.columnconfigure(0, weight=1)
        button = tk.Button(
            box,
            text=title,
            image=image,
            compound=tk.TOP,
            anchor="center",
            justify="center",
            bg=CONTENT_BG,
            fg=TEXT_FG,
            activebackground=ACCENT_BG,
            activeforeground=TEXT_FG,
            relief="flat",
            bd=0,
            highlightthickness=1,
            highlightbackground=BORDER_COLOR,
            padx=2,
            pady=4,
            font=(UI_FONT_FAMILY, 10),
            command=command,
        )
        button.grid(row=0, column=0, sticky="ew")
        if image is not None:
            self.images[f"sidebar:{logo_name}"] = image

    def _draw_round_rect(self, canvas: tk.Canvas, x1: int, y1: int, x2: int, y2: int, radius: int, **kwargs: Any) -> None:
        radius = max(1, min(radius, (x2 - x1) // 2, (y2 - y1) // 2))
        canvas.create_rectangle(x1 + radius, y1, x2 - radius, y2, **kwargs)
        canvas.create_rectangle(x1, y1 + radius, x2, y2 - radius, **kwargs)
        canvas.create_oval(x1, y1, x1 + radius * 2, y1 + radius * 2, **kwargs)
        canvas.create_oval(x2 - radius * 2, y1, x2, y1 + radius * 2, **kwargs)

    def _draw_vendor_progress(self, key: str) -> None:
        progress = self.vendor_progressbars.get(key)
        if progress is None:
            return
        progress.delete("all")
        width = int(progress.winfo_width() or 74)
        height = int(progress.winfo_height() or 13)
        self._draw_round_rect(progress, 1, 1, width - 1, height - 1, 6, fill="#d1d5db", outline="")
        phase = int(self.vendor_progress_phase.get(key, 0))
        cycle = 64
        offset = phase % (cycle * 2)
        ratio = (offset / cycle) if offset <= cycle else ((cycle * 2 - offset) / cycle)
        block_width = max(18, min(24, width // 3))
        block_height = max(7, height - 4)
        travel = max(1, width - block_width - 4)
        x = 2 + ratio * travel
        y = (height - block_height) / 2
        self._draw_round_rect(progress, x, y, x + block_width, y + block_height, 4, fill="#22c55e", outline="")

    def _animate_vendor_progress(self, key: str) -> None:
        if not self.vendor_progress_active.get(key):
            return
        self.vendor_progress_phase[key] = int(self.vendor_progress_phase.get(key, 0)) + 3
        self._draw_vendor_progress(key)
        try:
            job = self.after(35, lambda k=key: self._animate_vendor_progress(k))
            self.vendor_progress_jobs[key] = str(job)
        except tk.TclError:
            pass

    def _set_vendor_progress(self, key: str, active: bool) -> None:
        progress = self.vendor_progressbars.get(key)
        if progress is None:
            return
        if bool(self.vendor_progress_active.get(key)) == bool(active):
            return
        self.vendor_progress_active[key] = bool(active)
        if active:
            self.vendor_progress_phase[key] = 0
            progress.place(relx=0.5, y=0, anchor="n", width=74, height=13)
            self._animate_vendor_progress(key)
        else:
            job = self.vendor_progress_jobs.pop(key, "")
            if job:
                try:
                    self.after_cancel(job)
                except tk.TclError:
                    pass
            progress.delete("all")
            progress.place_forget()

    def _credential_history_vendor(self, fixed_vendor: Optional[str] = None) -> str:
        return (fixed_vendor or self.current_vendor or "").strip()

    def _refresh_credential_history_combo(self, combo: Any, value_var: tk.StringVar, fixed_vendor: Optional[str] = None) -> None:
        vendor = self._credential_history_vendor(fixed_vendor)
        records: Dict[str, Dict[str, str]] = {}
        if vendor:
            try:
                history_items = list(credential_history_for_cloud(vendor))
                mark_counts: Dict[str, int] = {}
                if vendor == "ctyun":
                    for item in history_items:
                        mark = str(item.get("mark") or "").strip()
                        if mark:
                            mark_counts[mark] = mark_counts.get(mark, 0) + 1
                for item in history_items:
                    mark = str(item.get("mark") or "").strip()
                    ak = str(item.get("ak") or "").strip()
                    sk = str(item.get("sk") or "").strip()
                    label = mark
                    if vendor == "ctyun" and mark_counts.get(mark, 0) > 1:
                        label = f"{mark}（AK:{ak[-6:]}）"
                        base_label = label
                        suffix = 2
                        while label in records:
                            label = f"{base_label}-{suffix}"
                            suffix += 1
                    if not mark or not ak or not sk or label in records:
                        continue
                    records[label] = {"mark": mark, "ak": ak, "sk": sk}
            except Exception:
                records = {}
        try:
            setattr(combo, "_credential_history_records", records)
            if value_var.get().strip() not in records:
                value_var.set(CREDENTIAL_HISTORY_PLACEHOLDER)
        except tk.TclError:
            pass

    def _refresh_credential_history_controls(self, vendor: Optional[str] = None) -> None:
        live_controls: List[Dict[str, Any]] = []
        for control in self.credential_history_controls:
            combo = control.get("combo")
            value_var = control.get("var")
            fixed_vendor = control.get("vendor")
            if not isinstance(value_var, tk.StringVar):
                continue
            try:
                if combo is None or not combo.winfo_exists():
                    continue
            except tk.TclError:
                continue
            live_controls.append(control)
            if vendor and fixed_vendor and fixed_vendor != vendor:
                continue
            if vendor and not fixed_vendor and self.current_vendor != vendor:
                continue
            self._refresh_credential_history_combo(combo, value_var, fixed_vendor)
        self.credential_history_controls = live_controls

    def _build_credential_history_row(
        self,
        parent: ttk.Frame,
        *,
        vendor: Optional[str],
        app: Any,
        parent_window: tk.Misc,
        ak_var: tk.StringVar,
        sk_var: tk.StringVar,
        mark_var: tk.StringVar,
        title_var: Optional[tk.StringVar] = None,
        balance_var: Optional[tk.StringVar] = None,
    ) -> tk.Frame:
        row = ttk.Frame(parent, style="Content.TFrame")
        row.grid(row=2, column=1, sticky="ew", padx=(0, 10))
        row.rowconfigure(0, minsize=27)
        row.columnconfigure(0, weight=0, minsize=232)
        row.columnconfigure(1, weight=1)
        row.columnconfigure(2, weight=0, minsize=90)
        row.columnconfigure(3, weight=1)
        row.columnconfigure(4, weight=0, minsize=190)
        mark_entry = ttk.Entry(row, textvariable=mark_var, width=1)
        mark_entry.grid(row=0, column=0, sticky="ew", padx=(0, 6))
        balance_box = None
        if balance_var is not None:
            balance_box = self._build_balance_display(
                row,
                balance_var,
                width=90,
                height=27,
                label_y=1,
                label_height=25,
                anchor="center",
            )
        else:
            balance_box = tk.Frame(row, width=90, height=27, bg=CONTENT_BG)
            balance_box.grid_propagate(False)
        balance_box.grid(row=0, column=2, sticky="ew", padx=(0, 6))
        history_var = tk.StringVar()
        history_box = tk.Frame(row, width=190, height=22, bg=BORDER_COLOR)
        history_box.grid_propagate(False)
        setattr(history_box, "_credential_history_popup_width", 260)
        history_box.grid(row=0, column=4, sticky="e")
        history_entry = tk.Entry(
            history_box,
            textvariable=history_var,
            width=31,
            bd=0,
            relief=tk.FLAT,
            bg=TEXT_BG,
            fg=TEXT_FG,
            font=(UI_FONT_FAMILY, 10),
            highlightthickness=0,
        )
        history_entry.place(x=1, y=1, width=168, height=20)
        open_command = lambda: self._open_credential_history_popup(
            fixed_vendor=vendor,
            anchor=history_box,
            history_widget=history_entry,
            value_var=history_var,
            app=app,
            parent_window=parent_window,
            ak_var=ak_var,
            sk_var=sk_var,
            mark_var=mark_var,
            title_var=title_var,
        )
        history_button = tk.Button(
            history_box,
            text="\u25be",
            width=3,
            command=open_command,
            bd=0,
            relief=tk.FLAT,
            bg=TEXT_BG,
            fg=MUTED_FG,
            activebackground=BUTTON_ACTIVE_BG,
            activeforeground=TEXT_FG,
            font=(UI_FONT_FAMILY, 9),
            highlightthickness=0,
            takefocus=False,
            cursor="hand2",
        )
        history_button.place(x=169, y=1, width=20, height=20)
        history_entry.bind("<Button-1>", lambda _event: open_command())
        history_entry.bind("<Key>", lambda _event: "break")
        setattr(balance_box, "_credential_history_row", row)
        setattr(balance_box, "_credential_history_mark_entry", mark_entry)
        if balance_var is not None:
            setattr(balance_box, "_credential_history_balance_var", balance_var)
            trace_id = balance_var.trace_add(
                "write",
                lambda *_args, box=balance_box: self._refresh_credential_history_balance_width(box),
            )
            setattr(balance_box, "_credential_history_balance_trace", trace_id)
        show_balance = balance_var is not None and (vendor is None or self._balance_supported(vendor))
        self._set_credential_history_balance_visible(balance_box, show_balance)
        self.credential_history_controls.append({"vendor": vendor, "combo": history_entry, "var": history_var})
        self._refresh_credential_history_combo(history_entry, history_var, vendor)
        return balance_box

    def _credential_history_balance_width(self, balance_box: tk.Frame) -> int:
        balance_var = getattr(balance_box, "_credential_history_balance_var", None)
        text = balance_var.get() if isinstance(balance_var, tk.StringVar) else ""
        try:
            width = tkfont.Font(family=UI_FONT_FAMILY, size=10).measure(text) + 14
        except tk.TclError:
            width = len(text) * 8 + 14
        return max(90, min(158, width))

    def _set_balance_display_width(self, balance_box: tk.Frame, width: int) -> None:
        balance_box.configure(width=width)
        label = getattr(balance_box, "_balance_display_label", None)
        if label is not None:
            try:
                label.place_configure(width=width)
            except tk.TclError:
                pass

    def _refresh_credential_history_balance_width(self, balance_box: tk.Frame) -> None:
        if not getattr(balance_box, "_credential_history_balance_visible", False):
            return
        self._set_credential_history_balance_visible(balance_box, True)

    def _set_credential_history_balance_visible(self, balance_box: tk.Frame, visible: bool) -> None:
        row = getattr(balance_box, "_credential_history_row", None)
        mark_entry = getattr(balance_box, "_credential_history_mark_entry", None)
        if row is None or mark_entry is None:
            return
        setattr(balance_box, "_credential_history_balance_visible", visible)
        if visible:
            balance_width = self._credential_history_balance_width(balance_box)
            mark_column_width = max(164, 322 - balance_width)
            self._set_balance_display_width(balance_box, balance_width)
            row.columnconfigure(0, weight=0, minsize=mark_column_width)
            row.columnconfigure(1, weight=1, minsize=0)
            row.columnconfigure(2, weight=0, minsize=balance_width)
            row.columnconfigure(3, weight=1, minsize=0)
            row.columnconfigure(4, weight=0, minsize=190)
            mark_entry.grid_configure(row=0, column=0, columnspan=1, sticky="ew", padx=(0, 6))
            balance_box.grid(row=0, column=2, sticky="ew", padx=(0, 6))
            return
        # Keep vendors without a balance API aligned with the full credential row.
        row.columnconfigure(0, weight=1, minsize=328)
        row.columnconfigure(1, weight=0, minsize=0)
        row.columnconfigure(2, weight=0, minsize=0)
        row.columnconfigure(3, weight=0, minsize=0)
        row.columnconfigure(4, weight=0, minsize=190)
        balance_box.grid_remove()
        mark_entry.grid_configure(row=0, column=0, columnspan=4, sticky="ew", padx=(0, 6))

    def _is_widget_inside(self, ancestor: tk.Misc, widget: Optional[tk.Misc]) -> bool:
        while widget is not None:
            if widget is ancestor:
                return True
            try:
                parent_name = widget.winfo_parent()
                if not parent_name:
                    return False
                widget = widget._nametowidget(parent_name)
            except (KeyError, tk.TclError):
                return False
        return False

    def _close_credential_history_popup(self) -> None:
        bind_info = self._credential_history_outside_bind
        self._credential_history_outside_bind = None
        if bind_info is not None:
            bind_widget, bind_id = bind_info
            try:
                bind_widget.unbind("<Button-1>", bind_id)
            except tk.TclError:
                pass
        popup = self._credential_history_popup
        self._credential_history_popup = None
        try:
            if popup is not None and popup.winfo_exists():
                popup.destroy()
        except tk.TclError:
            pass

    def _open_credential_history_popup(
        self,
        *,
        fixed_vendor: Optional[str],
        anchor: tk.Misc,
        history_widget: Any,
        value_var: tk.StringVar,
        app: Any,
        parent_window: tk.Misc,
        ak_var: tk.StringVar,
        sk_var: tk.StringVar,
        mark_var: tk.StringVar,
        title_var: Optional[tk.StringVar] = None,
    ) -> None:
        self._refresh_credential_history_combo(history_widget, value_var, fixed_vendor)
        records = getattr(history_widget, "_credential_history_records", {})
        if not isinstance(records, dict):
            records = {}
        values = list(records.keys())
        self._close_credential_history_popup()

        popup = tk.Toplevel(parent_window)
        self._credential_history_popup = popup
        popup.withdraw()
        popup.overrideredirect(True)
        popup.configure(bg=BORDER_COLOR)
        try:
            popup.transient(parent_window)
        except tk.TclError:
            pass

        popup_border = tk.Frame(popup, bg=BORDER_COLOR)
        popup_border.pack(fill=tk.BOTH, expand=True)
        frame = ttk.Frame(popup_border, style="Content.TFrame", padding=(6, 6, 6, 6))
        frame.pack(fill=tk.BOTH, expand=True, padx=1, pady=1)
        search_row = ttk.Frame(frame, style="Content.TFrame")
        search_row.pack(fill=tk.X)
        tk.Label(search_row, text="搜索：", bg=CONTENT_BG, fg=TEXT_FG, font=(UI_FONT_FAMILY, 10)).pack(side=tk.LEFT, padx=(0, 4))
        search_var = tk.StringVar()
        search_entry = ttk.Entry(search_row, textvariable=search_var)
        search_entry.pack(side=tk.LEFT, fill=tk.X, expand=True)
        list_frame = ttk.Frame(frame, style="Content.TFrame")
        list_frame.pack(fill=tk.BOTH, expand=True, pady=(6, 0))
        list_frame.columnconfigure(0, weight=1)
        list_frame.rowconfigure(0, weight=1)
        visible_rows = max(3, min(8, len(values) or 3))
        list_canvas = tk.Canvas(
            list_frame,
            height=visible_rows * 28,
            bg=TEXT_BG,
            bd=0,
            highlightthickness=1,
            highlightbackground=BORDER_COLOR,
        )
        list_canvas.grid(row=0, column=0, sticky="nsew")
        scrollbar = ttk.Scrollbar(list_frame, orient=tk.VERTICAL, command=list_canvas.yview)
        scrollbar.grid(row=0, column=1, sticky="ns")
        list_canvas.configure(yscrollcommand=scrollbar.set)
        rows_frame = tk.Frame(list_canvas, bg=TEXT_BG)
        rows_window = list_canvas.create_window((0, 0), window=rows_frame, anchor="nw")
        filtered_values: List[str] = []

        def resize_rows(event: tk.Event) -> None:
            try:
                list_canvas.itemconfigure(rows_window, width=event.width)
            except tk.TclError:
                pass

        def update_scroll_region(*_args: Any) -> None:
            try:
                list_canvas.configure(scrollregion=list_canvas.bbox("all"))
            except tk.TclError:
                pass

        list_canvas.bind("<Configure>", resize_rows)
        rows_frame.bind("<Configure>", update_scroll_region)
        bind_canvas_mousewheel(list_canvas, popup, frame, search_row, search_entry, list_frame, rows_frame)

        def choose_value(selected_value: str) -> None:
            value_var.set(selected_value)
            self._apply_selected_credential_history(
                fixed_vendor=fixed_vendor,
                combo=history_widget,
                value_var=value_var,
                app=app,
                parent_window=parent_window,
                ak_var=ak_var,
                sk_var=sk_var,
                mark_var=mark_var,
                title_var=title_var,
            )
            value_var.set(CREDENTIAL_HISTORY_PLACEHOLDER)
            self._close_credential_history_popup()

        def delete_value(selected_value: str) -> str:
            nonlocal records, values
            record = records.get(selected_value) if isinstance(records, dict) else None
            if not isinstance(record, dict):
                return "break"
            vendor = self._credential_history_vendor(fixed_vendor)
            deleted = delete_credential_history_record(
                vendor,
                str(record.get("mark") or selected_value),
                str(record.get("ak") or ""),
                str(record.get("sk") or ""),
            )
            if not deleted:
                return "break"
            self._refresh_credential_history_combo(history_widget, value_var, fixed_vendor)
            records = getattr(history_widget, "_credential_history_records", {})
            if not isinstance(records, dict):
                records = {}
            values = list(records.keys())
            value_var.set(CREDENTIAL_HISTORY_PLACEHOLDER)
            self._refresh_credential_history_controls(vendor)
            refresh_list()
            return "break"

        def refresh_list(*_args: Any) -> None:
            nonlocal filtered_values
            keyword = search_var.get().strip().lower()
            filtered_values = [item for item in values if keyword in item.lower()]
            for child in rows_frame.winfo_children():
                child.destroy()
            if not filtered_values:
                empty_text = "无匹配账号" if keyword else "暂无历史账号"
                tk.Label(
                    rows_frame,
                    text=empty_text,
                    anchor="w",
                    bg=TEXT_BG,
                    fg=MUTED_FG,
                    font=(UI_FONT_FAMILY, 10),
                ).pack(fill=tk.X, padx=8, pady=6)
                update_scroll_region()
                return
            def draw_delete_icon(canvas: tk.Canvas, hovered: bool = False) -> None:
                bg_color = "#fff1f2" if hovered else "#ffffff"
                border_color = "#fb7185" if hovered else "#ef4444"
                line_color = "#b91c1c" if hovered else "#dc2626"
                canvas.configure(bg=bg_color)
                canvas.delete("all")
                image = tk.PhotoImage(width=19, height=19)
                image.put(bg_color, to=(0, 0, 19, 19))
                for pos in range(19):
                    image.put(border_color, to=(pos, 0, pos + 1, 1))
                    image.put(border_color, to=(pos, 18, pos + 1, 19))
                    image.put(border_color, to=(0, pos, 1, pos + 1))
                    image.put(border_color, to=(18, pos, 19, pos + 1))
                for pos in range(4, 15):
                    image.put(line_color, to=(pos, pos, pos + 1, pos + 1))
                    mirror_x = 18 - pos
                    image.put(line_color, to=(mirror_x, pos, mirror_x + 1, pos + 1))
                canvas.create_image(0, 0, anchor=tk.NW, image=image)
                setattr(canvas, "_delete_icon_image", image)

            def set_row_hover(row_widget: tk.Frame, label_widget: tk.Label, box_widget: tk.Frame, base_bg: str, hovered: bool = False) -> None:
                bg_color = "#eef6ff" if hovered else base_bg
                try:
                    row_widget.configure(bg=bg_color)
                    label_widget.configure(bg=bg_color, fg="#0f172a" if hovered else TEXT_FG)
                    box_widget.configure(bg=bg_color)
                except tk.TclError:
                    pass

            for index, item in enumerate(filtered_values):
                row_bg = TEXT_BG if index % 2 == 0 else "#f8fafc"
                row = tk.Frame(rows_frame, bg=row_bg)
                row.pack(fill=tk.X)
                row.columnconfigure(0, weight=1)
                row.rowconfigure(0, weight=1)
                label = tk.Label(
                    row,
                    text=item,
                    anchor="w",
                    bg=row_bg,
                    fg=TEXT_FG,
                    font=(UI_FONT_FAMILY, 10),
                    padx=6,
                    pady=4,
                    cursor="hand2",
                )
                label.grid(row=0, column=0, sticky="ew")
                delete_box = tk.Frame(row, width=19, height=19, bg=row_bg)
                delete_box.grid(row=0, column=1, sticky="e", padx=(4, 4), pady=0)
                delete_box.grid_propagate(False)
                delete_button = tk.Canvas(
                    delete_box,
                    width=19,
                    height=19,
                    bd=0,
                    highlightthickness=0,
                    bg=row_bg,
                    cursor="hand2",
                )
                draw_delete_icon(delete_button)
                delete_button.bind("<Button-1>", lambda _event, value=item: delete_value(value))
                delete_button.bind("<Enter>", lambda _event, canvas=delete_button: draw_delete_icon(canvas, True))
                delete_button.bind("<Leave>", lambda _event, canvas=delete_button: draw_delete_icon(canvas, False))
                delete_button.place(x=0, y=0, width=19, height=19)
                bind_canvas_mousewheel(list_canvas, row, label, delete_box, delete_button)
                row.bind("<Button-1>", lambda _event, value=item: choose_value(value))
                label.bind("<Button-1>", lambda _event, value=item: choose_value(value))
                row.bind("<Enter>", lambda _event, row_widget=row, label_widget=label, box_widget=delete_box, base_bg=row_bg: set_row_hover(row_widget, label_widget, box_widget, base_bg, True))
                row.bind("<Leave>", lambda _event, row_widget=row, label_widget=label, box_widget=delete_box, base_bg=row_bg: set_row_hover(row_widget, label_widget, box_widget, base_bg, False))
                label.bind("<Enter>", lambda _event, row_widget=row, label_widget=label, box_widget=delete_box, base_bg=row_bg: set_row_hover(row_widget, label_widget, box_widget, base_bg, True))
                label.bind("<Leave>", lambda _event, row_widget=row, label_widget=label, box_widget=delete_box, base_bg=row_bg: set_row_hover(row_widget, label_widget, box_widget, base_bg, False))
            update_scroll_region()

        def choose_selected(*_args: Any) -> None:
            if not filtered_values:
                return
            choose_value(filtered_values[0])

        search_var.trace_add("write", refresh_list)
        search_entry.bind("<Return>", lambda _event: choose_selected())
        search_entry.bind("<Down>", lambda _event: "break")
        popup.bind("<Escape>", lambda _event: self._close_credential_history_popup())

        def close_on_parent_click(event: tk.Event) -> None:
            popup_now = self._credential_history_popup
            if popup_now is None:
                return
            widget = event.widget
            if self._is_widget_inside(popup_now, widget) or self._is_widget_inside(anchor, widget):
                return
            self._close_credential_history_popup()

        def bind_parent_click_guard() -> None:
            popup_now = self._credential_history_popup
            if popup_now is not popup:
                return
            try:
                bind_id = parent_window.bind("<Button-1>", close_on_parent_click, add="+")
                self._credential_history_outside_bind = (parent_window, bind_id)
            except tk.TclError:
                self._credential_history_outside_bind = None

        def focus_search_entry() -> None:
            if self._credential_history_popup is not popup:
                return
            try:
                popup.lift()
                search_entry.focus_set()
                search_entry.focus_force()
                search_entry.icursor(tk.END)
            except tk.TclError:
                pass

        refresh_list()
        try:
            anchor.update_idletasks()
            width = max(int(getattr(anchor, "_credential_history_popup_width", anchor.winfo_width())), 260)
            height = 36 + 28 * visible_rows + 18
            x = anchor.winfo_rootx() + anchor.winfo_width() - width
            y = anchor.winfo_rooty() + anchor.winfo_height()
            popup.geometry(f"{width}x{height}+{x}+{y}")
            popup.deiconify()
            popup.lift()
            popup.after(30, focus_search_entry)
            popup.after(120, bind_parent_click_guard)
        except tk.TclError:
            self._close_credential_history_popup()

    def _apply_selected_credential_history(
        self,
        *,
        fixed_vendor: Optional[str],
        combo: Any,
        value_var: tk.StringVar,
        app: Any,
        parent_window: tk.Misc,
        ak_var: tk.StringVar,
        sk_var: tk.StringVar,
        mark_var: tk.StringVar,
        title_var: Optional[tk.StringVar] = None,
    ) -> None:
        selected = value_var.get().strip()
        records = getattr(combo, "_credential_history_records", {})
        record = records.get(selected) if isinstance(records, dict) else None
        if not isinstance(record, dict):
            self._refresh_credential_history_combo(combo, value_var, fixed_vendor)
            return
        vendor = self._credential_history_vendor(fixed_vendor)
        ak = str(record.get("ak") or "").strip()
        sk = str(record.get("sk") or "").strip()
        mark = str(record.get("mark") or selected).strip()
        if not vendor or not ak or not sk or not mark:
            return
        target_app = app or self._active_vendor_app(vendor)
        old_syncing = self._syncing_credentials
        if parent_window is self:
            self._syncing_credentials = True
        try:
            ak_var.set(ak)
            sk_var.set(sk)
            mark_var.set(mark)
        finally:
            if parent_window is self:
                self._syncing_credentials = old_syncing
        if target_app is not None:
            self._set_app_credentials(target_app, ak, sk, mark)
        if fixed_vendor is None and vendor == self.current_vendor:
            self.vendor_credentials[vendor] = (ak, sk)
            self.vendor_marks[vendor] = mark
        if title_var is not None:
            title_var.set(self._credential_window_title(vendor, ak, mark))
        if isinstance(parent_window, tk.Toplevel):
            parent_window.title(self._credential_window_title(vendor, ak, mark))
        self._remember_mark_for_credentials(vendor, ak, sk, mark)
        if target_app is not None:
            self._validate_after_history_selection(
                vendor=vendor,
                app=target_app,
                parent=parent_window,
            )
        balance_var = self._balance_var_for_parent(parent_window)
        if balance_var is not None:
            self._schedule_balance_lookup(
                vendor=vendor,
                app=target_app,
                ak_var=ak_var,
                sk_var=sk_var,
                parent=parent_window,
                balance_var=balance_var,
                delay_ms=100,
            )

    def show_vendor(self, key: str) -> None:
        if key not in getattr(self, "vendor_factories", {}):
            key = self.available_vendors[0]
        if self.current_vendor:
            self._save_shell_credentials(self.current_vendor, silent_app_update=True)
        for page in self.vendor_pages.values():
            page.grid_remove()
        page = self._ensure_vendor_page(key, immediate=True)
        page.grid()
        page.tkraise()
        self.current_vendor = key
        titles = {
            "huawei": ("华为云管理", "购买搭建、资源查询、重装系统、退订/删除资源"),
            "ctyun": ("天翼云管理", "购买搭建、资源查询、重装系统、退订/删除资源"),
            "volc": ("火山引擎管理", "购买搭建、资源查询、重装系统、退订/删除资源"),
        }
        title, subtitle = titles.get(key, ("云厂商管理", ""))
        self.page_title_var.set(title)
        self.page_subtitle_var.set(subtitle)
        self._refresh_binding_action_labels(key)
        self._load_shell_credentials(key)
        if self._balance_supported(key):
            self._set_credential_history_balance_visible(self.main_balance_box, True)
            self._refresh_balance_display_from_cache(key, self.shell_ak_var, self.shell_sk_var, self.main_balance_var)
        else:
            self.main_balance_var.set("")
            self._set_credential_history_balance_visible(self.main_balance_box, False)
        self._refresh_credential_history_controls(key)
        for item_key, button in self.vendor_buttons.items():
            active = item_key == key
            button.configure(
                bg=ACCENT_BG if active else CONTENT_BG,
                fg=TEXT_FG,
                highlightbackground=TEXT_FG if active else BORDER_COLOR,
            )
        self._update_global_status()
        self._refresh_vendor_layout(page)

    def _active_credential_vars(self, key: Optional[str] = None) -> Tuple[Optional[tk.StringVar], Optional[tk.StringVar]]:
        vendor = key or self.current_vendor
        if vendor == "huawei" and hasattr(self, "huawei_app"):
            return self.huawei_app.ak_var, self.huawei_app.sk_var
        if vendor == "ctyun" and hasattr(self, "ctyun_gui"):
            return self.ctyun_gui.ak_var, self.ctyun_gui.sk_var
        if vendor == "volc" and hasattr(self, "volc_app"):
            return self.volc_app.ak_var, self.volc_app.sk_var
        return None, None

    def _active_mark_var(self, key: Optional[str] = None) -> Optional[tk.StringVar]:
        app = self._active_vendor_app(key or self.current_vendor)
        mark_var = getattr(app, "mark_var", None)
        return mark_var if isinstance(mark_var, tk.StringVar) else None

    def _set_app_mark(self, app: Any, mark: str) -> None:
        mark_var = getattr(app, "mark_var", None)
        if isinstance(mark_var, tk.StringVar):
            mark_var.set(mark or "")

    def _save_shell_credentials(self, key: str, *, silent_app_update: bool = False) -> None:
        ak = self.shell_ak_var.get().strip()
        sk = self.shell_sk_var.get().strip()
        mark = self.shell_mark_var.get().strip()
        self.vendor_credentials[key] = (ak, sk)
        self.vendor_marks[key] = mark
        app = self._active_vendor_app(key)
        if app is not None:
            self._set_app_credentials(app, ak, sk, mark, silent=silent_app_update, prepare=False)

    def _load_shell_credentials(self, key: str) -> None:
        ak_var, sk_var = self._active_credential_vars(key)
        saved_ak, saved_sk = self.vendor_credentials.get(key, ("", ""))
        saved_mark = self.vendor_marks.get(key, "")
        if ak_var is not None:
            saved_ak = ak_var.get()
        if sk_var is not None:
            saved_sk = sk_var.get()
        mark_var = self._active_mark_var(key)
        if mark_var is not None:
            saved_mark = mark_var.get()
        self._syncing_credentials = True
        try:
            self.shell_ak_var.set(saved_ak)
            self.shell_sk_var.set(saved_sk)
            self.shell_mark_var.set(saved_mark)
        finally:
            self._syncing_credentials = False
        self._refresh_credential_history_controls(key)

    def _sync_active_credentials(self) -> None:
        if self._syncing_credentials or not self.current_vendor:
            return
        self._save_shell_credentials(self.current_vendor)

    def _on_shell_credentials_changed(self, *_args: Any) -> None:
        if self._syncing_credentials or not self.current_vendor:
            return
        self._save_shell_credentials(self.current_vendor)
        app = self._active_vendor_app(self.current_vendor)
        if app is not None:
            self._schedule_mark_lookup(
                vendor=self.current_vendor,
                app=app,
                ak_var=self.shell_ak_var,
                sk_var=self.shell_sk_var,
                mark_var=self.shell_mark_var,
                parent=self,
            )
            self._schedule_auto_read_resources_after_credentials(
                vendor=self.current_vendor,
                app=app,
                ak_var=self.shell_ak_var,
                sk_var=self.shell_sk_var,
                parent=self,
            )
            self._schedule_balance_lookup(
                vendor=self.current_vendor,
                app=app,
                ak_var=self.shell_ak_var,
                sk_var=self.shell_sk_var,
                parent=self,
                balance_var=self.main_balance_var,
            )

    def _remember_mark_for_credentials(self, vendor: str, ak: str, sk: str, mark: str) -> None:
        # 中文注释：只保存 AK/SK 指纹对应的 Mark，不保存明文密钥；低质量数字 ID 不进入缓存。
        cleaned = self._clean_resolved_mark(mark)
        if not cleaned or self._is_low_quality_mark(cleaned):
            return
        try:
            remember_mark_for_credentials(vendor, ak, sk, cleaned)
            self._refresh_credential_history_controls(vendor)
        except Exception:
            pass

    def _remember_active_shell_mark(self) -> None:
        if self._syncing_credentials or not self.current_vendor:
            return
        self._remember_mark_for_credentials(
            self.current_vendor,
            self.shell_ak_var.get().strip(),
            self.shell_sk_var.get().strip(),
            self.shell_mark_var.get().strip(),
        )

    def paste_current_credentials(self) -> None:
        app = self._active_vendor_app(self.current_vendor)
        if app is None:
            self._show_message("当前云页面尚未加载完成。")
            return
        self._paste_credentials_into_app(
            vendor=self.current_vendor,
            app=app,
            ak_var=self.shell_ak_var,
            sk_var=self.shell_sk_var,
            mark_var=self.shell_mark_var,
            parent=self,
        )
        self._save_shell_credentials(self.current_vendor)

    def _vendor_display_name(self, vendor: str) -> str:
        names = {
            "huawei": "华为云管理",
            "ctyun": "天翼云管理",
            "volc": "火山引擎管理",
        }
        return names.get(vendor, "云管理")

    def _parse_credentials_for_vendor(self, vendor: str, text: str, app: Any) -> Tuple[str, str]:
        if vendor == "ctyun" and hasattr(app, "_extract_keys_from_text"):
            return app._extract_keys_from_text(text)
        return extract_ak_sk(text)

    def _credential_window_title(self, vendor: str, ak: str = "", mark: str = "") -> str:
        account = (mark or "").strip()
        if not account:
            suffix = (ak or "").strip()[-6:]
            account = f"AK后6位-{suffix}" if suffix else "未粘贴AK/SK"
        title_names = {
            "huawei": "华为",
            "ctyun": "天翼",
            "volc": "火山",
        }
        return f"{title_names.get(vendor, self._vendor_display_name(vendor))}|{account}"

    def _show_credential_message(self, parent: tk.Misc, message: str) -> None:
        message = friendly_popup_text(message)
        if parent is self:
            self._show_message(message)
            return
        show_app_message_dialog(parent, APP_TITLE, message, modal=True)

    def _set_app_credentials(self, app: Any, ak: str, sk: str, mark: str = "", *, silent: bool = False, prepare: bool = True) -> None:
        if silent:
            setter = getattr(app, "set_credentials_silently", None)
            if callable(setter):
                setter(ak or "", sk or "", mark or "")
                return
        ak_var = getattr(app, "ak_var", None)
        sk_var = getattr(app, "sk_var", None)
        mark_var = getattr(app, "mark_var", None)
        credentials_changed = (
            isinstance(ak_var, tk.StringVar)
            and isinstance(sk_var, tk.StringVar)
            and (ak_var.get().strip() != (ak or "").strip() or sk_var.get().strip() != (sk or "").strip())
        )
        if prepare and credentials_changed:
            prepare_func = getattr(app, "prepare_for_new_credentials", None)
            if callable(prepare_func):
                try:
                    prepare_func()
                except Exception:
                    pass
        if isinstance(ak_var, tk.StringVar) and ak_var.get() != (ak or ""):
            ak_var.set(ak or "")
        if isinstance(sk_var, tk.StringVar) and sk_var.get() != (sk or ""):
            sk_var.set(sk or "")
        if isinstance(mark_var, tk.StringVar) and mark_var.get() != (mark or ""):
            mark_var.set(mark or "")

    @staticmethod
    def _force_customer_guard_check(app: Any, delay_ms: int = 0) -> None:
        scheduler = getattr(app, "_schedule_customer_guard_check", None)
        if not callable(scheduler):
            return
        try:
            scheduler(delay_ms=delay_ms, force=True)
        except TypeError:
            try:
                scheduler(delay_ms=delay_ms)
            except TypeError:
                try:
                    scheduler(delay_ms)
                except Exception:
                    pass
            except Exception:
                pass
        except Exception:
            pass

    def _validate_after_history_selection(
        self,
        *,
        vendor: str,
        app: Any,
        parent: tk.Misc,
    ) -> None:
        self._force_customer_guard_check(app, delay_ms=0)
        try:
            parent.after(100, lambda: self._auto_read_vendor_app_resources(vendor, app, parent))
        except Exception:
            self._auto_read_vendor_app_resources(vendor, app, parent)

    def _schedule_auto_read_resources_after_credentials(
        self,
        *,
        vendor: str,
        app: Any,
        ak_var: tk.StringVar,
        sk_var: tk.StringVar,
        parent: tk.Misc,
        delay_ms: int = 900,
    ) -> None:
        target_key = f"{vendor}:{id(parent)}"
        old_job = self._auto_resource_read_jobs.pop(target_key, None)
        if old_job:
            try:
                self.after_cancel(old_job)
            except tk.TclError:
                pass
        ak = ak_var.get().strip()
        sk = sk_var.get().strip()
        if not self._credentials_ready_for_mark_lookup(ak, sk):
            return
        token = self._auto_resource_read_tokens.get(target_key, 0) + 1
        self._auto_resource_read_tokens[target_key] = token

        def start_read() -> None:
            self._auto_resource_read_jobs.pop(target_key, None)
            if self._auto_resource_read_tokens.get(target_key) != token:
                return
            if ak_var.get().strip() != ak or sk_var.get().strip() != sk:
                return
            self._auto_read_vendor_app_resources(vendor, app, parent)

        try:
            self._auto_resource_read_jobs[target_key] = self.after(delay_ms, start_read)
        except tk.TclError:
            pass

    @staticmethod
    def _clean_resolved_mark(mark: Any) -> str:
        # 中文注释：云接口可能返回中文用户名，也可能返回账号ID；这里只去掉空白和外层标点，不破坏中文。
        value = "".join(str(mark or "").split()).strip(" ,;:/\\|，；：、\"'<>[](){}")
        return value[:80]

    @staticmethod
    def _is_low_quality_mark(mark: str) -> bool:
        value = (mark or "").strip()
        if not value:
            return True
        if value.isdigit():
            return True
        return bool(re.fullmatch(r"[0-9a-fA-F-]{24,}", value))

    @staticmethod
    def _credentials_ready_for_mark_lookup(ak: str, sk: str) -> bool:
        return len((ak or "").strip()) >= 16 and len((sk or "").strip()) >= 16

    @staticmethod
    def _balance_display_for_amount(amount: Any) -> str:
        try:
            value = decimal.Decimal(str(amount).replace(",", ""))
            value = value.quantize(decimal.Decimal("0.01"), rounding=decimal.ROUND_HALF_UP)
        except (decimal.InvalidOperation, ValueError):
            return "余额: 未读取"
        text = format(value, "f").rstrip("0").rstrip(".")
        if not text:
            text = "0"
        return f"余额: {text}元"

    @staticmethod
    def _balance_cache_key(vendor: str, ak: str, sk: str) -> str:
        return f"{vendor}:{hashlib.sha256((ak + ':' + sk).encode('utf-8', errors='ignore')).hexdigest()}"

    def _balance_target_key(self, vendor: str, parent: tk.Misc) -> str:
        scope = "main" if parent is self else f"child-{id(parent)}"
        return f"{scope}:{vendor}"

    def _balance_var_for_parent(self, parent: tk.Misc) -> Optional[tk.StringVar]:
        if parent is self:
            return getattr(self, "main_balance_var", None)
        balance_var = getattr(parent, "_cloud_balance_var", None)
        return balance_var if isinstance(balance_var, tk.StringVar) else None

    @staticmethod
    def _balance_supported(vendor: str) -> bool:
        return vendor in {"huawei", "volc"}

    def _refresh_balance_display_from_cache(
        self,
        vendor: str,
        ak_var: tk.StringVar,
        sk_var: tk.StringVar,
        balance_var: tk.StringVar,
    ) -> None:
        if not self._balance_supported(vendor):
            balance_var.set("")
            return
        ak = ak_var.get().strip()
        sk = sk_var.get().strip()
        if not self._credentials_ready_for_mark_lookup(ak, sk):
            balance_var.set("余额: --")
            return
        balance_var.set(self._balance_lookup_cache.get(self._balance_cache_key(vendor, ak, sk), "余额: --"))

    def _query_available_balance_for_vendor(self, vendor: str, ak: str, sk: str) -> Any:
        if vendor == "huawei":
            if _huawei_cloud_client is None:
                raise RuntimeError("华为云余额模块未加载")
            client = _huawei_cloud_client.HuaweiCloudClient(ak, sk)
            return client.get_available_balance()
        if vendor == "ctyun":
            raise RuntimeError("天翼云余额不显示")
        if vendor == "volc":
            from Volcengine.api.volcengine.billing_service import get_available_balance

            ok, message, balance = get_available_balance(ak, sk, "cn-beijing")
            if ok and balance is not None:
                return balance
            raise RuntimeError(str(message or "火山引擎余额未读取"))
        raise RuntimeError("未知云厂商")

    @staticmethod
    def _customer_guard_balance_state(app: Any, ak: str, sk: str) -> str:
        if app is None:
            return "ok"
        fingerprint_func = getattr(app, "_customer_guard_fingerprint", None)
        if not callable(fingerprint_func):
            return "ok"
        try:
            expected = fingerprint_func(ak, sk)
        except Exception:
            return "waiting"
        current = str(getattr(app, "customer_guard_fingerprint", "") or "")
        status = str(getattr(app, "customer_guard_status", "") or "")
        if expected and current == expected and status == "ok":
            return "ok"
        if expected and current == expected and status == "blocked":
            return "blocked"
        return "waiting"

    def _schedule_balance_lookup(
        self,
        *,
        vendor: str,
        app: Any = None,
        ak_var: tk.StringVar,
        sk_var: tk.StringVar,
        parent: tk.Misc,
        balance_var: tk.StringVar,
        delay_ms: int = 900,
    ) -> None:
        target_key = self._balance_target_key(vendor, parent)
        old_job = self._balance_lookup_jobs.pop(target_key, None)
        if old_job:
            try:
                self.after_cancel(old_job)
            except tk.TclError:
                pass
        if not self._balance_supported(vendor):
            balance_var.set("")
            return
        ak = ak_var.get().strip()
        sk = sk_var.get().strip()
        if not self._credentials_ready_for_mark_lookup(ak, sk):
            balance_var.set("余额: --")
            return
        cache_key = self._balance_cache_key(vendor, ak, sk)
        cached = self._balance_lookup_cache.get(cache_key)
        if cached:
            balance_var.set(cached)
        else:
            balance_var.set("余额: 查询中")
        token = self._balance_lookup_tokens.get(target_key, 0) + 1
        self._balance_lookup_tokens[target_key] = token

        def start_lookup() -> None:
            self._balance_lookup_jobs.pop(target_key, None)
            if self._balance_lookup_tokens.get(target_key) != token:
                return
            if ak_var.get().strip() != ak or sk_var.get().strip() != sk:
                return
            guard_state = self._customer_guard_balance_state(app, ak, sk)
            if guard_state == "blocked":
                balance_var.set("余额: --")
                return
            if guard_state != "ok":
                balance_var.set("余额: 查询中")
                try:
                    self._balance_lookup_jobs[target_key] = self.after(500, start_lookup)
                except tk.TclError:
                    pass
                return

            def worker() -> None:
                success = False
                try:
                    amount = self._query_available_balance_for_vendor(vendor, ak, sk)
                    text = self._balance_display_for_amount(amount)
                    success = text != "余额: 未读取"
                except Exception as exc:
                    message = str(exc).strip()
                    text = message if message.startswith("余额:") else "余额: 未读取"

                def apply_result() -> None:
                    if self._balance_lookup_tokens.get(target_key) != token:
                        return
                    try:
                        if isinstance(parent, tk.Toplevel) and not parent.winfo_exists():
                            return
                    except tk.TclError:
                        return
                    if ak_var.get().strip() != ak or sk_var.get().strip() != sk:
                        return
                    if success:
                        self._balance_lookup_cache[cache_key] = text
                    balance_var.set(text)

                try:
                    self.after(0, apply_result)
                except tk.TclError:
                    pass

            threading.Thread(target=worker, daemon=True).start()

        try:
            self._balance_lookup_jobs[target_key] = self.after(delay_ms, start_lookup)
        except tk.TclError:
            pass

    def _mark_lookup_target_key(self, vendor: str, parent: tk.Misc) -> str:
        scope = "main" if parent is self else f"child-{id(parent)}"
        return f"{scope}:{vendor}"

    def _resolve_mark_for_vendor(self, vendor: str, app: Any, ak: str, sk: str) -> str:
        try:
            cached = self._clean_resolved_mark(cached_mark_for_credentials(vendor, ak, sk))
        except Exception:
            cached = ""
        if cached and not self._is_low_quality_mark(cached):
            return cached
        resolver = getattr(app, "resolve_mark_from_credentials", None)
        if callable(resolver):
            return self._clean_resolved_mark(resolver(ak, sk))
        if vendor == "volc":
            volc_resolver = getattr(app, "_resolve_volc_account_mark", None)
            if callable(volc_resolver):
                return self._clean_resolved_mark(volc_resolver(ak, sk))
        return ""

    def _note_mark_lookup_result(self, app: Any, message: str) -> None:
        logger = getattr(app, "log", None)
        if callable(logger):
            try:
                logger(message)
                return
            except Exception:
                pass
        emitter = getattr(app, "_emit", None)
        if callable(emitter):
            try:
                emitter(message + "\n")
            except Exception:
                pass

    def _apply_mark_lookup_result(
        self,
        *,
        vendor: str,
        app: Any,
        ak: str,
        sk: str,
        mark: str,
        token: int,
        target_key: str,
        mark_var: tk.StringVar,
        parent: tk.Misc,
        title_var: Optional[tk.StringVar] = None,
    ) -> None:
        if self._mark_lookup_tokens.get(target_key) != token:
            return
        try:
            if isinstance(parent, tk.Toplevel) and not parent.winfo_exists():
                return
        except tk.TclError:
            return
        if self._active_credentials_mismatch(app, ak, sk, parent, vendor):
            return
        current_mark = mark_var.get().strip()
        if current_mark and current_mark != mark and self._is_low_quality_mark(mark):
            return
        mark_var.set(mark)
        self._set_app_credentials(app, ak, sk, mark)
        if vendor == self.current_vendor and parent is self:
            self.vendor_credentials[vendor] = (ak, sk)
            self.vendor_marks[vendor] = mark
        title = self._credential_window_title(vendor, ak, mark)
        if title_var is not None:
            title_var.set(title)
        if isinstance(parent, tk.Toplevel):
            parent.title(title)
        self._remember_mark_for_credentials(vendor, ak, sk, mark)
        self._note_mark_lookup_result(app, f"已自动识别账号用户名到 用户名(可编辑)：{mark}")

    def _apply_direct_mark_result(
        self,
        *,
        vendor: str,
        app: Any,
        ak: str,
        sk: str,
        mark: str,
        mark_var: tk.StringVar,
        parent: tk.Misc,
        title_var: Optional[tk.StringVar] = None,
    ) -> None:
        cleaned = self._clean_resolved_mark(mark)
        if not cleaned or self._is_low_quality_mark(cleaned):
            return
        if self._active_credentials_mismatch(app, ak, sk, parent, vendor):
            return
        if mark_var.get().strip() != cleaned:
            mark_var.set(cleaned)
        self._set_app_credentials(app, ak, sk, cleaned)
        if vendor == self.current_vendor and parent is self:
            self.vendor_credentials[vendor] = (ak, sk)
            self.vendor_marks[vendor] = cleaned
        title = self._credential_window_title(vendor, ak, cleaned)
        if title_var is not None:
            title_var.set(title)
        if isinstance(parent, tk.Toplevel):
            parent.title(title)
        self._remember_mark_for_credentials(vendor, ak, sk, cleaned)

    def _apply_cached_mark_after_paste(
        self,
        *,
        vendor: str,
        app: Any,
        ak: str,
        sk: str,
        mark_var: tk.StringVar,
        parent: tk.Misc,
        title_var: Optional[tk.StringVar] = None,
    ) -> bool:
        try:
            cached = self._clean_resolved_mark(cached_mark_for_credentials(vendor, ak, sk))
        except Exception:
            return False
        if not cached or self._is_low_quality_mark(cached):
            return False
        self._apply_direct_mark_result(
            vendor=vendor,
            app=app,
            ak=ak,
            sk=sk,
            mark=cached,
            mark_var=mark_var,
            parent=parent,
            title_var=title_var,
        )
        if mark_var.get().strip() != cached:
            return False
        self._note_mark_lookup_result(app, "已从本地缓存读取账号用户名到 用户名(可编辑)。")
        return True

    def _active_credentials_mismatch(self, app: Any, ak: str, sk: str, parent: tk.Misc, vendor: str) -> bool:
        if parent is self and vendor == self.current_vendor:
            return self.shell_ak_var.get().strip() != ak or self.shell_sk_var.get().strip() != sk
        app_ak = getattr(app, "ak_var", None)
        app_sk = getattr(app, "sk_var", None)
        if isinstance(app_ak, tk.StringVar) and app_ak.get().strip() != ak:
            return True
        if isinstance(app_sk, tk.StringVar) and app_sk.get().strip() != sk:
            return True
        return False

    def _schedule_mark_lookup(
        self,
        *,
        vendor: str,
        app: Any,
        ak_var: tk.StringVar,
        sk_var: tk.StringVar,
        mark_var: tk.StringVar,
        parent: tk.Misc,
        title_var: Optional[tk.StringVar] = None,
        delay_ms: int = 800,
    ) -> None:
        ak = ak_var.get().strip()
        sk = sk_var.get().strip()
        target_key = self._mark_lookup_target_key(vendor, parent)
        old_job = self._mark_lookup_jobs.pop(target_key, None)
        if old_job:
            try:
                self.after_cancel(old_job)
            except tk.TclError:
                pass
        if not self._credentials_ready_for_mark_lookup(ak, sk):
            return
        token = self._mark_lookup_tokens.get(target_key, 0) + 1
        self._mark_lookup_tokens[target_key] = token

        def start_lookup() -> None:
            self._mark_lookup_jobs.pop(target_key, None)

            def worker() -> None:
                try:
                    mark = self._resolve_mark_for_vendor(vendor, app, ak, sk)
                except Exception:
                    # 中文注释：自动识别 Mark 只是辅助功能，失败不能影响用户继续使用 AK/SK。
                    return
                if not mark:
                    return
                self.after(
                    0,
                    lambda: self._apply_mark_lookup_result(
                        vendor=vendor,
                        app=app,
                        ak=ak,
                        sk=sk,
                        mark=mark,
                        token=token,
                        target_key=target_key,
                        mark_var=mark_var,
                        parent=parent,
                        title_var=title_var,
                    ),
                )

            threading.Thread(target=worker, daemon=True).start()

        self._mark_lookup_jobs[target_key] = self.after(delay_ms, start_lookup)

    def _apply_resolved_credentials_into_app(
        self,
        *,
        vendor: str,
        app: Any,
        ak_var: tk.StringVar,
        sk_var: tk.StringVar,
        mark_var: tk.StringVar,
        parent: tk.Misc,
        title_var: Optional[tk.StringVar] = None,
        ak: str,
        sk: str,
        mark: str = "",
        success_note: str = "",
    ) -> None:
        # 先让云模块看到“旧凭证 -> 新凭证”的变化并清空界面，再写入新值。
        # 之前先改共享 StringVar 会让模块误判凭证未变化，旧账号结果可能短暂串到新账号。
        self._set_app_credentials(app, ak, sk, mark or "")
        ak_var.set(ak)
        sk_var.set(sk)
        if mark:
            mark_var.set(mark)
            self._remember_mark_for_credentials(vendor, ak, sk, mark)
        else:
            mark_var.set("")
        if vendor == self.current_vendor and parent is self:
            self.vendor_credentials[vendor] = (ak, sk)
            self.vendor_marks[vendor] = mark_var.get().strip()
        title = self._credential_window_title(vendor, ak, mark_var.get())
        if title_var is not None:
            title_var.set(title)
        if isinstance(parent, tk.Toplevel):
            parent.title(title)
        if not mark_var.get().strip():
            self._apply_cached_mark_after_paste(
                vendor=vendor,
                app=app,
                ak=ak,
                sk=sk,
                mark_var=mark_var,
                parent=parent,
                title_var=title_var,
            )
        direct_lookup_started = False
        direct_populator = getattr(app, "populate_mark_from_credentials_after_paste", None)
        if callable(direct_populator) and not mark_var.get().strip():
            try:
                direct_populator(
                    ak,
                    sk,
                    mark_var=mark_var,
                    on_mark=lambda value, v=vendor, a=app, ak_value=ak, sk_value=sk, var=mark_var, p=parent, tv=title_var: self._apply_direct_mark_result(
                        vendor=v,
                        app=a,
                        ak=ak_value,
                        sk=sk_value,
                        mark=value,
                        mark_var=var,
                        parent=p,
                        title_var=tv,
                    ),
                )
                direct_lookup_started = True
            except Exception:
                pass
        if not direct_lookup_started and not mark_var.get().strip():
            self._schedule_mark_lookup(
                vendor=vendor,
                app=app,
                ak_var=ak_var,
                sk_var=sk_var,
                mark_var=mark_var,
                parent=parent,
                title_var=title_var,
                delay_ms=100,
            )
        if success_note:
            self._note_mark_lookup_result(app, success_note)
        balance_var = self._balance_var_for_parent(parent)
        if balance_var is not None:
            self._schedule_balance_lookup(
                vendor=vendor,
                app=app,
                ak_var=ak_var,
                sk_var=sk_var,
                parent=parent,
                balance_var=balance_var,
                delay_ms=100,
            )
        parent.after(100, lambda: self._auto_read_vendor_app_resources(vendor, app, parent))

    def _paste_credentials_into_app(
        self,
        *,
        vendor: str,
        app: Any,
        ak_var: tk.StringVar,
        sk_var: tk.StringVar,
        mark_var: tk.StringVar,
        parent: tk.Misc,
        title_var: Optional[tk.StringVar] = None,
    ) -> None:
        try:
            text = parent.clipboard_get()
        except tk.TclError:
            text = ""
        if not text:
            self._show_credential_message(parent, "剪贴板没有可识别的 AK/SK。")
            return
        ak, sk = self._parse_credentials_for_vendor(vendor, text, app)
        mark = extract_mark(text, vendor)
        if not ak or not sk:
            self._show_credential_message(parent, "未识别到完整 AK/SK。")
            return
        self._apply_resolved_credentials_into_app(
            vendor=vendor,
            app=app,
            ak_var=ak_var,
            sk_var=sk_var,
            mark_var=mark_var,
            parent=parent,
            title_var=title_var,
            ak=ak,
            sk=sk,
            mark=mark,
        )

    def _prepare_active_vendor_for_new_credentials(self) -> None:
        try:
            if self.current_vendor == "huawei" and hasattr(self, "huawei_app"):
                self.huawei_app.prepare_for_new_credentials()
            elif self.current_vendor == "ctyun" and hasattr(self, "ctyun_gui"):
                self.ctyun_gui.prepare_for_new_credentials()
            elif self.current_vendor == "volc" and hasattr(self, "volc_app"):
                self.volc_app.prepare_for_new_credentials()
        except Exception:
            pass

    def _auto_read_active_resources_after_paste(self) -> None:
        try:
            self._sync_active_credentials()
            app = self._active_vendor_app(self.current_vendor)
            if app is not None:
                self._auto_read_vendor_app_resources(self.current_vendor, app, self)
        except Exception as exc:
            self._show_message(f"自动读取账号资源失败：{exc}")

    def _auto_read_vendor_app_resources(self, vendor: str, app: Any, parent: tk.Misc) -> None:
        try:
            if vendor == "huawei":
                app.query_manage_resources(silent=True)
            elif vendor == "ctyun":
                app._query_cleanup_resources()
            elif vendor == "volc":
                app.scan_resources()
        except Exception as exc:
            self._show_credential_message(parent, f"自动读取账号资源失败：{exc}")

    def _live_vendor_child_windows(self, vendor: str) -> List[Dict[str, Any]]:
        live: List[Dict[str, Any]] = []
        for record in self.vendor_child_windows.get(vendor, []):
            window = record.get("window")
            try:
                if window is not None and window.winfo_exists():
                    live.append(record)
            except Exception:
                pass
        self.vendor_child_windows[vendor] = live
        return live

    def _parse_clone_count(self, count_var: tk.StringVar) -> Optional[int]:
        raw = (count_var.get() or "").strip()
        try:
            count = int(raw)
        except ValueError:
            self._show_message("副窗口数量请输入 1 到 7。")
            return None
        if count < 1 or count > 7:
            self._show_message("副窗口数量请输入 1 到 7。")
            return None
        return count

    def _adjust_clone_count(self, count_var: tk.StringVar, delta: int) -> None:
        """调整同云副窗口数量输入框，只允许 1 到 7。"""
        try:
            current = int((count_var.get() or "").strip())
        except ValueError:
            current = 1
        next_value = max(1, min(7, current + int(delta)))
        count_var.set(str(next_value))

    def _build_balance_display(
        self,
        parent: tk.Misc,
        balance_var: tk.StringVar,
        *,
        width: int = 132,
        height: int = 88,
        label_y: int = 57,
        label_height: int = 24,
        anchor: str = "e",
    ) -> tk.Frame:
        box = tk.Frame(parent, width=width, height=height, bg=CONTENT_BG)
        box.grid_propagate(False)
        label = tk.Label(
            box,
            textvariable=balance_var,
            bg=CONTENT_BG,
            fg=MUTED_FG,
            anchor=anchor,
            font=(UI_FONT_FAMILY, 10),
        )
        label.place(x=0, y=label_y, width=width, height=label_height)
        setattr(box, "_balance_display_label", label)
        return box

    def _open_vendor_subwindows(self, vendor: str, count_var: tk.StringVar) -> None:
        key = vendor or self.current_vendor
        if key not in {"huawei", "ctyun", "volc"}:
            self._show_message("请先选择云厂商。")
            return
        count = self._parse_clone_count(count_var)
        if count is None:
            return
        live_count = len(self._live_vendor_child_windows(key))
        remaining = max(0, 7 - live_count)
        if remaining <= 0:
            self._show_message(f"{self._vendor_display_name(key)} 已达到 8 个窗口上限。")
            return
        open_count = min(count, remaining)
        if open_count < count:
            self._show_message(f"{self._vendor_display_name(key)} 只剩 {remaining} 个副窗口名额，本次打开 {open_count} 个。")
        for _index in range(open_count):
            self._create_vendor_subwindow(key)

    def _build_clone_credential_panel(
        self,
        parent: ttk.Frame,
        *,
        vendor: str,
        app: Any,
        window: tk.Toplevel,
        title_var: tk.StringVar,
        ak_var: tk.StringVar,
        sk_var: tk.StringVar,
        mark_var: tk.StringVar,
        count_var: tk.StringVar,
        balance_var: tk.StringVar,
    ) -> None:
        header = ttk.Frame(parent, style="Content.TFrame")
        header.grid(row=0, column=0, sticky="ew", pady=(0, 10))
        header.columnconfigure(0, weight=1)
        header.columnconfigure(1, weight=0)
        header.columnconfigure(2, weight=0)
        header.columnconfigure(3, weight=0)
        header.columnconfigure(4, weight=0)
        ttk.Label(header, textvariable=title_var, style="Header.TLabel").grid(row=0, column=0, sticky="w")
        ttk.Label(header, text="同云副窗口：AK/SK、用户名(可编辑)、日志和结果独立", style="HeaderSubtitle.TLabel").grid(row=1, column=0, sticky="w", pady=(4, 0))

        clone_box = tk.Frame(header, width=100, height=118, bg=CONTENT_BG)
        clone_box.grid(row=0, column=1, rowspan=3, sticky="se", padx=(0, 8))
        clone_box.grid_propagate(False)
        tk.Label(
            clone_box,
            text="新增数量",
            bg=CONTENT_BG,
            fg=TEXT_FG,
            font=(UI_FONT_FAMILY, 10),
        ).place(x=0, y=23, width=100, height=24)
        ttk.Button(
            clone_box,
            text="-",
            command=lambda: self._adjust_clone_count(count_var, -1),
            style="Counter.TButton",
        ).place(x=5, y=50, width=30, height=30)
        ttk.Entry(
            clone_box,
            textvariable=count_var,
            justify="center",
        ).place(x=35, y=50, width=30, height=30)
        ttk.Button(
            clone_box,
            text="+",
            command=lambda: self._adjust_clone_count(count_var, 1),
            style="Counter.TButton",
        ).place(x=65, y=50, width=30, height=30)
        ttk.Button(
            clone_box,
            text="新增窗口",
            command=lambda: self._open_vendor_subwindows(vendor, count_var),
            style="LinkAction.TButton",
        ).place(x=0, y=88, width=100, height=30)

        login_box = self._build_web_login_box(
            header,
            command=lambda: self._open_web_login_for_credentials(
                vendor=vendor,
                app=app,
                ak_var=ak_var,
                sk_var=sk_var,
                mark_var=mark_var,
                parent=window,
                title_var=title_var,
            ),
        )
        login_box.grid(row=0, column=2, rowspan=2, sticky="se", padx=(0, 5))

        cred = ttk.Frame(header, style="Content.TFrame")
        cred.grid(row=0, column=3, rowspan=2, sticky="e")
        cred.columnconfigure(1, minsize=488)
        ttk.Label(cred, text="访问密钥:AK", style="Subheader.TLabel").grid(row=0, column=0, sticky="w", padx=(0, 6))
        ttk.Entry(cred, textvariable=ak_var, width=68).grid(row=0, column=1, sticky="ew", padx=(0, 10), pady=(0, 4))
        ttk.Label(cred, text="访问密钥:SK", style="Subheader.TLabel").grid(row=1, column=0, sticky="w", padx=(0, 6))
        ttk.Entry(cred, textvariable=sk_var, width=68).grid(row=1, column=1, sticky="ew", padx=(0, 10), pady=(0, 4))
        ttk.Label(cred, text="用户名(可编辑):", style="Subheader.TLabel").grid(row=2, column=0, sticky="w", padx=(0, 6))
        balance_box = self._build_credential_history_row(
            cred,
            vendor=vendor,
            app=app,
            parent_window=window,
            ak_var=ak_var,
            sk_var=sk_var,
            mark_var=mark_var,
            title_var=title_var,
            balance_var=balance_var,
        )
        setattr(window, "_cloud_balance_box", balance_box)
        if not self._balance_supported(vendor):
            balance_var.set("")

        paste_box = tk.Frame(cred, width=75, height=88, bg=CONTENT_BG)
        paste_box.grid(row=0, column=2, rowspan=3, sticky="ns")
        paste_box.grid_propagate(False)
        ttk.Button(
            paste_box,
            text="识别\n粘贴",
            command=lambda: self._paste_credentials_into_app(
                vendor=vendor,
                app=app,
                ak_var=ak_var,
                sk_var=sk_var,
                mark_var=mark_var,
                parent=window,
                title_var=title_var,
            ),
        ).place(x=0, y=0, width=75, height=88)
        self._build_binding_action_row(
            header,
            vendor=vendor,
            row=2,
            column=2,
            columnspan=2,
            parent_window=window,
        )

    def _bind_subwindow_credentials(
        self,
        app: Any,
        ak_var: tk.StringVar,
        sk_var: tk.StringVar,
        mark_var: tk.StringVar,
        title_var: tk.StringVar,
        balance_var: tk.StringVar,
        vendor: str,
        window: tk.Toplevel,
    ) -> None:
        def sync(*_args: Any) -> None:
            self._set_app_credentials(app, ak_var.get().strip(), sk_var.get().strip(), mark_var.get().strip())
            title_var.set(self._credential_window_title(vendor, ak_var.get().strip(), mark_var.get().strip()))
            self._remember_mark_for_credentials(vendor, ak_var.get().strip(), sk_var.get().strip(), mark_var.get().strip())

        def key_changed(*_args: Any) -> None:
            sync()
            self._schedule_mark_lookup(
                vendor=vendor,
                app=app,
                ak_var=ak_var,
                sk_var=sk_var,
                mark_var=mark_var,
                parent=window,
                title_var=title_var,
            )
            self._schedule_balance_lookup(
                vendor=vendor,
                app=app,
                ak_var=ak_var,
                sk_var=sk_var,
                parent=window,
                balance_var=balance_var,
            )

        ak_var.trace_add("write", key_changed)
        sk_var.trace_add("write", key_changed)
        mark_var.trace_add("write", sync)
        sync()

    @staticmethod
    def _vendor_status_title(vendor: str) -> str:
        return {
            "huawei": "华为云管理",
            "ctyun": "天翼云管理",
            "volc": "火山引擎管理",
        }.get(vendor, "云管理")

    @staticmethod
    def _vendor_app_busy(app: Any) -> bool:
        if app is None:
            return False
        if bool(getattr(app, "busy", False)) or bool(getattr(app, "running", False)):
            return True
        try:
            return int(getattr(app, "active_task_count", 0) or 0) > 0
        except Exception:
            return False

    def _runtime_tasks_pending(self) -> bool:
        apps = [
            getattr(self, "huawei_app", None),
            getattr(self, "ctyun_gui", None),
            getattr(self, "volc_app", None),
        ]
        for records in getattr(self, "vendor_child_windows", {}).values():
            apps.extend(record.get("app") for record in records if isinstance(record, dict))

        for app in apps:
            if app is None:
                continue
            if self._vendor_app_busy(app):
                return True
            for attribute in (
                "delayed_action_queue",
                "delayed_action_running_label",
                "delayed_current_action_label",
                "customer_guard_pending_resource_query",
            ):
                try:
                    if bool(getattr(app, attribute, None)):
                        return True
                except Exception:
                    continue

        manager = getattr(self, "subprocess_task_manager", None)
        return bool(manager and manager.has_running_tasks())

    @staticmethod
    def _normalize_vendor_status_detail(detail: str, busy: bool) -> str:
        value = str(detail or "").strip() or "就绪"
        if busy:
            return "任务运行中" if value == "就绪" else value
        if value == "正在读取账号资源":
            return "账号资源读取完成"
        if value == "任务运行中":
            return "就绪"
        return value

    def _subwindow_status_text(self, vendor: str, app: Any) -> str:
        detail = "就绪"
        status_var = getattr(app, "status_var", None)
        if isinstance(status_var, tk.StringVar):
            detail = status_var.get() or "就绪"
        detail = self._normalize_vendor_status_detail(detail, self._vendor_app_busy(app))
        return f"{self._vendor_status_title(vendor)} | {detail}"

    def _build_subwindow_status_bar(self, window: tk.Toplevel, status_var: tk.StringVar) -> None:
        status_bar = tk.Frame(window, bg=PANEL_BG, height=28, highlightthickness=1, highlightbackground=BORDER_COLOR)
        status_bar.pack(side=tk.BOTTOM, fill=tk.X)
        status_bar.pack_propagate(False)
        status_bar.columnconfigure(0, weight=1)
        tk.Label(
            status_bar,
            textvariable=status_var,
            bg=PANEL_BG,
            fg=TEXT_FG,
            anchor="e",
            font=(UI_FONT_FAMILY, 10),
        ).grid(row=0, column=0, sticky="e", padx=16, pady=4)

    def _create_vendor_subwindow(self, vendor: str) -> None:
        window = tk.Toplevel(self)
        window.withdraw()
        window.title(self._credential_window_title(vendor))
        window.geometry("1280x860")
        window.minsize(1024, 640)
        window.configure(bg=CONTENT_BG)
        apply_multi_window_icon(window)

        sub_status_var = tk.StringVar(value=f"{self._vendor_status_title(vendor)} | 就绪")
        self._build_subwindow_status_bar(window, sub_status_var)

        outer = ttk.Frame(window, style="Content.TFrame", padding=(14, 12, 14, 14))
        outer.pack(side=tk.TOP, fill=tk.BOTH, expand=True)
        outer.columnconfigure(0, weight=1)
        outer.rowconfigure(1, weight=1)

        body = ttk.Frame(outer, style="Content.TFrame")
        body.grid(row=1, column=0, sticky="nsew")
        body.columnconfigure(0, weight=1)
        body.rowconfigure(0, weight=1)

        if vendor == "huawei":
            app = HuaweiAutoBuyApp(body, embedded=True)
            app.grid(row=0, column=0, sticky="nsew")
        elif vendor == "ctyun":
            app = ctyun_workflow_gui.WorkflowGui(self, parent=body, embedded=True)
        else:
            window.customer_manager_auth_code = self.customer_manager_auth_code
            app = VolcEngineManagerApp(body, embedded=True)
            app.grid(row=0, column=0, sticky="nsew")
        app.subprocess_task_manager = self.subprocess_task_manager

        def update_subwindow_status(*_args: Any) -> None:
            try:
                if window.winfo_exists():
                    sub_status_var.set(self._subwindow_status_text(vendor, app))
            except tk.TclError:
                pass

        status_var = getattr(app, "status_var", None)
        status_trace = ""
        if isinstance(status_var, tk.StringVar):
            status_trace = status_var.trace_add("write", update_subwindow_status)
        update_subwindow_status()

        ak_var = tk.StringVar()
        sk_var = tk.StringVar()
        mark_var = tk.StringVar()
        count_var = tk.StringVar(value="1")
        title_var = tk.StringVar(value=self._credential_window_title(vendor))
        balance_var = tk.StringVar(value="余额: --" if self._balance_supported(vendor) else "")
        setattr(window, "_cloud_balance_var", balance_var)
        self._bind_subwindow_credentials(app, ak_var, sk_var, mark_var, title_var, balance_var, vendor, window)
        self._build_clone_credential_panel(
            outer,
            vendor=vendor,
            app=app,
            window=window,
            title_var=title_var,
            ak_var=ak_var,
            sk_var=sk_var,
            mark_var=mark_var,
            count_var=count_var,
            balance_var=balance_var,
        )

        record = {"window": window, "app": app, "vendor": vendor, "status_var": status_var, "status_trace": status_trace}
        self.vendor_child_windows.setdefault(vendor, []).append(record)
        window.protocol("WM_DELETE_WINDOW", lambda item=record: self._close_vendor_subwindow(item))
        try:
            window.update_idletasks()
            window.state("zoomed")
        except tk.TclError:
            try:
                window.deiconify()
                window.state("zoomed")
            except tk.TclError:
                pass
        try:
            window.deiconify()
        except tk.TclError:
            pass

    def _close_vendor_subwindow(self, record: Dict[str, Any]) -> None:
        app = record.get("app")
        window = record.get("window")
        status_var = record.get("status_var")
        status_trace = str(record.get("status_trace") or "")
        if isinstance(status_var, tk.StringVar) and status_trace:
            try:
                status_var.trace_remove("write", status_trace)
            except Exception:
                pass
        shutdown = getattr(app, "shutdown", None)
        if callable(shutdown):
            try:
                shutdown()
            except Exception:
                pass
        vendor = str(record.get("vendor") or "")
        if vendor in self.vendor_child_windows:
            self.vendor_child_windows[vendor] = [item for item in self.vendor_child_windows[vendor] if item is not record]
        try:
            if window is not None and window.winfo_exists():
                window.destroy()
        except Exception:
            pass

    def _update_global_status(self) -> None:
        if self._closing:
            return
        title = self.page_title_var.get() if hasattr(self, "page_title_var") else "多云管理"
        detail = "就绪"
        huawei_app = getattr(self, "huawei_app", None)
        ctyun_app = getattr(self, "ctyun_gui", None)
        volc_app = getattr(self, "volc_app", None)
        huawei_busy = self._vendor_app_busy(huawei_app)
        ctyun_busy = self._vendor_app_busy(ctyun_app)
        volc_busy = self._vendor_app_busy(volc_app)
        subprocess_count = len(self.subprocess_task_manager._running_tasks())
        self._set_vendor_progress("huawei", huawei_busy)
        self._set_vendor_progress("ctyun", ctyun_busy)
        self._set_vendor_progress("volc", volc_busy)
        if self.current_vendor == "huawei" and hasattr(self, "huawei_app"):
            detail = self.huawei_app.status_var.get() or "就绪"
            detail = self._normalize_vendor_status_detail(detail, huawei_busy)
        elif self.current_vendor == "ctyun" and hasattr(self, "ctyun_gui"):
            ctyun_status = self.ctyun_gui.status_var.get() or ""
            detail = ctyun_status or ("运行中" if ctyun_busy else "就绪")
            detail = self._normalize_vendor_status_detail(detail, ctyun_busy)
        elif self.current_vendor == "volc" and hasattr(self, "volc_app"):
            detail = self.volc_app.status_var.get() or "就绪"
            detail = self._normalize_vendor_status_detail(detail, volc_busy)
        if subprocess_count:
            detail = f"{detail} | 子进程任务 {subprocess_count} 个"
        self.global_status_var.set(f"{title} | {detail}")

    def _schedule_status_poll(self) -> None:
        if self._closing:
            return
        self._update_global_status()
        self.after(800, self._schedule_status_poll)

    def _runtime_update_package_key(self) -> str:
        if self.single_vendor == "huawei":
            return "huawei"
        if self.single_vendor == "ctyun":
            return "ctyun"
        if self.single_vendor == "volc":
            return "volcengine"
        return DEFAULT_UPDATE_PACKAGE

    def _runtime_update_message(self, check_result: Any) -> str:
        if bool(getattr(check_result, "auto_migration_available", False)):
            lines = [
                "检测到当前云管理软件将升级为多云管理。",
                "当前正在执行的任务不会受到影响，建议在方便时重新打开软件完成本地合并升级。",
                "升级会保留历史账号、日志、运行结果和旧IP库。",
            ]
            return "\n".join(lines)
        lines = [
            "检测到重要更新或功能更新。",
            "本次更新用于优化软件体验并提升稳定性。",
            "当前正在执行的任务不会受到影响，建议在方便时更新后继续使用。",
        ]
        return "\n".join(lines)

    def _defer_runtime_update_reminder(self, check_result: Any) -> None:
        self._runtime_update_pending_result = check_result
        if self._runtime_update_wait_job:
            return

        def wait_for_idle() -> None:
            self._runtime_update_wait_job = None
            if self._closing:
                self._runtime_update_pending_result = None
                return
            if self._runtime_tasks_pending():
                try:
                    self._runtime_update_wait_job = str(self.after(1000, wait_for_idle))
                except tk.TclError:
                    self._runtime_update_wait_job = None
                return

            pending_result = self._runtime_update_pending_result
            self._runtime_update_pending_result = None
            if pending_result is not None:
                self._show_runtime_update_reminder(pending_result)

        try:
            self._runtime_update_wait_job = str(self.after(1000, wait_for_idle))
        except tk.TclError:
            self._runtime_update_wait_job = None

    def _show_runtime_update_reminder(self, check_result: Any) -> None:
        if self._runtime_tasks_pending():
            self._defer_runtime_update_reminder(check_result)
            return
        if bool(getattr(check_result, "update_retry_blocked", False)):
            self._runtime_update_dialog = show_app_message_dialog(
                self,
                "软件更新提醒",
                UPDATE_FAILED_CONTACT_MESSAGE,
                existing_dialog=self._runtime_update_dialog,
                actions=(("关闭", lambda: None),),
            )
            return
        if bool(getattr(check_result, "auto_migration_available", False)):
            migration = getattr(check_result, "migration", {}) or {}
            title = str(migration.get("title") or "云管理升级提醒").strip()
            self._runtime_update_dialog = show_app_message_dialog(
                self,
                title,
                self._runtime_update_message(check_result),
                existing_dialog=self._runtime_update_dialog,
                actions=(
                    ("稍后更新", lambda: None),
                ),
            )
            return

        def restart_for_update() -> None:
            try:
                if getattr(sys, "frozen", False):
                    command = [sys.executable]
                    cwd = os.path.dirname(os.path.abspath(sys.executable)) or None
                else:
                    command = [sys.executable, *sys.argv]
                    cwd = os.getcwd()
                subprocess.Popen(command, cwd=cwd, close_fds=True)
            except Exception as exc:
                show_app_message_dialog(
                    self,
                    "软件更新提醒",
                    f"无法重新打开软件：{exc}\n{UPDATE_FAILED_CONTACT_MESSAGE}",
                    modal=True,
                )
                return
            self.force_close()

        self._runtime_update_dialog = show_app_message_dialog(
            self,
            "软件更新提醒",
            self._runtime_update_message(check_result),
            existing_dialog=self._runtime_update_dialog,
            actions=(
                ("立即更新", restart_for_update),
                ("稍后更新", lambda: None),
            ),
        )

    def _schedule_release_announcement_poll(self, delay_ms: int = RUNTIME_UPDATE_POLL_INTERVAL_MS) -> None:
        if self._closing:
            return

        def poll() -> None:
            self._announcement_poll_job = None
            if self._closing:
                return

            def worker() -> None:
                try:
                    check_result = check_latest_version(APP_VERSION, package_key=self._runtime_update_package_key())
                except Exception:
                    return

                def show() -> None:
                    if self._closing:
                        return
                    if not getattr(check_result, "error", "") and (
                        bool(getattr(check_result, "auto_update_available", False))
                        or bool(getattr(check_result, "needs_update", False))
                    ):
                        self._show_runtime_update_reminder(check_result)
                    else:
                        show_release_announcement_if_needed(self, check_result)

                try:
                    self.after(0, show)
                except tk.TclError:
                    pass

            threading.Thread(target=worker, daemon=True).start()
            self._schedule_release_announcement_poll()

        try:
            self._announcement_poll_job = str(self.after(delay_ms, poll))
        except tk.TclError:
            self._announcement_poll_job = None

    def force_close(self) -> None:
        if self._closing:
            os._exit(0)
        self._closing = True
        if self._announcement_poll_job:
            try:
                self.after_cancel(self._announcement_poll_job)
            except tk.TclError:
                pass
            self._announcement_poll_job = None
        if self._runtime_update_wait_job:
            try:
                self.after_cancel(self._runtime_update_wait_job)
            except tk.TclError:
                pass
            self._runtime_update_wait_job = None
        self._runtime_update_pending_result = None
        if self._window_recovery_job:
            try:
                self.after_cancel(self._window_recovery_job)
            except tk.TclError:
                pass
            self._window_recovery_job = None
        for target_key in list(getattr(self, "_web_login_sessions", {}).keys()):
            self._cleanup_web_login_session(target_key, terminate_process=True)
        try:
            self.global_status_var.set("正在关闭...")
        except Exception:
            pass
        try:
            if hasattr(self, "huawei_app"):
                shutdown = getattr(self.huawei_app, "shutdown", None)
                if callable(shutdown):
                    shutdown()
                else:
                    self.huawei_app.stop_event.set()
                    self.huawei_app.busy = False
        except Exception:
            pass
        try:
            if hasattr(self, "ctyun_gui"):
                shutdown = getattr(self.ctyun_gui, "shutdown", None)
                if callable(shutdown):
                    shutdown()
                else:
                    self.ctyun_gui.stop_requested = True
                    self.ctyun_gui.running = False
                    try:
                        self.ctyun_gui._request_stop()
                    except Exception:
                        pass
        except Exception:
            pass
        try:
            if hasattr(self, "volc_app"):
                shutdown = getattr(self.volc_app, "shutdown", None)
                if callable(shutdown):
                    shutdown()
                else:
                    self.volc_app.stop_tasks()
        except Exception:
            pass
        timer = threading.Timer(0.6, lambda: os._exit(0))
        timer.daemon = True
        timer.start()
        try:
            self.quit()
        except Exception:
            pass
        try:
            self.destroy()
        except Exception:
            os._exit(0)


def show_startup_notice(root: tk.Tk) -> bool:
    result = {"ok": False}
    dialog = tk.Toplevel(root)
    dialog.withdraw()
    dialog.title("欢迎使用")
    dialog.resizable(True, True)
    dialog.configure(bg=APP_BG)
    apply_multi_window_icon(dialog)
    try:
        style = ttk.Style(dialog)
        if "vista" in style.theme_names():
            style.theme_use("vista")
        style.configure("TFrame", background=APP_BG)
        style.configure("TLabel", background=APP_BG, foreground=TEXT_FG)
        style.configure(
            "TButton",
            padding=(8, 5),
            font=(UI_FONT_FAMILY, 10),
        )
        style.configure(
            "Notice.TButton",
            padding=(8, 5),
            font=(UI_FONT_FAMILY, 10),
        )
    except Exception:
        pass
    try:
        if root.state() != "withdrawn":
            dialog.transient(root)
    except tk.TclError:
        pass
    dialog.grab_set()
    dialog.attributes("-topmost", True)
    dialog.protocol("WM_DELETE_WINDOW", lambda: dialog.destroy())

    frame = _startup_notice_content_frame(dialog)
    pack_startup_notice(frame, STARTUP_DISCLAIMER_BODY)
    auth_code_var = tk.StringVar(value=read_saved_manager_auth_code(DEFAULT_UPDATE_PACKAGE))
    auth_entry = build_manager_auth_row(frame, auth_code_var)
    buttons = ttk.Frame(frame)
    buttons.pack(anchor="center", pady=(14, 0))
    buttons.columnconfigure(0, weight=1)
    buttons.columnconfigure(1, weight=1)

    def refresh_agree_button() -> None:
        agree_button.configure(state="normal" if auth_code_var.get().strip() else "disabled")

    def agree() -> None:
        global CURRENT_MANAGER_AUTH_CODE
        code = normalize_manager_auth_code(auth_code_var.get())
        agree_button.configure(text="校验中...", state="disabled")
        dialog.update_idletasks()
        valid, auth_message = authorize_manager_auth_code(code)
        agree_button.configure(text="同意并进入")
        refresh_agree_button()
        if not valid:
            if auth_message == INVALID_MANAGER_AUTH_MESSAGE:
                save_manager_auth_code("", DEFAULT_UPDATE_PACKAGE)
            show_app_message_dialog(dialog, "授权码校验", auth_message, modal=True)
            auth_entry.focus_set()
            return
        set_current_manager_auth_code(code)
        save_manager_auth_code(code, DEFAULT_UPDATE_PACKAGE)
        start_manager_auth_usage_reporting(code, DEFAULT_UPDATE_PACKAGE, APP_VERSION)
        result["ok"] = True
        dialog.destroy()

    def cancel() -> None:
        dialog.destroy()

    agree_button = ttk.Button(buttons, text="同意并进入", command=agree, style="Notice.TButton", width=14, state="disabled")
    agree_button.grid(row=0, column=0, padx=(0, 10))
    ttk.Button(buttons, text="退出", command=cancel, style="Notice.TButton", width=14).grid(row=0, column=1, padx=(10, 0))
    auth_code_var.trace_add("write", lambda *_args: refresh_agree_button())
    refresh_agree_button()

    def submit_auth_on_enter(_event: tk.Event | None = None) -> str:
        if str(agree_button.cget("state")) != "disabled":
            agree()
        return "break"

    auth_entry.bind("<Return>", submit_auth_on_enter)
    auth_entry.bind("<KP_Enter>", submit_auth_on_enter)
    dialog.deiconify()
    _fit_startup_notice_window(dialog, frame)
    dialog.lift()
    dialog.focus_force()
    auth_entry.focus_set()
    dialog.after(500, lambda: dialog.attributes("-topmost", False) if dialog.winfo_exists() else None)
    root.wait_window(dialog)
    return bool(result["ok"])


def show_startup_notice_window(
    *,
    app_title: str = APP_TITLE,
    subtitle: str = "正式完整包，包含华为云、天翼云、火山引擎管理功能。",
    version: str = APP_VERSION,
    package_label: str = "正式完整包",
    platform_label: str = "华为云 / 天翼云 / 火山引擎",
    logo_filename: str = "logo_app_68.png",
    release_notes: list[str] | None = None,
    package_key: str = DEFAULT_UPDATE_PACKAGE,
    download_page_url: str = DOWNLOAD_PAGE_URL,
) -> bool:
    result = {"ok": False}
    version_state = {"allowed": False, "download_url": download_page_url, "check_result": None, "checking": False}
    notice_root = tk.Tk()
    notice_root.withdraw()
    notice_root.title("欢迎使用")
    notice_root.resizable(True, True)
    notice_root.attributes("-topmost", True)
    notice_root.configure(bg=APP_BG)
    apply_multi_window_icon(notice_root)
    try:
        style = ttk.Style(notice_root)
        if "vista" in style.theme_names():
            style.theme_use("vista")
        style.configure("TFrame", background=APP_BG)
        style.configure("TLabel", background=APP_BG, foreground=TEXT_FG)
        style.configure(
            "TButton",
            padding=(8, 5),
            font=(UI_FONT_FAMILY, 10),
        )
        style.configure(
            "Notice.TButton",
            padding=(8, 5),
            font=(UI_FONT_FAMILY, 10),
        )
    except Exception:
        pass
    frame = _startup_notice_content_frame(notice_root)
    pack_startup_notice(
        frame,
        STARTUP_DISCLAIMER_BODY,
        app_title=app_title,
        subtitle=subtitle,
        version=version,
        package_label=package_label,
        platform_label=platform_label,
        logo_filename=logo_filename,
        release_notes=release_notes,
    )
    auth_code_var = tk.StringVar(value=read_saved_manager_auth_code(package_key))
    auth_entry = build_manager_auth_row(frame, auth_code_var)
    status_var = tk.StringVar(value="软件版本检测中，请稍候...")
    status_label = ttk.Label(
        frame,
        textvariable=status_var,
        font=(UI_FONT_FAMILY, 10),
        wraplength=520,
        justify=tk.CENTER,
        anchor="center",
    )
    status_label.pack(fill=tk.X, pady=(12, 0))
    buttons = ttk.Frame(frame)
    buttons.pack(anchor="center", pady=(14, 0))
    buttons.columnconfigure(0, weight=1)
    buttons.columnconfigure(1, weight=1)
    buttons.columnconfigure(2, weight=1)
    buttons.columnconfigure(3, weight=1)

    def refresh_agree_button() -> None:
        ready = bool(version_state["allowed"]) and bool(auth_code_var.get().strip())
        agree_button.configure(state="normal" if ready else "disabled")

    def agree() -> None:
        global CURRENT_MANAGER_AUTH_CODE
        if not version_state["allowed"]:
            return
        code = normalize_manager_auth_code(auth_code_var.get())
        upgrade_options = _manager_auth_public_upgrade_options(code)
        if upgrade_options:
            def choose_upgraded_code(new_code: str) -> None:
                auth_code_var.set(new_code)
                save_manager_auth_code(new_code, package_key)
                auth_entry.focus_set()

            show_app_message_dialog(
                notice_root,
                "授权码升级",
                (
                    "该授权码已升级，请选择需要使用的云管理功能。\n"
                    "选择后会自动替换授权码输入框，下次打开软件将默认显示新的授权码。"
                ),
                modal=True,
                actions=tuple(
                    (label, lambda value=new_code: choose_upgraded_code(value))
                    for label, new_code in upgrade_options
                ),
            )
            return
        agree_button.configure(text="校验中...", state="disabled")
        notice_root.update_idletasks()
        valid, auth_message = authorize_manager_auth_code(code, package_key)
        agree_button.configure(text="同意并进入")
        refresh_agree_button()
        if not valid:
            if auth_message == INVALID_MANAGER_AUTH_MESSAGE:
                save_manager_auth_code("", package_key)
            show_app_message_dialog(notice_root, "授权码校验", auth_message, modal=True)
            auth_entry.focus_set()
            return
        set_current_manager_auth_code(code)
        save_manager_auth_code(code, package_key)
        start_manager_auth_usage_reporting(code, package_key, version)
        show_release_announcement_if_needed(notice_root, version_state.get("check_result"))
        result["ok"] = True
        notice_root.destroy()

    def cancel() -> None:
        notice_root.destroy()

    def open_download() -> None:
        webbrowser.open(str(version_state.get("download_url") or download_page_url))

    agree_button = ttk.Button(buttons, text="同意并进入", command=agree, style="Notice.TButton", width=14, state="disabled")
    retry_button = ttk.Button(buttons, text="重新检测", command=lambda: start_version_check(), style="Notice.TButton", width=14, state="disabled")
    download_button = ttk.Button(buttons, text="下载最新版", command=open_download, style="Notice.TButton", width=14, state="disabled")
    exit_button = ttk.Button(buttons, text="退出", command=cancel, style="Notice.TButton", width=14)
    agree_button.grid(row=0, column=0, padx=(0, 10))
    exit_button.grid(row=0, column=3)
    retry_button_visible = {"value": False}
    download_button_visible = {"value": False}
    auth_code_var.trace_add("write", lambda *_args: refresh_agree_button())

    def submit_auth_on_enter(_event: tk.Event | None = None) -> str:
        if str(agree_button.cget("state")) != "disabled":
            agree()
        return "break"

    auth_entry.bind("<Return>", submit_auth_on_enter)
    auth_entry.bind("<KP_Enter>", submit_auth_on_enter)

    def show_download_button() -> None:
        download_button.configure(state="normal")
        if not download_button_visible["value"]:
            download_button.grid(row=0, column=2, padx=(0, 10))
            download_button_visible["value"] = True

    def hide_download_button() -> None:
        download_button.configure(state="disabled")
        if download_button_visible["value"]:
            download_button.grid_remove()
            download_button_visible["value"] = False

    def show_retry_button() -> None:
        retry_button.configure(state="normal")
        if not retry_button_visible["value"]:
            retry_button.grid(row=0, column=1, padx=(0, 10))
            retry_button_visible["value"] = True

    def hide_retry_button() -> None:
        retry_button.configure(state="disabled")
        if retry_button_visible["value"]:
            retry_button.grid_remove()
            retry_button_visible["value"] = False

    def relaunch_after_update() -> None:
        if not notice_root.winfo_exists():
            return
        status_var.set("自动更新完成，正在重新打开软件...")
        agree_button.configure(state="disabled")
        hide_retry_button()
        hide_download_button()
        exit_button.configure(state="disabled")

        def relaunch() -> None:
            if not notice_root.winfo_exists():
                return
            try:
                if getattr(sys, "frozen", False):
                    command = [sys.executable]
                    cwd = os.path.dirname(os.path.abspath(sys.executable)) or None
                else:
                    command = [sys.executable, *sys.argv]
                    cwd = os.getcwd()
                subprocess.Popen(command, cwd=cwd, close_fds=True)
            except Exception as exc:
                status_var.set(f"自动更新完成，但自动重新打开失败：{exc}\n请手动重新打开软件。")
                exit_button.configure(state="normal")
                return
            notice_root.destroy()

        notice_root.after(800, relaunch)

    def close_for_external_update(message: str) -> None:
        if not notice_root.winfo_exists():
            return
        status_var.set(message)
        agree_button.configure(state="disabled")
        hide_retry_button()
        hide_download_button()
        exit_button.configure(state="disabled")
        notice_root.after(800, notice_root.destroy)

    def finish_update_without_restart() -> None:
        version_state["allowed"] = True
        status_var.set("更新完毕，欢迎使用。")
        agree_button.configure(text="进入软件")
        refresh_agree_button()
        hide_retry_button()
        hide_download_button()
        exit_button.configure(state="normal")

    def start_auto_update(check_result) -> None:
        def set_progress(message: str) -> None:
            def apply_progress() -> None:
                if not notice_root.winfo_exists():
                    return
                status_var.set(message)
                _fit_startup_notice_window(notice_root, frame)

            notice_root.after(0, apply_progress)

        def worker() -> None:
            update_result = perform_auto_update(check_result, progress=set_progress)

            def apply_update_result() -> None:
                if not notice_root.winfo_exists():
                    return
                if update_result.ok:
                    if update_result.external_restart:
                        close_for_external_update(update_result.message)
                    elif update_result.restart_required:
                        relaunch_after_update()
                    else:
                        finish_update_without_restart()
                else:
                    status_var.set(update_result.message)
                    agree_button.configure(state="disabled")
                    if update_result.retry_blocked:
                        hide_retry_button()
                    else:
                        show_retry_button()
                    hide_download_button()
                    exit_button.configure(state="normal")

            notice_root.after(0, apply_update_result)

        threading.Thread(target=worker, daemon=True).start()

    def start_package_migration(check_result) -> None:
        def set_progress(message: str) -> None:
            def apply_progress() -> None:
                if not notice_root.winfo_exists():
                    return
                status_var.set(message)
                _fit_startup_notice_window(notice_root, frame)

            notice_root.after(0, apply_progress)

        def worker() -> None:
            update_result = perform_package_migration(check_result, progress=set_progress)

            def apply_update_result() -> None:
                if not notice_root.winfo_exists():
                    return
                if update_result.ok:
                    status_var.set(update_result.message)
                    agree_button.configure(state="disabled")
                    hide_retry_button()
                    hide_download_button()
                    exit_button.configure(state="disabled")
                    notice_root.after(1000, notice_root.destroy)
                else:
                    status_var.set(f"{update_result.message}\n请稍后重试；如仍失败，再手动下载最新版：{version_state.get('download_url') or download_page_url}")
                    agree_button.configure(state="disabled")
                    show_retry_button()
                    show_download_button()
                    exit_button.configure(state="normal")

            notice_root.after(0, apply_update_result)

        threading.Thread(target=worker, daemon=True).start()

    def start_package_migration_countdown(check_result, seconds: int = UPDATE_COUNTDOWN_SECONDS) -> None:
        version_state["download_url"] = check_result.download_page_url or download_page_url
        agree_button.configure(state="disabled")
        hide_retry_button()
        hide_download_button()

        def tick(remaining: int) -> None:
            if not notice_root.winfo_exists():
                return
            if remaining <= 0:
                status_var.set("正在准备分包升级，请稍候...")
                start_package_migration(check_result)
                return
            status_var.set(f"{check_result.message}\n将在 {remaining} 秒后自动合并本地文件升级，请勿关闭软件。")
            notice_root.after(1000, lambda: tick(remaining - 1))

        tick(seconds)

    def start_update_countdown(check_result, seconds: int = UPDATE_COUNTDOWN_SECONDS) -> None:
        version_state["download_url"] = check_result.download_page_url or download_page_url
        agree_button.configure(state="disabled")
        hide_retry_button()
        hide_download_button()

        def tick(remaining: int) -> None:
            if not notice_root.winfo_exists():
                return
            if remaining <= 0:
                status_var.set("正在自动更新，请稍候...")
                start_auto_update(check_result)
                return
            status_var.set(f"{check_result.message}\n将在 {remaining} 秒后自动更新，请勿关闭软件。")
            notice_root.after(1000, lambda: tick(remaining - 1))

        tick(seconds)

    def run_version_check() -> None:
        check_result = check_latest_version(version, package_key=package_key)

        def apply_check_result() -> None:
            if not notice_root.winfo_exists():
                return
            version_state["checking"] = False
            version_state["allowed"] = check_result.allowed
            version_state["download_url"] = check_result.download_page_url or download_page_url
            version_state["check_result"] = check_result
            if getattr(check_result, "disabled", False):
                status_var.set(check_result.message)
                agree_button.configure(state="disabled")
                hide_retry_button()
                hide_download_button()
            elif check_result.allowed:
                status_var.set("版本检测完成：最新版，欢迎使用!")
                agree_button.configure(text="同意并进入")
                refresh_agree_button()
                hide_retry_button()
                hide_download_button()
            elif getattr(check_result, "auto_migration_available", False):
                start_package_migration_countdown(check_result)
            elif check_result.auto_update_available:
                start_update_countdown(check_result)
            elif getattr(check_result, "update_retry_blocked", False):
                status_var.set(UPDATE_FAILED_CONTACT_MESSAGE)
                agree_button.configure(state="disabled")
                hide_retry_button()
                hide_download_button()
            else:
                status_var.set(check_result.message)
                agree_button.configure(state="disabled")
                if check_result.error:
                    show_retry_button()
                else:
                    hide_retry_button()
                hide_download_button()

        notice_root.after(0, apply_check_result)

    def start_version_check() -> None:
        if version_state["checking"]:
            return
        version_state["checking"] = True
        version_state["allowed"] = False
        status_var.set("正在重新连接更新服务器，请稍候...")
        agree_button.configure(state="disabled")
        hide_retry_button()
        hide_download_button()
        threading.Thread(target=run_version_check, daemon=True).start()

    start_version_check()
    notice_root.protocol("WM_DELETE_WINDOW", cancel)
    notice_root.deiconify()
    _acknowledge_full_update_health_check()
    _fit_startup_notice_window(notice_root, frame)
    notice_root.lift()
    notice_root.focus_force()
    auth_entry.focus_set()
    notice_root.after(500, lambda: notice_root.attributes("-topmost", False) if notice_root.winfo_exists() else None)
    notice_root.mainloop()
    return bool(result["ok"])


def run_self_test() -> None:
    assert normalize_manager_auth_code(" khjlnd ") == "KHJLND"
    assert CUSTOMER_MANAGER_AUTH_USAGE_HEARTBEAT_INTERVAL_SECONDS == 6 * 60
    assert CUSTOMER_MANAGER_BINDING_LINKS == {}
    assert CUSTOMER_MANAGER_CONTACTS == {}
    assert CUSTOMER_MANAGER_ALLOWED_VENDORS == {}
    assert CUSTOMER_MANAGER_LIMITED_FEATURE_CODES == set()
    base_code, scoped_vendors, scoped = _manager_auth_code_vendor_scope("KHJLNDHSHW")
    assert base_code == "KHJLND"
    assert scoped
    assert scoped_vendors == ("huawei", "volc")
    base_code, scoped_vendors, scoped = _manager_auth_code_vendor_scope("KHJLNDHWHSCT")
    assert base_code == "KHJLND"
    assert scoped
    assert scoped_vendors == ("huawei", "ctyun", "volc")
    for code, source_code in (
        ("KHJLNDHWHSCT", "ND"),
        ("KHJLTTCT", "WT"),
        ("KHJLWWHW", "WW"),
        ("ND", "ND"),
        ("WT", "WT"),
        ("WW", "WW"),
    ):
        public_links = _manager_auth_public_binding_links(code)
        source_links = LOCAL_INTERNAL_MANAGER_AUTH_CODES[source_code]["links"]
        assert public_links == source_links
        assert all(public_links[vendor].startswith("https://") for vendor in CUSTOMER_MANAGER_VENDOR_ORDER)
        binding_view = type("BindingView", (), {"customer_manager_auth_code": code})()
        for vendor in CUSTOMER_MANAGER_VENDOR_ORDER:
            assert MultiCloudManagerApp._vendor_binding_url(binding_view, vendor) == source_links[vendor]
    assert _manager_auth_public_upgrade_options("KHJLND") == (
        ("华为云", "KHJLNDHW"),
        ("火山引擎", "KHJLNDHS"),
        ("华为云+火山引擎", "KHJLNDHWHS"),
    )
    valid, message = authorize_manager_auth_code(" ww ")
    assert valid and not message
    assert CUSTOMER_MANAGER_BINDING_LINKS["WW"]["huawei"].startswith("https://")
    assert CUSTOMER_MANAGER_BINDING_LINKS["WW"]["ctyun"].startswith("https://")
    assert CUSTOMER_MANAGER_BINDING_LINKS["WW"]["volc"].startswith("https://")
    assert CUSTOMER_MANAGER_CONTACTS["WW"]
    valid, message = authorize_manager_auth_code("wwhs")
    assert valid and not message
    assert manager_auth_allowed_vendors("WWHS") == ("volc",)
    assert not manager_auth_limited_features("WWHS")
    _remember_manager_auth_payload(
        "KHJLNDHS",
        {"huawei": "https://example.com/hw", "ctyun": "https://example.com/ct", "volc": "https://example.com/hs"},
        (("微信", ("x",)),),
        ("volc",),
        limited_features=True,
    )
    assert manager_auth_allowed_vendors("KHJLNDHS") == ("volc",)
    assert manager_auth_limited_features("KHJLNDHS")
    cached_binding_view = type("CachedBindingView", (), {"customer_manager_auth_code": "KHJLNDHS"})()
    assert MultiCloudManagerApp._vendor_binding_url(cached_binding_view, "huawei") == "https://example.com/hw"
    assert MultiCloudManagerApp._vendor_binding_url(cached_binding_view, "ctyun") == "https://example.com/ct"
    assert MultiCloudManagerApp._vendor_binding_url(cached_binding_view, "volc") == "https://example.com/hs"
    from common.debug_billing import limited_vendor_features_enabled
    old_limited_vendors = os.environ.get(LIMITED_VENDOR_FEATURES_ENV)
    os.environ[LIMITED_VENDOR_FEATURES_ENV] = "huawei,volc"
    assert limited_vendor_features_enabled("huawei")
    assert limited_vendor_features_enabled("volc")
    assert not limited_vendor_features_enabled("ctyun")
    if old_limited_vendors is None:
        os.environ.pop(LIMITED_VENDOR_FEATURES_ENV, None)
    else:
        os.environ[LIMITED_VENDOR_FEATURES_ENV] = old_limited_vendors
    CUSTOMER_MANAGER_BINDING_LINKS.clear()
    CUSTOMER_MANAGER_CONTACTS.clear()
    CUSTOMER_MANAGER_ALLOWED_VENDORS.clear()
    CUSTOMER_MANAGER_LIMITED_FEATURE_CODES.clear()
    assert CUSTOMER_MANAGER_BINDING_LINKS == {}
    assert CUSTOMER_MANAGER_CONTACTS == {}
    assert CUSTOMER_MANAGER_ALLOWED_VENDORS == {}
    assert CUSTOMER_MANAGER_LIMITED_FEATURE_CODES == set()
    assert AK_SK_TUTORIAL_URLS["ctyun"] == "https://www.wawaapi.com/YunTools/CTYUN/AKSK"
    assert AK_SK_TUTORIAL_URLS["huawei"] == "https://www.wawaapi.com/YunTools/Huawei/AKSK"
    assert AK_SK_TUTORIAL_URLS["volc"] == "https://www.wawaapi.com/YunTools/Volcengine/AKSK"
    ak, sk = extract_ak_sk("访问密钥 AK HPUA1MPCJTGVLXFH2YL1\n访问密钥 SK jecL6ySbHp0MElqcajzbMxE18S68pEojr0YR0ron")
    assert ak == "HPUA1MPCJTGVLXFH2YL1"
    assert sk == "jecL6ySbHp0MElqcajzbMxE18S68pEojr0YR0ron"
    volc_ak, volc_sk = extract_ak_sk(
        "AccessKeyID: AKLTFAKEACCESSKEY1234567890ABCDE\n"
        "SecretAccessKey: VGhpcy1pcy1hLWZha2Utc2VjcmV0S2V5PT0="
    )
    assert volc_ak == "AKLTFAKEACCESSKEY1234567890ABCDE"
    assert volc_sk == "VGhpcy1pcy1hLWZha2Utc2VjcmV0S2V5PT0="
    downloaded_ak, downloaded_sk = extract_ak_sk(
        "访问密钥ID\nHPUA7KOWAIIYEQ6314PV\n状态\n启用\n"
        "下载访问密钥\n访问密钥ID\nHPUA8HYHFVAOXMNU7GIA\n"
        "Secret Key\nHgnagOvnlcsmvjHWYuW7YrpDyU35DSarC9gGfIll"
    )
    assert downloaded_ak == "HPUA8HYHFVAOXMNU7GIA"
    assert downloaded_sk == "HgnagOvnlcsmvjHWYuW7YrpDyU35DSarC9gGfIll"
    assert _canonical_uri("/v1/test") == "/v1/test/"
    ip_txt = format_ip_txt(
        [
            {"vlan_id": 11, "mac_address": "fa:16:3e:00:00:0b", "private_ip_address": "192.168.0.11"},
            {"vlan_id": 12, "mac_address": "fa:16:3e:00:00:0c", "private_ip_address": "192.168.0.12"},
            {"_parent_iface": "eth0", "vlan_id": 13, "mac_address": "fa:16:3e:00:00:0d", "private_ip_address": "192.168.0.13"},
            {"_parent_iface": "eth1", "vlan_id": 14, "mac_address": "fa:16:3e:00:00:0e", "private_ip_address": "192.168.0.14"},
        ]
    )
    assert "11 fa:16:3e:00:00:0b 192.168.0.11" in ip_txt
    assert "eth0 " not in ip_txt
    assert "13 fa:16:3e:00:00:0d 192.168.0.13" in ip_txt
    assert "eth1 14 fa:16:3e:00:00:0e 192.168.0.14" in ip_txt
    plan_by_key = {plan.key: plan for plan in REGION_PLANS}
    from HUAWEI.network_plan import aux_should_use_primary_parent
    shanghai_5ip_plan = plan_by_key["shanghai"]
    shanghai_20ip_plan = _huawei_app.HuaweiAutoBuyApp._effective_plan_for_flavor_mode(
        shanghai_5ip_plan,
        FLAVOR_MODE_WAWACLOUD_20,
    )
    # 上海普通 5IP 按普通 5IP 地区处理；20IP 仍通过 effective plan 强制挂主网卡。
    assert not aux_should_use_primary_parent(shanghai_5ip_plan)
    assert aux_should_use_primary_parent(shanghai_20ip_plan)
    sample_results = [
        RegionBuildResult(plan_by_key["east2"], "sid-east2", "115.120.19.183", "pw", ip_txt, ["115.120.19.183"], "SAS", 0),
        RegionBuildResult(plan_by_key["beijing"], "sid-beijing", "123.60.214.189", "pw", ip_txt, ["123.60.214.189"], "SAS", 0),
    ]
    socks_output = format_socks5_output(sample_results)
    assert socks_output.startswith("北京\nIP/端口/用户名/密码")
    assert socks_output.count("IP/端口/用户名/密码") == 2
    assert "Socks5丨免责声明：仅用于合法合规的网络技术交流用途。" in socks_output
    assert "======================================" in socks_output
    assert "IP/端口/账号/密码" not in socks_output
    assert "\n华东\nIP/端口/用户名/密码" in socks_output
    l2tp_output = format_l2tp_output(sample_results)
    assert l2tp_output.startswith("北京\n入口IP/账号/密码/密钥/对应出口IP")
    assert "L2TP丨免责声明：仅用于合法合规的网络技术交流用途。" in l2tp_output
    assert "入口IP/账号/密码/密钥/出口IP" not in l2tp_output
    setattr(sample_results[0], "_huawei_ros_l2tp_only", True)
    assert _huawei_app.HuaweiAutoBuyApp._socks5_results_for_display(sample_results) == [sample_results[1]]
    setattr(sample_results[1], "_huawei_ros_l2tp_only", True)
    assert _huawei_app.HuaweiAutoBuyApp._socks5_results_for_display(sample_results) == []
    assert low_config_flavor_ids("cn-north-4") == ("s7.small.1",)
    assert low_config_flavor_ids("cn-east-5")[0] == "x1.1u.1g"
    assert SCRIPT_FILENAME == "fifteenhuawei.sh"
    assert LOW_CONFIG_REMOTE_COMMAND == "bash <(curl -fsSL https://www.wawaapi.com/static/scripts/Huawei/fifteenhuawei.sh)"
    assert HUAWEI3_REMOTE_COMMAND == "bash <(curl -fsSL https://www.wawaapi.com/static/scripts/Huawei/fifteenhuawei.sh)"
    assert TWENTY_CONFIG_REMOTE_COMMAND == "bash <(curl -s https://218.78.124.34:10055/down/RSUb8ZXJAtNn.sh)"
    assert _huawei_app.HUAWEI_DEFAULT_REMOTE_COMMAND == "bash <(curl -fsSL https://www.wawaapi.com/static/scripts/Huawei/fifteenhuawei-default.sh)"
    assert _huawei_app.HUAWEI_RANDOM_REMOTE_COMMAND == "bash <(curl -fsSL https://www.wawaapi.com/static/scripts/Huawei/fifteenhuawei.sh)"
    assert _huawei_app.HUAWEI_20_DEFAULT_REMOTE_COMMAND == "bash <(curl -fsSL https://www.wawaapi.com/static/scripts/Huawei/fifteenhuawei20-default.sh)"
    assert _huawei_app.HUAWEI_20_RANDOM_REMOTE_COMMAND == "bash <(curl -fsSL https://www.wawaapi.com/static/scripts/Huawei/fifteenhuawei20.sh)"
    assert _huawei_app.HUAWEI_SSH_PORT_WAIT_TIMEOUT == 35
    assert _huawei_app.SHANGHAI_NEW_PURCHASE_CONFIRM_ATTEMPTS == 8
    assert _huawei_app.SHANGHAI_NEW_PURCHASE_BUILD_RETRY_ATTEMPTS == 3
    assert _huawei_app.HuaweiAutoBuyApp._is_autobuy_immediate_progress("步骤: 自动购买预检开始")
    assert _huawei_app.HuaweiAutoBuyApp._is_autobuy_immediate_progress("步骤: 自动购买地区并发=2/7")
    from HUAWEI import ros_builder
    assert ros_builder.chr_image_path().exists()
    assert ros_builder.find_ros_asset(ros_builder.ROS_FIVE_REGION_TEMPLATE).exists()
    assert ros_builder.find_ros_asset(ros_builder.ROS_TEN_REGION_TEMPLATE).exists()
    assert os.path.exists(os.path.join(huawei_wwwroot_source_dir(), "main.py"))
    assert _volc_app.VOLC_DEFAULT_REMOTE_COMMAND == "bash <(curl -fsSL https://www.wawaapi.com/static/scripts/Volc/volcengine-default.sh)"
    assert _volc_app.VOLC_RANDOM_REMOTE_COMMAND == "bash <(curl -fsSL https://www.wawaapi.com/static/scripts/Volc/volcengine.sh)"
    assert [plan.label for plan in VOLC_REGION_PLANS] == ["北京", "上海", "广州"]
    assert VOLC_CONFIGS[VOLC_CONFIG_FULL]["total_ips"] == 24
    assert VOLC_CONFIGS[VOLC_CONFIG_BIG_FULL]["total_ips"] == 48
    assert VOLC_SPECS[VOLC_SPEC_2C2G]["instance_type"] == "ecs.e-c1m1.large"
    assert VOLC_SPECS[VOLC_SPEC_2C4G]["instance_type"] == "ecs.e-c1m2.large"
    assert VOLC_IP_QUOTA_PER_REGION == 8
    from CTYUN.app import load_regions as load_ctyun_regions
    assert ctyun_workflow_gui.CTYUN_DEFAULT_SSH_COMMAND == "bash <(curl -fsSL https://www.wawaapi.com/static/scripts/ctyun/ctyun-default.sh)"
    assert ctyun_workflow_gui.CTYUN_RANDOM_SSH_COMMAND == "bash <(curl -fsSL https://www.wawaapi.com/static/scripts/ctyun/ctyun.sh)"
    ctyun_regions = load_ctyun_regions()
    assert len(ctyun_regions) >= 10
    assert "杭州2" in ctyun_regions
    assert HuaweiCloudClient.server_matches_purchase_config(
        {"name": LOW_CONFIG_ECS_NAME, "flavor": {"id": "s7.small.1", "vcpus": "1", "ram": "1024"}},
        FLAVOR_MODE_WAWACLOUD_LOW,
        "cn-north-4",
    )
    assert HuaweiCloudClient.server_matches_purchase_config(
        {"name": "ecs-qingdao-existing", "flavor": {"id": "x1.1u.1g", "vcpus": "1", "ram": "1024"}},
        FLAVOR_MODE_WAWACLOUD_LOW,
        "cn-east-5",
    )
    assert not HuaweiCloudClient.server_matches_purchase_config(
        {"name": "ecs-beijing-0526164743", "flavor": {"id": "s6.small.1", "vcpus": "1", "ram": "1024"}},
        FLAVOR_MODE_WAWACLOUD_LOW,
        "cn-north-4",
    )
    sub_ecs = {
        "service_type_code": "hws.service.type.ecs",
        "resource_type_code": "hws.resource.type.vm",
        "region_code": "cn-north-4",
        "status": 2,
        "resource_id": "server-1",
    }
    sub_eip = {
        "resource_type_name": "弹性公网IP",
        "region_code": "cn-east-3",
        "status": 2,
        "cloud_resource_id": "eip-1",
    }
    assert is_ecs_subscription_resource(sub_ecs)
    assert is_eip_subscription_resource(sub_eip)
    assert subscription_operation_kind(sub_ecs) == "subscription_unsubscribe"
    assert subscription_operation_kind({**sub_ecs, "status": 3}) == "none"
    assert resource_id_candidates(sub_eip) == ["eip-1"]
    region_project_client = HuaweiCloudClient("test-ak", "test-sk", "test-domain")
    region_project_calls = []

    def fake_region_project_request(method, url, body=None, **kwargs):
        region_project_calls.append((method, url, kwargs))
        if url.endswith("/v3/auth/projects"):
            return 200, {"projects": [{"name": "cn-north-9", "id": "project-wulanchabu"}]}, ""
        return 200, {"domains": [{"id": "test-domain"}]}, ""

    region_project_client.request = fake_region_project_request
    region_projects = region_project_client.warmup_region("cn-north-9")
    assert [item[1] for item in region_project_calls] == [
        "https://iam.cn-north-9.myhuaweicloud.com/v3/auth/domains",
        "https://iam.cn-north-9.myhuaweicloud.com/v3/auth/projects",
    ]
    assert region_project_calls[1][2]["use_domain_header"]
    assert region_projects == [{"name": "cn-north-9", "id": "project-wulanchabu"}]
    from huaweicloudsdkecs.v2 import EcsClient, CreateServersRequest, ShowJobRequest, ListServersDetailsRequest
    from huaweicloudsdkeip.v2 import EipClient, CreatePublicipRequest, UpdatePublicipRequest, ListPublicipsRequest
    from huaweicloudsdkiam.v3 import IamClient, KeystoneListProjectsRequest, KeystoneListAuthDomainsRequest
    from huaweicloudsdkims.v2 import ImsClient, ListImagesRequest
    from huaweicloudsdkvpc.v2 import VpcClient as HuaweiVpcV2Client, CreateVpcRequest, ListVpcsRequest, CreateSubnetRequest
    from huaweicloudsdkvpc.v3 import VpcClient as HuaweiVpcV3Client, CreateSubNetworkInterfaceRequest
    import volcenginesdkbilling
    import volcenginesdkcore
    import volcenginesdkecs
    import volcenginesdksts
    import volcenginesdkvpc
    assert EcsClient and CreateServersRequest and ShowJobRequest and ListServersDetailsRequest
    assert EipClient and CreatePublicipRequest and UpdatePublicipRequest and ListPublicipsRequest
    assert IamClient and KeystoneListProjectsRequest and KeystoneListAuthDomainsRequest
    assert ImsClient and ListImagesRequest
    assert HuaweiVpcV2Client and CreateVpcRequest and ListVpcsRequest and CreateSubnetRequest
    assert HuaweiVpcV3Client and CreateSubNetworkInterfaceRequest
    assert volcenginesdkcore and volcenginesdkecs and volcenginesdkvpc and volcenginesdkbilling and volcenginesdksts
    assert ThreadPoolExecutor and as_completed
    print("self-test ok")


def main() -> None:
    cleanup_stale_pyinstaller_dirs()
    set_windows_app_user_model_id()
    ensure_portable_desktop_shortcut(APP_TITLE)
    if len(sys.argv) > 1 and sys.argv[1] == "--huawei-wwwroot-worker":
        raise SystemExit(run_huawei_wwwroot_worker())
    if len(sys.argv) > 1 and sys.argv[1] == "--volc-worker":
        raise SystemExit(run_volc_wwwroot_worker())
    if len(sys.argv) > 1 and sys.argv[1] == "--ctyun-worker":
        sys.argv = [sys.argv[0], "--worker", *sys.argv[2:]]
        raise SystemExit(ctyun_workflow_gui.worker_main())
    if "--self-test" in sys.argv:
        run_self_test()
        return
    if not show_startup_notice_window():
        return
    app = MultiCloudManagerApp()
    app.mainloop()


if __name__ == "__main__":
    main()




























