[Python] Streamlit: AI Response Only Displays After Timer Timeout, Not on Button Click

I'm building a Streamlit app that interacts with YouTube videos using pytube to fetch audio and transcribe it. The problem I'm encountering is that the AI response (transcription result) is displayed only when the timer runs out, but not when the "Continue" button is clicked (within the custom_oauth_verifier function).

In my code, I use a threading.Event to wait for the user to click a button or for the timer to expire. When the timer expires, everything works fine, but when I click the "Continue..." button, I don't see the response in the app.

What could be causing the AI response not to display when I click the button? Any insights would be greatly appreciated!

For testing purposes, I've set the timer to 5 seconds for a short wait time and hardcoded a youtube url.

import queue
import tempfile
import threading
from io import BytesIO

import streamlit as st
from pytubefix import YouTube
from streamlit.runtime.scriptrunner import add_script_run_ctx


def custom_oauth_verifier(verification_url, user_code):
continue_event = threading.Event()
container = st.container()
with container:
message_area = st.empty()
st.button("Continue...", on_click=continue_event.set)

base_message = f"""
#### Please complete the verification process:

- Open the following URL: [{verification_url}]({verification_url})
- Enter the code: `{user_code}`

Once you have done that, click the **'Continue...'** button below.

_This action will expire in {{time_remaining}} seconds._
"""

for time_remaining in range(5, 0, -1):
message_area.markdown(base_message.format(time_remaining=time_remaining))
if continue_event.wait(1):
break
container.empty()


def get_video_transcript(url, result_queue):
try:
buf = BytesIO()
audio = YouTube(
url,
use_oauth=True,
allow_oauth_cache=False,
oauth_verifier=custom_oauth_verifier,
).streams.get_audio_only()
audio.stream_to_buffer(buf)
buf.seek(0)
with tempfile.NamedTemporaryFile(delete=True, suffix=".mp4") as temp_file:
temp_file.write(buf.read())
# TODO: Transcribe the audio here
result = "Lorem ipsum dolor..."
result_queue.put(result)
except Exception as e:
result = f"Error: {e}"
result_queue.put(result)


def transcribe_youtube_url(url):
result_queue = queue.Queue()
pytube_thread = threading.Thread(target=get_video_transcript, args=(url, result_queue))
add_script_run_ctx(pytube_thread)
pytube_thread.start()
pytube_thread.join()
return result_queue.get()


chat_history: list = st.session_state.setdefault(
"chat_history",
[{"type": "ai", "content": "Hello, how are you?"}],
)


for message in chat_history:
with st.chat_message(message["type"]):
st.write(message["content"])


if user_prompt := st.chat_input(placeholder="Type your message..."):
message = {"type": "human", "content": user_prompt}
chat_history.append(message)
with st.chat_message(message["type"]):
st.write(message["content"])

with st.spinner("Loading..."):
ai_response = transcribe_youtube_url("https://youtu.be/ahq6MF349Fw")
message = {"type": "ai", "content": ai_response}
chat_history.append(message)
with st.chat_message(message["type"]):
st.write(message["content"])
st.rerun()

Continue reading...