Update app.py

app.py

@@ -1,183 +1,152 @@
 import gradio as gr
 from internetarchive import search_items, get_item
-import requests
-import time
-import subprocess
-import json
-import re
-import concurrent.futures
+import requests, time, subprocess, json, re, tempfile, os
 from bs4 import BeautifulSoup
-from functools import lru_cache
+import networkx as nx
 from pyvis.network import Network
-from urllib.parse import urlparse
 
-# Persistent HTTP session for performance
-session = requests.Session()
-session.headers.update({
-    "User-Agent": "Mozilla/5.0 (compatible; IA-Drone-Explorer/1.0)"
-})
+# --- SETTINGS ---
+NEWS_FILTER = [r"\bcnn\b", r"\bfox\b", r"\bbbc\b", r"\bmsnbc\b", r"\breuters\b"]
+THEME = "gradio/soft"
 
 # --- VirusTotal scan ---
 def scan_url_vt(url, api_key):
     headers = {"x-apikey": api_key}
-    resp = session.post(
+    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"]
     while True:
         time.sleep(5)
-        status_resp = session.get(
+        status = requests.get(
             f"https://www.virustotal.com/api/v3/analyses/{analysis_id}", headers=headers
         )
-        status_resp.raise_for_status()
-        attr = status_resp.json()["data"]["attributes"]
+        attr = status.json()["data"]["attributes"]
         if attr.get("status") == "completed":
-            stats = attr.get("stats", {})
-            return stats.get("malicious", 0) == 0
+            return attr["stats"].get("malicious", 0) == 0
 
-# --- FFprobe metadata extraction ---
-def extract_ffprobe_metadata(url_or_path):
-    cmd = [
-        "ffprobe", "-v", "error", "-print_format", "json",
-        "-show_format", "-show_streams",
-        url_or_path
-    ]
-    out = subprocess.check_output(cmd)
-    return json.loads(out)
+# --- FFprobe metadata ---
+def extract_ffprobe_metadata(path):
+    output = subprocess.check_output([
+        "ffprobe", "-v", "error", "-print_format", "json", "-show_format", "-show_streams", path
+    ])
+    return json.loads(output)
 
-# --- Scrape page metadata (OpenGraph + title) ---
+# --- Fetch page metadata + favicon ---
 def fetch_page_metadata(url):
     try:
-        resp = session.get(url, timeout=5)
-        resp.raise_for_status()
-        soup = BeautifulSoup(resp.text, "html.parser")
-        meta = {"url": url, "title": soup.title.string if soup.title else None}
+        r = requests.get(url, timeout=5)
+        r.raise_for_status()
+        soup = BeautifulSoup(r.text, "html.parser")
+        meta = {"url": url, "title": soup.title.string if soup.title else url}
         for tag in soup.find_all("meta"):
-            prop = tag.get("property") or tag.get("name")
-            if prop and prop.startswith(("og:", "twitter:")):
-                meta[prop] = tag.get("content")
+            key = tag.get("property") or tag.get("name")
+            if key and (key.startswith("og:") or key.startswith("twitter:")):
+                meta[key] = tag.get("content")
+        icon = soup.find("link", rel=lambda v: v and "icon" in v.lower())
+        meta["favicon"] = icon.get("href") if icon else ""
         return meta
     except Exception as e:
-        return {"url": url, "error": str(e)}
-
-# --- Cache IA metadata to speed repeated fetches ---
-@lru_cache(maxsize=128)
-def fetch_ia_metadata(identifier):
-    item = get_item(identifier)
-    return {
-        "metadata": item.metadata,
-        "files": [
-            {k: v for k, v in f.items() if k != "_checksum"}
-            for f in item.files
-        ]
-    }
-
-# --- Search IA and return identifiers ---
-def fetch_identifiers(keywords):
-    query = " OR ".join([kw.strip().replace(" ", "+") for kw in keywords.split(",")])
-    ia_query = f"mediatype:(movies) AND ({query})"
-    results = list(search_items(ia_query, fields=["identifier"]))[:50]
-    return [r["identifier"] for r in results]
-
-# --- List video files for a given item ---
-def list_files_for_identifier(identifier):
-    data = fetch_ia_metadata(identifier)
-    return [
-        f["name"] for f in data["files"]
-        if f.get("format", "").lower().startswith(("mpeg","mp4","avi","mov","webm","m4v"))
-    ]
-
-# --- Gradio UI setup ---
-with gr.Blocks(theme=gr.themes.Soft()) as demo:
+        return {"url": url, "error": str(e), "favicon": ""}
+
+# --- IA search + filter ---
+def fetch_clean_videos(keywords, api_key, scan_enabled):
+    query = " OR ".join(kw.strip().replace(" ", "+") for kw in keywords.split(","))
+    items = list(search_items(f"mediatype:(movies) AND ({query})"))[:50]
+    results = []
+    for item_meta in items:
+        title = item_meta.get("title", "").lower()
+        if any(re.search(p, title) for p in NEWS_FILTER):
+            continue
+        item = get_item(item_meta['identifier'])
+        for f in item.files:
+            fmt = f.get('format', '').lower()
+            if fmt.startswith(("mpeg","mp4","avi","mov","webm","m4v")):
+                url = f"https://archive.org/download/{item_meta['identifier']}/{f['name']}"
+                if scan_enabled and api_key:
+                    try:
+                        if not scan_url_vt(url, api_key):
+                            continue
+                    except:
+                        continue
+                results.append(url)
+    return results
+
+# --- Build graph HTML (using write_html) ---
+def build_graph_html(chain):
+    net = Network(height="300px", width="100%", directed=True)
+    for hop in chain:
+        url = hop['url']
+        meta = hop['metadata']
+        title = meta.get('title', url)
+        favicon = meta.get('favicon', '')
+        if favicon:
+            net.add_node(url, label="", shape='image', image=favicon, title=title)
+        else:
+            net.add_node(url, label=title, title=title)
+    for i in range(len(chain)-1):
+        net.add_edge(chain[i]['url'], chain[i+1]['url'])
+    tmp_path = tempfile.mktemp(suffix='.html')
+    net.write_html(tmp_path, notebook=False, open_browser=False)
+    with open(tmp_path, 'r', encoding='utf8') as f:
+        html = f.read()
+    os.remove(tmp_path)
+    return html
+
+# --- Gradio UI ---
+with gr.Blocks(theme=THEME) as demo:
     gr.Markdown("# 📼 IA Drone‑Strike Explorer")
     with gr.Row():
-        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)
+        kw_input = gr.Textbox(label="Search keywords (comma-separated)", value="drone strike, military uav")
+        vt_key   = gr.Textbox(label="VirusTotal API Key", type="password")
+    scan_toggle    = gr.Checkbox(label="Enable VirusTotal scan", value=True)
     ffprobe_toggle = gr.Checkbox(label="Enable FFprobe metadata", value=False)
-    run_btn        = gr.Button("🔍 Search Items", variant="primary")
-
-    id_dropdown   = gr.Dropdown(label="IA Item Identifiers", choices=[], interactive=True)
-    file_dropdown = gr.Dropdown(label="Video Files", choices=[], interactive=True)
-    video_player  = gr.Video(label="Video Preview")
+    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        = gr.JSON(label="► Raw IA Metadata")
+    ff_meta        = gr.JSON(label="► FFprobe Metadata")
+    graph_panel    = gr.HTML(label="► Reupload Chain Graph")
+    origin_meta    = gr.JSON(label="► Origin Node Metadata")
+
+    def search_and_populate(keywords, api_key, scan_on):
+        urls = fetch_clean_videos(keywords, api_key, scan_on)
+        return gr.update(choices=urls, value=urls[0] if urls else None)
+
+    def on_url_select(selected_url, ff_on, api_key):
+        if not selected_url:
+            return None, {}, {}, "", {}
+        # IA metadata + files
+        parts = selected_url.split('/')
+        ident = parts[4]
+        item = get_item(ident)
+        ia_data = {'metadata': item.metadata, 'files': item.files}
+        # FFprobe
+        ff_data = extract_ffprobe_metadata(selected_url) if ff_on else {}
+        # origin chain
+        desc = item.metadata.get('description', '')
+        urls = re.findall(r'https?://[^\s"<]+', desc)
+        chain = []
+        for u in urls:
+            chain.append({'url': u, 'metadata': fetch_page_metadata(u)})
+        # append IA as last hop
+        chain.append({'url': selected_url, 'metadata': {'title': item.metadata.get('title','')}})
+        graph_html = build_graph_html(chain)
+        origin_node = chain[0]['metadata'] if chain else {}
+        return selected_url, ia_data, ff_data, graph_html, origin_node
 
-    with gr.Tabs():
-        with gr.TabItem("IA Metadata"):
-            ia_meta_json = gr.JSON(label="Raw IA Metadata")
-        with gr.TabItem("FFprobe"):
-            ffprobe_json = gr.JSON(label="FFprobe Metadata")
-        with gr.TabItem("Origins Graph"):
-            origins_graph = gr.HTML(label="Source Origins Graph")
-            origins_meta  = gr.JSON(label="Origins Metadata")
-
-    # 1) Fetch identifiers for search keywords
     run_btn.click(
-        lambda kws: gr.update(choices=fetch_identifiers(kws), value=None),
-        inputs=[kw_input],
-        outputs=[id_dropdown]
-    )
-
-    # 2) Populate video files dropdown when an identifier is selected
-    id_dropdown.change(
-        lambda ident: gr.update(choices=list_files_for_identifier(ident), value=None),
-        inputs=[id_dropdown],
-        outputs=[file_dropdown]
+        search_and_populate,
+        inputs=[kw_input, vt_key, scan_toggle],
+        outputs=[url_dropdown]
     )
-
-    # 3) When a file is selected, fetch metadata, run FFprobe (if toggled),
-    #    and build the clickable origins graph with circular favicon nodes.
-    def update_all(identifier, file_name, ff_on, api_key):
-        if not identifier or not file_name:
-            return None, {}, {}, "", []
-
-        url = f"https://archive.org/download/{identifier}/{file_name}"
-
-        # IA metadata (cached)
-        data = fetch_ia_metadata(identifier)
-        raw_ia = {"identifier": identifier, **data}
-
-        # FFprobe metadata
-        ff_md = {}
-        if ff_on:
-            try:
-                ff_md = extract_ffprobe_metadata(url)
-            except Exception as e:
-                ff_md = {"error": str(e)}
-
-        # Origins graph
-        desc = data["metadata"].get("description", "") or ""
-        urls = re.findall(r"https?://[^\s\"<]+", desc)
-        origins_list = []
-
-        net = Network(height="300px", width="100%", directed=True)
-        net.set_options('{"edges":{"arrows":"to"}}')
-        net.add_node(identifier, label=identifier, shape="ellipse")
-
-        for u in urls[:10]:
-            meta = fetch_page_metadata(u)
-            origins_list.append(meta)
-            dom = urlparse(u).netloc
-            fav = f"https://www.google.com/s2/favicons?domain={dom}"
-            net.add_node(
-                u,
-                label=dom,
-                shape="circularImage",
-                image=fav,
-                title=json.dumps(meta, indent=2),
-                href=u
-            )
-            net.add_edge(identifier, u)
-
-        graph_html = net.generate_html()
-
-        return url, raw_ia, ff_md, graph_html, origins_list
-
-    file_dropdown.change(
-        update_all,
-        inputs=[id_dropdown, file_dropdown, ffprobe_toggle, vt_key_input],
-        outputs=[video_player, ia_meta_json, ffprobe_json, origins_graph, origins_meta]
+    url_dropdown.change(
+        on_url_select,
+        inputs=[url_dropdown, ffprobe_toggle, vt_key],
+        outputs=[video_player, ia_meta, ff_meta, graph_panel, origin_meta]
     )
 
 if __name__ == "__main__":
-    demo.launch()
+    demo.launch()