Spaces:
Runtime error
Runtime error
| # !pip install youtube-dl | |
| from __future__ import unicode_literals | |
| import youtube_dl | |
| from pydub import AudioSegment | |
| from pyannote.audio import Pipeline | |
| import re | |
| import webvtt | |
| import whisper | |
| import os | |
| from pydub.utils import which | |
| import ffmpeg | |
| import webvtt | |
| import pprint | |
| from urllib.error import HTTPError | |
| import subprocess | |
| import gradio as gr | |
| import traceback | |
| pipeline = Pipeline.from_pretrained("pyannote/speaker-diarization", use_auth_token="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") | |
| def Transcribe(audio="temp_audio.wav"): | |
| def millisec(timeStr): | |
| spl = timeStr.split(":") | |
| s = (int)((int(spl[0]) * 60 * 60 + int(spl[1]) * 60 + float(spl[2]) )* 1000) | |
| return s | |
| def preprocess(audio): | |
| t1 = 0 * 1000 | |
| t2 = 20 * 60 * 1000 | |
| newAudio = AudioSegment.from_wav(audio) | |
| a = newAudio[t1:t2] | |
| spacermilli = 2000 | |
| spacer = AudioSegment.silent(duration=spacermilli) | |
| newAudio = spacer.append(a, crossfade=0) | |
| newAudio.export(audio, format="wav") | |
| return spacermilli, spacer | |
| def diarization(audio): | |
| as_audio = AudioSegment.from_wav(audio) | |
| DEMO_FILE = {'uri': 'blabal', 'audio': audio} | |
| dz = pipeline(DEMO_FILE) | |
| with open(f"diarization_{audio}.txt", "w") as text_file: | |
| text_file.write(str(dz)) | |
| dz = open(f"diarization_{audio}.txt").read().splitlines() | |
| dzList = [] | |
| for l in dz: | |
| start, end = tuple(re.findall('[0-9]+:[0-9]+:[0-9]+\.[0-9]+', string=l)) | |
| start = millisec(start) | |
| end = millisec(end) | |
| lex = re.findall('(SPEAKER_[0-9][0-9])', string=l)[0] | |
| dzList.append([start, end, lex]) | |
| sounds = spacer | |
| segments = [] | |
| dz = open(f"diarization_{audio}.txt").read().splitlines() | |
| for l in dz: | |
| start, end = tuple(re.findall('[0-9]+:[0-9]+:[0-9]+\.[0-9]+', string=l)) | |
| start = millisec(start) | |
| end = millisec(end) | |
| segments.append(len(sounds)) | |
| sounds = sounds.append(as_audio[start:end], crossfade=0) | |
| sounds = sounds.append(spacer, crossfade=0) | |
| sounds.export(f"dz_{audio}.wav", format="wav") | |
| return f"dz_{audio}.wav", dzList, segments | |
| def transcribe(dz_audio): | |
| model = whisper.load_model("base") | |
| result = model.transcribe(dz_audio) | |
| # for _ in result['segments']: | |
| # print(_['start'], _['end'], _['text']) | |
| captions = [[((caption["start"]*1000)), ((caption["end"]*1000)), caption["text"]] for caption in result['segments']] | |
| conversation = [] | |
| for i in range(len(segments)): | |
| idx = 0 | |
| for idx in range(len(captions)): | |
| if captions[idx][0] >= (segments[i] - spacermilli): | |
| break; | |
| while (idx < (len(captions))) and ((i == len(segments) - 1) or (captions[idx][1] < segments[i+1])): | |
| c = captions[idx] | |
| start = dzList[i][0] + (c[0] -segments[i]) | |
| if start < 0: | |
| start = 0 | |
| idx += 1 | |
| if not len(conversation): | |
| conversation.append([dzList[i][2], c[2]]) | |
| elif conversation[-1][0] == dzList[i][2]: | |
| conversation[-1][1] += c[2] | |
| else: | |
| conversation.append([dzList[i][2], c[2]]) | |
| #print(f"[{dzList[i][2]}] {c[2]}") | |
| return ("".join([f"{speaker} --> {text}\n" for speaker, text in conversation])) | |
| spacermilli, spacer = preprocess(audio) | |
| dz_audio, dzList, segments = diarization(audio) | |
| t_text = transcribe(dz_audio) | |
| try: | |
| os.remove("temp_audio.wav") | |
| except OSError: | |
| pass | |
| try: | |
| os.remove("dz_temp_audio.wav") | |
| except OSError: | |
| pass | |
| try: | |
| os.remove(f"diarization_{audio}.txt") | |
| except OSError: | |
| pass | |
| return t_text | |
| # subprocess.call(['ffmpeg', '-i', 'audio.mp3', | |
| # 'audio.wav']) | |
| def AudioTranscribe(audio, retries=5): | |
| if retries: | |
| try: | |
| subprocess.call(['ffmpeg', '-i', audio,'temp_audio.wav']) | |
| except Exception as ex: | |
| traceback.print_exc() | |
| return AudioTranscribe(audio, retries-1) | |
| if not (os.path.exist("temp_audio.wav")): | |
| return AudioTranscribe(audio, retries-1) | |
| return Transcribe() | |
| else: | |
| raise gr.Error("There is some issue ith Audio Transcriber. Please try again later!") | |
| def VideoTranscribe(video): | |
| command = f"ffmpeg -i {video} -ab 160k -ac 2 -ar 44100 -vn temp_audio.wav" | |
| subprocess.call(command, shell=True) | |
| return Transcribe() | |
| def YoutubeTranscribe(URL, retries = 5): | |
| if retries: | |
| if "youtu" not in URL.lower(): | |
| raise gr.Error(f"{URL} is not a valid youtube URL.") | |
| else: | |
| try: | |
| os.remove("temp_audio.wav") | |
| except OSError: | |
| pass | |
| ydl_opts = { | |
| 'format': 'bestaudio/best', | |
| 'outtmpl': 'temp_audio.%(ext)s', | |
| 'postprocessors': [{ | |
| 'key': 'FFmpegExtractAudio', | |
| 'preferredcodec': 'wav', | |
| }], | |
| } | |
| try: | |
| with youtube_dl.YoutubeDL(ydl_opts) as ydl: | |
| ydl.download([URL]) | |
| except: | |
| return YoutubeTranscribe(URL, retries-1) | |
| stream = ffmpeg.input('temp_audio.m4a') | |
| stream = ffmpeg.output(stream, 'temp_audio.wav') | |
| try: | |
| os.remove("temp_audio.m4a") | |
| except OSError: | |
| pass | |
| return Transcribe() | |
| else: | |
| raise gr.Error(f"Unable to get video from {URL}") | |
| # with gr.Blocks() as i: | |
| # with gr.Row(): | |
| # with gr.Column(): | |
| # with gr.Row(): | |
| # video = gr.Video() | |
| # audio = gr.Audio() | |
| # text = gr.Textbox(label="Youtube Link", placeholder="https://www.youtube.com/watch?v=GECcjrYHH8w") | |
| # btn = gr.Button("Run") | |
| # with gr.Row(): | |
| # output = gr.Textbox(label="Transcribed Text", lines=15) | |
| # if not video and not text: | |
| # raise gr.Error("Either input url or video (not both)") | |
| # else: | |
| # btn.click(fn=YoutubeTranscribe, inputs=text, outputs=output) | |
| # i.launch() | |
| ut = gr.Interface( | |
| fn=YoutubeTranscribe, | |
| inputs=gr.Textbox(label="Youtube Link", placeholder="https://www.youtube.com/watch?v=GECcjrYHH8w"), | |
| outputs=gr.Textbox(label="Transcribed Text", lines=15) | |
| ) | |
| vt = gr.Interface( | |
| fn=VideoTranscribe, | |
| inputs='video', | |
| outputs=gr.Textbox(label="Transcribed Text", lines=15) | |
| ) | |
| at = gr.Interface( | |
| fn=AudioTranscribe, | |
| inputs='audio', | |
| outputs=gr.Textbox(label="Transcribed Text", lines=15) | |
| ) | |
| demo = gr.TabbedInterface([ut, vt, at], ["Youtube URL", "Video", "Audio"]) | |
| demo.launch() | |
| # YoutubeTranscribe('https://www.youtube.com/watch?v=GECcjrYHH8w') |