app.py

import spaces
import tempfile
import gradio as gr
import subprocess
import os, stat
import uuid
from googletrans import Translator
from TTS.api import TTS
from faster_whisper import WhisperModel
import soundfile as sf
import numpy as np
import cv2
from huggingface_hub import HfApi
import shlex

HF_TOKEN = os.environ.get("HF_TOKEN")
os.environ["COQUI_TOS_AGREED"] = "1"
api = HfApi(token=HF_TOKEN)
repo_id = "artificialguybr/video-dubbing"

# Whisper
model_size = "small"
model = WhisperModel(model_size, device="cpu", compute_type="int8")

def check_for_faces(video_path):
    face_cascade = cv2.CascadeClassifier(cv2.data.haarcascades + 'haarcascade_frontalface_default.xml')
    cap = cv2.VideoCapture(video_path)

    while True:
        ret, frame = cap.read()
        if not ret:
            break

        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
        faces = face_cascade.detectMultiScale(gray, 1.1, 4)

        if len(faces) > 0:
            return True

    return False

@spaces.GPU
def process_video(radio, video, target_language, has_closeup_face):
    if target_language is None:
        return gr.Error("Please select a Target Language for Dubbing.")
        
    run_uuid = uuid.uuid4().hex[:6]
    output_filename = f"{run_uuid}_resized_video.mp4"
    
    # Use subprocess for ffmpeg operations
    subprocess.run(["ffmpeg", "-i", video, "-vf", "scale=-2:720", output_filename])

    video_path = output_filename
    
    if not os.path.exists(video_path):
        return f"Error: {video_path} does not exist."

    # Check video duration
    video_info = subprocess.check_output(["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=noprint_wrappers=1:nokey=1", video_path])
    video_duration = float(video_info)

    if video_duration > 60:
        os.remove(video_path)
        return gr.Error("Video duration exceeds 1 minute. Please upload a shorter video.")

    subprocess.run(["ffmpeg", "-i", video_path, "-acodec", "pcm_s24le", "-ar", "48000", "-map", "a", f"{run_uuid}_output_audio.wav"])

    subprocess.run(["ffmpeg", "-y", "-i", f"{run_uuid}_output_audio.wav", "-af", "lowpass=3000,highpass=100", f"{run_uuid}_output_audio_final.wav"])
    
    print("Attempting to transcribe with Whisper...")
    try:
        segments, info = model.transcribe(f"{run_uuid}_output_audio_final.wav", beam_size=5)
        whisper_text = " ".join(segment.text for segment in segments)
        whisper_language = info.language
        print(f"Transcription successful: {whisper_text}")
    except RuntimeError as e:
        print(f"RuntimeError encountered: {str(e)}")
        if "CUDA failed with error device-side assert triggered" in str(e):
            gr.Warning("Error. Space need to restart. Please retry in a minute")
            api.restart_space(repo_id=repo_id)
            
    language_mapping = {'English': 'en', 'Spanish': 'es', 'French': 'fr', 'German': 'de', 'Italian': 'it', 'Portuguese': 'pt', 'Polish': 'pl', 'Turkish': 'tr', 'Russian': 'ru', 'Dutch': 'nl', 'Czech': 'cs', 'Arabic': 'ar', 'Chinese (Simplified)': 'zh-cn'}
    target_language_code = language_mapping[target_language]
    translator = Translator()
    translated_text = translator.translate(whisper_text, src=whisper_language, dest=target_language_code).text
    print(translated_text)

    tts = TTS("tts_models/multilingual/multi-dataset/xtts_v2")
    tts.tts_to_file(translated_text, speaker_wav=f"{run_uuid}_output_audio_final.wav", file_path=f"{run_uuid}_output_synth.wav", language=target_language_code)
    
    if has_closeup_face:
        try:
            cmd = f"python Wav2Lip/inference.py --checkpoint_path 'Wav2Lip/checkpoints/wav2lip_gan.pth' --face {shlex.quote(video_path)} --audio '{run_uuid}_output_synth.wav' --pads 0 15 0 0 --resize_factor 1 --nosmooth --outfile '{run_uuid}_output_video.mp4'"
            subprocess.run(cmd, shell=True, check=True)
        except subprocess.CalledProcessError as e:
            if "Face not detected! Ensure the video contains a face in all the frames." in str(e.stderr):
                gr.Warning("Wav2lip didn't detect a face. Please try again with the option disabled.")
                subprocess.run(["ffmpeg", "-i", video_path, "-i", f"{run_uuid}_output_synth.wav", "-c:v", "copy", "-c:a", "aac", "-strict", "experimental", "-map", "0:v:0", "-map", "1:a:0", f"{run_uuid}_output_video.mp4"])
    else:
        subprocess.run(["ffmpeg", "-i", video_path, "-i", f"{run_uuid}_output_synth.wav", "-c:v", "copy", "-c:a", "aac", "-strict", "experimental", "-map", "0:v:0", "-map", "1:a:0", f"{run_uuid}_output_video.mp4"])

    if not os.path.exists(f"{run_uuid}_output_video.mp4"):
        raise FileNotFoundError(f"Error: {run_uuid}_output_video.mp4 was not generated.")

    output_video_path = f"{run_uuid}_output_video.mp4"

    # Cleanup
    files_to_delete = [
        f"{run_uuid}_resized_video.mp4",
        f"{run_uuid}_output_audio.wav",
        f"{run_uuid}_output_audio_final.wav",
        f"{run_uuid}_output_synth.wav"
    ]
    for file in files_to_delete:
        try:
            os.remove(file)
        except FileNotFoundError:
            print(f"File {file} not found for deletion.")

    return output_video_path

def swap(radio):
    if(radio == "Upload"):
        return gr.update(source="upload")
    else:
        return gr.update(source="webcam")
        
video = gr.Video()
radio = gr.Radio(["Upload", "Record"], value="Upload", show_label=False)
iface = gr.Interface(
    fn=process_video,
    inputs=[
        radio,
        video,
        gr.Dropdown(choices=["English", "Spanish", "French", "German", "Italian", "Portuguese", "Polish", "Turkish", "Russian", "Dutch", "Czech", "Arabic", "Chinese (Simplified)"], label="Target Language for Dubbing", value="Spanish"),
        gr.Checkbox(
                    label="Video has a close-up face. Use Wav2lip.",
                    value=False,
                    info="Say if video have close-up face. For Wav2lip. Will not work if checked wrongly.")
    ],
    outputs=gr.Video(),
    live=False,
    title="AI Video Dubbing",
    description="""This tool was developed by [@artificialguybr](https://twitter.com/artificialguybr) using entirely open-source tools. Special thanks to Hugging Face for the GPU support. Thanks [@yeswondwer](https://twitter.com/@yeswondwerr) for original code. Test the [Video Transcription and Translate](https://huggingface.co/spaces/artificialguybr/VIDEO-TRANSLATION-TRANSCRIPTION) space!""",
    allow_flagging=False
)

with gr.Blocks() as demo:
    iface.render()
    radio.change(swap, inputs=[radio], outputs=video)
    gr.Markdown("""
    **Note:**
    - Video limit is 1 minute. It will dubbing all people using just one voice.
    - Generation may take up to 5 minutes.
    - By using this demo you agree to the terms of the Coqui Public Model License at https://coqui.ai/cpml
    - The tool uses open-source models for all models. It's an alpha version.
    - Quality can be improved but would require more processing time per video. For scalability and hardware limitations, speed was chosen, not just quality.
    - If you need more than 1 minute, duplicate the Space and change the limit on app.py.
    - If you incorrectly mark the 'Video has a close-up face' checkbox, the dubbing may not work as expected.
    """)

demo.queue(concurrency_count=1, max_size=15)
demo.launch()