NeuralFalcon
commited on
Create app.py
Browse files
app.py
ADDED
@@ -0,0 +1,328 @@
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
1 |
+
# app.py
|
2 |
+
from utils import language_dict
|
3 |
+
import math
|
4 |
+
import torch
|
5 |
+
import gc
|
6 |
+
import time
|
7 |
+
import subprocess
|
8 |
+
from faster_whisper import WhisperModel
|
9 |
+
import os
|
10 |
+
import mimetypes
|
11 |
+
import shutil
|
12 |
+
import re
|
13 |
+
import uuid
|
14 |
+
from pydub import AudioSegment
|
15 |
+
import torch
|
16 |
+
|
17 |
+
|
18 |
+
|
19 |
+
def get_language_name(lang_code):
|
20 |
+
global language_dict
|
21 |
+
# Iterate through the language dictionary
|
22 |
+
for language, details in language_dict.items():
|
23 |
+
# Check if the language code matches
|
24 |
+
if details["lang_code"] == lang_code:
|
25 |
+
return language # Return the language name
|
26 |
+
return None
|
27 |
+
|
28 |
+
def clean_file_name(file_path):
|
29 |
+
# Get the base file name and extension
|
30 |
+
file_name = os.path.basename(file_path)
|
31 |
+
file_name, file_extension = os.path.splitext(file_name)
|
32 |
+
|
33 |
+
# Replace non-alphanumeric characters with an underscore
|
34 |
+
cleaned = re.sub(r'[^a-zA-Z\d]+', '_', file_name)
|
35 |
+
|
36 |
+
# Remove any multiple underscores
|
37 |
+
clean_file_name = re.sub(r'_+', '_', cleaned).strip('_')
|
38 |
+
|
39 |
+
# Generate a random UUID for uniqueness
|
40 |
+
random_uuid = uuid.uuid4().hex[:6]
|
41 |
+
|
42 |
+
# Combine cleaned file name with the original extension
|
43 |
+
clean_file_path = os.path.join(os.path.dirname(file_path), clean_file_name + f"_{random_uuid}" + file_extension)
|
44 |
+
|
45 |
+
return clean_file_path
|
46 |
+
|
47 |
+
def get_audio_file(uploaded_file):
|
48 |
+
global base_path
|
49 |
+
# ,device
|
50 |
+
device = "cuda" if torch.cuda.is_available() else "cpu"
|
51 |
+
# Detect the file type (audio/video)
|
52 |
+
mime_type, _ = mimetypes.guess_type(uploaded_file)
|
53 |
+
# Create the folder path to store audio files
|
54 |
+
audio_folder = f"{base_path}/subtitle_audio"
|
55 |
+
os.makedirs(audio_folder, exist_ok=True)
|
56 |
+
# Initialize variable for the audio file path
|
57 |
+
audio_file_path = ""
|
58 |
+
if mime_type and mime_type.startswith('audio'):
|
59 |
+
# If it's an audio file, save it as is
|
60 |
+
audio_file_path = os.path.join(audio_folder, os.path.basename(uploaded_file))
|
61 |
+
audio_file_path=clean_file_name(audio_file_path)
|
62 |
+
shutil.copy(uploaded_file, audio_file_path) # Move file to audio folder
|
63 |
+
|
64 |
+
elif mime_type and mime_type.startswith('video'):
|
65 |
+
# If it's a video file, extract the audio
|
66 |
+
audio_file_name = os.path.splitext(os.path.basename(uploaded_file))[0] + ".mp3"
|
67 |
+
audio_file_path = os.path.join(audio_folder, audio_file_name)
|
68 |
+
audio_file_path=clean_file_name(audio_file_path)
|
69 |
+
|
70 |
+
# Extract the file extension from the uploaded file
|
71 |
+
file_extension = os.path.splitext(uploaded_file)[1] # Includes the dot, e.g., '.mp4'
|
72 |
+
|
73 |
+
# Generate a random UUID and create a new file name with the same extension
|
74 |
+
random_uuid = uuid.uuid4().hex[:6]
|
75 |
+
new_file_name = random_uuid + file_extension
|
76 |
+
|
77 |
+
# Set the new file path in the subtitle_audio folder
|
78 |
+
new_file_path = os.path.join(audio_folder, new_file_name)
|
79 |
+
|
80 |
+
# Copy the original video file to the new location with the new name
|
81 |
+
shutil.copy(uploaded_file, new_file_path)
|
82 |
+
if device=="cuda":
|
83 |
+
command = f"ffmpeg -hwaccel cuda -i {new_file_path} {audio_file_path} -y"
|
84 |
+
else:
|
85 |
+
command = f"ffmpeg -i {new_file_path} {audio_file_path} -y"
|
86 |
+
|
87 |
+
subprocess.run(command, shell=True)
|
88 |
+
if os.path.exists(new_file_path):
|
89 |
+
os.remove(new_file_path)
|
90 |
+
# Return the saved audio file path
|
91 |
+
audio = AudioSegment.from_file(audio_file_path)
|
92 |
+
# Get the duration in seconds
|
93 |
+
duration_seconds = len(audio) / 1000.0 # pydub measures duration in milliseconds
|
94 |
+
return audio_file_path,duration_seconds
|
95 |
+
|
96 |
+
def format_segments(segments):
|
97 |
+
saved_segments = list(segments)
|
98 |
+
sentence_timestamp = []
|
99 |
+
words_timestamp = []
|
100 |
+
speech_to_text = ""
|
101 |
+
|
102 |
+
for i in saved_segments:
|
103 |
+
temp_sentence_timestamp = {}
|
104 |
+
# Store sentence information in sentence_timestamp
|
105 |
+
text = i.text.strip()
|
106 |
+
sentence_id = len(sentence_timestamp) # Get the current index for the new entry
|
107 |
+
sentence_timestamp.append({
|
108 |
+
"id": sentence_id, # Use the index as the id
|
109 |
+
"text": text,
|
110 |
+
"start": i.start,
|
111 |
+
"end": i.end,
|
112 |
+
"words": [] # Initialize words as an empty list within the sentence
|
113 |
+
})
|
114 |
+
speech_to_text += text + " "
|
115 |
+
|
116 |
+
# Process each word in the sentence
|
117 |
+
for word in i.words:
|
118 |
+
word_data = {
|
119 |
+
"word": word.word.strip(),
|
120 |
+
"start": word.start,
|
121 |
+
"end": word.end
|
122 |
+
}
|
123 |
+
|
124 |
+
# Append word timestamps to the sentence's word list
|
125 |
+
sentence_timestamp[sentence_id]["words"].append(word_data)
|
126 |
+
|
127 |
+
# Optionally, add the word data to the global words_timestamp list
|
128 |
+
words_timestamp.append(word_data)
|
129 |
+
|
130 |
+
return sentence_timestamp, words_timestamp, speech_to_text
|
131 |
+
|
132 |
+
def combine_word_segments(words_timestamp, max_words_per_subtitle=8, min_silence_between_words=0.5):
|
133 |
+
before_translate = {}
|
134 |
+
id = 1
|
135 |
+
text = ""
|
136 |
+
start = None
|
137 |
+
end = None
|
138 |
+
word_count = 0
|
139 |
+
last_end_time = None
|
140 |
+
|
141 |
+
for i in words_timestamp:
|
142 |
+
try:
|
143 |
+
word = i['word']
|
144 |
+
word_start = i['start']
|
145 |
+
word_end = i['end']
|
146 |
+
|
147 |
+
# Check for sentence-ending punctuation
|
148 |
+
is_end_of_sentence = word.endswith(('.', '?', '!'))
|
149 |
+
|
150 |
+
# Check for conditions to create a new subtitle
|
151 |
+
if ((last_end_time is not None and word_start - last_end_time > min_silence_between_words)
|
152 |
+
or word_count >= max_words_per_subtitle
|
153 |
+
or is_end_of_sentence):
|
154 |
+
|
155 |
+
# Store the previous subtitle if there's any
|
156 |
+
if text:
|
157 |
+
before_translate[id] = {
|
158 |
+
"text": text,
|
159 |
+
"start": start,
|
160 |
+
"end": end
|
161 |
+
}
|
162 |
+
id += 1
|
163 |
+
|
164 |
+
# Reset for the new subtitle segment
|
165 |
+
text = word
|
166 |
+
start = word_start # Set the start time for the new subtitle
|
167 |
+
word_count = 1
|
168 |
+
else:
|
169 |
+
if word_count == 0: # First word in the subtitle
|
170 |
+
start = word_start # Ensure the start time is set
|
171 |
+
text += " " + word
|
172 |
+
word_count += 1
|
173 |
+
|
174 |
+
end = word_end # Update the end timestamp
|
175 |
+
last_end_time = word_end # Update the last end timestamp
|
176 |
+
|
177 |
+
except KeyError as e:
|
178 |
+
print(f"KeyError: {e} - Skipping word")
|
179 |
+
pass
|
180 |
+
|
181 |
+
# After the loop, make sure to add the last subtitle segment
|
182 |
+
if text:
|
183 |
+
before_translate[id] = {
|
184 |
+
"text": text,
|
185 |
+
"start": start,
|
186 |
+
"end": end
|
187 |
+
}
|
188 |
+
|
189 |
+
return before_translate
|
190 |
+
|
191 |
+
|
192 |
+
def convert_time_to_srt_format(seconds):
|
193 |
+
""" Convert seconds to SRT time format (HH:MM:SS,ms) """
|
194 |
+
hours = int(seconds // 3600)
|
195 |
+
minutes = int((seconds % 3600) // 60)
|
196 |
+
secs = int(seconds % 60)
|
197 |
+
milliseconds = int((seconds - int(seconds)) * 1000)
|
198 |
+
return f"{hours:02}:{minutes:02}:{secs:02},{milliseconds:03}"
|
199 |
+
def write_subtitles_to_file(subtitles, filename="subtitles.srt"):
|
200 |
+
|
201 |
+
# Open the file with UTF-8 encoding
|
202 |
+
with open(filename, 'w', encoding='utf-8') as f:
|
203 |
+
for id, entry in subtitles.items():
|
204 |
+
# Write the subtitle index
|
205 |
+
f.write(f"{id}\n")
|
206 |
+
if entry['start'] is None or entry['end'] is None:
|
207 |
+
print(id)
|
208 |
+
# Write the start and end time in SRT format
|
209 |
+
start_time = convert_time_to_srt_format(entry['start'])
|
210 |
+
end_time = convert_time_to_srt_format(entry['end'])
|
211 |
+
f.write(f"{start_time} --> {end_time}\n")
|
212 |
+
|
213 |
+
# Write the text and speaker information
|
214 |
+
f.write(f"{entry['text']}\n\n")
|
215 |
+
|
216 |
+
def word_level_srt(words_timestamp, srt_path="world_level_subtitle.srt"):
|
217 |
+
with open(srt_path, 'w', encoding='utf-8') as srt_file:
|
218 |
+
for i, word_info in enumerate(words_timestamp, start=1):
|
219 |
+
start_time = convert_time_to_srt_format(word_info['start'])
|
220 |
+
end_time = convert_time_to_srt_format(word_info['end'])
|
221 |
+
srt_file.write(f"{i}\n{start_time} --> {end_time}\n{word_info['word']}\n\n")
|
222 |
+
|
223 |
+
def generate_srt_from_sentences(sentence_timestamp, srt_path="default_subtitle.srt"):
|
224 |
+
with open(srt_path, 'w', encoding='utf-8') as srt_file:
|
225 |
+
for index, sentence in enumerate(sentence_timestamp):
|
226 |
+
start_time = convert_time_to_srt_format(sentence['start'])
|
227 |
+
end_time = convert_time_to_srt_format(sentence['end'])
|
228 |
+
srt_file.write(f"{index + 1}\n{start_time} --> {end_time}\n{sentence['text']}\n\n")
|
229 |
+
|
230 |
+
|
231 |
+
|
232 |
+
def whisper_subtitle(uploaded_file,Source_Language,max_words_per_subtitle=8):
|
233 |
+
global language_dict,base_path
|
234 |
+
#Load model
|
235 |
+
if torch.cuda.is_available():
|
236 |
+
# If CUDA is available, use GPU with float16 precision
|
237 |
+
device = "cuda"
|
238 |
+
compute_type = "float16"
|
239 |
+
# compute_type="int8_float16"
|
240 |
+
else:
|
241 |
+
# If CUDA is not available, use CPU with int8 precision
|
242 |
+
device = "cpu"
|
243 |
+
compute_type = "int8"
|
244 |
+
faster_whisper_model = WhisperModel("deepdml/faster-whisper-large-v3-turbo-ct2",device=device, compute_type=compute_type)
|
245 |
+
audio_path,audio_duration=get_audio_file(uploaded_file)
|
246 |
+
|
247 |
+
if Source_Language=="Automatic":
|
248 |
+
segments,d = faster_whisper_model.transcribe(audio_path, word_timestamps=True)
|
249 |
+
lang_code=d.language
|
250 |
+
src_lang=get_language_name(lang_code)
|
251 |
+
else:
|
252 |
+
lang=language_dict[Source_Language]['lang_code']
|
253 |
+
segments,d = faster_whisper_model.transcribe(audio_path, word_timestamps=True,language=lang)
|
254 |
+
src_lang=Source_Language
|
255 |
+
if os.path.exists(audio_path):
|
256 |
+
os.remove(audio_path)
|
257 |
+
sentence_timestamp,words_timestamp,text=format_segments(segments)
|
258 |
+
del faster_whisper_model
|
259 |
+
gc.collect()
|
260 |
+
torch.cuda.empty_cache()
|
261 |
+
|
262 |
+
word_segments=combine_word_segments(words_timestamp, max_words_per_subtitle=max_words_per_subtitle, min_silence_between_words=0.5)
|
263 |
+
|
264 |
+
#setup srt file names
|
265 |
+
base_name = os.path.basename(uploaded_file).rsplit('.', 1)[0][:30]
|
266 |
+
save_name = f"{base_path}/generated_subtitle/{base_name}_{src_lang}.srt"
|
267 |
+
original_srt_name=clean_file_name(save_name)
|
268 |
+
original_txt_name=original_srt_name.replace(".srt",".txt")
|
269 |
+
word_level_srt_name=original_srt_name.replace(".srt","_word_level.srt")
|
270 |
+
default_srt_name=original_srt_name.replace(".srt","_default.srt")
|
271 |
+
|
272 |
+
generate_srt_from_sentences(sentence_timestamp, srt_path=default_srt_name)
|
273 |
+
word_level_srt(words_timestamp, srt_path=word_level_srt_name)
|
274 |
+
write_subtitles_to_file(word_segments, filename=original_srt_name)
|
275 |
+
with open(original_txt_name, 'w', encoding='utf-8') as f1:
|
276 |
+
f1.write(text)
|
277 |
+
return default_srt_name,original_srt_name,word_level_srt_name,original_txt_name
|
278 |
+
|
279 |
+
#@title Using Gradio Interface
|
280 |
+
def subtitle_maker(Audio_or_Video_File,Source_Language,max_words_per_subtitle):
|
281 |
+
try:
|
282 |
+
default_srt_path,customize_srt_path,word_level_srt_path,text_path=whisper_subtitle(Audio_or_Video_File,Source_Language,max_words_per_subtitle=max_words_per_subtitle)
|
283 |
+
except:
|
284 |
+
default_srt_path,customize_srt_path,word_level_srt_path,text_path=None,None,None,None
|
285 |
+
return default_srt_path,customize_srt_path,word_level_srt_path,text_path
|
286 |
+
|
287 |
+
|
288 |
+
|
289 |
+
|
290 |
+
|
291 |
+
import gradio as gr
|
292 |
+
import click
|
293 |
+
|
294 |
+
base_path="."
|
295 |
+
if not os.path.exists(f"{base_path}/generated_subtitle"):
|
296 |
+
os.makedirs(f"{base_path}/generated_subtitle", exist_ok=True)
|
297 |
+
|
298 |
+
source_lang_list = ['Automatic']
|
299 |
+
available_language=language_dict.keys()
|
300 |
+
source_lang_list.extend(available_language)
|
301 |
+
|
302 |
+
|
303 |
+
@click.command()
|
304 |
+
@click.option("--debug", is_flag=True, default=False, help="Enable debug mode.")
|
305 |
+
@click.option("--share", is_flag=True, default=False, help="Enable sharing of the interface.")
|
306 |
+
def main(debug, share):
|
307 |
+
# Define Gradio inputs and outputs
|
308 |
+
gradio_inputs = [
|
309 |
+
gr.File(label="Upload Audio or Video File"),
|
310 |
+
gr.Dropdown(label="Language", choices=source_lang_list, value="Automatic"),
|
311 |
+
gr.Number(label="Max Word Per Subtitle Segment", value=8)
|
312 |
+
]
|
313 |
+
|
314 |
+
gradio_outputs = [
|
315 |
+
gr.File(label="Default SRT File", show_label=True),
|
316 |
+
gr.File(label="Customize SRT File", show_label=True),
|
317 |
+
gr.File(label="Word Level SRT File", show_label=True),
|
318 |
+
gr.File(label="Text File", show_label=True)
|
319 |
+
]
|
320 |
+
|
321 |
+
# Create Gradio interface
|
322 |
+
demo = gr.Interface(fn=subtitle_maker, inputs=gradio_inputs, outputs=gradio_outputs, title="Whisper-Large-V3-Turbo-Ct2 Subtitle Maker")
|
323 |
+
|
324 |
+
# Launch Gradio with command-line options
|
325 |
+
demo.launch(debug=debug, share=share)
|
326 |
+
|
327 |
+
if __name__ == "__main__":
|
328 |
+
main()
|