Upload 8 files

README.md

@@ -1,57 +1,35 @@
----
-title: FaceSwap GPU
-emoji: 🎬
-colorFrom: indigo
-colorTo: purple
-sdk: gradio
-sdk_version: 5.29.0
-app_file: app.py
-pinned: false
-license: unknown
-short_description: GPU-accelerated all-in-one face swapping (photo + video)
----
-
-# FaceSwap GPU – Rented-GPU Ready
-
-All-in-one face swapping suite optimised for **CUDA GPUs** (RunPod, Vast.ai, Lambda, local RTX, HF GPU Spaces, etc.).
-
-## What changed vs the original
-
-- Explicit `CUDAExecutionProvider` preference via onnxruntime
-- Robust Gradio video path handling (fixes all video tabs)
-- Cleaned frame extraction + resume logic
-- Better error reporting in the Log box
-- Workdir isolation under `workdir/`
-- `opencv-python-headless` (safer on servers)
-- Proper `demo.queue()` for concurrent jobs
-- `ssr_mode=False` to avoid Gradio SSR 405 errors
-- ffmpeg system package declared
-
-## Requirements on the rented machine
-
-- NVIDIA GPU + drivers
-- CUDA-compatible onnxruntime-gpu (already in requirements)
-- ~4–8 GB VRAM recommended for 1080p video
-- ffmpeg installed (packages.txt)
-
-## Local / rented GPU launch
+# FaceSwapAll — Optimized
+
+Optimized InsightFace/ONNX Runtime face-swap Space for NVIDIA CUDA GPUs, with CPU fallback.
+
+## Main optimizations
+- Persistent InsightFace models
+- Source-face detection/embedding cached per job
+- In-memory video frame processing
+- Reduced filesystem I/O
+- Configurable detector size and detection interval
+- CUDAExecutionProvider when available
+- FFmpeg/OpenCV video output
+- Original audio preservation when FFmpeg is available
+
+## Hugging Face Space
+The original project contains `inswapper_128.onnx` as a 554 MB Xet/LFS file. This ZIP intentionally does not duplicate that large binary. After uploading these files to the Space, run:
 
 ```bash
-pip install -r requirements.txt
-python app.py --share          # creates a public link
-# or
-python app.py --server-port 7860
+python download_model.py
 ```
 
-First run downloads `buffalo_l` + `inswapper_128.onnx` (~300 MB).
+or keep the existing `inswapper_128.onnx` in the repository root.
 
-## Tips for long videos
+The original repository and model are at:
+`dosesnrolls1/FaceSwapAll`
 
-- Keep “Delete extracted frames” checked to save disk.
-- Processing is frame-by-frame; a 1-minute 30 fps clip ≈ 1800 swaps.
-- On a modern GPU expect 5–15 fps effective (depends on resolution & number of faces).
-- Resume is automatic: if the process is killed, re-run and already-swapped frames are skipped.
+## Git LFS
+If you want the model tracked directly in Git, install Git LFS and add:
 
-## Face indices
+```bash
+git lfs install
+git add inswapper_128.onnx
+```
 
-Faces are sorted left-to-right. Index 1 = leftmost face.
+The included `.gitattributes` already marks ONNX files for LFS.

VideoSwapping.py

@@ -1,81 +1,12 @@
-import cv2
-import os
 
+"""Backward-compatible wrapper around the optimized video engine."""
+from optimized_swapper import FaceSwapEngine
 
-def extract_frames(video_path, frames_dir, max_frames=None):
-    """
-    Extract frames with resume support.
-    Returns ordered list of frame paths.
-    """
-    os.makedirs(frames_dir, exist_ok=True)
+_ENGINE = FaceSwapEngine()
 
-    existing = sorted(
-        [f for f in os.listdir(frames_dir)
-         if f.startswith("frame_") and f.endswith(".jpg")]
+def swap_video(source_img, video_path, source_face_idx=1, target_face_idx=1,
+               det_size=320, detection_interval=1):
+    return _ENGINE.swap_video(
+        source_img, source_face_idx, video_path, target_face_idx,
+        det_size=det_size, detection_interval=detection_interval
     )
-    last_idx = -1
-    if existing:
-        try:
-            last_idx = max(int(f.split("_")[1].split(".")[0]) for f in existing)
-        except Exception:
-            last_idx = -1
-
-    cap = cv2.VideoCapture(video_path)
-    if not cap.isOpened():
-        raise RuntimeError(f"Cannot open video: {video_path}")
-
-    frame_paths = []
-    idx = 0
-    while True:
-        if max_frames is not None and idx >= max_frames:
-            break
-        ret, frame = cap.read()
-        if not ret:
-            break
-        frame_path = os.path.join(frames_dir, f"frame_{idx:05d}.jpg")
-        if idx > last_idx:
-            cv2.imwrite(frame_path, frame)
-        frame_paths.append(frame_path)
-        idx += 1
-    cap.release()
-
-    if not frame_paths:
-        raise RuntimeError("No frames could be extracted from the video")
-    return frame_paths
-
-
-def frames_to_video(frames_dir, output_video_path, fps, fourcc="mp4v"):
-    """
-    Rebuild video from sorted frame_*.jpg or swapped_*.jpg files.
-    """
-    frames = sorted(
-        [
-            os.path.join(frames_dir, f)
-            for f in os.listdir(frames_dir)
-            if f.endswith(".jpg") and (f.startswith("frame_") or f.startswith("swapped_"))
-        ]
-    )
-    if not frames:
-        raise RuntimeError(f"No frames found in {frames_dir}")
-
-    first = cv2.imread(frames[0])
-    if first is None:
-        raise RuntimeError(f"Cannot read first frame: {frames[0]}")
-    height, width = first.shape[:2]
-
-    writer = cv2.VideoWriter(
-        output_video_path,
-        cv2.VideoWriter_fourcc(*fourcc),
-        fps,
-        (width, height)
-    )
-    written = 0
-    for p in frames:
-        img = cv2.imread(p)
-        if img is not None:
-            writer.write(img)
-            written += 1
-    writer.release()
-
-    if written == 0 or not os.path.exists(output_video_path) or os.path.getsize(output_video_path) < 1000:
-        raise RuntimeError(f"Failed to write video: {output_video_path}")

app.py

@@ -1,843 +1,151 @@
-import gradio as gr
+
 import os
-import cv2
-import numpy as np
-import shutil
-import subprocess
 import time
-from SinglePhoto import FaceSwapper
-import argparse
-
-# ------------------------------------------------------------------
-# Global swapper (GPU if available)
-# ------------------------------------------------------------------
-print("Initializing FaceSwapper (this may take a moment on first run)...")
-swapper = FaceSwapper(det_size=(640, 640), ctx_id=0)
-print("FaceSwapper ready.")
+import tempfile
+import shutil
+from pathlib import Path
 
+import cv2
+import numpy as np
+import gradio as gr
 
-# ------------------------------------------------------------------
-# Helpers
-# ------------------------------------------------------------------
-def add_audio_to_video(original_video_path, video_no_audio_path, output_path):
-    """Mux original audio onto the silent swapped video using ffmpeg."""
-    cmd = [
-        "ffmpeg", "-y",
-        "-i", video_no_audio_path,
-        "-i", original_video_path,
-        "-c:v", "copy",
-        "-c:a", "aac",
-        "-map", "0:v:0",
-        "-map", "1:a:0?",
-        "-shortest",
-        output_path
-    ]
-    try:
-        subprocess.run(cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-        return True, ""
-    except subprocess.CalledProcessError as e:
-        return False, e.stderr.decode(errors="ignore")
+from optimized_swapper import FaceSwapEngine
 
+ENGINE = FaceSwapEngine()
 
-def safe_rmtree(path):
+def _as_index(v, default=1):
     try:
-        if os.path.exists(path):
-            shutil.rmtree(path)
+        return max(1, int(v))
     except Exception:
-        pass
-
-
-def ensure_dirs(*paths):
-    for p in paths:
-        os.makedirs(p, exist_ok=True)
-
-
-def _resolve_video_path(video):
-    """Turn whatever Gradio gives us into a real file path."""
-    if video is None:
-        raise ValueError("No video provided")
-    if isinstance(video, str) and os.path.exists(video):
-        return video
-    if hasattr(video, "name") and os.path.exists(getattr(video, "name", "")):
-        return video.name
-    if isinstance(video, dict) and "name" in video and os.path.exists(video["name"]):
-        return video["name"]
-    if isinstance(video, (list, tuple)) and len(video) > 0:
-        return _resolve_video_path(video[0])
-    raise ValueError(f"Could not resolve video path from type {type(video)}: {video}")
-
-
-# ------------------------------------------------------------------
-# Photo functions
-# ------------------------------------------------------------------
-def swap_single_photo(src_img, src_idx, dst_img, dst_idx, progress=gr.Progress(track_tqdm=True)):
-    log = ""
-    start = time.time()
-    try:
-        progress(0, desc="Preparing")
-        src_path = "workdir/SinglePhoto/data_src.jpg"
-        dst_path = "workdir/SinglePhoto/data_dst.jpg"
-        out_path = "workdir/SinglePhoto/output_swapped.jpg"
-        ensure_dirs(os.path.dirname(src_path), os.path.dirname(out_path))
-
-        cv2.imwrite(src_path, cv2.cvtColor(src_img, cv2.COLOR_RGB2BGR))
-        cv2.imwrite(dst_path, cv2.cvtColor(dst_img, cv2.COLOR_RGB2BGR))
-        log += "Saved source & destination\n"
-
-        progress(0.4, desc="Swapping on GPU")
-        result = swapper.swap_faces(src_path, int(src_idx), dst_path, int(dst_idx))
-        cv2.imwrite(out_path, result)
-        log += f"Saved result → {out_path}\n"
-
-        for p in (src_path, dst_path):
-            if os.path.exists(p):
-                os.remove(p)
-
-        progress(1, desc="Done")
-        log += f"Elapsed: {time.time()-start:.2f}s\n"
-        return out_path, log
-    except Exception as e:
-        log += f"ERROR: {e}\n"
-        return None, log
-
-
-def swap_single_src_multi_dst(src_img, dst_imgs, dst_indices, progress=gr.Progress(track_tqdm=True)):
-    log = ""
-    results = []
-    base = "workdir/SingleSrcMultiDst"
-    src_dir = f"{base}/src"
-    dst_dir = f"{base}/dst"
-    out_dir = f"{base}/output"
-    ensure_dirs(src_dir, dst_dir, out_dir)
-
-    try:
-        if isinstance(src_img, tuple):
-            src_img = src_img[0]
-        src_path = os.path.join(src_dir, "data_src.jpg")
-        cv2.imwrite(src_path, cv2.cvtColor(src_img, cv2.COLOR_RGB2BGR))
-        log += "Saved source image\n"
-
-        if isinstance(dst_indices, str):
-            idx_list = [int(x.strip()) for x in dst_indices.split(",") if x.strip().isdigit()]
-        else:
-            idx_list = [int(x) for x in dst_indices]
-
-        total = max(1, len(dst_imgs))
-        for j, dst_img in enumerate(dst_imgs):
-            if isinstance(dst_img, tuple):
-                dst_img = dst_img[0]
-            if dst_img is None:
-                results.append(None)
-                continue
-            dst_path = os.path.join(dst_dir, f"data_dst_{j}.jpg")
-            out_path = os.path.join(out_dir, f"output_swapped_{j}.jpg")
-            cv2.imwrite(dst_path, cv2.cvtColor(dst_img, cv2.COLOR_RGB2BGR))
-            try:
-                dst_idx = idx_list[j] if j < len(idx_list) else 1
-                result = swapper.swap_faces(src_path, 1, dst_path, dst_idx)
-                cv2.imwrite(out_path, result)
-                results.append(out_path)
-                log += f"OK: src → dst[{j}] (face {dst_idx})\n"
-            except Exception as e:
-                results.append(None)
-                log += f"FAIL dst[{j}]: {e}\n"
-            progress((j + 1) / total, desc=f"Dst {j+1}/{total}")
-        return results, log
-    except Exception as e:
-        log += f"FATAL: {e}\n"
-        return results, log
-
+        return default
+
+def swap_photo(source, source_idx, target, target_idx, det_size):
+    if source is None or target is None:
+        raise gr.Error("Upload both source and target images.")
+    result = ENGINE.swap_image(
+        source, _as_index(source_idx), target, _as_index(target_idx),
+        det_size=int(det_size)
+    )
+    return result
+
+def swap_video(source, source_idx, video_path, target_idx, det_size, detection_interval, jpeg_quality, audio):
+    if source is None:
+        raise gr.Error("Upload a source image.")
+    if not video_path:
+        raise gr.Error("Upload a target video.")
+
+    out = ENGINE.swap_video(
+        source=source,
+        source_idx=_as_index(source_idx),
+        video_path=video_path,
+        target_idx=_as_index(target_idx),
+        det_size=int(det_size),
+        detection_interval=max(1, int(detection_interval)),
+        jpeg_quality=int(jpeg_quality),
+        preserve_audio=bool(audio),
+    )
+    return out
 
-def swap_multi_src_single_dst(src_imgs, dst_img, dst_idx, progress=gr.Progress(track_tqdm=True)):
-    log = ""
+def swap_multi_source_single(source_files, target, target_idx, det_size):
+    if not source_files or target is None:
+        raise gr.Error("Upload source images and a target image.")
     results = []
-    base = "workdir/MultiSrcSingleDst"
-    src_dir = f"{base}/src"
-    dst_dir = f"{base}/dst"
-    out_dir = f"{base}/output"
-    ensure_dirs(src_dir, dst_dir, out_dir)
-
-    try:
-        if isinstance(dst_img, tuple):
-            dst_img = dst_img[0]
-        dst_path = os.path.join(dst_dir, "data_dst.jpg")
-        cv2.imwrite(dst_path, cv2.cvtColor(dst_img, cv2.COLOR_RGB2BGR))
-        log += "Saved destination\n"
-
-        total = max(1, len(src_imgs))
-        for i, src_img in enumerate(src_imgs):
-            if isinstance(src_img, tuple):
-                src_img = src_img[0]
-            if src_img is None:
-                results.append(None)
-                continue
-            src_path = os.path.join(src_dir, f"data_src_{i}.jpg")
-            out_path = os.path.join(out_dir, f"output_swapped_{i}.jpg")
-            cv2.imwrite(src_path, cv2.cvtColor(src_img, cv2.COLOR_RGB2BGR))
-            try:
-                result = swapper.swap_faces(src_path, 1, dst_path, int(dst_idx))
-                cv2.imwrite(out_path, result)
-                results.append(out_path)
-                log += f"OK: src[{i}] → dst\n"
-            except Exception as e:
-                results.append(None)
-                log += f"FAIL src[{i}]: {e}\n"
-            progress((i + 1) / total, desc=f"Src {i+1}/{total}")
-        return results, log
-    except Exception as e:
-        log += f"FATAL: {e}\n"
-        return results, log
-
-
-def swap_multi_src_multi_dst(src_imgs, dst_imgs, dst_indices, progress=gr.Progress(track_tqdm=True)):
-    log = ""
+    for item in source_files:
+        path = item if isinstance(item, str) else item[0]
+        img = cv2.imread(path)
+        if img is None:
+            continue
+        results.append(ENGINE.swap_image(
+            img, 1, target, _as_index(target_idx), int(det_size)
+        ))
+    return results
+
+def swap_multi_source_multi(source_files, target_files, target_indices, det_size):
+    if not source_files or not target_files:
+        raise gr.Error("Upload source and target images.")
+    indices = [x.strip() for x in str(target_indices).split(",") if x.strip()]
     results = []
-    base = "workdir/MultiSrcMultiDst"
-    src_dir = f"{base}/src"
-    dst_dir = f"{base}/dst"
-    out_dir = f"{base}/output"
-    ensure_dirs(src_dir, dst_dir, out_dir)
-
-    try:
-        if isinstance(dst_indices, str):
-            idx_list = [int(x.strip()) for x in dst_indices.split(",") if x.strip().isdigit()]
-        else:
-            idx_list = [int(x) for x in dst_indices]
-
-        total = max(1, len(src_imgs) * len(dst_imgs))
-        count = 0
-        for i, src_img in enumerate(src_imgs):
-            if isinstance(src_img, tuple):
-                src_img = src_img[0]
-            if src_img is None:
-                continue
-            src_path = os.path.join(src_dir, f"data_src_{i}.jpg")
-            cv2.imwrite(src_path, cv2.cvtColor(src_img, cv2.COLOR_RGB2BGR))
-
-            for j, dst_img in enumerate(dst_imgs):
-                if isinstance(dst_img, tuple):
-                    dst_img = dst_img[0]
-                if dst_img is None:
-                    continue
-                dst_path = os.path.join(dst_dir, f"data_dst_{j}.jpg")
-                out_path = os.path.join(out_dir, f"output_swapped_{i}_{j}.jpg")
-                cv2.imwrite(dst_path, cv2.cvtColor(dst_img, cv2.COLOR_RGB2BGR))
-                try:
-                    dst_idx = idx_list[j] if j < len(idx_list) else 1
-                    result = swapper.swap_faces(src_path, 1, dst_path, dst_idx)
-                    cv2.imwrite(out_path, result)
-                    results.append(out_path)
-                    log += f"OK src[{i}]→dst[{j}]\n"
-                except Exception as e:
-                    results.append(None)
-                    log += f"FAIL {i}/{j}: {e}\n"
-                count += 1
-                progress(count / total, desc=f"{count}/{total}")
-        return results, log
-    except Exception as e:
-        log += f"FATAL: {e}\n"
-        return results, log
-
-
-def swap_faces_custom(src_imgs, dst_img, mapping_str, progress=gr.Progress(track_tqdm=True)):
-    log = ""
-    start = time.time()
-    base = "workdir/CustomSwap"
-    src_dir = f"{base}/src"
-    temp_dir = f"{base}/temp"
-    dst_path = f"{base}/data_dst.jpg"
-    out_path = f"{base}/output_swapped.jpg"
-    ensure_dirs(src_dir, temp_dir)
-
-    try:
-        cv2.imwrite(dst_path, cv2.cvtColor(dst_img, cv2.COLOR_RGB2BGR))
-        log += "Saved destination\n"
-
-        src_paths = []
-        for i, src_img in enumerate(src_imgs):
-            if isinstance(src_img, tuple):
-                src_img = src_img[0]
-            if src_img is None:
-                continue
-            p = os.path.join(src_dir, f"data_src_{i+1}.jpg")
-            cv2.imwrite(p, cv2.cvtColor(src_img, cv2.COLOR_RGB2BGR))
-            src_paths.append(p)
-            log += f"Saved source {i+1}\n"
-
-        try:
-            mapping = [int(x.strip()) for x in mapping_str.split(",") if x.strip().isdigit()]
-        except Exception as e:
-            return None, f"Bad mapping: {e}"
-
-        temp_dst = os.path.join(temp_dir, "temp_dst.jpg")
-        shutil.copy(dst_path, temp_dst)
-
-        for face_idx, src_idx in enumerate(mapping, start=1):
-            if src_idx < 1 or src_idx > len(src_paths):
-                log += f"Skip invalid src index {src_idx} for face {face_idx}\n"
-                continue
+    targets = []
+    for item in target_files:
+        path = item if isinstance(item, str) else item[0]
+        img = cv2.imread(path)
+        if img is not None:
+            targets.append(img)
+
+    for src_item in source_files:
+        src_path = src_item if isinstance(src_item, str) else src_item[0]
+        src = cv2.imread(src_path)
+        if src is None:
+            continue
+        # Cache the source face once for this source image.
+        ENGINE.prepare_source(src, 1, int(det_size))
+        for j, dst in enumerate(targets):
+            idx = _as_index(indices[j] if j < len(indices) else 1)
             try:
-                swapped = swapper.swap_faces(src_paths[src_idx-1], 1, temp_dst, face_idx)
-                cv2.imwrite(temp_dst, swapped)
-                log += f"Swapped face {face_idx} ← source {src_idx}\n"
+                results.append(ENGINE.swap_prepared_source(dst, idx, int(det_size)))
             except Exception as e:
-                log += f"Failed face {face_idx}: {e}\n"
-
-        shutil.copy(temp_dst, out_path)
-        safe_rmtree(temp_dir)
-        log += f"Elapsed: {time.time()-start:.2f}s\n"
-        return out_path, log
-    except Exception as e:
-        log += f"FATAL: {e}\n"
-        return None, log
-
-
-# ------------------------------------------------------------------
-# Video functions – robust Gradio path handling
-# ------------------------------------------------------------------
-def swap_video(src_img, src_idx, video, dst_idx,
-               delete_frames_dir=True, add_audio=True,
-               progress=gr.Progress()):
-    log = ""
-    start = time.time()
-    base = "workdir/VideoSwapping"
-    src_path = f"{base}/data_src.jpg"
-    dst_video = f"{base}/data_dst.mp4"
-    frames_dir = f"{base}/video_frames"
-    swapped_dir = f"{base}/swapped_frames"
-    tmp_video = f"{base}/tmp_no_audio.mp4"
-    final_video = f"{base}/output_with_audio.mp4"
-
-    ensure_dirs(base, frames_dir, swapped_dir)
-
-    try:
-        cv2.imwrite(src_path, cv2.cvtColor(src_img, cv2.COLOR_RGB2BGR))
-        log += "Saved source image\n"
-        progress(0.05, desc="Source ready")
-
-        video_path = _resolve_video_path(video)
-        shutil.copy(video_path, dst_video)
-        log += f"Copied video → {dst_video}\n"
-
-        from VideoSwapping import extract_frames, frames_to_video
-
-        frame_paths = extract_frames(dst_video, frames_dir)
-        total = len(frame_paths)
-        log += f"Extracted {total} frames\n"
-        progress(0.12, desc=f"Extracted {total} frames")
-
-        already = set(os.listdir(swapped_dir))
-        loop_start = time.time()
-
-        for idx, frame_path in enumerate(frame_paths):
-            name = f"swapped_{idx:05d}.jpg"
-            out_p = os.path.join(swapped_dir, name)
-            if name in already and os.path.exists(out_p):
-                pass
-            else:
-                try:
-                    swapped = swapper.swap_faces(
-                        src_path, int(src_idx), frame_path, int(dst_idx)
-                    )
-                    cv2.imwrite(out_p, swapped)
-                except ValueError as ve:
-                    if "Target image contains" in str(ve) and int(dst_idx) != 1:
-                        swapped = swapper.swap_faces(
-                            src_path, int(src_idx), frame_path, 1
-                        )
-                        cv2.imwrite(out_p, swapped)
-                        log += f"Frame {idx}: fell back to face 1\n"
-                    else:
-                        shutil.copy(frame_path, out_p)
-                        log += f"Frame {idx} failed: {ve}\n"
-                except Exception as e:
-                    shutil.copy(frame_path, out_p)
-                    log += f"Frame {idx} failed, kept original: {e}\n"
-
-            elapsed = time.time() - loop_start
-            avg = elapsed / (idx + 1)
-            remain = avg * (total - idx - 1)
-            m, s = divmod(int(remain), 60)
-            progress(
-                0.12 + 0.7 * (idx + 1) / total,
-                desc=f"GPU swap {idx+1}/{total} | ETA {m:02d}:{s:02d}"
-            )
-
-        cap = cv2.VideoCapture(dst_video)
-        fps = cap.get(cv2.CAP_PROP_FPS) or 25.0
-        cap.release()
-        frames_to_video(swapped_dir, tmp_video, fps)
-        log += f"Rebuilt silent video @ {fps:.2f} fps\n"
-        progress(0.88, desc="Muxing audio")
-
-        if add_audio:
-            ok, err = add_audio_to_video(dst_video, tmp_video, final_video)
-            if ok:
-                log += "Audio muxed successfully\n"
-                result_path = final_video
-            else:
-                log += f"Audio mux failed: {err}\n"
-                result_path = tmp_video
-        else:
-            result_path = tmp_video
-            log += "Audio skipped by user\n"
-
-        if os.path.exists(src_path):
-            os.remove(src_path)
-        if delete_frames_dir:
-            safe_rmtree(frames_dir)
-            log += "Deleted frames dir\n"
-        safe_rmtree(swapped_dir)
-
-        log += f"Total time: {time.time()-start:.1f}s\n"
-        progress(1, desc="Done")
-        return result_path, log
-
-    except Exception as e:
-        log += f"FATAL: {e}\n"
-        return None, log
-
-
-def swap_video_all_faces(src_img, video, num_faces_to_swap,
-                         delete_frames_dir=True, add_audio=True,
-                         progress=gr.Progress()):
-    """Swap the same source face onto the first N faces of every frame."""
-    log = ""
-    start = time.time()
-    base = "workdir/VideoAllFaces"
-    src_path = f"{base}/data_src.jpg"
-    dst_video = f"{base}/data_dst.mp4"
-    frames_dir = f"{base}/video_frames"
-    swapped_dir = f"{base}/swapped_frames"
-    temp_dir = f"{base}/temp"
-    tmp_video = f"{base}/tmp_no_audio.mp4"
-    final_video = f"{base}/output_with_audio.mp4"
-
-    ensure_dirs(base, frames_dir, swapped_dir, temp_dir)
-
-    try:
-        cv2.imwrite(src_path, cv2.cvtColor(src_img, cv2.COLOR_RGB2BGR))
-        log += "Saved source\n"
-
-        video_path = _resolve_video_path(video)
-        shutil.copy(video_path, dst_video)
-        log += f"Copied video → {dst_video}\n"
-
-        from VideoSwapping import extract_frames, frames_to_video
-        frame_paths = extract_frames(dst_video, frames_dir)
-        total = len(frame_paths)
-        log += f"Extracted {total} frames\n"
-        progress(0.1, desc=f"{total} frames")
-
-        already = set(os.listdir(swapped_dir))
-        loop_start = time.time()
-        n_faces = max(1, int(num_faces_to_swap))
-
-        for idx, frame_path in enumerate(frame_paths):
-            name = f"swapped_{idx:05d}.jpg"
-            out_p = os.path.join(swapped_dir, name)
-            if name in already and os.path.exists(out_p):
-                pass
-            else:
-                temp_frame = os.path.join(temp_dir, "t.jpg")
-                shutil.copy(frame_path, temp_frame)
-                for face_i in range(1, n_faces + 1):
-                    try:
-                        swapped = swapper.swap_faces(src_path, 1, temp_frame, face_i)
-                        cv2.imwrite(temp_frame, swapped)
-                    except Exception as e:
-                        log += f"Frame {idx} face {face_i}: {e}\n"
-                        break
-                shutil.copy(temp_frame, out_p)
-                if os.path.exists(temp_frame):
-                    os.remove(temp_frame)
-
-            elapsed = time.time() - loop_start
-            avg = elapsed / (idx + 1)
-            remain = avg * (total - idx - 1)
-            m, s = divmod(int(remain), 60)
-            progress(0.1 + 0.75 * (idx + 1) / total,
-                     desc=f"{idx+1}/{total} | ETA {m:02d}:{s:02d}")
-
-        safe_rmtree(temp_dir)
-
-        cap = cv2.VideoCapture(dst_video)
-        fps = cap.get(cv2.CAP_PROP_FPS) or 25.0
-        cap.release()
-        frames_to_video(swapped_dir, tmp_video, fps)
-
-        if add_audio:
-            ok, err = add_audio_to_video(dst_video, tmp_video, final_video)
-            result_path = final_video if ok else tmp_video
-            if not ok:
-                log += f"Audio fail: {err}\n"
-        else:
-            result_path = tmp_video
-
-        if delete_frames_dir:
-            safe_rmtree(frames_dir)
-        safe_rmtree(swapped_dir)
-
-        log += f"Total: {time.time()-start:.1f}s\n"
-        progress(1, desc="Done")
-        return result_path, log
-
-    except Exception as e:
-        log += f"FATAL: {e}\n"
-        return None, log
-
-
-def swap_video_custom_mapping(src_imgs, video, mapping_str,
-                              delete_frames_dir=True, add_audio=True,
-                              progress=gr.Progress()):
-    log = ""
-    start = time.time()
-    base = "workdir/CustomVideo"
-    src_dir = f"{base}/src"
-    frames_dir = f"{base}/frames"
-    swapped_dir = f"{base}/swapped"
-    temp_dir = f"{base}/temp"
-    dst_video = f"{base}/data_dst.mp4"
-    tmp_video = f"{base}/tmp_no_audio.mp4"
-    final_video = f"{base}/output_with_audio.mp4"
-
-    ensure_dirs(src_dir, frames_dir, swapped_dir, temp_dir)
-
-    try:
-        # save sources
-        src_paths = []
-        for i, img in enumerate(src_imgs):
-            if isinstance(img, tuple):
-                img = img[0]
-            if img is None:
-                continue
-            p = os.path.join(src_dir, f"src_{i+1}.jpg")
-            cv2.imwrite(p, cv2.cvtColor(img, cv2.COLOR_RGB2BGR))
-            src_paths.append(p)
-        log += f"Saved {len(src_paths)} source faces\n"
-
-        try:
-            mapping = [int(x.strip()) for x in mapping_str.split(",") if x.strip().isdigit()]
-        except Exception as e:
-            return None, f"Bad mapping: {e}"
-
-        video_path = _resolve_video_path(video)
-        shutil.copy(video_path, dst_video)
-        log += f"Copied video → {dst_video}\n"
-
-        from VideoSwapping import extract_frames, frames_to_video
-        frame_paths = extract_frames(dst_video, frames_dir)
-        total = len(frame_paths)
-        log += f"Extracted {total} frames\n"
-        progress(0.08, desc=f"{total} frames")
-
-        already = set(os.listdir(swapped_dir))
-        loop_start = time.time()
-        temp_frame = os.path.join(temp_dir, "t.jpg")
-
-        for idx, frame_path in enumerate(frame_paths):
-            name = f"swapped_{idx:05d}.jpg"
-            out_p = os.path.join(swapped_dir, name)
-            if name in already and os.path.exists(out_p):
-                pass
-            else:
-                shutil.copy(frame_path, temp_frame)
-                for face_idx, src_idx in enumerate(mapping, start=1):
-                    if src_idx < 1 or src_idx > len(src_paths):
-                        continue
-                    try:
-                        swapped = swapper.swap_faces(
-                            src_paths[src_idx-1], 1, temp_frame, face_idx
-                        )
-                        cv2.imwrite(temp_frame, swapped)
-                    except Exception as e:
-                        log += f"F{idx} face{face_idx}: {e}\n"
-                shutil.copy(temp_frame, out_p)
-
-            elapsed = time.time() - loop_start
-            avg = elapsed / (idx + 1)
-            remain = avg * (total - idx - 1)
-            m, s = divmod(int(remain), 60)
-            progress(0.08 + 0.75 * (idx + 1) / total,
-                     desc=f"{idx+1}/{total} | ETA {m:02d}:{s:02d}")
-
-        safe_rmtree(temp_dir)
-
-        cap = cv2.VideoCapture(dst_video)
-        fps = cap.get(cv2.CAP_PROP_FPS) or 25.0
-        cap.release()
-        frames_to_video(swapped_dir, tmp_video, fps)
-
-        if add_audio:
-            ok, err = add_audio_to_video(dst_video, tmp_video, final_video)
-            result_path = final_video if ok else tmp_video
-            if not ok:
-                log += f"Audio fail: {err}\n"
-        else:
-            result_path = tmp_video
-
-        if delete_frames_dir:
-            safe_rmtree(frames_dir)
-        safe_rmtree(swapped_dir)
-
-        log += f"Total: {time.time()-start:.1f}s\n"
-        progress(1, desc="Done")
-        return result_path, log
-
-    except Exception as e:
-        log += f"FATAL: {e}\n"
-        return None, log
-
-
-def swap_single_src_multi_video(src_img, dst_videos, dst_indices,
-                                delete_frames_dir=True, add_audio=True,
-                                progress=gr.Progress(track_tqdm=True)):
-    log = ""
-    results = []
-    start = time.time()
-    base = "workdir/SingleSrcMultiVideo"
-    ensure_dirs(base)
-
-    try:
-        if isinstance(dst_indices, str):
-            idx_list = [int(x.strip()) for x in dst_indices.split(",") if x.strip().isdigit()]
-        else:
-            idx_list = [int(x) for x in dst_indices]
-
-        src_path = os.path.join(base, "data_src.jpg")
-        cv2.imwrite(src_path, cv2.cvtColor(src_img, cv2.COLOR_RGB2BGR))
-        log += "Saved source\n"
-
-        from VideoSwapping import extract_frames, frames_to_video
-
-        # Gradio File(file_count="multiple") can return list of paths or list of objects
-        if not isinstance(dst_videos, (list, tuple)):
-            dst_videos = [dst_videos]
-
-        n_videos = len(dst_videos)
-
-        for i, video in enumerate(dst_videos):
-            dst_idx = idx_list[i] if i < len(idx_list) else 1
-            v_base = os.path.join(base, f"v{i}")
-            frames_dir = os.path.join(v_base, "frames")
-            swapped_dir = os.path.join(v_base, "swapped")
-            dst_v = os.path.join(v_base, "dst.mp4")
-            tmp_v = os.path.join(v_base, "tmp.mp4")
-            final_v = os.path.join(v_base, "out.mp4")
-            ensure_dirs(frames_dir, swapped_dir)
-
-            try:
-                video_path = _resolve_video_path(video)
-                shutil.copy(video_path, dst_v)
-            except Exception as e:
-                log += f"Video {i}: cannot resolve path: {e}\n"
-                results.append(None)
-                continue
-
-            frame_paths = extract_frames(dst_v, frames_dir)
-            total = len(frame_paths)
-            log += f"Video {i}: {total} frames\n"
-            progress(i / max(1, n_videos), desc=f"Video {i+1}/{n_videos}")
-
-            for idx, fp in enumerate(frame_paths):
-                out_p = os.path.join(swapped_dir, f"swapped_{idx:05d}.jpg")
-                try:
-                    swapped = swapper.swap_faces(src_path, 1, fp, dst_idx)
-                    cv2.imwrite(out_p, swapped)
-                except Exception:
-                    shutil.copy(fp, out_p)
-
-            cap = cv2.VideoCapture(dst_v)
-            fps = cap.get(cv2.CAP_PROP_FPS) or 25.0
-            cap.release()
-            frames_to_video(swapped_dir, tmp_v, fps)
-
-            if add_audio:
-                ok, _ = add_audio_to_video(dst_v, tmp_v, final_v)
-                results.append(final_v if ok else tmp_v)
-            else:
-                results.append(tmp_v)
-
-            if delete_frames_dir:
-                safe_rmtree(frames_dir)
-            safe_rmtree(swapped_dir)
-
-        log += f"All videos done in {time.time()-start:.1f}s\n"
-        return results, log
-
-    except Exception as e:
-        log += f"FATAL: {e}\n"
-        return results, log
-
-
-# ------------------------------------------------------------------
-# Gradio UI
-# ------------------------------------------------------------------
-welcome = """
-# Face Swapping Suite (GPU-ready)
-All-in-one face swapping for photos **and videos**.  
-Optimised for rented GPUs (CUDA). First run downloads models (~300 MB).
-"""
-
-with gr.Blocks(title="FaceSwap GPU") as demo:
-    gr.Markdown(welcome)
+                results.append(f"Error: {e}")
+    return results
+
+with gr.Blocks(title="Fast Face Swap") as demo:
+    gr.Markdown(
+        "# Fast Face Swapping Suite\n"
+        "CUDA/ONNX Runtime optimized photo and video face swapping. "
+        "The source face is detected once and video frames stay in memory."
+    )
 
     with gr.Tab("Single Photo"):
-        gr.Interface(
-            fn=swap_single_photo,
-            inputs=[
-                gr.Image(label="Source Image", type="numpy"),
-                gr.Number(value=1, label="Source Face Index (1-based)"),
-                gr.Image(label="Destination Image", type="numpy"),
-                gr.Number(value=1, label="Destination Face Index (1-based)"),
-            ],
-            outputs=[
-                gr.Image(label="Result"),
-                gr.Textbox(label="Log", lines=6, interactive=False)
-            ],
-            api_name="single_photo"
-        )
-
-    with gr.Tab("Single Src → Multi Dst"):
-        gr.Interface(
-            fn=swap_single_src_multi_dst,
-            inputs=[
-                gr.Image(label="Source Image", type="numpy"),
-                gr.Gallery(label="Destination Images", columns=3, type="numpy"),
-                gr.Textbox(label="Dst face indices (comma-sep, e.g. 1,1,2)", value="1"),
-            ],
-            outputs=[
-                gr.Gallery(label="Results"),
-                gr.Textbox(label="Log", lines=6, interactive=False)
-            ],
-            api_name="single_src_multi_dst"
-        )
-
-    with gr.Tab("Multi Src → Single Dst"):
-        gr.Interface(
-            fn=swap_multi_src_single_dst,
-            inputs=[
-                gr.Gallery(label="Source Images", columns=3, type="numpy"),
-                gr.Image(label="Destination Image", type="numpy"),
-                gr.Number(value=1, label="Destination Face Index"),
-            ],
-            outputs=[
-                gr.Gallery(label="Results"),
-                gr.Textbox(label="Log", lines=6, interactive=False)
-            ],
-            api_name="multi_src_single_dst"
-        )
-
-    with gr.Tab("Multi Src → Multi Dst"):
-        gr.Interface(
-            fn=swap_multi_src_multi_dst,
-            inputs=[
-                gr.Gallery(label="Source Images", columns=3, type="numpy"),
-                gr.Gallery(label="Destination Images", columns=3, type="numpy"),
-                gr.Textbox(label="Dst face indices (comma-sep)", value="1"),
-            ],
-            outputs=[
-                gr.Gallery(label="Results"),
-                gr.Textbox(label="Log", lines=6, interactive=False)
-            ],
-            api_name="multi_src_multi_dst"
-        )
-
-    with gr.Tab("Custom Face Mapping (Photo)"):
-        gr.Interface(
-            fn=swap_faces_custom,
-            inputs=[
-                gr.Gallery(label="Source Images (order = indices)", columns=3, type="numpy"),
-                gr.Image(label="Destination Image", type="numpy"),
-                gr.Textbox(label="Mapping (e.g. 2,1,3 means face1←src2, face2←src1 …)", value="1"),
-            ],
-            outputs=[
-                gr.Image(label="Result"),
-                gr.Textbox(label="Log", lines=8, interactive=False)
-            ],
-            api_name="custom_photo"
-        )
-
-    with gr.Tab("Video Swapping (single face)"):
-        gr.Interface(
-            fn=swap_video,
-            inputs=[
-                gr.Image(label="Source Face Image", type="numpy"),
-                gr.Number(value=1, label="Source Face Index"),
-                gr.Video(label="Target Video"),
-                gr.Number(value=1, label="Destination Face Index"),
-                gr.Checkbox(label="Delete extracted frames after finish", value=True),
-                gr.Checkbox(label="Add original audio", value=True),
-            ],
-            outputs=[
-                gr.Video(label="Swapped Video"),
-                gr.Textbox(label="Log", lines=10, interactive=False)
-            ],
-            api_name="video_single"
-        )
-
-    with gr.Tab("Video – All Faces"):
-        gr.Interface(
-            fn=swap_video_all_faces,
-            inputs=[
-                gr.Image(label="Source Face Image", type="numpy"),
-                gr.Video(label="Target Video"),
-                gr.Number(value=1, label="How many faces to swap per frame", precision=0),
-                gr.Checkbox(label="Delete extracted frames after finish", value=True),
-                gr.Checkbox(label="Add original audio", value=True),
-            ],
-            outputs=[
-                gr.Video(label="Swapped Video"),
-                gr.Textbox(label="Log", lines=10, interactive=False)
-            ],
-            api_name="video_all_faces"
-        )
-
-    with gr.Tab("Video – Custom Mapping"):
-        gr.Interface(
-            fn=swap_video_custom_mapping,
-            inputs=[
-                gr.Gallery(label="Source Images (order = indices)", columns=3, type="numpy"),
-                gr.Video(label="Target Video"),
-                gr.Textbox(label="Mapping (e.g. 2,1,3)", value="1"),
-                gr.Checkbox(label="Delete extracted frames after finish", value=True),
-                gr.Checkbox(label="Add original audio", value=True),
-            ],
-            outputs=[
-                gr.Video(label="Swapped Video"),
-                gr.Textbox(label="Log", lines=10, interactive=False)
-            ],
-            api_name="video_custom"
-        )
-
-    with gr.Tab("Single Src → Multi Video"):
-        gr.Interface(
-            fn=swap_single_src_multi_video,
-            inputs=[
-                gr.Image(label="Source Face Image", type="numpy"),
-                gr.File(label="Target Videos (multi-select)", file_count="multiple", type="filepath"),
-                gr.Textbox(label="Dst face indices per video (comma-sep)", value="1"),
-                gr.Checkbox(label="Delete extracted frames after each video", value=True),
-                gr.Checkbox(label="Add original audio", value=True),
-            ],
-            outputs=[
-                gr.Gallery(label="Swapped Videos", type="filepath"),
-                gr.Textbox(label="Log", lines=10, interactive=False)
-            ],
-            api_name="single_src_multi_video"
+        with gr.Row():
+            with gr.Column():
+                src = gr.Image(type="numpy", label="Source Image")
+                src_idx = gr.Number(value=1, precision=0, label="Source Face Index")
+            with gr.Column():
+                dst = gr.Image(type="numpy", label="Target Image")
+                dst_idx = gr.Number(value=1, precision=0, label="Target Face Index")
+        det = gr.Dropdown([256, 320, 384, 512], value=320, label="Detector Size")
+        btn = gr.Button("Swap", variant="primary")
+        out = gr.Image(type="numpy", label="Result")
+        btn.click(swap_photo, [src, src_idx, dst, dst_idx, det], out)
+
+    with gr.Tab("Fast Video"):
+        vsrc = gr.Image(type="numpy", label="Source Image")
+        with gr.Row():
+            vid = gr.Video(label="Target Video", type="filepath")
+            vdst_idx = gr.Number(value=1, precision=0, label="Target Face Index")
+        with gr.Row():
+            vdet = gr.Dropdown([256, 320, 384, 512], value=320, label="Detector Size")
+            interval = gr.Slider(1, 5, value=1, step=1,
+                                  label="Face detection interval (1 = best tracking accuracy)")
+            quality = gr.Slider(75, 98, value=92, step=1, label="JPEG fallback quality")
+            audio = gr.Checkbox(value=True, label="Preserve original audio")
+        vbtn = gr.Button("Swap Video", variant="primary")
+        vout = gr.Video(label="Output Video")
+        vbtn.click(
+            swap_video,
+            [vsrc, gr.Number(value=1, visible=False), vid, vdst_idx,
+             vdet, interval, quality, audio],
+            vout
         )
 
+    with gr.Tab("Multi Source → Single Target"):
+        ms = gr.File(file_count="multiple", file_types=["image"], label="Source Images")
+        md = gr.Image(type="numpy", label="Target Image")
+        mi = gr.Number(value=1, precision=0, label="Target Face Index")
+        mb = gr.Button("Process", variant="primary")
+        mo = gr.Gallery(label="Results", columns=3)
+        mb.click(swap_multi_source_single, [ms, md, mi, det], mo)
+
+    with gr.Tab("Multi Source → Multi Target"):
+        mss = gr.File(file_count="multiple", file_types=["image"], label="Source Images")
+        mdd = gr.File(file_count="multiple", file_types=["image"], label="Target Images")
+        mids = gr.Textbox(value="1", label="Target face indices, comma-separated")
+        mdb = gr.Button("Process", variant="primary")
+        mdo = gr.Gallery(label="Results", columns=3)
+        mdb.click(swap_multi_source_multi, [mss, mdd, mids, det], mdo)
 
 if __name__ == "__main__":
-    parser = argparse.ArgumentParser()
-    parser.add_argument("--share", action="store_true", help="Create a public Gradio link")
-    parser.add_argument("--server-name", default="0.0.0.0")
-    parser.add_argument("--server-port", type=int, default=7860)
-    args = parser.parse_args()
-    demo.queue(max_size=4).launch(
-        share=args.share,
-        server_name=args.server_name,
-        server_port=args.server_port,
+    demo.queue(default_concurrency_limit=1, max_size=8).launch(
+        server_name="0.0.0.0",
+        server_port=int(os.getenv("PORT", "7860")),
         show_error=True,
-        ssr_mode=False,   # prevents the SvelteKit 405 / Content-Length errors
     )

benchmark.py

@@ -0,0 +1,21 @@
+
+import os, time, cv2
+from optimized_swapper import FaceSwapEngine
+
+src = cv2.imread("benchmark_source.jpg")
+dst = cv2.imread("benchmark_target.jpg")
+if src is None or dst is None:
+    raise SystemExit("Put benchmark_source.jpg and benchmark_target.jpg in the root.")
+
+engine = FaceSwapEngine()
+for _ in range(2):
+    engine.swap_image(src, 1, dst, 1, 320)
+
+n = 20
+t0 = time.perf_counter()
+for _ in range(n):
+    engine.swap_prepared_source(dst, 1, 320, 1)
+dt = time.perf_counter() - t0
+print(f"{n} swaps: {dt:.3f}s")
+print(f"Average: {dt/n*1000:.2f} ms/image")
+print(f"Throughput: {n/dt:.2f} images/sec")

download_model.py

@@ -0,0 +1,19 @@
+"""Download the large InsightFace swap model when it is not already present."""
+from pathlib import Path
+from huggingface_hub import hf_hub_download
+
+TARGET = Path("inswapper_128.onnx")
+REPO_ID = "dosesnrolls1/FaceSwapAll"
+FILENAME = "inswapper_128.onnx"
+
+if TARGET.exists() and TARGET.stat().st_size > 100_000_000:
+    print(f"Model already present: {TARGET}")
+else:
+    downloaded = hf_hub_download(
+        repo_id=REPO_ID,
+        filename=FILENAME,
+        repo_type="space",
+        local_dir=".",
+        local_dir_use_symlinks=False,
+    )
+    print(f"Downloaded: {downloaded}")

optimized_swapper.py

@@ -0,0 +1,243 @@
+
+import os
+import subprocess
+import tempfile
+import time
+from pathlib import Path
+
+import cv2
+import numpy as np
+import insightface
+from insightface.app import FaceAnalysis
+
+MODEL_PATH = os.getenv("INSWAPPER_MODEL", "inswapper_128.onnx")
+GPU = os.getenv("FACE_SWAP_GPU", "1") not in {"0", "false", "False", "cpu"}
+CUDA_DEVICE = int(os.getenv("CUDA_DEVICE_ID", "0"))
+
+class FaceSwapEngine:
+    """Performance-focused InsightFace engine.
+
+    Key optimizations:
+      * one persistent FaceAnalysis model
+      * one persistent inswapper model
+      * source face is cached and reused
+      * NumPy/OpenCV frames stay in memory
+      * smaller detector size by default
+      * target face detection can be throttled for video
+      * final video is encoded once by ffmpeg
+    """
+
+    def __init__(self):
+        self._source_face = None
+        self._source_key = None
+        self._source_idx = None
+        self._det_size = None
+
+        providers = []
+        if GPU:
+            providers.append((
+                "CUDAExecutionProvider",
+                {
+                    "device_id": CUDA_DEVICE,
+                    "arena_extend_strategy": "kSameAsRequested",
+                    "cudnn_conv_algo_search": "EXHAUSTIVE",
+                    "do_copy_in_default_stream": True,
+                },
+            ))
+        providers.append("CPUExecutionProvider")
+
+        # Keep detector/model objects alive for the whole Space lifetime.
+        self.app = FaceAnalysis(
+            name=os.getenv("INSIGHTFACE_MODEL", "buffalo_l"),
+            providers=providers,
+        )
+        self.app.prepare(
+            ctx_id=CUDA_DEVICE if GPU else -1,
+            det_size=(320, 320),
+        )
+
+        self.swapper = insightface.model_zoo.get_model(
+            MODEL_PATH,
+            providers=providers,
+        )
+
+        self.providers = providers
+        self.gpu = GPU
+
+    @staticmethod
+    def _img_key(img):
+        # Avoid copying the complete image just to build a cache key.
+        return (id(img), img.shape, img.dtype.str)
+
+    @staticmethod
+    def _sort_faces(faces):
+        return sorted(faces, key=lambda f: float(f.bbox[0]))
+
+    def _detect(self, image, det_size=320, max_num=0):
+        # InsightFace's FaceAnalysis allows max_num through get().
+        if self._det_size != det_size:
+            self.app.prepare(
+                ctx_id=CUDA_DEVICE if self.gpu else -1,
+                det_size=(det_size, det_size),
+            )
+            self._det_size = det_size
+        try:
+            return self._sort_faces(self.app.get(image, max_num=max_num))
+        except TypeError:
+            return self._sort_faces(self.app.get(image))
+
+    def prepare_source(self, source, source_idx=1, det_size=320):
+        faces = self._detect(source, det_size, max_num=0)
+        if len(faces) < source_idx:
+            raise ValueError(
+                f"Source image contains {len(faces)} faces; requested face {source_idx}."
+            )
+        self._source_face = faces[source_idx - 1]
+        self._source_idx = source_idx
+        self._source_key = self._img_key(source)
+        return self._source_face
+
+    def swap_prepared_source(self, target, target_idx=1, det_size=320, max_faces=0):
+        if self._source_face is None:
+            raise RuntimeError("Source face has not been prepared.")
+        faces = self._detect(target, det_size, max_num=max_faces)
+        if len(faces) < target_idx:
+            raise ValueError(
+                f"Target image contains {len(faces)} faces; requested face {target_idx}."
+            )
+        return self.swapper.get(
+            target,
+            faces[target_idx - 1],
+            self._source_face,
+            paste_back=True,
+        )
+
+    def swap_image(self, source, source_idx, target, target_idx, det_size=320):
+        self.prepare_source(source, source_idx, det_size)
+        # If target index is 1, detector can stop after first face.
+        max_faces = target_idx if target_idx > 1 else 1
+        return self.swap_prepared_source(target, target_idx, det_size, max_faces)
+
+    def _detect_video_face(self, frame, target_idx, det_size):
+        max_faces = target_idx if target_idx > 1 else 1
+        faces = self._detect(frame, det_size, max_num=max_faces)
+        if len(faces) < target_idx:
+            raise ValueError("Target face not found.")
+        return faces[target_idx - 1]
+
+    def _ffmpeg_encode(self, frames_dir, output_path, fps, audio_source=None):
+        pattern = str(Path(frames_dir) / "frame_%08d.jpg")
+        cmd = [
+            "ffmpeg", "-y",
+            "-loglevel", "error",
+            "-framerate", f"{fps:.6f}",
+            "-i", pattern,
+        ]
+        if audio_source:
+            cmd += [
+                "-i", audio_source,
+                "-map", "0:v:0", "-map", "1:a?",
+                "-c:v", "libx264", "-preset", "veryfast",
+                "-crf", "18", "-pix_fmt", "yuv420p",
+                "-c:a", "aac", "-b:a", "160k",
+                "-shortest",
+            ]
+        else:
+            cmd += [
+                "-c:v", "libx264", "-preset", "veryfast",
+                "-crf", "18", "-pix_fmt", "yuv420p",
+            ]
+        cmd.append(output_path)
+        subprocess.run(cmd, check=True)
+
+    def swap_video(
+        self, source, source_idx, video_path, target_idx,
+        det_size=320, detection_interval=1, jpeg_quality=92,
+        preserve_audio=True
+    ):
+        self.prepare_source(source, source_idx, det_size)
+
+        work = tempfile.mkdtemp(prefix="faceswap_")
+        frames_dir = os.path.join(work, "frames")
+        os.makedirs(frames_dir, exist_ok=True)
+        output = os.path.join(work, "swapped.mp4")
+
+        cap = cv2.VideoCapture(video_path)
+        if not cap.isOpened():
+            raise ValueError("Could not open video.")
+
+        fps = cap.get(cv2.CAP_PROP_FPS) or 30.0
+        frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT) or 0)
+
+        # We encode final frames from JPEGs only. Unlike the old implementation,
+        # source/target frames are never written and then read back before swapping.
+        idx = 0
+        last_face = None
+        failed = 0
+        t0 = time.perf_counter()
+
+        try:
+            while True:
+                ok, frame = cap.read()
+                if not ok:
+                    break
+
+                try:
+                    if last_face is None or idx % detection_interval == 0:
+                        last_face = self._detect_video_face(
+                            frame, target_idx, det_size
+                        )
+
+                    swapped = self.swapper.get(
+                        frame, last_face, self._source_face, paste_back=True
+                    )
+                except Exception:
+                    # If a cached face becomes invalid, immediately re-detect.
+                    try:
+                        last_face = self._detect_video_face(
+                            frame, target_idx, det_size
+                        )
+                        swapped = self.swapper.get(
+                            frame, last_face, self._source_face, paste_back=True
+                        )
+                    except Exception:
+                        swapped = frame
+                        failed += 1
+
+                path = os.path.join(frames_dir, f"frame_{idx:08d}.jpg")
+                cv2.imwrite(
+                    path,
+                    swapped,
+                    [int(cv2.IMWRITE_JPEG_QUALITY), int(jpeg_quality)],
+                )
+                idx += 1
+
+            cap.release()
+            self._ffmpeg_encode(
+                frames_dir,
+                output,
+                fps,
+                audio_source=video_path if preserve_audio else None,
+            )
+
+            # Copy result to a stable temp path that Gradio can serve after
+            # the worker returns.
+            stable = tempfile.NamedTemporaryFile(
+                prefix="faceswap_result_", suffix=".mp4", delete=False
+            ).name
+            with open(output, "rb") as srcf, open(stable, "wb") as dstf:
+                while True:
+                    chunk = srcf.read(8 * 1024 * 1024)
+                    if not chunk:
+                        break
+                    dstf.write(chunk)
+
+            elapsed = time.perf_counter() - t0
+            print(
+                f"[FaceSwap] processed {idx} frames in {elapsed:.2f}s "
+                f"({idx / max(elapsed, 1e-6):.2f} FPS), failures={failed}"
+            )
+            return stable
+        finally:
+            import shutil
+            shutil.rmtree(work, ignore_errors=True)

packages.txt

@@ -1,3 +1,4 @@
+
 ffmpeg
 libgl1
 libglib2.0-0

requirements.txt

@@ -1,8 +1,15 @@
-numpy
-opencv-python-headless
-onnxruntime-gpu==1.20.2
-scikit-learn
-scikit-image
+
+# Core UI / image processing
+gradio>=5.44,<6
+numpy<2
+opencv-python-headless>=4.10
+pillow>=10
+
+# Face analysis / ONNX Runtime
+insightface==0.7.3
+onnx>=1.16,<2
+onnxruntime-gpu>=1.20,<1.24
+
+# Utilities
+psutil
 tqdm
-insightface
-gradio>=4.0.0