Update app.py

app.py

@@ -5,90 +5,91 @@ from typing import List, Dict
 from collections import defaultdict
 
 ###############################################################################
-# GLOBAL STATE
+# GLOBAL STATE (HF SAFE)
 ###############################################################################
 
 RESULT_CACHE: List[Dict] = []
-SELECTED_INDEX = 0
+CURRENT_INDEX = 0
 
 ###############################################################################
-# ROBOTS.TXT POLICY
+# AGENCY POLICY / ROBOTS / LIVE SAFETY
 ###############################################################################
 
 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"},
+    "CIA":  {"robots": True,  "live": True},
+    "FBI":  {"robots": True,  "live": True},
+    "ODNI": {"robots": True,  "live": True},
+    "USAF": {"robots": True,  "live": True},
+    "NSA":  {"robots": False, "live": False},
+    "NRO":  {"robots": False, "live": False},
+    "SAP":  {"robots": False, "live": False},
 }
 
+###############################################################################
+# KILL SWITCH (AUTO + MANUAL SAFE)
+###############################################################################
+
+class KillSwitch:
+    def __init__(self):
+        self.disabled = {}
+
+    def disable(self, agency: str, reason: str):
+        self.disabled[agency] = reason
+
+    def enabled(self, agency: str) -> bool:
+        return agency not in self.disabled
+
+    def reason(self, agency: str) -> str:
+        return self.disabled.get(agency, "")
+
+KILL = KillSwitch()
+
 ###############################################################################
 # FOIA EXEMPTION CLASSIFIER
 ###############################################################################
 
 B_CODES = {
     "b(1)": "National Security",
-    "b(3)": "Statutory Exemption",
+    "b(3)": "Statutory",
     "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)
+    found = [k for k in B_CODES if k in text.lower()]
+    confidence = round(min(1.0, 0.3 * len(found)), 2)
+    return {"codes": found, "confidence": confidence}
 
-    confidence = round(min(1.0, len(found) * 0.3), 2)
-
-    return {
-        "codes": found,
-        "confidence": confidence,
-    }
+###############################################################################
+# REDACTION SCORING
+###############################################################################
 
+def redaction_score(text: str) -> float:
+    hits = sum(k in text.lower() for k in ["redact", "withheld", "b("])
+    return round(min(1.0, hits * 0.25), 2)
 
 ###############################################################################
-# SEMANTIC CLUSTERING (HF SAFE)
+# SEMANTIC CLUSTERING (HF SAFE HEURISTIC)
 ###############################################################################
 
-def semantic_cluster(results: List[Dict]) -> Dict[str, List[Dict]]:
+def semantic_clusters(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:
+        t = r["title"].lower()
+        if "uap" in t or "aerial" in t:
             clusters["🛸 UAP / Aerial Phenomena"].append(r)
-        elif "intelligence" in title:
+        elif "intelligence" in t:
             clusters["🧠 Intelligence Activities"].append(r)
         else:
             clusters["📄 General Records"].append(r)
-
     return clusters
 
-
 ###############################################################################
-# UTILITIES
+# MOCK LIVE SEARCH (REPLACE WITH REAL ADAPTERS SAFELY)
 ###############################################################################
 
-def highlight(text: str, query: str) -> str:
-    if not query:
-        return text
-    pattern = re.compile(re.escape(query), re.IGNORECASE)
-    return pattern.sub(r"<mark>\g<0></mark>", text)
-
-
-def redaction_score(text: str) -> float:
-    hits = sum(k in text.lower() for k in ["redact", "withheld", "b("])
-    return round(min(1.0, hits * 0.25), 2)
-
-
-###############################################################################
-# MOCK LIVE QUERY (REPLACE ADAPTERS LATER)
-###############################################################################
-
-def run_query(query: str) -> List[Dict]:
-    time.sleep(0.4)
+def run_search(query: str) -> List[Dict]:
+    time.sleep(0.3)
 
     raw = [
         {
@@ -114,89 +115,127 @@ def run_query(query: str) -> List[Dict]:
         },
     ]
 
-    # Robots enforcement
     allowed = []
     for r in raw:
         policy = AGENCY_POLICY.get(r["agency"], {})
-        if policy.get("robots_allowed", False):
-            allowed.append(r)
+        if not policy.get("robots", False):
+            continue
+        if not KILL.enabled(r["agency"]):
+            continue
+        allowed.append(r)
 
     return allowed
 
+###############################################################################
+# TEXT UTILITIES
+###############################################################################
+
+def highlight(text: str, query: str) -> str:
+    if not query:
+        return text
+    return re.sub(
+        re.escape(query),
+        lambda m: f"<mark>{m.group(0)}</mark>",
+        text,
+        flags=re.IGNORECASE,
+    )
 
 ###############################################################################
 # RENDERERS
 ###############################################################################
 
-def render_clusters(results: List[Dict], query: str) -> str:
-    clusters = semantic_cluster(results)
+def render_results(results: List[Dict], query: str) -> str:
+    clusters = semantic_clusters(results)
     blocks = []
 
-    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"])
+    for name, items in clusters.items():
+        section = [f"## {name}"]
+        for idx, r in enumerate(items):
+            global_index = RESULT_CACHE.index(r)
+            ex = classify_exemptions(r["snippet"])
 
             section.append(
                 f"""
 **{highlight(r['title'], query)}**  
-🏛️ {r['agency']} · 🛡️ Redaction {redaction_score(r['snippet'])}  
-⚖️ Exemptions: `{', '.join(exemptions['codes']) or 'None'}`  
-🔎 Confidence: **{exemptions['confidence']}**
-
-🔗 [Source]({r['url']})
+🏛️ {r['agency']} · 📊 Redaction {redaction_score(r['snippet'])}  
+⚖️ Exemptions: `{', '.join(ex['codes']) or 'None'}` (conf {ex['confidence']})  
 
-➡️ Select index **{idx}**
+🔗 {r['url']}  
+➡️ **Select #{global_index}**
 """
             )
-
         blocks.append("\n\n".join(section))
 
     return "\n\n---\n\n".join(blocks)
 
-
 def render_preview(index: int) -> str:
     if not RESULT_CACHE:
         return "_No document selected._"
 
     r = RESULT_CACHE[index]
-    exemptions = classify_exemptions(r["snippet"])
+    ex = classify_exemptions(r["snippet"])
+
+    iframe = (
+        f'<iframe src="{r["url"]}" width="100%" height="520px" '
+        f'style="border:1px solid #444;border-radius:8px;"></iframe>'
+    )
 
     return f"""
-### 📄 Preview
+### 📄 Document Preview
 **{r['title']}**  
 🏛️ {r['agency']} · {r['source']}
 
 🛡️ Redaction Risk: **{redaction_score(r['snippet'])}**  
-⚖️ FOIA Exemptions: `{', '.join(exemptions['codes']) or 'None'}`  
-🔎 Confidence: **{exemptions['confidence']}**
+⚖️ FOIA Exemptions: `{', '.join(ex['codes']) or 'None'}`  
+🔎 Confidence: **{ex['confidence']}**
 
-<iframe src="{r['url']}" width="100%" height="520px"
-style="border-radius:8px;border:1px solid #444;"></iframe>
+{iframe}
 """
 
+def agency_coverage(results: List[Dict]) -> str:
+    counts = defaultdict(int)
+    for r in results:
+        counts[r["agency"]] += 1
+
+    rows = ["| Agency | Docs |", "|---|---|"]
+    for k, v in sorted(counts.items(), key=lambda x: -x[1]):
+        rows.append(f"| {k} | {v} |")
+
+    return "\n".join(rows)
 
 ###############################################################################
 # EVENT HANDLERS
 ###############################################################################
 
-def search(query: str):
-    global RESULT_CACHE, SELECTED_INDEX
-    RESULT_CACHE = run_query(query)
-    SELECTED_INDEX = 0
+def do_search(query: str):
+    global RESULT_CACHE, CURRENT_INDEX
+    RESULT_CACHE = run_search(query)
+    CURRENT_INDEX = 0
 
     return (
-        render_clusters(RESULT_CACHE, query),
+        render_results(RESULT_CACHE, query),
         render_preview(0),
+        agency_coverage(RESULT_CACHE),
+        0,
     )
 
-
 def select_index(idx: int):
-    global SELECTED_INDEX
-    SELECTED_INDEX = idx
+    global CURRENT_INDEX
+    idx = int(max(0, min(idx, len(RESULT_CACHE) - 1)))
+    CURRENT_INDEX = idx
     return render_preview(idx)
 
+def next_doc():
+    global CURRENT_INDEX
+    if CURRENT_INDEX < len(RESULT_CACHE) - 1:
+        CURRENT_INDEX += 1
+    return CURRENT_INDEX, render_preview(CURRENT_INDEX)
+
+def prev_doc():
+    global CURRENT_INDEX
+    if CURRENT_INDEX > 0:
+        CURRENT_INDEX -= 1
+    return CURRENT_INDEX, render_preview(CURRENT_INDEX)
 
 ###############################################################################
 # UI
@@ -205,29 +244,43 @@ def select_index(idx: int):
 with gr.Blocks(theme=gr.themes.Soft()) as app:
     gr.Markdown("# 🏛️ Federated FOIA Intelligence Search")
 
-    query = gr.Textbox(label="Search")
+    query = gr.Textbox(label="Search public FOIA reading rooms")
     search_btn = gr.Button("🔍 Search")
 
     with gr.Row():
         with gr.Column(scale=5):
             results_md = gr.Markdown()
+            with gr.Accordion("🗺️ Agency Coverage", open=False):
+                coverage_md = gr.Markdown()
 
         with gr.Column(scale=7):
             preview_md = gr.Markdown()
-
-    select_box = gr.Number(label="Select result index", precision=0)
+            with gr.Row():
+                prev_btn = gr.Button("⬅️ Prev")
+                next_btn = gr.Button("➡️ Next")
+            index_box = gr.Number(label="Selected index", precision=0)
 
     search_btn.click(
-        fn=search,
+        do_search,
         inputs=query,
-        outputs=[results_md, preview_md],
+        outputs=[results_md, preview_md, coverage_md, index_box],
     )
 
-    select_box.change(
-        fn=select_index,
-        inputs=select_box,
+    index_box.change(
+        select_index,
+        inputs=index_box,
         outputs=preview_md,
     )
 
+    next_btn.click(
+        next_doc,
+        outputs=[index_box, preview_md],
+    )
+
+    prev_btn.click(
+        prev_doc,
+        outputs=[index_box, preview_md],
+    )
+
 if __name__ == "__main__":
     app.launch()