Summarize.ipynb

{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "colab_type": "text",
        "id": "view-in-github"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/drive/1n5csfChR4iRIhvaIpuh5hCrgrUiPYXUi?usp=sharing\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "xkogwIy7IfQO"
      },
      "source": [
        "# https://github.com/martinopiaggi/summarize\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "cellView": "form",
        "id": "jwROe6WH2lZi"
      },
      "outputs": [],
      "source": [
        "# @markdown ## 🔗 **Source Configuration**\n",
        "\n",
        "# @markdown **Source Type**\n",
        "Type_of_source = \"YouTube Video\"  # @param [\"YouTube Video\", \"Google Drive Video Link\", \"Dropbox Video Link\", \"Local File\"]\n",
        "\n",
        "# @markdown **Source URL or Path**\n",
        "Source = \"\"  # @param {type:\"string\"}\n",
        "\n",
        "# Set variables based on user input\n",
        "Type = Type_of_source\n",
        "URL = Source\n",
        "\n",
        "# @markdown **Use YouTube Captions**\n",
        "\n",
        "# @markdown If source is a Youtube video, it's recommended to use the available YouTube captions to save on transcription time and API usage. \n",
        "\n",
        "use_Youtube_captions = True  # @param {type:\"boolean\"}\n",
        "\n",
        "# @markdown ---\n",
        "# @markdown ## 🌐 **API Configuration**\n",
        "\n",
        "# @markdown The summarization process uses the API key specified in `api_key` variable. \n",
        "# @markdown Ensure you have set the required environment variables or Colab secrets for your API keys.\n",
        "\n",
        "api_endpoint = \"Groq\"  # @param [\"Groq\", \"OpenAI\", \"Custom\"]\n",
        "\n",
        "# Define endpoints and models based on the selected API\n",
        "endpoints = {\n",
        "    \"Groq\": \"https://api.groq.com/openai/v1\",\n",
        "    \"OpenAI\": \"https://api.openai.com/v1\",\n",
        "    \"Custom\": \"http://localhost:1234/v1\"  # Example custom endpoint\n",
        "}\n",
        "base_url = endpoints.get(api_endpoint)\n",
        "\n",
        "# Define models based on the selected API\n",
        "model = {\n",
        "    \"Groq\": \"llama-3.1-70b-versatile\",\n",
        "    \"OpenAI\": \"gpt-4\",\n",
        "    \"Custom\": \"custom-model-id\"  # Placeholder for any custom model\n",
        "}.get(api_endpoint)\n",
        "\n",
        "# @markdown ---\n",
        "# @markdown ## 🎤 **Transcription Settings**\n",
        "# @markdown The transcription settings are applied only if you want to use Whisper transcription and not Youtube Captions. \n",
        "\n",
        "\n",
        "# @markdown If you plan to use Whisper API endpoint (only **Groq** endpoint is supported for now) you have to specify your Groq API key in `api_key_groq`.\n",
        "\n",
        "# @markdown Why use `api_key_groq` and `api_key` ? So that you can use a different API for summarization (e.g., OpenAI), specify the corresponding API key in `api_key`.\n",
        "\n",
        "# @markdown If using locally Whisper: remember to switch the runtime type in Google Colab to a GPU instance (e.g., T4). Go to **Runtime** > **Change runtime type** and select **GPU** as the hardware accelerator.\n",
        "\n",
        "# @markdown **Transcription Method**\n",
        "transcription_method = \"Cloud Whisper\"  # @param [\"Cloud Whisper\", \"Local Whisper\"]\n",
        "\n",
        "# @markdown **Language** (ISO-639-1 code, e.g., \"en\" for English)\n",
        "language = \"auto\"  # @param {type:\"string\"}\n",
        "\n",
        "# @markdown **Initial Prompt for Whisper** (Optional)\n",
        "initial_prompt = \"\"  # @param {type:\"string\"}"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Libraries and helper functions\n",
        "Re-run if you change settings in the previous cell"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "cellView": "form",
        "collapsed": true,
        "id": "sJnZWCPc3uOH"
      },
      "outputs": [],
      "source": [
        "# @markdown ## Libraries and helper functions\n",
        "# @markdown Re-run if you change settings in the previous cell\n",
        "\n",
        "import subprocess\n",
        "import re\n",
        "import os\n",
        "!pip install python-dotenv\n",
        "from dotenv import load_dotenv\n",
        "\n",
        "if use_Youtube_captions:\n",
        "  !pip install youtube-transcript-api\n",
        "  from youtube_transcript_api import YouTubeTranscriptApi\n",
        "\n",
        "if (not Type == \"YouTube Video\") or (not use_Youtube_captions):\n",
        "  if transcription_method == \"Local Whisper\":\n",
        "    !pip install openai-whisper\n",
        "    import whisper\n",
        "  else:\n",
        "    !pip install --upgrade groq\n",
        "    from groq import Groq\n",
        "\n",
        "if Type == \"YouTube Video\":\n",
        "  !pip install pytubefix\n",
        "  from pytubefix import YouTube\n",
        "\n",
        "if Type == \"Google Drive Video Link\":\n",
        "  from google.colab import drive\n",
        "  drive.mount('/content/drive')\n",
        "\n",
        "\n",
        "# Function to get configuration value\n",
        "def get_api_key():\n",
        "    if api_endpoint == \"Groq\":\n",
        "      return get_groq_api_key()\n",
        "    try:\n",
        "        from google.colab import userdata\n",
        "        api_key = userdata.get('api_key')\n",
        "    except ImportError:\n",
        "        load_dotenv()\n",
        "        api_key = os.getenv('api_key')\n",
        "\n",
        "    if not api_key:\n",
        "        raise ValueError(\"API key not found in environment variables or Colab secrets\")\n",
        "\n",
        "    return api_key\n",
        "\n",
        "def get_groq_api_key():\n",
        "    try:\n",
        "        from google.colab import userdata\n",
        "        groq_api_key = userdata.get('api_key_groq')\n",
        "    except ImportError:\n",
        "        load_dotenv()\n",
        "        groq_api_key = os.getenv('api_key_groq')\n",
        "\n",
        "    if not groq_api_key:\n",
        "        raise ValueError(\"Groq API key not found in environment variables or Colab secrets\")\n",
        "\n",
        "    return groq_api_key\n",
        "\n",
        "# Converts the audio file to MP3 with low sample rate and bitrate to reduce the file size (to stay in audio file API limits) \n",
        "def process_audio_file(input_path, output_path):\n",
        "    command_convert = [\n",
        "        'ffmpeg', '-y', '-i', input_path,\n",
        "        '-ar', str(8000),\n",
        "        '-ac', str(1),\n",
        "        '-b:a', '16k',\n",
        "        output_path\n",
        "    ]\n",
        "    subprocess.run(command_convert, check=True)\n",
        "\n",
        "\n",
        "!pip install openai\n",
        "import openai\n",
        "client = openai.OpenAI(api_key = get_api_key(), base_url=base_url)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        " ## Video fetching\n",
        " Re-run cell if you change the source URL"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "cellView": "form",
        "id": "eLIU6bAX9a9v"
      },
      "outputs": [],
      "source": [
        "# @markdown ## Video fetching\n",
        "# @markdown Re-run cell if you change the source URL\n",
        "skip_transcription=False\n",
        "transcription_text = \"\"\n",
        "textTimestamps = \"\"\n",
        "\n",
        "def seconds_to_time_format(seconds):\n",
        "    hours, remainder = divmod(seconds, 3600)\n",
        "    minutes, seconds = divmod(remainder, 60)\n",
        "    return f\"{int(hours):02d}:{int(minutes):02d}:{int(seconds):02d}\"\n",
        "\n",
        "def download_youtube_audio_only(url):\n",
        "    yt = YouTube(url)\n",
        "    audio_stream = yt.streams.get_audio_only()\n",
        "    saved_path = audio_stream.download(mp3=True,output_path=\".\", skip_existing=True)\n",
        "    return saved_path\n",
        "\n",
        "def download_youtube_captions(url):\n",
        "    regex = r'(?:https?:\\/\\/)?(?:www\\.)?(?:youtube\\.com\\/(?:[^\\/\\n\\s]+\\/\\S+\\/|(?:v|e(?:mbed)?)\\/|\\S*?[?&]v=)|youtu\\.be\\/)([a-zA-Z0-9_-]{11})'\n",
        "    video_id =  re.search(regex, url).group(1)\n",
        "    transcript_list = YouTubeTranscriptApi.list_transcripts(video_id)\n",
        "\n",
        "    try:\n",
        "      transcript = YouTubeTranscriptApi.get_transcript(video_id, languages=['en'])\n",
        "    except:\n",
        "      for available_transcript in transcript_list:\n",
        "        if available_transcript.is_translatable:\n",
        "          transcript = available_transcript.translate('en').fetch()\n",
        "          break\n",
        "\n",
        "    transcription_text = \"\"\n",
        "    for entry in transcript:\n",
        "            start_time = seconds_to_time_format(entry['start'])\n",
        "            transcription_text += f\"{start_time} {entry['text'].strip()}\\n\"\n",
        "\n",
        "    transcript_file_name = f\"{video_id}_captions.md\"\n",
        "\n",
        "    with open(transcript_file_name, 'w', encoding='utf-8') as f:\n",
        "      f.write(transcription_text)\n",
        "\n",
        "    return transcription_text,transcript_file_name\n",
        "\n",
        "if Type == \"YouTube Video\":\n",
        "    #clean youtube url from timestamp\n",
        "    URL = re.sub('\\&t=\\d+s?', '', URL)\n",
        "    if use_Youtube_captions:\n",
        "      transcription_text, transcript_file_name = download_youtube_captions(URL)\n",
        "      skip_transcription=True\n",
        "    else:\n",
        "      video_path_local =  download_youtube_audio_only(URL)\n",
        "      # Process the audio file to reduce its size\n",
        "      processed_audio_path = os.path.splitext(video_path_local)[0] + '_processed.mp3'\n",
        "      process_audio_file(video_path_local, processed_audio_path)\n",
        "      video_path_local = processed_audio_path  # Update to the processed file path\n",
        "\n",
        "elif Type == \"Google Drive Video Link\":\n",
        "  subprocess.run(['ffmpeg', '-y', '-i', \"drive/MyDrive/\" + URL, '-vn', '-acodec', 'pcm_s16le',\n",
        "                  '-ar', '16000', '-ac', '1', 'gdrive_audio.wav'], check=True)\n",
        "  video_path_local = \"gdrive_audio.wav\"\n",
        "  # Process the audio file to reduce its size\n",
        "  processed_audio_path = os.path.splitext(video_path_local)[0] + '_processed.mp3'\n",
        "  process_audio_file(video_path_local, processed_audio_path)\n",
        "  video_path_local = processed_audio_path  # Update to the processed file path\n",
        "\n",
        "elif Type == \"Dropbox Video Link\":\n",
        "    subprocess.run(['wget', URL, '-O', 'dropbox_video.mp4'], check=True)\n",
        "    subprocess.run(['ffmpeg', '-y', '-i', 'dropbox_video.mp4', '-vn', '-acodec', 'pcm_s16le',\n",
        "                    '-ar', '16000', '-ac', '1', 'dropbox_video_audio.wav'], check=True)\n",
        "    video_path_local = \"dropbox_video_audio.wav\"\n",
        "    # Process the audio file to reduce its size\n",
        "    processed_audio_path = os.path.splitext(video_path_local)[0] + '_processed.mp3'\n",
        "    process_audio_file(video_path_local, processed_audio_path)\n",
        "    video_path_local = processed_audio_path  # Update to the processed file path\n",
        "\n",
        "elif Type == \"Local File\":\n",
        "    local_file_path = Source\n",
        "    subprocess.run(['ffmpeg', '-y', '-i', local_file_path, '-vn', '-acodec', 'pcm_s16le',\n",
        "                    '-ar', '16000', '-ac', '1', 'local_file_audio.wav'], check=True)\n",
        "    video_path_local = \"local_file_audio.wav\"\n",
        "    # Process the audio file to reduce its size\n",
        "    processed_audio_path = os.path.splitext(video_path_local)[0] + '_processed.mp3'\n",
        "    process_audio_file(video_path_local, processed_audio_path)\n",
        "    video_path_local = processed_audio_path  # Update to the processed file path"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Transcription using Whisper\n",
        "***Only run this cell if the source is not YouTube or you decided not to use YouTube captions.***"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "cellView": "form",
        "id": "iOTFm1vPAVDh"
      },
      "outputs": [],
      "source": [
        "# @markdown ### Transcription\n",
        "# @markdown Re-run cell if you change transcription settings\n",
        "if not skip_transcription:\n",
        "    transcription_text = \"\"\n",
        "\n",
        "    if video_path_local:\n",
        "        # Single file transcription\n",
        "        audio_files = [video_path_local]\n",
        "    else:\n",
        "        # Multiple chunk files\n",
        "        audio_files = audio_chunks\n",
        "\n",
        "    for audio_file_path in audio_files:\n",
        "        if transcription_method == \"Local Whisper\":\n",
        "            # Local Whisper transcription\n",
        "            transcription = model_whisper.transcribe(\n",
        "                audio_file_path,\n",
        "                beam_size=5,\n",
        "                language=None if language == \"auto\" else language,\n",
        "                task=\"translate\",\n",
        "                initial_prompt=initial_prompt or None\n",
        "            )\n",
        "\n",
        "            for segment in transcription[\"segments\"]:\n",
        "                start_time = seconds_to_time_format(segment['start'])\n",
        "                transcription_text += f\"{start_time} {segment['text'].strip()} \"\n",
        "\n",
        "        elif transcription_method == \"Cloud Whisper\":\n",
        "            # Cloud Whisper using Groq API\n",
        "            groq_client = Groq(api_key=get_groq_api_key())\n",
        "            with open(audio_file_path, \"rb\") as audio_file:\n",
        "                transcription_response = groq_client.audio.transcriptions.create(\n",
        "                    file=(os.path.basename(audio_file_path), audio_file.read()),\n",
        "                    model=\"distil-whisper-large-v3-en\" if language == \"en\" else \"whisper-large-v3\",\n",
        "                    prompt=initial_prompt or None,\n",
        "                    response_format=\"verbose_json\",\n",
        "                    language=None if language == \"auto\" else language,\n",
        "                    temperature=0.0\n",
        "                )\n",
        "\n",
        "            # Corrected code using dot notation\n",
        "            for segment in transcription_response.segments:\n",
        "                start_time = seconds_to_time_format(segment['start'])\n",
        "                transcription_text += f\"{start_time} {segment['text'].strip()} \"\n",
        "else:\n",
        "  print(\"Using YouTube captions for transcription.\")\n",
        "\n",
        "# Save the transcription\n",
        "if not skip_transcription:\n",
        "    transcript_file_name = 'transcription.md'\n",
        "    with open(transcript_file_name, 'w', encoding='utf-8') as f:\n",
        "        f.write(transcription_text)\n",
        "else:\n",
        "    transcript_file_name = f\"{video_id}_captions.md\"\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Summarization and elaboration"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "cellView": "form",
        "id": "fWeEGfgFAoni"
      },
      "outputs": [],
      "source": [
        "prompt_type = \"Questions and answers\"  # @param ['Summarization', 'Only grammar correction with highlights','Distill Wisdom', 'Questions and answers']\n",
        "\n",
        "# @markdown Parallel API calls (mind rate limits)\n",
        "parallel_api_calls = 30 # @param\n",
        "\n",
        "# @markdown Chunk size (tokens) (mind model context length). Higher = less granular summary.\n",
        "# @markdown Rule of thumb: 28k for 3h, 10k for 1h, 5k for 30min, 4k for shorter.\n",
        "chunk_size = 18000 # @param\n",
        "\n",
        "# @markdown Overlap (tokens) between chunks\n",
        "overlap_size = 20 # @param\n",
        "\n",
        "# @markdown Max output tokens of each chunk (mind model limits). Higher = less granular summary.\n",
        "# @markdown Rule of thumb: 4k, 2k or 1k depending on content density.\n",
        "max_output_tokens = 2096 # @param\n",
        "\n",
        "final_summary = \"\"\n",
        "\n",
        "prompts = {\n",
        "    'Distill Wisdom': \"\"\"Analyze the transcript and extract key insights and wisdom including a concise title that reflects the content.\n",
        "\n",
        "**{TITLE}**\n",
        "\n",
        "**IDEAS**\n",
        "- ...\n",
        "\n",
        "**QUOTES**\n",
        "- ...\n",
        "\n",
        "**REFERENCES**\n",
        "- ...\n",
        "\n",
        "- **Formatting Guidelines**:\n",
        "  - **Title**: Start with the title in bold (`**{TITLE}**`).\n",
        "  - **Categories**: Use bold for category headers (`**IDEAS**`, `**QUOTES**`, `**REFERENCES**`).\n",
        "  - **Bullet Points**: Use hyphens (`-`) for each bullet point.\n",
        "  - **Omit Empty Categories**: Do not include a category if there are no relevant items.\n",
        "  - **No Additional Text**: Do not add any introductory phrases, explanations, or headers using `#`.\n",
        "  - **Strict Template Adherence**: Follow the template exactly as shown above without deviations.\n",
        "\n",
        "Here is the text:\n",
        "\"\"\",\n",
        "    'Summarization': \"\"\"Summarize the video transcript excerpt including a concise title that reflects the content. Wrap the title with **markdown bold notation**. Write the summary as if you are continuing a conversation without needing to signal a beginning. Here is the transcript: \"\"\",\n",
        "    'Only grammar correction with highlights': \"\"\"Repeat the following text correcting any grammatical errors and formatting error. Highlight only the important quote (if there are any) with **markdown bold notation**. Focus solely on the essence of the content as if you are continuing a conversation without using any form of introduction like 'Here's the corrected text:'. Here is the text to fix: \"\"\",\n",
        "    'Questions and answers': \"\"\"Analyze the input text and generate 5 essential questions that, when answered, capture the main points and core meaning of the text. Do not add any introductory phrases, explanations! Just start with the questions and answers. Mark each  question with **bold syntax** and don't number them. 2.) When formulating your questions: a. Address the central theme or argument b. Identify key supporting ideas c. Highlight important facts or evidence d. Reveal the author's purpose or perspective e. Explore any significant implications or conclusions. 3.) Answer all of your generated questions one-by-one in detail.\n",
        "\n",
        "Here is the text:\n",
        "\"\"\"\n",
        "   }\n",
        "\n",
        "# Select the appropriate prompt\n",
        "summary_prompt = prompts[prompt_type]\n",
        "\n",
        "\n",
        "def extract_and_clean_timestamps(text_chunks):\n",
        "    timestamp_pattern = re.compile(r'(\\d{2}:\\d{2}:\\d{2})')\n",
        "    cleaned_texts = []\n",
        "    timestamp_ranges = []\n",
        "    for chunk in text_chunks:\n",
        "        timestamps = timestamp_pattern.findall(chunk)\n",
        "        if timestamps:\n",
        "            for timestamp in timestamps:\n",
        "                # Remove each found timestamp from the chunk\n",
        "                chunk = chunk.replace(timestamp, \"\")\n",
        "            timestamp_ranges.append(timestamps[0])  # Assuming you want the first timestamp per chunk\n",
        "        else:\n",
        "            timestamp_ranges.append(\"\")\n",
        "        cleaned_texts.append(chunk.strip())  # Strip to remove any leading/trailing whitespace\n",
        "    return cleaned_texts, timestamp_ranges\n",
        "\n",
        "def format_timestamp_link(timestamp):\n",
        "    if Type == \"YouTube Video\":\n",
        "      hours, minutes, seconds = map(int, timestamp.split(':'))\n",
        "      total_seconds = hours * 3600 + minutes * 60 + seconds\n",
        "      return f\"{timestamp} - {URL}&t={total_seconds}\"\n",
        "    else:\n",
        "      return f\"{timestamp}\"\n",
        "\n",
        "import concurrent.futures\n",
        "import time\n",
        "\n",
        "def summarize(prompt):\n",
        "    completion = client.chat.completions.create(\n",
        "            model=model,\n",
        "            messages=[\n",
        "            {\"role\": \"system\", \"content\": summary_prompt},\n",
        "            {\"role\": \"user\", \"content\": prompt}\n",
        "            ],\n",
        "            max_tokens=max_output_tokens\n",
        "    )\n",
        "    return completion.choices[0].message.content\n",
        "\n",
        "def process_and_summarize(text):\n",
        "    texts = [text[i:i+chunk_size] for i in range(0, len(text), chunk_size - overlap_size)]\n",
        "    cleaned_texts, timestamp_ranges = extract_and_clean_timestamps(texts)\n",
        "    summaries = []\n",
        "\n",
        "    with concurrent.futures.ThreadPoolExecutor(max_workers=parallel_api_calls) as executor:\n",
        "        future_to_chunk = {executor.submit(summarize, text_chunk): idx for idx, text_chunk in enumerate(cleaned_texts)}\n",
        "        for future in concurrent.futures.as_completed(future_to_chunk):\n",
        "            idx = future_to_chunk[future]\n",
        "            try:\n",
        "                summarized_chunk = future.result()\n",
        "                summary_piece = format_timestamp_link(timestamp_ranges[idx]) + \"\\n\\n\" + summarized_chunk\n",
        "                summary_piece += \"\\n\"\n",
        "                summaries.append((idx, summary_piece))\n",
        "            except Exception as exc:\n",
        "                print(f'Chunk {idx} generated an exception: {exc}')\n",
        "                # Resubmit the task with the new model\n",
        "                time.sleep(10)\n",
        "                future_to_chunk[executor.submit(summarize, texts[idx])] = idx\n",
        "\n",
        "    summaries.sort()  # Ensure summaries are in the correct order\n",
        "    final_summary = \"\\n\\n\".join([summary for _, summary in summaries])\n",
        "\n",
        "    # Save the final summary\n",
        "    final_name = transcript_file_name.replace(\".md\", \"_FINAL.md\") if Type != \"Dropbox video link\" else \"final_dropbox_video.md\"\n",
        "    with open(final_name, 'w') as f:\n",
        "        f.write(final_summary)\n",
        "\n",
        "\n",
        "process_and_summarize(transcription_text)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "# @markdown Clean folder\n",
        "!rm *.md"
      ]
    }
  ],
  "metadata": {
    "colab": {
      "include_colab_link": true,
      "provenance": []
    },
    "kernelspec": {
      "display_name": "Python 3 (ipykernel)",
      "language": "python",
      "name": "python3"
    },
    "language_info": {
      "codemirror_mode": {
        "name": "ipython",
        "version": 3
      },
      "file_extension": ".py",
      "mimetype": "text/x-python",
      "name": "python",
      "nbconvert_exporter": "python",
      "pygments_lexer": "ipython3",
      "version": "3.11.9"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}