scripts/build_faceswap.py

"""C1 runner: face-swap a fictional target face onto a source photo.

Swaps the (fictional, non-celebrity) target identity onto the source face(s) via
the optional InsightFace backend, applies the AI-disclosure watermark, and writes
the result + a QA summary under runtime/ (gitignored). When the backend / swapper
model is unavailable it writes a clearly-marked STUB instead (no network call, no
weights). The swapper model is NEVER committed and is acquired by the operator
(set FACE_SWAP_MODEL_PATH). Pilot Ready: NOT CONFIRMED.

Usage (operator, with a real swap):
    set FACE_SWAP_MODEL_PATH to a local inswapper_128.onnx
    python backend/scripts/build_faceswap.py --source <consented.jpg> \
        --target <fictional_model_face.png> --tag faceswap-01
"""
from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from app.services.face_pipeline.face_swap import (  # noqa: E402
    face_swap_backend_available,
    face_swap_model_path,
    run_face_swap_pipeline,
)
from app.services.gemini_client import normalize_image_orientation_bytes  # noqa: E402
from app.services.watermark import apply_ai_watermark  # noqa: E402


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="C1 face-swap pipeline runner")
    parser.add_argument("--source", required=True, help="consented/synthetic source photo")
    parser.add_argument("--target", required=True, help="fictional, non-celebrity target face")
    parser.add_argument("--evidence-root", default="runtime/gemini-smoke-evidence")
    parser.add_argument("--tag", default="faceswap-01")
    args = parser.parse_args(argv)

    source_path = Path(args.source)
    target_path = Path(args.target)
    for label, p in (("source", source_path), ("target", target_path)):
        if not p.exists():
            print(f"REFUSED: {label} not found: {p}")
            return 2

    out_dir = Path(args.evidence_root) / "gemini-smoke" / "faceswap" / args.tag
    out_dir.mkdir(parents=True, exist_ok=True)

    source_bytes = normalize_image_orientation_bytes(source_path.read_bytes())
    target_bytes = normalize_image_orientation_bytes(target_path.read_bytes())

    backend_ready = face_swap_backend_available()
    if not backend_ready:
        print(f"NOTE: face-swap backend unavailable (model={face_swap_model_path()}); "
              "writing a clearly-marked STUB, not a real swap.")

    result = run_face_swap_pipeline(source_bytes, target_bytes)

    # Raw swap (pre-watermark) for analysis; final deliverable carries the badge.
    (out_dir / "swap-raw.png").write_bytes(result.image_bytes)
    final_bytes = apply_ai_watermark(result.image_bytes)
    (out_dir / "faceswap-output.png").write_bytes(final_bytes)

    metrics = dict(result.metrics)
    metrics["tag"] = args.tag
    metrics["watermark_applied"] = True
    (out_dir / "faceswap-summary.json").write_text(
        json.dumps(metrics, indent=2, ensure_ascii=False), encoding="utf-8"
    )
    md = [f"# C1 face-swap - {args.tag}", ""]
    md += [f"- {k}: {v}" for k, v in metrics.items()]
    md += ["", "> Target face MUST be fictional / non-celebrity.",
           "> Human visual QA required. NOT_PRODUCTION_READY. Pilot Ready: NOT CONFIRMED."]
    (out_dir / "faceswap-summary.md").write_text("\n".join(md), encoding="utf-8")

    print(f"faceswap: DONE tag={args.tag} backend={result.backend} "
          f"face_swapped={metrics['face_swapped']}")
    print(f"outside_face_delta_ratio={metrics['outside_face_delta_ratio']}")
    print(f"out={out_dir}")
    print("Pilot Ready: NOT CONFIRMED.")
    return 0


if __name__ == "__main__":
    sys.exit(main())