136 lines
4.9 KiB
Python
136 lines
4.9 KiB
Python
# 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)
|