""" app.py — Multi-source OSINT Analyst Space Agentic loop powered by smolagents + Modal inference backend. Required Space Secrets MODAL_ENDPOINT MODAL_API_TOKEN ACLED_USERNAME ACLED_EMAIL HF_TOKEN Optional Space Secrets: TAVILY_API_KEY — from https://tavily.com (enables live web search in analysis + chat) """ import os import time import threading import requests from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime from typing import Optional import gradio as gr from smolagents import ToolCallingAgent try: from smolagents.models import ChatMessage except ImportError: from smolagents import ChatMessage from tools import ( fetch_acled_events, fetch_rss_headlines, fetch_travel_advisory, fetch_airspace_status, list_available_sources, tavily_web_search, collect_rss_articles, collect_tavily_articles, ) from brief import ( BRIEF_PROMPT_SCHEMA, ThreatBrief, NewsItem, parse_brief_from_llm, render_brief_html, ) from export import generate_pdf # --------------------------------------------------------------------------- # Config # --------------------------------------------------------------------------- MODAL_ENDPOINT = os.environ.get("MODAL_ENDPOINT", "") MODAL_API_TOKEN = os.environ.get("MODAL_API_TOKEN", "") # --------------------------------------------------------------------------- # Custom model class — calls Modal REST endpoint instead of HF Inference API # # InferenceClientModel speaks the HF Inference API format; our Modal endpoint # returns {"response": "...", "model_id": "..."} which is a custom format. # This wrapper bridges the gap: converts smolagents message dicts into a # flat prompt string, POSTs to Modal, and returns the text response. # # Think of it like a radio operator who translates between two agencies # using different protocols — same information, different encoding. # --------------------------------------------------------------------------- class ModalModel: """Thin wrapper that routes smolagents inference calls to a Modal endpoint.""" def __init__(self, endpoint_url: str, api_token: str = ""): self.endpoint_url = endpoint_url self.api_token = api_token def _call_endpoint(self, messages, stop_sequences=None, **kwargs) -> str: if not self.endpoint_url: raise ValueError( "MODAL_ENDPOINT is not set. " "Add it in Space Settings > Variables and Secrets." ) # Flatten messages into a single prompt string. # smolagents may pass plain dicts OR ChatMessage objects depending on version. prompt_lines = [] for m in messages: if isinstance(m, dict): role = m.get("role", "user") content = m.get("content", "") else: # ChatMessage object — access as attributes role = getattr(m, "role", "user") content = getattr(m, "content", "") if isinstance(content, list): parts = [] for block in content: if isinstance(block, dict): if block.get("type") == "text": parts.append(block.get("text", "")) elif getattr(block, "type", None) == "text": parts.append(getattr(block, "text", "")) content = " ".join(parts) if content: prompt_lines.append(f"{role.upper()}: {content}") prompt = "\n".join(prompt_lines) headers = {"Content-Type": "application/json"} if self.api_token: headers["Authorization"] = f"Bearer {self.api_token}" resp = requests.post( self.endpoint_url, json={"prompt": prompt, "max_new_tokens": 4096}, headers=headers, timeout=180, ) resp.raise_for_status() text = resp.json()["response"] return ChatMessage(role="assistant", content=text) def __call__(self, messages, stop_sequences=None, **kwargs) -> str: # Some smolagents versions call model(messages) return self._call_endpoint(messages, stop_sequences, **kwargs) def generate(self, messages, stop_sequences=None, **kwargs) -> str: # Other smolagents versions call model.generate(messages) return self._call_endpoint(messages, stop_sequences, **kwargs) # --------------------------------------------------------------------------- # Agent setup # --------------------------------------------------------------------------- def build_agent() -> ToolCallingAgent: model = ModalModel(MODAL_ENDPOINT, MODAL_API_TOKEN) agent = ToolCallingAgent( tools=[ fetch_acled_events, fetch_rss_headlines, fetch_travel_advisory, fetch_airspace_status, list_available_sources, tavily_web_search, ], model=model, max_steps=12, verbosity_level=1, ) return agent # --------------------------------------------------------------------------- # Core analysis function # --------------------------------------------------------------------------- SYSTEM_PROMPT = """You are a professional OSINT intelligence analyst specializing in geopolitical conflict and security threat assessment. Your job is to: 1. Call the available tools to gather data from ACLED, RSS news sources, State Department travel advisories, airspace feeds, and live web search. 2. Use tavily_web_search to find breaking news, recent reports, or any gaps not covered by the other tools. Good queries include the country name plus terms like 'security situation', 'conflict update', 'travel warning', or 'latest news'. 3. Collect enough information to assess the security situation, including travel safety and airspace status. 4. Synthesize your findings into a structured threat brief. Always start by checking what sources are available if needed. Be thorough — use multiple sources before drawing conclusions. """ def _format_rss_for_llm(articles: list) -> str: """Compact text rendering of structured RSS articles for the synthesis prompt.""" if not articles: return "[No matching RSS articles retrieved.]" lines = [f"{len(articles)} articles retrieved:"] for i, a in enumerate(articles, 1): flag = " [NOTABLE]" if a.get("notable") else "" lines.append( f"{i}. ({a.get('source_name','?')}){flag} {a.get('title','')} " f"— {a.get('summary','')}" ) return "\n".join(lines) def run_analysis( country: str, passport_country: str, rss_sources: list, days_back: int, progress=gr.Progress(), ) -> tuple: """ Gathers all OSINT sources in parallel, then makes a single LLM synthesis call. This replaces the multi-step agentic loop and cuts runtime from ~300s to ~60-90s. """ if not country or not str(country).strip(): return "

Please select a country or region.

", "", None country = str(country).strip() if not MODAL_ENDPOINT: error_html = ( "
Configuration error:
" "MODAL_ENDPOINT is not set. Add it in Space Settings > " "Variables and Secrets.
" ) return error_html, "MODAL_ENDPOINT not configured.", None sources_str = ",".join(rss_sources) if rss_sources else "bbc_world,al_jazeera" include_embassy = bool(passport_country and passport_country != "Not specified") year = datetime.utcnow().strftime("%Y") # ----------------------------------------------------------------------- # Phase 1: Parallel data collection — all sources fire simultaneously. # This was the #1 performance bottleneck: 5 sequential tool calls × 1 LLM # decision each = 5 extra Modal round-trips (~100-150s). Now it's one # parallel fan-out with zero LLM involvement. # ----------------------------------------------------------------------- # RSS and Tavily are both collected as STRUCTURED data and injected directly # into the brief — bypassing the LLM for news so every URL is real. rss_box: dict = {} tavily_box: dict = {} def _collect_rss(): articles, errors = collect_rss_articles(country, sources_str, max_articles=30) rss_box["articles"] = articles rss_box["errors"] = errors return _format_rss_for_llm(articles) def _collect_tavily_news(): articles = collect_tavily_articles(country, max_results=10) tavily_box["articles"] = articles # Also return a compact text version for the synthesis prompt if not articles: return "[Tavily news] No results." lines = [f"[Tavily news] {len(articles)} results:"] for a in articles: lines.append(f"- ({a['source_name']}) {a['title']} — {a['summary']}") return "\n".join(lines) tasks = { "acled": lambda: fetch_acled_events(country, days_back), "rss": _collect_rss, "advisory": lambda: fetch_travel_advisory(country), "airspace": lambda: fetch_airspace_status(country), "web_security": lambda: tavily_web_search( f"{country} security situation {year}", max_results=6 ), "web_news": _collect_tavily_news, } progress(0.05, desc=f"Collecting Intelligence on {country}...") collected: dict = {} lock = threading.Lock() def _run_task(name: str, fn): try: result = fn() except Exception as e: result = f"[{name.upper()}] Collection error: {e}" with lock: collected[name] = result n_done = len(collected) labels = { "acled": "Conflict events (ACLED)", "rss": "News feeds (RSS)", "advisory": "Travel advisory (State Dept)", "airspace": "Airspace status", "web_security": "Security web search (Tavily)", "web_news": "Live news search (Tavily)", } progress( 0.05 + 0.50 * n_done / len(tasks), desc=f"Collecting Intelligence — {labels.get(name, name)} ({n_done}/{len(tasks)})...", ) with ThreadPoolExecutor(max_workers=len(tasks)) as executor: futs = [executor.submit(_run_task, name, fn) for name, fn in tasks.items()] for f in as_completed(futs): f.result() # surface exceptions # ----------------------------------------------------------------------- # Phase 2: Single LLM synthesis call. # Previously the agent made 8+ round-trips. Now it's exactly one. # ----------------------------------------------------------------------- progress(0.60, desc="Synthesizing Intelligence Report...") # Cap each source block so the total prompt stays within the model's context. # RSS is the biggest contributor — 25 articles can easily be 15,000+ chars. def _cap(text: str, max_chars: int) -> str: if len(text) <= max_chars: return text return text[:max_chars] + f"\n... [truncated at {max_chars} chars]" embassy_block = ( f"\nPassport country for traveller: {passport_country}\n" f"REQUIRED — Populate the 'embassy' JSON field with the {passport_country} " f"embassy or nearest consulate in '{country}'. Include: name, street address, " f"main phone number, after-hours emergency phone, and official website URL.\n" if include_embassy else "" ) synthesis_prompt = f"""You are an OSINT intelligence analyst. Below is all collected source data for {country}. Analyse it thoroughly, then produce the final JSON threat brief. === ACLED CONFLICT EVENTS (last {days_back} days) === {_cap(collected.get('acled', '[Not available]'), 4000)} === RSS NEWS HEADLINES === {_cap(collected.get('rss', '[Not available]'), 6000)} === LIVE WEB NEWS (Tavily) === {_cap(collected.get('web_news', '[Not available]'), 3000)} === SECURITY SITUATION WEB SEARCH === {_cap(collected.get('web_security', '[Not available]'), 3000)} === US STATE DEPT TRAVEL ADVISORY === {_cap(collected.get('advisory', '[Not available]'), 2000)} === AIRSPACE STATUS === {_cap(collected.get('airspace', '[Not available]'), 3000)} {embassy_block} Today's date: {datetime.utcnow().strftime('%Y-%m-%d')} {BRIEF_PROMPT_SCHEMA}""" messages = [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": synthesis_prompt}, ] raw_output = None last_error = None for attempt in range(1, 4): try: model = ModalModel(MODAL_ENDPOINT, MODAL_API_TOKEN) response = model(messages) raw_output = response.content if hasattr(response, "content") else str(response) break except requests.exceptions.Timeout: last_error = Exception("Modal endpoint timed out (120 s).") if attempt < 3: progress(0.60 + attempt * 0.05, desc=f"Synthesizing Intelligence Report — retrying ({attempt + 1}/3)...") time.sleep(5 * attempt) continue return ( "
Timed out after 3 attempts.
" "The Modal container may be cold-starting. Wait 30 seconds and try again.
", str(last_error), None, ) except Exception as e: last_error = e err_str = str(e).lower() is_timeout = any(k in err_str for k in ("504", "timeout", "gateway", "timed out")) if is_timeout and attempt < 3: progress(0.60 + attempt * 0.05, desc=f"Synthesizing Intelligence Report — retrying ({attempt + 1}/3)...") time.sleep(5 * attempt) continue error_html = ( f"
Synthesis failed (attempt {attempt}/3):
" f"{e}

" f"If you see a 504 / gateway timeout the Modal container may be under load. " f"Wait a minute and try again.
" ) return error_html, str(e), None if raw_output is None: return ( "

Analysis failed after 3 attempts. Please try again later.

", str(last_error), None, ) progress(0.85, desc="Synthesizing Intelligence Report — parsing...") brief = parse_brief_from_llm(raw_output) # Merge RSS + Tavily articles directly into the brief, bypassing LLM news # transcription entirely. The LLM's analytical summaries are carried over # where they match a real article by title; otherwise the article's own # snippet is used. Every URL in the final brief is real and clickable. rss_articles = rss_box.get("articles", []) tavily_articles = tavily_box.get("articles", []) all_news = rss_articles + tavily_articles # De-duplicate across sources by URL seen_urls: set = set() deduped_news = [] for a in all_news: u = a.get("url", "") if u and u in seen_urls: continue if u: seen_urls.add(u) deduped_news.append(a) if deduped_news: llm_summaries = { (n.title or "").strip().lower(): n.summary for n in brief.notable_news if n.summary } # Notable first, then by source (tavily results interleaved with RSS) deduped_news.sort(key=lambda a: not a.get("notable", False)) brief.notable_news = [ NewsItem( title = a.get("title", ""), source = a.get("source_name", ""), published = a.get("published", ""), summary = llm_summaries.get((a.get("title", "") or "").strip().lower()) or a.get("summary", ""), url = a.get("url", ""), notable = a.get("notable", False), ) for a in deduped_news ] if passport_country and passport_country != "Not specified": brief.passport_country = passport_country progress(0.95, desc="Synthesizing Intelligence Report — rendering...") html_output = render_brief_html(brief) trace_lines = [ f"=== OSINT Analysis: {country} ===", f"Date: {datetime.utcnow().strftime('%Y-%m-%d %H:%M UTC')}", f"Sources: ACLED + Airspace + State Dept + RSS ({sources_str}) + Tavily", f"Days back: {days_back}", f"Endpoint: {MODAL_ENDPOINT}", "", "--- Collected Source Data ---", f"[ACLED]\n{collected.get('acled','—')}", f"\n[RSS] ({len(rss_box.get('articles',[]))} articles)\n{collected.get('rss','—')}", f"\n[Tavily News] ({len(tavily_box.get('articles',[]))} articles)\n{collected.get('web_news','—')}", f"\n[Advisory]\n{collected.get('advisory','—')}", f"\n[Airspace]\n{collected.get('airspace','—')}", f"\n[Security Web Search]\n{collected.get('web_security','—')}", "", "--- Raw LLM Output ---", str(raw_output), ] progress(1.0, desc="Intelligence Report ready.") return html_output, "\n".join(trace_lines), brief def chat_with_analyst( user_message: str, history: list, brief: Optional[object], country: str, ) -> tuple: """Send a question to the LLM with brief context and return updated history.""" if not user_message or not user_message.strip(): return history, "" if not MODAL_ENDPOINT: history = history + [ {"role": "user", "content": user_message}, {"role": "assistant", "content": "⚠️ MODAL_ENDPOINT is not configured. Add it in Space Settings > Variables and Secrets."}, ] return history, "" country_label = str(country).strip() if country else "the selected country" # Build system context from the brief if available if brief is not None: brief_context = f"""You are an expert OSINT intelligence analyst. You have already completed a threat assessment for {country_label}. Here is a summary of the findings: Severity: {getattr(brief, 'severity', 'Unknown')} Confidence: {getattr(brief, 'confidence', 'Unknown')} Summary: {getattr(brief, 'narrative_summary', 'No summary available.')} Travel Advisory: {getattr(brief, 'travel_advisory_summary', 'Not available.')} Airspace Status: {getattr(brief, 'airspace_status', 'Not available.')} Answer the analyst's follow-up questions concisely and accurately based on this context and your general knowledge. Always ground your answers in the provided data where relevant.""" else: brief_context = ( f"You are an expert OSINT intelligence analyst. The analyst is researching " f"{country_label}. No automated analysis has been run yet — answer based on " f"your general knowledge and note that running an analysis first will give you " f"access to live ACLED, RSS, and State Department data." ) messages = [{"role": "system", "content": brief_context}] for msg in history: messages.append({"role": msg["role"], "content": msg["content"]}) messages.append({"role": "user", "content": user_message}) try: model = ModalModel(MODAL_ENDPOINT, MODAL_API_TOKEN) # Append a note about Tavily availability so the model knows it can suggest searches tavily_available = bool(os.environ.get("TAVILY_API_KEY", "").strip()) if tavily_available: messages[-1]["content"] += ( "\n\n(You may also suggest the analyst use the web search tool " "for real-time results if the question requires up-to-date information.)" ) response = model(messages) answer = response.content if hasattr(response, "content") else str(response) except Exception as e: answer = f"⚠️ Error contacting the model: {e}" history = history + [ {"role": "user", "content": user_message}, {"role": "assistant", "content": answer}, ] return history, "" # --------------------------------------------------------------------------- # Gradio UI # --------------------------------------------------------------------------- RSS_SOURCE_OPTIONS = [ # General world news ("BBC World", "bbc_world"), ("Al Jazeera", "al_jazeera"), ("France 24", "france24"), ("Euronews", "euronews"), ("NPR World", "npr_world"), ("Sky News", "sky_news"), ("UN News", "un_news"), ("Intl Business Times", "ibt"), # Regional: Middle East ("Middle East Eye", "middle_east_eye"), ("Al-Monitor", "al_monitor"), ("Arab News", "arab_news"), # Regional: Africa ("AllAfrica", "allafrica"), # Regional: Asia-Pacific ("Radio Free Asia", "radio_free_asia"), ("S. China Morning Post", "scmp"), # Regional: South Asia ("Dawn (Pakistan)", "dawn"), # Regional: Russia / E. Europe ("The Moscow Times", "moscow_times"), # OSINT / investigative ("Bellingcat", "bellingcat"), ("The Intercept", "the_intercept"), ("OCCRP", "occrp"), # Policy / security analysis ("Crisis Group", "crisis_group"), ("War on the Rocks", "war_on_rocks"), ("Just Security", "just_security"), ("Defense One", "defense_one"), ("The Cipher Brief", "cipher_brief"), ("Stimson Center", "stimson"), # Human rights ("Human Rights Watch", "hrw"), ("Amnesty Intl", "amnesty"), ] DESTINATION_COUNTRIES = [ "Afghanistan", "Albania", "Algeria", "Andorra", "Angola", "Antigua and Barbuda", "Argentina", "Armenia", "Australia", "Austria", "Azerbaijan", "Bahamas", "Bahrain", "Bangladesh", "Barbados", "Belarus", "Belgium", "Belize", "Benin", "Bhutan", "Bolivia", "Bosnia and Herzegovina", "Botswana", "Brazil", "Brunei", "Bulgaria", "Burkina Faso", "Burundi", "Cabo Verde", "Cambodia", "Cameroon", "Canada", "Central African Republic", "Chad", "Chile", "China", "Colombia", "Comoros", "Congo (Republic)", "Congo (DRC)", "Costa Rica", "Croatia", "Cuba", "Cyprus", "Czech Republic", "Denmark", "Djibouti", "Dominica", "Dominican Republic", "Ecuador", "Egypt", "El Salvador", "Equatorial Guinea", "Eritrea", "Estonia", "Eswatini", "Ethiopia", "Fiji", "Finland", "France", "Gabon", "Gambia", "Georgia", "Germany", "Ghana", "Greece", "Grenada", "Guatemala", "Guinea", "Guinea-Bissau", "Guyana", "Haiti", "Honduras", "Hungary", "Iceland", "India", "Indonesia", "Iran", "Iraq", "Ireland", "Israel", "Italy", "Jamaica", "Japan", "Jordan", "Kazakhstan", "Kenya", "Kiribati", "Kosovo", "Kuwait", "Kyrgyzstan", "Laos", "Latvia", "Lebanon", "Lesotho", "Liberia", "Libya", "Liechtenstein", "Lithuania", "Luxembourg", "Madagascar", "Malawi", "Malaysia", "Maldives", "Mali", "Malta", "Marshall Islands", "Mauritania", "Mauritius", "Mexico", "Micronesia", "Moldova", "Monaco", "Mongolia", "Montenegro", "Morocco", "Mozambique", "Myanmar", "Namibia", "Nauru", "Nepal", "Netherlands", "New Zealand", "Nicaragua", "Niger", "Nigeria", "North Korea", "North Macedonia", "Norway", "Oman", "Pakistan", "Palau", "Palestine", "Panama", "Papua New Guinea", "Paraguay", "Peru", "Philippines", "Poland", "Portugal", "Qatar", "Romania", "Russia", "Rwanda", "Saint Kitts and Nevis", "Saint Lucia", "Saint Vincent and the Grenadines", "Samoa", "San Marino", "Sao Tome and Principe", "Saudi Arabia", "Senegal", "Serbia", "Seychelles", "Sierra Leone", "Singapore", "Slovakia", "Slovenia", "Solomon Islands", "Somalia", "South Africa", "South Korea", "South Sudan", "Spain", "Sri Lanka", "Sudan", "Suriname", "Sweden", "Switzerland", "Syria", "Taiwan", "Tajikistan", "Tanzania", "Thailand", "Timor-Leste", "Togo", "Tonga", "Trinidad and Tobago", "Tunisia", "Turkey", "Turkmenistan", "Tuvalu", "Uganda", "Ukraine", "United Arab Emirates", "United Kingdom", "United States", "Uruguay", "Uzbekistan", "Vanuatu", "Vatican City", "Venezuela", "Vietnam", "Yemen", "Zambia", "Zimbabwe", ] PASSPORT_COUNTRIES = [ "Not specified", "Afghanistan", "Albania", "Algeria", "Argentina", "Australia", "Austria", "Bangladesh", "Belgium", "Bolivia", "Brazil", "Cambodia", "Canada", "Chile", "China", "Colombia", "Croatia", "Czech Republic", "Denmark", "Ecuador", "Egypt", "Ethiopia", "Finland", "France", "Germany", "Ghana", "Greece", "Guatemala", "Hungary", "India", "Indonesia", "Iran", "Iraq", "Ireland", "Israel", "Italy", "Japan", "Jordan", "Kazakhstan", "Kenya", "Kuwait", "Malaysia", "Mexico", "Morocco", "Nepal", "Netherlands", "New Zealand", "Nigeria", "Norway", "Pakistan", "Peru", "Philippines", "Poland", "Portugal", "Qatar", "Romania", "Russia", "Saudi Arabia", "Senegal", "Singapore", "South Africa", "South Korea", "Spain", "Sri Lanka", "Sudan", "Sweden", "Switzerland", "Taiwan", "Thailand", "Turkey", "Ukraine", "United Arab Emirates", "United Kingdom", "United States", "Venezuela", "Vietnam", "Zimbabwe", ] EXAMPLE_QUERIES = [ ["Sudan", ["bbc_world", "al_jazeera", "middle_east_eye", "hrw"], 14], ["Myanmar", ["bbc_world", "crisis_group", "radio_free_asia", "hrw"], 21], ["Ukraine", ["bbc_world", "npr_world", "sky_news", "war_on_rocks"], 7], ["Haiti", ["bbc_world", "al_jazeera", "un_news", "amnesty"], 14], ] CSS = """ .gradio-container { max-width: 1280px !important; margin: auto; font-family: 'Inter', system-ui, sans-serif; } footer { display: none !important; } /* Sidebar panel */ #config-panel { background: #0d1117; border-radius: 10px; padding: 20px; color: #e2e8f0; } #config-panel .label-wrap > span { color: #94a3b8 !important; font-size: 0.82em !important; text-transform: uppercase; letter-spacing: 0.06em; } /* Text inputs / dropdowns only — must NOT touch checkboxes or their checked state */ #config-panel input[type="text"], #config-panel input[type="search"], #config-panel textarea, #config-panel select { background: #161b22 !important; border-color: #30363d !important; color: #e2e8f0 !important; } /* Make checkboxes visible on the dark panel and keep their labels readable */ #config-panel input[type="checkbox"] { accent-color: #2563eb !important; width: 16px; height: 16px; } #config-panel .gr-checkbox-group label, #config-panel fieldset label span { color: #c9d1d9 !important; text-transform: none !important; letter-spacing: normal !important; font-size: 0.85em !important; } /* Buttons */ #analyze-btn { background: linear-gradient(135deg, #1e3a5f 0%, #2563eb 100%) !important; color: white !important; border: none !important; font-weight: 700 !important; letter-spacing: 0.05em !important; font-size: 1em !important; padding: 14px !important; border-radius: 8px !important; } #analyze-btn:hover { background: linear-gradient(135deg, #2563eb 0%, #3b82f6 100%) !important; } #export-btn { border: 1px solid #30363d !important; color: #94a3b8 !important; } /* Tabs */ .tab-nav button { font-size: 0.82em !important; font-weight: 600 !important; letter-spacing: 0.04em !important; } /* Chat */ #chat-panel .message { font-size: 0.9em !important; } #chat-send { background: #2563eb !important; color: white !important; border: none !important; } """ with gr.Blocks(title="OSINT Intelligence Platform", css=CSS, theme=gr.themes.Base()) as demo: brief_state = gr.State(None) gr.HTML("""
LIVE · OPEN-SOURCE INTELLIGENCE

Know before you go.
Any country. One click. 90 seconds.

A real-time threat picture for any destination on earth — conflict events, breaking news, government travel warnings, airspace closures and embassy contacts, fused into a single brief. Pick a country and read the room before you ever set foot in it. *Note: Only reflects British FCDO and US State Dept Risk Ratings.*

⚔️ ACLED conflict data 📰 27 live news feeds 🗺️ State Dept advisories ✈️ Airspace & NOTAMs 🌐 Live web search 💬 Ask-the-analyst chat
""") with gr.Row(equal_height=False): with gr.Column(scale=1, min_width=300, elem_id="config-panel"): gr.HTML('

COLLECTION PARAMETERS

') country_input = gr.Dropdown( choices=DESTINATION_COUNTRIES, value=None, label="Target Country / Region", info="Type to search.", filterable=True, allow_custom_value=True, ) passport_input = gr.Dropdown( choices=PASSPORT_COUNTRIES, value="Not specified", label="Passport Country", info="Enables embassy/consulate data in the report.", filterable=True, ) days_back = gr.Slider( minimum=7, maximum=90, value=14, step=7, label="ACLED lookback window (days)", ) with gr.Accordion("News Sources", open=False): with gr.Row(): select_all_btn = gr.Button("Deselect All", size="sm", scale=1) rss_sources = gr.CheckboxGroup( choices=RSS_SOURCE_OPTIONS, value=[v for _, v in RSS_SOURCE_OPTIONS], label="", ) analyze_btn = gr.Button( "⚡ RUN INTELLIGENCE COLLECTION", variant="primary", elem_id="analyze-btn", ) with gr.Column(scale=3): with gr.Tabs(): with gr.Tab("🛡️ INTELLIGENCE BRIEF"): brief_output = gr.HTML( value="""
🛡️
AWAITING COLLECTION
Configure target and click RUN INTELLIGENCE COLLECTION
""" ) with gr.Tab("📡 SOURCE TRACE"): trace_output = gr.Textbox( label="Raw collected data + LLM output", lines=35, interactive=False, placeholder="All source data and raw LLM output will appear here...", ) with gr.Tab("💬 ANALYST CHAT"): gr.HTML( '
' '💡 Run an analysis first — the assistant will have full brief context. ' 'Ask about safe zones, evacuation routes, historical context, threat actors, etc.' '
' ) chatbot = gr.Chatbot( label="", height=400, type="messages", show_label=False, ) with gr.Row(): chat_input = gr.Textbox( placeholder="Ask a follow-up intelligence question...", label="", scale=5, lines=1, show_label=False, ) chat_send_btn = gr.Button("Send", variant="primary", scale=1, elem_id="chat-send") chat_clear_btn = gr.Button("🗑 Clear conversation", size="sm") with gr.Row(): export_btn = gr.Button( "📄 Export PDF Report", variant="secondary", interactive=False, elem_id="export-btn", scale=1, ) pdf_file = gr.File( label="Download PDF Report", visible=False, interactive=False, type="filepath", ) pdf_error = gr.Markdown(visible=False, elem_id="pdf-error") gr.Examples( examples=EXAMPLE_QUERIES, inputs=[country_input, rss_sources, days_back], label="Quick-start examples", ) _ALL_SOURCE_KEYS = [v for _, v in RSS_SOURCE_OPTIONS] def toggle_sources(current_values): if len(current_values) == len(_ALL_SOURCE_KEYS): return gr.update(value=[]), gr.update(value="Select All") return gr.update(value=_ALL_SOURCE_KEYS), gr.update(value="Deselect All") select_all_btn.click( fn=toggle_sources, inputs=[rss_sources], outputs=[rss_sources, select_all_btn], ) analyze_btn.click( fn=run_analysis, inputs=[country_input, passport_input, rss_sources, days_back], outputs=[brief_output, trace_output, brief_state], ).then( fn=lambda b: gr.update(interactive=b is not None), inputs=[brief_state], outputs=[export_btn], ) def _export_pdf(brief): if brief is None: return gr.update(visible=False), gr.update(visible=False) try: path = generate_pdf(brief) return gr.update(value=path, visible=True), gr.update(visible=False) except Exception as exc: import traceback tb = traceback.format_exc() return ( gr.update(visible=False), gr.update( value=f"⚠️ **PDF export failed:** `{exc}`\n\n```\n{tb}\n```", visible=True, ), ) export_btn.click( fn=_export_pdf, inputs=[brief_state], outputs=[pdf_file, pdf_error], ) # Chat event handlers _chat_inputs = [chat_input, chatbot, brief_state, country_input] _chat_outputs = [chatbot, chat_input] chat_send_btn.click( fn=chat_with_analyst, inputs=_chat_inputs, outputs=_chat_outputs, ) chat_input.submit( fn=chat_with_analyst, inputs=_chat_inputs, outputs=_chat_outputs, ) chat_clear_btn.click(fn=lambda: [], outputs=[chatbot]) gr.HTML("""
⚠️ UNCLASSIFIED · AI-generated from open sources · Not for operational use without verification Data: ACLED · Public RSS · State Dept · EASA/AviationWeather · Tavily · Modal GPU
""") if __name__ == "__main__": demo.launch()