5ea9a4e879
Phase 1: in-memory ConnectionManager viewport fan-out, 500ms map debounce, cachetools TTLCache, background masscan/ffmpeg, compose memory caps. Phase 2: PostGIS geofences + ST_Intersects alerts, Timescale 1-min CAGGs and timestamp playback, FIRMS/WFIGS x firefighting ADS-B within 20 miles. No Redis/Kafka/Celery.
246 lines
8 KiB
Python
246 lines
8 KiB
Python
"""Resolve a browser-renderable preview for a camera.
|
|
|
|
HTTP/MJPEG cameras already expose a snapshot_url the existing proxy can
|
|
stream. masscan finds are stored as `rtsp://IP/` with no snapshot_url, so
|
|
the map popup used to skip the <img> entirely and the leftover source link
|
|
handed the browser an rtsp:// URL (which opens VLC).
|
|
|
|
This module:
|
|
1. Tries a short list of unauthenticated HTTP snapshot paths (fast).
|
|
2. Falls back to grabbing one JPEG frame from RTSP via ffmpeg (no auth).
|
|
3. Remembers the first URL that worked on the camera row so the next
|
|
popup is a cache hit.
|
|
|
|
No credentials are ever tried.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import shutil
|
|
from urllib.parse import urlparse
|
|
|
|
import httpx
|
|
|
|
from camera_config import USER_AGENT
|
|
from camera_models import cameras
|
|
from camera_scraper import fetch_snapshot
|
|
from database import async_session
|
|
|
|
logger = logging.getLogger("osint.camera_preview")
|
|
|
|
# Most common unauthenticated still-image endpoints on consumer NVRs/IP cams.
|
|
# Keep this list SHORT — it runs on popup click.
|
|
_HTTP_PATHS = (
|
|
"/snapshot.jpg",
|
|
"/cgi-bin/snapshot.cgi",
|
|
"/jpg/image.jpg",
|
|
"/image.jpg",
|
|
"/onvif/snapshot",
|
|
"/axis-cgi/jpg/image.cgi",
|
|
"/tmpfs/auto.jpg",
|
|
)
|
|
|
|
# Browser-playable MJPEG paths the /stream proxy can pass through.
|
|
_MJPEG_PATHS = (
|
|
"/mjpg/video.mjpg",
|
|
"/video.mjpg",
|
|
"/cgi-bin/mjpg/video.cgi",
|
|
"/axis-cgi/mjpg/video.cgi",
|
|
"/nphMotionJpeg",
|
|
"/mjpeg.cgi",
|
|
)
|
|
|
|
_FFMPEG = shutil.which("ffmpeg")
|
|
|
|
|
|
def _host_from_url(url: str) -> str | None:
|
|
try:
|
|
return urlparse(url).hostname
|
|
except Exception: # noqa: BLE001
|
|
return None
|
|
|
|
|
|
def _looks_like_jpeg(data: bytes) -> bool:
|
|
return bool(data) and len(data) >= 64 and data[:2] == b"\xff\xd8"
|
|
|
|
|
|
async def _http_get_image(url: str, timeout: float = 2.5) -> bytes | None:
|
|
try:
|
|
async with httpx.AsyncClient(
|
|
timeout=timeout, follow_redirects=True,
|
|
headers={"User-Agent": USER_AGENT},
|
|
) as c:
|
|
r = await c.get(url)
|
|
if r.status_code != 200:
|
|
return None
|
|
ctype = (r.headers.get("content-type") or "").lower()
|
|
if "html" in ctype or "text/" in ctype:
|
|
return None
|
|
if not _looks_like_jpeg(r.content) and "image/" not in ctype:
|
|
return None
|
|
if len(r.content) < 64:
|
|
return None
|
|
return r.content
|
|
except Exception: # noqa: BLE001
|
|
return None
|
|
|
|
|
|
async def _http_feed_url(url: str, timeout: float = 2.5) -> str | None:
|
|
"""Return url if it looks like an unauthenticated image/MJPEG feed."""
|
|
try:
|
|
async with httpx.AsyncClient(
|
|
timeout=timeout, follow_redirects=True,
|
|
headers={"User-Agent": USER_AGENT},
|
|
) as c:
|
|
async with c.stream("GET", url) as r:
|
|
if r.status_code != 200:
|
|
return None
|
|
ctype = (r.headers.get("content-type") or "").lower()
|
|
if "html" in ctype or ctype.startswith("text/"):
|
|
return None
|
|
if any(x in ctype for x in ("image/", "multipart", "mjpeg", "octet-stream")):
|
|
# Read a little to reject empty/error bodies.
|
|
chunk = b""
|
|
async for b in r.aiter_bytes():
|
|
chunk += b
|
|
if len(chunk) >= 64:
|
|
break
|
|
if len(chunk) < 64:
|
|
return None
|
|
if b"html" in chunk[:64].lower():
|
|
return None
|
|
return url
|
|
except Exception: # noqa: BLE001
|
|
return None
|
|
return None
|
|
|
|
|
|
async def probe_public_feed(host: str) -> str | None:
|
|
"""Unauthenticated HTTP still or MJPEG URL for this host, or None.
|
|
|
|
Used at masscan ingest time so dead RTSP-only hosts never hit the map.
|
|
No credentials, no RTSP path-walking (too slow / rarely public).
|
|
"""
|
|
urls = [f"http://{host}{p}" for p in _HTTP_PATHS]
|
|
urls.append(f"http://{host}:8080/shot.jpg")
|
|
urls.extend(f"http://{host}{p}" for p in _MJPEG_PATHS)
|
|
results = await asyncio.gather(
|
|
*(_http_feed_url(u) for u in urls),
|
|
return_exceptions=True,
|
|
)
|
|
for url, hit in zip(urls, results):
|
|
if isinstance(hit, str) and hit:
|
|
return hit
|
|
return None
|
|
|
|
|
|
async def ffmpeg_snapshot(url: str, timeout: float = 8.0) -> bytes | None:
|
|
"""Grab a single JPEG frame from an RTSP URL. None if ffmpeg missing/fails.
|
|
|
|
The subprocess is scheduled via asyncio.create_task (shared per URL) so
|
|
concurrent popup clicks do not stack ffmpeg processes on the request path.
|
|
"""
|
|
from bg_jobs import cached_ffmpeg_jpeg, schedule_ffmpeg_snapshot
|
|
|
|
hit = cached_ffmpeg_jpeg(url)
|
|
if hit:
|
|
return hit
|
|
return await schedule_ffmpeg_snapshot(url, timeout)
|
|
|
|
|
|
async def ffmpeg_mjpeg_stream(url: str):
|
|
"""Yield an MJPEG multipart body transcoded from RTSP. Caller streams it."""
|
|
if not _FFMPEG:
|
|
return
|
|
cmd = [
|
|
_FFMPEG, "-hide_banner", "-loglevel", "error", "-nostdin",
|
|
]
|
|
if url.lower().startswith("rtsp://"):
|
|
cmd += ["-rtsp_transport", "tcp", "-timeout", "4000000"]
|
|
cmd += [
|
|
"-i", url,
|
|
"-an", "-c:v", "mjpeg", "-q:v", "8",
|
|
"-f", "mpjpeg", "pipe:1",
|
|
]
|
|
proc = await asyncio.create_subprocess_exec(
|
|
*cmd,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.DEVNULL,
|
|
)
|
|
try:
|
|
assert proc.stdout is not None
|
|
while True:
|
|
chunk = await proc.stdout.read(64 * 1024)
|
|
if not chunk:
|
|
break
|
|
yield chunk
|
|
finally:
|
|
if proc.returncode is None:
|
|
proc.kill()
|
|
try:
|
|
await proc.wait()
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
|
|
|
|
async def _remember(camera_id, url: str) -> None:
|
|
try:
|
|
async with async_session() as session:
|
|
await session.execute(
|
|
cameras.update()
|
|
.where(cameras.c.id == camera_id)
|
|
.values(snapshot_url=url)
|
|
)
|
|
await session.commit()
|
|
except Exception: # noqa: BLE001
|
|
logger.warning("failed to persist snapshot_url for %s", camera_id,
|
|
exc_info=True)
|
|
|
|
|
|
async def resolve_preview(row) -> tuple[bytes | None, str | None]:
|
|
"""Return (jpeg_bytes, working_url) for a cameras-table row.
|
|
|
|
Tries, in order:
|
|
* existing HTTP snapshot_url (via the TTL cache)
|
|
* common HTTP still-image paths on the host
|
|
* ffmpeg frame grab from the stored RTSP URL, then common RTSP paths
|
|
"""
|
|
snap = row.get("snapshot_url") or ""
|
|
source = row.get("source_url") or ""
|
|
host = _host_from_url(snap) or _host_from_url(source)
|
|
if not host:
|
|
return None, None
|
|
|
|
# 1. Known HTTP snapshot — go through the existing TTL cache.
|
|
if snap.lower().startswith(("http://", "https://")):
|
|
data = await fetch_snapshot(snap)
|
|
if data:
|
|
return data, snap
|
|
|
|
# 2. Probe unauthenticated HTTP stills in parallel (fast fail) BEFORE
|
|
# any ffmpeg — most open cams that preview at all do it over HTTP.
|
|
http_urls = [f"http://{host}{p}" for p in _HTTP_PATHS]
|
|
http_urls.append(f"http://{host}:8080/shot.jpg")
|
|
results = await asyncio.gather(
|
|
*(_http_get_image(u) for u in http_urls),
|
|
return_exceptions=True,
|
|
)
|
|
for url, data in zip(http_urls, results):
|
|
if isinstance(data, (bytes, bytearray)) and data:
|
|
await _remember(row["id"], url)
|
|
return bytes(data), url
|
|
|
|
# 3. One ffmpeg grab of the stored RTSP URL. Path-walking is too slow
|
|
# for a popup click; unauthenticated RTSP often needs a vendor path
|
|
# and/or credentials we will not try.
|
|
rtsp_url = snap if snap.lower().startswith("rtsp://") else source
|
|
if rtsp_url.lower().startswith("rtsp://"):
|
|
data = await ffmpeg_snapshot(rtsp_url, timeout=5.0)
|
|
if data:
|
|
if not snap:
|
|
await _remember(row["id"], rtsp_url)
|
|
return data, rtsp_url
|
|
|
|
return None, None
|