Initial commit: ShadowBroker v0.1

Former-commit-id: 8ed321f2ba

backend/services/__init__.py

@@ -0,0 +1 @@
+# Empty init

backend/services/ais_cache.json.REMOVED.git-id

@@ -0,0 +1 @@
+5d33551b09405e7e252c6a11f080a6c9eca50f6b

backend/services/ais_stream.py

@@ -0,0 +1,359 @@
+"""
+AIS Stream WebSocket client for real-time maritime vessel tracking.
+Connects to aisstream.io and maintains a live dictionary of global vessel positions.
+"""
+
+import asyncio
+import json
+import logging
+import threading
+import time
+from datetime import datetime, timezone
+import os
+
+logger = logging.getLogger(__name__)
+
+AIS_WS_URL = "wss://stream.aisstream.io/v0/stream"
+API_KEY = os.environ.get("AIS_API_KEY", "75cc39af03c9cc23c90e8a7b3c3bc2b2a507c5fb")
+
+# AIS vessel type code classification
+# See: https://coast.noaa.gov/data/marinecadastre/ais/VesselTypeCodes2018.pdf
+def classify_vessel(ais_type: int, mmsi: int) -> str:
+    """Classify a vessel by its AIS type code into a rendering category."""
+    if 80 <= ais_type <= 89:
+        return "tanker"        # Oil/Chemical/Gas tankers → RED
+    if 70 <= ais_type <= 79:
+        return "cargo"         # Cargo ships, container vessels → RED
+    if 60 <= ais_type <= 69:
+        return "passenger"     # Cruise ships, ferries → GRAY
+    if ais_type in (36, 37):
+        return "yacht"         # Sailing/Pleasure craft → DARK BLUE
+    if ais_type == 35:
+        return "military_vessel"  # Military → YELLOW
+    # MMSI-based military detection: military MMSIs often start with certain prefixes
+    mmsi_str = str(mmsi)
+    if mmsi_str.startswith("3380") or mmsi_str.startswith("3381"):
+        return "military_vessel"  # US Navy
+    if ais_type in (30, 31, 32, 33, 34):
+        return "other"         # Fishing, towing, dredging, diving, etc.
+    if ais_type in (50, 51, 52, 53, 54, 55, 56, 57, 58, 59):
+        return "other"         # Pilot, SAR, tug, port tender, etc.
+    return "unknown"            # Not yet classified — will update when ShipStaticData arrives
+
+
+# MMSI Maritime Identification Digit (MID) → Country mapping
+# First 3 digits of MMSI (for 9-digit MMSIs) encode the flag state
+MID_COUNTRY = {
+    201: "Albania", 202: "Andorra", 203: "Austria", 204: "Portugal", 205: "Belgium",
+    206: "Belarus", 207: "Bulgaria", 208: "Vatican", 209: "Cyprus", 210: "Cyprus",
+    211: "Germany", 212: "Cyprus", 213: "Georgia", 214: "Moldova", 215: "Malta",
+    216: "Armenia", 218: "Germany", 219: "Denmark", 220: "Denmark", 224: "Spain",
+    225: "Spain", 226: "France", 227: "France", 228: "France", 229: "Malta",
+    230: "Finland", 231: "Faroe Islands", 232: "United Kingdom", 233: "United Kingdom",
+    234: "United Kingdom", 235: "United Kingdom", 236: "Gibraltar", 237: "Greece",
+    238: "Croatia", 239: "Greece", 240: "Greece", 241: "Greece", 242: "Morocco",
+    243: "Hungary", 244: "Netherlands", 245: "Netherlands", 246: "Netherlands",
+    247: "Italy", 248: "Malta", 249: "Malta", 250: "Ireland", 251: "Iceland",
+    252: "Liechtenstein", 253: "Luxembourg", 254: "Monaco", 255: "Portugal",
+    256: "Malta", 257: "Norway", 258: "Norway", 259: "Norway", 261: "Poland",
+    263: "Portugal", 264: "Romania", 265: "Sweden", 266: "Sweden", 267: "Slovakia",
+    268: "San Marino", 269: "Switzerland", 270: "Czech Republic", 271: "Turkey",
+    272: "Ukraine", 273: "Russia", 274: "North Macedonia", 275: "Latvia",
+    276: "Estonia", 277: "Lithuania", 278: "Slovenia",
+    301: "Anguilla", 303: "Alaska", 304: "Antigua", 305: "Antigua",
+    306: "Netherlands Antilles", 307: "Aruba", 308: "Bahamas", 309: "Bahamas",
+    310: "Bermuda", 311: "Bahamas", 312: "Belize", 314: "Barbados", 316: "Canada",
+    319: "Cayman Islands", 321: "Costa Rica", 323: "Cuba", 325: "Dominica",
+    327: "Dominican Republic", 329: "Guadeloupe", 330: "Grenada", 331: "Greenland",
+    332: "Guatemala", 334: "Honduras", 336: "Haiti", 338: "United States",
+    339: "Jamaica", 341: "Saint Kitts", 343: "Saint Lucia", 345: "Mexico",
+    347: "Martinique", 348: "Montserrat", 350: "Nicaragua", 351: "Panama",
+    352: "Panama", 353: "Panama", 354: "Panama", 355: "Panama",
+    356: "Panama", 357: "Panama", 358: "Puerto Rico", 359: "El Salvador",
+    361: "Saint Pierre", 362: "Trinidad", 364: "Turks and Caicos",
+    366: "United States", 367: "United States", 368: "United States", 369: "United States",
+    370: "Panama", 371: "Panama", 372: "Panama", 373: "Panama",
+    374: "Panama", 375: "Saint Vincent", 376: "Saint Vincent", 377: "Saint Vincent",
+    378: "British Virgin Islands", 379: "US Virgin Islands",
+    401: "Afghanistan", 403: "Saudi Arabia", 405: "Bangladesh", 408: "Bahrain",
+    410: "Bhutan", 412: "China", 413: "China", 414: "China",
+    416: "Taiwan", 417: "Sri Lanka", 419: "India", 422: "Iran",
+    423: "Azerbaijan", 425: "Iraq", 428: "Israel", 431: "Japan",
+    432: "Japan", 434: "Turkmenistan", 436: "Kazakhstan", 437: "Uzbekistan",
+    438: "Jordan", 440: "South Korea", 441: "South Korea", 443: "Palestine",
+    445: "North Korea", 447: "Kuwait", 450: "Lebanon", 451: "Kyrgyzstan",
+    453: "Macao", 455: "Maldives", 457: "Mongolia", 459: "Nepal",
+    461: "Oman", 463: "Pakistan", 466: "Qatar", 468: "Syria",
+    470: "UAE", 472: "Tajikistan", 473: "Yemen", 475: "Tonga",
+    477: "Hong Kong", 478: "Bosnia",
+    501: "Antarctica", 503: "Australia", 506: "Myanmar",
+    508: "Brunei", 510: "Micronesia", 511: "Palau", 512: "New Zealand",
+    514: "Cambodia", 515: "Cambodia", 516: "Christmas Island",
+    518: "Cook Islands", 520: "Fiji", 523: "Cocos Islands",
+    525: "Indonesia", 529: "Kiribati", 531: "Laos", 533: "Malaysia",
+    536: "Northern Mariana Islands", 538: "Marshall Islands",
+    540: "New Caledonia", 542: "Niue", 544: "Nauru", 546: "French Polynesia",
+    548: "Philippines", 553: "Papua New Guinea", 555: "Pitcairn",
+    557: "Solomon Islands", 559: "American Samoa", 561: "Samoa",
+    563: "Singapore", 564: "Singapore", 565: "Singapore", 566: "Singapore",
+    567: "Thailand", 570: "Tonga", 572: "Tuvalu", 574: "Vietnam",
+    576: "Vanuatu", 577: "Vanuatu", 578: "Wallis and Futuna",
+    601: "South Africa", 603: "Angola", 605: "Algeria", 607: "Benin",
+    609: "Botswana", 610: "Burundi", 611: "Cameroon", 612: "Cape Verde",
+    613: "Central African Republic", 615: "Congo", 616: "Comoros",
+    617: "DR Congo", 618: "Ivory Coast", 619: "Djibouti",
+    620: "Egypt", 621: "Equatorial Guinea", 622: "Ethiopia",
+    624: "Eritrea", 625: "Gabon", 626: "Gambia", 627: "Ghana",
+    629: "Guinea", 630: "Guinea-Bissau", 631: "Kenya", 632: "Lesotho",
+    633: "Liberia", 634: "Liberia", 635: "Liberia", 636: "Liberia",
+    637: "Libya", 642: "Madagascar", 644: "Malawi", 645: "Mali",
+    647: "Mauritania", 649: "Mauritius", 650: "Mozambique",
+    654: "Namibia", 655: "Niger", 656: "Nigeria", 657: "Guinea",
+    659: "Rwanda", 660: "Senegal", 661: "Sierra Leone",
+    662: "Somalia", 663: "South Africa", 664: "Sudan",
+    667: "Tanzania", 668: "Togo", 669: "Tunisia", 670: "Uganda",
+    671: "Egypt", 672: "Tanzania", 674: "Zambia", 675: "Zimbabwe",
+    676: "Comoros", 677: "Tanzania",
+}
+
+def get_country_from_mmsi(mmsi: int) -> str:
+    """Look up flag state from MMSI Maritime Identification Digit."""
+    mmsi_str = str(mmsi)
+    if len(mmsi_str) == 9:
+        mid = int(mmsi_str[:3])
+        return MID_COUNTRY.get(mid, "UNKNOWN")
+    return "UNKNOWN"
+
+
+# Global vessel store: MMSI → vessel dict
+_vessels: dict[int, dict] = {}
+_vessels_lock = threading.Lock()
+_ws_thread: threading.Thread | None = None
+_ws_running = False
+
+import os
+CACHE_FILE = os.path.join(os.path.dirname(__file__), "ais_cache.json")
+
+
+def _save_cache():
+    """Save vessel data to disk for persistence across restarts."""
+    try:
+        with _vessels_lock:
+            # Convert int keys to strings for JSON
+            data = {str(k): v for k, v in _vessels.items()}
+        with open(CACHE_FILE, 'w') as f:
+            json.dump(data, f)
+        logger.info(f"AIS cache saved: {len(data)} vessels")
+    except Exception as e:
+        logger.error(f"Failed to save AIS cache: {e}")
+
+
+def _load_cache():
+    """Load vessel data from disk on startup."""
+    global _vessels
+    if not os.path.exists(CACHE_FILE):
+        return
+    try:
+        with open(CACHE_FILE, 'r') as f:
+            data = json.load(f)
+        now = time.time()
+        stale_cutoff = now - 3600  # Accept vessels up to 1 hour old on restart
+        loaded = 0
+        with _vessels_lock:
+            for k, v in data.items():
+                if v.get("_updated", 0) > stale_cutoff:
+                    _vessels[int(k)] = v
+                    loaded += 1
+        logger.info(f"AIS cache loaded: {loaded} vessels from disk")
+    except Exception as e:
+        logger.error(f"Failed to load AIS cache: {e}")
+
+
+def get_ais_vessels() -> list[dict]:
+    """Return a snapshot of tracked AIS vessels, excluding 'other' type, pruning stale."""
+    now = time.time()
+    stale_cutoff = now - 900  # 15 minutes
+    
+    with _vessels_lock:
+        # Prune stale vessels
+        stale_keys = [k for k, v in _vessels.items() if v.get("_updated", 0) < stale_cutoff]
+        for k in stale_keys:
+            del _vessels[k]
+        
+        result = []
+        for mmsi, v in _vessels.items():
+            v_type = v.get("type", "unknown")
+            # Skip 'other' vessels (fishing, tug, pilot, etc.) to reduce load
+            if v_type == "other":
+                continue
+            # Skip vessels without valid position
+            if not v.get("lat") or not v.get("lng"):
+                continue
+            
+            result.append({
+                "mmsi": mmsi,
+                "name": v.get("name", "UNKNOWN"),
+                "type": v_type,
+                "lat": round(v.get("lat", 0), 5),
+                "lng": round(v.get("lng", 0), 5),
+                "heading": v.get("heading", 0),
+                "sog": round(v.get("sog", 0), 1),
+                "cog": round(v.get("cog", 0), 1),
+                "callsign": v.get("callsign", ""),
+                "destination": v.get("destination", "") or "UNKNOWN",
+                "imo": v.get("imo", 0),
+                "country": get_country_from_mmsi(mmsi),
+            })
+        return result
+
+
+def _ais_stream_loop():
+    """Main loop: spawn node proxy and process messages from stdout."""
+    import subprocess
+    import os
+    
+    proxy_script = os.path.join(os.path.dirname(os.path.dirname(__file__)), "ais_proxy.js")
+    
+    while _ws_running:
+        try:
+            logger.info("Starting Node.js AIS Stream Proxy...")
+            process = subprocess.Popen(
+                ['node', proxy_script, API_KEY],
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                text=True,
+                bufsize=1
+            )
+            
+            # Drain stderr in a background thread to prevent deadlock
+            import threading
+            def _drain_stderr():
+                for errline in iter(process.stderr.readline, ''):
+                    errline = errline.strip()
+                    if errline:
+                        logger.warning(f"AIS proxy stderr: {errline}")
+            threading.Thread(target=_drain_stderr, daemon=True).start()
+            
+            logger.info("AIS Stream proxy started — receiving vessel data")
+            
+            msg_count = 0
+            for raw_msg in iter(process.stdout.readline, ''):
+                if not _ws_running:
+                    process.terminate()
+                    break
+                
+                raw_msg = raw_msg.strip()
+                if not raw_msg:
+                    continue
+                
+                try:
+                    data = json.loads(raw_msg)
+                except json.JSONDecodeError:
+                    continue
+                
+                if "error" in data:
+                    logger.error(f"AIS Stream error: {data['error']}")
+                    continue
+                
+                msg_type = data.get("MessageType", "")
+                metadata = data.get("MetaData", {})
+                message = data.get("Message", {})
+                
+                mmsi = metadata.get("MMSI", 0)
+                if not mmsi:
+                    continue
+                
+                with _vessels_lock:
+                    if mmsi not in _vessels:
+                        _vessels[mmsi] = {"_updated": time.time()}
+                    vessel = _vessels[mmsi]
+                
+                # Update position from PositionReport or StandardClassBPositionReport
+                if msg_type in ("PositionReport", "StandardClassBPositionReport"):
+                    report = message.get(msg_type, {})
+                    lat = report.get("Latitude", metadata.get("latitude", 0))
+                    lng = report.get("Longitude", metadata.get("longitude", 0))
+                    
+                    # Skip invalid positions
+                    if lat == 0 and lng == 0:
+                        continue
+                    if abs(lat) > 90 or abs(lng) > 180:
+                        continue
+                    
+                    with _vessels_lock:
+                        vessel["lat"] = lat
+                        vessel["lng"] = lng
+                        vessel["sog"] = report.get("Sog", 0)
+                        vessel["cog"] = report.get("Cog", 0)
+                        heading = report.get("TrueHeading", 511)
+                        vessel["heading"] = heading if heading != 511 else report.get("Cog", 0)
+                        vessel["_updated"] = time.time()
+                        # Use metadata name if we don't have one yet
+                        if not vessel.get("name") or vessel["name"] == "UNKNOWN":
+                            vessel["name"] = metadata.get("ShipName", "UNKNOWN").strip() or "UNKNOWN"
+                
+                # Update static data from ShipStaticData
+                elif msg_type == "ShipStaticData":
+                    static = message.get("ShipStaticData", {})
+                    ais_type = static.get("Type", 0)
+                    
+                    with _vessels_lock:
+                        vessel["name"] = (static.get("Name", "") or metadata.get("ShipName", "UNKNOWN")).strip() or "UNKNOWN"
+                        vessel["callsign"] = (static.get("CallSign", "") or "").strip()
+                        vessel["imo"] = static.get("ImoNumber", 0)
+                        vessel["destination"] = (static.get("Destination", "") or "").strip().replace("@", "")
+                        vessel["ais_type_code"] = ais_type
+                        vessel["type"] = classify_vessel(ais_type, mmsi)
+                        vessel["_updated"] = time.time()
+                
+                msg_count += 1
+                if msg_count % 5000 == 0:
+                    with _vessels_lock:
+                        # Inline pruning: remove vessels not updated in 15 minutes
+                        prune_cutoff = time.time() - 900
+                        stale = [k for k, v in _vessels.items() if v.get("_updated", 0) < prune_cutoff]
+                        for k in stale:
+                            del _vessels[k]
+                        count = len(_vessels)
+                    if stale:
+                        logger.info(f"AIS pruned {len(stale)} stale vessels")
+                    logger.info(f"AIS Stream: processed {msg_count} messages, tracking {count} vessels")
+                    _save_cache()  # Auto-save every 5000 messages (~60 seconds)
+                        
+        except Exception as e:
+            logger.error(f"AIS proxy connection error: {e}")
+            if _ws_running:
+                logger.info("Restarting AIS proxy in 5 seconds...")
+                time.sleep(5)
+
+
+def _run_ais_loop():
+    """Thread target: run the AIS loop."""
+    try:
+        _ais_stream_loop()
+    except Exception as e:
+        logger.error(f"AIS Stream thread crashed: {e}")
+
+
+def start_ais_stream():
+    """Start the AIS WebSocket stream in a background thread."""
+    global _ws_thread, _ws_running
+    if _ws_thread and _ws_thread.is_alive():
+        logger.info("AIS Stream already running")
+        return
+    
+    # Load cached vessel data from disk
+    _load_cache()
+    
+    _ws_running = True
+    _ws_thread = threading.Thread(target=_run_ais_loop, daemon=True, name="ais-stream")
+    _ws_thread.start()
+    logger.info("AIS Stream background thread started")
+
+
+def stop_ais_stream():
+    """Stop the AIS WebSocket stream and save cache."""
+    global _ws_running
+    _ws_running = False
+    _save_cache()  # Save on shutdown
+    logger.info("AIS Stream stopping...")

backend/services/api_settings.py

@@ -0,0 +1,175 @@
+"""
+API Settings management — serves the API key registry and allows updates.
+Keys are stored in the backend .env file and loaded via python-dotenv.
+"""
+import os
+import re
+from pathlib import Path
+
+# Path to the backend .env file
+ENV_PATH = Path(__file__).parent.parent / ".env"
+
+# ---------------------------------------------------------------------------
+# API Registry — every external service the dashboard depends on
+# ---------------------------------------------------------------------------
+API_REGISTRY = [
+    {
+        "id": "opensky_client_id",
+        "env_key": "OPENSKY_CLIENT_ID",
+        "name": "OpenSky Network — Client ID",
+        "description": "OAuth2 client ID for the OpenSky Network API. Provides global flight state vectors with 400 requests/day.",
+        "category": "Aviation",
+        "url": "https://opensky-network.org/",
+        "required": True,
+    },
+    {
+        "id": "opensky_client_secret",
+        "env_key": "OPENSKY_CLIENT_SECRET",
+        "name": "OpenSky Network — Client Secret",
+        "description": "OAuth2 client secret paired with the Client ID above. Used for authenticated token refresh.",
+        "category": "Aviation",
+        "url": "https://opensky-network.org/",
+        "required": True,
+    },
+    {
+        "id": "ais_api_key",
+        "env_key": "AIS_API_KEY",
+        "name": "AIS Stream",
+        "description": "WebSocket API key for real-time Automatic Identification System (AIS) vessel tracking data worldwide.",
+        "category": "Maritime",
+        "url": "https://aisstream.io/",
+        "required": True,
+    },
+    {
+        "id": "adsb_lol",
+        "env_key": None,
+        "name": "ADS-B Exchange (adsb.lol)",
+        "description": "Community-maintained ADS-B flight tracking API. No key required — public endpoint.",
+        "category": "Aviation",
+        "url": "https://api.adsb.lol/",
+        "required": False,
+    },
+    {
+        "id": "usgs_earthquakes",
+        "env_key": None,
+        "name": "USGS Earthquake Hazards",
+        "description": "Real-time earthquake data feed from the United States Geological Survey. No key required.",
+        "category": "Geophysical",
+        "url": "https://earthquake.usgs.gov/",
+        "required": False,
+    },
+    {
+        "id": "celestrak",
+        "env_key": None,
+        "name": "CelesTrak (NORAD TLEs)",
+        "description": "Satellite orbital element data from CelesTrak. Provides TLE sets for 2,000+ active satellites. No key required.",
+        "category": "Space",
+        "url": "https://celestrak.org/",
+        "required": False,
+    },
+    {
+        "id": "gdelt",
+        "env_key": None,
+        "name": "GDELT Project",
+        "description": "Global Database of Events, Language, and Tone. Monitors news media for geopolitical events worldwide. No key required.",
+        "category": "Intelligence",
+        "url": "https://www.gdeltproject.org/",
+        "required": False,
+    },
+    {
+        "id": "nominatim",
+        "env_key": None,
+        "name": "Nominatim (OpenStreetMap)",
+        "description": "Reverse geocoding service. Converts lat/lng coordinates to human-readable location names. No key required.",
+        "category": "Geolocation",
+        "url": "https://nominatim.openstreetmap.org/",
+        "required": False,
+    },
+    {
+        "id": "rainviewer",
+        "env_key": None,
+        "name": "RainViewer",
+        "description": "Weather radar tile overlay. Provides global precipitation data as map tiles. No key required.",
+        "category": "Weather",
+        "url": "https://www.rainviewer.com/",
+        "required": False,
+    },
+    {
+        "id": "rss_feeds",
+        "env_key": None,
+        "name": "RSS News Feeds",
+        "description": "Aggregates from NPR, BBC, Al Jazeera, NYT, Reuters, and AP for global news coverage. No key required.",
+        "category": "Intelligence",
+        "url": None,
+        "required": False,
+    },
+    {
+        "id": "yfinance",
+        "env_key": None,
+        "name": "Yahoo Finance (yfinance)",
+        "description": "Defense sector stock tickers and commodity prices. Uses the yfinance Python library. No key required.",
+        "category": "Markets",
+        "url": "https://finance.yahoo.com/",
+        "required": False,
+    },
+    {
+        "id": "openmhz",
+        "env_key": None,
+        "name": "OpenMHz",
+        "description": "Public radio scanner feeds for SIGINT interception. Streams police/fire/EMS radio traffic. No key required.",
+        "category": "SIGINT",
+        "url": "https://openmhz.com/",
+        "required": False,
+    },
+]
+
+
+def _obfuscate(value: str) -> str:
+    """Show first 4 chars, mask the rest with bullets."""
+    if not value or len(value) <= 4:
+        return "••••••••"
+    return value[:4] + "•" * (len(value) - 4)
+
+
+def get_api_keys():
+    """Return the full API registry with obfuscated key values."""
+    result = []
+    for api in API_REGISTRY:
+        entry = {
+            "id": api["id"],
+            "name": api["name"],
+            "description": api["description"],
+            "category": api["category"],
+            "url": api["url"],
+            "required": api["required"],
+            "has_key": api["env_key"] is not None,
+            "env_key": api["env_key"],
+            "value_obfuscated": None,
+            "value_plain": None,
+        }
+        if api["env_key"]:
+            raw = os.environ.get(api["env_key"], "")
+            entry["value_obfuscated"] = _obfuscate(raw)
+            entry["value_plain"] = raw  # Sent only when reveal is requested
+        result.append(entry)
+    return result
+
+
+def update_api_key(env_key: str, new_value: str) -> bool:
+    """Update a single key in the .env file and in the current process env."""
+    if not ENV_PATH.exists():
+        return False
+
+    # Update os.environ immediately
+    os.environ[env_key] = new_value
+
+    # Update the .env file on disk
+    content = ENV_PATH.read_text(encoding="utf-8")
+    pattern = re.compile(rf"^{re.escape(env_key)}=.*$", re.MULTILINE)
+    if pattern.search(content):
+        content = pattern.sub(f"{env_key}={new_value}", content)
+    else:
+        content = content.rstrip("\n") + f"\n{env_key}={new_value}\n"
+
+    ENV_PATH.write_text(content, encoding="utf-8")
+    return True

backend/services/carrier_tracker.py

@@ -0,0 +1,455 @@
+"""
+Carrier Strike Group OSINT Tracker
+===================================
+Scrapes multiple OSINT sources to maintain current estimated positions
+for US Navy Carrier Strike Groups. Updates on startup + 00:00 & 12:00 UTC.
+
+Sources:
+  1. GDELT News API — recent carrier movement headlines
+  2. WikiVoyage / public port-call databases
+  3. Fallback — last-known or static OSINT estimates
+"""
+
+import re
+import json
+import time
+import logging
+import threading
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Dict, List, Optional
+from services.network_utils import fetch_with_curl
+
+logger = logging.getLogger(__name__)
+
+# -----------------------------------------------------------------
+# Carrier registry: hull number → metadata + fallback position
+# -----------------------------------------------------------------
+CARRIER_REGISTRY: Dict[str, dict] = {
+    "CVN-68": {
+        "name": "USS Nimitz (CVN-68)",
+        "wiki": "https://en.wikipedia.org/wiki/USS_Nimitz",
+        "homeport": "Bremerton, WA",
+        "homeport_lat": 47.56, "homeport_lng": -122.63,
+        "fallback_lat": 21.35, "fallback_lng": -157.95,
+        "fallback_heading": 270,
+        "fallback_desc": "Pacific Fleet / Pearl Harbor"
+    },
+    "CVN-69": {
+        "name": "USS Dwight D. Eisenhower (CVN-69)",
+        "wiki": "https://en.wikipedia.org/wiki/USS_Dwight_D._Eisenhower",
+        "homeport": "Norfolk, VA",
+        "homeport_lat": 36.95, "homeport_lng": -76.33,
+        "fallback_lat": 18.0, "fallback_lng": 39.5,
+        "fallback_heading": 120,
+        "fallback_desc": "Red Sea / CENTCOM AOR"
+    },
+    "CVN-78": {
+        "name": "USS Gerald R. Ford (CVN-78)",
+        "wiki": "https://en.wikipedia.org/wiki/USS_Gerald_R._Ford",
+        "homeport": "Norfolk, VA",
+        "homeport_lat": 36.95, "homeport_lng": -76.33,
+        "fallback_lat": 34.0, "fallback_lng": 25.0,
+        "fallback_heading": 90,
+        "fallback_desc": "Eastern Mediterranean deterrence"
+    },
+    "CVN-70": {
+        "name": "USS Carl Vinson (CVN-70)",
+        "wiki": "https://en.wikipedia.org/wiki/USS_Carl_Vinson",
+        "homeport": "San Diego, CA",
+        "homeport_lat": 32.68, "homeport_lng": -117.15,
+        "fallback_lat": 15.0, "fallback_lng": 115.0,
+        "fallback_heading": 45,
+        "fallback_desc": "South China Sea patrol"
+    },
+    "CVN-71": {
+        "name": "USS Theodore Roosevelt (CVN-71)",
+        "wiki": "https://en.wikipedia.org/wiki/USS_Theodore_Roosevelt_(CVN-71)",
+        "homeport": "San Diego, CA",
+        "homeport_lat": 32.68, "homeport_lng": -117.15,
+        "fallback_lat": 22.0, "fallback_lng": 122.0,
+        "fallback_heading": 300,
+        "fallback_desc": "Philippine Sea / Taiwan Strait"
+    },
+    "CVN-72": {
+        "name": "USS Abraham Lincoln (CVN-72)",
+        "wiki": "https://en.wikipedia.org/wiki/USS_Abraham_Lincoln_(CVN-72)",
+        "homeport": "San Diego, CA",
+        "homeport_lat": 32.68, "homeport_lng": -117.15,
+        "fallback_lat": 21.0, "fallback_lng": -158.0,
+        "fallback_heading": 270,
+        "fallback_desc": "Pacific deployment"
+    },
+    "CVN-73": {
+        "name": "USS George Washington (CVN-73)",
+        "wiki": "https://en.wikipedia.org/wiki/USS_George_Washington_(CVN-73)",
+        "homeport": "Yokosuka, Japan",
+        "homeport_lat": 35.28, "homeport_lng": 139.67,
+        "fallback_lat": 35.0, "fallback_lng": 139.0,
+        "fallback_heading": 0,
+        "fallback_desc": "Yokosuka, Japan (Forward deployed)"
+    },
+    "CVN-74": {
+        "name": "USS John C. Stennis (CVN-74)",
+        "wiki": "https://en.wikipedia.org/wiki/USS_John_C._Stennis",
+        "homeport": "Norfolk, VA",
+        "homeport_lat": 36.95, "homeport_lng": -76.33,
+        "fallback_lat": 36.95, "fallback_lng": -76.33,
+        "fallback_heading": 0,
+        "fallback_desc": "RCOH / Norfolk (maintenance)"
+    },
+    "CVN-75": {
+        "name": "USS Harry S. Truman (CVN-75)",
+        "wiki": "https://en.wikipedia.org/wiki/USS_Harry_S._Truman",
+        "homeport": "Norfolk, VA",
+        "homeport_lat": 36.95, "homeport_lng": -76.33,
+        "fallback_lat": 36.0, "fallback_lng": 15.0,
+        "fallback_heading": 90,
+        "fallback_desc": "Mediterranean deployment"
+    },
+    "CVN-76": {
+        "name": "USS Ronald Reagan (CVN-76)",
+        "wiki": "https://en.wikipedia.org/wiki/USS_Ronald_Reagan",
+        "homeport": "Bremerton, WA",
+        "homeport_lat": 47.56, "homeport_lng": -122.63,
+        "fallback_lat": 47.56, "fallback_lng": -122.63,
+        "fallback_heading": 0,
+        "fallback_desc": "Bremerton, WA (Homeport)"
+    },
+    "CVN-77": {
+        "name": "USS George H.W. Bush (CVN-77)",
+        "wiki": "https://en.wikipedia.org/wiki/USS_George_H.W._Bush",
+        "homeport": "Norfolk, VA",
+        "homeport_lat": 36.95, "homeport_lng": -76.33,
+        "fallback_lat": 36.95, "fallback_lng": -76.33,
+        "fallback_heading": 0,
+        "fallback_desc": "Norfolk, VA (Homeport)"
+    },
+}
+
+# -----------------------------------------------------------------
+# Region → approximate center coordinates
+# Used to map textual geographic descriptions to lat/lng
+# -----------------------------------------------------------------
+REGION_COORDS: Dict[str, tuple] = {
+    # Oceans & Seas
+    "eastern mediterranean": (34.0, 25.0),
+    "mediterranean": (36.0, 15.0),
+    "western mediterranean": (37.0, 2.0),
+    "red sea": (18.0, 39.5),
+    "arabian sea": (16.0, 64.0),
+    "persian gulf": (26.5, 51.5),
+    "gulf of oman": (24.5, 58.5),
+    "north arabian sea": (20.0, 64.0),
+    "south china sea": (15.0, 115.0),
+    "east china sea": (28.0, 125.0),
+    "philippine sea": (20.0, 130.0),
+    "sea of japan": (40.0, 135.0),
+    "taiwan strait": (24.0, 119.5),
+    "western pacific": (20.0, 140.0),
+    "pacific": (20.0, -150.0),
+    "indian ocean": (-5.0, 70.0),
+    "north atlantic": (40.0, -40.0),
+    "atlantic": (30.0, -50.0),
+    "gulf of aden": (12.5, 45.0),
+    "horn of africa": (10.0, 50.0),
+    "strait of hormuz": (26.5, 56.3),
+    "bab el-mandeb": (12.6, 43.3),
+    "suez canal": (30.5, 32.3),
+    "baltic sea": (57.0, 18.0),
+    "north sea": (56.0, 3.0),
+    "black sea": (43.0, 34.0),
+    "south atlantic": (-20.0, -20.0),
+    "coral sea": (-18.0, 155.0),
+    "gulf of mexico": (25.0, -90.0),
+    "caribbean": (15.0, -75.0),
+
+    # Specific bases / ports
+    "norfolk": (36.95, -76.33),
+    "san diego": (32.68, -117.15),
+    "yokosuka": (35.28, 139.67),
+    "pearl harbor": (21.35, -157.95),
+    "guam": (13.45, 144.79),
+    "bahrain": (26.23, 50.55),
+    "rota": (36.62, -6.35),
+    "naples": (40.85, 14.27),
+    "bremerton": (47.56, -122.63),
+    "puget sound": (47.56, -122.63),
+    "newport news": (36.98, -76.43),
+
+    # Areas of operation
+    "centcom": (25.0, 55.0),
+    "indopacom": (20.0, 130.0),
+    "eucom": (48.0, 15.0),
+    "southcom": (10.0, -80.0),
+    "5th fleet": (25.0, 55.0),
+    "6th fleet": (36.0, 15.0),
+    "7th fleet": (25.0, 130.0),
+    "3rd fleet": (30.0, -130.0),
+    "2nd fleet": (35.0, -60.0),
+}
+
+# -----------------------------------------------------------------
+# Cache file for persisting positions between restarts
+# -----------------------------------------------------------------
+CACHE_FILE = Path(__file__).parent.parent / "carrier_cache.json"
+
+_carrier_positions: Dict[str, dict] = {}
+_positions_lock = threading.Lock()
+_last_update: Optional[datetime] = None
+
+
+def _load_cache() -> Dict[str, dict]:
+    """Load cached carrier positions from disk."""
+    try:
+        if CACHE_FILE.exists():
+            data = json.loads(CACHE_FILE.read_text())
+            logger.info(f"Carrier cache loaded: {len(data)} carriers from {CACHE_FILE}")
+            return data
+    except Exception as e:
+        logger.warning(f"Failed to load carrier cache: {e}")
+    return {}
+
+
+def _save_cache(positions: Dict[str, dict]):
+    """Persist carrier positions to disk."""
+    try:
+        CACHE_FILE.write_text(json.dumps(positions, indent=2))
+        logger.info(f"Carrier cache saved: {len(positions)} carriers")
+    except Exception as e:
+        logger.warning(f"Failed to save carrier cache: {e}")
+
+
+def _match_region(text: str) -> Optional[tuple]:
+    """Match a text string against known regions, return (lat, lng) or None."""
+    text_lower = text.lower()
+    for region, coords in sorted(REGION_COORDS.items(), key=lambda x: -len(x[0])):
+        if region in text_lower:
+            return coords
+    return None
+
+
+def _match_carrier(text: str) -> Optional[str]:
+    """Match a text string against known carrier names/hull numbers."""
+    text_lower = text.lower()
+    for hull, info in CARRIER_REGISTRY.items():
+        hull_check = hull.lower().replace("-", "")
+        name_parts = info["name"].lower()
+        # Match hull number (e.g., "CVN-78", "CVN78")
+        if hull.lower() in text_lower or hull_check in text_lower.replace("-", ""):
+            return hull
+        # Match ship name (e.g., "Ford", "Eisenhower", "Vinson")
+        ship_name = name_parts.split("(")[0].strip()
+        last_name = ship_name.split()[-1] if ship_name else ""
+        if last_name and len(last_name) > 3 and last_name in text_lower:
+            return hull
+    return None
+
+
+def _fetch_gdelt_carrier_news() -> List[dict]:
+    """Search GDELT for recent carrier movement news."""
+    results = []
+    search_terms = [
+        "aircraft+carrier+deployed",
+        "carrier+strike+group+navy",
+        "USS+Nimitz+carrier", "USS+Ford+carrier", "USS+Eisenhower+carrier",
+        "USS+Vinson+carrier", "USS+Roosevelt+carrier+navy",
+        "USS+Lincoln+carrier", "USS+Truman+carrier",
+        "USS+Reagan+carrier", "USS+Washington+carrier+navy",
+        "USS+Bush+carrier", "USS+Stennis+carrier",
+    ]
+
+    for term in search_terms:
+        try:
+            url = f"https://api.gdeltproject.org/api/v2/doc/doc?query={term}&mode=artlist&maxrecords=5&format=json&timespan=14d"
+            raw = fetch_with_curl(url, timeout=8)
+            if not raw:
+                continue
+            data = json.loads(raw)
+            articles = data.get("articles", [])
+            for art in articles:
+                title = art.get("title", "")
+                url = art.get("url", "")
+                results.append({"title": title, "url": url})
+        except Exception as e:
+            logger.debug(f"GDELT search failed for '{term}': {e}")
+            continue
+
+    logger.info(f"Carrier OSINT: found {len(results)} GDELT articles")
+    return results
+
+
+def _parse_carrier_positions_from_news(articles: List[dict]) -> Dict[str, dict]:
+    """Parse carrier positions from news article titles and descriptions."""
+    updates: Dict[str, dict] = {}
+
+    for article in articles:
+        title = article.get("title", "")
+
+        # Try to match a carrier from the title
+        hull = _match_carrier(title)
+        if not hull:
+            continue
+
+        # Try to match a region from the title
+        coords = _match_region(title)
+        if not coords:
+            continue
+
+        # Only update if we haven't seen this carrier yet (first match wins — most recent)
+        if hull not in updates:
+            updates[hull] = {
+                "lat": coords[0],
+                "lng": coords[1],
+                "desc": title[:100],
+                "source": "GDELT OSINT",
+                "updated": datetime.now(timezone.utc).isoformat()
+            }
+            logger.info(f"Carrier update: {CARRIER_REGISTRY[hull]['name']} → {coords} (from: {title[:80]})")
+
+    return updates
+
+
+def update_carrier_positions():
+    """Main update function — called on startup and every 12h."""
+    global _last_update
+
+    logger.info("Carrier tracker: updating positions from OSINT sources...")
+
+    # Start with fallback positions
+    positions: Dict[str, dict] = {}
+    for hull, info in CARRIER_REGISTRY.items():
+        positions[hull] = {
+            "name": info["name"],
+            "lat": info["fallback_lat"],
+            "lng": info["fallback_lng"],
+            "heading": info["fallback_heading"],
+            "desc": info["fallback_desc"],
+            "wiki": info["wiki"],
+            "source": "Static OSINT estimate",
+            "updated": datetime.now(timezone.utc).isoformat()
+        }
+
+    # Load cached positions (may have better data from previous runs)
+    cached = _load_cache()
+    for hull, cached_pos in cached.items():
+        if hull in positions:
+            # Only use cache if it has a real OSINT source (not just static)
+            if cached_pos.get("source", "").startswith("GDELT") or cached_pos.get("source", "").startswith("News"):
+                positions[hull].update({
+                    "lat": cached_pos["lat"],
+                    "lng": cached_pos["lng"],
+                    "desc": cached_pos.get("desc", positions[hull]["desc"]),
+                    "source": cached_pos.get("source", "Cached OSINT"),
+                    "updated": cached_pos.get("updated", "")
+                })
+
+    # Try GDELT news for fresh positions
+    try:
+        articles = _fetch_gdelt_carrier_news()
+        news_positions = _parse_carrier_positions_from_news(articles)
+        for hull, pos in news_positions.items():
+            if hull in positions:
+                positions[hull].update(pos)
+                logger.info(f"Carrier OSINT: updated {CARRIER_REGISTRY[hull]['name']} from news")
+    except Exception as e:
+        logger.warning(f"GDELT carrier fetch failed: {e}")
+
+    # Save and update the global state
+    with _positions_lock:
+        _carrier_positions.clear()
+        _carrier_positions.update(positions)
+        _last_update = datetime.now(timezone.utc)
+
+    _save_cache(positions)
+
+    sources = {}
+    for p in positions.values():
+        src = p.get("source", "unknown")
+        sources[src] = sources.get(src, 0) + 1
+    logger.info(f"Carrier tracker: {len(positions)} carriers updated. Sources: {sources}")
+
+
+def get_carrier_positions() -> List[dict]:
+    """Return current carrier positions for the data pipeline."""
+    with _positions_lock:
+        result = []
+        for hull, pos in _carrier_positions.items():
+            info = CARRIER_REGISTRY.get(hull, {})
+            result.append({
+                "name": pos.get("name", info.get("name", hull)),
+                "type": "carrier",
+                "lat": pos["lat"],
+                "lng": pos["lng"],
+                "heading": pos.get("heading", 0),
+                "sog": 0,
+                "cog": 0,
+                "country": "United States",
+                "desc": pos.get("desc", ""),
+                "wiki": pos.get("wiki", info.get("wiki", "")),
+                "estimated": True,
+                "source": pos.get("source", "OSINT estimated position"),
+                "last_osint_update": pos.get("updated", "")
+            })
+        return result
+
+
+# -----------------------------------------------------------------
+# Scheduler: runs at startup, then at 00:00 and 12:00 UTC daily
+# -----------------------------------------------------------------
+_scheduler_thread: Optional[threading.Thread] = None
+_scheduler_stop = threading.Event()
+
+
+def _scheduler_loop():
+    """Background thread that triggers updates at 00:00 and 12:00 UTC."""
+    # Initial update on startup
+    try:
+        update_carrier_positions()
+    except Exception as e:
+        logger.error(f"Carrier tracker initial update failed: {e}")
+
+    while not _scheduler_stop.is_set():
+        now = datetime.now(timezone.utc)
+        # Next target: 00:00 or 12:00 UTC, whichever is sooner
+        hour = now.hour
+        if hour < 12:
+            next_hour = 12
+        else:
+            next_hour = 24  # midnight = next day 00:00
+
+        next_run = now.replace(hour=next_hour % 24, minute=0, second=0, microsecond=0)
+        if next_hour == 24:
+            from datetime import timedelta
+            next_run = (now + timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0)
+
+        wait_seconds = (next_run - now).total_seconds()
+        logger.info(f"Carrier tracker: next update at {next_run.isoformat()} ({wait_seconds/3600:.1f}h)")
+
+        # Wait until next scheduled time, or until stop event
+        if _scheduler_stop.wait(timeout=wait_seconds):
+            break  # Stop event was set
+
+        try:
+            update_carrier_positions()
+        except Exception as e:
+            logger.error(f"Carrier tracker scheduled update failed: {e}")
+
+
+def start_carrier_tracker():
+    """Start the carrier tracker background thread."""
+    global _scheduler_thread
+    if _scheduler_thread and _scheduler_thread.is_alive():
+        return
+    _scheduler_stop.clear()
+    _scheduler_thread = threading.Thread(target=_scheduler_loop, daemon=True, name="carrier-tracker")
+    _scheduler_thread.start()
+    logger.info("Carrier tracker started")
+
+
+def stop_carrier_tracker():
+    """Stop the carrier tracker background thread."""
+    _scheduler_stop.set()
+    if _scheduler_thread:
+        _scheduler_thread.join(timeout=5)
+    logger.info("Carrier tracker stopped")

backend/services/cctv_pipeline.py

@@ -0,0 +1,274 @@
+import sqlite3
+import requests
+from services.network_utils import fetch_with_curl
+import logging
+from abc import ABC, abstractmethod
+from typing import List, Dict, Any
+
+logger = logging.getLogger(__name__)
+
+DB_PATH = "cctv.db"
+
+def init_db():
+    conn = sqlite3.connect(DB_PATH)
+    cursor = conn.cursor()
+    cursor.execute("""
+        CREATE TABLE IF NOT EXISTS cameras (
+            id TEXT PRIMARY KEY,
+            source_agency TEXT,
+            lat REAL,
+            lon REAL,
+            direction_facing TEXT,
+            media_url TEXT,
+            refresh_rate_seconds INTEGER,
+            last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+        )
+    """)
+    conn.commit()
+    conn.close()
+
+class BaseCCTVIngestor(ABC):
+    def __init__(self):
+        self.conn = sqlite3.connect(DB_PATH)
+
+    @abstractmethod
+    def fetch_data(self) -> List[Dict[str, Any]]:
+        pass
+
+    def ingest(self):
+        try:
+            cameras = self.fetch_data()
+            cursor = self.conn.cursor()
+            for cam in cameras:
+                cursor.execute("""
+                    INSERT INTO cameras 
+                    (id, source_agency, lat, lon, direction_facing, media_url, refresh_rate_seconds)
+                    VALUES (?, ?, ?, ?, ?, ?, ?)
+                    ON CONFLICT(id) DO UPDATE SET
+                    media_url=excluded.media_url,
+                    last_updated=CURRENT_TIMESTAMP
+                """, (
+                    cam.get("id"),
+                    cam.get("source_agency"),
+                    cam.get("lat"),
+                    cam.get("lon"),
+                    cam.get("direction_facing", "Unknown"),
+                    cam.get("media_url"),
+                    cam.get("refresh_rate_seconds", 60)
+                ))
+            self.conn.commit()
+            logger.info(f"Successfully ingested {len(cameras)} cameras from {self.__class__.__name__}")
+        except Exception as e:
+            logger.error(f"Failed to ingest cameras in {self.__class__.__name__}: {e}")
+
+class TFLJamCamIngestor(BaseCCTVIngestor):
+    def fetch_data(self) -> List[Dict[str, Any]]:
+        # Transport for London Open Data API
+        url = "https://api.tfl.gov.uk/Place/Type/JamCam"
+        response = fetch_with_curl(url, timeout=15)
+        response.raise_for_status()
+        
+        data = response.json()
+        cameras = []
+        for item in data:
+            # TfL returns URLs without protocols sometimes or with a base path
+            vid_url = None
+            img_url = None
+            
+            for prop in item.get('additionalProperties', []):
+                if prop.get('key') == 'videoUrl':
+                    vid_url = prop.get('value')
+                elif prop.get('key') == 'imageUrl':
+                    img_url = prop.get('value')
+            
+            media = vid_url if vid_url else img_url
+            if media:
+                cameras.append({
+                    "id": f"TFL-{item.get('id')}",
+                    "source_agency": "TfL",
+                    "lat": item.get('lat'),
+                    "lon": item.get('lon'),
+                    "direction_facing": item.get('commonName', 'Unknown'),
+                    "media_url": media,
+                    "refresh_rate_seconds": 15
+                })
+        return cameras
+
+class LTASingaporeIngestor(BaseCCTVIngestor):
+    def fetch_data(self) -> List[Dict[str, Any]]:
+        # Singapore Land Transport Authority (LTA) Traffic Images API
+        url = "https://api.data.gov.sg/v1/transport/traffic-images"
+        response = fetch_with_curl(url, timeout=15)
+        response.raise_for_status()
+        
+        data = response.json()
+        cameras = []
+        if "items" in data and len(data["items"]) > 0:
+            for item in data["items"][0].get("cameras", []):
+                loc = item.get("location", {})
+                if "latitude" in loc and "longitude" in loc and "image" in item:
+                    cameras.append({
+                        "id": f"SGP-{item.get('camera_id', 'UNK')}",
+                        "source_agency": "Singapore LTA",
+                        "lat": loc.get("latitude"),
+                        "lon": loc.get("longitude"),
+                        "direction_facing": f"Camera {item.get('camera_id')}",
+                        "media_url": item.get("image"),
+                        "refresh_rate_seconds": 60
+                    })
+        return cameras
+
+
+
+class AustinTXIngestor(BaseCCTVIngestor):
+    def fetch_data(self) -> List[Dict[str, Any]]:
+        # City of Austin Traffic Cameras Open Data
+        url = "https://data.austintexas.gov/resource/b4k4-adkb.json?$limit=2000"
+        response = fetch_with_curl(url, timeout=15)
+        response.raise_for_status()
+        
+        data = response.json()
+        cameras = []
+        for item in data:
+            cam_id = item.get("camera_id")
+            if not cam_id: continue
+            
+            loc = item.get("location", {})
+            coords = loc.get("coordinates", [])
+            
+            # coords is usually [lon, lat]
+            if len(coords) == 2:
+                cameras.append({
+                    "id": f"ATX-{cam_id}",
+                    "source_agency": "Austin TxDOT",
+                    "lat": coords[1],
+                    "lon": coords[0],
+                    "direction_facing": item.get("location_name", "Austin TX Camera"),
+                    "media_url": f"https://cctv.austinmobility.io/image/{cam_id}.jpg",
+                    "refresh_rate_seconds": 60
+                })
+        return cameras
+
+class NYCDOTIngestor(BaseCCTVIngestor):
+    def fetch_data(self) -> List[Dict[str, Any]]:
+        url = "https://webcams.nyctmc.org/api/cameras"
+        response = fetch_with_curl(url, timeout=15)
+        response.raise_for_status()
+        
+        data = response.json()
+        cameras = []
+        for item in data:
+            cam_id = item.get("id")
+            if not cam_id: continue
+            
+            lat = item.get("latitude")
+            lon = item.get("longitude")
+            if lat and lon:
+                cameras.append({
+                    "id": f"NYC-{cam_id}",
+                    "source_agency": "NYC DOT",
+                    "lat": lat,
+                    "lon": lon,
+                    "direction_facing": item.get("name", "NYC Camera"),
+                    "media_url": f"https://webcams.nyctmc.org/api/cameras/{cam_id}/image",
+                    "refresh_rate_seconds": 30
+                })
+        return cameras
+
+class GlobalOSMCrawlingIngestor(BaseCCTVIngestor):
+    def fetch_data(self) -> List[Dict[str, Any]]:
+        # This will pull physical street surveillance cameras across all global hotspots
+        # using OpenStreetMap Overpass mapping their exact geospatial coordinates to Google Street View
+        regions = [
+            ("35.6,139.6,35.8,139.8", "Tokyo"),
+            ("48.8,2.3,48.9,2.4", "Paris"),
+            ("40.6,-74.1,40.8,-73.9", "NYC Expanded"),
+            ("34.0,-118.4,34.2,-118.2", "Los Angeles"),
+            ("-33.9,151.1,-33.7,151.3", "Sydney"),
+            ("52.4,13.3,52.6,13.5", "Berlin"),
+            ("25.1,55.2,25.3,55.4", "Dubai"),
+            ("19.3,-99.2,19.5,-99.0", "Mexico City"),
+            ("-23.6,-46.7,-23.4,-46.5", "Sao Paulo"),
+            ("39.6,-105.1,39.9,-104.8", "Denver")
+        ]
+        
+        query_parts = [f'node["man_made"="surveillance"]({bbox});' for bbox, city in regions]
+        query = "".join(query_parts)
+        url = f"https://overpass-api.de/api/interpreter?data=[out:json];({query});out%202000;"
+        
+        try:
+            response = fetch_with_curl(url, timeout=15)
+            response.raise_for_status()
+            data = response.json()
+            
+            cameras = []
+            for item in data.get('elements', []):
+                lat = item.get("lat")
+                lon = item.get("lon")
+                cam_id = item.get("id")
+                
+                if lat and lon:
+                    # Find which city this belongs to
+                    source_city = "Global OSINT"
+                    for bbox, city in regions:
+                        s, w, n, e = map(float, bbox.split(','))
+                        if s <= lat <= n and w <= lon <= e:
+                            source_city = f"OSINT: {city}"
+                            break
+                            
+                    # Attempt to parse camera direction for a cool realistic bearing angle if OSM mapped it
+                    direction_str = item.get("tags", {}).get("camera:direction", "0")
+                    try:
+                        bearing = int(float(direction_str))
+                    except:
+                        bearing = 0
+                        
+                    mapbox_key = "YOUR_MAPBOX_TOKEN_HERE"
+                    mapbox_url = f"https://api.mapbox.com/styles/v1/mapbox/satellite-streets-v12/static/{lon},{lat},18,{bearing},60/600x400?access_token={mapbox_key}"
+                    
+                    cameras.append({
+                        "id": f"OSM-{cam_id}",
+                        "source_agency": source_city,
+                        "lat": lat,
+                        "lon": lon,
+                        "direction_facing": item.get("tags", {}).get("surveillance:type", "Street Level Camera"),
+                        "media_url": mapbox_url,
+                        "refresh_rate_seconds": 3600
+                    })
+            return cameras
+        except Exception:
+            return []
+
+
+
+def _detect_media_type(url: str) -> str:
+    """Detect the media type from a camera URL for proper frontend rendering."""
+    if not url:
+        return "image"
+    url_lower = url.lower()
+    if any(ext in url_lower for ext in ['.mp4', '.webm', '.ogg']):
+        return "video"
+    if any(kw in url_lower for kw in ['.mjpg', '.mjpeg', 'mjpg', 'axis-cgi/mjpg', 'mode=motion']):
+        return "mjpeg"
+    if '.m3u8' in url_lower or 'hls' in url_lower:
+        return "hls"
+    if any(kw in url_lower for kw in ['embed', 'maps/embed', 'iframe']):
+        return "embed"
+    if 'mapbox.com' in url_lower or 'satellite' in url_lower:
+        return "satellite"
+    return "image"
+
+def get_all_cameras() -> List[Dict[str, Any]]:
+    conn = sqlite3.connect(DB_PATH)
+    conn.row_factory = sqlite3.Row
+    cursor = conn.cursor()
+    cursor.execute("SELECT * FROM cameras")
+    rows = cursor.fetchall()
+    conn.close()
+    cameras = []
+    for row in rows:
+        cam = dict(row)
+        cam['media_type'] = _detect_media_type(cam.get('media_url', ''))
+        cameras.append(cam)
+    return cameras
+

backend/services/geopolitics.py

@@ -0,0 +1,301 @@
+import requests
+import logging
+from cachetools import cached, TTLCache
+from datetime import datetime
+from services.network_utils import fetch_with_curl
+
+logger = logging.getLogger(__name__)
+
+# Cache Frontline data for 30 minutes, it doesn't move that fast
+frontline_cache = TTLCache(maxsize=1, ttl=1800)
+
+@cached(frontline_cache)
+def fetch_ukraine_frontlines():
+    """
+    Fetches the latest GeoJSON data representing the Ukraine frontline.
+    We use the cyterat/deepstate-map-data github mirror since the public API is locked.
+    """
+    try:
+        logger.info("Fetching DeepStateMap from GitHub mirror...")
+        
+        # First, query the repo tree to find the latest file name
+        tree_url = "https://api.github.com/repos/cyterat/deepstate-map-data/git/trees/main?recursive=1"
+        res_tree = requests.get(tree_url, timeout=10)
+        
+        if res_tree.status_code == 200:
+            tree_data = res_tree.json().get("tree", [])
+            # Filter for geojson files in data folder
+            geo_files = [item["path"] for item in tree_data if item["path"].startswith("data/deepstatemap_data_") and item["path"].endswith(".geojson")]
+            
+            if geo_files:
+                # Get the alphabetically latest file (since it's named with YYYYMMDD)
+                latest_file = sorted(geo_files)[-1]
+                
+                raw_url = f"https://raw.githubusercontent.com/cyterat/deepstate-map-data/main/{latest_file}"
+                logger.info(f"Downloading latest DeepStateMap: {raw_url}")
+                
+                res_geo = requests.get(raw_url, timeout=20)
+                if res_geo.status_code == 200:
+                    data = res_geo.json()
+                    
+                    # The Cyterat GitHub mirror strips all properties and just provides a raw array of Feature polygons.
+                    # Based on DeepStateMap's frontend mapping, the array index corresponds to the zone type:
+                    # 0: Russian-occupied areas
+                    # 1: Russian advance
+                    # 2: Liberated area
+                    # 3: Uncontested/Crimea (often folded into occupied)
+                    name_map = {
+                        0: "Russian-occupied areas",
+                        1: "Russian advance",
+                        2: "Liberated area",
+                        3: "Russian-occupied areas", # Crimea / LPR / DPR
+                        4: "Directions of UA attacks"
+                    }
+                    
+                    if "features" in data:
+                        for idx, feature in enumerate(data["features"]):
+                            if "properties" not in feature or feature["properties"] is None:
+                                feature["properties"] = {}
+                            
+                            feature["properties"]["name"] = name_map.get(idx, "Russian-occupied areas")
+                            feature["properties"]["zone_id"] = idx
+                            
+                    return data
+                else:
+                    logger.error(f"Failed to fetch parsed Github Raw GeoJSON: {res_geo.status_code}")
+        else:
+            logger.error(f"Failed to fetch Github Tree for Deepstatemap: {res_tree.status_code}")
+    except Exception as e:
+        logger.error(f"Error fetching DeepStateMap: {e}")
+    return None
+
+# Cache GDELT data for 6 hours - heavy aggregation, data doesn't change rapidly
+gdelt_cache = TTLCache(maxsize=1, ttl=21600)
+
+def _extract_domain(url):
+    """Extract a clean source name from a URL, e.g. 'nytimes.com' from 'https://www.nytimes.com/...'"""
+    try:
+        from urllib.parse import urlparse
+        host = urlparse(url).hostname or ''
+        # Strip www. prefix
+        if host.startswith('www.'):
+            host = host[4:]
+        return host
+    except Exception:
+        return url[:40]
+
+def _url_to_headline(url):
+    """Extract a human-readable headline from a URL path.
+    e.g. 'https://nytimes.com/2026/03/us-strikes-iran-nuclear-sites.html' -> 'Us Strikes Iran Nuclear Sites (nytimes.com)'
+    """
+    try:
+        from urllib.parse import urlparse, unquote
+        parsed = urlparse(url)
+        domain = parsed.hostname or ''
+        if domain.startswith('www.'):
+            domain = domain[4:]
+
+        # Get last meaningful path segment
+        path = unquote(parsed.path).strip('/')
+        if not path:
+            return domain
+
+        # Take the last path segment (usually the slug)
+        slug = path.split('/')[-1]
+        # Remove file extensions
+        for ext in ['.html', '.htm', '.php', '.asp', '.aspx', '.shtml']:
+            if slug.lower().endswith(ext):
+                slug = slug[:-len(ext)]
+        # If slug is purely numeric or a short ID, try the second-to-last segment
+        import re
+        if re.match(r'^[a-z]?\d{5,}$', slug, re.IGNORECASE):
+            segments = path.split('/')
+            if len(segments) >= 2:
+                slug = segments[-2]
+                for ext in ['.html', '.htm', '.php']:
+                    if slug.lower().endswith(ext):
+                        slug = slug[:-len(ext)]
+        # Remove common ID patterns at start/end
+        slug = re.sub(r'^[\d]+-', '', slug)  # leading numbers like "13847569-"
+        slug = re.sub(r'-[\da-f]{6,}$', '', slug)  # trailing hex IDs
+        slug = re.sub(r'[-_]c-\d+$', '', slug)  # trailing "-c-21803431"
+        slug = re.sub(r'^p=\d+$', '', slug)  # WordPress ?p=1234
+        # Convert slug separators to spaces
+        slug = slug.replace('-', ' ').replace('_', ' ')
+        # Clean up multiple spaces
+        slug = re.sub(r'\s+', ' ', slug).strip()
+
+        # If slug is still just a number or too short, fall back to domain
+        if len(slug) < 5 or re.match(r'^\d+$', slug):
+            return domain
+
+        # Title case and truncate
+        headline = slug.title()
+        if len(headline) > 80:
+            headline = headline[:77] + '...'
+        return f"{headline} ({domain})"
+    except Exception:
+        return url[:60]
+
+def _parse_gdelt_export_zip(zip_bytes, conflict_codes, seen_locs, features, loc_index):
+    """Parse a single GDELT export ZIP and append conflict features.
+    loc_index maps loc_key -> index in features list for fast duplicate merging.
+    """
+    import csv, io, zipfile
+    try:
+        zf = zipfile.ZipFile(io.BytesIO(zip_bytes))
+        csv_name = zf.namelist()[0]
+        with zf.open(csv_name) as cf:
+            reader = csv.reader(io.TextIOWrapper(cf, encoding='utf-8', errors='replace'), delimiter='\t')
+            for row in reader:
+                try:
+                    if len(row) < 61:
+                        continue
+                    event_code = row[26][:2] if len(row[26]) >= 2 else ''
+                    if event_code not in conflict_codes:
+                        continue
+                    lat = float(row[56]) if row[56] else None
+                    lng = float(row[57]) if row[57] else None
+                    if lat is None or lng is None or (lat == 0 and lng == 0):
+                        continue
+
+                    source_url = row[60].strip() if len(row) > 60 else ''
+                    location = row[52].strip() if len(row) > 52 else 'Unknown'
+                    actor1 = row[6].strip() if len(row) > 6 else ''
+                    actor2 = row[16].strip() if len(row) > 16 else ''
+
+                    loc_key = f"{round(lat, 1)}_{round(lng, 1)}"
+                    if loc_key in seen_locs:
+                        # Merge: increment count and add source URL if new (dedup by domain)
+                        idx = loc_index[loc_key]
+                        feat = features[idx]
+                        feat["properties"]["count"] = feat["properties"].get("count", 1) + 1
+                        urls = feat["properties"].get("_urls", [])
+                        seen_domains = feat["properties"].get("_domains", set())
+                        if source_url:
+                            domain = _extract_domain(source_url)
+                            if domain not in seen_domains and len(urls) < 10:
+                                urls.append(source_url)
+                                seen_domains.add(domain)
+                                feat["properties"]["_urls"] = urls
+                                feat["properties"]["_domains"] = seen_domains
+                        continue
+                    seen_locs.add(loc_key)
+
+                    name = location or (f"{actor1} vs {actor2}" if actor1 and actor2 else actor1) or "Unknown Incident"
+                    domain = _extract_domain(source_url) if source_url else ''
+                    loc_index[loc_key] = len(features)
+                    features.append({
+                        "type": "Feature",
+                        "properties": {
+                            "name": name,
+                            "count": 1,
+                            "_urls": [source_url] if source_url else [],
+                            "_domains": {domain} if domain else set(),
+                        },
+                        "geometry": {"type": "Point", "coordinates": [lng, lat]},
+                        "_loc_key": loc_key
+                    })
+                except (ValueError, IndexError):
+                    continue
+    except Exception as e:
+        logger.warning(f"Failed to parse GDELT export zip: {e}")
+
+def _download_gdelt_export(url):
+    """Download a single GDELT export file, return bytes or None."""
+    try:
+        res = fetch_with_curl(url, timeout=15)
+        if res.status_code == 200:
+            return res.content
+    except Exception:
+        pass
+    return None
+
+@cached(gdelt_cache)
+def fetch_global_military_incidents():
+    """
+    Fetches global military/conflict incidents from GDELT Events Export files.
+    Aggregates the last ~8 hours of 15-minute exports to build ~1000 incidents.
+    """
+    from datetime import timedelta
+    from concurrent.futures import ThreadPoolExecutor
+
+    try:
+        logger.info("Fetching GDELT events via export CDN (multi-file)...")
+
+        # Get the latest export URL to determine current timestamp
+        index_res = fetch_with_curl("http://data.gdeltproject.org/gdeltv2/lastupdate.txt", timeout=10)
+        if index_res.status_code != 200:
+            logger.error(f"GDELT lastupdate failed: {index_res.status_code}")
+            return []
+
+        # Extract latest export URL and its timestamp
+        latest_url = None
+        for line in index_res.text.strip().split('\n'):
+            parts = line.strip().split()
+            if len(parts) >= 3 and parts[2].endswith('.export.CSV.zip'):
+                latest_url = parts[2]
+                break
+
+        if not latest_url:
+            logger.error("Could not find GDELT export URL")
+            return []
+
+        # Extract timestamp from URL like: http://data.gdeltproject.org/gdeltv2/20260301120000.export.CSV.zip
+        import re
+        ts_match = re.search(r'(\d{14})\.export\.CSV\.zip', latest_url)
+        if not ts_match:
+            logger.error("Could not parse GDELT export timestamp")
+            return []
+
+        latest_ts = datetime.strptime(ts_match.group(1), '%Y%m%d%H%M%S')
+
+        # Generate URLs for the last 8 hours (32 files at 15-min intervals)
+        NUM_FILES = 32
+        urls = []
+        for i in range(NUM_FILES):
+            ts = latest_ts - timedelta(minutes=15 * i)
+            fname = ts.strftime('%Y%m%d%H%M%S') + '.export.CSV.zip'
+            url = f"http://data.gdeltproject.org/gdeltv2/{fname}"
+            urls.append(url)
+
+        logger.info(f"Downloading {len(urls)} GDELT export files...")
+
+        # Download in parallel (8 threads)
+        with ThreadPoolExecutor(max_workers=8) as executor:
+            zip_results = list(executor.map(_download_gdelt_export, urls))
+
+        successful = sum(1 for r in zip_results if r is not None)
+        logger.info(f"Downloaded {successful}/{len(urls)} GDELT exports")
+
+        # Parse all downloaded files
+        CONFLICT_CODES = {'14', '17', '18', '19', '20'}
+        features = []
+        seen_locs = set()
+        loc_index = {}  # loc_key -> index in features
+
+        for zip_bytes in zip_results:
+            if zip_bytes:
+                _parse_gdelt_export_zip(zip_bytes, CONFLICT_CODES, seen_locs, features, loc_index)
+
+        # Build URL + headline arrays for frontend rendering
+        for f in features:
+            urls = f["properties"].pop("_urls", [])
+            f["properties"].pop("_domains", None)
+            headlines = [_url_to_headline(u) for u in urls]
+            f["properties"]["_urls_list"] = urls
+            f["properties"]["_headlines_list"] = headlines
+            # Keep html as fallback
+            if urls:
+                links = [f'<div style="margin-bottom:6px;"><a href="{u}" target="_blank">{h}</a></div>' for u, h in zip(urls, headlines)]
+                f["properties"]["html"] = ''.join(links)
+            else:
+                f["properties"]["html"] = f["properties"]["name"]
+            f.pop("_loc_key", None)
+
+        logger.info(f"GDELT multi-file parsed: {len(features)} conflict locations from {successful} files")
+        return features
+
+    except Exception as e:
+        logger.error(f"Error fetching GDELT data: {e}")
+    return []

backend/services/liveuamap_scraper.py

@@ -0,0 +1,98 @@
+import json
+import logging
+import base64
+import urllib.parse
+import re
+from playwright.sync_api import sync_playwright
+from playwright_stealth import stealth_sync
+
+logger = logging.getLogger(__name__)
+
+def fetch_liveuamap():
+    logger.info("Starting Liveuamap scraper with Playwright Stealth...")
+    
+    regions = [
+        {"name": "Ukraine", "url": "https://liveuamap.com"},
+        {"name": "Middle East", "url": "https://mideast.liveuamap.com"},
+        {"name": "Israel-Palestine", "url": "https://israelpalestine.liveuamap.com"},
+        {"name": "Syria", "url": "https://syria.liveuamap.com"}
+    ]
+    
+    all_markers = []
+    seen_ids = set()
+    
+    with sync_playwright() as p:
+        # Launching with a real user agent to bypass Turnstile
+        browser = p.chromium.launch(headless=False, args=["--disable-blink-features=AutomationControlled"])
+        context = browser.new_context(
+            user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
+            viewport={"width": 1920, "height": 1080},
+            color_scheme="dark"
+        )
+        page = context.new_page()
+        stealth_sync(page)
+        
+        for region in regions:
+            try:
+                logger.info(f"Scraping Liveuamap region: {region['name']}")
+                page.goto(region["url"], timeout=60000, wait_until="domcontentloaded")
+                
+                # Wait for the map canvas or markers script to load, max 10s wait
+                try:
+                    page.wait_for_timeout(5000)
+                except:
+                    pass
+                
+                html = page.content()
+                
+                m = re.search(r"var\s+ovens\s*=\s*(.*?);(?!function)", html, re.DOTALL)
+                if not m:
+                    logger.warning(f"Could not find 'ovens' data for {region['name']} in raw HTML")
+                    # Let's try grabbing the evaluated JavaScript variable if it's there
+                    try:
+                        ovens_json = page.evaluate("() => typeof ovens !== 'undefined' ? JSON.stringify(ovens) : null")
+                        if ovens_json:
+                            markers = json.loads(ovens_json)
+                            # process below
+                            html = f"var ovens={ovens_json};"
+                            m = re.search(r"var\s+ovens=(.*?);", html, re.DOTALL)
+                    except:
+                        pass
+                
+                if m:
+                    json_str = m.group(1).strip()
+                    if json_str.startswith("'") or json_str.startswith('"'):
+                        json_str = json_str.strip('"\'')
+                        json_str = base64.b64decode(urllib.parse.unquote(json_str)).decode('utf-8')
+                        
+                    try:
+                        markers = json.loads(json_str)
+                        for marker in markers:
+                            mid = marker.get("id")
+                            if mid and mid not in seen_ids:
+                                seen_ids.add(mid)
+                                all_markers.append({
+                                    "id": mid,
+                                    "type": "liveuamap",
+                                    "title": marker.get("s", "Unknown Event") or marker.get("title", ""),
+                                    "lat": marker.get("lat"),
+                                    "lng": marker.get("lng"),
+                                    "timestamp": marker.get("time", ""),
+                                    "link": marker.get("link", region["url"]),
+                                    "region": region["name"]
+                                })
+                    except Exception as e:
+                        logger.error(f"Error parsing JSON for {region['name']}: {e}")
+                        
+            except Exception as e:
+                logger.error(f"Error scraping Liveuamap {region['name']}: {e}")
+                
+        browser.close()
+        
+    logger.info(f"Liveuamap scraper finished, extracted {len(all_markers)} unique markers.")
+    return all_markers
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO)
+    res = fetch_liveuamap()
+    print(json.dumps(res[:3], indent=2))

backend/services/network_utils.py

@@ -0,0 +1,90 @@
+import logging
+import json
+import subprocess
+import shutil
+import time
+from urllib.parse import urlparse
+
+logger = logging.getLogger(__name__)
+
+# Find bash for curl fallback — Git bash's curl has the TLS features
+# needed to pass CDN fingerprint checks (brotli, zstd, libpsl)
+_BASH_PATH = shutil.which("bash") or "bash"
+
+# Cache domains where requests fails — skip straight to curl for 5 minutes
+_domain_fail_cache: dict[str, float] = {}
+_DOMAIN_FAIL_TTL = 300  # 5 minutes
+
+class _DummyResponse:
+    """Minimal response object matching requests.Response interface."""
+    def __init__(self, status_code, text):
+        self.status_code = status_code
+        self.text = text
+        self.content = text.encode('utf-8', errors='replace')
+
+    def json(self):
+        return json.loads(self.text)
+
+    def raise_for_status(self):
+        if self.status_code >= 400:
+            raise Exception(f"HTTP {self.status_code}: {self.text[:100]}")
+
+
+def fetch_with_curl(url, method="GET", json_data=None, timeout=15, headers=None):
+    """Wrapper to bypass aggressive local firewall that blocks Python but permits curl.
+
+    Falls back to running curl through Git bash, which has the TLS features
+    (brotli, zstd, libpsl) needed to pass CDN fingerprint checks that block
+    both Python requests and the barebones Windows system curl.
+    """
+    default_headers = {
+        "User-Agent": "ShadowBroker-OSINT/1.0 (live-risk-dashboard)",
+    }
+    if headers:
+        default_headers.update(headers)
+
+    domain = urlparse(url).netloc
+
+    # Check if this domain recently failed with requests — skip straight to curl
+    if domain in _domain_fail_cache and (time.time() - _domain_fail_cache[domain]) < _DOMAIN_FAIL_TTL:
+        pass  # Fall through to curl below
+    else:
+        try:
+            import requests
+            if method == "POST":
+                res = requests.post(url, json=json_data, timeout=timeout, headers=default_headers)
+            else:
+                res = requests.get(url, timeout=timeout, headers=default_headers)
+            res.raise_for_status()
+            # Clear failure cache on success
+            _domain_fail_cache.pop(domain, None)
+            return res
+        except Exception as e:
+            logger.warning(f"Python requests failed for {url} ({e}), falling back to bash curl...")
+            _domain_fail_cache[domain] = time.time()
+
+        # Build curl command string for bash execution
+        header_flags = " ".join(f'-H "{k}: {v}"' for k, v in default_headers.items())
+        if method == "POST" and json_data:
+            payload = json.dumps(json_data).replace('"', '\\"')
+            curl_cmd = f'curl -s -w "\\n%{{http_code}}" {header_flags} -X POST -H "Content-Type: application/json" -d "{payload}" "{url}"'
+        else:
+            curl_cmd = f'curl -s -w "\\n%{{http_code}}" {header_flags} "{url}"'
+
+        try:
+            res = subprocess.run(
+                [_BASH_PATH, "-c", curl_cmd],
+                capture_output=True, text=True, timeout=timeout + 5
+            )
+            if res.returncode == 0 and res.stdout.strip():
+                # Parse HTTP status code from -w output (last line)
+                lines = res.stdout.rstrip().rsplit("\n", 1)
+                body = lines[0] if len(lines) > 1 else res.stdout
+                http_code = int(lines[-1]) if len(lines) > 1 and lines[-1].strip().isdigit() else 200
+                return _DummyResponse(http_code, body)
+            else:
+                logger.error(f"bash curl fallback failed: exit={res.returncode} stderr={res.stderr[:200]}")
+                return _DummyResponse(500, "")
+        except Exception as curl_e:
+            logger.error(f"bash curl fallback exception: {curl_e}")
+            return _DummyResponse(500, "")

backend/services/radio_intercept.py

@@ -0,0 +1,177 @@
+import requests
+from bs4 import BeautifulSoup
+import logging
+from cachetools import cached, TTLCache
+import cloudscraper
+import reverse_geocoder as rg
+
+logger = logging.getLogger(__name__)
+
+# Cache the top feeds for 5 minutes so we don't hammer Broadcastify
+radio_cache = TTLCache(maxsize=1, ttl=300)
+
+@cached(radio_cache)
+def get_top_broadcastify_feeds():
+    """
+    Scrapes the Broadcastify Top 50 live audio feeds public dashboard.
+    Returns a list of dictionaries containing feed metadata and direct stream URLs.
+    """
+    logger.info("Scraping Broadcastify Top Feeds (Cache Miss)")
+    headers = {
+        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
+        'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8',
+        'Accept-Language': 'en-US,en;q=0.9',
+    }
+    
+    try:
+        res = requests.get("https://www.broadcastify.com/listen/top", headers=headers, timeout=10)
+        if res.status_code != 200:
+            logger.error(f"Broadcastify Scrape Failed: HTTP {res.status_code}")
+            return []
+            
+        soup = BeautifulSoup(res.text, 'html.parser')
+        
+        table = soup.find('table', {'class': 'btable'})
+        if not table:
+            logger.error("Could not find feeds table on Broadcastify.")
+            return []
+            
+        feeds = []
+        rows = table.find_all('tr')[1:] # Skip header row
+        
+        for row in rows:
+            cols = row.find_all('td')
+            if len(cols) >= 5:
+                # Top layout: [Listeners, Feed ID (hidden), Location, Feed Name, Category, Genre]
+                listeners_str = cols[0].text.strip().replace(',', '')
+                listeners = int(listeners_str) if listeners_str.isdigit() else 0
+                
+                link_tag = cols[2].find('a')
+                if not link_tag:
+                    continue
+                    
+                href = link_tag.get('href', '')
+                feed_id = href.split('/')[-1] if '/listen/feed/' in href else None
+                
+                if not feed_id:
+                    continue
+                    
+                location = cols[1].text.strip()
+                name = cols[2].text.strip()
+                category = cols[3].text.strip()
+                
+                feeds.append({
+                    "id": feed_id,
+                    "listeners": listeners,
+                    "location": location,
+                    "name": name,
+                    "category": category,
+                    "stream_url": f"https://broadcastify.cdnstream1.com/{feed_id}"
+                })
+                
+        logger.info(f"Successfully scraped {len(feeds)} top feeds from Broadcastify.")
+        return feeds
+        
+    except Exception as e:
+        logger.error(f"Broadcastify Scrape Exception: {e}")
+        return []
+
+# Cache OpenMHZ systems mapping so we don't have to fetch all 450+ every time
+openmhz_systems_cache = TTLCache(maxsize=1, ttl=3600)
+
+@cached(openmhz_systems_cache)
+def get_openmhz_systems():
+    """Fetches the full directory of OpenMHZ systems."""
+    logger.info("Scraping OpenMHZ Systems (Cache Miss)")
+    scraper = cloudscraper.create_scraper(browser={'browser': 'chrome', 'platform': 'windows', 'desktop': True})
+    
+    try:
+        res = scraper.get("https://api.openmhz.com/systems", timeout=15)
+        if res.status_code == 200:
+            data = res.json()
+            # Return list of systems
+            return data.get('systems', []) if isinstance(data, dict) else []
+        return []
+    except Exception as e:
+        logger.error(f"OpenMHZ Systems Scrape Exception: {e}")
+        return []
+
+# Cache specific city calls briefly (15-30s) to limit our polling rate
+openmhz_calls_cache = TTLCache(maxsize=100, ttl=20)
+
+@cached(openmhz_calls_cache)
+def get_recent_openmhz_calls(sys_name: str):
+    """Fetches the actual audio burst .m4a URLs for a specific system (e.g., 'wmata')."""
+    logger.info(f"Fetching OpenMHZ calls for {sys_name} (Cache Miss)")
+    scraper = cloudscraper.create_scraper(browser={'browser': 'chrome', 'platform': 'windows', 'desktop': True})
+    
+    try:
+        url = f"https://api.openmhz.com/{sys_name}/calls"
+        res = scraper.get(url, timeout=15)
+        if res.status_code == 200:
+            data = res.json()
+            return data.get('calls', []) if isinstance(data, dict) else []
+        return []
+    except Exception as e:
+        logger.error(f"OpenMHZ Calls Scrape Exception ({sys_name}): {e}")
+        return []
+
+US_STATES = {
+    'Alabama': 'AL', 'Alaska': 'AK', 'Arizona': 'AZ', 'Arkansas': 'AR', 'California': 'CA',
+    'Colorado': 'CO', 'Connecticut': 'CT', 'Delaware': 'DE', 'Florida': 'FL', 'Georgia': 'GA',
+    'Hawaii': 'HI', 'Idaho': 'ID', 'Illinois': 'IL', 'Indiana': 'IN', 'Iowa': 'IA',
+    'Kansas': 'KS', 'Kentucky': 'KY', 'Louisiana': 'LA', 'Maine': 'ME', 'Maryland': 'MD',
+    'Massachusetts': 'MA', 'Michigan': 'MI', 'Minnesota': 'MN', 'Mississippi': 'MS',
+    'Missouri': 'MO', 'Montana': 'MT', 'Nebraska': 'NE', 'Nevada': 'NV', 'New Hampshire': 'NH',
+    'New Jersey': 'NJ', 'New Mexico': 'NM', 'New York': 'NY', 'North Carolina': 'NC',
+    'North Dakota': 'ND', 'Ohio': 'OH', 'Oklahoma': 'OK', 'Oregon': 'OR', 'Pennsylvania': 'PA',
+    'Rhode Island': 'RI', 'South Carolina': 'SC', 'South Dakota': 'SD', 'Tennessee': 'TN',
+    'Texas': 'TX', 'Utah': 'UT', 'Vermont': 'VT', 'Virginia': 'VA', 'Washington': 'WA',
+    'West Virginia': 'WV', 'Wisconsin': 'WI', 'Wyoming': 'WY', 'Washington, D.C.': 'DC', 'District of Columbia': 'DC'
+}
+
+import math
+
+def haversine_distance(lat1, lon1, lat2, lon2):
+    R = 3958.8 # Earth radius in miles
+    dLat = math.radians(lat2 - lat1)
+    dLon = math.radians(lon2 - lon1)
+    a = math.sin(dLat/2) * math.sin(dLat/2) + \
+        math.cos(math.radians(lat1)) * math.cos(math.radians(lat2)) * \
+        math.sin(dLon/2) * math.sin(dLon/2)
+    c = 2 * math.atan2(math.sqrt(a), math.sqrt(1-a))
+    return R * c
+
+def find_nearest_openmhz_systems_list(lat: float, lng: float, limit: int = 5):
+    """
+    Finds the strictly nearest OpenMHZ systems by distance.
+    """
+    systems = get_openmhz_systems()
+    if not systems:
+        return []
+
+    # Calculate distance for all systems that provide coordinates
+    valid_systems = []
+    for s in systems:
+        s_lat = s.get('lat')
+        s_lng = s.get('lng')
+        if s_lat is not None and s_lng is not None:
+            dist = haversine_distance(lat, lng, float(s_lat), float(s_lng))
+            s['distance_miles'] = dist
+            valid_systems.append(s)
+
+    if not valid_systems:
+        return []
+
+    # Sort strictly by distance
+    valid_systems.sort(key=lambda x: x['distance_miles'])
+    return valid_systems[:limit]
+
+def find_nearest_openmhz_system(lat: float, lng: float):
+    """
+    Returns the single closest OpenMHZ system by distance.
+    """
+    nearest = find_nearest_openmhz_systems_list(lat, lng, limit=1)
+    if nearest:
+        return nearest[0]
+    return None

backend/services/region_dossier.py

@@ -0,0 +1,202 @@
+import logging
+import concurrent.futures
+from urllib.parse import quote
+from cachetools import TTLCache
+from services.network_utils import fetch_with_curl
+
+logger = logging.getLogger(__name__)
+
+# Cache dossier results for 24 hours — country data barely changes
+# Key: rounded lat/lng grid (0.1 degree ≈ 11km)
+dossier_cache = TTLCache(maxsize=500, ttl=86400)
+
+
+def _reverse_geocode(lat: float, lng: float) -> dict:
+    url = (
+        f"https://nominatim.openstreetmap.org/reverse?"
+        f"lat={lat}&lon={lng}&format=json&zoom=10&addressdetails=1&accept-language=en"
+    )
+    try:
+        res = fetch_with_curl(url, timeout=10)
+        if res.status_code == 200:
+            data = res.json()
+            addr = data.get("address", {})
+            return {
+                "city": addr.get("city") or addr.get("town") or addr.get("village") or addr.get("county") or "",
+                "state": addr.get("state") or addr.get("region") or "",
+                "country": addr.get("country") or "",
+                "country_code": (addr.get("country_code") or "").upper(),
+                "display_name": data.get("display_name", ""),
+            }
+    except Exception as e:
+        logger.warning(f"Reverse geocode failed: {e}")
+    return {}
+
+
+def _fetch_country_data(country_code: str) -> dict:
+    if not country_code:
+        return {}
+    url = (
+        f"https://restcountries.com/v3.1/alpha/{country_code}"
+        f"?fields=name,population,capital,languages,region,subregion,area,currencies,borders,flag"
+    )
+    try:
+        res = fetch_with_curl(url, timeout=10)
+        if res.status_code == 200:
+            return res.json()
+    except Exception as e:
+        logger.warning(f"RestCountries failed for {country_code}: {e}")
+    return {}
+
+
+def _fetch_wikidata_leader(country_name: str) -> dict:
+    if not country_name:
+        return {"leader": "Unknown", "government_type": "Unknown"}
+    # SPARQL: get head of state (P35) and form of government (P122) for a sovereign state
+    safe_name = country_name.replace('"', '\\"').replace("'", "\\'")
+    sparql = f"""
+    SELECT ?leaderLabel ?govTypeLabel WHERE {{
+      ?country wdt:P31 wd:Q6256 ;
+               rdfs:label "{safe_name}"@en .
+      OPTIONAL {{ ?country wdt:P35 ?leader . }}
+      OPTIONAL {{ ?country wdt:P122 ?govType . }}
+      SERVICE wikibase:label {{ bd:serviceParam wikibase:language "en". }}
+    }} LIMIT 1
+    """
+    url = f"https://query.wikidata.org/sparql?query={quote(sparql)}&format=json"
+    try:
+        res = fetch_with_curl(url, timeout=15)
+        if res.status_code == 200:
+            results = res.json().get("results", {}).get("bindings", [])
+            if results:
+                r = results[0]
+                return {
+                    "leader": r.get("leaderLabel", {}).get("value", "Unknown"),
+                    "government_type": r.get("govTypeLabel", {}).get("value", "Unknown"),
+                }
+    except Exception as e:
+        logger.warning(f"Wikidata SPARQL failed for {country_name}: {e}")
+    return {"leader": "Unknown", "government_type": "Unknown"}
+
+
+def _fetch_local_wiki_summary(place_name: str, country_name: str = "") -> dict:
+    if not place_name:
+        return {}
+    # Try exact match first, then with country qualifier
+    candidates = [place_name]
+    if country_name:
+        candidates.append(f"{place_name}, {country_name}")
+
+    for name in candidates:
+        slug = quote(name.replace(" ", "_"))
+        url = f"https://en.wikipedia.org/api/rest_v1/page/summary/{slug}"
+        try:
+            res = fetch_with_curl(url, timeout=10)
+            if res.status_code == 200:
+                data = res.json()
+                if data.get("type") != "disambiguation":
+                    return {
+                        "description": data.get("description", ""),
+                        "extract": data.get("extract", ""),
+                        "thumbnail": data.get("thumbnail", {}).get("source", ""),
+                    }
+        except Exception:
+            continue
+    return {}
+
+
+def get_region_dossier(lat: float, lng: float) -> dict:
+    cache_key = f"{round(lat, 1)}_{round(lng, 1)}"
+    if cache_key in dossier_cache:
+        return dossier_cache[cache_key]
+
+    # Step 1: Reverse geocode
+    geo = _reverse_geocode(lat, lng)
+    if not geo or not geo.get("country"):
+        return {
+            "coordinates": {"lat": lat, "lng": lng},
+            "location": geo or {},
+            "country": None,
+            "local": None,
+            "error": "No country data — possibly international waters or uninhabited area",
+        }
+
+    country_code = geo.get("country_code", "")
+    country_name = geo.get("country", "")
+    city_name = geo.get("city", "")
+    state_name = geo.get("state", "")
+
+    # Step 2: Parallel fetch with timeouts to prevent hanging
+    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool:
+        country_fut = pool.submit(_fetch_country_data, country_code)
+        leader_fut = pool.submit(_fetch_wikidata_leader, country_name)
+        local_fut = pool.submit(_fetch_local_wiki_summary, city_name or state_name, country_name)
+        # Also fetch country-level Wikipedia summary as fallback for local
+        country_wiki_fut = pool.submit(_fetch_local_wiki_summary, country_name, "")
+
+    try:
+        country_data = country_fut.result(timeout=12)
+    except Exception:
+        logger.warning("Country data fetch timed out or failed")
+        country_data = {}
+    try:
+        leader_data = leader_fut.result(timeout=12)
+    except Exception:
+        logger.warning("Leader data fetch timed out or failed")
+        leader_data = {"leader": "Unknown", "government_type": "Unknown"}
+    try:
+        local_data = local_fut.result(timeout=12)
+    except Exception:
+        logger.warning("Local wiki fetch timed out or failed")
+        local_data = {}
+    try:
+        country_wiki_data = country_wiki_fut.result(timeout=12)
+    except Exception:
+        country_wiki_data = {}
+
+    # If no local data but we have country wiki summary, use that
+    if not local_data.get("extract") and country_wiki_data.get("extract"):
+        local_data = country_wiki_data
+
+    # Build languages list
+    languages = country_data.get("languages", {})
+    lang_list = list(languages.values()) if isinstance(languages, dict) else []
+
+    # Build currencies
+    currencies = country_data.get("currencies", {})
+    currency_list = []
+    if isinstance(currencies, dict):
+        for v in currencies.values():
+            if isinstance(v, dict):
+                symbol = v.get("symbol", "")
+                name = v.get("name", "")
+                currency_list.append(f"{name} ({symbol})" if symbol else name)
+
+    result = {
+        "coordinates": {"lat": lat, "lng": lng},
+        "location": geo,
+        "country": {
+            "name": country_data.get("name", {}).get("common", country_name),
+            "official_name": country_data.get("name", {}).get("official", ""),
+            "leader": leader_data.get("leader", "Unknown"),
+            "government_type": leader_data.get("government_type", "Unknown"),
+            "population": country_data.get("population", 0),
+            "capital": (country_data.get("capital") or ["Unknown"])[0] if isinstance(country_data.get("capital"), list) else "Unknown",
+            "languages": lang_list,
+            "currencies": currency_list,
+            "region": country_data.get("region", ""),
+            "subregion": country_data.get("subregion", ""),
+            "area_km2": country_data.get("area", 0),
+            "flag_emoji": country_data.get("flag", ""),
+        },
+        "local": {
+            "name": city_name,
+            "state": state_name,
+            "description": local_data.get("description", ""),
+            "summary": local_data.get("extract", ""),
+            "thumbnail": local_data.get("thumbnail", ""),
+        },
+    }
+
+    dossier_cache[cache_key] = result
+    return result

backend/services/test_flights.py

@@ -0,0 +1,17 @@
+import sys
+import logging
+logging.basicConfig(level=logging.DEBUG)
+
+# Add backend directory to sys path so we can import modules
+sys.path.append(r'f:\Codebase\Oracle\live-risk-dashboard\backend')
+
+from services.data_fetcher import fetch_flights, latest_data
+
+print("Testing fetch_flights...")
+try:
+    fetch_flights()
+    print("Commercial flights count:", len(latest_data.get('commercial_flights', [])))
+    print("Private jets count:", len(latest_data.get('private_jets', [])))
+except Exception as e:
+    import traceback
+    traceback.print_exc()

backend/services/test_liveuamap.py

@@ -0,0 +1,38 @@
+import json
+from playwright.sync_api import sync_playwright
+
+def scrape_liveuamap():
+    print("Launching playwright...")
+    with sync_playwright() as p:
+        # User agents are important for headless browsing
+        browser = p.chromium.launch(headless=True)
+        page = browser.new_page(user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36")
+        
+        def handle_response(response):
+            try:
+                if not response.url.endswith(('js', 'css', 'png', 'jpg', 'woff2', 'svg', 'ico')):
+                    print(f"Intercepted API Call: {response.url}")
+            except Exception:
+                pass
+                    
+        page.on("response", handle_response)
+        
+        print("Navigating to liveuamap...")
+        try:
+            page.goto("https://liveuamap.com/", timeout=30000, wait_until="domcontentloaded")
+            page.wait_for_timeout(5000)
+            
+            print("Grabbing all script tags...")
+            scripts = page.evaluate("() => Array.from(document.querySelectorAll('script')).map(s => s.innerText)")
+            for i, s in enumerate(scripts):
+                if 'JSON.parse' in s or 'markers' in s or 'JSON' in s:
+                    with open(f"script_{i}.txt", "w", encoding="utf-8") as f:
+                        f.write(s)
+        except Exception as e:
+            print("Playwright timeout or error:", e)
+            
+        print("Closing browser...")
+        browser.close()
+
+if __name__ == "__main__":
+    scrape_liveuamap()

backend/services/test_openmhz_scraper.py

@@ -0,0 +1,59 @@
+import requests
+import json
+import time
+import cloudscraper
+
+def scrape_openmhz_systems():
+    print("Testing OpenMHZ undocumented API with Cloudscraper...")
+    headers = {
+        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
+    }
+    
+    scraper = cloudscraper.create_scraper(browser={'browser': 'chrome', 'platform': 'windows', 'desktop': True})
+    
+    try:
+        # Step 1: Hit the public systems list that the front-end map uses
+        res = scraper.get("https://api.openmhz.com/systems", headers=headers, timeout=15)
+        json_data = res.json()
+        systems = json_data.get('systems', []) if isinstance(json_data, dict) else []
+        print(f"Successfully spoofed OpenMHZ frontend. Found {len(systems)} active police/fire systems.")
+        
+        if not systems:
+            return
+            
+        # Inspect the first system (usually a major city)
+        city = systems[0]
+        sys_name = city.get('shortName')
+        print(f"Targeting System: {city.get('name')} ({sys_name})")
+        
+        if not sys_name:
+            return
+            
+        time.sleep(2) # Ethical delay
+        
+        # Step 2: Query the recent calls for this specific system
+        # The frontend queries: https://api.openmhz.com/<system_name>/calls
+        calls_url = f"https://api.openmhz.com/{sys_name}/calls"
+        print(f"Fetching recent bursts: {calls_url}")
+        
+        call_res = scraper.get(calls_url, headers=headers, timeout=15)
+        
+        if call_res.status_code == 200:
+            call_json = call_res.json()
+            calls = call_json.get('calls', []) if isinstance(call_json, dict) else []
+            if calls and len(calls) > 0:
+                print(f"Intercepted {len(calls)} audio bursts.")
+                latest = calls[0]
+                print("LATEST INTERCEPT:")
+                print(f"Talkgroup: {latest.get('talkgroupNum')}")
+                print(f"Audio URL: {latest.get('url')}")
+            else:
+                 print("No recent calls found for this system.")
+        else:
+             print(f"Failed to fetch calls. HTTP {call_res.status_code}")
+             
+    except Exception as e:
+        print(f"Scrape Exception: {e}")
+
+if __name__ == "__main__":
+    scrape_openmhz_systems()

backend/services/test_radio.py

@@ -0,0 +1,19 @@
+import requests
+
+def test_openmhz():
+    print("Testing OpenMHZ...")
+    res = requests.get("https://api.openmhz.com/systems")
+    if res.status_code == 200:
+        data = res.json()
+        print(f"OpenMHZ returned {len(data)} systems.")
+        print(f"Example: {data[0]['name']} ({data[0]['shortName']})")
+    else:
+        print(f"OpenMHZ Failed: {res.status_code}")
+
+def test_scanner_radio():
+    print("Testing Scanner Radio...")
+    # Gordon Edwards app often uses something like this
+    # We will just try broadcastify public page scrape as a secondary fallback
+    pass
+
+test_openmhz()

backend/services/test_rss.py

@@ -0,0 +1,55 @@
+import feedparser
+import requests
+import re
+
+feeds = {
+    "NPR": "https://feeds.npr.org/1004/rss.xml",
+    "BBC": "http://feeds.bbci.co.uk/news/world/rss.xml"
+}
+
+keyword_coords = {
+    "venezuela": (7.119, -66.589), "brazil": (-14.235, -51.925), "argentina": (-38.416, -63.616),
+    "colombia": (4.570, -74.297), "mexico": (23.634, -102.552), "united states": (38.907, -77.036),
+    " usa ": (38.907, -77.036), " us ": (38.907, -77.036), "washington": (38.907, -77.036),
+    "canada": (56.130, -106.346), "ukraine": (49.487, 31.272), "kyiv": (50.450, 30.523),
+    "russia": (61.524, 105.318), "moscow": (55.755, 37.617), "israel": (31.046, 34.851),
+    "gaza": (31.416, 34.333), "iran": (32.427, 53.688), "lebanon": (33.854, 35.862),
+    "syria": (34.802, 38.996), "yemen": (15.552, 48.516), "china": (35.861, 104.195),
+    "beijing": (39.904, 116.407), "taiwan": (23.697, 120.960), "north korea": (40.339, 127.510),
+    "south korea": (35.907, 127.766), "pyongyang": (39.039, 125.762), "seoul": (37.566, 126.978),
+    "japan": (36.204, 138.252), "afghanistan": (33.939, 67.709), "pakistan": (30.375, 69.345),
+    "india": (20.593, 78.962), " uk ": (55.378, -3.435), "london": (51.507, -0.127),
+    "france": (46.227, 2.213), "paris": (48.856, 2.352), "germany": (51.165, 10.451),
+    "berlin": (52.520, 13.405), "sudan": (12.862, 30.217), "congo": (-4.038, 21.758),
+    "south africa": (-30.559, 22.937), "nigeria": (9.082, 8.675), "egypt": (26.820, 30.802),
+    "zimbabwe": (-19.015, 29.154), "australia": (-25.274, 133.775), "middle east": (31.500, 34.800),
+    "europe": (48.800, 2.300), "africa": (0.000, 25.000), "america": (38.900, -77.000),
+    "south america": (-14.200, -51.900), "asia": (34.000, 100.000),
+    "california": (36.778, -119.417), "texas": (31.968, -99.901), "florida": (27.994, -81.760),
+    "new york": (40.712, -74.006), "virginia": (37.431, -78.656),
+    "british columbia": (53.726, -127.647), "ontario": (51.253, -85.323), "quebec": (52.939, -73.549),
+    "delhi": (28.704, 77.102), "new delhi": (28.613, 77.209), "mumbai": (19.076, 72.877),
+    "shanghai": (31.230, 121.473), "hong kong": (22.319, 114.169), "istanbul": (41.008, 28.978),
+    "dubai": (25.204, 55.270), "singapore": (1.352, 103.819)
+}
+
+for name, url in feeds.items():
+    r = requests.get(url)
+    feed = feedparser.parse(r.text)
+    for entry in feed.entries[:10]: 
+        title = entry.get('title', '')
+        summary = entry.get('summary', '')
+        text = (title + " " + summary).lower()
+        padded_text = f" {text} "
+        
+        matched_kw = None
+        for kw, coords in keyword_coords.items():
+            if kw.startswith(" ") or kw.endswith(" "):
+                if kw in padded_text:
+                    matched_kw = kw
+                    break
+            else:
+                if re.search(r'\b' + re.escape(kw) + r'\b', text):
+                    matched_kw = kw
+                    break
+        print(f"[{name}] {title}\n  Matched: {matched_kw}\n  Text: {text}\n")

backend/services/test_scraper.py

@@ -0,0 +1,67 @@
+import requests
+from bs4 import BeautifulSoup
+import json
+
+def scrape_broadcastify_top():
+    print("Scraping Broadcastify Top Feeds...")
+    headers = {
+        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
+    }
+    
+    try:
+        # The top 50 feeds page provides a wealth of listening data
+        res = requests.get("https://www.broadcastify.com/listen/top", headers=headers, timeout=10)
+        if res.status_code != 200:
+            print(f"Failed HTTP {res.status_code}")
+            return []
+            
+        soup = BeautifulSoup(res.text, 'html.parser')
+        
+        # The table of feeds is in a standard class
+        table = soup.find('table', {'class': 'btable'})
+        if not table:
+            print("Could not find feeds table.")
+            return []
+            
+        feeds = []
+        rows = table.find_all('tr')[1:] # Skip header
+        
+        for row in rows:
+            cols = row.find_all('td')
+            if len(cols) >= 5:
+                # Top layout: [Listeners, Feed ID (hidden), Location, Feed Name, Category, Genre]
+                listeners_str = cols[0].text.strip().replace(',', '')
+                listeners = int(listeners_str) if listeners_str.isdigit() else 0
+                
+                # The link is usually in the Feed Name column
+                link_tag = cols[2].find('a')
+                if not link_tag:
+                    continue
+                    
+                href = link_tag.get('href', '')
+                feed_id = href.split('/')[-1] if '/listen/feed/' in href else None
+                
+                if not feed_id:
+                    continue
+                    
+                location = cols[1].text.strip()
+                name = cols[2].text.strip()
+                
+                feeds.append({
+                    "id": feed_id,
+                    "listeners": listeners,
+                    "location": location,
+                    "name": name,
+                    "stream_url": f"https://broadcastify.cdnstream1.com/{feed_id}"
+                })
+                
+        print(f"Successfully scraped {len(feeds)} top feeds.")
+        return feeds
+        
+    except Exception as e:
+        print(f"Scrape error: {e}")
+        return []
+
+if __name__ == "__main__":
+    top_feeds = scrape_broadcastify_top()
+    print(json.dumps(top_feeds[:3], indent=2))