roux/scripts/faceswap-images.py

#!/usr/bin/env python3
"""
Face-swap issue plate images with FaceFusion (~/Projekte/facefusion).

  python3 scripts/faceswap-images.py --issue 02 --source ~/Bilder/palina.webp

Runs FaceFusion `batch-run` once (models load a single time) over every
content/posts/<issue>/<slug>/<slug>.png, swapping in the face from --source, and
writes the results back in place (the .png sources are tracked in git, so that is
the backup). Pass --out DIR to write to a staging tree instead.

Idempotent: a slug already recorded in data/prompts/issue-<issue>.faceswap.json
for the same source image is skipped unless --force.

FaceFusion is invoked through its own virtualenv
(~/Projekte/facefusion/.venv/bin/python); this script itself needs only stdlib.
"""
import argparse
import hashlib
import json
import shutil
import subprocess
import sys
import tempfile
import time
from pathlib import Path

SITE = Path(__file__).resolve().parent.parent
FF_DIR = Path.home() / "Projekte" / "facefusion"
FF_PY = FF_DIR / ".venv" / "bin" / "python"
FF_ENTRY = FF_DIR / "facefusion.py"


def sha256(path: Path) -> str:
    h = hashlib.sha256()
    with path.open("rb") as f:
        for chunk in iter(lambda: f.read(1 << 20), b""):
            h.update(chunk)
    return h.hexdigest()


def load_manifest(path: Path) -> dict:
    if path.is_file():
        try:
            return {r["slug"]: r for r in json.loads(path.read_text(encoding="utf-8"))}
        except (json.JSONDecodeError, KeyError, TypeError):
            pass
    return {}


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--issue", required=True)
    ap.add_argument("--source", type=Path, default=Path.home() / "Bilder" / "palina.webp",
                    help="face to swap in (default: ~/Bilder/palina.webp)")
    ap.add_argument("--out", type=Path, help="output root; default is in-place")
    # Everything below is passed to FaceFusion only when explicitly set;
    # otherwise FaceFusion uses its own defaults.
    ap.add_argument("--model",
                    help="face-swapper model (blendswap_256, hyperswap_1a_256, simswap_unofficial_512, …)")
    ap.add_argument("--pixel-boost",
                    choices=["256x256", "512x512", "768x768", "1024x1024"])
    ap.add_argument("--swapper-weight")
    ap.add_argument("--enhance", action="store_true",
                    help="also run the face_enhancer processor (sharper, slower)")
    ap.add_argument("--enhancer-model")
    ap.add_argument("--enhancer-blend", type=int)
    ap.add_argument("--output-image-quality", type=int)
    ap.add_argument("--execution-providers", nargs="+",
                    help="cuda / cpu / tensorrt …")
    ap.add_argument("--execution-thread-count", type=int)
    ap.add_argument("--only", help="comma-separated slugs")
    ap.add_argument("--limit", type=int)
    ap.add_argument("--force", action="store_true")
    ap.add_argument("--dry-run", action="store_true")
    args = ap.parse_args()

    if not FF_PY.exists():
        sys.exit(f"FaceFusion venv python not found: {FF_PY}")
    source = args.source.expanduser()
    if not source.is_file():
        sys.exit(f"source face not found: {source}")

    content = SITE / "content" / "posts" / args.issue
    if not content.is_dir():
        sys.exit(f"no such issue dir: {content}")

    bundles = sorted(p for p in content.iterdir()
                     if p.is_dir() and (p / f"{p.name}.png").is_file())
    if args.only:
        want = {s.strip() for s in args.only.split(",")}
        bundles = [b for b in bundles if b.name in want]
    if args.limit:
        bundles = bundles[: args.limit]
    if not bundles:
        sys.exit("no matching plate images")

    src_hash = sha256(source)
    manifest_path = SITE / "data" / "prompts" / f"issue-{args.issue}.faceswap.json"
    done = load_manifest(manifest_path)

    todo = [b for b in bundles
            if args.force or done.get(b.name, {}).get("source_sha256") != src_hash]
    skipped = len(bundles) - len(todo)
    print(f"{len(bundles)} plates · {len(todo)} to swap · {skipped} already done "
          f"(source {source.name} {src_hash[:12]})")
    if not todo:
        return 0

    processors = ["face_swapper"] + (["face_enhancer"] if args.enhance else [])

    with tempfile.TemporaryDirectory(prefix="roux-faceswap-") as td:
        tin, tout, tjobs = Path(td) / "in", Path(td) / "out", Path(td) / "jobs"
        tin.mkdir()
        tout.mkdir()
        tjobs.mkdir()
        for b in todo:
            shutil.copy2(b / f"{b.name}.png", tin / f"{b.name}.png")

        cmd = [
            str(FF_PY), str(FF_ENTRY), "batch-run",
            "--jobs-path", str(tjobs),
            "-s", str(source),
            "-t", str(tin / "*.png"),
            "-o", str(tout / "{target_name}.png"),
            "--processors", *processors,
            "--log-level", "info",
        ]
        if args.model:
            cmd += ["--face-swapper-model", args.model]
        if args.pixel_boost:
            cmd += ["--face-swapper-pixel-boost", args.pixel_boost]
        if args.swapper_weight:
            cmd += ["--face-swapper-weight", args.swapper_weight]
        if args.enhancer_model:
            cmd += ["--face-enhancer-model", args.enhancer_model]
        if args.enhancer_blend is not None:
            cmd += ["--face-enhancer-blend", str(args.enhancer_blend)]
        if args.output_image_quality is not None:
            cmd += ["--output-image-quality", str(args.output_image_quality)]
        if args.execution_providers:
            cmd += ["--execution-providers", *args.execution_providers]
        if args.execution_thread_count is not None:
            cmd += ["--execution-thread-count", str(args.execution_thread_count)]

        if args.dry_run:
            print(" ".join(cmd))
            return 0

        print(f"→ facefusion batch-run ({len(todo)} steps, processors: {'+'.join(processors)})")
        rc = subprocess.run(cmd, cwd=str(FF_DIR)).returncode
        if rc != 0:
            sys.exit(f"facefusion exited {rc}")

        out_root = args.out.expanduser() if args.out else content
        ok, failed = [], []
        for b in todo:
            produced = tout / f"{b.name}.png"
            if not produced.is_file():
                failed.append(b.name)
                continue
            dest = out_root / b.name / f"{b.name}.png"
            dest.parent.mkdir(parents=True, exist_ok=True)
            shutil.move(str(produced), str(dest))
            ok.append(b.name)
            done[b.name] = {
                "slug": b.name,
                "source": str(source),
                "source_sha256": src_hash,
                "model": args.model or "facefusion-default",
                "pixel_boost": args.pixel_boost or "facefusion-default",
                "enhanced": args.enhance,
                "done_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            }

    # Only the in-place pipeline is tracked; --out is for experimentation.
    if not args.out:
        manifest_path.parent.mkdir(parents=True, exist_ok=True)
        manifest_path.write_text(
            json.dumps([done[k] for k in sorted(done)], indent=2, ensure_ascii=False) + "\n",
            encoding="utf-8")

    print(f"\n{len(ok)} swapped / {len(failed)} failed / {skipped} skipped")
    if failed:
        print("failed (no face detected?):", ", ".join(sorted(failed)))
        return 1
    return 0


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