import streamlit as st import streamlit_ext as ste import openai from pydub import AudioSegment # from pytube import YouTube # import pytube import yt_dlp import io from pyannote.audio import Pipeline from pyannote.audio.pipelines.utils.hook import ProgressHook from pyannote.database.util import load_rttm from pyannote.core import Annotation, Segment, notebook import time import json import torch import urllib.parse as urlparse from urllib.parse import urlencode import os import unicodedata import re import matplotlib matplotlib.use('Agg') from matplotlib import pyplot as plt st.set_page_config( page_title="Speech-to-chat", page_icon = '๐ŸŒŠ', layout='wide' ) # Set your OpenAI, Hugging Face API keys try: openai.api_key = st.secrets['openai'] hf_api_key = st.secrets['hf'] except Exception: openai.api_key = os.getenv['openai'] hf_api_key = os.getenv['hf'] TRANSCRIPTION_REQUEST_LIMIT = 550 PROMPT_REQUEST_LIMIT = 20 DURATION_LIMIT = 3600 # seconds def create_audio_stream(audio): return io.BytesIO(audio.export(format="wav").read()) def add_query_parameter(link, params): url_parts = list(urlparse.urlparse(link)) query = dict(urlparse.parse_qsl(url_parts[4])) query.update(params) url_parts[4] = urlencode(query) return urlparse.urlunparse(url_parts) def slugify(value, allow_unicode=False): """ Taken from https://github.com/django/django/blob/master/django/utils/text.py Convert to ASCII if 'allow_unicode' is False. Convert spaces or repeated dashes to single dashes. Remove characters that aren't alphanumerics, underscores, or hyphens. Convert to lowercase. Also strip leading and trailing whitespace, dashes, and underscores. """ value = str(value) if allow_unicode: value = unicodedata.normalize('NFKC', value) else: value = unicodedata.normalize('NFKD', value).encode('ascii', 'ignore').decode('ascii') value = re.sub(r'[^\w\s-]', '', value.lower()) return re.sub(r'[-\s]+', '-', value).strip('-_') def youtube_video_id(value): """ Examples: - http://youtu.be/SA2iWivDJiE - http://www.youtube.com/watch?v=_oPAwA_Udwc&feature=feedu - http://www.youtube.com/embed/SA2iWivDJiE - http://www.youtube.com/v/SA2iWivDJiE?version=3&hl=en_US """ query = urlparse.urlparse(value) if query.hostname == 'youtu.be': return query.path[1:] if query.hostname in ('www.youtube.com', 'youtube.com'): if query.path == '/watch': p = urlparse.parse_qs(query.query) return p['v'][0] if query.path[:7] == '/embed/': return query.path.split('/')[2] if query.path[:3] == '/v/': return query.path.split('/')[2] # fail? return None @st.cache_data def process_youtube_link2(youtube_link): ''' uses pytube https://github.com/pytube/pytube issue with https://github.com/pytube/pytube/issues/84 ''' try: yt = YouTube(youtube_link) audio_stream = yt.streams.filter(only_audio=True).first() audio_name = audio_stream.default_filename st.write(f"Downloaded {audio_name}") except pytube.exceptions.AgeRestrictedError: st.warning('Age restricted videos cannot be processed.') st.stop() try: os.remove('sample.mp4') except OSError: pass audio_file = audio_stream.download(filename='sample.mp4') time.sleep(2) audio = load_audio('sample.mp4') st.audio(create_audio_stream(audio), format="audio/mp4", start_time=0) return audio, audio_name @st.cache_data def process_youtube_link(youtube_link): 'uses yt-dlp https://github.com/yt-dlp/yt-dlp' try: os.remove('sample.m4a') except OSError: pass ydl_opts = { 'format': 'm4a/bestaudio/best', # โ„น๏ธ See help(yt_dlp.postprocessor) for a list of available Postprocessors and their arguments 'outtmpl': './sample.%(ext)s' # 'postprocessors': [{ # Extract audio using ffmpeg # 'key': 'FFmpegExtractAudio', # 'preferredcodec': 'm4a', # }] } try: with yt_dlp.YoutubeDL(ydl_opts) as ydl: info = ydl.extract_info(youtube_link, download=True) audio_name = slugify( info['title'] ) st.write(f"Downloaded {info['title']}") except Exception as e: st.warning(e) st.stop() audio = load_audio(f'sample.m4a') st.audio(create_audio_stream(audio), format="audio/m4a", start_time=0) return audio, audio_name @st.cache_data def load_rttm_file(rttm_path): return load_rttm(rttm_path)['stream'] def load_audio(uploaded_audio): return AudioSegment.from_file(uploaded_audio) if "openai_model" not in st.session_state: st.session_state["openai_model"] = "gpt-4o-mini" if "prompt_request_counter" not in st.session_state: st.session_state["prompt_request_counter"] = 0 initial_prompt = [{"role": "system", "content": "You are helping to analyze and summarize a transcript of a conversation."}, {"role": 'user', "content": 'Please summarize briefly below transcript and inlcude a list of tags with a hash for SEO. \n{}'}] if "messages" not in st.session_state: st.session_state.messages = initial_prompt st.title("Speech-to-Chat") reddit_thread = 'https://www.reddit.com/r/dataisbeautiful/comments/17413bq/oc_speech_diarization_app_that_transcribes_audio' with st.sidebar: st.markdown(''' # How to Use 1. Enter a youtube link. 2. "Chat" with the video. Example prompts: - Which speaker spoke the most? - Give me a list of tags with a hash for SEO based on this transcript. ''') api_key_input = st.text_input( "OpenAI API Key to lift request limits (Coming soon)", disabled=True, type="password", placeholder="Paste your OpenAI API key here (sk-...)", help="You can get your API key from https://platform.openai.com/account/api-keys.", # noqa: E501 value=os.environ.get("OPENAI_API_KEY", None) or st.session_state.get("OPENAI_API_KEY", ""), ) st.divider() st.markdown(f''' # About Given an audio file or a youtube link this app will - [x] 1. Partition the audio according to the identity of each speaker (diarization) using `pyannote` [HuggingFace Speaker Diarization api](https://huggingface.co/pyannote/speaker-diarization-3.0) - [x] 2. Transcribe each audio segment using [OpenAi Whisper API](https://platform.openai.com/docs/guides/speech-to-text/quickstart) - [x] 3. Set up an LLM chat with the transcript loaded into its knowledge database, so that a user can "talk" to the transcript of the audio file. This version will only process up to first 6 minutes of an audio file due to limited resources of free tier Streamlit.io/HuggingFace Spaces. A local version with access to a GPU can process 1 hour of audio in 1 to 5 minutes. If you would like to use this app at scale reach out directly by creating an issue on [github๐Ÿค–](https://github.com/KobaKhit/speech-to-text-app/issues)! Rule of thumb, for this free tier hosted app it takes half the duration of the audio to complete processing, ex. g. 6 minute youtube video will take 3 minutes to diarize. Made by [kobakhit](https://github.com/KobaKhit/speech-to-text-app) ''') # Chat container container_transcript_chat = st.container() # Source Selection option = st.radio("Select source:", [ "Use YouTube link","See Example"], index=0) # Upload audio file if option == "Upload an audio file": with st.form('uploaded-file', clear_on_submit=True): uploaded_audio = st.file_uploader("Upload an audio file (MP3 or WAV)", type=["mp3", "wav","mp4"]) st.form_submit_button() if st.form_submit_button(): st.session_state.messages = initial_prompt with st.expander('Optional Parameters'): # st.session_state.rttm = st.file_uploader("Upload .rttm if you already have one", type=["rttm"]) # st.session_state.transcript_file = st.file_uploader("Upload transcipt json", type=["json"]) youtube_link = st.text_input('Youtube link of the audio sample') if uploaded_audio is not None: st.audio(uploaded_audio, format="audio/wav", start_time=0) audio_name = uploaded_audio.name audio = load_audio(uploaded_audio) # sample_rate = st.number_input("Enter the sample rate of the audio", min_value=8000, max_value=48000) # audio = audio.set_frame_rate(sample_rate) # use youtube link elif option == "Use YouTube link": with st.form('youtube-link'): youtube_link_raw = st.text_input("Enter the YouTube video URL:") youtube_link = f'https://youtu.be/{youtube_video_id(youtube_link_raw)}' if st.form_submit_button(): # reset variables on new link submit process_youtube_link.clear() st.session_state.messages = initial_prompt st.session_state.rttm = None st.session_state.transcript_file = None st.session_state.prompt_request_counter = 0 with container_transcript_chat: st.empty() # with st.expander('Optional Parameters'): # st.session_state.rttm = st.file_uploader("Upload .rttm if you already have one", type=["rttm"]) # st.session_state.transcript_file = st.file_uploader("Upload transcipt json", type=["json"]) if youtube_link_raw: audio, audio_name = process_youtube_link(youtube_link) # sample_rate = st.number_input("Enter the sample rate of the audio", min_value=8000, max_value=48000) # audio = audio.set_frame_rate(sample_rate) # except Exception as e: # st.write(f"Error: {str(e)}") elif option == 'See Example': youtube_link = 'https://www.youtube.com/watch?v=TamrOZX9bu8' audio_name = 'Stephen A. Smith has JOKES with Shannon Sharpe' st.write(f'Loaded audio file from {youtube_link} - {audio_name} ๐Ÿ‘๐Ÿ˜‚') if os.path.isfile('example/steve a smith jokes.mp4'): audio = load_audio('example/steve a smith jokes.mp4') else: yt = YouTube(youtube_link) audio_stream = yt.streams.filter(only_audio=True).first() audio_file = audio_stream.download(filename='sample.mp4') time.sleep(2) audio = load_audio('sample.mp4') if os.path.isfile("example/steve a smith jokes.rttm"): st.session_state.rttm = "example/steve a smith jokes.rttm" if os.path.isfile('example/steve a smith jokes.json'): st.session_state.transcript_file = 'example/steve a smith jokes.json' st.audio(create_audio_stream(audio), format="audio/mp4", start_time=0) # Diarize if "audio" in locals(): # create stream duration = audio.duration_seconds if duration > DURATION_LIMIT: st.info(f'Only processing the first {int(DURATION_LIMIT/6/6)} minutes of the audio due to Streamlit.io resource limits.') audio = audio[:DURATION_LIMIT*1000] duration = audio.duration_seconds # Perform diarization with PyAnnote pipeline = Pipeline.from_pretrained( "pyannote/speaker-diarization-3.0", use_auth_token=hf_api_key) if torch.cuda.device_count() > 0: # use gpu if available st.write('Using cuda - GPU') pipeline.to(torch.device('cuda')) # run the pipeline on an audio file with st.spinner('Performing Diarization...'): if 'rttm' in st.session_state and st.session_state.rttm != None: st.write(f'Loading {st.session_state.rttm}') diarization = load_rttm_file(st.session_state.rttm ) else: # make progress hook # with ProgressHook() as hook: # diarization = pipeline(audio_, hook=hook) diarization = pipeline(create_audio_stream(audio)) # dump the diarization output to disk using RTTM format with open(f'{audio_name.split(".")[0]}.rttm', "w") as f: diarization.write_rttm(f) st.session_state.rttm = f'{audio_name.split(".")[0]}.rttm' # Display the diarization results st.write("Diarization Results:") annotation = Annotation() sp_chunks = [] progress_text = f"Processing 1/{len(sp_chunks)}..." my_bar = st.progress(0, text=progress_text) counter = 0 n_tracks = len([a for a in diarization.itertracks(yield_label=True)]) for turn, _, speaker in diarization.itertracks(yield_label=True): annotation[turn] = speaker progress_text = f"Processing {counter}/{len(sp_chunks)}..." my_bar.progress((counter+1)/n_tracks, text=progress_text) counter +=1 temp = {'speaker': speaker, 'start': turn.start, 'end': turn.end, 'duration': turn.end-turn.start, 'audio': audio[turn.start*1000:turn.end*1000]} if 'transcript_file' in st.session_state and st.session_state.transcript_file == None: temp['audio_stream'] = create_audio_stream(audio[turn.start*1000:turn.end*1000]) sp_chunks.append(temp) # plot notebook.crop = Segment(-1, duration + 1) figure, ax = plt.subplots(figsize=(10,3)) notebook.plot_annotation(annotation, ax=ax, time=True, legend=True) figure.tight_layout() # save to file st.pyplot(figure) st.write('Speakers and Audio Samples') with st.expander('Samples', expanded=True): for speaker in set(s['speaker'] for s in sp_chunks): temp = max(filter(lambda d: d['speaker'] == speaker, sp_chunks), key=lambda x: x['duration']) speak_time = sum(c['duration'] for c in filter(lambda d: d['speaker'] == speaker, sp_chunks)) rate = 100*min((speak_time, duration))/duration speaker_summary = f"{temp['speaker']} ({round(rate)}% of video duration): start={temp['start']:.1f}s stop={temp['end']:.1f}s" if youtube_link != None: speaker_summary += f" {add_query_parameter(youtube_link, {'t':str(int(temp['start']))})}" st.write(speaker_summary) st.audio(create_audio_stream(temp['audio'])) st.divider() # # Perform transcription with Whisper ASR # Transcript containers st.write(f'Transcribing using Whisper API ({TRANSCRIPTION_REQUEST_LIMIT} requests limit)...') container_transcript_completed = st.container() progress_text = f"Processing 1/{len(sp_chunks[:TRANSCRIPTION_REQUEST_LIMIT])}..." my_bar = st.progress(0, text=progress_text) # rework the loop. Simplify if Else with st.expander('Transcript', expanded=True): if 'transcript_file' in st.session_state and st.session_state.transcript_file != None: with open(st.session_state.transcript_file,'r') as f: sp_chunks_loaded = json.load(f) for i,s in enumerate(sp_chunks_loaded): if s['transcript'] != None: transcript_summary = f"**{s['speaker']}** start={float(s['start']):.1f}s end={float(s['end']):.1f}s: {s['transcript']}" if youtube_link != None and youtube_link != '': transcript_summary += f" {add_query_parameter(youtube_link, {'t':str(int(s['start']))})}" st.markdown(transcript_summary) progress_text = f"Processing {i+1}/{len(sp_chunks_loaded)}..." my_bar.progress((i+1)/len(sp_chunks_loaded), text=progress_text) transcript_json = sp_chunks_loaded transcript_path = f'{audio_name.split(".")[0]}-transcript.json' else: sp_chunks_updated = [] for i,s in enumerate(sp_chunks[:TRANSCRIPTION_REQUEST_LIMIT]): if s['duration'] > 0.1: audio_path = s['audio'].export('temp.wav',format='wav') try: transcript = openai.Audio.transcribe("whisper-1", audio_path)['text'] except Exception: transcript = '' pass if transcript !='' and transcript != None: s['transcript'] = transcript transcript_summary = f"**{s['speaker']}** start={s['start']:.1f}s end={s['end']:.1f}s : {s['transcript']}" if youtube_link != None: transcript_summary += f" {add_query_parameter(youtube_link, {'t':str(int(s['start']))})}" sp_chunks_updated.append({'speaker':s['speaker'], 'start':s['start'], 'end':s['end'], 'duration': s['duration'],'transcript': transcript}) st.markdown(transcript_summary) progress_text = f"Processing {i+1}/{len(sp_chunks[:TRANSCRIPTION_REQUEST_LIMIT])}..." my_bar.progress((i+1)/len(sp_chunks[:TRANSCRIPTION_REQUEST_LIMIT]), text=progress_text) transcript_json = [dict((k, d[k]) for k in ['speaker','start','end','duration','transcript'] if k in d) for d in sp_chunks_updated] transcript_path = f'{audio_name.split(".")[0]}-transcript.json' st.session_state.transcript_file = transcript_path # save the trancript file with open(transcript_path,'w') as f: json.dump(transcript_json, f) # generate transcript string transcript_string = '\n'.join([f"{s['speaker']} start={s['start']:.1f}s end={s['end']:.1f}s : {s['transcript']}" for s in transcript_json]) @st.cache_data def get_initial_response(transcript_string): st.session_state.messages[1]['content'] = st.session_state.messages[1]['content'].format(transcript_string) initial_response = openai.ChatCompletion.create( model=st.session_state["openai_model"], messages=st.session_state.messages ) return initial_response['choices'][0]['message']['content'] # Chat container st.session_state.messages[1]['content'] = st.session_state.messages[1]['content'].format(transcript_string) with container_transcript_chat: # get a summary of transcript from ChatGpt try: init = get_initial_response(transcript_string) except openai.error.APIError: # st.stop('It is not you. It is not this app. It is OpenAI API thats having issues.') init = '' st.warning('OpenAI API is having issues. Hope they resolve it soon. Refer to https://status.openai.com/') # pass transcript to initial prompt # LLM Chat with st.expander('Summary of the Transcribed Audio File Generated by [`gpt-40-mini`](https://platform.openai.com/docs/models/gpt-4o-mini)', expanded = True): # display the AI generated summary. with st.chat_message("assistant", avatar='https://upload.wikimedia.org/wikipedia/commons/0/04/ChatGPT_logo.svg'): st.write(init) # chat field with st.form("Chat",clear_on_submit=True): prompt = st.text_input(f'Chat with the Transcript ({int(PROMPT_REQUEST_LIMIT)} prompts limit)') st.form_submit_button() # message list # for message in st.session_state.messages[2:]: # with st.chat_message(message["role"]): # st.markdown(message["content"]) # make request if prompt was entered if prompt: st.session_state.prompt_request_counter += 1 if st.session_state.prompt_request_counter > PROMPT_REQUEST_LIMIT: st.warning('Exceeded prompt limit.'); st.stop() # append user prompt to messages st.session_state.messages.append({"role": "user", "content": prompt}) # dislay user prompt with st.chat_message("user"): st.markdown(prompt) # stream LLM Assisstant response with st.chat_message("assistant"): message_placeholder = st.empty() full_response = "" # stream response for response in openai.ChatCompletion.create( model=st.session_state["openai_model"], messages=[ {"role": m["role"], "content": m["content"]} for m in st.session_state.messages ], stream=True, ): full_response += response.choices[0].delta.get("content", "") message_placeholder.markdown(full_response + "โ–Œ") message_placeholder.markdown(full_response) # append ai response to messages st.session_state.messages.append({"role": "assistant", "content": full_response}) # Trancription Completed Section with container_transcript_completed: st.info(f'Completed transcribing') @st.cache_data def convert_df(string): # IMPORTANT: Cache the conversion to prevent computation on every rerun return string.encode('utf-8') # encode transcript string transcript_json_download = convert_df(json.dumps(transcript_json)) # transcript download buttons c1_b,c2_b = st.columns((1,1)) # json button with c1_b: ste.download_button( "Download transcript as json", transcript_json_download, transcript_path, ) # create csv string header = ','.join(transcript_json[0].keys()) + '\n' for s in transcript_json: header += ','.join([str(e) if ',' not in str(e) else '"' + str(e) + '"' for e in s.values()]) + '\n' # csv button transcript_csv_download = convert_df(header) with c2_b: ste.download_button( "Download transcript as csv", transcript_csv_download, f'{audio_name.split(".")[0]}-transcript.csv' )