osint-dashboard/app/camera_scraper.py

"""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,
    UDOT_IBI_URL, UDOT_IBI_BASE, UDOT_IBI_PAGE_SIZE, UDOT_IBI_MAX_PAGES,
)
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 post(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.post(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


# ── UDOT IBI 511 ──────────────────────────────────────────────────────────
# Utah bbox (lat 36.9–42.1, lon -114.2–-108.9). WKT is `POINT (lng lat)`.
_UDOT_IBI_MIN_LAT, _UDOT_IBI_MAX_LAT = 36.9, 42.1
_UDOT_IBI_MIN_LON, _UDOT_IBI_MAX_LON = -114.2, -108.9
_UDOT_WKT_POINT_RE = re.compile(
    r"POINT\s*\(\s*(-?\d+(?:\.\d+)?)\s+(-?\d+(?:\.\d+)?)\s*\)", re.I,
)


def parse_udot_ibi_page(text: str, source_name: str = "udot") -> list[dict]:
    """Parse one UDOT IBI 511 DataTables camera page (`{"data": [...]}`).

    Skips rows whose first image is `blocked` or `disabled`, and drops any
    point outside the Utah bbox. The `/map/Cctv/{id}` URL is a stable identity
    (always serves the latest frame), so it is stored as both source_url and
    snapshot_url — we never scrape frames ourselves.
    """
    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:
        if not isinstance(row, dict):
            continue
        cam_id = row.get("id")
        images = row.get("images") or []
        if cam_id is None or not images:
            continue
        img = images[0] or {}
        if img.get("blocked") or img.get("disabled"):
            continue
        lon = lat = None
        try:
            wkt = (row.get("latLng") or {}).get("geography") or {}
            wkt = wkt.get("wellKnownText") or ""
            m = _UDOT_WKT_POINT_RE.match(str(wkt).strip())
            if m:
                lon, lat = float(m.group(1)), float(m.group(2))
        except (AttributeError, TypeError, ValueError):
            lon = lat = None
        if lat is None or lon is None:
            continue
        if not (_UDOT_IBI_MIN_LAT <= lat <= _UDOT_IBI_MAX_LAT
                and _UDOT_IBI_MIN_LON <= lon <= _UDOT_IBI_MAX_LON):
            continue
        snap = f"{UDOT_IBI_BASE}/map/Cctv/{cam_id}"
        roadway, direction, location = (
            row.get("roadway"), row.get("direction"), row.get("location"),
        )
        name = ", ".join(
            str(b) for b in (roadway, direction, location)
            if b and str(b).strip() and str(b).strip().lower() != "unknown"
        ) or None
        out.append({
            "source_url": snap,
            "snapshot_url": snap,
            "discovery_source": source_name,
            "location_lat": lat,
            "location_lon": lon,
            "location_name": name,
            "vendor": "UDOT",
            "device_type": "http",
            "raw": {
                "udot_id": cam_id,
                "agency": row.get("source"),
                "source_id": row.get("sourceId"),
                "roadway": roadway,
                "direction": direction,
            },
        })
    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 ("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


# ── UDOT IBI 511 paginated fetcher ────────────────────────────────────────

async def scrape_udot_ibi(client: RateLimitedClient) -> list[dict]:
    """Page through the UDOT IBI 511 DataTables endpoint and normalize.

    POSTs `start`/`length` form fields (server caps at 100 rows/page), walking
    pages until `recordsTotal` is exhausted or UDOT_IBI_MAX_PAGES is hit.
    """
    out: list[dict] = []
    seen: set[str] = set()
    start = 0
    for _ in range(UDOT_IBI_MAX_PAGES):
        try:
            resp = await client.post(
                UDOT_IBI_URL,
                data={
                    "start": str(start),
                    "length": str(UDOT_IBI_PAGE_SIZE),
                    "lang": "en-US",
                },
                headers={"X-Requested-With": "XMLHttpRequest"},
            )
            resp.raise_for_status()
            body = resp.text
        except Exception:  # noqa: BLE001
            logger.exception("failed to fetch UDOT IBI page start=%d", start)
            break
        try:
            payload = json.loads(body)
        except ValueError:
            logger.warning("UDOT IBI non-JSON response at start=%d", start)
            break
        total = int(payload.get("recordsTotal") or 0)
        rows = payload.get("data") or []
        if not isinstance(rows, list) or not rows:
            break
        for cam in parse_udot_ibi_page(body, "udot"):
            if cam["source_url"] in seen:
                continue
            seen.add(cam["source_url"])
            out.append(cam)
        if start + len(rows) >= total:
            break
        start += len(rows)
    logger.info("UDOT IBI yielded %d cameras", 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),
            scrape_udot_ibi(client),
            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