Update app.py

app.py

@@ -1,35 +1,28 @@
-import spaces
-import tempfile
-import gradio as gr
-import subprocess
-import os, stat
+import os
 import uuid
-from googletrans import Translator
-import edge_tts
 import asyncio
-import ffmpeg
+import subprocess
 import json
-from scipy.signal import wiener
-import soundfile as sf
-from pydub import AudioSegment
-import numpy as np
-import librosa
 from zipfile import ZipFile
-import shlex
+import gradio as gr
+import ffmpeg
 import cv2
-import torch
-import torchvision
-from tqdm import tqdm
-from numba import jit
+import edge_tts
+from googletrans import Translator
 from huggingface_hub import HfApi
 import moviepy.editor as mp
+import spaces
 
+# Constants and initialization
 HF_TOKEN = os.environ.get("HF_TOKEN")
+REPO_ID = "artificialguybr/video-dubbing"
+MAX_VIDEO_DURATION = 60  # seconds
+
 api = HfApi(token=HF_TOKEN)
-repo_id = "artificialguybr/video-dubbing"
+
+# Extract and set permissions for ffmpeg
 ZipFile("ffmpeg.zip").extractall()
-st = os.stat('ffmpeg')
-os.chmod('ffmpeg', st.st_mode | stat.S_IEXEC)
+os.chmod('ffmpeg', os.stat('ffmpeg').st_mode | os.stat.S_IEXEC)
 
 print("Starting the program...")
 
@@ -50,19 +43,16 @@ def check_for_faces(video_path):
         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:
+        if face_cascade.detectMultiScale(gray, 1.1, 4):
             return True
-
     return False
 
 @spaces.GPU(duration=90)
 def transcribe_audio(file_path):
     print(f"Starting transcription of file: {file_path}")
     temp_audio = None
+    
     if file_path.endswith(('.mp4', '.avi', '.mov', '.flv')):
         print("Video file detected. Extracting audio...")
         try:
@@ -73,10 +63,7 @@ def transcribe_audio(file_path):
         except Exception as e:
             print(f"Error extracting audio from video: {e}")
             raise
-    
-    print(f"Does the file exist? {os.path.exists(file_path)}")
-    print(f"File size: {os.path.getsize(file_path) if os.path.exists(file_path) else 'N/A'} bytes")
-    
+
     output_file = generate_unique_filename(".json")
     command = [
         "insanely-fast-whisper",
@@ -87,37 +74,24 @@ def transcribe_audio(file_path):
         "--timestamp", "chunk",
         "--transcript-path", output_file
     ]
-    print(f"Executing command: {' '.join(command)}")
+    
     try:
         result = subprocess.run(command, check=True, capture_output=True, text=True)
-        print(f"Standard output: {result.stdout}")
-        print(f"Error output: {result.stderr}")
+        print(f"Transcription output: {result.stdout}")
     except subprocess.CalledProcessError as e:
         print(f"Error running insanely-fast-whisper: {e}")
-        print(f"Standard output: {e.stdout}")
-        print(f"Error output: {e.stderr}")
         raise
-    
-    print(f"Reading transcription file: {output_file}")
+
     try:
         with open(output_file, "r") as f:
             transcription = json.load(f)
     except json.JSONDecodeError as e:
         print(f"Error decoding JSON: {e}")
-        print(f"File content: {open(output_file, 'r').read()}")
         raise
+
+    result = transcription.get("text", " ".join([chunk["text"] for chunk in transcription.get("chunks", [])]))
     
-    if "text" in transcription:
-        result = transcription["text"]
-    else:
-        result = " ".join([chunk["text"] for chunk in transcription.get("chunks", [])])
-    
-    print("Transcription completed.")
-    
-    # Cleanup
-    cleanup_files(output_file)
-    if temp_audio:
-        cleanup_files(temp_audio)
+    cleanup_files(output_file, temp_audio)
     
     return result
 
@@ -143,22 +117,16 @@ def process_video(radio, video, target_language, has_closeup_face):
         video_info = ffmpeg.probe(video_path)
         video_duration = float(video_info['streams'][0]['duration'])
 
-        if video_duration > 60:
-            os.remove(video_path)
-            raise ValueError("Video duration exceeds 1 minute. Please upload a shorter video.")
+        if video_duration > MAX_VIDEO_DURATION:
+            cleanup_files(video_path)
+            raise ValueError(f"Video duration exceeds {MAX_VIDEO_DURATION} seconds. Please upload a shorter video.")
 
         ffmpeg.input(video_path).output(f"{run_uuid}_output_audio.wav", acodec='pcm_s24le', ar=48000, map='a').run()
 
-        shell_command = f"ffmpeg -y -i {run_uuid}_output_audio.wav -af lowpass=3000,highpass=100 {run_uuid}_output_audio_final.wav".split(" ")
-        subprocess.run([item for item in shell_command], capture_output=False, text=True, check=True)
+        subprocess.run(f"ffmpeg -y -i {run_uuid}_output_audio.wav -af lowpass=3000,highpass=100 {run_uuid}_output_audio_final.wav", shell=True, check=True)
         
-        print("Attempting to transcribe with Whisper...")
-        try:
-            whisper_text = transcribe_audio(f"{run_uuid}_output_audio_final.wav")
-            print(f"Transcription successful: {whisper_text}")
-        except Exception as e:
-            print(f"Error encountered during transcription: {str(e)}")
-            raise
+        whisper_text = transcribe_audio(f"{run_uuid}_output_audio_final.wav")
+        print(f"Transcription successful: {whisper_text}")
                 
         language_mapping = {
             'English': ('en', 'en-US-EricNeural'),
@@ -189,61 +157,36 @@ def process_video(radio, video, target_language, has_closeup_face):
 
         asyncio.run(text_to_speech(translated_text, voice, f"{run_uuid}_output_synth.wav"))
         
-        pad_top = 0
-        pad_bottom = 15
-        pad_left = 0
-        pad_right = 0
-        rescaleFactor = 1
-
-        video_path_fix = video_path
-
-        if has_closeup_face:
-            has_face = True
-        else:
-            has_face = check_for_faces(video_path)
-
-        if has_closeup_face:
+        if has_closeup_face or check_for_faces(video_path):
             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 {pad_top} {pad_bottom} {pad_left} {pad_right} --resize_factor {rescaleFactor} --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.")
-                    cmd = f"ffmpeg -i {video_path} -i {run_uuid}_output_synth.wav -c:v copy -c:a aac -strict experimental -map 0:v:0 -map 1:a:0 {run_uuid}_output_video.mp4"
-                    subprocess.run(cmd, shell=True)
+                subprocess.run(f"python Wav2Lip/inference.py --checkpoint_path 'Wav2Lip/checkpoints/wav2lip_gan.pth' --face '{video_path}' --audio '{run_uuid}_output_synth.wav' --pads 0 15 0 0 --resize_factor 1 --nosmooth --outfile '{run_uuid}_output_video.mp4'", shell=True, check=True)
+            except subprocess.CalledProcessError:
+                gr.Warning("Wav2lip didn't detect a face. Please try again with the option disabled.")
+                subprocess.run(f"ffmpeg -i {video_path} -i {run_uuid}_output_synth.wav -c:v copy -c:a aac -strict experimental -map 0:v:0 -map 1:a:0 {run_uuid}_output_video.mp4", shell=True)
         else:
-            cmd = f"ffmpeg -i {video_path} -i {run_uuid}_output_synth.wav -c:v copy -c:a aac -strict experimental -map 0:v:0 -map 1:a:0 {run_uuid}_output_video.mp4"
-            subprocess.run(cmd, shell=True)
-
-        if not os.path.exists(f"{run_uuid}_output_video.mp4"):
-            raise FileNotFoundError(f"Error: {run_uuid}_output_video.mp4 was not generated.")
+            subprocess.run(f"ffmpeg -i {video_path} -i {run_uuid}_output_synth.wav -c:v copy -c:a aac -strict experimental -map 0:v:0 -map 1:a:0 {run_uuid}_output_video.mp4", shell=True)
 
         output_video_path = f"{run_uuid}_output_video.mp4"
+        if not os.path.exists(output_video_path):
+            raise FileNotFoundError(f"Error: {output_video_path} was not generated.")
 
-        files_to_delete = [
+        cleanup_files(
             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, ""  # Retorna o caminho do vídeo e uma string vazia para a mensagem de erro
+        return output_video_path, ""
 
     except Exception as e:
         print(f"Error in process_video: {str(e)}")
-        return None, f"Error: {str(e)}"  # Retorna None para o vídeo e a mensagem de erro
+        return None, f"Error: {str(e)}"
 
 def swap(radio):
-    if(radio == "Upload"):
-        return gr.update(source="upload")
-    else:
-        return gr.update(source="webcam")
-        
+    return gr.update(source="upload" if radio == "Upload" else "webcam")
+
+# Gradio interface setup
 video = gr.Video()
 radio = gr.Radio(["Upload", "Record"], value="Upload", show_label=False)
 iface = gr.Interface(
@@ -251,11 +194,8 @@ iface = gr.Interface(
     inputs=[
         radio,
         video,
-        gr.Dropdown(choices=["English", "Spanish", "French", "German", "Italian", "Portuguese", "Polish", "Turkish", "Russian", "Dutch", "Czech", "Arabic", "Chinese (Simplified)", "Japanese", "Korean", "Hindi", "Swedish", "Danish", "Finnish", "Greek"], 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.")
+        gr.Dropdown(choices=list(language_mapping.keys()), 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(label="Processed Video"),