Create app.py

app.py

@@ -0,0 +1,297 @@
+from openai import AzureOpenAI
+import os
+import ffmpeg
+from typing import List
+from moviepy.editor import VideoFileClip
+import nltk
+from sklearn.feature_extraction.text import TfidfVectorizer
+import gradio as gr
+from pytube import YouTube
+import requests
+import logging
+
+nltk.download('punkt')
+nltk.download('stopwords')
+
+
+class VideoAnalytics:
+    """
+    Class for performing analytics on videos including transcription, summarization, topic generation,
+    and extraction of important sentences.
+    """
+
+    def __init__(self):
+      """
+      Initialize the VideoAnalytics object.
+
+      Args:
+          hf_token (str): Hugging Face API token.
+      """
+      # Initialize AzureOpenAI client
+      self.client = AzureOpenAI()
+
+      # Initialize transcribed text variable
+      self.transcribed_text = ""
+
+      # API URL for accessing the Hugging Face model
+      self.API_URL = "https://api-inference.huggingface.co/models/openai/whisper-large-v3"
+
+      # Placeholder for Hugging Face API token
+      hf_token = os.get_environ("HF_TOKEN")  # Replace this with the actual Hugging Face API token
+
+      # Set headers for API requests with Hugging Face token
+      self.headers = {"Authorization": f"Bearer {hf_token}"}
+
+      # Configure logging settings
+      logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+
+    def transcribe_video(self, vid: str) -> str:
+      """
+      Transcribe the audio of the video.
+
+      Args:
+          vid (str): Path to the video file.
+
+      Returns:
+          str: Transcribed text.
+      """
+      try:
+          # Load the video file and extract audio
+          video = VideoFileClip(vid)
+          audio = video.audio
+
+          # Write audio to a temporary file
+          audio.write_audiofile("output_audio.mp3")
+          audio_file = open("output_audio.mp3", "rb")
+
+          # Define a helper function to query the Hugging Face model
+          def query(data):
+              response = requests.post(self.API_URL, headers=self.headers, data=data)
+              return response.json()
+
+          # Send audio data to the Hugging Face model for transcription
+          output = query(audio_file)
+          # Update the transcribed_text attribute with the transcription result
+          self.transcribed_text = output["text"]
+          # Return the transcribed text
+          return output["text"]
+
+      except Exception as e:
+          logging.error(f"Error transcribing video: {e}")
+          return ""
+
+    def generate_video_summary(self) -> str:
+        """
+        Generate a summary of the transcribed video.
+
+        Returns:
+            str: Generated summary.
+        """
+        try:
+          # Define a conversation between system and user
+          conversation = [
+            {"role": "system", "content": "You are a Summarizer"},
+            {"role": "user", "content": f"""summarize the following text delimited by triple backticks.
+                      In two format of Outputs given below:
+                      Abstractive Summary:
+                      Extractive Summary:
+                      ```{self.transcribed_text}```
+              """}
+              ]
+          # Generate completion using ChatGPT model
+          response = self.client.chat.completions.create(
+              model="ChatGPT",
+              messages=conversation,
+              temperature=0,
+              max_tokens=1000
+          )
+          # Get the generated summary message
+          message = response.choices[0].message.content
+          return message
+        except Exception as e:
+            logging.error(f"Error generating video summary: {e}")
+            return ""
+
+
+    def generate_topics(self) -> str:
+        """
+        Generate topics from the transcribed video.
+
+        Returns:
+            str: Generated topics.
+        """
+        try:
+          # Define a conversation between system and user
+          conversation = [
+            {"role": "system", "content": "You are a Topic Generator"},
+            {"role": "user", "content": f"""generate single Topics from the following text don't make sentence for topic generation,delimited by triple backticks.
+                      list out the topics:
+                      Topics:
+                      ```{self.transcribed_text}```
+              """}
+              ]
+          # Generate completion using ChatGPT model
+          response = self.client.chat.completions.create(
+              model="ChatGPT",
+              messages=conversation,
+              temperature=0,
+              max_tokens=1000
+          )
+          # Get the generated topics message
+          message = response.choices[0].message.content
+          return message
+        except Exception as e:
+            logging.error(f"Error generating topics: {e}")
+            return ""
+
+    def extract_video_important_sentence(self) -> str:
+        """
+        Extract important sentences from the transcribed video.
+
+        Returns:
+            str: Extracted important sentences.
+        """
+        try:
+          # Tokenize the sentences
+          sentences = nltk.sent_tokenize(self.transcribed_text)
+
+          # Initialize TF-IDF vectorizer
+          tfidf_vectorizer = TfidfVectorizer()
+
+          # Fit the vectorizer on the summary sentences
+          tfidf_matrix = tfidf_vectorizer.fit_transform(sentences)
+
+          # Calculate sentence scores based on TF-IDF values
+          sentence_scores = tfidf_matrix.sum(axis=1)
+
+          # Create a list of (score, sentence) tuples
+          sentence_rankings = [(score, sentence) for score, sentence in zip(sentence_scores, sentences)]
+
+          # Sort sentences by score in descending order
+          sentence_rankings.sort(reverse=True)
+
+          # Set a threshold for selecting sentences
+          threshold = 2 # Adjust as needed
+
+          # Select sentences with scores above the threshold
+          selected_sentences = [sentence for score, sentence in sentence_rankings if score >= threshold]
+
+          # Join selected sentences to form the summary
+          summary = '\n\n'.join(selected_sentences)
+
+          return summary
+
+        except Exception as e:
+            logging.error(f"Error extracting important sentences: {e}")
+            return ""
+
+    def write_text_files(self, text: str, filename: str) -> None:
+        """
+        Write text to a file.
+
+        Args:
+            text (str): Text to be written to the file.
+            filename (str): Name of the file.
+        """
+        try:
+          file_path = f"{filename}.txt"
+          with open(file_path, 'w') as file:
+              # Write content to the file
+              file.write(text)
+        except Exception as e:
+            logging.error(f"Error writing text to file: {e}")
+
+    def Download(self, link: str) -> str:
+        """
+        Download a video from YouTube.
+
+        Args:
+            link (str): YouTube video link.
+
+        Returns:
+            str: Path to the downloaded video file.
+        """
+        try:
+          # Initialize YouTube object with the provided link
+          youtubeObject = YouTube(link)
+
+          # Get the highest resolution stream
+          youtubeObject = youtubeObject.streams.get_highest_resolution()
+          try:
+              # Attempt to download the video
+              file_name = youtubeObject.download()
+              return file_name
+          except:
+              # Log any errors that occur during video download
+              logging.info("An error has occurred")
+
+          logging.info("Download is completed successfully")
+
+        except Exception as e:
+            # Log any errors that occur during initialization of YouTube object
+            logging.error(f"Error downloading video: {e}")
+            return ""
+
+
+    def main(self, video: str = None, input_path: str = None) -> tuple:
+        """
+        Perform video analytics.
+
+        Args:
+            video (str): Path to the video file.
+            input_path (str): Input path for the video.
+
+        Returns:
+            tuple: Summary, important sentences, and topics.
+        """
+        try:
+          # Download the video if input_path is provided, otherwise use the provided video path
+          if input_path:
+            input_path = self.Download(input_path)
+            text = self.transcribe_video(input_path)
+          elif video:
+            text = self.transcribe_video(video)
+            input_path = video
+
+          # Generate summary, important sentences, and topics
+          summary = self.generate_video_summary()
+          self.write_text_files(summary,"Summary")
+          important_sentences = self.extract_video_important_sentence()
+          self.write_text_files(important_sentences,"Important_Sentence")
+          topics = self.generate_topics()
+          self.write_text_files(topics,"Topics")
+
+          # Return the generated summary, important sentences, and topics
+          return summary,important_sentences,topics
+
+        except Exception as e:
+            # Log any errors that occur during video analytics
+            logging.error(f"Error in main function: {e}")
+            return "", "", ""
+
+    def gradio_interface(self):
+        with gr.Blocks(css="style.css",theme=gr.themes.Soft()) as demo:
+            gr.HTML("""<center><h1>Video Analytics</h1></center>""")
+            with gr.Row():
+              yt_link = gr.Textbox(label= "Youtube Link",placeholder="https://www.youtube.com/watch?v=")
+            with gr.Row():
+              video = gr.Video(sources="upload",height=200,width=300)
+            with gr.Row():
+              submit_btn = gr.Button(value="Submit")
+            with gr.Tab("Summary"):
+              with gr.Row():
+                  summary = gr.Textbox(show_label=False,lines=10)
+              with gr.Row():
+                  summary_download = gr.DownloadButton(label="Download",value="Summary.txt",visible=True,size='lg',elem_classes="download_button")
+            with gr.Tab("Important Sentences"):
+              with gr.Row():
+                  Important_Sentences = gr.Textbox(show_label=False,lines=10)
+              with gr.Row():
+                  sentence_download = gr.DownloadButton(label="Download",value="Important_Sentence.txt",visible=True,size='lg',elem_classes="download_button")
+            with gr.Tab("Topics"):
+              with gr.Row():
+                  Topics = gr.Textbox(show_label=False,lines=10)
+              with gr.Row():
+                  topics_download = gr.DownloadButton(label="Download",value="Topics.txt",visible=True,size='lg',elem_classes="download_button")
+              submit_btn.click(self.main,[video,yt_link],[summary,Important_Sentences,Topics])
+        demo.launch()