app.py

import gradio as gr
import os
import time
import sys
import subprocess
import tempfile
import requests
from urllib.parse import urlparse

# Clone and install faster-whisper from GitHub
subprocess.run(["git", "clone", "https://github.com/SYSTRAN/faster-whisper.git"], check=True)
subprocess.run(["pip", "install", "-e", "./faster-whisper"], check=True)
subprocess.run(["pip", "install", "yt-dlp pytube ffmpeg-python"], check=True)

# Add the faster-whisper directory to the Python path
sys.path.append("./faster-whisper")

from faster_whisper import WhisperModel
from faster_whisper.transcribe import BatchedInferencePipeline
import yt_dlp

def download_audio(url):
    parsed_url = urlparse(url)
    if parsed_url.netloc in ['www.youtube.com', 'youtu.be', 'youtube.com']:
        return download_youtube_audio(url)
    else:
        return download_direct_audio(url)

def download_youtube_audio(url):
    methods = [
        youtube_dl_method,
        pytube_method,
        youtube_dl_alternative_method,
        ffmpeg_method
    ]

    for method in methods:
        try:
            return method(url)
        except Exception as e:
            print(f"Method {method.__name__} failed: {str(e)}")
    
    raise Exception("All download methods failed. Please try a different video or a direct audio URL.")

def youtube_dl_method(url):
    ydl_opts = {
        'format': 'bestaudio/best',
        'postprocessors': [{
            'key': 'FFmpegExtractAudio',
            'preferredcodec': 'mp3',
            'preferredquality': '192',
        }],
        'outtmpl': '%(id)s.%(ext)s',
    }
    with yt_dlp.YoutubeDL(ydl_opts) as ydl:
        info = ydl.extract_info(url, download=True)
        return f"{info['id']}.mp3"

def pytube_method(url):
    from pytube import YouTube
    yt = YouTube(url)
    audio_stream = yt.streams.filter(only_audio=True).first()
    out_file = audio_stream.download()
    base, ext = os.path.splitext(out_file)
    new_file = base + '.mp3'
    os.rename(out_file, new_file)
    return new_file

def youtube_dl_alternative_method(url):
    ydl_opts = {
        'format': 'bestaudio/best',
        'postprocessors': [{
            'key': 'FFmpegExtractAudio',
            'preferredcodec': 'mp3',
            'preferredquality': '192',
        }],
        'outtmpl': '%(id)s.%(ext)s',
        'no_warnings': True,
        'quiet': True,
        'no_check_certificate': True,
        'prefer_insecure': True,
        'nocheckcertificate': True,
    }
    with yt_dlp.YoutubeDL(ydl_opts) as ydl:
        info = ydl.extract_info(url, download=True)
        return f"{info['id']}.mp3"

def ffmpeg_method(url):
    output_file = tempfile.mktemp(suffix='.mp3')
    command = ['ffmpeg', '-i', url, '-vn', '-acodec', 'libmp3lame', '-q:a', '2', output_file]
    subprocess.run(command, check=True, capture_output=True)
    return output_file

def download_direct_audio(url):
    response = requests.get(url)
    if response.status_code == 200:
        with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as temp_file:
            temp_file.write(response.content)
            return temp_file.name
    else:
        raise Exception(f"Failed to download audio from {url}")

def transcribe_audio(input_source, batch_size):
    try:
        # Initialize the model
        model = WhisperModel("cstr/whisper-large-v3-turbo-int8_float32", device="auto", compute_type="int8")
        batched_model = BatchedInferencePipeline(model=model)

        # Handle input source
        if isinstance(input_source, str) and (input_source.startswith('http://') or input_source.startswith('https://')):
            # It's a URL, download the audio
            audio_path = download_audio(input_source)
        else:
            # It's a local file path
            audio_path = input_source

        # Benchmark transcription time
        start_time = time.time()
        segments, info = batched_model.transcribe(audio_path, batch_size=batch_size)
        end_time = time.time()

        # Generate transcription
        transcription = ""
        for segment in segments:
            transcription += f"[{segment.start:.2f}s -> {segment.end:.2f}s] {segment.text}\n"

        # Calculate metrics
        transcription_time = end_time - start_time
        real_time_factor = info.duration / transcription_time
        audio_file_size = os.path.getsize(audio_path) / (1024 * 1024)  # Size in MB

        # Prepare output
        output = f"Transcription:\n\n{transcription}\n"
        output += f"\nLanguage: {info.language}, Probability: {info.language_probability:.2f}\n"
        output += f"Duration: {info.duration:.2f}s, Duration after VAD: {info.duration_after_vad:.2f}s\n"
        output += f"Transcription time: {transcription_time:.2f} seconds\n"
        output += f"Real-time factor: {real_time_factor:.2f}x\n"
        output += f"Audio file size: {audio_file_size:.2f} MB"

        return output

    except Exception as e:
        return f"An error occurred: {str(e)}"

    finally:
        # Clean up downloaded file if it was a URL
        if isinstance(input_source, str) and (input_source.startswith('http://') or input_source.startswith('https://')):
            try:
                os.remove(audio_path)
            except:
                pass

# Gradio interface
iface = gr.Interface(
    fn=transcribe_audio,
    inputs=[
        gr.Textbox(label="Audio Source (Upload, MP3 URL, or YouTube URL)"),
        gr.Slider(minimum=1, maximum=32, step=1, value=16, label="Batch Size")
    ],
    outputs=gr.Textbox(label="Transcription and Metrics"),
    title="Faster Whisper Multi-Input Transcription",
    description="Enter an audio file path, MP3 URL, or YouTube URL to transcribe using Faster Whisper (GitHub version). Adjust the batch size for performance tuning.",
    examples=[
        ["https://www.youtube.com/watch?v=dQw4w9WgXcQ", 16],
        ["https://example.com/path/to/audio.mp3", 16],
        ["path/to/local/audio.mp3", 16]
    ],
    cache_examples=False  # Prevents automatic processing of examples
)

iface.launch()