Update app.py

app.py

@@ -6,200 +6,154 @@ import subprocess
 import json
 import re
 from bs4 import BeautifulSoup
-from urllib.parse import urlparse
-import tempfile
-from pyvis.network import Network
-from functools import lru_cache
-from concurrent.futures import ThreadPoolExecutor
-
-THEME = "gradio/soft"
-
-# --- Shared HTTP session for speed & consistent headers ---
-session = requests.Session()
-session.headers.update({
-    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"
-})
-
-@lru_cache(maxsize=64)
-def get_cached_item(identifier):
-    """Cache IA get_item() calls to avoid duplicate downloads."""
-    return get_item(identifier)
 
+# --- VirusTotal helper functions ---
 def scan_url_vt(url, api_key):
-    """Return True if VirusTotal sees no malicious hits."""
     headers = {"x-apikey": api_key}
-    resp = session.post(
-        "https://www.virustotal.com/api/v3/urls",
-        headers=headers,
-        data={"url": url}
+    resp = requests.post(
+        "https://www.virustotal.com/api/v3/urls", headers=headers, data={"url": url}
     )
     resp.raise_for_status()
     analysis_id = resp.json()["data"]["id"]
-
-    # Poll every second until complete
     while True:
-        time.sleep(1)
-        st = session.get(
-            f"https://www.virustotal.com/api/v3/analyses/{analysis_id}",
-            headers=headers
+        time.sleep(5)
+        status_resp = requests.get(
+            f"https://www.virustotal.com/api/v3/analyses/{analysis_id}", headers=headers
         )
-        st.raise_for_status()
-        attr = st.json()["data"]["attributes"]
+        status_resp.raise_for_status()
+        attr = status_resp.json()["data"]["attributes"]
         if attr.get("status") == "completed":
-            return attr.get("stats", {}).get("malicious", 0) == 0
+            stats = attr.get("stats", {})
+            return stats.get("malicious", 0) == 0
 
-def extract_ffprobe_metadata(path):
-    """Run ffprobe and parse its JSON output."""
+# --- FFprobe metadata extraction ---
+def extract_ffprobe_metadata(url_or_path):
     cmd = [
         "ffprobe", "-v", "error", "-print_format", "json",
-        "-show_format", "-show_streams", path
+        "-show_format", "-show_streams",
+        url_or_path
     ]
     out = subprocess.check_output(cmd)
     return json.loads(out)
 
+# --- Scrape basic page metadata (title + og: tags) ---
 def fetch_page_metadata(url):
-    """Grab <title>, og: and twitter: meta tags from any page."""
     try:
-        r = session.get(url, timeout=5)
-        r.raise_for_status()
-        soup = BeautifulSoup(r.text, "html.parser")
-        data = {
-            "url": url,
-            "title": getattr(soup.title, "string", None)
-        }
+        resp = requests.get(url, timeout=5)
+        resp.raise_for_status()
+        html = resp.text
+        soup = BeautifulSoup(html, "html.parser")
+        meta = {"url": url, "title": soup.title.string if soup.title else None}
+        # grab OpenGraph tags
         for tag in soup.find_all("meta"):
             prop = tag.get("property") or tag.get("name")
             if prop and prop.startswith(("og:", "twitter:")):
-                data[prop] = tag.get("content")
-        return data
+                meta[prop] = tag.get("content")
+        return meta
     except Exception as e:
         return {"url": url, "error": str(e)}
 
+# --- Core search & scan logic ---
 def fetch_clean_videos(keywords, api_key, scan_enabled):
-    """Search IA, filter by file extension, optional parallel VT scan."""
-    terms = [kw.strip().replace(" ", "+") for kw in keywords.split(",")]
-    ia_query = f"mediatype:(movies) AND ({' OR '.join(terms)})"
-    results = list(search_items(ia_query))[:10]  # top 10 to stay speedy
-
-    # collect candidate video URLs
-    candidates = []
-    for r in results:
-        ident = r["identifier"]
-        item = get_cached_item(ident)
+    query = " OR ".join([f"{kw.strip().replace(' ', '+')}" for kw in keywords.split(",")])
+    ia_query = f"mediatype:(movies) AND ({query})"
+    results = list(search_items(ia_query))[:50]
+
+    clean_urls = []
+    for res in results:
+        identifier = res["identifier"]
+        item = get_item(identifier)
         for f in item.files:
-            name = f.get("name", "")
-            ext  = name.lower().split(".")[-1]
-            if ext in ("mp4", "mkv", "avi", "mov", "webm", "m4v"):
-                candidates.append(f"https://archive.org/download/{ident}/{name}")
-
-    # optionally scan in parallel
-    if scan_enabled and api_key:
-        with ThreadPoolExecutor(max_workers=5) as exe:
-            safe_urls = exe.map(lambda u: u if scan_url_vt(u, api_key) else None, candidates)
-        return [u for u in safe_urls if u]
-    return candidates
-
-def get_favicon_url(page_url):
-    """Assume https://domain/favicon.ico exists."""
-    dom = urlparse(page_url).netloc
-    return f"https://{dom}/favicon.ico"
-
-def build_origin_graph(origins):
-    """Render a PyVis network; return its full HTML."""
-    net = Network(height="400px", width="100%", directed=True)
-    for i, m in enumerate(origins):
-        net.add_node(
-            i,
-            label=urlparse(m["url"]).netloc,
-            title=json.dumps(m, indent=2),
-            shape="image",
-            image=get_favicon_url(m["url"])
-        )
-        if i > 0:
-            net.add_edge(i - 1, i)
-    tmpfile = tempfile.NamedTemporaryFile(delete=False, suffix=".html").name
-    net.save_graph(tmpfile)
-    return open(tmpfile, encoding="utf-8").read()
-
-with gr.Blocks(theme=THEME) as demo:
-    gr.Markdown("## 📼 IA Drone-Strike Explorer")
-
+            fmt = f.get("format", "").lower()
+            if fmt.startswith(("mpeg","mp4","avi","mov","webm","m4v")):
+                url = f"https://archive.org/download/{identifier}/{f['name']}"
+                if scan_enabled and api_key:
+                    try:
+                        is_clean = scan_url_vt(url, api_key)
+                    except Exception:
+                        continue
+                else:
+                    is_clean = True
+                if is_clean:
+                    clean_urls.append(url)
+    return clean_urls
+
+# --- Gradio UI setup ---
+with gr.Blocks() as demo:
+    gr.Markdown("# 📼 IA Drone‑Strike Explorer  \nEnable VT scan, FFprobe & Origin Tracing")
     with gr.Row():
-        kw     = gr.Textbox(label="Search keywords", value="drone strike, military uav")
-        vt_key = gr.Textbox(label="VirusTotal API Key", type="password")
-    scan = gr.Checkbox(label="Enable VT scan", value=True)
-    ff   = gr.Checkbox(label="Enable FFprobe metadata", value=False)
-    btn  = gr.Button("Search & Scan")
-
-    dropdown     = gr.Dropdown(label="Clean Video URLs", choices=[])
-    graph_html   = gr.HTML("<p>No origin graph yet.</p>")
-    video_player = gr.Video()
-
-    with gr.Tabs():
-        with gr.TabItem("IA Metadata"):
-            ia_json = gr.JSON()
-        with gr.TabItem("FFprobe"):
-            ff_json = gr.JSON()
-        with gr.TabItem("Origins"):
-            graph_html
-            orig_json = gr.JSON()
+        kw_input     = gr.Textbox(label="Search keywords", value="drone strike, military uav")
+        vt_key_input = gr.Textbox(label="VirusTotal API Key", type="password")
+    scan_toggle    = gr.Checkbox(label="Enable VT scan", value=True)
+    ffprobe_toggle = gr.Checkbox(label="Enable FFprobe metadata", value=False)
+    run_btn        = gr.Button("Search & Scan")
+
+    url_dropdown   = gr.Dropdown(label="Clean Video URLs", choices=[], interactive=True)
+    video_player   = gr.Video(label="Video Player")
+    ia_meta_json   = gr.JSON(label="► Raw IA Metadata")
+    ffprobe_json   = gr.JSON(label="► FFprobe Metadata")
+    origins_json   = gr.JSON(label="► Source‑Origin Metadata")
 
     def search_and_populate(keywords, api_key, scan_enabled):
         urls = fetch_clean_videos(keywords, api_key, scan_enabled)
         return gr.update(choices=urls, value=urls[0] if urls else None)
 
-    def update_all(url_sel, ff_on, api_key):
-        if not url_sel:
-            return None, {}, {}, "<p>No origin graph.</p>", []
+    def update_all(selected_url, ff_on, api_key):
+        # no selection guard
+        if not selected_url:
+            return None, {}, {}, []
 
-        parts    = url_sel.split("/")
-        ident, fn = parts[4], parts[-1]
-        ia_data  = {"identifier": ident, "file": {}}
-
-        # — Fix metadata merge
-        try:
-            item = get_cached_item(ident)
-            file_rec = next((f for f in item.files if f["name"] == fn), None)
-            if file_rec:
-                ia_data["file"] = {**item.metadata, **file_rec}
-            else:
-                ia_data["file"] = {}
-        except Exception as e:
-            ia_data["error"] = str(e)
-
-        # — FFprobe
-        ff_data = {}
+        # 1) IA metadata + file list
+        parts = selected_url.split("/")
+        identifier = parts[4] if len(parts) > 4 else None
+        raw_ia = {"identifier": identifier, "metadata": {}, "files": []}
+        if identifier:
+            try:
+                item = get_item(identifier)
+                raw_ia["metadata"] = item.metadata
+                raw_ia["files"] = [
+                    {
+                        "name": f.get("name"),
+                        "format": f.get("format"),
+                        "size": f.get("size"),
+                        "md5": f.get("md5"),
+                        **{k: v for k,v in f.items() if k not in ("name","format","size","md5")}
+                    }
+                    for f in item.files
+                ]
+            except Exception:
+                raw_ia["error"] = "could not fetch IA metadata"
+
+        # 2) FFprobe metadata if toggled
+        ff_md = {}
         if ff_on:
             try:
-                ff_data = extract_ffprobe_metadata(url_sel)
+                ff_md = extract_ffprobe_metadata(selected_url)
             except Exception as e:
-                ff_data = {"error": str(e)}
+                ff_md = {"error": str(e)}
 
-        # — Origin tracing
-        desc  = (item.metadata.get("description") or "")
-        found = re.findall(r"https?://[^\s\"'<]+", desc)
-        if found:
-            with ThreadPoolExecutor(max_workers=5) as exe:
-                origins = list(exe.map(fetch_page_metadata, found))
-            graph   = build_origin_graph(origins)
-        else:
-            origins, graph = [], "<p>No origins.</p>"
+        # 3) Origin tracing: scrape each URL in description
+        origins = []
+        desc = raw_ia["metadata"].get("description", "")
+        urls_found = re.findall(r'https?://[^\s"<]+', desc)
+        for url in urls_found:
+            meta = fetch_page_metadata(url)
+            origins.append(meta)
+            # stop at first “real” origin (you can remove this break to collect all)
+            break
 
-        return url_sel, ia_data, ff_data, graph, origins
+        return selected_url, raw_ia, ff_md, origins
 
-    btn.click(
+    run_btn.click(
         fn=search_and_populate,
-        inputs=[kw, vt_key, scan],
-        outputs=[dropdown]
+        inputs=[kw_input, vt_key_input, scan_toggle],
+        outputs=[url_dropdown]
     )
-    dropdown.change(
+    url_dropdown.change(
         fn=update_all,
-        inputs=[dropdown, ff, vt_key],
-        outputs=[video_player, ia_json, ff_json, graph_html, orig_json]
+        inputs=[url_dropdown, ffprobe_toggle, vt_key_input],
+        outputs=[video_player, ia_meta_json, ffprobe_json, origins_json]
     )
 
-    # <<< enable async queue & spinner >>>
-    demo.queue()
-
 if __name__ == "__main__":
     demo.launch()