random-background-api/build.py

#!/usr/bin/env python3
from __future__ import annotations

import json
import os
import re
import shutil
import subprocess
import sys
import venv
from datetime import datetime, timezone
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent
PUBLIC_DIR = BASE_DIR / "public"
MANIFEST_PATH = PUBLIC_DIR / "manifest.json"

VARIANTS = {
    "mobile": {
        "out": PUBLIC_DIR / "mobile-image",
        "scan": [
            PUBLIC_DIR / "mobile-image",
            BASE_DIR / "mobile-image",
        ],
    },
    "desktop": {
        "out": PUBLIC_DIR / "desketop-image",
        "scan": [
            PUBLIC_DIR / "desketop-image",
            BASE_DIR / "desketop-image",
        ],
    },
}

INPUT_EXTENSIONS = {
    ".jpg",
    ".jpeg",
    ".png",
    ".bmp",
    ".gif",
    ".tif",
    ".tiff",
    ".webp",
}

GENERATED_NAME_RE = re.compile(r"^\d+\.webp$", re.IGNORECASE)
VENV_DIR = BASE_DIR / ".build-venv"


def ensure_pillow() -> None:
    try:
        from PIL import Image  # noqa: F401
        return
    except Exception:
        pass

    venv_python = VENV_DIR / "bin" / "python"
    if not venv_python.exists():
        print("Pillow not found, creating a local build venv...", file=sys.stderr)
        builder = venv.EnvBuilder(with_pip=True)
        builder.create(VENV_DIR)

    has_pillow = subprocess.run(
        [str(venv_python), "-c", "import PIL"],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        check=False,
    ).returncode == 0

    if not has_pillow:
        print("Installing Pillow into the local build venv...", file=sys.stderr)
        subprocess.check_call([str(venv_python), "-m", "pip", "install", "--quiet", "Pillow"])
    os.execv(str(venv_python), [str(venv_python), str(Path(__file__).resolve()), *sys.argv[1:]])


def load_pillow():
    ensure_pillow()
    from PIL import Image, ImageOps

    return Image, ImageOps


def iter_source_files(directory: Path) -> list[Path]:
    if not directory.exists():
        return []
    files: list[Path] = []
    for path in sorted(directory.iterdir(), key=lambda item: item.name.lower()):
        if not path.is_file():
            continue
        if GENERATED_NAME_RE.match(path.name):
            continue
        if path.suffix.lower() not in INPUT_EXTENSIONS:
            continue
        files.append(path)
    return files


def select_scan_directory(scan_candidates: list[Path]) -> Path:
    """优先有原图的目录，其次有序号 webp；否则退回 scan 第一项（通常为 public下的输出目录）。"""
    for directory in scan_candidates:
        if iter_source_files(directory):
            return directory
    for directory in scan_candidates:
        if sorted_generated_webp_names(directory):
            return directory
    return scan_candidates[0]


def clear_generated_files(directory: Path) -> None:
    if not directory.exists():
        directory.mkdir(parents=True, exist_ok=True)
        return
    for path in directory.iterdir():
        if path.is_file() and GENERATED_NAME_RE.match(path.name):
            path.unlink()


def sorted_generated_webp_names(directory: Path) -> list[str]:
    """已存在的序号 webp（无原图时再跑脚本不会清空）。"""
    if not directory.exists():
        return []
    names = [
        path.name
        for path in directory.iterdir()
        if path.is_file() and GENERATED_NAME_RE.match(path.name)
    ]

    def sort_key(entry: str) -> int:
        return int(Path(entry).stem)

    names.sort(key=sort_key)
    return names


def convert_to_webp(source: Path, target: Path, image_mod, image_ops) -> None:
    with image_mod.open(source) as image:
        image = image_ops.exif_transpose(image)
        if image.mode not in {"RGB", "RGBA", "LA", "P"}:
            image = image.convert("RGBA")
        elif image.mode == "P":
            image = image.convert("RGBA")
        elif image.mode == "LA":
            image = image.convert("RGBA")
        elif image.mode == "RGB":
            pass
        image.save(target, format="WEBP", quality=92, method=6)


def copy_generated_webps(source_dir: Path, out_dir: Path, names: list[str]) -> None:
    """将已有序号 webp 同步到 public 下的部署目录。"""
    out_dir.mkdir(parents=True, exist_ok=True)
    for entry in names:
        shutil.copy2(source_dir / entry, out_dir / entry)


def build_variant(name: str, spec: dict[str, Path | list[Path]], image_mod, image_ops) -> dict:
    scan_dir = select_scan_directory(list(spec["scan"]))
    out_dir = spec["out"]
    out_dir.mkdir(parents=True, exist_ok=True)
    sources = iter_source_files(scan_dir)

    images: list[str] = []
    if sources:
        clear_generated_files(out_dir)
        for index, source in enumerate(sources, start=1):
            target = out_dir / f"{index}.webp"
            try:
                convert_to_webp(source, target, image_mod, image_ops)
                images.append(target.name)
                rel_src = os.path.relpath(source.resolve(), BASE_DIR.resolve())
                print(f"[{name}] {rel_src} -> {target.name}")
            except Exception as exc:
                print(f"[{name}] skip {source.name}: {exc}", file=sys.stderr)
    else:
        images = sorted_generated_webp_names(out_dir)
        if not images and scan_dir.resolve() != out_dir.resolve():
            legacy = sorted_generated_webp_names(scan_dir)
            if legacy:
                copy_generated_webps(scan_dir, out_dir, legacy)
                images = legacy
                print(f"[{name}] 已从 {scan_dir.relative_to(BASE_DIR)} 复制序号 webp ×{len(images)} -> public/")
        if images:
            if scan_dir.resolve() == out_dir.resolve():
                print(f"[{name}] 未检测到原图，保留 public 下序号 webp ×{len(images)}")
        else:
            print(f"[{name}] 跳过：未发现原图，且 public/ 与归档目录均无 1.webp 等序号文件")

    return {
        "folder": out_dir.name,
        "count": len(images),
        "images": images,
    }


def write_manifest(manifest: dict) -> None:
    MANIFEST_PATH.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")


def main() -> int:
    image_mod, image_ops = load_pillow()

    manifest = {
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "variants": {},
    }

    for name, variant_spec in VARIANTS.items():
        manifest["variants"][name] = build_variant(name, variant_spec, image_mod, image_ops)

    write_manifest(manifest)
    print(f"Wrote manifest: {MANIFEST_PATH}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())