app.py

from __future__ import annotations

import os
import shutil
import subprocess
import tempfile
import threading
import time
import uuid
from pathlib import Path

import cv2
import gradio as gr
import numpy as np
import onnxruntime as ort
import spaces
from huggingface_hub import hf_hub_download
from insightface.app import FaceAnalysis


APP_TITLE = "Dream Video Face Swap — ZeroGPU"
MAX_VIDEO_SECONDS = 45
OUTPUT_ROOT = Path(tempfile.gettempdir()) / "dream_video_face_swap"
MODEL_ROOT = Path(tempfile.gettempdir()) / "insightface"
OUTPUT_ROOT.mkdir(parents=True, exist_ok=True)
MODEL_ROOT.mkdir(parents=True, exist_ok=True)

# HyperSwap works at 256x256 instead of the previous 128x128 InSwapper crop.
# Both models are downloaded from FaceFusion's official Hugging Face repos.
SWAPPER_MODEL_PATH = hf_hub_download(
    repo_id="facefusion/models-3.3.0",
    filename="hyperswap_1a_256.onnx",
)
ENHANCER_MODEL_PATH = hf_hub_download(
    repo_id="facefusion/models-3.0.0",
    filename="gfpgan_1.4.onnx",
)

_MODEL_LOCK = threading.Lock()
_MODEL_CACHE: tuple[FaceAnalysis, ort.InferenceSession, ort.InferenceSession] | None = None

QUALITY_BALANCED = "Balanced — HyperSwap 256"
QUALITY_BEST = "Best quality — HyperSwap 256 + GFPGAN 512"

ARCFACE_128_TEMPLATE = np.array(
    [
        [0.36167656, 0.40387734],
        [0.63696719, 0.40235469],
        [0.50019687, 0.56044219],
        [0.38710391, 0.72160547],
        [0.61507734, 0.72034453],
    ],
    dtype=np.float32,
)
FFHQ_512_TEMPLATE = np.array(
    [
        [0.37691676, 0.46864664],
        [0.62285697, 0.46912813],
        [0.50123859, 0.61331904],
        [0.39308822, 0.72541100],
        [0.61150205, 0.72490465],
    ],
    dtype=np.float32,
)


def _cleanup_old_outputs(max_age_seconds: int = 3600) -> None:
    """Remove old anonymous outputs so uploaded media is not retained."""
    now = time.time()
    for path in OUTPUT_ROOT.iterdir():
        try:
            if now - path.stat().st_mtime > max_age_seconds:
                if path.is_dir():
                    shutil.rmtree(path, ignore_errors=True)
                else:
                    path.unlink(missing_ok=True)
        except OSError:
            continue


def _video_metadata(video_path: str) -> tuple[float, int, int, float, int]:
    cap = cv2.VideoCapture(video_path)
    if not cap.isOpened():
        raise ValueError("The uploaded video could not be opened.")

    fps = float(cap.get(cv2.CAP_PROP_FPS) or 0)
    width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH) or 0)
    height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT) or 0)
    frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT) or 0)
    cap.release()

    if fps <= 0 or width <= 0 or height <= 0:
        raise ValueError("The video metadata is invalid or unsupported.")

    duration = frame_count / fps if frame_count > 0 else 0.0
    return fps, width, height, duration, frame_count


def _estimate_gpu_seconds(
    source_image: str | None,
    target_video: str | None,
    target_face_index: int,
    clip_length: int,
    output_resolution: str,
    quality_mode: str,
    keep_audio: bool,
    add_watermark: bool,
    consent: bool,
) -> int:
    """Request only the ZeroGPU time that the submitted clip likely needs."""
    del source_image, target_face_index, output_resolution, keep_audio, add_watermark, consent
    seconds = float(clip_length or 10)
    if target_video:
        try:
            seconds = min(_video_metadata(target_video)[3] or seconds, seconds)
        except Exception:
            pass
    seconds_per_second = 8.0 if quality_mode == QUALITY_BEST else 5.5
    return int(max(120, min(300, 60 + seconds * seconds_per_second)))


def _load_models() -> tuple[FaceAnalysis, ort.InferenceSession, ort.InferenceSession]:
    """Create CUDA ONNX sessions only after ZeroGPU has allocated a GPU."""
    global _MODEL_CACHE
    with _MODEL_LOCK:
        if _MODEL_CACHE is not None:
            return _MODEL_CACHE

        providers = ["CUDAExecutionProvider", "CPUExecutionProvider"]
        analyser = FaceAnalysis(
            name="buffalo_l",
            root=str(MODEL_ROOT),
            allowed_modules=["detection", "recognition"],
            providers=providers,
        )
        analyser.prepare(ctx_id=0, det_size=(640, 640))
        session_options = ort.SessionOptions()
        session_options.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL
        swapper = ort.InferenceSession(
            SWAPPER_MODEL_PATH, sess_options=session_options, providers=providers
        )
        enhancer = ort.InferenceSession(
            ENHANCER_MODEL_PATH, sess_options=session_options, providers=providers
        )
        _MODEL_CACHE = (analyser, swapper, enhancer)
        return _MODEL_CACHE


def _input_dtype(session: ort.InferenceSession, input_name: str) -> np.dtype:
    input_type = next(item.type for item in session.get_inputs() if item.name == input_name)
    if "float16" in input_type:
        return np.dtype(np.float16)
    if "double" in input_type:
        return np.dtype(np.float64)
    return np.dtype(np.float32)


def _warp_face(
    frame: np.ndarray,
    landmarks: np.ndarray,
    template: np.ndarray,
    size: tuple[int, int],
) -> tuple[np.ndarray, np.ndarray]:
    destination = template * np.array(size, dtype=np.float32)
    matrix = cv2.estimateAffinePartial2D(
        np.asarray(landmarks, dtype=np.float32),
        destination,
        method=cv2.RANSAC,
        ransacReprojThreshold=100,
    )[0]
    if matrix is None:
        raise RuntimeError("Could not align the target face.")
    crop = cv2.warpAffine(
        frame,
        matrix,
        size,
        flags=cv2.INTER_LINEAR,
        borderMode=cv2.BORDER_REPLICATE,
    )
    return crop, matrix


def _soft_face_mask(size: tuple[int, int], blur: float = 0.12) -> np.ndarray:
    width, height = size
    mask = np.zeros((height, width), dtype=np.float32)
    center = (width // 2, int(height * 0.53))
    axes = (int(width * 0.43), int(height * 0.48))
    cv2.ellipse(mask, center, axes, 0, 0, 360, 1.0, -1, cv2.LINE_AA)
    sigma = max(1.0, width * blur * 0.35)
    return cv2.GaussianBlur(mask, (0, 0), sigma).clip(0, 1)


def _paste_face(
    frame: np.ndarray,
    crop: np.ndarray,
    mask: np.ndarray,
    affine_matrix: np.ndarray,
) -> np.ndarray:
    height, width = frame.shape[:2]
    inverse_matrix = cv2.invertAffineTransform(affine_matrix)
    pasted = cv2.warpAffine(
        crop,
        inverse_matrix,
        (width, height),
        flags=cv2.INTER_LINEAR,
        borderMode=cv2.BORDER_REPLICATE,
    )
    pasted_mask = cv2.warpAffine(
        mask,
        inverse_matrix,
        (width, height),
        flags=cv2.INTER_LINEAR,
    ).clip(0, 1)[..., None]
    return (frame * (1 - pasted_mask) + pasted * pasted_mask).astype(np.uint8)


def _swap_face_hq(
    frame: np.ndarray,
    landmarks: np.ndarray,
    source_embedding: np.ndarray,
    swapper: ort.InferenceSession,
) -> np.ndarray:
    size = (256, 256)
    crop, matrix = _warp_face(frame, landmarks, ARCFACE_128_TEMPLATE, size)
    target = crop[:, :, ::-1].astype(np.float32) / 255.0
    target = ((target - 0.5) / 0.5).transpose(2, 0, 1)[None]

    inputs: dict[str, np.ndarray] = {}
    for item in swapper.get_inputs():
        if item.name == "source":
            inputs[item.name] = source_embedding.astype(_input_dtype(swapper, item.name))
        elif item.name == "target":
            inputs[item.name] = target.astype(_input_dtype(swapper, item.name))
    output = swapper.run(None, inputs)[0][0].transpose(1, 2, 0)
    output = ((output * 0.5 + 0.5).clip(0, 1)[:, :, ::-1] * 255).astype(np.uint8)
    return _paste_face(frame, output, _soft_face_mask(size), matrix)


def _enhance_face(
    frame: np.ndarray,
    landmarks: np.ndarray,
    enhancer: ort.InferenceSession,
    blend: float = 0.75,
) -> np.ndarray:
    size = (512, 512)
    crop, matrix = _warp_face(frame, landmarks, FFHQ_512_TEMPLATE, size)
    prepared = crop[:, :, ::-1].astype(np.float32) / 255.0
    prepared = ((prepared - 0.5) / 0.5).transpose(2, 0, 1)[None]

    inputs: dict[str, np.ndarray] = {}
    for item in enhancer.get_inputs():
        if item.name == "input":
            inputs[item.name] = prepared.astype(_input_dtype(enhancer, item.name))
        elif item.name == "weight":
            inputs[item.name] = np.array([0.5], dtype=_input_dtype(enhancer, item.name))
    output = enhancer.run(None, inputs)[0][0].clip(-1, 1)
    output = (((output + 1) * 0.5).transpose(1, 2, 0)[:, :, ::-1] * 255).astype(np.uint8)
    enhanced = _paste_face(frame, output, _soft_face_mask(size, blur=0.10), matrix)
    return cv2.addWeighted(frame, 1.0 - blend, enhanced, blend, 0)


def _resize_to_limit(frame: np.ndarray, resolution: str) -> np.ndarray:
    limits = {"720p (fast)": 1280, "1080p": 1920, "Original": None}
    limit = limits.get(resolution, 1280)
    if limit is None:
        return frame

    height, width = frame.shape[:2]
    longest = max(width, height)
    if longest <= limit:
        return frame

    scale = limit / longest
    new_width = max(2, int(width * scale) // 2 * 2)
    new_height = max(2, int(height * scale) // 2 * 2)
    return cv2.resize(frame, (new_width, new_height), interpolation=cv2.INTER_AREA)


def _largest_face(faces: list) -> object:
    return max(
        faces,
        key=lambda face: float((face.bbox[2] - face.bbox[0]) * (face.bbox[3] - face.bbox[1])),
    )


def _embedding(face: object) -> np.ndarray:
    vector = np.asarray(face.embedding, dtype=np.float32)
    norm = float(np.linalg.norm(vector))
    return vector / max(norm, 1e-8)


def _select_tracked_face(faces: list, target_embedding: np.ndarray) -> object:
    return max(faces, key=lambda face: float(np.dot(_embedding(face), target_embedding)))


def _add_ai_watermark(frame: np.ndarray, text: str = "AI FACE SWAP") -> np.ndarray:
    font = cv2.FONT_HERSHEY_SIMPLEX
    scale = max(0.45, min(frame.shape[0], frame.shape[1]) / 1800)
    thickness = max(1, int(round(scale * 2)))
    (text_width, text_height), _ = cv2.getTextSize(text, font, scale, thickness)
    x = max(12, frame.shape[1] - text_width - 16)
    y = max(text_height + 12, frame.shape[0] - 16)
    cv2.putText(frame, text, (x, y), font, scale, (0, 0, 0), thickness + 3, cv2.LINE_AA)
    cv2.putText(frame, text, (x, y), font, scale, (255, 255, 255), thickness, cv2.LINE_AA)
    return frame


def _encode_final_video(
    silent_video: Path,
    source_video: str,
    output_video: Path,
    keep_audio: bool,
    processed_seconds: float,
) -> None:
    command = ["ffmpeg", "-y", "-loglevel", "error", "-i", str(silent_video)]
    if keep_audio:
        command += [
            "-i",
            source_video,
            "-map",
            "0:v:0",
            "-map",
            "1:a?",
            "-c:v",
            "libx264",
            "-preset",
            "veryfast",
            "-crf",
            "18",
            "-c:a",
            "aac",
            "-b:a",
            "192k",
            "-t",
            f"{processed_seconds:.3f}",
            "-movflags",
            "+faststart",
            str(output_video),
        ]
    else:
        command += [
            "-map",
            "0:v:0",
            "-c:v",
            "libx264",
            "-preset",
            "veryfast",
            "-crf",
            "18",
            "-an",
            "-movflags",
            "+faststart",
            str(output_video),
        ]

    subprocess.run(command, check=True, capture_output=True, text=True)


def _estimate_head_gpu_seconds(
    source_image: str | None,
    target_video: str | None,
    clip_length: int,
    keep_audio: bool,
    add_watermark: bool,
    consent: bool,
) -> int:
    del source_image, keep_audio, add_watermark, consent
    seconds = float(clip_length or 3)
    if target_video:
        try:
            seconds = min(seconds, _video_metadata(target_video)[3] or seconds)
        except Exception:
            pass
    return int(max(180, min(300, 120 + seconds * 45)))


@spaces.GPU(duration=_estimate_gpu_seconds)
def swap_video_face(
    source_image: str | None,
    target_video: str | None,
    target_face_index: int,
    clip_length: int,
    output_resolution: str,
    quality_mode: str,
    keep_audio: bool,
    add_watermark: bool,
    consent: bool,
) -> tuple[str, str]:
    if not consent:
        raise gr.Error("Please confirm that you have permission to use the uploaded media.")
    if not source_image:
        raise gr.Error("Upload a clear source-face image.")
    if not target_video:
        raise gr.Error("Upload a target video.")

    _cleanup_old_outputs()
    fps, _, _, duration, _ = _video_metadata(target_video)
    requested_seconds = min(float(clip_length), float(MAX_VIDEO_SECONDS))
    if duration > 0:
        requested_seconds = min(requested_seconds, duration)
    frame_limit = max(1, int(round(requested_seconds * fps)))

    analyser, swapper, enhancer = _load_models()
    source_frame = cv2.imread(source_image)
    if source_frame is None:
        raise gr.Error("The source image format is unsupported.")
    source_frame = _resize_to_limit(source_frame, "1080p")
    source_faces = analyser.get(source_frame)
    if not source_faces:
        raise gr.Error("No face was detected in the source image. Use a clear, front-facing portrait.")
    source_face = _largest_face(source_faces)
    source_embedding = _embedding(source_face).reshape(1, -1)

    job_dir = OUTPUT_ROOT / uuid.uuid4().hex
    job_dir.mkdir(parents=True, exist_ok=False)
    silent_path = job_dir / "silent.mp4"
    final_path = job_dir / "face_swap_result.mp4"

    cap = cv2.VideoCapture(target_video)
    writer: cv2.VideoWriter | None = None
    tracked_embedding: np.ndarray | None = None
    smoothed_landmarks: np.ndarray | None = None
    missed_face_frames = 0
    swapped_frames = 0
    processed_frames = 0
    face_index = max(1, int(target_face_index))

    try:
        for frame_number in range(frame_limit):
            ok, frame = cap.read()
            if not ok:
                break

            frame = _resize_to_limit(frame, output_resolution)
            if writer is None:
                height, width = frame.shape[:2]
                writer = cv2.VideoWriter(
                    str(silent_path),
                    cv2.VideoWriter_fourcc(*"mp4v"),
                    fps,
                    (width, height),
                )
                if not writer.isOpened():
                    raise RuntimeError("Could not initialize the video encoder.")

            faces = analyser.get(frame)
            if faces:
                if tracked_embedding is None:
                    ordered_faces = sorted(faces, key=lambda face: float(face.bbox[0]))
                    if len(ordered_faces) >= face_index:
                        target_face = ordered_faces[face_index - 1]
                        tracked_embedding = _embedding(target_face)
                    else:
                        target_face = None
                else:
                    target_face = _select_tracked_face(faces, tracked_embedding)

                if target_face is not None:
                    current_landmarks = np.asarray(target_face.kps, dtype=np.float32)
                    if smoothed_landmarks is None or missed_face_frames > 2:
                        smoothed_landmarks = current_landmarks
                    else:
                        smoothed_landmarks = 0.72 * current_landmarks + 0.28 * smoothed_landmarks
                    missed_face_frames = 0
                    frame = _swap_face_hq(
                        frame,
                        smoothed_landmarks,
                        source_embedding,
                        swapper,
                    )
                    if quality_mode == QUALITY_BEST:
                        frame = _enhance_face(frame, smoothed_landmarks, enhancer)
                    swapped_frames += 1
            else:
                missed_face_frames += 1

            if add_watermark:
                frame = _add_ai_watermark(frame)
            writer.write(frame)
            processed_frames += 1
    except Exception:
        shutil.rmtree(job_dir, ignore_errors=True)
        raise
    finally:
        cap.release()
        if writer is not None:
            writer.release()

    if processed_frames == 0:
        shutil.rmtree(job_dir, ignore_errors=True)
        raise gr.Error("No readable frames were found in the target video.")
    if tracked_embedding is None or swapped_frames == 0:
        shutil.rmtree(job_dir, ignore_errors=True)
        raise gr.Error(
            f"Target face #{face_index} was not found. Try face #1 or choose a video where the face is clearer."
        )

    processed_seconds = processed_frames / fps
    try:
        _encode_final_video(
            silent_path,
            target_video,
            final_path,
            bool(keep_audio),
            processed_seconds,
        )
        silent_path.unlink(missing_ok=True)
    except subprocess.CalledProcessError as error:
        shutil.rmtree(job_dir, ignore_errors=True)
        details = (error.stderr or "FFmpeg encoding failed.")[-500:]
        raise gr.Error(f"Could not create the final MP4: {details}") from error

    status = (
        f"Completed {processed_seconds:.1f}s at {fps:.2f} FPS — "
        f"face replaced in {swapped_frames}/{processed_frames} frames using {quality_mode}."
    )
    return str(final_path), status


@spaces.GPU(duration=_estimate_head_gpu_seconds)
def swap_video_head(
    source_image: str | None,
    target_video: str | None,
    clip_length: int,
    keep_audio: bool,
    add_watermark: bool,
    consent: bool,
) -> tuple[str, str]:
    if not consent:
        raise gr.Error("Please confirm that you have permission to use the uploaded media.")
    if not source_image:
        raise gr.Error("Upload a clear source head image including the complete hairstyle.")
    if not target_video:
        raise gr.Error("Upload a target video.")

    from ghost_runtime import get_ghost_engine

    _cleanup_old_outputs()
    fps, _, _, duration, _ = _video_metadata(target_video)
    requested_seconds = min(float(clip_length), 5.0, duration or 5.0)
    frame_limit = max(1, int(round(requested_seconds * fps)))

    source_frame = cv2.imread(source_image)
    if source_frame is None:
        raise gr.Error("The source image format is unsupported.")

    try:
        engine = get_ghost_engine()
        prepared_source = engine.prepare_source(source_frame)
    except Exception as error:
        raise gr.Error(f"Could not initialize GHOST 2.0 or detect the source head: {error}") from error

    job_dir = OUTPUT_ROOT / uuid.uuid4().hex
    job_dir.mkdir(parents=True, exist_ok=False)
    silent_path = job_dir / "silent_head_swap.mp4"
    final_path = job_dir / "head_swap_result.mp4"
    cap = cv2.VideoCapture(target_video)
    writer: cv2.VideoWriter | None = None
    processed_frames = 0
    swapped_frames = 0

    try:
        for _ in range(frame_limit):
            ok, frame = cap.read()
            if not ok:
                break
            frame = _resize_to_limit(frame, "720p (fast)")
            if writer is None:
                height, width = frame.shape[:2]
                writer = cv2.VideoWriter(
                    str(silent_path), cv2.VideoWriter_fourcc(*"mp4v"), fps, (width, height)
                )
                if not writer.isOpened():
                    raise RuntimeError("Could not initialize the video encoder.")
            try:
                frame = engine.swap_frame(prepared_source, frame)
                swapped_frames += 1
            except ValueError:
                pass
            if add_watermark:
                frame = _add_ai_watermark(frame, "AI HEAD SWAP")
            writer.write(frame)
            processed_frames += 1
    except Exception:
        shutil.rmtree(job_dir, ignore_errors=True)
        raise
    finally:
        cap.release()
        if writer is not None:
            writer.release()

    if processed_frames == 0 or swapped_frames == 0:
        shutil.rmtree(job_dir, ignore_errors=True)
        raise gr.Error("No clear target head was detected in the selected part of the video.")

    processed_seconds = processed_frames / fps
    try:
        _encode_final_video(silent_path, target_video, final_path, keep_audio, processed_seconds)
        silent_path.unlink(missing_ok=True)
    except subprocess.CalledProcessError as error:
        shutil.rmtree(job_dir, ignore_errors=True)
        raise gr.Error("Could not create the final Head Swap MP4.") from error

    status = (
        f"GHOST 2.0 Head Swap completed {processed_seconds:.1f}s — "
        f"head replaced in {swapped_frames}/{processed_frames} frames at 720p."
    )
    return str(final_path), status


CSS = """
.gradio-container {max-width: 1180px !important; margin: 0 auto !important;}
.hero {text-align: center; margin-bottom: 1rem;}
.hero h1 {font-size: 2.2rem; margin-bottom: .25rem;}
.notice {border: 1px solid rgba(128,128,128,.25); border-radius: 12px; padding: 12px 14px;}
"""

with gr.Blocks(title=APP_TITLE, theme=gr.themes.Soft(), css=CSS) as demo:
    gr.HTML(
        """
        <div class="hero">
          <h1>🎭 Dream Video Face Swap</h1>
          <p>High-quality ZeroGPU face replacement with HyperSwap, enhancement, identity tracking, and audio preservation.</p>
          <p dir="rtl">تبديل وجه عالي الجودة عبر ZeroGPU باستخدام HyperSwap وتحسين تفاصيل الوجه مع تتبع الشخص والاحتفاظ بالصوت.</p>
        </div>
        """
    )

    with gr.Tab("Face Swap / تبديل الوجه"):
      with gr.Row():
        with gr.Column(scale=1):
            source_input = gr.Image(
                label="1. Source face / صورة الوجه",
                type="filepath",
                sources=["upload"],
                height=300,
            )
            target_input = gr.Video(
                label="2. Target video / الفيديو المستهدف",
                sources=["upload"],
                format="mp4",
            )

        with gr.Column(scale=1):
            target_face_index = gr.Slider(
                minimum=1,
                maximum=5,
                step=1,
                value=1,
                label="Target face number (left to right in the first clear frame)",
            )
            clip_length = gr.Slider(
                minimum=3,
                maximum=MAX_VIDEO_SECONDS,
                step=1,
                value=10,
                label="Maximum clip length (seconds)",
            )
            output_resolution = gr.Radio(
                choices=["720p (fast)", "1080p", "Original"],
                value="1080p",
                label="Output resolution",
            )
            quality_mode = gr.Radio(
                choices=[QUALITY_BEST, QUALITY_BALANCED],
                value=QUALITY_BEST,
                label="Quality / الجودة",
                info="Best quality restores facial detail at 512×512; Balanced is faster.",
            )
            keep_audio = gr.Checkbox(value=True, label="Keep original audio")
            add_watermark = gr.Checkbox(value=True, label="Add ‘AI FACE SWAP’ watermark")
            consent = gr.Checkbox(
                value=False,
                label="I own this media or have permission from everyone depicted.",
            )
            run_button = gr.Button("Swap Face / تبديل الوجه", variant="primary", size="lg")

      output_video = gr.Video(label="Result / النتيجة", format="mp4")
      status_output = gr.Textbox(label="Status", interactive=False)

      gr.Markdown(
        """
<div class="notice">

**Tips / نصائح:** Use a sharp, front-facing source portrait. For multiple people, select the target by its left-to-right position in the first clear frame. “Best quality” uses HyperSwap 256 plus GFPGAN 512; start with a 5–10 second clip.

استخدم صورة أمامية واضحة للوجه. إذا ظهر أكثر من شخص، اختر رقم الشخص حسب ترتيبه من اليسار إلى اليمين في أول لقطة واضحة.

Uploads are processed temporarily and old results are automatically removed. Do not use the app for impersonation, fraud, harassment, or non-consensual content.

</div>
        """
      )

    with gr.Tab("Head Swap — GHOST 2.0 Beta / تبديل الرأس"):
        gr.Markdown(
            """
**GHOST 2.0 Beta** replaces the complete head, including hair and head shape. It is much
heavier than Face Swap, so this first version is limited to **5 seconds at 720p**.

يستبدل الرأس كاملًا بما يشمل الشعر وشكل الرأس. استخدم صورة واضحة يظهر فيها الشعر كاملًا،
وابدأ بفيديو أمامي قليل الحركة للحصول على أفضل ثبات.
            """
        )
        with gr.Row():
            with gr.Column():
                head_source = gr.Image(
                    label="1. Source head including hair / صورة الرأس والشعر كاملًا",
                    type="filepath", sources=["upload"], height=300,
                )
                head_target = gr.Video(
                    label="2. Target video / الفيديو المستهدف", sources=["upload"], format="mp4"
                )
            with gr.Column():
                head_clip_length = gr.Slider(
                    minimum=1, maximum=5, step=1, value=3,
                    label="Maximum clip length (seconds) / مدة المقطع",
                )
                head_keep_audio = gr.Checkbox(value=True, label="Keep original audio")
                head_watermark = gr.Checkbox(value=True, label="Add ‘AI HEAD SWAP’ watermark")
                head_consent = gr.Checkbox(
                    value=False,
                    label="I own this media or have permission from everyone depicted.",
                )
                head_run = gr.Button(
                    "Swap Complete Head / تبديل الرأس كاملًا", variant="primary", size="lg"
                )
        head_output = gr.Video(label="Head Swap result / النتيجة", format="mp4")
        head_status = gr.Textbox(label="Status", interactive=False)

    run_button.click(
        fn=swap_video_face,
        inputs=[
            source_input,
            target_input,
            target_face_index,
            clip_length,
            output_resolution,
            quality_mode,
            keep_audio,
            add_watermark,
            consent,
        ],
        outputs=[output_video, status_output],
        api_name="swap_video",
    )

    head_run.click(
        fn=swap_video_head,
        inputs=[
            head_source, head_target, head_clip_length,
            head_keep_audio, head_watermark, head_consent,
        ],
        outputs=[head_output, head_status],
        api_name="swap_head_video",
    )

demo.queue(default_concurrency_limit=1, max_size=8).launch()