Update app.py

app.py

@@ -1,3 +1,4 @@
+2:
 import os
 import re
 import torch
@@ -10,55 +11,80 @@ import shutil
 from torch.utils.data import Dataset, DataLoader
 from youtube_transcript_api import YouTubeTranscriptApi, TranscriptsDisabled, NoTranscriptFound, CouldNotRetrieveTranscript, VideoUnavailable
 from youtube_transcript_api.formatters import TextFormatter
-from transformers import pipeline, WhisperProcessor, WhisperForConditionalGeneration
+from transformers import (
+    pipeline,
+    WhisperProcessor,
+    WhisperForConditionalGeneration,
+)
+
+# === UTILS ===
+
+def is_youtube_url(url):
+    return "youtube.com" in url or "youtu.be" in url
+
+def is_web_url(url):
+    return url.startswith("http://") or url.startswith("https://")
 
-# === Functions from Code 1 ===
 def get_video_id(url):
     match = re.search(r'(?:v=|\/)([0-9A-Za-z_-]{11})', url)
     return match.group(1) if match else None
 
-def download_transcript(video_id, lang="en"):
+def try_download_transcript(video_id):
     try:
-        transcript = YouTubeTranscriptApi.get_transcript(video_id, languages=[lang])
-        return TextFormatter().format_transcript(transcript)
-    except (TranscriptsDisabled, NoTranscriptFound, VideoUnavailable, CouldNotRetrieveTranscript):
+        transcript = YouTubeTranscriptApi.get_transcript(video_id, languages=["en"])
+        formatted = TextFormatter().format_transcript(transcript)
+        return formatted
+    except (TranscriptsDisabled, NoTranscriptFound, CouldNotRetrieveTranscript, VideoUnavailable):
         return None
     except Exception as e:
         print(f"Transcript error: {e}")
         return None
 
-def download_audio(url, sabr_only=False):
-    try:
-        ydl_opts = {
-            'format': 'bestaudio[asr>0]/bestaudio/best' if sabr_only else 'bestaudio/best',
-            'outtmpl': 'temp_audio.%(ext)s',
-            'postprocessors': [{
-                'key': 'FFmpegExtractAudio',
-                'preferredcodec': 'wav',
-            }],
-        }
-        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
-            ydl.download([url])
-        return 'temp_audio.wav'
-    except Exception as e:
-        print(f"Audio download error: {e}")
-        return None
+def download_audio_youtube(url, output_path="audio.wav", cookies_path=None):
+    import subprocess
+
+    fallback_video_path = "fallback_video.mp4"
+
+    ydl_opts = {
+        "format": "best",
+        "outtmpl": fallback_video_path,
+        "user_agent": "com.google.android.youtube/17.31.35 (Linux; U; Android 11)",
+        "compat_opts": ["allow_unplayable_formats"]
+    }
+
+    if cookies_path:
+        ydl_opts["cookiefile"] = cookies_path
 
-def download_video(url, sabr_only=False):
     try:
-        ydl_opts = {
-            'format': 'bestvideo+bestaudio/best' if sabr_only else 'best',
-            'outtmpl': 'temp_video.%(ext)s',
-            'merge_output_format': 'mp4',
-        }
         with yt_dlp.YoutubeDL(ydl_opts) as ydl:
             ydl.download([url])
-        return 'temp_video.mp4'
     except Exception as e:
-        print(f"Video download error: {e}")
-        return None
+        # On failure, run yt-dlp in subprocess to list formats
+        try:
+            list_cmd = ["yt-dlp", "-F", url]
+            if cookies_path:
+                list_cmd += ["--cookies", cookies_path]
+            result = subprocess.run(list_cmd, capture_output=True, text=True, timeout=15)
+            formats = result.stdout or "No formats found."
+        except Exception as format_err:
+            formats = f"⚠️ Could not list formats due to: {format_err}"
+
+        raise RuntimeError(
+            f"yt-dlp failed: {e}\n\n"
+            f"Available formats for this video:\n\n{formats}"
+        )
+
+    return extract_audio_from_video(fallback_video_path, audio_path=output_path)
+
+def download_video_direct(url, output_path="video.mp4"):
+    ydl_opts = {
+        "format": "best",
+        "outtmpl": output_path
+    }
+    with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+        ydl.download([url])
+    return output_path
 
-# === Utilities ===
 def extract_audio_from_video(video_path, audio_path="audio.wav"):
     ffmpeg.input(video_path).output(audio_path, ac=1, ar=16000).run(overwrite_output=True)
     return audio_path
@@ -147,34 +173,38 @@ def generate_questions_with_pipeline(text, num_questions=5):
 
     return questions[:num_questions]
 
-# === Main Processing ===
-def process_input_gradio(url_input, file_input, transcript_text):
+# === MAIN FUNCTION ===
+
+def process_input_gradio(url_input, file_input, cookies_file):
     try:
-        if transcript_text:
-            transcript = transcript_text
+        cookies_path = None
+        if cookies_file is not None:
+            cookies_path = "cookies.txt"
+            shutil.copyfile(cookies_file.name, cookies_path)
 
-        elif file_input is not None:
+        if file_input is not None:
             audio_path = extract_audio_from_video(file_input.name)
             chunks, sr = split_audio(audio_path, chunk_length_sec=15)
             transcript = transcribe_chunks_dataset(chunks, sr)
 
         elif url_input:
-            video_id = get_video_id(url_input)
-            transcript = download_transcript(video_id)
-
-            if not transcript:
-                audio_path = download_audio(url_input)
-                if not audio_path:
-                    video_path = download_video(url_input)
-                    if not video_path:
-                        return "❌ Failed to download video/audio/transcript.", ""
-                    audio_path = extract_audio_from_video(video_path)
-
+            if is_youtube_url(url_input):
+                video_id = get_video_id(url_input)
+                transcript = try_download_transcript(video_id)
+                if not transcript:
+                    try:
+                        audio_path = download_audio_youtube(url_input, cookies_path=cookies_path)
+                        chunks, sr = split_audio(audio_path, chunk_length_sec=15)
+                        transcript = transcribe_chunks_dataset(chunks, sr)
+                    except Exception as e:
+                        return f"⚠️ Could not download this YouTube video due to restrictions. Please upload the video manually.\nDetails: {e}", ""
+            else:
+                video_file = download_video_direct(url_input)
+                audio_path = extract_audio_from_video(video_file)
                 chunks, sr = split_audio(audio_path, chunk_length_sec=15)
                 transcript = transcribe_chunks_dataset(chunks, sr)
-
         else:
-            return "Please provide a URL, upload a file, or paste transcript text.", ""
+            return "Please provide a URL or upload a video file.", ""
 
         summary = summarize_with_bart(transcript)
         questions = generate_questions_with_pipeline(summary)
@@ -182,20 +212,21 @@ def process_input_gradio(url_input, file_input, transcript_text):
     except Exception as e:
         return f"Error: {str(e)}", ""
 
-# === Gradio UI ===
+# === GRADIO UI ===
+
 iface = gr.Interface(
     fn=process_input_gradio,
     inputs=[
         gr.Textbox(label="YouTube or Direct Video URL", placeholder="https://..."),
         gr.File(label="Or Upload a Video File", file_types=[".mp4", ".mkv", ".webm"]),
-        gr.Textbox(label="Or Paste Transcript Text Directly", lines=10, placeholder="Paste full transcript here..."),
+        gr.File(label="Optional cookies.txt for YouTube", file_types=[".txt"]),
     ],
     outputs=[
         gr.Textbox(label="Summary", lines=10),
         gr.Textbox(label="Generated Questions", lines=10),
     ],
     title="Lecture Summary & Question Generator",
-    description="Provide a YouTube/Direct video URL, upload a video file, or paste a transcript."
+    description="Provide a YouTube/Direct video URL or upload a video file. If the video is restricted, upload cookies.txt or the video file directly."
 )
 
-iface.launch()
+iface.launch()