使用 Python 脚本批量下载 macOS 航拍屏保/壁纸

  1import json
  2import asyncio
  3import sys
  4from argparse import ArgumentParser
  5from pathlib import Path
  6import plistlib
  7
  8import httpx
  9from tqdm import tqdm
 10
 11BASE_PATH = Path("/Library/Application Support/com.apple.idleassetsd/Customer")
 12DEST_PATH = BASE_PATH / "4KSDR240FPS"
 13ENTRIES_FILE = BASE_PATH / "entries.json"
 14LOCALIZABLE_FILE = BASE_PATH / "TVIdleScreenStrings.bundle/zh_CN.lproj/Localizable.nocache.strings"
 15
 16def load_localizable_strings() -> dict:
 17    strings = {}
 18    if LOCALIZABLE_FILE.exists():
 19        with LOCALIZABLE_FILE.open("rb") as f:
 20            plist_data = plistlib.load(f)
 21            strings.update(plist_data)
 22    return strings
 23
 24def get_localized_name(key: str, strings: dict) -> str:
 25    return strings.get(key, key)
 26
 27def download_asset_sync(item: dict, dst: Path):
 28    name = f"{item['categoryName']}: {item['assetName']}"
 29    tqdm.write(f"Downloading: {name}")
 30    try:
 31        with dst.open("wb") as download_file:
 32            with httpx.stream("GET", item["url-4K-SDR-240FPS"], verify=False) as response:
 33                total = int(response.headers.get("Content-Length", 0))
 34                with tqdm(total=total, unit="B", unit_scale=True, unit_divisor=1024, desc=name, position=1, leave=False) as progress:
 35                    num_bytes_downloaded = response.num_bytes_downloaded
 36                    for chunk in response.iter_bytes():
 37                        download_file.write(chunk)
 38                        progress.update(response.num_bytes_downloaded - num_bytes_downloaded)
 39                        num_bytes_downloaded = response.num_bytes_downloaded
 40    except (httpx.RequestError, httpx.HTTPStatusError) as e:
 41        tqdm.write(f"Error downloading {name}: {e}")
 42
 43async def download_asset_async(client, item: dict, dst: Path, position: int):
 44    name = f"{item['categoryName']}: {item['assetName']}"
 45    tqdm.write(f"Downloading: {name}")
 46    try:
 47        async with client.stream("GET", item["url-4K-SDR-240FPS"]) as response:
 48            total = int(response.headers.get("Content-Length", 0))
 49            with tqdm(total=total, unit="B", unit_scale=True, unit_divisor=1024, desc=name, position=position, leave=False) as progress:
 50                with dst.open("wb") as download_file:
 51                    async for chunk in response.aiter_bytes():
 52                        download_file.write(chunk)
 53                        progress.update(len(chunk))
 54    except (httpx.RequestError, httpx.HTTPStatusError) as e:
 55        tqdm.write(f"Error downloading {name}: {e}")
 56
 57async def download_asset_concurrent(items: list, max_concurrent: int = 5):
 58    async with httpx.AsyncClient(verify=False, timeout=30.0) as client:
 59        tasks = []
 60        pending_items = [item for item in items if not (DEST_PATH / f"{item['id']}.mov").exists() and item.get("url-4K-SDR-240FPS")]
 61
 62        # 预留进度条行
 63        for i in range(min(max_concurrent, len(pending_items))):
 64            print(f"\033[K", end="")  # 清除行
 65            print()  # 预留一行
 66        print(f"\033[{min(max_concurrent, len(pending_items))}A", end="", flush=True)  # 移动光标到第一行
 67
 68        for index, item in enumerate(pending_items):
 69            position = (index % max_concurrent) + 1  # 分配行号（1 到 max_concurrent）
 70            tasks.append(download_asset_async(client, item, DEST_PATH / f"{item['id']}.mov", position))
 71            if len(tasks) >= max_concurrent:
 72                await asyncio.gather(*tasks, return_exceptions=True)
 73                tasks = []
 74        if tasks:
 75            await asyncio.gather(*tasks, return_exceptions=True)
 76
 77    # 清除进度条区域
 78    print(f"\033[{min(max_concurrent, len(pending_items))}A", end="", flush=True)
 79    for _ in range(min(max_concurrent, len(pending_items))):
 80        print(f"\033[K", end="")  # 清除行
 81        print()
 82
 83def main():
 84    parser = ArgumentParser(description="Download macOS Aerial screensaver assets")
 85    parser.add_argument("--batch", nargs="?", const=5, type=int, metavar="SIZE", help="Use concurrent downloads, 5 tasks by default")
 86    args = parser.parse_args()
 87
 88    if not ENTRIES_FILE.exists():
 89        print(f"Error: {ENTRIES_FILE} not found")
 90        sys.exit(1)
 91    with ENTRIES_FILE.open() as f:
 92        data = json.load(f)
 93
 94    localizable_strings = load_localizable_strings()
 95
 96    categories = {}
 97    for category in data.get("categories", []):
 98        category_name = get_localized_name(category["localizedNameKey"], localizable_strings)
 99        categories[category["id"]] = category_name
100
101    for asset in data.get("assets", []):
102        category_id = asset.get("categories", [""])[0]
103        asset["categoryName"] = categories.get(category_id, "")
104        asset["assetName"] = get_localized_name(asset["localizedNameKey"], localizable_strings)
105
106    DEST_PATH.mkdir(parents=True, exist_ok=True)
107
108    if args.batch:
109        asyncio.run(download_asset_concurrent(data.get("assets", []), max_concurrent=args.batch))
110    else:
111        for item in tqdm(data.get("assets", []), desc="Processing assets", position=0):
112            dst = DEST_PATH / f"{item['id']}.mov"
113            if not dst.exists() and item.get("url-4K-SDR-240FPS"):
114                download_asset_sync(item, dst)
115
116    print("Done")
117
118if __name__ == "__main__":
119    main()