fix: stabilize v0.9.7 startup and feeds

This commit is contained in:
BigBodyCobain committed 2026-05-02 13:35:49 -06:00
1 parent f5b9d14b48
commit 08810f2537
22 files changed
+940 -130

No files matched your search

+158 -5
View File
@@ -15,7 +15,7 @@ import time
import heapq
from datetime import datetime, timedelta
from pathlib import Path
from services.network_utils import fetch_with_curl
from services.network_utils import external_curl_fallback_enabled, fetch_with_curl
from services.fetchers._store import latest_data, _data_lock, _mark_fresh
from services.fetchers.nuforc_enrichment import enrich_sighting
from services.fetchers.retry import with_retry
@@ -685,6 +685,8 @@ _NUFORC_TOKEN = os.environ.get("NUFORC_MAPBOX_TOKEN", "").strip()
_NUFORC_RADIUS_M = 200_000 # 200 km query radius
_NUFORC_LIMIT = 50 # max features per tilequery call
_NUFORC_RECENT_DAYS = int(os.environ.get("NUFORC_RECENT_DAYS", "60"))
_NUFORC_HF_FALLBACK_LIMIT = max(25, int(os.environ.get("NUFORC_HF_FALLBACK_LIMIT", "250")))
_NUFORC_HF_GEOCODE_LIMIT = max(25, int(os.environ.get("NUFORC_HF_GEOCODE_LIMIT", "150")))
_NUFORC_GEOCODE_WORKERS = max(1, int(os.environ.get("NUFORC_GEOCODE_WORKERS", "1")))
# Photon (Komoot) is more lenient than Nominatim — ~200ms per query in
# practice, so a 0.3s spacing keeps us well under any soft throttle while
@@ -1034,6 +1036,14 @@ def _nuforc_fetch_month_live(yyyymm: str, cookie_jar: Path) -> list[dict]:
index_url = _NUFORC_LIVE_INDEX_URL.format(yyyymm=yyyymm)
ajax_url = _NUFORC_LIVE_AJAX_URL.format(yyyymm=yyyymm)
if not external_curl_fallback_enabled():
logger.warning(
"NUFORC live: external curl disabled on Windows for %s; "
"set SHADOWBROKER_ENABLE_WINDOWS_CURL_FALLBACK=1 to opt in.",
yyyymm,
)
return []
# Step 1: GET the month index to capture session cookies + fresh nonce.
try:
index_res = subprocess.run(
@@ -1340,6 +1350,143 @@ def _build_recent_uap_sightings() -> list[dict]:
return sightings
def _split_uap_location(location: str) -> tuple[str, str, str]:
parts = [p.strip() for p in str(location or "").split(",") if p.strip()]
city = parts[0] if parts else ""
state = ""
country = ""
if len(parts) >= 2:
state = parts[1]
if len(parts) >= 3:
country = parts[-1]
if country and country.upper() in _US_COUNTRY_ALIASES:
country = "US"
return city, state, country
def _build_uap_sightings_from_hf_mirror() -> list[dict]:
"""Build visible UAP points from the public Hugging Face NUFORC mirror.
This is a resilience fallback for local/Windows runs where nuforc.org is
Cloudflare-gated and the Mapbox token is not configured. It is not as fresh
as the live NUFORC AJAX feed, but it keeps the layer visible and cached.
"""
from services.fetchers.nuforc_enrichment import _HF_CSV_URL, _parse_date
from services.geocode_validate import coord_in_country
try:
response = fetch_with_curl(_HF_CSV_URL, timeout=180, follow_redirects=True)
if not response or response.status_code != 200:
logger.warning(
"UAP sightings: HF fallback failed HTTP %s",
getattr(response, "status_code", "None"),
)
return []
except Exception as e:
logger.warning("UAP sightings: HF fallback download failed: %s", e)
return []
candidates: list[dict] = []
try:
reader = csv.DictReader(io.StringIO(response.text))
for row in reader:
occurred = _parse_date(
row.get("Occurred", "")
or row.get("Date / Time", "")
or row.get("Date", "")
)
if not occurred:
continue
raw_location = _normalize_uap_location(
row.get("Location", "")
or row.get("City", "")
or row.get("location", "")
)
if not raw_location:
continue
city, state, country = _split_uap_location(raw_location)
if not city:
continue
sighting_id = str(row.get("Sighting", "") or "").strip()
if not sighting_id:
sighting_id = hashlib.sha1(
f"{occurred}|{raw_location}|{row.get('Summary', '')}".encode("utf-8", "ignore")
).hexdigest()[:12]
summary = (row.get("Summary", "") or row.get("Text", "") or "Sighting reported").strip()
if len(summary) > 280:
summary = summary[:277] + "..."
candidates.append({
"id": f"NUFORC-{sighting_id}",
"occurred": occurred,
"posted": _parse_date(row.get("Posted", "") or row.get("Reported", "")) or occurred,
"location": raw_location,
"city": city,
"state": state,
"country": country or _uap_country_from_location(raw_location, state),
"shape_raw": (row.get("Shape", "") or "Unknown").strip(),
"duration": (row.get("Duration", "") or "").strip(),
"summary": summary,
})
except Exception as e:
logger.warning("UAP sightings: HF fallback parse failed: %s", e)
return []
candidates.sort(key=lambda row: (row["occurred"], row["posted"], row["id"]), reverse=True)
candidates = candidates[:_NUFORC_HF_FALLBACK_LIMIT]
location_cache = _load_nuforc_location_cache()
sightings: list[dict] = []
geocoded = 0
for row in candidates:
coords = location_cache.get(row["location"])
if row["location"] not in location_cache and geocoded < _NUFORC_HF_GEOCODE_LIMIT:
try:
coords = _geocode_uap_location(
row["location"], row["city"], row["state"], row["country"]
)
except Exception:
coords = None
location_cache[row["location"]] = coords
geocoded += 1
if geocoded < _NUFORC_HF_GEOCODE_LIMIT:
time.sleep(_NUFORC_GEOCODE_SPACING_S)
if not coords:
continue
if row.get("country"):
try:
inside = coord_in_country(coords[0], coords[1], row["country"])
except Exception:
inside = None
if inside is False:
continue
shape_raw = row["shape_raw"] or "Unknown"
sightings.append({
"id": row["id"],
"date_time": row["occurred"],
"city": row["city"],
"state": row["state"],
"country": row["country"],
"shape": _normalize_uap_shape(shape_raw) if shape_raw != "Unknown" else "unknown",
"shape_raw": shape_raw,
"duration": row["duration"],
"summary": row["summary"],
"posted": row["posted"],
"lat": float(coords[0]),
"lng": float(coords[1]),
"count": 1,
"source": "NUFORC-HF",
})
_save_nuforc_location_cache(location_cache)
logger.info(
"UAP sightings: %d mapped reports from HF fallback (%d candidates, %d geocoded)",
len(sightings),
len(candidates),
geocoded,
)
return sightings
@with_retry(max_retries=1, base_delay=5)
def fetch_uap_sightings(*, force_refresh: bool = False):
"""Fetch last-year UAP sightings from NUFORC.
@@ -1355,12 +1502,18 @@ def fetch_uap_sightings(*, force_refresh: bool = False):
sightings = _load_nuforc_sightings_cache(force_refresh=force_refresh)
if sightings is None:
sightings = _build_recent_uap_sightings()
_save_nuforc_sightings_cache(sightings)
try:
sightings = _build_recent_uap_sightings()
except Exception as e:
logger.warning("UAP sightings: live NUFORC rebuild failed, using fallback: %s", e)
sightings = _build_uap_sightings_from_hf_mirror()
if sightings:
_save_nuforc_sightings_cache(sightings)
with _data_lock:
latest_data["uap_sightings"] = sightings
_mark_fresh("uap_sightings")
latest_data["uap_sightings"] = sightings or []
if sightings:
_mark_fresh("uap_sightings")
return
cutoff = datetime.utcnow() - timedelta(days=_NUFORC_RECENT_DAYS)
+24
View File
@@ -15,6 +15,24 @@ from services.fetchers.retry import with_retry
logger = logging.getLogger(__name__)
def _env_flag(name: str) -> str:
return str(os.getenv(name, "")).strip().lower()
def liveuamap_scraper_enabled() -> bool:
"""Return whether the Playwright-based LiveUAMap scraper should run.
It is useful enrichment, but it starts a browser/Node driver and must not be
allowed to destabilize Windows local startup.
"""
setting = _env_flag("SHADOWBROKER_ENABLE_LIVEUAMAP_SCRAPER")
if setting in {"1", "true", "yes", "on"}:
return True
if setting in {"0", "false", "no", "off"}:
return False
return os.name != "nt"
# ---------------------------------------------------------------------------
# Ships (AIS + Carriers)
# ---------------------------------------------------------------------------
@@ -191,6 +209,12 @@ def update_liveuamap():
if not is_any_active("global_incidents"):
return
if not liveuamap_scraper_enabled():
logger.info(
"Liveuamap scraper disabled for this runtime; set "
"SHADOWBROKER_ENABLE_LIVEUAMAP_SCRAPER=1 to opt in."
)
return
logger.info("Running scheduled Liveuamap scraper...")
try:
from services.liveuamap_scraper import fetch_liveuamap
+119 -7
View File
@@ -15,6 +15,7 @@ Analysis features (derived from cached TLEs — no extra network requests):
import math
import time
import json
import os
import re
import logging
import requests
@@ -41,6 +42,38 @@ def _gmst(jd_ut1):
# CelesTrak fair use: fetch at most once per 24 hours (86400s).
# SGP4 propagation runs every 60s using cached TLEs — positions stay live.
_CELESTRAK_FETCH_INTERVAL = 86400 # 24 hours
_MIN_VISIBLE_SATELLITE_CATALOG = int(os.environ.get("SHADOWBROKER_MIN_VISIBLE_SATELLITES", "350"))
_MAX_VISIBLE_SATELLITE_CATALOG = int(os.environ.get("SHADOWBROKER_MAX_VISIBLE_SATELLITES", "450"))
_CELESTRAK_VISIBLE_GROUPS = {
"military": {"mission": "military", "sat_type": "Military / Defense"},
"radar": {"mission": "sar", "sat_type": "Radar / SAR"},
"resource": {"mission": "earth_observation", "sat_type": "Earth Observation"},
"weather": {"mission": "weather", "sat_type": "Weather / Meteorology"},
"gnss": {"mission": "navigation", "sat_type": "GNSS / Navigation"},
"science": {"mission": "science", "sat_type": "Science"},
}
_TLE_VISIBLE_FALLBACK_TERMS = {
"COSMOS": {"mission": "military", "sat_type": "Russian / Soviet Military"},
"USA": {"mission": "military", "sat_type": "US Military / NRO"},
"NROL": {"mission": "military", "sat_type": "Classified NRO"},
"GPS": {"mission": "navigation", "sat_type": "GPS Navigation"},
"GALILEO": {"mission": "navigation", "sat_type": "Galileo Navigation"},
"BEIDOU": {"mission": "navigation", "sat_type": "BeiDou Navigation"},
"GLONASS": {"mission": "navigation", "sat_type": "GLONASS Navigation"},
"NOAA": {"mission": "weather", "sat_type": "NOAA Weather"},
"METEOR": {"mission": "weather", "sat_type": "Meteor Weather"},
"SENTINEL": {"mission": "earth_observation", "sat_type": "Sentinel Earth Observation"},
"LANDSAT": {"mission": "earth_observation", "sat_type": "Landsat Earth Observation"},
"WORLDVIEW": {"mission": "commercial_imaging", "sat_type": "Maxar High-Res"},
"PLEIADES": {"mission": "commercial_imaging", "sat_type": "Airbus Imaging"},
"SKYSAT": {"mission": "commercial_imaging", "sat_type": "Planet Video"},
"JILIN": {"mission": "commercial_imaging", "sat_type": "Jilin Imaging"},
"FLOCK": {"mission": "commercial_imaging", "sat_type": "PlanetScope"},
"LEMUR": {"mission": "commercial_rf", "sat_type": "Spire RF / AIS"},
"ICEYE": {"mission": "sar", "sat_type": "ICEYE SAR"},
"UMBRA": {"mission": "sar", "sat_type": "Umbra SAR"},
"CAPELLA": {"mission": "sar", "sat_type": "Capella SAR"},
}
_sat_gp_cache = {"data": None, "last_fetch": 0, "source": "none", "last_modified": None}
_sat_classified_cache = {"data": None, "gp_fetch_ts": 0}
_SAT_CACHE_PATH = Path(__file__).parent.parent.parent / "data" / "sat_gp_cache.json"
@@ -564,9 +597,61 @@ def _parse_tle_to_gp(name, norad_id, line1, line2):
return None
def _annotate_celestrak_group(records: list[dict], group: str) -> list[dict]:
meta = _CELESTRAK_VISIBLE_GROUPS.get(group, {})
out = []
for sat in records:
if not isinstance(sat, dict):
continue
item = dict(sat)
item["_SB_GROUP"] = group
if meta:
item["_SB_GROUP_META"] = meta
out.append(item)
return out
def _fetch_visible_celestrak_catalog(headers: dict | None = None) -> list[dict]:
"""Fetch bounded CelesTrak groups used by the visible satellite layer.
The full ``active`` catalog is too large and frequently times out on local
startup. These groups cover the visible operational set users expect
without pulling Starlink-scale constellations into the map.
"""
headers = headers or {}
merged: dict[int, dict] = {}
for group in _CELESTRAK_VISIBLE_GROUPS:
url = f"https://celestrak.org/NORAD/elements/gp.php?GROUP={group}&FORMAT=json"
try:
response = fetch_with_curl(url, timeout=15, headers=headers)
if response.status_code != 200:
logger.debug("Satellites: CelesTrak group %s returned HTTP %s", group, response.status_code)
continue
gp_data = response.json()
if not isinstance(gp_data, list):
continue
for sat in _annotate_celestrak_group(gp_data, group):
norad_id = sat.get("NORAD_CAT_ID")
if norad_id is None:
continue
merged[int(norad_id)] = sat
time.sleep(0.35)
except (
requests.RequestException,
ConnectionError,
TimeoutError,
ValueError,
KeyError,
json.JSONDecodeError,
OSError,
) as e:
logger.warning("Satellites: Failed to fetch CelesTrak group %s: %s", group, e)
return list(merged.values())
def _fetch_satellites_from_tle_api():
"""Fallback: fetch satellite TLEs from tle.ivanstanojevic.me when CelesTrak is blocked."""
search_terms = set()
search_terms = set(_TLE_VISIBLE_FALLBACK_TERMS)
for key, _ in _SAT_INTEL_DB:
term = key.split()[0] if len(key.split()) > 1 and key.split()[0] in ("USA", "NROL") else key
search_terms.add(term)
@@ -591,8 +676,13 @@ def _fetch_satellites_from_tle_api():
sat_id = gp.get("NORAD_CAT_ID")
if sat_id not in seen_ids:
seen_ids.add(sat_id)
if term in _TLE_VISIBLE_FALLBACK_TERMS:
gp["_SB_GROUP"] = f"tle:{term}"
gp["_SB_GROUP_META"] = _TLE_VISIBLE_FALLBACK_TERMS[term]
all_results.append(gp)
time.sleep(1) # Polite delay between requests
if len(all_results) >= _MAX_VISIBLE_SATELLITE_CATALOG:
return all_results
time.sleep(0.15) # Polite delay between requests
except (
requests.RequestException,
ConnectionError,
@@ -644,18 +734,34 @@ def fetch_satellites():
if (
_sat_gp_cache["data"] is None
or len(_sat_gp_cache.get("data") or []) < _MIN_VISIBLE_SATELLITE_CATALOG
or (now_ts - _sat_gp_cache["last_fetch"]) > _CELESTRAK_FETCH_INTERVAL
):
gp_urls = [
"https://celestrak.org/NORAD/elements/gp.php?GROUP=active&FORMAT=json",
"https://celestrak.com/NORAD/elements/gp.php?GROUP=active&FORMAT=json",
]
# Build conditional request headers (CelesTrak fair use)
headers = {}
if _sat_gp_cache.get("last_modified"):
headers["If-Modified-Since"] = _sat_gp_cache["last_modified"]
visible_data = _fetch_visible_celestrak_catalog(headers=headers)
if len(visible_data) >= _MIN_VISIBLE_SATELLITE_CATALOG:
_sat_gp_cache["data"] = visible_data
_sat_gp_cache["last_fetch"] = now_ts
_sat_gp_cache["source"] = "celestrak_visible_groups"
_save_sat_cache(visible_data)
_snapshot_current_tles(visible_data)
logger.info(
"Satellites: Downloaded %d GP records from visible CelesTrak groups",
len(visible_data),
)
gp_urls = [
"https://celestrak.org/NORAD/elements/gp.php?GROUP=active&FORMAT=json",
"https://celestrak.com/NORAD/elements/gp.php?GROUP=active&FORMAT=json",
]
for url in gp_urls:
if len(_sat_gp_cache.get("data") or []) >= _MIN_VISIBLE_SATELLITE_CATALOG:
break
try:
response = fetch_with_curl(url, timeout=15, headers=headers)
if response.status_code == 304:
@@ -696,7 +802,10 @@ def fetch_satellites():
logger.warning(f"Satellites: Failed to fetch from {url}: {e}")
continue
if _sat_gp_cache["data"] is None:
if (
_sat_gp_cache["data"] is None
or len(_sat_gp_cache.get("data") or []) < _MIN_VISIBLE_SATELLITE_CATALOG
):
logger.info("Satellites: CelesTrak unreachable, trying TLE fallback API...")
try:
fallback_data = _fetch_satellites_from_tle_api()
@@ -757,6 +866,9 @@ def fetch_satellites():
owner = sat.get("OWNER", sat.get("OBJECT_OWNER", ""))
if owner in _OWNER_CODE_MAP:
intel = {"country": _OWNER_CODE_MAP[owner], "mission": "general", "sat_type": "Unclassified"}
if not intel and sat.get("_SB_GROUP_META"):
intel = dict(sat["_SB_GROUP_META"])
intel.setdefault("country", "Unknown")
if not intel:
continue