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.")