Update app.py

app.py

@@ -2,14 +2,71 @@ import gradio as gr
 import time
 import re
 from typing import List, Dict
+from collections import defaultdict
 
 ###############################################################################
-# GLOBAL STATE (HF SAFE)
+# GLOBAL STATE
 ###############################################################################
 
 RESULT_CACHE: List[Dict] = []
 SELECTED_INDEX = 0
 
+###############################################################################
+# ROBOTS.TXT POLICY
+###############################################################################
+
+AGENCY_POLICY = {
+    "CIA": {"robots_allowed": True, "reason": "Explicit public FOIA reading room"},
+    "ODNI": {"robots_allowed": True, "reason": "Public PDF hosting"},
+    "USAF": {"robots_allowed": True, "reason": "National Archives mirror"},
+    "NSA": {"robots_allowed": False, "reason": "robots.txt disallows crawling"},
+}
+
+###############################################################################
+# FOIA EXEMPTION CLASSIFIER
+###############################################################################
+
+B_CODES = {
+    "b(1)": "National Security",
+    "b(3)": "Statutory Exemption",
+    "b(5)": "Deliberative Process",
+    "b(7)": "Law Enforcement",
+}
+
+
+def classify_exemptions(text: str) -> Dict:
+    found = []
+    for code, desc in B_CODES.items():
+        if code in text.lower():
+            found.append(code)
+
+    confidence = round(min(1.0, len(found) * 0.3), 2)
+
+    return {
+        "codes": found,
+        "confidence": confidence,
+    }
+
+
+###############################################################################
+# SEMANTIC CLUSTERING (HF SAFE)
+###############################################################################
+
+def semantic_cluster(results: List[Dict]) -> Dict[str, List[Dict]]:
+    clusters = defaultdict(list)
+
+    for r in results:
+        title = r["title"].lower()
+        if "uap" in title or "aerial" in title:
+            clusters["🛸 UAP / Aerial Phenomena"].append(r)
+        elif "intelligence" in title:
+            clusters["🧠 Intelligence Activities"].append(r)
+        else:
+            clusters["📄 General Records"].append(r)
+
+    return clusters
+
+
 ###############################################################################
 # UTILITIES
 ###############################################################################
@@ -17,167 +74,128 @@ SELECTED_INDEX = 0
 def highlight(text: str, query: str) -> str:
     if not query:
         return text
-    terms = [re.escape(t) for t in query.split() if len(t) > 1]
-    if not terms:
-        return text
-    pattern = re.compile(rf"({'|'.join(terms)})", re.IGNORECASE)
-    return pattern.sub(r"<mark>\1</mark>", text)
+    pattern = re.compile(re.escape(query), re.IGNORECASE)
+    return pattern.sub(r"<mark>\g<0></mark>", text)
 
 
 def redaction_score(text: str) -> float:
-    """
-    Heuristic confidence score that document may contain redactions.
-    0.0 = none detected, 1.0 = heavy redaction likelihood
-    """
-    hits = sum(
-        k in text.lower()
-        for k in ["redact", "b(1)", "b(3)", "withheld", "classified"]
-    )
+    hits = sum(k in text.lower() for k in ["redact", "withheld", "b("])
     return round(min(1.0, hits * 0.25), 2)
 
 
-def badge(r: Dict) -> str:
-    live = "🟢 LIVE" if r["live"] else "🟡 STUB"
-    return f"`{live}` · `{r['agency']}`"
-
-
 ###############################################################################
-# MOCK ADAPTER OUTPUT (REPLACE WITH REAL ADAPTERS)
+# MOCK LIVE QUERY (REPLACE ADAPTERS LATER)
 ###############################################################################
 
-def run_federated_query(query: str) -> List[Dict]:
-    """
-    HF-safe simulated adapter aggregation.
-    """
+def run_query(query: str) -> List[Dict]:
     time.sleep(0.4)
 
-    return [
+    raw = [
         {
-            "title": "UAP Task Force Report (Preliminary)",
-            "snippet": "The UAPTF evaluated a number of UAP incidents...",
+            "title": "UAP Task Force Preliminary Assessment",
+            "snippet": "Some material withheld under b(1) and b(3).",
             "url": "https://www.dni.gov/files/ODNI/documents/assessments/Prelimary-Assessments-UAP-20210625.pdf",
             "agency": "ODNI",
-            "source": "ODNI FOIA Reading Room",
-            "live": True,
+            "source": "ODNI FOIA",
         },
         {
             "title": "CIA Memorandum on Aerial Phenomena",
-            "snippet": "This document was partially redacted under b(3)...",
+            "snippet": "This document contains redactions under b(3).",
             "url": "https://www.cia.gov/readingroom/docs/DOC_000001.pdf",
             "agency": "CIA",
-            "source": "CIA FOIA Reading Room",
-            "live": True,
+            "source": "CIA FOIA",
         },
         {
-            "title": "Project BLUE BOOK Summary",
-            "snippet": "Historical summary of investigations into unidentified objects...",
+            "title": "Project Blue Book Summary",
+            "snippet": "Historical investigation records.",
             "url": "https://www.archives.gov/research/military/air-force/ufos",
             "agency": "USAF",
             "source": "National Archives",
-            "live": False,
         },
     ]
 
+    # Robots enforcement
+    allowed = []
+    for r in raw:
+        policy = AGENCY_POLICY.get(r["agency"], {})
+        if policy.get("robots_allowed", False):
+            allowed.append(r)
 
-###############################################################################
-# RESULT RENDERING
-###############################################################################
-
-def agency_counts(results: List[Dict]) -> str:
-    counts = {}
-    for r in results:
-        counts[r["agency"]] = counts.get(r["agency"], 0) + 1
+    return allowed
 
-    lines = ["### 🏛️ Per-Agency Coverage"]
-    for agency, count in sorted(counts.items()):
-        lines.append(f"- **{agency}**: {count}")
-
-    return "\n".join(lines)
 
+###############################################################################
+# RENDERERS
+###############################################################################
 
-def render_results(results: List[Dict], query: str) -> str:
-    lines = ["### 📚 Search Results (click to preview)\n"]
+def render_clusters(results: List[Dict], query: str) -> str:
+    clusters = semantic_cluster(results)
+    blocks = []
 
-    for i, r in enumerate(results):
-        title = highlight(r["title"], query)
-        snippet = highlight(r["snippet"], query)
-        score = redaction_score(r["snippet"])
+    for cluster, items in clusters.items():
+        section = [f"## {cluster}"]
+        for i, r in enumerate(items):
+            idx = RESULT_CACHE.index(r)
+            exemptions = classify_exemptions(r["snippet"])
 
-        lines.append(
-            f"""
-<div style="cursor:pointer" onclick="">
-**{i+1}. {title}**  
-{badge(r)}  
-🛡️ Redaction Confidence: **{score}**
+            section.append(
+                f"""
+**{highlight(r['title'], query)}**  
+🏛️ {r['agency']} · 🛡️ Redaction {redaction_score(r['snippet'])}  
+⚖️ Exemptions: `{', '.join(exemptions['codes']) or 'None'}`  
+🔎 Confidence: **{exemptions['confidence']}**
 
-{snippet}
+🔗 [Source]({r['url']})
 
-🔗 [Open Source]({r['url']})
-</div>
+➡️ Select index **{idx}**
 """
-        )
+            )
 
-    return "\n\n---\n\n".join(lines)
+        blocks.append("\n\n".join(section))
 
+    return "\n\n---\n\n".join(blocks)
 
-###############################################################################
-# PREVIEW PANEL
-###############################################################################
 
 def render_preview(index: int) -> str:
     if not RESULT_CACHE:
         return "_No document selected._"
 
-    index = max(0, min(index, len(RESULT_CACHE) - 1))
     r = RESULT_CACHE[index]
+    exemptions = classify_exemptions(r["snippet"])
 
-    score = redaction_score(r["snippet"])
-
-    header = f"""
-### 📄 Document Preview ({index + 1}/{len(RESULT_CACHE)})
+    return f"""
+### 📄 Preview
 **{r['title']}**  
-{r['source']} · `{r['agency']}`  
-🛡️ Redaction Confidence: **{score}**
-"""
+🏛️ {r['agency']} · {r['source']}
 
-    iframe = f"""
-<iframe src="{r['url']}" width="100%" height="550px"
-style="border:1px solid #444;border-radius:8px;"></iframe>
-"""
+🛡️ Redaction Risk: **{redaction_score(r['snippet'])}**  
+⚖️ FOIA Exemptions: `{', '.join(exemptions['codes']) or 'None'}`  
+🔎 Confidence: **{exemptions['confidence']}**
 
-    return header + iframe
+<iframe src="{r['url']}" width="100%" height="520px"
+style="border-radius:8px;border:1px solid #444;"></iframe>
+"""
 
 
 ###############################################################################
-# SEARCH HANDLER
+# EVENT HANDLERS
 ###############################################################################
 
-def run_search(query: str):
+def search(query: str):
     global RESULT_CACHE, SELECTED_INDEX
-
-    if not query.strip():
-        return "⚠️ Enter a search term.", "_", "_"
-
-    results = run_federated_query(query)
-
-    RESULT_CACHE = results
+    RESULT_CACHE = run_query(query)
     SELECTED_INDEX = 0
 
     return (
-        render_results(results, query),
+        render_clusters(RESULT_CACHE, query),
         render_preview(0),
-        agency_counts(results),
     )
 
 
-def next_doc(idx):
-    idx = min(idx + 1, len(RESULT_CACHE) - 1)
-    return render_preview(idx), idx
-
-
-def prev_doc(idx):
-    idx = max(idx - 1, 0)
-    return render_preview(idx), idx
+def select_index(idx: int):
+    global SELECTED_INDEX
+    SELECTED_INDEX = idx
+    return render_preview(idx)
 
 
 ###############################################################################
@@ -185,49 +203,31 @@ def prev_doc(idx):
 ###############################################################################
 
 with gr.Blocks(theme=gr.themes.Soft()) as app:
-    gr.Markdown(
-        """
-# 🏛️ Federated FOIA Document Search
-Search public FOIA reading rooms across agencies.
-"""
-    )
-
-    query = gr.Textbox(label="Search term", placeholder="e.g. UAP")
+    gr.Markdown("# 🏛️ Federated FOIA Intelligence Search")
 
+    query = gr.Textbox(label="Search")
     search_btn = gr.Button("🔍 Search")
 
-    selected_index = gr.State(0)
-
     with gr.Row():
         with gr.Column(scale=5):
             results_md = gr.Markdown()
 
         with gr.Column(scale=7):
             preview_md = gr.Markdown()
-            with gr.Row():
-                prev_btn = gr.Button("⬅️ Previous")
-                next_btn = gr.Button("➡️ Next")
 
-    coverage_md = gr.Markdown()
+    select_box = gr.Number(label="Select result index", precision=0)
 
     search_btn.click(
-        fn=run_search,
+        fn=search,
         inputs=query,
-        outputs=[results_md, preview_md, coverage_md],
+        outputs=[results_md, preview_md],
     )
 
-    prev_btn.click(
-        fn=prev_doc,
-        inputs=selected_index,
-        outputs=[preview_md, selected_index],
+    select_box.change(
+        fn=select_index,
+        inputs=select_box,
+        outputs=preview_md,
     )
 
-    next_btn.click(
-        fn=next_doc,
-        inputs=selected_index,
-        outputs=[preview_md, selected_index],
-    )
-
-
 if __name__ == "__main__":
     app.launch()