diff --git a/.env.example b/.env.example index f736246..dcfd964 100644 --- a/.env.example +++ b/.env.example @@ -18,6 +18,16 @@ 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 + # ── NASA FIRMS (active fire / hotspot ingest) ────────────────────────────── # MAP_KEY is FREE — get one at https://firms.modaps.eosdis.nasa.gov/api/map_key_info/ # (1-minute signup, no payment). Leave blank to keep fire ingest idle. @@ -33,10 +43,8 @@ FIRMS_INTERVAL=900 INGEST_FIRES=1 # ── API keys (managed from the dashboard UI) ────────────────────────────── -# Ingest-service keys such as FIRMS_MAP_KEY, GEMINI_API_KEY and TELEGRAM_TOKEN -# are stored in the Postgres `api_keys` table and managed from the dashboard's -# "Keys" tab (GET/POST/DELETE /api/keys/{name}). You do NOT need to put them in -# .env — the UI writes straight to the DB, and ingest services read them from -# there (keystore.get_api_key), picking up changes on the next poll without a -# container restart. Keys set here in .env are only a fallback for the FIRMS -# ingestor until a value is saved via the UI. +# Keys such as GEMINI_API_KEY and TELEGRAM_TOKEN are stored in the Postgres +# `api_keys` table and managed from the dashboard's "Keys" tab +# (GET/POST/DELETE /api/keys/{name}) — see app/keystore.py. The FIRMS ingestor +# currently reads FIRMS_MAP_KEY from .env (above); wiring the Keys-UI store as +# its lookup/fallback is a planned follow-up. diff --git a/alembic/versions/002_cameras.py b/alembic/versions/002_cameras.py new file mode 100644 index 0000000..869e351 --- /dev/null +++ b/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') diff --git a/app/camera_config.py b/app/camera_config.py new file mode 100644 index 0000000..96020ea --- /dev/null +++ b/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: +# [|,||] +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") diff --git a/app/camera_models.py b/app/camera_models.py new file mode 100644 index 0000000..6fdae2d --- /dev/null +++ b/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) diff --git a/app/camera_scraper.py b/app/camera_scraper.py new file mode 100644 index 0000000..bb6ecf9 --- /dev/null +++ b/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: +# ... latitude: 48.85 / longitude: 2.35 ... title="Paris" +# We parse generically: find lat/lon pairs plus nearby snapshot 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"]+src=[\"']([^\"']+\.(?:jpg|jpeg|png|mjpeg))[\"']", re.I) +_TITLE_RE = re.compile(r"([^<]+)", 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 diff --git a/app/main.py b/app/main.py index 029d7b7..ad619da 100644 --- a/app/main.py +++ b/app/main.py @@ -696,6 +696,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) diff --git a/app/run_camera_service.py b/app/run_camera_service.py new file mode 100644 index 0000000..c550d04 --- /dev/null +++ b/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()) diff --git a/docker-compose.yml b/docker-compose.yml index 930fe21..4e4b382 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -112,5 +112,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: