fbff9e5415
Add parse_mdot_json for the MDOT MiDrive camera/list JSON where coordinates/id live in the county field's map link (lat=/lon=/id=) and the JPEG still lives in the image field's <img src>. Michigan bbox filter, missing-coords drop, RTSP reject. discovery_source=mdot, stable source_url keyed on camera id, url_hash dedupe. Wired into scrape_source dispatch + CAMERA_SOURCE_URLS defaults. Unit tests: HTML field extract, bbox drop, missing-coords/image drop, malformed payload.
691 lines
27 KiB
Python
691 lines
27 KiB
Python
"""Camera discovery scraper — finds publicly listed open IP cameras.
|
||
|
||
Ethics/scope (hard constraints, enforced in code):
|
||
* Only indexes cameras that appear in PUBLIC directories/lists.
|
||
* No credential brute-forcing, no login attempts of any kind.
|
||
* No active scanning — never probes private (RFC1918/loopback/link-local)
|
||
ranges; private-range URLs found in public lists are dropped.
|
||
|
||
Pipeline per cycle:
|
||
1. Fetch each configured public source page/list.
|
||
2. Parse cameras (Insecam-style HTML with embedded lat/lon, or plain-text
|
||
`url[|lat,lon|vendor|location]` lines).
|
||
3. Geocode missing coords via Nominatim at <=1 req/s.
|
||
4. Publish each camera to NATS (`events.camera`) for the shared ingester,
|
||
and upsert into the `cameras` table keyed by sha256(url_hash).
|
||
|
||
Snapshots are cached locally with a TTL; nothing is fetched more often than
|
||
the cache allows.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import hashlib
|
||
import ipaddress
|
||
import json
|
||
import logging
|
||
import re
|
||
import time
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from urllib.parse import urlparse
|
||
|
||
import httpx
|
||
|
||
from camera_config import (
|
||
CAMERA_SOURCE_URLS, CAMERA_REQUEST_DELAY, CAMERA_MAX_PER_SOURCE,
|
||
NOMINATIM_URL, NOMINATIM_MIN_INTERVAL, USER_AGENT,
|
||
SNAPSHOT_CACHE_DIR, SNAPSHOT_TTL_SECONDS, SNAPSHOT_TIMEOUT,
|
||
)
|
||
from camera_models import cameras
|
||
from database import async_session
|
||
|
||
logger = logging.getLogger("osint.camera_scraper")
|
||
|
||
# ── Private-range guard ────────────────────────────────────────────────────
|
||
|
||
_public_host_cache: dict[str, bool] = {}
|
||
|
||
|
||
def is_public_url(url: str) -> bool:
|
||
"""True only if the URL host resolves to a globally-routable address.
|
||
|
||
Hostnames that cannot be resolved are treated as not-public (fail closed).
|
||
"""
|
||
try:
|
||
parsed = urlparse(url)
|
||
if parsed.scheme not in ("http", "https", "rtsp", "rtsps"):
|
||
return False
|
||
host = parsed.hostname
|
||
if not host:
|
||
return False
|
||
# Literal IPs: check directly.
|
||
try:
|
||
addr = ipaddress.ip_address(host)
|
||
return addr.is_global and not addr.is_private
|
||
except ValueError:
|
||
pass
|
||
if host in _public_host_cache:
|
||
return _public_host_cache[host]
|
||
# Private-name quick rejects before any DNS work.
|
||
if (host.endswith(".local") or host.endswith(".internal")
|
||
or host == "localhost" or re.fullmatch(r"10\..*|172\.(1[6-9]|2\d|3[01])\..*|192\.168\..*", host)):
|
||
_public_host_cache[host] = False
|
||
return False
|
||
import socket
|
||
ok = True
|
||
for info in socket.getaddrinfo(host, None):
|
||
addr = ipaddress.ip_address(info[4][0])
|
||
if not (addr.is_global and not addr.is_private):
|
||
ok = False
|
||
break
|
||
_public_host_cache[host] = ok
|
||
return ok
|
||
except Exception: # noqa: BLE001
|
||
return False
|
||
|
||
|
||
# ── Dedupe key ─────────────────────────────────────────────────────────────
|
||
|
||
def url_hash(url: str) -> str:
|
||
return hashlib.sha256(url.strip().lower().encode()).hexdigest()
|
||
|
||
|
||
# ── Politeness-limited fetcher ─────────────────────────────────────────────
|
||
|
||
class RateLimitedClient:
|
||
"""httpx client wrapper enforcing a per-host minimum request interval."""
|
||
|
||
def __init__(self, min_delay: float = CAMERA_REQUEST_DELAY):
|
||
self._delay = min_delay
|
||
self._last: dict[str, float] = {}
|
||
self.client = httpx.AsyncClient(
|
||
timeout=60, follow_redirects=True,
|
||
headers={"User-Agent": USER_AGENT},
|
||
)
|
||
|
||
async def get(self, url: str, **kw) -> httpx.Response:
|
||
host = urlparse(url).netloc
|
||
now = time.monotonic()
|
||
wait = self._last.get(host, 0.0) + self._delay - now
|
||
if wait > 0:
|
||
await asyncio.sleep(wait)
|
||
self._last[host] = time.monotonic()
|
||
return await self.client.get(url, **kw)
|
||
|
||
async def aclose(self):
|
||
await self.client.aclose()
|
||
|
||
|
||
# ── Nominatim geocoding (1 req/s politeness) ───────────────────────────────
|
||
|
||
class Geocoder:
|
||
def __init__(self):
|
||
self._min_interval = NOMINATIM_MIN_INTERVAL
|
||
self._last = 0.0
|
||
self._cache: dict[str, tuple[float | None, float | None]] = {}
|
||
self._lock = asyncio.Lock()
|
||
|
||
async def geocode(self, place: str) -> tuple[float | None, float | None]:
|
||
place = place.strip()
|
||
if not place:
|
||
return None, None
|
||
if place in self._cache:
|
||
return self._cache[place]
|
||
async with self._lock:
|
||
wait = self._last + self._min_interval - time.monotonic()
|
||
if wait > 0:
|
||
await asyncio.sleep(wait)
|
||
try:
|
||
async with httpx.AsyncClient(timeout=15) as c:
|
||
r = await c.get(
|
||
f"{NOMINATIM_URL}/search",
|
||
params={"q": place, "format": "json", "limit": 1},
|
||
headers={"User-Agent": USER_AGENT},
|
||
)
|
||
r.raise_for_status()
|
||
data = r.json()
|
||
result = (
|
||
(float(data[0]["lat"]), float(data[0]["lon"]))
|
||
if data else (None, None)
|
||
)
|
||
except Exception: # noqa: BLE001
|
||
logger.warning("geocode failed for %r", place, exc_info=True)
|
||
result = (None, None)
|
||
self._last = time.monotonic()
|
||
self._cache[place] = result
|
||
return result
|
||
|
||
|
||
# ── Snapshot cache (local disk, TTL'd) ─────────────────────────────────────
|
||
|
||
_snapshot_lock = asyncio.Lock()
|
||
|
||
|
||
async def fetch_snapshot(url: str) -> bytes | None:
|
||
"""Fetch a snapshot image through the local TTL cache. Returns raw bytes."""
|
||
key = url_hash(url)[:24]
|
||
path = Path(SNAPSHOT_CACHE_DIR) / f"{key}.bin"
|
||
meta = Path(SNAPSHOT_CACHE_DIR) / f"{key}.meta"
|
||
async with _snapshot_lock:
|
||
try:
|
||
if path.exists() and meta.exists():
|
||
age = time.time() - path.stat().st_mtime
|
||
if age < SNAPSHOT_TTL_SECONDS:
|
||
return path.read_bytes()
|
||
except OSError:
|
||
pass
|
||
try:
|
||
async with httpx.AsyncClient(timeout=SNAPSHOT_TIMEOUT, follow_redirects=True) as c:
|
||
r = await c.get(url, headers={"User-Agent": USER_AGENT})
|
||
if r.status_code != 200 or len(r.content) < 64:
|
||
return None
|
||
data = r.content
|
||
except Exception: # noqa: BLE001
|
||
return None
|
||
async with _snapshot_lock:
|
||
try:
|
||
Path(SNAPSHOT_CACHE_DIR).mkdir(parents=True, exist_ok=True)
|
||
path.write_bytes(data)
|
||
meta.write_text(json.dumps({"url": url, "fetched": time.time()}))
|
||
except OSError:
|
||
logger.warning("snapshot cache write failed", exc_info=True)
|
||
return data
|
||
|
||
|
||
# ── Source parsers ─────────────────────────────────────────────────────────
|
||
|
||
# Insecam-style pages embed camera entries like:
|
||
# <a href="/view/..."> ... latitude: 48.85 / longitude: 2.35 ... title="Paris"
|
||
# We parse generically: find lat/lon pairs plus nearby snapshot <img> sources.
|
||
_LATLON_RE = re.compile(
|
||
r"(?:latitude|lat)[\"'\s:=]+(-?\d{1,2}\.\d+).{0,200}?(?:longitude|lon)[\"'\s:=]+(-?\d{1,3}\.\d+)",
|
||
re.I | re.S,
|
||
)
|
||
_IMG_RE = re.compile(r"<img[^>]+src=[\"']([^\"']+\.(?:jpg|jpeg|png|mjpeg))[\"']", re.I)
|
||
_TITLE_RE = re.compile(r"<title>([^<]+)</title>", re.I)
|
||
|
||
|
||
def _vendor_from_url(url: str) -> str | None:
|
||
u = url.lower()
|
||
for vendor, marker in [
|
||
("hikvision", "hikvision"), ("dahua", "dahua"),
|
||
("axis", "axis-cgi"), ("foscam", "foscam"),
|
||
("tplink", "tplink"), ("tp-link", "tp-link"),
|
||
("mjpeg", "mjpeg"), ("rtsp", "rtsp://"),
|
||
]:
|
||
if marker in u:
|
||
return vendor
|
||
return None
|
||
|
||
|
||
def parse_plain_list(text: str, source_name: str) -> list[dict]:
|
||
"""Parse camera URLs from a plain-text or markdown list.
|
||
|
||
Accepted line shapes:
|
||
* `url[|lat,lon|vendor|location_name]` (strict form)
|
||
* markdown bullets / bare lines that merely CONTAIN a URL
|
||
(e.g. `* http://1.2.3.4/mjpg/video.mjpg`) — coords/vendor unknown.
|
||
"""
|
||
cams = []
|
||
url_re = re.compile(r'(?:https?|rtsp)://[^\s\)\]>"\']+', re.I)
|
||
for line in text.splitlines():
|
||
line = line.strip()
|
||
if not line or line.startswith("#"):
|
||
continue
|
||
m = url_re.search(line)
|
||
if not m:
|
||
continue
|
||
# Reuse strict parsing when the URL is pipe-delimited with metadata;
|
||
# otherwise take just the URL and leave metadata empty.
|
||
if "|" in line:
|
||
parts = [p.strip() for p in line.split("|")]
|
||
url = parts[0]
|
||
lat = lon = vendor = loc = None
|
||
if len(parts) > 1 and "," in parts[1]:
|
||
try:
|
||
la, lo = parts[1].split(",", 1)
|
||
lat, lon = float(la), float(lo)
|
||
except ValueError:
|
||
pass
|
||
if len(parts) > 2 and parts[2]:
|
||
vendor = parts[2]
|
||
if len(parts) > 3 and parts[3]:
|
||
loc = parts[3]
|
||
else:
|
||
url = m.group(0)
|
||
lat = lon = vendor = loc = None
|
||
cams.append({
|
||
"source_url": url,
|
||
"snapshot_url": url,
|
||
"discovery_source": source_name,
|
||
"location_lat": lat, "location_lon": lon,
|
||
"location_name": loc, "vendor": vendor,
|
||
})
|
||
return cams
|
||
|
||
|
||
def parse_alertwest_json(text: str, source_name: str) -> list[dict]:
|
||
"""Parse ALERTWest / ALERTCalifornia getCameraDataByLoc JSON.
|
||
|
||
Official public JPEG stills (wildfire, DOT, FAA). Skip private + offline.
|
||
Image URL formula from their docs:
|
||
https://img.cdn.prod.alertwest.com/data/img/{cid}/{yyyy}/{mm}/{dd}/{img}
|
||
with yyyy/mm/dd taken from the epoch embedded in the filename.
|
||
"""
|
||
try:
|
||
payload = json.loads(text)
|
||
except (json.JSONDecodeError, ValueError):
|
||
return []
|
||
data = (payload or {}).get("data") or {}
|
||
locs = {loc.get("id"): loc for loc in (data.get("locs") or {}).get("data") or []}
|
||
cams_in = (data.get("cams") or {}).get("data") or []
|
||
epoch_re = re.compile(r"_(\d{10,})_\d+")
|
||
out: list[dict] = []
|
||
for cam in cams_in:
|
||
if cam.get("pv") not in (0, "0", None):
|
||
continue
|
||
if cam.get("off") not in (0, "0", None):
|
||
continue
|
||
img = cam.get("img") or ""
|
||
cid = cam.get("id")
|
||
if not img or cid is None:
|
||
continue
|
||
m = epoch_re.search(img)
|
||
if not m:
|
||
continue
|
||
try:
|
||
dt = datetime.fromtimestamp(int(m.group(1)), timezone.utc)
|
||
except (OSError, ValueError, OverflowError):
|
||
continue
|
||
snap = (
|
||
f"https://img.cdn.prod.alertwest.com/data/img/"
|
||
f"{cid}/{dt:%Y}/{dt:%m}/{dt:%d}/{img}"
|
||
)
|
||
# Stable identity (the JPEG filename changes every capture).
|
||
ident = f"https://img.cdn.prod.alertwest.com/cam/{cid}"
|
||
loc = locs.get(cam.get("lid")) or {}
|
||
try:
|
||
lat = float(loc["lat"]) if loc.get("lat") is not None else None
|
||
lon = float(loc["lon"]) if loc.get("lon") is not None else None
|
||
except (TypeError, ValueError):
|
||
lat = lon = None
|
||
bits = [cam.get("cn"), cam.get("co"), loc.get("st") or cam.get("st")]
|
||
name = ", ".join(str(b) for b in bits if b)
|
||
out.append({
|
||
"source_url": ident,
|
||
"snapshot_url": snap,
|
||
"discovery_source": source_name,
|
||
"location_lat": lat,
|
||
"location_lon": lon,
|
||
"location_name": name or None,
|
||
"vendor": cam.get("pr") or cam.get("cc"),
|
||
"device_type": cam.get("cc") or "cctv",
|
||
})
|
||
return out
|
||
|
||
|
||
def parse_caltrans_json(text: str, source_name: str) -> list[dict]:
|
||
"""Parse Caltrans CWWP2 cctvStatus JSON (districts 1–12, same schema).
|
||
|
||
Skip cameras that are not in service. Store the JPEG still as snapshot_url
|
||
(map thumbs) and the HLS playlist as source_url when present. Never RTSP.
|
||
"""
|
||
try:
|
||
payload = json.loads(text)
|
||
except (json.JSONDecodeError, ValueError):
|
||
return []
|
||
rows = payload.get("data") if isinstance(payload, dict) else None
|
||
if not isinstance(rows, list):
|
||
return []
|
||
out: list[dict] = []
|
||
for row in rows:
|
||
cctv = (row or {}).get("cctv") or row or {}
|
||
loc = cctv.get("location") or {}
|
||
in_service = str(cctv.get("inService") or "").strip().lower()
|
||
if in_service not in ("true", "1", "yes"):
|
||
continue
|
||
img = ((cctv.get("imageData") or {}).get("static") or {})
|
||
jpeg = (img.get("currentImageURL") or img.get("currentImageUrl") or "").strip()
|
||
hls = str((cctv.get("imageData") or {}).get("streamingVideoURL") or "").strip()
|
||
if hls.lower().startswith("rtsp://") or jpeg.lower().startswith("rtsp://"):
|
||
continue
|
||
if not jpeg and not hls:
|
||
continue
|
||
try:
|
||
lat = float(loc["latitude"]) if loc.get("latitude") not in (None, "") else None
|
||
lon = float(loc["longitude"]) if loc.get("longitude") not in (None, "") else None
|
||
except (TypeError, ValueError):
|
||
lat = lon = None
|
||
bits = [
|
||
loc.get("locationName"),
|
||
loc.get("route") and f"SR-{loc.get('route')}",
|
||
loc.get("nearbyPlace") or loc.get("county"),
|
||
]
|
||
name = ", ".join(str(b) for b in bits if b)
|
||
ident = hls or jpeg
|
||
dtype = "hls" if hls else "http"
|
||
out.append({
|
||
"source_url": ident,
|
||
"snapshot_url": jpeg or hls,
|
||
"discovery_source": "caltrans",
|
||
"location_lat": lat,
|
||
"location_lon": lon,
|
||
"location_name": name or None,
|
||
"vendor": "Caltrans",
|
||
"device_type": dtype,
|
||
})
|
||
return out
|
||
|
||
|
||
# MDOT MiDrive field extractors (fields carry rendered HTML).
|
||
_MDOT_LAT_RE = re.compile(r"lat=(-?\d+(?:\.\d+)?)", re.I)
|
||
_MDOT_LON_RE = re.compile(r"lon=(-?\d+(?:\.\d+)?)", re.I)
|
||
_MDOT_ID_RE = re.compile(r"[?&]id=(\d+)", re.I)
|
||
_MDOT_IMG_RE = re.compile(r'<img[^>]+src=["\']([^"\']+)["\']', re.I)
|
||
|
||
# Michigan bbox (docs/osiris-ideas.md §3.2): lat 41.6–48.3, lon -90.5–-82.1.
|
||
MDOT_LAT_RANGE = (41.6, 48.3)
|
||
MDOT_LON_RANGE = (-90.5, -82.1)
|
||
|
||
|
||
def parse_mdot_json(text: str, source_name: str) -> list[dict]:
|
||
"""Parse MDOT MiDrive `camera/list` JSON (fields carry rendered HTML).
|
||
|
||
Coordinates and the stable id live in the `county` field's map link
|
||
(`/MiDrive/map?...lat=&lon=&id=`); the `image` field carries an `<img>`
|
||
whose src is the JPEG still. Out-of-bbox and coord-less rows are dropped.
|
||
"""
|
||
try:
|
||
payload = json.loads(text)
|
||
except (json.JSONDecodeError, ValueError):
|
||
return []
|
||
if not isinstance(payload, list):
|
||
return []
|
||
out: list[dict] = []
|
||
for row in payload:
|
||
if not isinstance(row, dict):
|
||
continue
|
||
county_html = row.get("county") or ""
|
||
m_lat = _MDOT_LAT_RE.search(county_html)
|
||
m_lon = _MDOT_LON_RE.search(county_html)
|
||
m_id = _MDOT_ID_RE.search(county_html)
|
||
if not (m_lat and m_lon and m_id):
|
||
continue # missing coordinates / stable id → drop
|
||
try:
|
||
lat = float(m_lat.group(1))
|
||
lon = float(m_lon.group(1))
|
||
except ValueError:
|
||
continue
|
||
if not (MDOT_LAT_RANGE[0] <= lat <= MDOT_LAT_RANGE[1]
|
||
and MDOT_LON_RANGE[0] <= lon <= MDOT_LON_RANGE[1]):
|
||
continue # out of Michigan bbox → drop
|
||
img_m = _MDOT_IMG_RE.search(row.get("image") or "")
|
||
if not img_m:
|
||
continue
|
||
snap = img_m.group(1).strip()
|
||
low = snap.lower()
|
||
if not (low.startswith("http://") or low.startswith("https://")):
|
||
continue
|
||
if low.startswith("rtsp"):
|
||
continue
|
||
cam_id = m_id.group(1)
|
||
route = (row.get("route") or "").strip()
|
||
loc = (row.get("location") or "").strip().lstrip("@").strip()
|
||
county_name = county_html.split("<a", 1)[0].strip()
|
||
bits = [
|
||
f"{route} @ {loc}" if (route and loc) else (route or loc or None),
|
||
county_name or None,
|
||
]
|
||
name = ", ".join(b for b in bits if b) or None
|
||
out.append({
|
||
"source_url": f"https://mdotjboss.state.mi.us/MiDrive/camera/{cam_id}",
|
||
"snapshot_url": snap,
|
||
"discovery_source": "mdot",
|
||
"location_lat": lat,
|
||
"location_lon": lon,
|
||
"location_name": name,
|
||
"vendor": "MDOT",
|
||
"device_type": "http",
|
||
})
|
||
return out
|
||
|
||
|
||
def parse_live_streams_geojson(text: str, source_name: str) -> list[dict]:
|
||
"""Parse willytop8/Live-Environment-Streams GeoJSON.
|
||
|
||
Only direct, playable URLs: HLS, YouTube, HTTP stills. Skip html_page
|
||
feeds that need token extraction or a headless browser.
|
||
"""
|
||
try:
|
||
payload = json.loads(text)
|
||
except (json.JSONDecodeError, ValueError):
|
||
return []
|
||
usable = {"hls", "youtube", "http_image"}
|
||
out: list[dict] = []
|
||
for feat in payload.get("features") or []:
|
||
props = feat.get("properties") or {}
|
||
ut = (props.get("url_type") or "").lower()
|
||
if ut not in usable:
|
||
continue
|
||
if props.get("source_url_requires"):
|
||
continue
|
||
url = (props.get("url") or "").strip()
|
||
if not url:
|
||
continue
|
||
coords = (feat.get("geometry") or {}).get("coordinates") or []
|
||
lon = lat = None
|
||
if len(coords) >= 2:
|
||
try:
|
||
lon, lat = float(coords[0]), float(coords[1])
|
||
except (TypeError, ValueError):
|
||
lon = lat = None
|
||
dtype = "ip-cam" if ut == "http_image" else ut
|
||
out.append({
|
||
"source_url": url,
|
||
"snapshot_url": url,
|
||
"discovery_source": source_name,
|
||
"location_lat": lat,
|
||
"location_lon": lon,
|
||
"location_name": props.get("display_name") or props.get("name"),
|
||
"vendor": props.get("source_family"),
|
||
"device_type": dtype,
|
||
})
|
||
return out
|
||
|
||
|
||
def parse_directory_html(html: str, base_url: str, source_name: str) -> list[dict]:
|
||
"""Generic Insecam-style directory parser."""
|
||
cams = []
|
||
imgs = _IMG_RE.findall(html)
|
||
for m in _LATLON_RE.finditer(html):
|
||
lat, lon = float(m.group(1)), float(m.group(2))
|
||
# nearest img tag after the match → best-effort snapshot endpoint
|
||
tail = html[m.end(): m.end() + 800]
|
||
img_m = _IMG_RE.search(tail)
|
||
snap = img_m.group(1) if img_m else (imgs[0] if imgs else None)
|
||
if snap and snap.startswith("/"):
|
||
from urllib.parse import urljoin
|
||
snap = urljoin(base_url, snap)
|
||
cams.append({
|
||
"source_url": base_url,
|
||
"snapshot_url": snap,
|
||
"discovery_source": source_name,
|
||
"location_lat": lat, "location_lon": lon,
|
||
"location_name": None, "vendor": None,
|
||
})
|
||
return cams
|
||
|
||
|
||
# ── Batch IP geolocation (ip-api.com — free, 100 IPs per batch call) ───────
|
||
|
||
IPAPI_BATCH_URL = "http://ip-api.com/batch"
|
||
IPAPI_BATCH_SIZE = 100
|
||
|
||
|
||
async def geolocate_ips(ips: list[str]) -> dict[str, tuple[float, float]]:
|
||
"""Resolve public IPs → (lat, lon). Unresolvable IPs are simply absent."""
|
||
out: dict[str, tuple[float, float]] = {}
|
||
async with httpx.AsyncClient(timeout=15) as c:
|
||
for i in range(0, len(ips), IPAPI_BATCH_SIZE):
|
||
chunk = ips[i: i + IPAPI_BATCH_SIZE]
|
||
try:
|
||
fields = "status,country,city,lat,lon,query"
|
||
r = await c.post(IPAPI_BATCH_URL, json=[
|
||
{"query": ip, "fields": fields} for ip in chunk
|
||
])
|
||
r.raise_for_status()
|
||
for row in r.json():
|
||
if (row.get("status") == "success"
|
||
and row.get("lat") is not None):
|
||
out[row["query"]] = (
|
||
float(row["lat"]), float(row["lon"]))
|
||
except Exception: # noqa: BLE001
|
||
logger.warning("ip-api batch failed", exc_info=True)
|
||
if i + IPAPI_BATCH_SIZE < len(ips):
|
||
await asyncio.sleep(2) # stay well under the free rate limit
|
||
return out
|
||
|
||
|
||
async def scrape_source(client: RateLimitedClient, geo: Geocoder,
|
||
src_url: str) -> list[dict]:
|
||
"""Fetch one source and return normalized camera dicts (public only)."""
|
||
try:
|
||
resp = await client.get(src_url)
|
||
resp.raise_for_status()
|
||
except Exception: # noqa: BLE001
|
||
logger.exception("failed to fetch source %s", src_url)
|
||
return []
|
||
|
||
name = urlparse(src_url).netloc
|
||
ctype = resp.headers.get("content-type", "")
|
||
body = resp.text
|
||
if "cwwp2.dot.ca.gov" in src_url or "cctvStatus" in src_url:
|
||
cams = parse_caltrans_json(body, name)
|
||
elif "mdotjboss.state.mi.us" in src_url or "/MiDrive/camera/list" in src_url:
|
||
cams = parse_mdot_json(body, name)
|
||
elif ("getCameraDataByLoc" in src_url
|
||
or ("json" in ctype and '"locs"' in body[:4000] and '"cams"' in body[:8000])):
|
||
cams = parse_alertwest_json(body, name)
|
||
elif (src_url.endswith(".geojson") or src_url.endswith("/streams.geojson")
|
||
or '"FeatureCollection"' in body[:400]):
|
||
cams = parse_live_streams_geojson(body, name)
|
||
elif "html" in ctype:
|
||
cams = parse_directory_html(body, str(resp.url), name)
|
||
else:
|
||
cams = parse_plain_list(body, name)
|
||
|
||
out: list[dict] = []
|
||
seen_in_batch: set[str] = set()
|
||
# Batch-geolocate coordinate-less camera hosts via ip-api (one call per
|
||
# 100 IPs) instead of dropping them — raw-IP lists have no embedded coords.
|
||
need_geo: dict[str, str] = {}
|
||
for cam in cams[:CAMERA_MAX_PER_SOURCE]:
|
||
url = cam["source_url"]
|
||
h = url_hash(url)
|
||
if h in seen_in_batch:
|
||
continue
|
||
seen_in_batch.add(h)
|
||
if not is_public_url(url):
|
||
continue
|
||
if (cam["location_lat"] is None and cam["location_name"] is None):
|
||
host = urlparse(url).hostname or ""
|
||
try:
|
||
ipaddress.ip_address(host)
|
||
need_geo.setdefault(host, url)
|
||
except ValueError:
|
||
pass # hostname-only: Nominatim fallback below handles it
|
||
geo_by_ip = await geolocate_ips(list(need_geo)) if need_geo else {}
|
||
|
||
for cam in cams:
|
||
if url_hash(cam["source_url"]) not in seen_in_batch:
|
||
continue # didn't survive dedupe/public-range filtering above
|
||
if cam["location_lat"] is None and cam["location_name"]:
|
||
cam["location_lat"], cam["location_lon"] = await geo.geocode(
|
||
cam["location_name"])
|
||
if (cam["location_lat"] is None and cam["location_lon"] is None):
|
||
host = urlparse(cam["source_url"]).hostname or ""
|
||
coords = geo_by_ip.get(host)
|
||
if coords is None:
|
||
continue # map display requires coordinates
|
||
cam["location_lat"], cam["location_lon"] = coords
|
||
cam["location_name"] = f"{host} (IP-geo)"
|
||
cam.setdefault("vendor", None)
|
||
if not cam.get("vendor"):
|
||
cam["vendor"] = _vendor_from_url(cam["snapshot_url"] or cam["source_url"])
|
||
if not cam.get("device_type"):
|
||
cam["device_type"] = (
|
||
"rtsp" if (cam["snapshot_url"] or "").startswith("rtsp") else "ip-cam"
|
||
)
|
||
cam.setdefault("raw", {"discovered_via": src_url})
|
||
out.append(cam)
|
||
logger.info("source %s yielded %d public cameras", src_url, len(out))
|
||
return out
|
||
|
||
|
||
# ── Persistence ────────────────────────────────────────────────────────────
|
||
|
||
async def upsert_cameras(cams: list[dict]) -> int:
|
||
"""Insert-or-update cameras keyed by url_hash. Returns rows written."""
|
||
now = datetime.now(timezone.utc)
|
||
written = 0
|
||
async with async_session() as session:
|
||
for cam in cams:
|
||
values = {
|
||
"url_hash": url_hash(cam["source_url"]),
|
||
"source_url": cam["source_url"],
|
||
"snapshot_url": cam.get("snapshot_url"),
|
||
"discovery_source": cam["discovery_source"],
|
||
"location_lat": cam.get("location_lat"),
|
||
"location_lon": cam.get("location_lon"),
|
||
"location_name": cam.get("location_name"),
|
||
"vendor": cam.get("vendor"),
|
||
"device_type": cam.get("device_type"),
|
||
"last_seen": now,
|
||
"raw": cam.get("raw"),
|
||
}
|
||
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
||
stmt = pg_insert(cameras).values(**values)
|
||
stmt = stmt.on_conflict_do_update(
|
||
index_elements=[cameras.c.url_hash],
|
||
set_={
|
||
"last_seen": now,
|
||
"snapshot_url": values["snapshot_url"],
|
||
"location_lat": values["location_lat"],
|
||
"location_lon": values["location_lon"],
|
||
"location_name": values["location_name"],
|
||
"vendor": values["vendor"],
|
||
"device_type": values["device_type"],
|
||
"raw": values["raw"],
|
||
},
|
||
)
|
||
await session.execute(stmt)
|
||
written += 1
|
||
await session.commit()
|
||
return written
|
||
|
||
|
||
# ── Cycle driver ───────────────────────────────────────────────────────────
|
||
|
||
async def run_cycle() -> int:
|
||
client = RateLimitedClient()
|
||
geo = Geocoder()
|
||
total = 0
|
||
try:
|
||
results = await asyncio.gather(
|
||
*(scrape_source(client, geo, s) for s in CAMERA_SOURCE_URLS),
|
||
return_exceptions=True,
|
||
)
|
||
all_cams: list[dict] = []
|
||
for r in results:
|
||
if isinstance(r, BaseException):
|
||
logger.error("scrape task failed: %s", r)
|
||
else:
|
||
all_cams.extend(r)
|
||
if all_cams:
|
||
total = await upsert_cameras(all_cams)
|
||
finally:
|
||
await client.aclose()
|
||
logger.info("camera cycle complete: %d cameras stored", total)
|
||
return total
|