Update app.py

app.py

@@ -20,9 +20,13 @@ from speechbrain.inference.classifiers import EncoderClassifier
 from pydub.silence import split_on_silence
 from moviepy.editor import VideoFileClip
 import re
+from moviepy.editor import AudioFileClip
+import subprocess
+
 nltk.download('punkt')
 nltk.download('stopwords')
 
+
 class VideoAnalytics:
     """
     Class for performing analytics on videos including transcription, summarization, topic generation,
@@ -165,33 +169,33 @@ class VideoAnalytics:
             logging.error(f"Error processing audio: {e}")
             return ""
 
-    def transcribe_video(self, vid: str) -> str:
+    def transcribe_video(self, audio_path: str) -> str:
       """
       Transcribe the audio of the video.
       Args:
-          vid (str): Path to the video file.
+          audio_path (str): Path to the audio file.
       Returns:
           str: Transcribed text.
       """
       try:
-          # Load the video file and extract audio
-          video = VideoFileClip(vid)
-          audio = video.audio
+          # # 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")
+          # # Write audio to a temporary file
+          # audio.write_audiofile("output_audio.mp3")
 
 
-          # Replace 'input.mp3' and 'output.wav' with your file paths
-          audio_filename = self.mp3_to_wav("output_audio.mp3", 'output.wav')
-          segments = self.split_audio(audio_filename)
+          # # Replace 'input.mp3' and 'output.wav' with your file paths
+          # audio_filename = self.mp3_to_wav("output_audio.mp3", 'output.wav')
+          segments = self.split_audio(audio_path)
           splitted_audio_filename = segments[0].export("segment_for_1_min.wav",format="wav")
 
           # for detect lang
           signal = self.language_id.load_audio(splitted_audio_filename.name)
           prediction =  self.language_id.classify_batch(signal)
           lang = [prediction[3][0].split(":")][0][0]
-          text  = self.get_large_audio_transcription_on_silence(audio_filename,lang)
+          text  = self.get_large_audio_transcription_on_silence(audio_path,lang)
           # Update the transcribed_text attribute with the transcription result
           self.transcribed_text = text
           # Update the translation text into english_text
@@ -514,34 +518,31 @@ class VideoAnalytics:
         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.
-        """
+    def download_youtube_audio(self,url, output_filename="audio.wav"):
         try:
-          # Initialize YouTube object with the provided link
-          youtubeObject = YouTube(link)
+            # Step 1: Download the audio using yt-dlp
+            audio_filename = "downloaded_audio"
+            subprocess.run([
+                "yt-dlp",
+                "-x", "--audio-format", "mp3",  # Extract audio in mp3 format
+                "-o", f"{audio_filename}.%(ext)s",  # Save the audio with the specified name
+                url
+            ], check=True)
 
-          # 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")
+            # Step 2: Convert the downloaded audio file to .wav format
+            audio = AudioSegment.from_file(f"{audio_filename}.mp3", format="mp3")
+            output_path = f"{output_filename}"
 
-          logging.info("Download is completed successfully")
+            audio.export(output_path, format="wav")
+            print(f"Audio downloaded and saved as {output_path}")
+
+            # Step 3: Cleanup - remove the original downloaded file if needed
+            os.remove(f"{audio_filename}.mp3")
+
+            return output_path
 
         except Exception as e:
-            # Log any errors that occur during initialization of YouTube object
-            logging.error(f"Error downloading video: {e}")
-            return ""
+            print(f"Error: {e}")
 
     def save_audio_with_gtts(self, text: str, filename: str) -> str:
         """
@@ -579,23 +580,23 @@ class VideoAnalytics:
         try:
             # Download the video if input_path is provided, otherwise use the provided video path
           if input_path:
-            input_path = self.Download(input_path)
-            video_ = VideoFileClip(input_path)
-            duration = video_.duration
-            video_.close()
-            if round(duration) <= 6*600:
-              text = self.transcribe_video(input_path)
-            else:
-              return "Video Duration Above 10 Minutes,Try Below 10 Minutes Video","","",None,None,None
+            input_path = self.download_youtube_audio(input_path)
+            audio_ = AudioFileClip(input_path)
+            duration = audio_.duration
+            audio_.close()
+            # if round(duration) <= 6*600:
+            text = self.transcribe_video(input_path)
+            # else:
+            #   return "Video Duration Above 10 Minutes,Try Below 10 Minutes Video","","",None,None,None
           elif video:
             video_ = VideoFileClip(video)
             duration = video_.duration
             video_.close()
-            if round(duration) <= 6*600:
-              text = self.transcribe_video(video)
-              input_path = video
-            else:
-              return "Video Duration Above 10 Minutes,Try Below 10 Minutes Video","","",None,None,None
+            # if round(duration) <= 6*600:
+            text = self.transcribe_video(video)
+            input_path = video
+            # else:
+            #   return "Video Duration Above 10 Minutes,Try Below 10 Minutes Video","","",None,None,None
           overall_summary = ""
           # Generate summary, important sentences, and topics
           summary = self.generate_video_summary(model)
@@ -661,7 +662,7 @@ class VideoAnalytics:
                   result = gr.Textbox(label='Answer',lines=10)
               submit_btn.click(self.main,[video,yt_link,model_selection],[summary,Important_Sentences,Topics,summary_audio,important_sentence_audio,topics_audio])
               question.submit(self.video_qa,[question,model],result)
-        demo.launch(debug=True)
+        demo.launch()
 
 if __name__ == "__main__":
   video_analytics = VideoAnalytics()