69a30115cc
Some checks failed
build-and-deploy / build (push) Failing after 5s
- 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.
106 lines
3.7 KiB
Python
106 lines
3.7 KiB
Python
"""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())
|