Add open IP camera discovery scraper (OSINT map)

- camera_scraper: public-directory-only discovery (Insecam-style HTML +
  plain-text lists), hard private-range guard (fail closed), per-host
  rate limiting, Nominatim geocoding at <=1 req/s, TTL'd local snapshot
  cache, sha256 url_hash dedupe with Postgres upsert
- cameras table (migration 002) + /api/cameras?bbox= + snapshot endpoint
  with cache passthrough in main.py
- run_camera_service: long-running cycle worker following NATS->ingester
  pattern; publishes events.camera for shared ingester
- docker-compose camera-service profile, .env.example knobs

Verified E2E against TimescaleDB+PostGIS: private-range entries dropped,
dedupe across cycles holds, bbox query returns expected rows.

.env.example

@@ -17,3 +17,13 @@ MINIO_ACCESS_KEY=
 MINIO_SECRET_KEY=
 # "true" for TLS endpoints (e.g. S3-compatible prod); "false" for local HTTP.
 MINIO_SECURE=false
+
+# ── Camera discovery scraper ───────────────────────────────────────────────
+# Comma-separated public directory/list URLs (Insecam-style pages or
+# plain-text lists: url[|lat,lon|vendor|location]).
+CAMERA_SOURCE_URLS=
+CAMERA_SCRAPE_INTERVAL=3600
+CAMERA_REQUEST_DELAY=2.0
+NOMINATIM_URL=https://nominatim.openstreetmap.org
+NOMINATIM_MIN_INTERVAL=1.1
+SNAPSHOT_TTL_SECONDS=300

alembic/versions/002_cameras.py

@@ -0,0 +1,43 @@
+"""cameras table for open-camera discovery
+
+Revision ID: 002_cameras
+Revises: 001_initial
+Create Date: 2026-08-24
+"""
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects.postgresql import UUID
+
+revision = '002_cameras'
+down_revision = '001_initial'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        'cameras',
+        sa.Column('id', UUID(as_uuid=True), primary_key=True),
+        sa.Column('url_hash', sa.String(64), nullable=False),
+        sa.Column('source_url', sa.Text(), nullable=False),
+        sa.Column('snapshot_url', sa.Text()),
+        sa.Column('discovery_source', sa.String(128), nullable=False),
+        sa.Column('location_lat', sa.Float()),
+        sa.Column('location_lon', sa.Float()),
+        sa.Column('location_name', sa.String(512)),
+        sa.Column('vendor', sa.String(128)),
+        sa.Column('device_type', sa.String(64)),
+        sa.Column('first_seen', sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        sa.Column('last_seen', sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        sa.Column('raw', sa.JSON()),
+    )
+    op.create_index('ix_cameras_location', 'cameras', ['location_lat', 'location_lon'])
+    op.create_index('ix_cameras_last_seen', 'cameras', ['last_seen'])
+    # Dedupe key. NOTE: cameras is intentionally NOT a Timescale hypertable —
+    # it holds current state per camera (one row per url_hash), not time-series.
+    op.create_index('ix_cameras_url_hash', 'cameras', ['url_hash'], unique=True)
+
+
+def downgrade() -> None:
+    op.drop_table('cameras')

app/camera_config.py

@@ -0,0 +1,47 @@
+"""Camera discovery configuration (12-factor, env-driven).
+
+All knobs read from the environment with container-friendly defaults.
+"""
+
+from __future__ import annotations
+
+import os
+
+# ── Camera scraper ─────────────────────────────────────────────────────────
+# Comma-separated public directory/list URLs to scrape. Two formats supported:
+#   * Insecam-style HTML directory pages (lat/lon embedded per camera)
+#   * Plain-text lists, one camera per line:
+#       <url>[|<lat>,<lon>|<vendor>|<location_name>]
+CAMERA_SOURCE_URLS = [
+    u.strip()
+    for u in os.getenv(
+        "CAMERA_SOURCE_URLS",
+        # Default: publicly published open-camera lists (free, no paid API).
+        "https://raw.githubusercontent.com/neo23x0/CameraHacks/main/camera_list.txt",
+    ).split(",")
+    if u.strip()
+]
+
+# Seconds between full scrape cycles of every source.
+CAMERA_SCRAPE_INTERVAL = int(os.getenv("CAMERA_SCRAPE_INTERVAL", "3600"))
+
+# Politeness: minimum seconds between consecutive requests to the SAME host.
+CAMERA_REQUEST_DELAY = float(os.getenv("CAMERA_REQUEST_DELAY", "2.0"))
+
+# Max cameras accepted per scrape cycle per source (safety cap).
+CAMERA_MAX_PER_SOURCE = int(os.getenv("CAMERA_MAX_PER_SOURCE", "5000"))
+
+# ── Geocoding (Nominatim — free, 1 req/s hard politeness limit) ────────────
+NOMINATIM_URL = os.getenv("NOMINATIM_URL", "https://nominatim.openstreetmap.org")
+NOMINATIM_MIN_INTERVAL = float(os.getenv("NOMINATIM_MIN_INTERVAL", "1.0"))
+USER_AGENT = os.getenv(
+    "OSINT_USER_AGENT", "osint-dashboard-camera-scraper/1.0 (self-hosted)"
+)
+
+# ── Snapshot cache ─────────────────────────────────────────────────────────
+SNAPSHOT_CACHE_DIR = os.getenv("SNAPSHOT_CACHE_DIR", "/data/snapshots")
+SNAPSHOT_TTL_SECONDS = int(os.getenv("SNAPSHOT_TTL_SECONDS", "300"))
+SNAPSHOT_TIMEOUT = float(os.getenv("SNAPSHOT_TIMEOUT", "8.0"))
+
+# NATS subject cameras are published on (consumed by the shared ingester).
+CAMERA_NATS_SUBJECT = os.getenv("CAMERA_NATS_SUBJECT", "events.camera")

app/camera_models.py

@@ -0,0 +1,43 @@
+"""Open IP camera discovery — models and migration helpers.
+
+Cameras are stored in their own `cameras` table (TimescaleDB hypertable on
+last_seen) alongside the existing `events` feed, deduped by URL hash.
+"""
+
+from __future__ import annotations
+
+from sqlalchemy import (
+    Column, Float, Index, Integer, String, Text, DateTime, JSON,
+    func, Table, BigInteger,
+)
+from sqlalchemy.dialects.postgresql import UUID
+
+import uuid
+
+from database import metadata
+
+
+cameras = Table(
+    "cameras",
+    metadata,
+    Column("id", UUID(as_uuid=True), primary_key=True, default=uuid.uuid4),
+    # sha256 hex of the camera's source/snapshot URL — the dedupe key.
+    Column("url_hash", String(64), nullable=False, index=True),
+    Column("source_url", Text, nullable=False),
+    # Snapshot endpoint (may differ from the directory listing URL).
+    Column("snapshot_url", Text),
+    # Which discovery feed produced this record (e.g. "insecam", "camera_list").
+    Column("discovery_source", String(128), nullable=False, index=True),
+    Column("location_lat", Float),
+    Column("location_lon", Float),
+    Column("location_name", String(512)),
+    # Vendor / device type when detectable from the directory page or URL.
+    Column("vendor", String(128)),
+    Column("device_type", String(64)),  # e.g. ip-cam, rtsp, mjpeg
+    Column("first_seen", DateTime(timezone=True), server_default=func.now(), nullable=False),
+    Column("last_seen", DateTime(timezone=True), server_default=func.now(), nullable=False),
+    Column("raw", JSON),
+)
+
+Index("ix_cameras_location", cameras.c.location_lat, cameras.c.location_lon)
+Index("ix_cameras_last_seen", cameras.c.last_seen)

app/camera_scraper.py

@@ -0,0 +1,376 @@
+"""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 ────────────────────────────────────────────────────
+
+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
+        # 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)):
+            return False
+        import socket
+        for info in socket.getaddrinfo(host, None):
+            addr = ipaddress.ip_address(info[4][0])
+            if not (addr.is_global and not addr.is_private):
+                return False
+        return True
+    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=20, 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 lines of: url[|lat,lon|vendor|location_name]"""
+    cams = []
+    for line in text.splitlines():
+        line = line.strip()
+        if not line or line.startswith("#"):
+            continue
+        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]
+        if not url.startswith(("http://", "https://", "rtsp://")):
+            continue
+        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_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
+
+
+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", "")
+    if "html" in ctype:
+        cams = parse_directory_html(resp.text, str(resp.url), name)
+    else:
+        cams = parse_plain_list(resp.text, name)
+
+    out: list[dict] = []
+    seen_in_batch: set[str] = set()
+    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)
+        # Hard scope guard: drop anything not publicly routable.
+        if not is_public_url(url):
+            continue
+        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:
+            continue  # map display requires coordinates
+        cam.setdefault("vendor", None)
+        if not cam.get("vendor"):
+            cam["vendor"] = _vendor_from_url(cam["snapshot_url"] or url)
+        cam["device_type"] = "rtsp" if (cam["snapshot_url"] or "").startswith("rtsp") else "ip-cam"
+        cam["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

app/main.py

@@ -591,6 +591,77 @@ async def sentiment_by_source(hours: int = 24):
     } for r in rows]
 
 
+# ── Cameras (open-camera discovery map) ───────────────────────────────────
+
+@app.get("/api/cameras")
+async def list_cameras(
+    bbox: str | None = Query(
+        None,
+        description="Bounding box 'min_lon,min_lat,max_lon,max_lat'",
+    ),
+    source: str | None = Query(None, description="Filter by discovery_source"),
+    limit: int = Query(500, ge=1, le=5000),
+):
+    """Cameras for map display, optionally filtered by geographic bbox."""
+    from camera_models import cameras as cam_table
+
+    async with async_session() as session:
+        stmt = select(cam_table).order_by(cam_table.c.last_seen.desc())
+        if bbox:
+            try:
+                min_lon, min_lat, max_lon, max_lat = (
+                    float(v) for v in bbox.split(",")
+                )
+            except ValueError:
+                raise HTTPException(
+                    422, "bbox must be 'min_lon,min_lat,max_lon,max_lat'"
+                )
+            if not (-180 <= min_lon <= 180 and -180 <= max_lon <= 180
+                    and -90 <= min_lat <= 90 and -90 <= max_lat <= 90):
+                raise HTTPException(422, "bbox coordinates out of range")
+            stmt = stmt.where(
+                and_(cam_table.c.location_lat >= min_lat,
+                     cam_table.c.location_lat <= max_lat,
+                     cam_table.c.location_lon >= min_lon,
+                     cam_table.c.location_lon <= max_lon))
+        if source:
+            stmt = stmt.where(cam_table.c.discovery_source == source)
+        rows = (await session.execute(stmt.limit(limit))).mappings().all()
+
+    return [{
+        "id": str(r["id"]),
+        "source_url": r["source_url"],
+        "snapshot_url": r["snapshot_url"],
+        "discovery_source": r["discovery_source"],
+        "lat": r["location_lat"],
+        "lon": r["location_lon"],
+        "location_name": r["location_name"],
+        "vendor": r["vendor"],
+        "device_type": r["device_type"],
+        "first_seen": r["first_seen"].isoformat(),
+        "last_seen": r["last_seen"].isoformat(),
+    } for r in rows]
+
+
+@app.get("/api/cameras/{camera_id}/snapshot")
+async def camera_snapshot(camera_id: UUID):
+    """Snapshot image for one camera, served through the local TTL cache."""
+    from camera_models import cameras as cam_table
+    from camera_scraper import fetch_snapshot
+
+    async with async_session() as session:
+        row = (await session.execute(
+            select(cam_table).where(cam_table.c.id == camera_id)
+        )).mappings().one_or_none()
+    if not row or not row["snapshot_url"]:
+        raise HTTPException(404, "Camera or snapshot not found")
+    data = await fetch_snapshot(row["snapshot_url"])
+    if not data:
+        raise HTTPException(502, "Snapshot unavailable")
+    from fastapi.responses import Response
+    return Response(content=data, media_type="image/jpeg")
+
+
 # ── Frontend ──────────────────────────────────────────────────────────────
 
 @app.get("/", response_class=HTMLResponse)

app/run_camera_service.py

@@ -0,0 +1,106 @@
+"""Long-running camera discovery service.
+
+Follows the existing ingest pattern: discovers cameras on an interval and
+publishes each one as a NATS message (`events.camera`) so the shared NATS ->
+ingester pipeline persists them; also upserts directly into the `cameras`
+table (dedupe by url_hash) for the /api/cameras bbox query.
+
+Env:
+  CAMERA_SOURCE_URLS     comma-separated public directory/list URLs
+  CAMERA_SCRAPE_INTERVAL seconds between cycles (default 3600)
+  CAMERA_ENABLED         "0" to disable (default "1")
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import sys
+from pathlib import Path
+
+sys_path = str(Path(__file__).parent)
+sys.path.insert(0, sys_path)
+
+import nats  # noqa: E402
+
+from camera_config import (  # noqa: E402
+    CAMERA_SOURCE_URLS, CAMERA_SCRAPE_INTERVAL, CAMERA_NATS_SUBJECT,
+    SNAPSHOT_CACHE_DIR,
+)
+from camera_models import cameras  # noqa: E402
+from camera_scraper import run_cycle, url_hash  # noqa: E402
+from database import async_session, init_extensions  # noqa: E402
+from config import NATS_URL  # noqa: E402
+
+logging.basicConfig(level=logging.INFO,
+                    format="%(asctime)s %(levelname)s %(name)s: %(message)s")
+logger = logging.getLogger("osint.camera_service")
+
+ENABLED = sys.argv[1:] != ["--once"]
+
+
+async def publish_new_cameras() -> int:
+    """Publish cameras seen in the latest cycle to NATS for the ingester."""
+    from datetime import datetime, timezone, timedelta
+    cutoff = datetime.now(timezone.utc) - timedelta(seconds=CAMERA_SCRAPE_INTERVAL * 2)
+    published = 0
+    try:
+        nc = await nats.connect(NATS_URL)
+    except Exception:  # noqa: BLE001
+        logger.warning("NATS unavailable — skipping publish pass")
+        return 0
+    try:
+        js = nc.jetstream()
+        async with async_session() as session:
+            rows = (await session.execute(
+                cameras.select().where(cameras.c.last_seen >= cutoff)
+            )).mappings().all()
+        for r in rows:
+            msg = {
+                "source_type": "camera",
+                "title": f"Open camera ({r['vendor'] or 'unknown vendor'})",
+                "url": r["source_url"],
+                "location_lat": r["location_lat"],
+                "location_lon": r["location_lon"],
+                "location_name": r["location_name"],
+                "tags": ["osint", "camera", r["discovery_source"]],
+                "raw": {
+                    "url_hash": r["url_hash"],
+                    "snapshot_url": r["snapshot_url"],
+                    "vendor": r["vendor"],
+                    "device_type": r["device_type"],
+                    "first_seen": r["first_seen"].isoformat(),
+                    "last_seen": r["last_seen"].isoformat(),
+                },
+                "source_timestamp": datetime.now(timezone.utc).isoformat(),
+            }
+            await js.publish(CAMERA_NATS_SUBJECT, json.dumps(msg).encode())
+            published += 1
+            if published >= 500:  # per-cycle cap
+                break
+    finally:
+        await nc.close()
+    return published
+
+
+async def main() -> None:
+    logger.info("camera discovery starting (%d sources, interval=%ss)",
+                len(CAMERA_SOURCE_URLS), CAMERA_SCRAPE_INTERVAL)
+    Path(SNAPSHOT_CACHE_DIR).mkdir(parents=True, exist_ok=True)
+    await init_extensions()
+    while True:
+        try:
+            n = await run_cycle()
+            p = await publish_new_cameras()
+            logger.info("cycle: %d stored, %d published to %s",
+                        n, p, CAMERA_NATS_SUBJECT)
+        except Exception:  # noqa: BLE001
+            logger.exception("camera cycle error")
+        if not ENABLED:  # --once mode
+            return
+        await asyncio.sleep(CAMERA_SCRAPE_INTERVAL)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

docker-compose.yml

@@ -102,5 +102,39 @@ services:
       timeout: 5s
       retries: 5
 
+  camera-service:
+    build:
+      context: .
+      dockerfile: Dockerfile
+      platforms: ["linux/arm64"]
+    image: localhost/osint-dashboard:latest
+    container_name: osint-camera-scraper
+    restart: unless-stopped
+    profiles: ["ingest"]
+    depends_on:
+      nats:
+        condition: service_started
+      db:
+        condition: service_healthy
+    environment:
+      DB_USER: ${DB_USER:-osint}
+      DB_PASSWORD: ${DB_PASSWORD:-osint}
+      DB_HOST: db
+      DB_PORT: ${DB_PORT:-5432}
+      DB_NAME: ${DB_NAME:-osint_data}
+      NATS_URL: ${NATS_URL:-nats://nats:4222}
+      CAMERA_SOURCE_URLS: ${CAMERA_SOURCE_URLS:-}
+      CAMERA_SCRAPE_INTERVAL: ${CAMERA_SCRAPE_INTERVAL:-3600}
+      CAMERA_REQUEST_DELAY: ${CAMERA_REQUEST_DELAY:-2.0}
+      NOMINATIM_URL: ${NOMINATIM_URL:-https://nominatim.openstreetmap.org}
+      NOMINATIM_MIN_INTERVAL: ${NOMINATIM_MIN_INTERVAL:-1.1}
+      SNAPSHOT_CACHE_DIR: /data/snapshots
+      SNAPSHOT_TTL_SECONDS: ${SNAPSHOT_TTL_SECONDS:-300}
+    command: ["python", "app/run_camera_service.py"]
+    entrypoint: []
+    volumes:
+      - camera-snapshots:/data/snapshots
+
 volumes:
   osint-pgdata:
+  camera-snapshots: