groq_backend.py

import os
import shutil
import logging
import traceback
import re
from typing import List, Optional

from dotenv import load_dotenv
from groq import Groq
import yt_dlp
from pydub import AudioSegment


logging.basicConfig(level=logging.INFO)
load_dotenv('.env')

def transcribe_with_groq(file_path: str) -> str:
    """
    Transcribes an audio file using Groq's transcription service.
    This function uses Groq's audio transcription API to convert the audio file into text. The audio file is expected to be in an MP3 format.
    Groq utilizes the Whisper model, an automatic speech recognition system developed by OpenAI, for efficient and accurate audio transcription.
    Args:
        file_path (str): The absolute or relative path to the audio file that needs to be transcribed. This file should be accessible and readable.
    Returns:
        str: The full transcript of the audio file.
    Raises:
        Exception: If there are issues in transcribing the audio file.
    """
    client = Groq()
    filename = os.path.basename(file_path)

    try:
        with open(file_path, "rb") as file:
            result = client.audio.transcriptions.create(
                file=(filename, file.read()),
                model="whisper-large-v3",
            )
        transcription = result.text
        logging.info(transcription)
        
        return transcription
    except Exception as e:
        logging.error(f"Error during transcription: {str(e)}")
        logging.error(traceback.format_exc())
        raise

def yt_dlp_download(yt_url:str, output_path:str = None) -> str:
    """
    Downloads the audio track from a specified YouTube video URL using the yt-dlp library, then converts it to an MP3 format file.
    This function configures yt-dlp to extract the best quality audio available and uses FFmpeg (via yt-dlp's postprocessors) to convert the audio to MP3 format. The resulting MP3 file is saved to the specified or default output directory with a filename derived from the video title.
    Args:
        yt_url (str): The URL of the YouTube video from which audio will be downloaded. This should be a valid YouTube video URL.
    Returns:
        str: The absolute file path of the downloaded and converted MP3 file. This path includes the filename which is derived from the original video title.
    Raises:
        yt_dlp.utils.DownloadError: If there is an issue with downloading the video's audio due to reasons such as video unavailability or restrictions.
        
        Exception: For handling unexpected errors during the download and conversion process.
    """
    if output_path is None:
        output_path = os.getcwd()

    ydl_opts = {
        'format': 'bestaudio/best',
        'postprocessors': [{
            'key': 'FFmpegExtractAudio',
            'preferredcodec': 'mp3',
            'preferredquality': '192',
        }],
        'outtmpl': os.path.join(output_path, '%(title)s.%(ext)s'),
    }

    try:
        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
            result = ydl.extract_info(yt_url, download=True)
            file_name = ydl.prepare_filename(result)
            mp3_file_path = file_name.rsplit('.', 1)[0] + '.mp3'
            logging.info(f"yt_dlp_download saved YouTube video to file path: {mp3_file_path}")
            return mp3_file_path
    except yt_dlp.utils.DownloadError as e:
        logging.error(f"yt_dlp_download failed to download audio from URL {yt_url}: {e}")
        raise
    except Exception as e:
        logging.error(f"An unexpected error occurred with yt_dlp_download: {e}")
        logging.error(traceback.format_exc())
        raise


def create_audio_chunks(audio_file: str, chunk_size: int, temp_dir: str) -> List[str]:
    """
    Splits an audio file into smaller segments or chunks based on a specified duration. This function is useful for processing large audio files incrementally or in parallel, which can be beneficial for tasks such as audio analysis or transcription where handling smaller segments might be more manageable.
    AudioSegment can slice an audio file by specifying the start and end times in milliseconds. This allows you to extract precise segments of the audio without needing to process the entire file at once. For example, `audio[1000:2000]` extracts a segment from the 1-second mark to the 2-second mark of the audio file.
    Args:
        audio_file (str): The absolute or relative path to the audio file that needs to be chunked. This file should be accessible and readable.
        
        chunk_size (int): The length of each audio chunk expressed in milliseconds. This value determines how the audio file will be divided. For example, a `chunk_size` of 1000 milliseconds will split the audio into chunks of 1 second each.
        
        temp_dir (str): The directory where the temporary audio chunk files will be stored. This directory will be used to save the output chunk files, and it must have write permissions. If the directory does not exist, it will be created.
    Returns:
        List[str]: A list containing the file paths of all the audio chunks created. Each path in the list represents a single chunk file stored in the specified `temp_dir`. The files are named sequentially based on their order in the original audio file.
    Raises:
        FileNotFoundError: If the `audio_file` does not exist or is inaccessible.
        
        PermissionError: If the script lacks the necessary permissions to read the `audio_file` or write to the `temp_dir`.
        ValueError: If `chunk_size` is set to a non-positive value.
    """
    os.makedirs(temp_dir, exist_ok=True)
    file_name = os.path.splitext(os.path.basename(audio_file))[0]

    try:
        audio = AudioSegment.from_file(audio_file)
    except Exception as e:
        logging.error(f"create_audio_chunks failed to load audio file {audio_file}: {e}")
        logging.error(traceback.format_exc())
        return []

    start = 0
    end = chunk_size
    counter = 0
    chunk_files = []



    while start < len(audio):
        chunk = audio[start:end]
        chunk_file_path = os.path.join(temp_dir, f"{counter}_{file_name}.mp3")
        try:
            chunk.export(chunk_file_path, format="mp3") # Using .mp3 because it's cheaper
            chunk_files.append(chunk_file_path)
        except Exception as e:
            error_message = f"create_audio_chunks failed to export chunk {counter}: {e}"
            logging.error(error_message)
            logging.error(traceback.format_exc())
            raise error_message
        start += chunk_size
        end += chunk_size
        counter += 1
    return chunk_files

def generate_youtube_transcript_with_groq(yt_url: str, chunk_size: int, temp_dir: str) -> str:
    """
    Generate a transcript for a YouTube video using Groq's transcription service by processing the video's audio.
    This function performs several steps to achieve transcription:
    1. It validates the provided YouTube URL to ensure it is correctly formatted.
    2. Downloads the YouTube video to a local file.
    3. Splits the downloaded video's audio track into manageable chunks of a specified duration.
    4. Transcribes each audio chunk into text using Groq's transcription service.
    5. Aggregates the transcriptions of all chunks to form a complete transcript of the video.
    6. Cleans up all temporary files and directories created during the process to free up system resources.
    Args:
        yt_url (str): The URL of the YouTube video to transcribe. This should be a valid YouTube link.
        
        chunk_size (int): The duration of each audio chunk in milliseconds. This determines how the audio is divided for transcription.
        
        temp_dir (str): The directory to store the temporary audio chunk files. This directory will be used for intermediate storage and must be writable.
    Returns:
        str: The full transcript of the YouTube video, which is a concatenation of all transcribed audio chunks.
    Raises:
        ValueError: If the YouTube URL is invalid, indicating the URL does not match the expected format or cannot be processed.
        
        Exception: If there are issues in any of the steps such as downloading the video, creating audio chunks, transcribing the chunks, or cleaning up temporary files. Specific errors will provide more details on the step that failed.
    """
    youtube_url_pattern = r'^(https?://)?(www\.)?(youtube\.com|youtu\.be)/.+$'
    if not re.match(youtube_url_pattern, yt_url):
        logging.error(f"Invalid YouTube URL: {yt_url}")
        raise ValueError("Invalid YouTube URL provided.")
    
    try:
        file_path = yt_dlp_download(yt_url)
    except Exception as e:
        logging.error(f"generate_youtube_transcript_with_groq failed to download YouTube video from URL {yt_url}: {e}")
        logging.error(traceback.format_exc())
        raise

    try:
        chunk_files = create_audio_chunks(file_path, chunk_size, temp_dir)
    except Exception as e:
        error_message = f"generate_youtube_transcript_with_groq failed to create audio chunks from file {file_path}: {e}"
        logging.error(error_message)
        logging.error(traceback.format_exc())
        raise error_message

    transcripts = []
    for file_name in chunk_files:
        try:
            logging.info(f"Transcribing {file_name}")
            transcript = transcribe_with_groq(file_name)
            transcripts.append(transcript)
        except Exception as e:
            error_message = f"generate_youtube_transcript_with_groq failed to transcribe file {file_name}: {e}"
            logging.error(error_message)
            logging.error(traceback.format_exc())
            raise error_message

    full_transcript = " ".join(transcripts)

    try:
        # Clean up the temporary directory
        shutil.rmtree(temp_dir)
    except Exception as e:
        error_message = f"generate_youtube_transcript_with_groq failed to remove temporary directory {temp_dir}: {e}"
        logging.error(error_message)
        logging.error(traceback.format_exc())
        raise error_message

    try:
        # Remove the downloaded video file
        os.remove(file_path)
    except Exception as e:
        error_message = (f"generate_youtube_transcript_with_groq failed to remove downloaded file {file_path}: {e}")
        logging.error(error_message)
        logging.error(traceback.format_exc())
        raise error_message
    
    return full_transcript

def save_transcript_to_file(transcript: str, folder_path: str, file_name: str):
    """
    Saves the transcript to a text file in a specified folder with a specified file name.
    
    Args:
        transcript (str): The transcript text to save.
        folder_path (str): The path to the folder where the file will be saved.
        file_name (str): The name of the file (without the .txt extension).
    """
    if not os.path.exists(folder_path):
        os.makedirs(folder_path)
    
    file_path = os.path.join(folder_path, file_name + '.txt')
    
    with open(file_path, 'w') as file:
        file.write(transcript)
    
    logging.info(f"Transcript saved to {file_path}")

if __name__ == "__main__":
    yt_url = "https://www.youtube.com/watch?v=ZUOYyXg7ewo&ab_channel=TalkTottenham"
    chunk_size = 25*60000
    temp_directory = "temp_chunks"
    transcript = generate_youtube_transcript_with_groq(yt_url, chunk_size, temp_directory)

    # Save the transcript
    folder_to_save = "transcripts"
    file_name_to_save = "youtube_transcript"
    save_transcript_to_file(transcript, folder_to_save, file_name_to_save)
    print(f"Transcript saved in {folder_to_save}/{file_name_to_save}.txt")