-
Notifications
You must be signed in to change notification settings - Fork 2
/
Copy pathdub.py
163 lines (145 loc) · 6.68 KB
/
dub.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
import gradio as gr
import os
import time
import json
from typing import Optional, Dict
from elevenlabs import ElevenLabs
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
# Initialize ElevenLabs client
eleven = ElevenLabs(api_key=os.getenv("ELEVENLABS_API_KEY"))
# Define language options with codes
languages: Dict[str, str] = {
"English": "en", "Hindi": "hi", "Portuguese": "pt", "Chinese": "zh", "Spanish": "es",
"French": "fr", "German": "de", "Japanese": "ja", "Arabic": "ar", "Russian": "ru",
"Korean": "ko", "Indonesian": "id", "Italian": "it", "Dutch": "nl", "Turkish": "tr",
"Polish": "pl", "Swedish": "sv", "Filipino": "fil", "Malay": "ms", "Romanian": "ro",
"Ukrainian": "uk", "Greek": "el", "Czech": "cs", "Danish": "da", "Finnish": "fi",
"Bulgarian": "bg", "Croatian": "hr", "Slovak": "sk", "Tamil": "ta"
}
def wait_for_dubbing_completion(dubbing_id: str, progress: Optional[gr.Progress] = None) -> bool:
MAX_ATTEMPTS = 360 # Increased to allow for up to 1 hour of processing
CHECK_INTERVAL = 10 # In seconds
for attempt in range(MAX_ATTEMPTS):
try:
metadata = eleven.dubbing.get_dubbing_project_metadata(dubbing_id)
if metadata.status == "dubbed":
if progress:
progress(1, desc="Dubbing completed successfully.")
return True
elif metadata.status == "dubbing":
if progress:
progress((attempt + 1) / MAX_ATTEMPTS, desc=f"Dubbing in progress... Time elapsed: {attempt * CHECK_INTERVAL} seconds")
time.sleep(CHECK_INTERVAL)
else:
if progress:
progress(1, desc=f"Dubbing failed: {metadata.error_message}")
return False
except Exception as e:
print(f"Error in wait_for_dubbing_completion: {str(e)}")
if progress:
progress(1, desc=f"Error checking dubbing status: {str(e)}")
return False
if progress:
progress(1, desc="Dubbing timed out after 1 hour")
return False
def download_dubbed_file(dubbing_id: str, language_code: str, output_dir: str, progress: Optional[gr.Progress] = None) -> Optional[str]:
os.makedirs(output_dir, exist_ok=True)
file_path = os.path.join(output_dir, f"{dubbing_id}_{language_code}.mp4")
try:
with open(file_path, "wb") as file:
for chunk in eleven.dubbing.get_dubbed_file(dubbing_id, language_code):
file.write(chunk)
if progress:
progress(1, desc=f"File downloaded successfully: {file_path}")
return file_path
except Exception as e:
print(f"Error in download_dubbed_file: {str(e)}")
if progress:
progress(1, desc=f"Error downloading dubbed file: {str(e)}")
return None
def create_dub(input_type: str, input_data, source_language: str, target_language: str, output_dir: str, progress: Optional[gr.Progress] = None) -> str:
try:
if progress:
progress(0.1, desc="Initiating dubbing process")
if input_type == "file":
if not input_data:
return "No file uploaded."
file_path = input_data.name
with open(file_path, "rb") as audio_file:
response = eleven.dubbing.dub_a_video_or_an_audio_file(
file=(os.path.basename(file_path), audio_file, "video/mp4"),
target_lang=languages[target_language],
source_lang=languages[source_language],
)
else: # URL
response = eleven.dubbing.dub_a_video_or_an_audio_file(
source_url=input_data,
target_lang=languages[target_language],
source_lang=languages[source_language],
)
dubbing_id = response.dubbing_id
if progress:
progress(0.2, desc="Dubbing initiated, waiting for completion")
if wait_for_dubbing_completion(dubbing_id, progress):
if progress:
progress(0.9, desc="Dubbing completed, downloading file")
output_file_path = download_dubbed_file(dubbing_id, languages[target_language], output_dir, progress)
if output_file_path:
return f"Dubbing was successful! File saved at: {output_file_path}"
else:
return "Dubbing completed but file download failed."
else:
return "Dubbing failed or timed out."
except Exception as e:
print(f"Error in create_dub: {str(e)}")
error_message = str(e)
if hasattr(e, 'status_code') and hasattr(e, 'body'):
try:
error_body = json.loads(e.body)
if 'detail' in error_body:
error_message = f"Error {e.status_code}: {error_body['detail']['message']}"
except json.JSONDecodeError:
error_message = f"Error {e.status_code}: {e.body}"
return f"An error occurred: {error_message}"
# Create Gradio interface
with gr.Blocks() as iface:
gr.Markdown("# ElevenLabs Video Dubbing Interface")
with gr.Tab("File Upload"):
file_input = gr.File(label="Upload Video File")
file_source_lang = gr.Dropdown(list(languages.keys()), label="Source Language", value="English")
file_target_lang = gr.Dropdown(list(languages.keys()), label="Target Language", value="Spanish")
file_output_dir = gr.Textbox(label="Output Directory", value="output", placeholder="Enter the output directory path")
file_submit = gr.Button("Dub Video File")
file_output = gr.Textbox(label="Result")
with gr.Tab("URL Input"):
url_input = gr.Textbox(label="Enter Video URL (YouTube, TikTok, Twitter, or Vimeo)")
url_source_lang = gr.Dropdown(list(languages.keys()), label="Source Language", value="English")
url_target_lang = gr.Dropdown(list(languages.keys()), label="Target Language", value="Spanish")
url_output_dir = gr.Textbox(label="Output Directory", value="output", placeholder="Enter the output directory path")
url_submit = gr.Button("Dub Video from URL")
url_output = gr.Textbox(label="Result")
file_submit.click(
create_dub,
inputs=[
gr.Textbox(value="file", visible=False),
file_input,
file_source_lang,
file_target_lang,
file_output_dir
],
outputs=file_output
)
url_submit.click(
create_dub,
inputs=[
gr.Textbox(value="url", visible=False),
url_input,
url_source_lang,
url_target_lang,
url_output_dir
],
outputs=url_output
)
iface.launch()