Skip to content

Instantly share code, notes, and snippets.

@nickcmiller

nickcmiller/01_bash_setup

Last active July 28, 2024 01:54
Show Gist options
  • Save nickcmiller/5c56dea5412f4c8c99560c970a91d60b to your computer and use it in GitHub Desktop.
Save nickcmiller/5c56dea5412f4c8c99560c970a91d60b to your computer and use it in GitHub Desktop.
Download ZIP
YouTube Transcription with Groq's Whisper API
Raw
01_bash_setup
sudo apt update
sudo apt install python3-pip
sudo apt update
sudo apt install ffmpeg
Raw
02_requirements.txt
httpx
Flask
groq
pydub
python-dotenv
yt_dlp
Raw
03_groq_backend.py
import os
import shutil
import logging
import traceback
import re
from typing import List, Optional
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
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)
print(transcript)
Raw
flask_app.py
import groq_backend
from flask import Flask, request, jsonify, g
from typing import Dict, Union
import logging
import re
app = Flask(__name__)
logger = logging.getLogger(__name__)
# Configuration settings directly defined
app.config['DEFAULT_CHUNK_SIZE'] = 25*60*1000
app.config['TEMP_DIR'] = "temp"
YOUTUBE_URL_PATTERN = r'^(https?://)?(www\.)?(youtube\.com|youtu\.be)/.+$'
def validate_youtube_url(url: str) -> bool:
return re.match(YOUTUBE_URL_PATTERN, url) is not None
@app.route('/transcribe', methods=['POST'])
def transcribe() -> Union[Dict, str]:
data = request.json
yt_url = data.get('yt_url')
chunk_size = data.get('chunk_size', app.config['DEFAULT_CHUNK_SIZE'])
temp_dir = data.get('temp_dir', app.config['TEMP_DIR'])
if not yt_url:
error = "YouTube URL is required."
logger.error(error)
return jsonify({'error': error}), 400
if not validate_youtube_url(yt_url):
error = f"Invalid YouTube URL: {yt_url}"
logger.error(error)
return jsonify({'error': error}), 400
try:
print(f"Transcribing {yt_url} with chunk size {chunk_size} and temp dir {temp_dir}")
transcript = groq_backend.generate_youtube_transcript_with_groq(
yt_url, chunk_size, temp_dir
)
return jsonify({'transcript': transcript}), 200
except ValueError as e:
error = f"Failed to transcribe video: {str(e)}"
logger.exception(error)
return jsonify({'error': error}), 400
except Exception as e:
error = "Unexpected error occurred during transcription"
logger.exception(error)
return jsonify({'error': error}), 500
if __name__ == '__main__':
app.run()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment