bluenevus's picture
Update app.py
b8cd6c2 verified
raw
history blame
8.14 kB
import base64
import io
import os
import threading
import tempfile
import logging
import openai
from dash import Dash, dcc, html, Input, Output, State, callback, callback_context
import dash_bootstrap_components as dbc
from pydub import AudioSegment
import requests
import yt_dlp
import mimetypes
import urllib.parse
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# Try to import moviepy with the simpler import statement
try:
from moviepy import VideoFileClip
logger.info("MoviePy (VideoFileClip) successfully imported")
except ImportError as e:
logger.error(f"Error importing MoviePy (VideoFileClip): {str(e)}")
logger.error("Please ensure moviepy is installed correctly")
raise
# Supported file formats
AUDIO_FORMATS = ['.wav', '.mp3', '.ogg', '.flac', '.aac', '.m4a', '.wma']
VIDEO_FORMATS = ['.mp4', '.avi', '.mov', '.flv', '.wmv', '.mkv', '.webm']
SUPPORTED_FORMATS = AUDIO_FORMATS + VIDEO_FORMATS
# Initialize the Dash app
app = Dash(__name__, external_stylesheets=[dbc.themes.BOOTSTRAP])
# Global variables
generated_file = None
transcription_text = ""
# Set up OpenAI API key
openai.api_key = os.getenv("OPENAI_API_KEY")
# Layout
app.layout = dbc.Container([
html.H1("Audio/Video Transcription and Diarization App", className="text-center my-4"),
dbc.Card([
dbc.CardBody([
dcc.Upload(
id='upload-media',
children=html.Div([
'Drag and Drop or ',
html.A('Select Audio/Video File')
]),
style={
'width': '100%',
'height': '60px',
'lineHeight': '60px',
'borderWidth': '1px',
'borderStyle': 'dashed',
'borderRadius': '5px',
'textAlign': 'center',
'margin': '10px'
},
multiple=False
),
html.Div(id='output-media-upload'),
dbc.Input(id="url-input", type="text", placeholder="Enter audio/video URL (including YouTube)", className="mb-3"),
dbc.Button("Process URL", id="process-url-button", color="primary", className="mb-3"),
dbc.Spinner(html.Div(id='transcription-status'), color="primary", type="grow"),
html.H4("Diarized Transcription Preview", className="mt-4"),
html.Div(id='transcription-preview', style={'whiteSpace': 'pre-wrap'}),
html.Br(),
dbc.Button("Download Transcription", id="btn-download", color="primary", className="mt-3", disabled=True),
dcc.Download(id="download-transcription")
])
])
], fluid=True)
def process_media(file_path, is_url=False):
global generated_file, transcription_text
temp_file = None
wav_path = None
try:
if is_url:
logger.info(f"Processing URL: {file_path}")
try:
ydl_opts = {
'format': 'bestaudio/best',
'postprocessors': [{
'key': 'FFmpegExtractAudio',
'preferredcodec': 'wav',
}],
'outtmpl': '%(id)s.%(ext)s',
}
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
info = ydl.extract_info(file_path, download=True)
wav_path = f"{info['id']}.wav"
logger.info(f"Audio downloaded: {wav_path}")
except Exception as e:
logger.error(f"Error downloading audio from URL: {str(e)}")
return f"Error downloading audio from URL: {str(e)}", False
else:
logger.info("Processing uploaded file")
temp_file = tempfile.NamedTemporaryFile(delete=False)
temp_file.write(file_path)
temp_file.close()
logger.info(f"Uploaded file saved: {temp_file.name}")
file_extension = os.path.splitext(temp_file.name)[1].lower()
logger.info(f"Detected file extension: {file_extension}")
if file_extension in VIDEO_FORMATS:
logger.info("Processing video file")
video = VideoFileClip(temp_file.name)
audio = video.audio
wav_path = temp_file.name + ".wav"
audio.write_audiofile(wav_path)
video.close()
elif file_extension in AUDIO_FORMATS:
logger.info("Processing audio file")
audio = AudioSegment.from_file(temp_file.name, format=file_extension[1:])
wav_path = temp_file.name + ".wav"
audio.export(wav_path, format="wav")
else:
logger.error(f"Unsupported file format: {file_extension}")
return f"Unsupported file format: {file_extension}. Please upload a supported audio or video file.", False
logger.info(f"Audio extracted to WAV: {wav_path}")
with open(wav_path, "rb") as audio_file:
transcript = openai.Audio.transcribe("whisper-1", audio_file)
audio_file.seek(0)
diarized_transcript = openai.Audio.transcribe("whisper-1", audio_file, response_format="verbose_json")
formatted_transcript = ""
if 'segments' in diarized_transcript:
for segment in diarized_transcript["segments"]:
speaker = segment.get('speaker', 'Unknown')
text = segment.get('text', '')
formatted_transcript += f"Speaker {speaker}: {text}\n\n"
else:
formatted_transcript = transcript.get('text', 'No transcription available.')
transcription_text = formatted_transcript
generated_file = io.BytesIO(transcription_text.encode())
logger.info("Transcription and diarization completed successfully")
return "Transcription and diarization completed successfully!", True
except Exception as e:
logger.error(f"Error during processing: {str(e)}")
return f"An error occurred: {str(e)}", False
finally:
if temp_file and os.path.exists(temp_file.name):
os.unlink(temp_file.name)
if wav_path and os.path.exists(wav_path):
os.unlink(wav_path)
@app.callback(
[Output('output-media-upload', 'children'),
Output('transcription-status', 'children'),
Output('transcription-preview', 'children'),
Output('btn-download', 'disabled')],
[Input('upload-media', 'contents'),
Input('process-url-button', 'n_clicks')],
[State('upload-media', 'filename'),
State('url-input', 'value')]
)
def update_output(contents, n_clicks, filename, url):
ctx = callback_context
if not ctx.triggered:
return "No file uploaded or URL processed.", "", "", True
# Clear the preview pane
transcription_preview = ""
if contents is not None:
content_type, content_string = contents.split(',')
decoded = base64.b64decode(content_string)
status_message, success = process_media(decoded)
elif url:
status_message, success = process_media(url, is_url=True)
else:
return "No file uploaded or URL processed.", "", "", True
if success:
preview = transcription_text[:1000] + "..." if len(transcription_text) > 1000 else transcription_text
return f"Media processed successfully.", status_message, preview, False
else:
return "Processing failed.", status_message, transcription_preview, True
@app.callback(
Output("download-transcription", "data"),
Input("btn-download", "n_clicks"),
prevent_initial_call=True,
)
def download_transcription(n_clicks):
if n_clicks is None:
return None
return dcc.send_bytes(generated_file.getvalue(), "diarized_transcription.txt")
if __name__ == '__main__':
print("Starting the Dash application...")
app.run(debug=True, host='0.0.0.0', port=7860)
print("Dash application has finished running.")