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("""

Video Analytics

""") 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()