commit new files

README.md

@@ -1,3 +1,38 @@
----
-license: mit
----
+# youtube2blog
+
+![screenshot](https://github.com/S4mpl3r/youtube2blog/blob/main/assets/screenshot.jpg)
+
+youtube2blog offers a seamless solution for transforming any YouTube video, or audio file, into a comprehensive blog post. This tool leverages the power of [Groq](https://groq.com) and [Deepgram](https://deepgram.com) to provide a streamlined content creation process.
+
+## Features
+- **Video-to-Blog conversion:** Easily convert any YouTube video into a blog post with just the video's URL.
+- **Keyword Extraction:** Extract top 10 keywords from a youtube video.
+- **Transcription:** Obtain a full transcript of the video for further analysis or content creation.
+
+This tool is not limited to YouTube videos; users can also input their own audio files to generate blog posts, extract keywords, and transcribe content.
+
+## Installation
+To use this tool you should obtain Groq and Deepgram API keys. Groq is currently free, and Deepgram provides $200 credit which is more than enough to run this tool.
+To install, do the following:
+1. Clone the repository:
+   ```bash
+   git clone https://github.com/S4mpl3r/youtube2blog.git
+   ```
+2. Create a python environment and activate it. (optional, but highly recommended)
+3. Create a .env file in the project root and populate it with your API keys:
+   ```bash
+   GROQ_API_KEY=<YOUR_KEY>
+   DEEPGRAM_API_KEY=<YOUR_KEY>
+   ```
+3. Install the required packages
+   ```bash
+   python -m pip install -r requirements.txt
+   ```
+4. CD into the youtube2blog directory and run the tool:
+   ```bash
+   cd youtube2blog/
+   python youtube2blog.py
+   ```
+
+## License
+MIT

falsk_app.py

@@ -0,0 +1,55 @@
+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()

groq_backend.py

@@ -0,0 +1,246 @@
+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")