iOSAI/Utils/DevDiskImageDeployer.py

# support_deployer.py
import os
import re
import shutil
from pathlib import Path
from dataclasses import dataclass, field
from typing import Iterable, Optional

VERSION_RE = re.compile(r"^\d+(?:\.\d+)*$")  # 15 / 15.6 / 16.7 / 16.7.1

def _find_support_root(hint: Optional[Path]) -> Optional[Path]:
    """
    1) 优先：显式传入的 hint
    2) 其次：环境变量 SUPPORT_DDI_DIR
    3) 再次：从 __file__ 所在目录向上搜索 3 层，找名为 'SupportFiles' 的目录
    """
    if hint and hint.exists():
        return hint.resolve()

    env = os.environ.get("SUPPORT_DDI_DIR")
    if env:
        p = Path(env).expanduser()
        if p.exists():
            return p.resolve()

    here = Path(__file__).resolve().parent
    for _ in range(4):  # 当前目录 + 向上 3 层
        cand = here / "SupportFiles"
        if cand.exists():
            return cand.resolve()
        here = here.parent
    return None


@dataclass
class DevDiskImageDeployer:
    """
    同步 SupportFiles/<version>/ 或 SupportFiles/<version>.zip 到 ~/.tidevice/device-support
    - 目录：复制为 ~/.tidevice/device-support/<version>/
    - zip：原样复制为 ~/.tidevice/device-support/<version>.zip （不解压）
    - 已存在则跳过；如设置 overwrite=True 则覆盖
    """
    project_support_root: Optional[Path] = None
    cache_root: Optional[Path] = None
    verbose: bool = True
    dry_run: bool = False
    overwrite: bool = False

    _src_dir: Path = field(init=False, repr=False)
    _cache_dir: Path = field(init=False, repr=False)

    def __post_init__(self):
        src = _find_support_root(self.project_support_root)
        if src is None:
            raise FileNotFoundError(
                "未找到 SupportFiles 目录。"
                "可传入 project_support_root，或设置环境变量 SUPPORT_DDI_DIR，"
                "或确保在当前文件上层 3 级目录内存在名为 'SupportFiles' 的目录。"
            )
        self._src_dir = src

        if self.cache_root is None:
            self._cache_dir = Path.home() / ".tidevice" / "device-support"
        else:
            self._cache_dir = Path(self.cache_root).expanduser().resolve()
        self._cache_dir.mkdir(parents=True, exist_ok=True)

        if self.verbose:
            print(f"[INFO] resolved SupportFiles = {self._src_dir}")
            print(f"[INFO] cache_dir            = {self._cache_dir}")

            parent = self._src_dir.parent
            try:
                siblings = ", ".join(sorted(p.name for p in parent.iterdir() if p.is_dir()))
                print(f"[INFO] SupportFiles parent = {parent}")
                print(f"[INFO] siblings            = {siblings}")
            except Exception:
                pass

    def deploy_all(self):
        entries = list(self._iter_version_entries(self._src_dir))
        copied = skipped = 0

        for p, kind, version in entries:
            # kind: "dir" 或 "zip"
            if kind == "dir":
                dst = self._cache_dir / version
                exists = dst.exists()
                if exists and not self.overwrite:
                    skipped += 1
                    # if self.verbose:
                    #     print(f"[SKIP] {dst} 已存在（目录）")
                    continue
                if exists and self.overwrite and not self.dry_run:
                    shutil.rmtree(dst)

                if self.verbose:
                    print(f"[COPY] DIR  {p} -> {dst}")
                if not self.dry_run:
                    shutil.copytree(p, dst)
                copied += 1

            elif kind == "zip":
                dst = self._cache_dir / f"{version}.zip"
                exists = dst.exists()
                if exists and not self.overwrite:
                    skipped += 1
                    # if self.verbose:
                    #     print(f"[SKIP] {dst} 已存在（zip）")
                    continue
                if exists and self.overwrite and not self.dry_run:
                    dst.unlink()

                if self.verbose:
                    print(f"[COPY] ZIP  {p} -> {dst}")
                if not self.dry_run:
                    # 用 copy2 保留 mtime 等元数据
                    shutil.copy2(p, dst)
                copied += 1

        if self.verbose:
            print(f"[SUMMARY] copied={copied}, skipped={skipped}, total={copied+skipped}")

    # -------- helpers --------
    def _iter_version_entries(self, root: Path) -> Iterable[tuple[Path, str, str]]:
        """
        迭代返回 (路径, 类型, 版本号)
        - 目录：名称需匹配版本号
        - zip：stem（去除后缀）的名称需匹配版本号
        """
        for p in sorted(root.iterdir()):
            if p.is_dir() and VERSION_RE.match(p.name):
                yield (p, "dir", p.name)
            elif p.is_file() and p.suffix.lower() == ".zip" and VERSION_RE.match(p.stem):
                yield (p, "zip", p.stem)