Podcast_summarizer.ipynb

{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/anupkashyap/podcast-summarizer/blob/main/Podcast_summarizer.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "GYTWIPZe4eUZ"
      },
      "source": [
        "# **Podcast summarizer using PEGASUS abstractive summerization model**"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "l2WOwtZ544cB"
      },
      "source": [
        "Dependencies\n",
        "1. PyTorch\n",
        "2. Hungging face Transformers\n",
        "3. Sentence Piece\n",
        "4. Pydub"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "MfSAToSZ496O"
      },
      "outputs": [],
      "source": [
        "!pip install torch\n",
        "!pip install sentencePiece\n",
        "!pip install SpeechRecognition\n",
        "!pip install transformers\n",
        "!pip install pydub"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "lhKCLEPS5z58"
      },
      "source": [
        "Import packages"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "q94z2n3h53VP"
      },
      "outputs": [],
      "source": [
        "import speech_recognition as sr\n",
        "import os\n",
        "import math\n",
        "from pydub import AudioSegment\n",
        "from pydub.silence import split_on_silence\n",
        "from transformers import  PegasusForConditionalGeneration, PegasusTokenizer\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "ErRDPmNw-CQn"
      },
      "outputs": [],
      "source": [
        "#Configurations\n",
        "debug=True\n",
        "outputLines=20\n",
        "test_file=\"test.mp3\""
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "tCPkjuwzdI1R"
      },
      "outputs": [],
      "source": [
        "#Load models\n",
        "\n",
        "tokenizer = PegasusTokenizer.from_pretrained(\"google/pegasus-xsum\")\n",
        "model=PegasusForConditionalGeneration.from_pretrained(\"google/pegasus-xsum\");"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "AcYrvK3hZAdM"
      },
      "source": [
        "Helpers"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "9NQD7UdmTEYq"
      },
      "outputs": [],
      "source": [
        "def process_audio_chunk(chunk):\n",
        "  silence=AudioSegment.silent(duration=10)\n",
        "  audio_chunk=silence+chunk+silence\n",
        "  chunk_file=\"chunk\"+str(i)+\".wav\"\n",
        "  print(\"Creating {0}\".format(chunk_file)) if debug==True else None\n",
        "  audio_chunk.export(\"./\"+chunk_file,bitrate='192k',format=\"wav\")\n",
        "  recognizer=sr.Recognizer()\n",
        "  with sr.AudioFile(chunk_file) as source:\n",
        "    recognizer.adjust_for_ambient_noise(source)\n",
        "    return(recognizer.listen(source))"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "o2-HunVnToQO"
      },
      "outputs": [],
      "source": [
        "def read_audio_file(file_name):\n",
        "  if(file_name.endswith(\".mp3\")):\n",
        "    #Extract from mp3\n",
        "    audio = AudioSegment.from_mp3(file_name)\n",
        "  elif(file_name.endswith(\".wav\")):\n",
        "    #Extract from wav\n",
        "    audio=AudioSegment.from_wav(file_name)\n",
        "  else:\n",
        "    print(\"Invalid Input format\")\n",
        "    return\n",
        "  return audio"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "9jOL1skn9XxD"
      },
      "outputs": [],
      "source": [
        "def preprocess_text(text_data):\n",
        "  text=textData[0]\n",
        "  length=len(text);\n",
        "  print(length)\n",
        "  quantSize=math.ceil(length/outputLines)\n",
        "  print(quantSize)\n",
        "  textArray=[];\n",
        "  for i in range(0,outputLines):\n",
        "    textArray.append(text[i*quantSize:(i*quantSize)+quantSize])\n",
        "  return textArray\n",
        "\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "QtbZ5PcI6jYd"
      },
      "source": [
        "Transcribing audio files"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "HKhKa8n36q6s"
      },
      "outputs": [],
      "source": [
        "def transcribe_audio(file_name):\n",
        "  print(\"Transcribing audio\")\n",
        "  audio=read_audio_file(file_name)\n",
        "  #Create directory\n",
        "  try:\n",
        "    os.mkdir(\"temp_audio_chunks\")\n",
        "  except(FileExistsError):\n",
        "    print(\"Directory already exists\") if debug else None\n",
        "  os.chdir(\"temp_audio_chunks\") \n",
        "  outputText = open(\"outputText.txt\", \"w+\")\n",
        "  print(audio)\n",
        "  #Process audio\n",
        "  audio_chunks=split_on_silence(audio,min_silence_len=800,silence_thresh=-32)\n",
        "  print(\"Number of chunks is \"+str(len(audio_chunks))) if debug else None\n",
        "  i=0\n",
        "  text=\"\"\n",
        "  for chunk in audio_chunks:\n",
        "    audio_listened=process_audio_chunk(chunk)\n",
        "    try:\n",
        "      res= recognizer.recognize_google(audio_listened)\n",
        "      text= text+ \" \"+ res\n",
        "    except sr.UnknownValueError:\n",
        "      print(\"Audio not clear\") if debug==True else None\n",
        "    except sr.RequestError as e:\n",
        "      print (\"Could not connect to API. Check internet connection\")\n",
        "    i+=1\n",
        "  outputText.write(text)\n",
        "  os.chdir(\"..\")"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "XzVidxKYZOjs"
      },
      "source": [
        "Abstractive summarization."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "A0qetleo90_Y"
      },
      "outputs": [],
      "source": [
        "def summarize(textChunks):\n",
        "  summaries=[]\n",
        "  test=[]\n",
        "  for i in range(1,outputLines):\n",
        "    token=tokenizer(textChunks[i],truncation=True,padding=\"longest\",return_tensors=\"pt\")\n",
        "    summary=append(model.generate(**token))\n",
        "    output=output + tokenizer.decode(summary[0]) +\".\"\n",
        "  return output"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "colab": {
          "background_save": true
        },
        "id": "1jfrnxywCiDH"
      },
      "outputs": [],
      "source": [
        "transcribe_audio(test_file)\n",
        "textData=\"\"\n",
        "with open (\"outputText.txt\", \"r\") as textFile:\n",
        "  textData=textFile.readlines() \n",
        "textChunks=pre_process_text(text_data)\n",
        "summary=summarize(textChunks)\n",
        "print(summary)"
      ]
    }
  ],
  "metadata": {
    "colab": {
      "collapsed_sections": [],
      "name": "Podcast summarizer.ipynb",
      "provenance": [],
      "authorship_tag": "ABX9TyPdFXfayALE7OyXKXGL+pJL",
      "include_colab_link": true
    },
    "kernelspec": {
      "display_name": "Python 3",
      "name": "python3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}