Summarize.ipynb

{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "colab_type": "text",
        "id": "view-in-github"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/martinopiaggi/summarize/blob/main/Summarize.ipynb\" 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": [
        "Source = \"\" #@param {type:\"string\"}\n",
        "Type_of_source = \"Local file\" #@param ['Youtube video or playlist', 'Google Drive video link','Dropbox video link', 'Local file']\n",
        "Type = Type_of_source\n",
        "URL = Source                                                                                                                                                                                                    \n",
        "\n",
        "#@markdown ---\n",
        "#@markdown Insert your API key depending on which endpoint you want to use\n",
        "# API Configuration\n",
        "api_key = \"lm-studio\"  # @param {type:\"string\"}\n",
        "api_endpoint = \"Custom\"  # @param ['Groq', 'OpenAI', 'Custom'] \n",
        "\n",
        "endpoints = {\n",
        "    \"Groq\": \"https://api.groq.com/openai/v1\",\n",
        "    \"OpenAI\": \"https://api.openai.com\",\n",
        "    \"Custom\": \"http://localhost:1234/v1\"  # Example custom endpoint\n",
        "}\n",
        "base_url = endpoints.get(api_endpoint)\n",
        "\n",
        "models = {                                  \n",
        "    \"Groq\": \"llama3-8b-8192\",\n",
        "    \"OpenAI\": \"gpt-3.5-turbo\",\n",
        "    \"Custom\": \"Meta-Llama-3-8B-Instruct-GGUF\"  # Placeholder for any custom model\n",
        "}\n",
        "model = models.get(api_endpoint)\n",
        "\n",
        "use_Youtube_captions = True #@param {type:\"boolean\"}\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "M4M4nmKcnt2M"
      },
      "source": [
        "*Remember to change runtime type to T4 GPU in case you are planning to use Whisper and not youtube autogenerated captions*"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Installation of libraries\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": [
        "# %%capture\n",
        "import subprocess\n",
        "import re\n",
        "import os \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 playlist\") or (not use_Youtube_captions):\n",
        "  !pip install openai-whisper\n",
        "  import whisper\n",
        "\n",
        "!pip install openai\n",
        "import openai\n",
        "client = openai.OpenAI(api_key=api_key, base_url=base_url)\n",
        "\n",
        "\n",
        "if Type == \"Youtube video or playlist\":\n",
        "  !pip install git+https://github.com/pytube/pytube\n",
        "  from pytube import YouTube\n",
        "\n",
        "if Type == \"Google Drive video link\":\n",
        "  from google.colab import drive\n",
        "  drive.mount('/gdrive')"
      ]
    },
    {
      "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": [
        "skip_transcription=False\n",
        "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(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",
        "    text = \"\"\n",
        "    for entry in transcript:\n",
        "            start_time = seconds_to_time_format(entry['start'])\n",
        "            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(text)\n",
        "\n",
        "    return text,transcript_file_name\n",
        "\n",
        "if Type == \"Youtube video or playlist\":\n",
        "    #clean youtube url from timestamp\n",
        "    URL = re.sub('\\&t=\\d+s?', '', URL)\n",
        "    if use_Youtube_captions:\n",
        "      text, transcript_file_name = download_youtube_captions(URL)\n",
        "      skip_transcription=True\n",
        "    else:\n",
        "      video_path_local =  download_youtube_audio_only(URL)\n",
        "\n",
        "elif Type == \"Google Drive video link\":\n",
        "  subprocess.run(['ffmpeg', '-y', '-i', '/gdrive/My Drive/' + URL, '-vn', '-acodec', 'pcm_s16le',\n",
        "                  '-ar', '16000', '-ac', '1', 'gdrive_audio.wav'], check=True)\n",
        "  video_path_local = \"gdrive_audio.wav\"\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",
        "\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\""
      ]
    },
    {
      "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.***\n",
        "You can specify the language and initial prompt to increase speed"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "cellView": "form",
        "id": "iOTFm1vPAVDh"
      },
      "outputs": [],
      "source": [
        "#%%capture\n",
        "if not skip_transcription:\n",
        "  language = \"en\" # @param {type:\"string\"}\n",
        "  initial_prompt = \"\" # @param {type:\"string\"}\n",
        "\n",
        "  modelWhisper = whisper.load_model('tiny', device=\"cuda\")\n",
        "  transcription = modelWhisper.transcribe(str(video_path_local), beam_size=5,\n",
        "                                    language=None if language == \"auto\" else language,\n",
        "                                    task=\"translate\",\n",
        "                                    initial_prompt=initial_prompt)\n",
        "  \n",
        "  transcript_file_name = video_path_local.replace(\".mp4\", \".md\") #to check\n",
        "  transcript_file_name = video_path_local.replace(\".wav\", \".md\")\n",
        "\n",
        "with open(transcript_file_name, 'w') as f:\n",
        "    for segment in transcription[\"segments\"]:\n",
        "        start_time = seconds_to_time_format(segment['start'])\n",
        "        text += f\"{start_time} {segment['text'].strip()} \"\n",
        "\n",
        "    f.write(text)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Summarization and elaboration"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "cellView": "form",
        "id": "fWeEGfgFAoni"
      },
      "outputs": [],
      "source": [
        "prompt_type = \"Summarization\"  # @param ['Summarization', 'Only grammar correction with highlights']\n",
        "# @markdown Set the number of parallel API calls (be mindful of usage rate limits)\n",
        "parallel_api_calls = 1 # @param\n",
        "\n",
        "\n",
        "# Define your prompts using a dictionary for easier management\n",
        "prompts = {\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",
        "}\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 or playlist\":\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=4096\n",
        "    )\n",
        "    return completion.choices[0].message.content\n",
        "\n",
        "def process_and_summarize(text):\n",
        "    chunk_size, overlap_size = 4096, 20\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]) + \" \" + 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(text)"
      ]
    }
  ],
  "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.10.12"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}