Update streamlit_app.py

streamlit_app.py

@@ -1,4 +1,6 @@
 # streamlit_app.py
+import requests          # for direct‑video download
+import snscrape.modules.twitter as sntwitter  # Twitter scraper
 import base64
 import hashlib
 import os
@@ -93,14 +95,72 @@ def compress_video(inp: str, out: str, crf: int = 28, preset: str = "fast") -> s
 
 
 def download_video(url: str, dst_dir: str, password: str = "") -> str:
+    """
+    Download a video from *url*.
+    - If the URL ends with a known video extension → plain HTTP download.
+    - If it is a YouTube / archive URL → try yt‑dlp (will raise a clear error if DNS fails).
+    - If it is a Twitter status URL → scrape the tweet for the first video URL.
+    """
+    video_exts = (".mp4", ".mov", ".webm", ".mkv", ".avi", ".flv")
+
+    # --------------------------------------------------------------
+    # 1️⃣ Direct video file (no yt‑dlp needed)
+    # --------------------------------------------------------------
+    if url.lower().endswith(video_exts):
+        try:
+            r = requests.get(url, stream=True, timeout=30)
+            r.raise_for_status()
+            filename = sanitize_filename(url.split("/")[-1])
+            out_path = Path(dst_dir) / filename
+            with open(out_path, "wb") as f:
+                for chunk in r.iter_content(chunk_size=8192):
+                    if chunk:
+                        f.write(chunk)
+            return str(out_path)          # already an MP4 (or other supported format)
+        except Exception as e:
+            raise RuntimeError(f"Direct download failed: {e}")
+
+    # --------------------------------------------------------------
+    # 2️⃣ Twitter status – scrape for video URL
+    # --------------------------------------------------------------
+    if "twitter.com" in url and "/status/" in url:
+        try:
+            tweet_id = url.split("/")[-1].split("?")[0]
+            for tweet in sntwitter.TwitterTweetScraper(tweet_id).get_items():
+                # Look for media URLs in the tweet JSON
+                media = tweet.media
+                if media:
+                    for m in media:
+                        if getattr(m, "video_url", None):
+                            return download_video(m.video_url, dst_dir)  # recurse as a direct URL
+                # If no media field, fall back to expanded URLs
+                for u in tweet.urls:
+                    if u.expandedUrl.lower().endswith(video_exts):
+                        return download_video(u.expandedUrl, dst_dir)
+            raise RuntimeError("No video found in the tweet.")
+        except Exception as e:
+            raise RuntimeError(f"Twitter scrape failed: {e}")
+
+    # --------------------------------------------------------------
+    # 3️⃣ Fallback to yt‑dlp (YouTube, archive.org, etc.)
+    # --------------------------------------------------------------
     tmpl = str(Path(dst_dir) / "%(id)s.%(ext)s")
     opts = {"outtmpl": tmpl, "format": "best"}
     if password:
         opts["videopassword"] = password
-    with yt_dlp.YoutubeDL(opts) as ydl:
-        info = ydl.extract_info(url, download=True)
 
-    # Try to locate the downloaded file
+    try:
+        with yt_dlp.YoutubeDL(opts) as ydl:
+            info = ydl.extract_info(url, download=True)
+    except Exception as e:
+        # Re‑raise with a short, user‑friendly message
+        raise RuntimeError(
+            f"yt‑dlp could not download the URL. "
+            f"Common reasons: DNS failure, unsupported site, or missing video. "
+            f"Original error: {e}"
+        )
+
+    # Locate the downloaded file (same logic as before)
     if isinstance(info, dict) and info.get("id"):
         vid_id = info["id"]
         ext = info.get("ext", "mp4")
@@ -214,6 +274,27 @@ if st.sidebar.button("Load Video"):
         st.session_state["last_error"] = f"Download failed: {e}"
         st.sidebar.error(st.session_state["last_error"])
 
+st.sidebar.markdown("---")
+st.sidebar.subheader("🔎 Extract video from a Tweet")
+tweet_url = st.sidebar.text_input(
+    "Tweet URL (e.g. https://twitter.com/user/status/1234567890)",
+    key="tweet_url",
+)
+
+if st.sidebar.button("Fetch video from tweet"):
+    if not tweet_url:
+        st.sidebar.error("Please paste a tweet URL.")
+    else:
+        try:
+            # The same download_video function will recognise the tweet URL
+            path = download_video(tweet_url, str(DATA_DIR))
+            st.session_state["videos"] = path
+            st.session_state["last_error"] = ""
+            st.sidebar.success("Video extracted from tweet and loaded.")
+        except Exception as e:
+            st.session_state["last_error"] = f"Tweet extraction failed: {e}"
+            st.sidebar.error(st.session_state["last_error"])
+
 if st.session_state["videos"]:
     try:
         mp4 = convert_to_mp4(st.session_state["videos"])