r"""Apple Mobile Device Support（Windows 上的 usbmuxd 等价物）检测与自动安装。

**为什么需要这个模块**

macOS 自带 usbmuxd；Windows 上 usbmuxd 由 Apple 的 "Apple Mobile Device Service"
提供（随 iTunes / Apple Mobile Device Support 安装）。它监听 **127.0.0.1:27015**，
pymobiledevice3 通过它发现 iPhone。没有它，`pymobiledevice3 usbmux list` 一台设备也看不到。

**关键事实（已在真机取证，见 docs/DEVICE_NOTES_WINDOWS.md）**

1. 服务名是带空格的 ``Apple Mobile Device Service``；``AppleMobileDeviceService`` 只是
   exe 文件名，`sc query AppleMobileDeviceService` 会报 1060。
2. Windows 上 **没有** ``\\.\pipe\usbmuxd``，只有 TCP 27015。
3. ProductCode 不稳定（每个版本都变），**UpgradeCode 稳定**：
   ``{5CF0656E-F3B8-11DB-BA81-804E56D89593}``（x86/x64 通用）。
4. **AMDS 不提供 USB 驱动**。它的 MSI 里 0 个 .inf/.sys/.cat；现代机器上 iPhone 实际绑定
   的是 ``appleusb.inf``（服务 WINUSB / usbccgp）。所以**不要**用 `usbaapl64` 判断驱动是否存在
   —— 健康的机器上这个判断会误报"缺失"。
5. 判断"到底能不能用"的**唯一权威信号**是 27015 上的协议往返（``ReadBUID`` 不需要插设备）。
"""
from __future__ import annotations

import ctypes
import hashlib
import json
import os
import plistlib
import shutil
import socket
import struct
import subprocess
import sys
import time
import urllib.request
from pathlib import Path
from typing import Any, Callable

from . import platform as plat

#: 服务名（带空格！）
SERVICE_NAME = "Apple Mobile Device Service"
#: AMDS 的稳定标识（UpgradeCode，x86/x64 通用，跨版本不变）
AMDS_UPGRADE_CODE = "{5CF0656E-F3B8-11DB-BA81-804E56D89593}"
#: AMDS 安装包在注册表 Uninstall 下的显示名
AMDS_DISPLAY_NAME = "Apple Mobile Device Support"

USBMUX_HOST = "127.0.0.1"
USBMUX_PORT = 27015

# ---------------------------------------------------------------------------
# 官方下载地址（2026-10 实测可达）
# ---------------------------------------------------------------------------
ITUNES_WIN64_URL = "https://www.apple.com/itunes/download/win64"
ITUNES_WIN32_URL = "https://www.apple.com/itunes/download/win32"

#: 独立 AMDS MSI（约 38 MB，比整个 iTunes 包 193 MB 小得多）
AMDS_MSI_X64_URL = (
    "https://swcdn.apple.com/content/downloads/20/49/047-76422/"
    "qcw2a7028lr8yp4rdkrfyzkvlhi9q1gg2g/AppleMobileDeviceSupport64.msi"
)
AMDS_MSI_X64_SHA256 = "b60533fb54e7bd81ffc52d99678a9cce04e58cb54a76281bd7b1309f20d360e9"
AMDS_MSI_X86_URL = (
    "https://swcdn.apple.com/content/downloads/54/12/047-76417/"
    "rftfhqmqnu0afv25sdr2iosg38mxkqa388/AppleMobileDeviceSupport.msi"
)
AMDS_MSI_X86_SHA256 = "590e4b21d2aa522cfe789cc3fea95421b6141ddfa36e06525ae9715a5b5859e3"

#: Apple Application Support（AMDS 的前置依赖，安装顺序由 iTunes 包自己决定）
AAS_MSI_X64_URL = (
    "https://swcdn.apple.com/content/downloads/63/47/001-39934/"
    "955xl2euy0wb5j2z4oddbbd39ln1sthedn/AppleApplicationSupport64.msi"
)

#: msiexec / 安装程序的"成功"退出码
SUCCESS_CODES = {0, 3010, 1641, 1638}
CODE_MEANING = {
    0: "成功",
    3010: "成功（需要重启才能完全生效）",
    1641: "成功（安装程序已自动重启）",
    1638: "已安装该产品的其它版本（视为已就绪）",
    1603: "安装失败（致命错误）",
    1618: "另一个安装正在进行中，请稍后重试",
    1619: "安装包无法打开（下载可能不完整）",
    1620: "安装包无法打开（文件损坏）",
    1260: "组策略禁止了本次安装",
    5: "拒绝访问（需要管理员权限）",
}

_TIMEOUT_PROBE = 5.0


# ---------------------------------------------------------------------------
# 底层工具
# ---------------------------------------------------------------------------
def _no_window() -> dict[str, Any]:
    if plat.IS_WINDOWS:
        return {"creationflags": getattr(subprocess, "CREATE_NO_WINDOW", 0x08000000)}
    return {}


def _run(command: list[str], timeout: float = 60.0) -> tuple[int, str]:
    """执行外部命令，返回 (returncode, 合并输出)。失败不抛异常。"""
    try:
        result = subprocess.run(
            command, capture_output=True, text=True, timeout=timeout, **_no_window()
        )
        output = ((result.stdout or "") + (result.stderr or "")).strip()
        return result.returncode, output
    except subprocess.TimeoutExpired:
        return -1, "命令超时"
    except FileNotFoundError:
        return -2, f"找不到命令：{command[0]}"
    except Exception as exc:  # pragma: no cover
        return -3, repr(exc)


def _decode(data: bytes) -> str:
    """Windows 中文控制台输出多为 GBK，尽量解出来。"""
    for encoding in ("utf-8", "gbk", "mbcs"):
        try:
            return data.decode(encoding)
        except (UnicodeDecodeError, LookupError):
            continue
    return data.decode("utf-8", "replace")


# ---------------------------------------------------------------------------
# 检测
# ---------------------------------------------------------------------------
def service_state() -> dict[str, Any]:
    """查询 ``Apple Mobile Device Service`` 的状态。"""
    if not plat.IS_WINDOWS:
        return {"applicable": False, "exists": False, "running": False, "state": "n/a"}
    code, output = _run(["sc", "query", SERVICE_NAME], timeout=15)
    text = output.upper()
    exists = code == 0 and "1060" not in output
    running = "RUNNING" in text
    state = "unknown"
    for token in ("RUNNING", "STOPPED", "START_PENDING", "STOP_PENDING", "PAUSED"):
        if token in text:
            state = token
            break
    if not exists:
        state = "not_installed"
    return {
        "applicable": True,
        "exists": exists,
        "running": running,
        "state": state,
        "raw": output.strip(),
    }


def start_service() -> tuple[bool, str]:
    """尝试启动服务。已有的服务只是没启动，是"看不到设备"最常见的原因。"""
    if not plat.IS_WINDOWS:
        return False, "非 Windows 平台"
    code, output = _run(["sc", "start", SERVICE_NAME], timeout=60)
    if code == 0 or "1056" in output:  # 1056 = 已在运行
        return True, "服务已启动"
    return False, output.strip() or f"sc start 返回 {code}"


def restart_service() -> tuple[bool, str]:
    """重启 Apple 服务（先停再起）。

    用于「服务装着、也在跑，但 27015 协议往返失败」的场景 —— 通常是服务卡死。
    比重新安装（几十到几百 MB 下载）便宜得多，应该优先尝试。
    """
    if not plat.IS_WINDOWS:
        return False, "非 Windows 平台"
    _run(["sc", "stop", SERVICE_NAME], timeout=90)
    # 等它真的停下来（服务停止是异步的，立刻 start 可能报 1056/1052）
    deadline = time.time() + 25
    while time.time() < deadline:
        code, output = _run(["sc", "query", SERVICE_NAME], timeout=15)
        if "STOPPED" in output.upper() or "1060" in output:
            break
        time.sleep(1.0)
    code, output = _run(["sc", "start", SERVICE_NAME], timeout=90)
    if code == 0 or "1056" in output:
        return True, "服务已重启"
    return False, output.strip() or f"sc start 返回 {code}"


def usbmux_probe(
    message_type: str = "ReadBUID",
    host: str = USBMUX_HOST,
    port: int = USBMUX_PORT,
    timeout: float = _TIMEOUT_PROBE,
) -> dict[str, Any]:
    """在 27015 上做一次 usbmuxd 协议往返 —— **判断可用性的唯一权威信号**。

    ``ReadBUID`` 不需要插设备，因此可以在"有没有手机"之外单独验证服务是否真的工作。

    协议：16 字节小端头 ``<IIII`` = (length, version=1, message=8, tag)，
    然后是 XML plist；``length`` 含头。
    """
    body = {
        "MessageType": message_type,
        "ProgName": "iOSLocationTrack",
        "ClientVersionString": "iOSLocationTrack 1.0",
        "kLibUSBMuxVersion": 3,
    }
    try:
        payload = plistlib.dumps(body)
    except Exception as exc:  # pragma: no cover
        return {"ok": False, "error": f"构造 plist 失败：{exc}"}

    header = struct.pack("<IIII", 16 + len(payload), 1, 8, 1)
    try:
        with socket.create_connection((host, port), timeout=timeout) as sock:
            sock.sendall(header + payload)
            raw = b""
            while len(raw) < 4:
                chunk = sock.recv(4 - len(raw))
                if not chunk:
                    break
                raw += chunk
            if len(raw) < 4:
                return {"ok": False, "error": "连接被对端关闭（无响应头）"}
            total = struct.unpack("<I", raw[:4])[0]
            if not (16 <= total <= 4 * 1024 * 1024):
                return {"ok": False, "error": f"响应长度异常：{total}"}
            while len(raw) < total:
                chunk = sock.recv(total - len(raw))
                if not chunk:
                    break
                raw += chunk
    except ConnectionRefusedError:
        return {"ok": False, "error": f"{host}:{port} 拒绝连接（服务未运行或未安装）"}
    except socket.timeout:
        return {"ok": False, "error": f"{host}:{port} 连接超时"}
    except OSError as exc:
        return {"ok": False, "error": f"{host}:{port} 不可达：{exc}"}

    try:
        length, version, message, tag = struct.unpack("<IIII", raw[:16])
        parsed = plistlib.loads(raw[16:total])
    except Exception as exc:
        return {"ok": False, "error": f"解析响应失败：{exc}"}

    # 成功的两种形态：{"MessageType":"Result","Number":0} 或直接带数据（如 ListDevices）
    if isinstance(parsed, dict):
        number = parsed.get("Number")
        if parsed.get("MessageType") == "Result" and number not in (0, None):
            return {
                "ok": False,
                "error": f"usbmuxd 返回错误码 {number}",
                "plist": parsed,
            }
    return {
        "ok": True,
        "plist": parsed,
        "header": {"length": length, "version": version, "message": message, "tag": tag},
    }


def list_devices() -> dict[str, Any]:
    """通过 usbmuxd 列出设备（USB + Wi-Fi）。"""
    result = usbmux_probe("ListDevices")
    if not result.get("ok"):
        return result
    devices = (result.get("plist") or {}).get("DeviceList") or []
    out = []
    for item in devices:
        props = item.get("Properties") or {}
        out.append(
            {
                "device_id": item.get("DeviceID"),
                "connection": props.get("ConnectionType", "?"),
                "product_id": props.get("ProductID"),
                "serial": props.get("SerialNumber", ""),
            }
        )
    return {"ok": True, "devices": out, "count": len(out)}


def _registry_views():
    """枚举 32/64 位两个注册表视图的 Uninstall 键（32 位 AMDS 落在 WOW6432Node）。"""
    import winreg  # type: ignore[import-not-found]

    root = r"SOFTWARE\Microsoft\Windows\CurrentVersion\Uninstall"
    for hive in (winreg.HKEY_LOCAL_MACHINE, winreg.HKEY_CURRENT_USER):
        for view in (winreg.KEY_WOW64_64KEY, winreg.KEY_WOW64_32KEY):
            for sub in ("", r"\WOW6432Node"):
                path = root + sub if sub else root
                try:
                    with winreg.OpenKey(hive, path, 0, winreg.KEY_READ | view) as key:
                        count = winreg.QueryInfoKey(key)[0]
                        for index in range(count):
                            try:
                                name = winreg.EnumKey(key, index)
                                with winreg.OpenKey(key, name) as entry:
                                    yield entry
                            except OSError:
                                continue
                except OSError:
                    continue


def installed_products() -> list[dict[str, Any]]:
    """从注册表读出 Apple 相关产品的版本信息（仅用于展示/诊断）。"""
    if not plat.IS_WINDOWS:
        return []
    wanted = ("apple mobile device support", "itunes", "apple application support",
              "apple devices", "bonjour")
    found: dict[str, dict[str, Any]] = {}
    for entry in _registry_views():
        try:
            import winreg  # type: ignore[import-not-found]

            display = str(winreg.QueryValueEx(entry, "DisplayName")[0])
        except OSError:
            continue
        low = display.lower()
        if not any(w in low for w in wanted):
            continue
        record: dict[str, Any] = {"display_name": display}
        for field in ("DisplayVersion", "Publisher", "InstallLocation", "UninstallString"):
            try:
                import winreg  # type: ignore[import-not-found]

                value = winreg.QueryValueEx(entry, field)[0]
                record[field.lower()] = str(value)
            except OSError:
                continue
        # 用 UpgradeCode 精确识别 AMDS（ProductCode 不稳定，不能硬编码）
        try:
            import winreg  # type: ignore[import-not-found]

            props = winreg.OpenKey(entry, "InstallProperties")
            with props:
                upgrade = str(winreg.QueryValueEx(props, "UpgradeCode")[0])
                record["upgrade_code"] = upgrade
        except OSError:
            pass
        key = f"{display}|{record.get('displayversion','')}"
        found[key] = record
    return sorted(found.values(), key=lambda r: r["display_name"])


def amds_product() -> dict[str, Any] | None:
    """返回已安装的 AMDS 记录（按 UpgradeCode 或显示名匹配）。"""
    for item in installed_products():
        if (item.get("upgrade_code") or "").upper() == AMDS_UPGRADE_CODE.upper():
            return item
        if (item.get("display_name") or "").strip().lower() == AMDS_DISPLAY_NAME.lower():
            return item
    return None


def usb_driver_status() -> dict[str, Any]:
    """检查 iPhone 的 USB 驱动绑定情况。

    **不要**用 ``usbaapl`` 判断是否存在 —— 现代机器上健康设备绑定的是 ``WINUSB``/
    ``usbccgp``（驱动文件 ``appleusb.inf``）。
    """
    if not plat.IS_WINDOWS:
        return {"applicable": False}
    ps = shutil.which("powershell") or shutil.which("pwsh")
    if not ps:
        return {"applicable": True, "checked": False, "error": "找不到 PowerShell"}
    script = (
        "Get-PnpDevice -ErrorAction SilentlyContinue | "
        "Where-Object { $_.InstanceId -like '*VID_05AC*' } | "
        "Select-Object Status,Class,FriendlyName,InstanceId,Service | ConvertTo-Json -Compress"
    )
    code, output = _run([ps, "-NoProfile", "-NonInteractive", "-Command", script], timeout=45)
    devices: list[dict[str, Any]] = []
    if output.strip():
        try:
            data = json.loads(output.strip())
            devices = data if isinstance(data, list) else [data]
        except Exception:
            devices = []
    healthy_tokens = ("usbaapl", "winusb", "appleusb", "usbccgp")
    bound = [
        d for d in devices
        if any(t in str(d.get("Service") or "").lower() for t in healthy_tokens)
    ]
    # MTP（照片）设备绑定 WUDFWpdMtp，不能算作"定位可用"，但说明线材/端口 OK
    return {
        "applicable": True,
        "checked": True,
        "devices": [
            {
                "name": d.get("FriendlyName"),
                "status": d.get("Status"),
                "service": d.get("Service"),
                "instance": d.get("InstanceId"),
            }
            for d in devices
        ],
        "bound_count": len(bound),
        "any_apple_device": len(devices) > 0,
    }


def status(deep: bool = True) -> dict[str, Any]:
    """汇总状态。``deep=True`` 时做 27015 协议往返（权威可用性判定）。"""
    if not plat.IS_WINDOWS:
        return {
            "applicable": False,
            "ready": True,
            "summary": "非 Windows 平台，usbmuxd 由系统自带",
        }
    service = service_state()
    product = amds_product()
    probe: dict[str, Any] = {"ok": False, "error": "未检测"}
    devices: dict[str, Any] = {"ok": False, "count": 0, "devices": []}
    if deep:
        probe = usbmux_probe("ReadBUID")
        if probe.get("ok"):
            devices = list_devices()

    ready = bool(service["running"] and probe.get("ok"))
    if ready:
        summary = "Apple Mobile Device Service 正在运行，usbmuxd 可通信"
    elif not service["exists"]:
        summary = "未安装 Apple Mobile Device Support（需要安装才能发现 iPhone）"
    elif not service["running"]:
        summary = "Apple Mobile Device Service 已安装但未运行"
    else:
        summary = "服务在运行，但 usbmuxd（127.0.0.1:27015）不可通信"

    return {
        "applicable": True,
        "ready": ready,
        "summary": summary,
        "needs_install": not service["exists"] and product is None,
        "needs_start": bool(service["exists"] and not service["running"]),
        "service": service,
        "product": product,
        "probe": {k: v for k, v in probe.items() if k != "plist"},
        "devices": devices,
        "driver": usb_driver_status() if deep else {"checked": False},
        "is_admin": plat.is_admin(),
    }


# ---------------------------------------------------------------------------
# 安装
# ---------------------------------------------------------------------------
def _download(url: str, dest: Path, log: Callable[[str, str], None] | None = None,
              expected_sha256: str | None = None) -> tuple[bool, str]:
    """下载文件并可选校验 SHA256。支持断点续传（.part）。"""
    dest.parent.mkdir(parents=True, exist_ok=True)
    part = dest.with_suffix(dest.suffix + ".part")

    def emit(level: str, message: str) -> None:
        if log:
            log(level, message)

    if dest.exists() and expected_sha256:
        if _sha256(dest) == expected_sha256.lower():
            return True, f"已存在且校验通过：{dest.name}"

    emit("info", f"下载 {url.split('/')[-1]} …")
    request = urllib.request.Request(url, headers={"User-Agent": "iOSLocationTrack/1.0"})
    proxy = os.environ.get("HTTPS_PROXY") or os.environ.get("https_proxy")
    try:
        handlers: list[Any] = []
        if proxy:
            handlers.append(urllib.request.ProxyHandler({"https": proxy, "http": proxy}))
        opener = urllib.request.build_opener(*handlers)
        with opener.open(request, timeout=60) as response, open(part, "wb") as handle:
            total = int(response.headers.get("Content-Length") or 0)
            done = 0
            last_report = 0.0
            while True:
                chunk = response.read(256 * 1024)
                if not chunk:
                    break
                handle.write(chunk)
                done += len(chunk)
                now = time.monotonic()
                if total and now - last_report > 2.0:
                    last_report = now
                    emit("info", f"  … {done/1048576:.1f}/{total/1048576:.1f} MB "
                                 f"({done*100//total}%)")
    except Exception as exc:
        try:
            part.unlink(missing_ok=True)
        except OSError:
            pass
        return False, f"下载失败：{exc}"

    if expected_sha256:
        actual = _sha256(part)
        if actual != expected_sha256.lower():
            part.unlink(missing_ok=True)
            return False, (f"SHA256 校验失败：期望 {expected_sha256[:16]}…，"
                           f"实际 {actual[:16]}…（文件可能已更新，请报告）")
    try:
        part.replace(dest)
    except OSError as exc:
        return False, f"保存失败：{exc}"
    return True, f"已下载：{dest}"


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


def _msiexec(msi: Path, log: Callable[[str, str], None] | None = None,
             extra: list[str] | None = None) -> tuple[int, str]:
    """静默安装一个 MSI，返回 (退出码, 输出)。"""
    logfile = plat.data_dir() / "logs" / f"msi-{msi.stem}.log"
    logfile.parent.mkdir(parents=True, exist_ok=True)
    command = [
        "msiexec", "/i", str(msi), "/qn", "/norestart", "/passive",
        f"/l*v", str(logfile),
    ]
    if extra:
        command += extra
    # /qn 与 /passive 同时给出时以 /qn 为准；这里保留 /qn 保证全静默
    command = [c for c in command if c != "/passive"]
    if log:
        log("info", f"执行：{' '.join(command)}")
    code, output = _run(command, timeout=1800)
    return code, output


def install_amds(log: Callable[[str, str], None] | None = None,
                 allow_download: bool = True) -> dict[str, Any]:
    """策略 B：只装 Apple Mobile Device Support（约 38 MB，推荐）。

    顺序：Apple Application Support x64 → AMDS x64。iTunes 包自己就是这个顺序。
    """
    def emit(level: str, message: str) -> None:
        if log:
            log(level, message)

    if not plat.IS_WINDOWS:
        return {"ok": False, "error": "仅 Windows 支持"}
    if not plat.is_admin():
        return {"ok": False, "error": "需要管理员权限才能安装 Apple Mobile Device Support",
                "needs_admin": True}
    if not allow_download:
        return {"ok": False, "error": "已禁用下载"}

    arch64 = struct.calcsize("P") * 8 == 64
    cache = plat.data_dir() / "installers"
    steps: list[dict[str, Any]] = []

    plans = []
    if arch64:
        plans.append(("Apple Application Support x64", AAS_MSI_X64_URL, None, True))
    url = AMDS_MSI_X64_URL if arch64 else AMDS_MSI_X86_URL
    sha = AMDS_MSI_X64_SHA256 if arch64 else AMDS_MSI_X86_SHA256
    plans.append(("Apple Mobile Device Support", url, sha, False))

    for label, url, sha, optional in plans:
        name = url.rsplit("/", 1)[-1]
        target = cache / name
        emit("info", f"=== {label} ===")
        ok, message = _download(url, target, log, sha)
        if not ok:
            if optional:
                emit("warn", f"{label} 下载失败（可选前置，继续）：{message}")
                steps.append({"step": label, "ok": False, "skipped": True, "message": message})
                continue
            return {"ok": False, "error": message, "steps": steps}
        emit("info", message)
        code, output = _msiexec(target, log)
        meaning = CODE_MEANING.get(code, f"退出码 {code}")
        emit("info" if code in SUCCESS_CODES else "error", f"{label}：{meaning}")
        steps.append({"step": label, "code": code, "ok": code in SUCCESS_CODES,
                      "message": meaning})
        if code not in SUCCESS_CODES:
            tail = "\n".join(output.strip().splitlines()[-8:])
            return {"ok": False, "error": f"{label} 安装失败：{meaning}",
                    "steps": steps, "output": tail}

    return {"ok": True, "steps": steps, "strategy": "amds_msi",
            "reboot_required": any(s.get("code") in (3010, 1641) for s in steps)}


def install_itunes(log: Callable[[str, str], None] | None = None) -> dict[str, Any]:
    """策略 A：装完整 iTunes 包（约 193 MB，Apple 官方推荐路径，最稳）。"""
    def emit(level: str, message: str) -> None:
        if log:
            log(level, message)

    if not plat.IS_WINDOWS:
        return {"ok": False, "error": "仅 Windows 支持"}
    if not plat.is_admin():
        return {"ok": False, "error": "需要管理员权限才能安装 iTunes", "needs_admin": True}

    cache = plat.data_dir() / "installers"
    target = cache / "iTunes64Setup.exe"
    ok, message = _download(ITUNES_WIN64_URL, target, log, None)
    if not ok:
        return {"ok": False, "error": message}
    emit("info", f"{message}（{target.stat().st_size/1048576:.0f} MB）")
    emit("warn", "安装期间 iTunes / Outlook 可能被自动关闭，请先保存工作")
    command = [str(target), "/quiet", "/norestart", "/log",
               str(plat.data_dir() / "logs" / "itunes-install.log")]
    emit("info", f"执行：{' '.join(command)}（可能需要几分钟）")
    code, output = _run(command, timeout=3600)
    meaning = CODE_MEANING.get(code, f"退出码 {code}")
    ok = code in SUCCESS_CODES
    emit("info" if ok else "error", f"iTunes 安装：{meaning}")
    return {"ok": ok, "strategy": "itunes_bundle", "code": code, "message": meaning,
            "reboot_required": code in (3010, 1641),
            "error": None if ok else f"iTunes 安装失败：{meaning}"}


def install_winget(log: Callable[[str, str], None] | None = None) -> dict[str, Any]:
    """策略 C：用 winget 装（依赖 winget 可用且能联网）。"""
    def emit(level: str, message: str) -> None:
        if log:
            log(level, message)

    if not plat.IS_WINDOWS:
        return {"ok": False, "error": "仅 Windows 支持"}
    winget = shutil.which("winget")
    if not winget:
        return {"ok": False, "error": "未找到 winget"}
    last: dict[str, Any] = {"ok": False, "error": "所有 winget 包都失败"}
    for package_id in ("Apple.AppleMobileDeviceSupport", "Apple.iTunes"):
        command = [winget, "install", "--id", package_id, "--exact", "--silent",
                   "--accept-package-agreements", "--accept-source-agreements",
                   "--disable-interactivity"]
        emit("info", f"执行：{' '.join(command)}")
        code, output = _run(command, timeout=3600)
        if code == 0:
            emit("info", f"{package_id} 安装完成")
            return {"ok": True, "strategy": f"winget:{package_id}"}
        emit("warn", f"{package_id} 失败（{code}）：{output.strip()[-300:]}")
        last = {"ok": False, "error": f"{package_id} 安装失败（退出码 {code}）",
                "output": output.strip()[-500:]}
    return last


def ensure(
    log: Callable[[str, str], None] | None = None,
    allow_install: bool = True,
    strategy: str = "auto",
) -> dict[str, Any]:
    """**主入口**：检测 → 必要时安装 → 复检。

    ``strategy``: ``auto``（推荐）| ``msi`` | ``itunes`` | ``winget`` | ``none``
    """
    def emit(level: str, message: str) -> None:
        if log:
            log(level, message)

    if not plat.IS_WINDOWS:
        return {"ok": True, "applicable": False, "action": "none",
                "status": status(deep=False)}

    before = status(deep=True)
    if before["ready"]:
        emit("info", "Apple Mobile Device Support 已就绪，无需安装")
        return {"ok": True, "applicable": True, "action": "none", "status": before}

    # 已安装但没运行 —— 先尝试启动（最常见、也最便宜）
    if before["needs_start"]:
        emit("warn", "服务已安装但未运行，尝试启动…")
        ok, message = start_service()
        emit("info" if ok else "warn", message)
        if ok:
            time.sleep(1.5)
            after_start = status(deep=True)
            if after_start["ready"]:
                return {"ok": True, "applicable": True, "action": "start_service",
                        "status": after_start}

    # 服务存在（甚至在运行）但 27015 探针不通 —— 多半是服务卡死。
    # **先重启服务**，不要一上来就走安装阶梯：AMDS 明明已经装好，
    # 却去重新执行 38MB MSI、再下载 193MB 的 iTunes，完全是浪费且可能装坏。
    if before["service"]["exists"]:
        emit("warn", "Apple 服务已安装但 usbmuxd(27015) 不通，先尝试重启服务…")
        for attempt in (1, 2):
            ok, message = restart_service()
            emit("info" if ok else "warn", f"重启服务（第 {attempt} 次）：{message}")
            time.sleep(2.0 * attempt)
            after_restart = status(deep=True)
            if after_restart["ready"]:
                return {"ok": True, "applicable": True, "action": "restart_service",
                        "status": after_restart}
        if not allow_install or strategy == "none":
            return {"ok": False, "applicable": True, "action": "restart_service",
                    "status": after_restart,
                    "error": "Apple 服务已安装但 usbmuxd 仍不可用（已尝试重启服务）。"
                             "请检查是否有防火墙/安全软件拦截 127.0.0.1:27015，"
                             "或断开重连数据线后重试。"}
        # 服务存在但重启无效：仅在用户显式要求时才强制重装，避免无谓的大体积下载
        emit("warn", "重启服务仍无效。若确认驱动损坏，可手动强制重装 iTunes："
                     "https://www.apple.com/itunes/download/win64")

    if not allow_install or strategy == "none":
        return {"ok": False, "applicable": True, "action": "none",
                "needs_install": True, "status": before,
                "error": before["summary"]}

    if not plat.is_admin():
        return {"ok": False, "applicable": True, "action": "blocked",
                "needs_admin": True, "status": before,
                "error": "安装 Apple Mobile Device Support 需要管理员权限，"
                         "请以管理员身份重新运行程序"}

    emit("warn", f"未检测到可用的 usbmuxd：{before['summary']}")
    emit("info", "开始自动安装 Apple Mobile Device Support（首次约 40~200 MB）…")

    attempts: list[dict[str, Any]] = []
    order = {
        "auto": [("msi", install_amds), ("itunes", install_itunes), ("winget", install_winget)],
        "msi": [("msi", install_amds)],
        "itunes": [("itunes", install_itunes)],
        "winget": [("winget", install_winget)],
    }.get(strategy) or []

    for name, function in order:
        emit("info", f"--- 安装策略：{name} ---")
        result = function(log)
        attempts.append({"strategy": name, **{k: v for k, v in result.items()
                                              if k not in ("steps",)}})
        # 装成功（或要求重启）时，给服务一次起来的机会再复检；
        # 失败则直接落到下一个策略（msi → itunes → winget 逐级降级）
        if result.get("ok") or result.get("reboot_required"):
            start_service()
            time.sleep(2.0)
            after = status(deep=True)
            if after["ready"]:
                return {"ok": True, "applicable": True, "action": f"installed:{name}",
                        "status": after, "attempts": attempts,
                        "reboot_required": bool(result.get("reboot_required"))}
            if result.get("reboot_required"):
                return {"ok": False, "applicable": True, "action": f"installed:{name}",
                        "needs_reboot": True, "status": after, "attempts": attempts,
                        "error": "安装完成但需要重启 Windows 后 usbmuxd 才会生效"}
        if result.get("error"):
            emit("warn", f"策略 {name} 未成功：{result['error']}")

    final = status(deep=True)
    if final["ready"]:
        return {"ok": True, "applicable": True, "action": "installed",
                "status": final, "attempts": attempts}
    return {
        "ok": False, "applicable": True, "action": "failed", "status": final,
        "attempts": attempts,
        "error": "自动安装未能建立可用的 usbmuxd。可手动安装 iTunes："
                 "https://www.apple.com/itunes/download/win64",
    }


USBMUX_PIPE = r"\\.\pipe\usbmuxd"


# ---------------------------------------------------------------------------
# 开发者镜像（个性化镜像，iOS 17+ 全系共用）
# ---------------------------------------------------------------------------
#: 镜像来源：doronz88/DeveloperDiskImage（pymobiledevice3 官方使用的仓库）
DDI_REPO = "doronz88/DeveloperDiskImage"
DDI_BRANCH = "main"
DDI_DIR = "PersonalizedImages/Xcode_iOS_DDI_Personalized"
#: 文件名 → 最小合理体积（用于判断下载是否完整）
DDI_FILES = {
    "Image.dmg": 10 * 1024 * 1024,
    "BuildManifest.plist": 100 * 1024,
    "Image.dmg.trustcache": 256,
}
#: 本地存放位置（相对项目/资源根）
DDI_SUBDIR = ("resources", "ddi", "personalized")


def ddi_candidates(root: Path | str | None = None) -> list[Path]:
    """按优先级返回镜像**可能所在**的目录。

    打包后资源在 ``sys._MEIPASS``（PyInstaller 解包的临时目录，只读、退出即删），
    所以下载得到的镜像不能放那里；用户目录 ``~/.ios-location-ddi`` 是第二候选，
    ``core/server.py:_stage_personalized_ddi`` 的搜索列表里已经包含它。
    """
    if root:
        return [Path(root).joinpath(*DDI_SUBDIR)]
    resources = plat.resolve_roots()[0]
    return [
        resources.joinpath(*DDI_SUBDIR),                      # 随包内置（只读）
        Path.home() / ".ios-location-ddi" / "personalized",   # 运行期下载（可写）
    ]


def ddi_write_dir(root: Path | str | None = None) -> Path:
    """返回**可写**的镜像目录（下载目标）。"""
    if root:
        return Path(root).joinpath(*DDI_SUBDIR)
    if getattr(sys, "frozen", False):
        return Path.home() / ".ios-location-ddi" / "personalized"
    return plat.resolve_roots()[0].joinpath(*DDI_SUBDIR)


def _ddi_status_one(directory: Path) -> dict[str, Any]:
    files: dict[str, Any] = {}
    missing: list[str] = []
    total = 0
    for name, minimum in DDI_FILES.items():
        path = directory / name
        if path.exists():
            size = path.stat().st_size
            ok = size >= minimum
            total += size
            files[name] = {"exists": True, "size": size, "ok": ok}
            if not ok:
                missing.append(name)
        else:
            files[name] = {"exists": False, "size": 0, "ok": False}
            missing.append(name)
    return {
        "ok": not missing,
        "dir": str(directory),
        "files": files,
        "missing": missing,
        "total_mb": round(total / 1048576, 2),
    }


def ddi_status(root: Path | str | None = None) -> dict[str, Any]:
    """检查本地开发者镜像素材是否齐全（按候选目录优先级，返回第一个齐全的）。"""
    candidates = ddi_candidates(root)
    for candidate in candidates:
        info = _ddi_status_one(candidate)
        if info["ok"]:
            return info
    # 都不齐全：报告优先级最高的那个（通常就是随包内置的），便于用户理解缺什么
    info = _ddi_status_one(candidates[0])
    info["searched"] = [str(c) for c in candidates]
    return info


def ddi_dir(root: Path | str | None = None) -> Path:
    """返回镜像实际所在目录：优先齐全的那个，否则返回可写目录。"""
    for candidate in ddi_candidates(root):
        if (candidate / "Image.dmg").exists():
            return candidate
    return ddi_write_dir(root)


def _ddi_urls(name: str) -> list[str]:
    path = f"{DDI_DIR}/{name}"
    return [
        f"https://raw.githubusercontent.com/{DDI_REPO}/{DDI_BRANCH}/{path}",
        f"https://cdn.jsdelivr.net/gh/{DDI_REPO}@{DDI_BRANCH}/{path}",
    ]


def download_developer_image(
    log: Callable[[str, str], None] | None = None,
    root: Path | str | None = None,
    force: bool = False,
) -> dict[str, Any]:
    """下载缺失的个性化开发者镜像（约 15 MB）。

    按「GitHub 直连 → jsDelivr 镜像」的顺序重试，与 tools/fetch_ddi.py 的策略一致。
    供「环境准备」窗口的「安装」按钮调用。
    """
    def emit(level: str, message: str) -> None:
        if log:
            log(level, message)

    directory = ddi_write_dir(root)
    try:
        directory.mkdir(parents=True, exist_ok=True)
    except OSError as exc:
        return {"ok": False, "error": f"无法创建目录 {directory}：{exc}"}

    current = ddi_status(root)
    if current["ok"] and not force:
        return {"ok": True, "action": "none", "status": current,
                "detail": f"已就绪（{current['total_mb']} MB）"}
    emit("info", f"镜像下载目录：{directory}")

    for name, minimum in DDI_FILES.items():
        target = directory / name
        if not force and target.exists() and target.stat().st_size >= minimum:
            emit("info", f"{name} 已存在，跳过")
            continue
        last_error = ""
        downloaded = False
        for url in _ddi_urls(name):
            ok, message = _download(url, target, log, None)
            if ok and target.exists() and target.stat().st_size >= minimum:
                emit("info", f"{name}：{target.stat().st_size/1048576:.2f} MB")
                downloaded = True
                break
            last_error = message
            emit("warn", f"{name} 从 {url.split('/')[2]} 下载失败，尝试备用源…")
        if not downloaded:
            return {"ok": False, "error": f"{name} 下载失败：{last_error}",
                    "status": ddi_status(root)}

    after = ddi_status(root)
    if not after["ok"]:
        return {"ok": False, "error": "镜像文件仍不完整：" + "、".join(after["missing"]),
                "status": after}
    return {"ok": True, "action": "downloaded", "status": after,
            "detail": f"已下载（{after['total_mb']} MB）"}