""" tools.py — OSINT data source tools for the agentic analyst loop. Required Space Secrets: ACLED_USERNAME — your myACLED account email (from https://developer.acleddata.com) ACLED_PASSWORD — your myACLED account password Optional Space Secrets (enhance airspace data): AVWX_TOKEN — free API token from https://avwx.rest (enables per-airport NOTAM lookups) """ import os import re import time import threading from concurrent.futures import ThreadPoolExecutor, as_completed import feedparser import requests from datetime import datetime, timedelta from smolagents import tool # --------------------------------------------------------------------------- # ACLED OAuth 2.0 token cache # --------------------------------------------------------------------------- ACLED_TOKEN_URL = "https://acleddata.com/oauth/token" ACLED_BASE = "https://acleddata.com/api/acled/read" _token_cache: dict = { "access_token": None, "expires_at": 0.0, "lock": threading.Lock(), } def _get_acled_token() -> str: """Return a valid Bearer token, refreshing via OAuth password grant if needed.""" with _token_cache["lock"]: now = time.time() if _token_cache["access_token"] and now < _token_cache["expires_at"]: return _token_cache["access_token"] username = os.environ.get("ACLED_USERNAME", "").strip() password = os.environ.get("ACLED_PASSWORD", "").strip() if not username or not password: raise EnvironmentError( "ACLED credentials missing. " "Add ACLED_USERNAME and ACLED_PASSWORD as Space secrets " "(Settings → Variables and Secrets). " "Register free at https://developer.acleddata.com" ) try: resp = requests.post( ACLED_TOKEN_URL, data={ "grant_type": "password", "client_id": "acled", "username": username, "password": password, }, headers={"Content-Type": "application/x-www-form-urlencoded"}, timeout=20, ) except requests.RequestException as e: raise EnvironmentError(f"ACLED token request failed (network): {e}") from e if resp.status_code != 200: raise EnvironmentError( f"ACLED token request failed (HTTP {resp.status_code}): {resp.text[:300]}" ) try: token_data = resp.json() except ValueError: raise EnvironmentError(f"ACLED token response not JSON: {resp.text[:300]}") access_token = token_data.get("access_token") if not access_token: raise EnvironmentError( f"ACLED token response missing 'access_token'. Got: {token_data}" ) expires_in = token_data.get("expires_in", 86400) _token_cache["access_token"] = access_token _token_cache["expires_at"] = now + int(expires_in) - 300 return access_token def _strip_html(text: str) -> str: """Remove HTML tags and clean up whitespace.""" clean = re.sub(r"<[^>]+>", " ", text) clean = re.sub(r"\s+", " ", clean) return clean.strip() # --------------------------------------------------------------------------- # ACLED Tool # --------------------------------------------------------------------------- @tool def fetch_acled_events(country: str, days_back: int = 14, limit: int = 15) -> str: """ Fetches recent armed conflict events from ACLED for a given country. Returns dates, locations, actor names, event types, and fatality counts. Args: country: Country name to query (e.g. 'Sudan', 'Ukraine', 'Mexico'). days_back: How many days back to search (default 14). limit: Maximum number of events to return (default 15, max 50). """ try: token = _get_acled_token() except EnvironmentError as e: return f"[ACLED] Auth error: {e}" since = (datetime.utcnow() - timedelta(days=days_back)).strftime("%Y-%m-%d") params = { "country": country, "event_date": since, "event_date_where": ">=", "limit": min(limit, 50), "fields": "event_date|event_type|sub_event_type|actor1|actor2|location|admin1|fatalities|notes", "_format": "json", } headers = {"Authorization": f"Bearer {token}"} try: resp = requests.get(ACLED_BASE, params=params, headers=headers, timeout=20) except requests.RequestException as e: return f"[ACLED] Request failed (network): {e}" if resp.status_code != 200: return f"[ACLED] HTTP {resp.status_code}: {resp.text[:300]}" try: data = resp.json() except ValueError: return f"[ACLED] Could not parse response as JSON. Raw: {resp.text[:300]}" api_status = data.get("status") if api_status != 200: msg = data.get("error") or data.get("message") or str(data)[:300] return f"[ACLED] API status {api_status}: {msg}" events = data.get("data", []) if not events: return f"[ACLED] No events found for '{country}' in the last {days_back} days." lines = [f"[ACLED] {len(events)} events in {country} (last {days_back} days):\n"] total_fatalities = 0 for ev in events: fatalities = int(ev.get("fatalities") or 0) total_fatalities += fatalities actor2_str = f" vs {ev['actor2']}" if ev.get("actor2") else "" lines.append( f"* {ev.get('event_date', '?')} | {ev.get('event_type', '?')} / {ev.get('sub_event_type', '')} | " f"{ev.get('location', '?')}, {ev.get('admin1', '?')} | " f"{ev.get('actor1', '?')}{actor2_str} | " f"Fatalities: {fatalities} | " f"Notes: {str(ev.get('notes', ''))[:80]}" ) lines.append(f"\nTotal reported fatalities: {total_fatalities}") return "\n".join(lines) # --------------------------------------------------------------------------- # RSS Tool # --------------------------------------------------------------------------- RSS_FEED_REGISTRY = { # --- General world news --- "bbc_world": "https://feeds.bbci.co.uk/news/world/rss.xml", "al_jazeera": "https://www.aljazeera.com/xml/rss/all.xml", "france24": "https://www.france24.com/en/rss", "euronews": "https://feeds.feedburner.com/euronews/en/news/", "npr_world": "https://feeds.npr.org/1004/rss.xml", "sky_news": "https://feeds.skynews.com/feeds/rss/world.xml", "un_news": "https://news.un.org/feed/subscribe/en/news/all/rss.xml", "ibt": "https://www.ibtimes.com/rss", # --- Regional: Middle East --- "middle_east_eye": "https://www.middleeasteye.net/rss", "al_monitor": "https://www.almonitor.com/rss", "arab_news": "https://www.arabnews.com/rss.xml", # --- Regional: Africa --- "allafrica": "https://allafrica.com/tools/headlines/rdf/latest/headlines.rdf", # --- Regional: Asia-Pacific --- "radio_free_asia": "https://www.rfa.org/english/rss2.xml", "scmp": "https://www.scmp.com/rss/91/feed", # --- Regional: South Asia --- "dawn": "https://www.dawn.com/feeds/home", # --- Regional: Russia / Eastern Europe --- "moscow_times": "https://www.themoscowtimes.com/rss/news", # --- OSINT / investigative --- "bellingcat": "https://www.bellingcat.com/feed/", "the_intercept": "https://theintercept.com/feed/?rss", "occrp": "https://www.occrp.org/en/component/rssfeed/index.xml", # --- Policy / security analysis --- "crisis_group": "https://www.crisisgroup.org/rss.xml", "war_on_rocks": "https://warontherocks.com/feed/", "just_security": "https://www.justsecurity.org/feed/", "defense_one": "https://www.defenseone.com/rss/all/", "cipher_brief": "https://www.thecipherbrief.com/feed", "stimson": "https://www.stimson.org/feed/", # --- Human rights --- "hrw": "https://www.hrw.org/rss.xml", "amnesty": "https://www.amnesty.org/en/feed/", } AVIATION_RSS_SOURCES = { "the_aviationist": "https://theaviationist.com/feed/", "flight_global": "https://www.flightglobal.com/rss/", "thedrive_warzone": "https://www.thedrive.com/the-war-zone/rss", "defense_news": "https://www.defensenews.com/arc/outboundfeeds/rss/", "air_force_mag": "https://www.airforcemag.com/rss/", "overt_defense": "https://www.overtdefense.com/feed/", } SCAN_LIMIT = 40 # Max entries scanned per feed (filtered down by keyword match) NOTABLE_SIGNALS = [ "killed", "dead", "deaths", "fatalities", "massacre", "attack", "attacked", "explosion", "bomb", "bombing", "shooting", "gunfire", "clash", "clashes", "offensive", "invasion", "coup", "crisis", "emergency", "arrest", "arrested", "protest", "riot", "siege", "hostage", "kidnap", "cartel", "militia", "sanctions", "airstrike", "drone", "ceasefire", "peace", "agreement", "earthquake", "flood", "disaster", "outbreak", "epidemic", ] AIRSPACE_SIGNALS = [ "notam", "no-fly", "no fly", "airspace", "flight ban", "flight restriction", "fir", "air defense", "air defence", "missile", "anti-aircraft", "flight advisory", "aviation", "closed airspace", "restricted airspace", "drone", "uav", "uas", "aircraft", "airline", "airport", "runway", "air traffic", "eurocontrol", "icao", "overflight", "air corridor", ] def _is_notable(title: str, summary: str) -> bool: text = (title + " " + summary).lower() return any(signal in text for signal in NOTABLE_SIGNALS) def _fetch_one_feed(key: str, url: str, keywords: list) -> tuple: """Fetch and filter a single RSS feed. Returns (articles, error_str).""" try: feed = feedparser.parse(url) if feed.bozo and not feed.entries: return [], f"[{key}] Parse error: {feed.bozo_exception}" except Exception as e: return [], f"[{key}] Exception: {e}" source_name = feed.feed.get("title", key) articles = [] for entry in feed.entries[:SCAN_LIMIT]: title = entry.get("title", "").strip() raw_summary = entry.get("summary", entry.get("description", "")) full_summary = _strip_html(raw_summary) published = entry.get("published", entry.get("updated", "")) link = entry.get("link", "") # Match against the FULL text (title + full summary + tags + link slug), # not the 150-char display truncation. This was silently dropping articles # where the country name appeared deeper in the body. tag_text = " ".join(t.get("term", "") for t in entry.get("tags", []) if isinstance(t, dict)) haystack = f"{title} {full_summary} {tag_text} {link}".lower() if not any(kw in haystack for kw in keywords): continue summary = full_summary[:150] articles.append({ "source_key": key, "source_name": source_name, "published": published, "title": title, "summary": summary, "url": link, "notable": _is_notable(title, full_summary), }) return articles, "" _COUNTRY_ALIASES = { "myanmar": ["myanmar", "burma"], "burma": ["myanmar", "burma"], "ivory coast": ["ivory coast", "côte d'ivoire", "cote d'ivoire"], "drc": ["drc", "congo", "kinshasa"], "congo (drc)": ["drc", "congo", "kinshasa"], "car": ["central african republic", "car"], "uae": ["uae", "united arab emirates", "emirates"], "united arab emirates": ["uae", "united arab emirates", "emirates"], "united states": ["united states", "u.s.", "america", "washington"], "united kingdom": ["united kingdom", "u.k.", "britain", "british", "london"], "south korea": ["south korea", "korea", "seoul"], "north korea": ["north korea", "pyongyang", "dprk"], "dr congo": ["drc", "congo", "kinshasa"], } def collect_rss_articles( topic: str, sources: str = "bbc_world,al_jazeera,france24", max_articles: int = 10, ) -> tuple: """ Structured RSS collector. Returns (articles, feed_errors) where each article is a dict with source_key, source_name, published, title, summary, url, notable. This is the canonical data path — callers that want real, verifiable articles (with genuine URLs) should use this directly rather than parsing the string produced by fetch_rss_headlines or relying on an LLM to transcribe them. """ source_keys = [s.strip() for s in sources.split(",") if s.strip()] base_keywords = [w.lower() for w in topic.lower().split() if len(w) > 2] topic_lower = topic.lower().strip() extra = _COUNTRY_ALIASES.get(topic_lower, []) keywords = list(dict.fromkeys(base_keywords + extra)) feed_errors = [] valid_feeds = [(k, RSS_FEED_REGISTRY[k]) for k in source_keys if k in RSS_FEED_REGISTRY] feed_errors += [f"Unknown source key: '{k}'" for k in source_keys if k not in RSS_FEED_REGISTRY] all_articles = [] with ThreadPoolExecutor(max_workers=12) as executor: futures = { executor.submit(_fetch_one_feed, key, url, keywords): key for key, url in valid_feeds } for future in as_completed(futures): feed_arts, err = future.result() if err: feed_errors.append(err) all_articles.extend(feed_arts) # De-duplicate by URL (same wire story syndicated across feeds) seen = set() deduped = [] for a in all_articles: key = a.get("url") or a.get("title") if key in seen: continue seen.add(key) deduped.append(a) # Notable articles first, then most recent deduped.sort(key=lambda a: a.get("published", ""), reverse=True) deduped.sort(key=lambda a: not a["notable"]) return deduped[:max_articles], feed_errors @tool def fetch_rss_headlines( topic: str, sources: str = "bbc_world,al_jazeera,france24", max_articles: int = 10, ) -> str: """ Fetches recent RSS news headlines related to a topic or region. Returns structured article records including title, source, date, summary, URL, and a 'notable' flag for high-signal security/conflict articles. Args: topic: Keyword or region to filter headlines (e.g. 'Mexico', 'Sudan'). Single keywords work best. sources: Comma-separated source keys. Available: bbc_world, al_jazeera, france24, euronews, npr_world, sky_news, un_news, ibt, middle_east_eye, al_monitor, arab_news, allafrica, radio_free_asia, scmp, dawn, moscow_times, bellingcat, the_intercept, occrp, crisis_group, war_on_rocks, just_security, defense_one, cipher_brief, stimson, hrw, amnesty. max_articles: Maximum total articles to return across all sources (default 10). """ articles, feed_errors = collect_rss_articles(topic, sources, max_articles) if not articles: err_detail = "; ".join(feed_errors) if feed_errors else "no entries matched" return ( f"[RSS] No articles matched '{topic}'. {err_detail}\n" "Tip: Try a shorter single-word keyword (e.g. 'Mexico' not 'Mexico violence')." ) lines = [f"[RSS] {len(articles)} articles found for '{topic}':\n"] notable_count = sum(1 for a in articles if a["notable"]) lines.append(f"Notable (high-signal) articles: {notable_count} of {len(articles)}\n") for i, a in enumerate(articles, 1): flag = " *** NOTABLE ***" if a["notable"] else "" lines.append( f"[{i}] {a['source_name']} | {a['published']}{flag}\n" f" Title: {a['title']}\n" f" Summary: {a['summary']}\n" f" URL: {a['url']}\n" f" Notable: {a['notable']}" ) if feed_errors: lines.append("\n--- Feed warnings ---") lines.extend(feed_errors) return "\n\n".join(lines) # --------------------------------------------------------------------------- # US State Department Travel Advisory Tool # --------------------------------------------------------------------------- _ADVISORY_API = "https://cadataapi.state.gov/api/TravelAdvisories" _RISK_KEYWORDS = { "crime": "Crime", "terrorism": "Terrorism", "civil unrest": "Civil Unrest", "health": "Health", "natural disaster": "Natural Disaster", "kidnapping": "Kidnapping", "wrongful detention": "Wrongful Detention", "piracy": "Piracy", "maritime": "Maritime", } @tool def fetch_travel_advisory(country: str) -> str: """ Fetches the current US State Department travel advisory for a country. Returns the advisory level (1–4), risk categories, publication date, a plain-text summary, and a link to the full advisory. Advisory levels: 1 = Exercise Normal Precautions 2 = Exercise Increased Caution 3 = Reconsider Travel 4 = Do Not Travel Args: country: Country name to look up (e.g. 'Sudan', 'Ukraine', 'Haiti'). """ try: resp = requests.get(_ADVISORY_API, timeout=20) resp.raise_for_status() advisories = resp.json() except requests.RequestException as e: return f"[Travel Advisory] Request failed: {e}" except ValueError: return "[Travel Advisory] Could not parse API response as JSON." country_lower = country.lower().strip() match = None for entry in advisories: title = entry.get("Title", "") dest = title.split(" - Level ")[0].strip() if country_lower in dest.lower(): match = entry break if not match: return ( f"[Travel Advisory] No advisory found for '{country}'. " "Check spelling or try the country's common English name." ) title = match.get("Title", "") link = match.get("Link", "") published = match.get("Published", match.get("Updated", "")) raw_summary = match.get("Summary", "") summary = _strip_html(raw_summary)[:250] level_match = re.search(r"Level\s+(\d)", title, re.IGNORECASE) level_num = level_match.group(1) if level_match else "Unknown" summary_lower = summary.lower() indicators = [ label for keyword, label in _RISK_KEYWORDS.items() if keyword in summary_lower ] date_str = published[:10] if published else "" lines = [ f"[Travel Advisory] {title}", f"Risk Categories: {', '.join(indicators) if indicators else 'See summary'}", f"Published: {date_str}", f"Summary: {summary}", ] if link: lines.append(f"Full Advisory: {link}") return "\n".join(lines) # --------------------------------------------------------------------------- # Airspace helpers — EASA CZIBs, AviationWeather SIGMETs, AVWX NOTAMs # --------------------------------------------------------------------------- COUNTRY_ICAO_MAP: dict = { "ukraine": ["UKBB", "UKKK", "UKLL"], "russia": ["UUEE", "UUDD", "ULLI"], "sudan": ["HSSS"], "myanmar": ["VYYY", "VYBR"], "haiti": ["MTPP"], "syria": ["OSDI", "OSLK"], "iraq": ["ORBI", "ORMM", "ORKK"], "libya": ["HLLT", "HLLB"], "somalia": ["HCMM"], "yemen": ["OYAA", "OYAB"], "afghanistan": ["OAKB"], "ethiopia": ["HAAB"], "nigeria": ["DNMM", "DNKN", "DNAA"], "mali": ["GABS", "GAMB"], "burkina faso": ["DFFD"], "niger": ["DRRN"], "mozambique": ["FQMA"], "central african republic": ["FEFF"], "democratic republic of congo": ["FZAA", "FZNA"], "drc": ["FZAA"], "israel": ["LLBG", "LLHA"], "palestine": ["LVGZ"], "iran": ["OIIE", "OIII", "OIMM"], "pakistan": ["OPKC", "OPLA", "OPPS"], "north korea": ["ZKPY"], "venezuela": ["SVMI", "SVMC"], "mexico": ["MMMX", "MMGL", "MMMY"], "colombia": ["SKBO", "SKCL"], "lebanon": ["OLBA"], "georgia": ["UGTB"], "armenia": ["UDYZ"], "azerbaijan": ["UBBB"], } def _kws(country: str) -> list: return [w.lower() for w in country.lower().split() if len(w) > 2] def _fetch_easa_czibs(country: str) -> list: """Scrape EASA's live Conflict Zone Information Bulletins table.""" url = "https://www.easa.europa.eu/en/domains/air-operations/czibs" try: resp = requests.get( url, timeout=15, headers={"User-Agent": "Mozilla/5.0 (compatible; OSINTBot/1.0)"}, ) resp.raise_for_status() except requests.RequestException as e: return [f"[EASA CZIB] Request failed: {e}"] keywords = _kws(country) html_text = resp.text rows = re.findall(r"]*>(.*?)", html_text, re.DOTALL | re.IGNORECASE) results = [] for row in rows: cell_text = re.sub(r"<[^>]+>", " ", row) cell_text = re.sub(r"\s+", " ", cell_text).strip() if cell_text and any(kw in cell_text.lower() for kw in keywords): link_match = re.search(r'href="(/en/domains/air-operations/czibs/[^"]+)"', row) link = f"https://www.easa.europa.eu{link_match.group(1)}" if link_match else "" entry = f"EASA CZIB | {cell_text}" if link: entry += f" | {link}" results.append(entry) return results if results else [f"[EASA CZIB] No active conflict zone bulletins found for '{country}'."] def _fetch_sigmets(country: str) -> list: """Fetch active international SIGMETs from AviationWeather.gov.""" url = "https://aviationweather.gov/api/data/isigmet?format=json" try: resp = requests.get(url, timeout=15) resp.raise_for_status() sigmets = resp.json() except (requests.RequestException, ValueError) as e: return [f"[SIGMET] Request failed: {e}"] if not isinstance(sigmets, list): return ["[SIGMET] Unexpected response format."] keywords = _kws(country) results = [] for s in sigmets: fir_name = s.get("firName", "") raw = s.get("rawSigmet", "") searchable = (fir_name + " " + raw).lower() if not any(kw in searchable for kw in keywords): continue try: valid_from = datetime.utcfromtimestamp(s["validTimeFrom"]).strftime("%Y-%m-%d %H:%MZ") valid_to = datetime.utcfromtimestamp(s["validTimeTo"]).strftime("%Y-%m-%d %H:%MZ") except (KeyError, TypeError, OSError): valid_from = valid_to = "?" base = s.get("base") or "SFC" top = s.get("top", "?") results.append( f"SIGMET | {fir_name} | Hazard: {s.get('hazard','?')} {s.get('qualifier','')}" f" | Valid: {valid_from} → {valid_to}" f" | FL: {base}–{top}" f" | {raw[:150]}" ) return results if results else [f"[SIGMET] No active SIGMETs found for '{country}'."] def _fetch_avwx_notams(country: str, token: str) -> list: """Fetch NOTAMs via AVWX for the country's main airports.""" country_lower = country.lower() icao_codes = COUNTRY_ICAO_MAP.get(country_lower) if not icao_codes: for k, v in COUNTRY_ICAO_MAP.items(): if country_lower in k or k in country_lower: icao_codes = v break if not icao_codes: return [f"[NOTAM] No ICAO airport codes mapped for '{country}'. Skipping AVWX lookup."] headers = {"Authorization": f"BEARER {token}"} results = [] for icao in icao_codes[:2]: try: resp = requests.get( f"https://avwx.rest/api/notam/{icao}", headers=headers, timeout=12, ) if resp.status_code == 401: return ["[NOTAM] AVWX token invalid or expired — check AVWX_TOKEN secret."] resp.raise_for_status() data = resp.json() except (requests.RequestException, ValueError) as e: results.append(f"[NOTAM] {icao}: {e}") continue notam_list = data if isinstance(data, list) else data.get("data", []) if not notam_list: results.append(f"[NOTAM] {icao}: No active NOTAMs.") continue for n in notam_list[:6]: raw = ( n.get("raw") or n.get("text", {}).get("repr", "") or str(n) )[:200] results.append(f"NOTAM | {icao} | {raw}") time.sleep(0.2) return results if results else [f"[NOTAM] No NOTAMs returned for '{country}'."] # --------------------------------------------------------------------------- # Airspace tool — EASA CZIBs + SIGMETs + AVWX NOTAMs + aviation RSS # --------------------------------------------------------------------------- @tool def fetch_airspace_status(country: str, max_articles: int = 12) -> str: """ Fetches airspace disruption intelligence for a given country from three structured sources plus aviation news RSS feeds: 1. EASA Conflict Zone Information Bulletins (CZIBs) — official EU conflict zone airspace warnings (no auth required). 2. AviationWeather.gov international SIGMETs — active hazards (thunderstorms, volcanic ash, turbulence, tropical cyclones) within the country's FIRs (no auth required). 3. AVWX NOTAMs — per-airport notices to airmen for the country's main airports (requires optional AVWX_TOKEN Space secret). 4. Aviation news RSS feeds — filtered for country + airspace keywords. Args: country: Country name to query (e.g. 'Ukraine', 'Sudan', 'Libya'). max_articles: Maximum RSS articles to include (default 12). """ keywords = _kws(country) sections: list = [] # 1. EASA CZIBs sections.append("=== EASA Conflict Zone Bulletins (CZIBs) ===") sections.extend(_fetch_easa_czibs(country)) # 2. AviationWeather SIGMETs sections.append("\n=== Active SIGMETs (AviationWeather.gov) ===") sections.extend(_fetch_sigmets(country)) # 3. AVWX NOTAMs (optional) avwx_token = os.environ.get("AVWX_TOKEN", "").strip() sections.append("\n=== NOTAMs (AVWX) ===") if avwx_token: sections.extend(_fetch_avwx_notams(country, avwx_token)) else: sections.append("[NOTAM] AVWX_TOKEN not set — skipping NOTAM lookup.") # 4. Aviation RSS news — parallel fetch all_sources = {**AVIATION_RSS_SOURCES, **{ k: v for k, v in RSS_FEED_REGISTRY.items() if k in ("bbc_world", "al_jazeera", "france24") }} def _fetch_avn_feed(key_url): key, url = key_url try: resp = requests.get( url, timeout=10, headers={"User-Agent": "Mozilla/5.0 (compatible; OSINTBot/1.0)"}, ) feed = feedparser.parse(resp.text) except Exception as e: return [], f"[{key}] Exception: {e}" if feed.bozo and not feed.entries: return [], f"[{key}] Feed error" source_name = feed.feed.get("title", key) hits = [] for entry in feed.entries[:SCAN_LIMIT]: title = entry.get("title", "").strip() raw_summary = entry.get("summary", entry.get("description", "")) summary = _strip_html(raw_summary)[:300] searchable = (title + " " + summary).lower() if not ( any(kw in searchable for kw in keywords) and any(sig in searchable for sig in AIRSPACE_SIGNALS) ): continue hits.append({ "source_name": source_name, "published": entry.get("published", entry.get("updated", "")), "title": title, "summary": summary, "url": entry.get("link", ""), }) return hits, None articles = [] feed_errors = [] with ThreadPoolExecutor(max_workers=8) as ex: futures = {ex.submit(_fetch_avn_feed, item): item[0] for item in all_sources.items()} for fut in as_completed(futures): hits, err = fut.result() if err: feed_errors.append(err) else: articles.extend(hits) articles = articles[:max_articles] sections.append(f"\n=== Aviation News ({len(articles)} articles) ===") if articles: for i, a in enumerate(articles, 1): sections.append( f"[{i}] {a['source_name']} | {a['published']}\n" f" Title: {a['title']}\n" f" Summary: {a['summary']}\n" f" URL: {a['url']}" ) else: sections.append(f"No aviation news matched for '{country}'.") if feed_errors: sections.append(f"\n[Feed warnings: {'; '.join(feed_errors)}]") return "\n".join(sections) # --------------------------------------------------------------------------- # Tavily Web Search Tool # --------------------------------------------------------------------------- TAVILY_API_URL = "https://api.tavily.com/search" def _tavily_raw(query: str, max_results: int, topic: str = "news") -> list: """ Internal structured Tavily call. Returns a list of result dicts, or [] on failure. topic="news" biases Tavily toward news sources. """ api_key = os.environ.get("TAVILY_API_KEY", "").strip() if not api_key: return [] payload = { "api_key": api_key, "query": query, "max_results": min(max_results, 20), "search_depth": "basic", "topic": topic, "include_answer": False, } try: resp = requests.post(TAVILY_API_URL, json=payload, timeout=20) resp.raise_for_status() return resp.json().get("results", []) except Exception: return [] def collect_tavily_articles(country: str, max_results: int = 10) -> list: """ Run two parallel Tavily news queries for a country and return structured article dicts compatible with NewsItem / collect_rss_articles output. Returns a list of dicts with keys: title, source_name, published, summary, url, notable, source_key="tavily". """ year = datetime.utcnow().year queries = [ f"{country} news {year}", f"{country} conflict security latest", ] raw_results: list = [] with ThreadPoolExecutor(max_workers=2) as ex: futs = [ex.submit(_tavily_raw, q, max_results, "news") for q in queries] for f in as_completed(futs): raw_results.extend(f.result()) # De-duplicate by URL seen: set = set() articles = [] for r in raw_results: url = r.get("url", "") if not url or url in seen: continue seen.add(url) content = (r.get("content") or "")[:200] title = r.get("title") or "" haystack = (title + " " + content).lower() notable = any(sig in haystack for sig in NOTABLE_SIGNALS) # Extract hostname as source name try: from urllib.parse import urlparse source_name = urlparse(url).hostname.replace("www.", "") except Exception: source_name = "Web" articles.append({ "source_key": "tavily", "source_name": source_name, "published": r.get("published_date", ""), "title": title, "summary": content, "url": url, "notable": notable, }) articles.sort(key=lambda a: not a["notable"]) return articles[:max_results] @tool def tavily_web_search(query: str, max_results: int = 8) -> str: """ Performs a real-time web search using the Tavily API and returns summarised results with titles, URLs, and content snippets. Use this to find breaking news, recent reports, or any information not covered by the other tools. Args: query: The search query (e.g. 'Sudan conflict 2025 casualties'). max_results: Maximum number of results to return (default 8, max 20). """ api_key = os.environ.get("TAVILY_API_KEY", "").strip() if not api_key: return ( "[Tavily] TAVILY_API_KEY is not set. " "Add it in Space Settings > Variables and Secrets." ) payload = { "api_key": api_key, "query": query, "max_results": min(max_results, 20), "search_depth": "advanced", "include_answer": True, } try: resp = requests.post(TAVILY_API_URL, json=payload, timeout=20) resp.raise_for_status() data = resp.json() except requests.RequestException as e: return f"[Tavily] Request failed: {e}" except ValueError: return "[Tavily] Could not parse API response as JSON." lines = [f"[Tavily Web Search] Query: '{query}'\n"] answer = data.get("answer", "") if answer: lines.append(f"AI Answer: {answer}\n") results = data.get("results", []) if not results: return f"[Tavily] No results found for '{query}'." for i, r in enumerate(results, 1): title = r.get("title", "No title") url = r.get("url", "") content = r.get("content", "")[:300] score = r.get("score", 0) lines.append( f"[{i}] {title}\n" f" URL: {url}\n" f" Relevance: {score:.2f}\n" f" Snippet: {content}" ) return "\n\n".join(lines) # --------------------------------------------------------------------------- # Helper tool # --------------------------------------------------------------------------- @tool def list_available_sources() -> str: """ Returns a list of all available RSS feed source keys and their URLs. Args: None """ lines = ["Available RSS sources:"] for key, url in RSS_FEED_REGISTRY.items(): lines.append(f" * {key}: {url}") lines.append("\nAviation RSS sources (used by fetch_airspace_status):") for key, url in AVIATION_RSS_SOURCES.items(): lines.append(f" * {key}: {url}") lines.append("\nStructured airspace sources (used by fetch_airspace_status):") lines.append(" * EASA CZIBs (no auth): https://www.easa.europa.eu/en/domains/air-operations/czibs") lines.append(" * AviationWeather SIGMETs (no auth): https://aviationweather.gov/api/data/isigmet") lines.append(" * AVWX NOTAMs (requires AVWX_TOKEN secret): https://avwx.rest") lines.append("\nACLED is also available for structured armed conflict event data.") lines.append("US State Department Travel Advisories are also available via fetch_travel_advisory.") return "\n".join(lines)