1f23083351
Add a server-side REST poller for VesselAPI (free tier 150 calls/mo) that upserts last-known positions into the shared vessel_last_known store when AISStream is unset. Default box is the Strait of Hormuz (span 3.6 <= 4 deg), never CONUS/NC (AISStream owns US coasts). - app/vesselapi.py: worker loop, 4deg span validator, position->marker transform (skip suspected_glitch), durable Postgres daily-quota table. - Local hard cap 5 successful 2xx/UTC day (VESSELAPI_MAX_CALLS_PER_DAY), monthly floor from X-RateLimit-Remaining, single request limit=50, no nextToken, no filter.sat, no retry-storm. - keystore VESSELAPI_API_KEY registry entry; config + compose env passthrough (app + ingest); wired next to AISStream in main.py lifespan + run_ingester. - GET /api/vessels docstring notes AISStream and/or VesselAPI cache. - tests/test_vesselapi.py: 20 unit tests (no network/DB).
443 lines
15 KiB
Python
443 lines
15 KiB
Python
"""VesselAPI REST poller — quota-capped AIS fallback (free tier 150 calls/mo).
|
|
|
|
AISStream (WebSocket) keeps US coasts live; VesselAPI fills the Middle East
|
|
blind spot (Strait of Hormuz default box). This worker polls the REST
|
|
``GET /v1/location/vessels/bounding-box`` endpoint at most
|
|
``VESSELAPI_MAX_CALLS_PER_DAY`` (default 5) *successful 2xx* calls per UTC day
|
|
and upserts the results into the shared ``vessel_last_known`` store.
|
|
|
|
Idle (no crash) when VESSELAPI_API_KEY is unset. Never called from the GET
|
|
/api/vessels path — map pans must not hit upstream. One request per poll,
|
|
``pagination.limit=50``, never follow ``nextToken``, never send ``filter.sat``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import calendar
|
|
import logging
|
|
import os
|
|
from datetime import date, datetime, timezone
|
|
|
|
import httpx
|
|
from sqlalchemy import Column, Date, DateTime, Integer, Table, func, select, text
|
|
|
|
from config import (
|
|
OSINT_USER_AGENT,
|
|
VESSELAPI_API_KEY,
|
|
VESSELAPI_BBOX,
|
|
VESSELAPI_INTERVAL,
|
|
VESSELAPI_MAX_CALLS_PER_DAY,
|
|
)
|
|
from database import async_session, engine, metadata
|
|
from live_layers import to_marker, upsert_vessel
|
|
|
|
logger = logging.getLogger("osint.vesselapi")
|
|
|
|
BASE_URL = "https://api.vesselapi.com/v1"
|
|
ENDPOINT = f"{BASE_URL}/location/vessels/bounding-box"
|
|
MAX_SPAN_DEG = 4.0 # |dLat| + |dLon| — VesselAPI 400s above this.
|
|
PAGE_LIMIT = 50 # pagination.limit; never follow nextToken on the free tier.
|
|
|
|
_client: httpx.AsyncClient | None = None
|
|
_client_lock = asyncio.Lock()
|
|
|
|
|
|
# ── Box parsing / span validation ─────────────────────────────────────────
|
|
|
|
class BboxError(ValueError):
|
|
"""A configured VesselAPI box violates the 4° span rule or is malformed."""
|
|
|
|
|
|
def validate_bbox_span(
|
|
minlat: float, minlon: float, maxlat: float, maxlon: float,
|
|
) -> None:
|
|
"""Reject boxes VesselAPI would 400 on (span > 4°, bad order, bad range)."""
|
|
if not (-90 <= minlat <= 90 and -90 <= maxlat <= 90
|
|
and -180 <= minlon <= 180 and -180 <= maxlon <= 180):
|
|
raise BboxError("coordinates out of range")
|
|
if minlat >= maxlat or minlon >= maxlon:
|
|
raise BboxError("bbox must have min < max on both axes")
|
|
dlat = abs(maxlat - minlat)
|
|
dlon = abs(maxlon - minlon)
|
|
if dlat + dlon > MAX_SPAN_DEG:
|
|
raise BboxError(
|
|
f"span |dLat|+|dLon| = {dlat + dlon:.2f}° exceeds {MAX_SPAN_DEG}° cap"
|
|
)
|
|
|
|
|
|
def parse_boxes(raw: str) -> list[tuple[float, float, float, float]]:
|
|
"""Env format: ``minlat,minlon,maxlat,maxlon[; ...]`` (lat/lon order)."""
|
|
out: list[tuple[float, float, float, float]] = []
|
|
for chunk in (raw or "").split(";"):
|
|
parts = [p.strip() for p in chunk.split(",") if p.strip()]
|
|
if len(parts) != 4:
|
|
continue
|
|
try:
|
|
minlat = float(parts[0])
|
|
minlon = float(parts[1])
|
|
maxlat = float(parts[2])
|
|
maxlon = float(parts[3])
|
|
except ValueError:
|
|
continue
|
|
out.append((minlat, minlon, maxlat, maxlon))
|
|
return out
|
|
|
|
|
|
def parse_boxes_validated(raw: str) -> list[tuple[float, float, float, float]]:
|
|
"""Parse boxes, log + skip any that violate the span/order/range rules."""
|
|
valid: list[tuple[float, float, float, float]] = []
|
|
for box in parse_boxes(raw):
|
|
try:
|
|
validate_bbox_span(*box)
|
|
valid.append(box)
|
|
except BboxError as exc:
|
|
logger.warning("VesselAPI bbox %r skipped: %s", box, exc)
|
|
return valid
|
|
|
|
|
|
# ── Position → marker transform ───────────────────────────────────────────
|
|
|
|
def _f(value: object) -> float | None:
|
|
if value is None or value == "":
|
|
return None
|
|
try:
|
|
return float(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _s(value: object) -> str | None:
|
|
if value is None:
|
|
return None
|
|
text = str(value).strip()
|
|
return text or None
|
|
|
|
|
|
def transform_vesselapi_position(obj: dict | None) -> dict | None:
|
|
"""Map one VesselAPI position object to the shared marker contract.
|
|
|
|
Returns None for glitch rows, missing MMSI, or missing coordinates.
|
|
"""
|
|
if not obj or not isinstance(obj, dict):
|
|
return None
|
|
if obj.get("suspected_glitch") is True:
|
|
return None
|
|
mmsi = obj.get("mmsi")
|
|
if mmsi is None:
|
|
return None
|
|
lat = _f(obj.get("latitude"))
|
|
lon = _f(obj.get("longitude"))
|
|
if lat is None or lon is None:
|
|
return None
|
|
mmsi_s = str(mmsi)
|
|
name = _s(obj.get("vessel_name") or obj.get("name"))
|
|
heading = _f(obj.get("heading"))
|
|
if heading is None:
|
|
heading = _f(obj.get("cog"))
|
|
sog = _f(obj.get("sog"))
|
|
extra: dict = {
|
|
"src": "vesselapi",
|
|
"mmsi": mmsi_s,
|
|
"cog": obj.get("cog"),
|
|
"sog": obj.get("sog"),
|
|
"navstat": obj.get("nav_status"),
|
|
}
|
|
imo = obj.get("imo")
|
|
if imo:
|
|
extra["imo"] = imo
|
|
dest = _s(obj.get("dest") or obj.get("destination"))
|
|
if dest:
|
|
extra["dest"] = dest
|
|
ts = obj.get("timestamp") or obj.get("processed_timestamp")
|
|
if ts:
|
|
extra["timestamp"] = ts
|
|
return to_marker(
|
|
mmsi_s, lat, lon,
|
|
heading=heading,
|
|
speed=sog,
|
|
label=name or mmsi_s,
|
|
extra=extra,
|
|
)
|
|
|
|
|
|
def transform_vesselapi_payload(payload: dict | None) -> list[dict]:
|
|
"""Flatten a bounding-box response ``{vessels: [...]}`` to markers."""
|
|
if not payload or not isinstance(payload, dict):
|
|
return []
|
|
rows = payload.get("vessels") or []
|
|
out = []
|
|
for row in rows:
|
|
marker = transform_vesselapi_position(row)
|
|
if marker:
|
|
out.append(marker)
|
|
return out
|
|
|
|
|
|
# ── Durable daily quota (Postgres, survives restarts) ─────────────────────
|
|
# Mirrors keystore.api_keys: lazy CREATE TABLE IF NOT EXISTS, no alembic fork.
|
|
|
|
vesselapi_quota = Table(
|
|
"vesselapi_quota",
|
|
metadata,
|
|
Column("day", Date, primary_key=True),
|
|
Column("calls", Integer, nullable=False, server_default="0"),
|
|
Column("remaining", Integer, nullable=True),
|
|
Column("updated_at", DateTime(timezone=True), server_default=func.now(), nullable=False),
|
|
)
|
|
|
|
_CREATE_QUOTA_SQL = text(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS vesselapi_quota (
|
|
day DATE PRIMARY KEY,
|
|
calls INTEGER NOT NULL DEFAULT 0,
|
|
remaining INTEGER,
|
|
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
|
)
|
|
"""
|
|
)
|
|
|
|
_quota_lock = asyncio.Lock()
|
|
_quota_ensured = False
|
|
|
|
|
|
async def ensure_quota_table() -> None:
|
|
global _quota_ensured
|
|
if _quota_ensured:
|
|
return
|
|
async with _quota_lock:
|
|
if _quota_ensured:
|
|
return
|
|
async with engine.begin() as conn:
|
|
await conn.execute(_CREATE_QUOTA_SQL)
|
|
_quota_ensured = True
|
|
|
|
|
|
class PgQuotaStore:
|
|
"""Postgres-backed daily call counter. Injected for tests."""
|
|
|
|
async def calls_today(self, day: date) -> int:
|
|
await ensure_quota_table()
|
|
async with async_session() as session:
|
|
row = (await session.execute(
|
|
select(vesselapi_quota.c.calls).where(vesselapi_quota.c.day == day)
|
|
)).scalar()
|
|
return int(row) if row else 0
|
|
|
|
async def remaining_today(self, day: date) -> int | None:
|
|
await ensure_quota_table()
|
|
async with async_session() as session:
|
|
row = (await session.execute(
|
|
select(vesselapi_quota.c.remaining).where(vesselapi_quota.c.day == day)
|
|
)).scalar()
|
|
return int(row) if row is not None else None
|
|
|
|
async def bump(self, day: date, remaining: int | None) -> int:
|
|
await ensure_quota_table()
|
|
now = datetime.now(timezone.utc)
|
|
async with async_session() as session:
|
|
existing = (await session.execute(
|
|
select(vesselapi_quota.c.calls).where(vesselapi_quota.c.day == day)
|
|
)).scalar()
|
|
if existing is None:
|
|
await session.execute(
|
|
vesselapi_quota.insert().values(
|
|
day=day, calls=1, remaining=remaining, updated_at=now,
|
|
)
|
|
)
|
|
else:
|
|
await session.execute(
|
|
vesselapi_quota.update()
|
|
.where(vesselapi_quota.c.day == day)
|
|
.values(
|
|
calls=vesselapi_quota.c.calls + 1,
|
|
remaining=remaining,
|
|
updated_at=now,
|
|
)
|
|
)
|
|
await session.commit()
|
|
return (int(existing) if existing else 0) + 1
|
|
|
|
|
|
# ── Budget / scheduling (pure, unit-testable) ─────────────────────────────
|
|
|
|
def days_left_in_month(now: datetime) -> int:
|
|
"""UTC days remaining in the current month, inclusive of today."""
|
|
_, last = calendar.monthrange(now.year, now.month)
|
|
return last - now.day + 1
|
|
|
|
|
|
def budget_allows(
|
|
calls_today: int,
|
|
remaining: int | None,
|
|
days_left: int,
|
|
max_per_day: int,
|
|
) -> bool:
|
|
"""True if another poll is permitted today.
|
|
|
|
Local hard cap: fewer than ``max_per_day`` successful calls today.
|
|
Monthly floor: if ``X-RateLimit-Remaining`` is known, keep at least
|
|
``max_per_day * days_left`` in reserve for the rest of the month.
|
|
"""
|
|
if calls_today >= max_per_day:
|
|
return False
|
|
if remaining is not None and remaining <= max_per_day * days_left:
|
|
return False
|
|
return True
|
|
|
|
|
|
def choose_box(
|
|
boxes: list[tuple[float, float, float, float]],
|
|
calls_today: int,
|
|
max_per_day: int,
|
|
) -> int:
|
|
"""Index into ``boxes`` for the next poll.
|
|
|
|
Prefer refreshing the first (primary) box rather than spraying one call
|
|
across every region — round-robin only when the remaining daily budget is
|
|
enough to cover all boxes.
|
|
"""
|
|
if len(boxes) <= 1:
|
|
return 0
|
|
budget_left = max_per_day - calls_today
|
|
if budget_left >= len(boxes):
|
|
return calls_today % len(boxes)
|
|
return 0
|
|
|
|
|
|
# ── HTTP / poll ───────────────────────────────────────────────────────────
|
|
|
|
async def _resolve_key() -> str:
|
|
from keystore import get_api_key
|
|
return (
|
|
os.getenv("VESSELAPI_API_KEY")
|
|
or VESSELAPI_API_KEY
|
|
or (await get_api_key("VESSELAPI_API_KEY"))
|
|
or ""
|
|
).strip()
|
|
|
|
|
|
async def _get_client() -> httpx.AsyncClient:
|
|
global _client
|
|
if _client is None:
|
|
async with _client_lock:
|
|
if _client is None:
|
|
_client = httpx.AsyncClient(
|
|
timeout=httpx.Timeout(15.0, connect=5.0),
|
|
follow_redirects=True,
|
|
headers={"User-Agent": OSINT_USER_AGENT, "Accept": "application/json"},
|
|
limits=httpx.Limits(max_connections=1, max_keepalive_connections=1),
|
|
)
|
|
return _client
|
|
|
|
|
|
async def close_client() -> None:
|
|
global _client
|
|
if _client is not None:
|
|
await _client.aclose()
|
|
_client = None
|
|
|
|
|
|
def _int_header(value: str | None) -> int | None:
|
|
if value is None:
|
|
return None
|
|
try:
|
|
return int(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
async def poll_once(store, boxes: list[tuple[float, float, float, float]], key: str) -> bool:
|
|
"""One quota-checked poll. Returns True if a successful 2xx was made.
|
|
|
|
Only successful 2xx responses count against the monthly quota; 4xx/5xx/429
|
|
are skipped without retry-storming (Retry-After respected by simply
|
|
sleeping the interval).
|
|
"""
|
|
now = datetime.now(timezone.utc)
|
|
today = now.date()
|
|
calls = await store.calls_today(today)
|
|
remaining = await store.remaining_today(today)
|
|
days_left = days_left_in_month(now)
|
|
if not budget_allows(calls, remaining, days_left, VESSELAPI_MAX_CALLS_PER_DAY):
|
|
logger.info(
|
|
"VesselAPI quota reached (calls_today=%d, remaining=%s, days_left=%d) — skip poll",
|
|
calls, remaining, days_left,
|
|
)
|
|
return False
|
|
|
|
idx = choose_box(boxes, calls, VESSELAPI_MAX_CALLS_PER_DAY)
|
|
minlat, minlon, maxlat, maxlon = boxes[idx]
|
|
client = await _get_client()
|
|
params = {
|
|
"filter.latBottom": str(minlat),
|
|
"filter.latTop": str(maxlat),
|
|
"filter.lonLeft": str(minlon),
|
|
"filter.lonRight": str(maxlon),
|
|
"pagination.limit": str(PAGE_LIMIT),
|
|
}
|
|
headers = {"Authorization": f"Bearer {key}"}
|
|
try:
|
|
resp = await client.get(ENDPOINT, params=params, headers=headers)
|
|
except httpx.HTTPError as exc:
|
|
logger.warning("VesselAPI request failed: %s", exc)
|
|
return False
|
|
|
|
if resp.status_code == 429:
|
|
logger.warning(
|
|
"VesselAPI rate-limited (Retry-After=%s) — skip poll",
|
|
resp.headers.get("Retry-After"),
|
|
)
|
|
return False
|
|
if resp.status_code >= 400:
|
|
logger.warning("VesselAPI HTTP %d — not counted against quota", resp.status_code)
|
|
return False
|
|
|
|
# 2xx success — counts against the monthly quota.
|
|
remaining = _int_header(resp.headers.get("X-RateLimit-Remaining"))
|
|
calls = await store.bump(today, remaining)
|
|
try:
|
|
data = resp.json()
|
|
except ValueError:
|
|
logger.warning("VesselAPI 2xx with non-JSON body — counted but ignored")
|
|
return True
|
|
markers = transform_vesselapi_payload(data)
|
|
for m in markers:
|
|
await upsert_vessel(m)
|
|
logger.info(
|
|
"VesselAPI poll OK: %d vessels (remaining=%s, calls_today=%d)",
|
|
len(markers), remaining, calls,
|
|
)
|
|
return True
|
|
|
|
|
|
# ── Worker loop ───────────────────────────────────────────────────────────
|
|
|
|
async def run_vesselapi_worker(store: PgQuotaStore | None = None) -> None:
|
|
"""Long-lived poll loop. Idle when the key is unset; never crashes the app."""
|
|
if store is None:
|
|
store = PgQuotaStore()
|
|
boxes = parse_boxes_validated(VESSELAPI_BBOX)
|
|
if not boxes:
|
|
logger.warning(
|
|
"VESSELAPI_BBOX has no valid boxes (span ≤ %.1f°) — poller idle", MAX_SPAN_DEG,
|
|
)
|
|
while True:
|
|
try:
|
|
if not boxes:
|
|
await asyncio.sleep(VESSELAPI_INTERVAL)
|
|
continue
|
|
key = await _resolve_key()
|
|
if not key:
|
|
logger.warning(
|
|
"VESSELAPI_API_KEY not set — VesselAPI poller idle. "
|
|
"Create a free key at https://dashboard.vesselapi.com/"
|
|
)
|
|
await asyncio.sleep(VESSELAPI_INTERVAL)
|
|
continue
|
|
await poll_once(store, boxes, key)
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception: # noqa: BLE001 — keep the loop alive across transient failures
|
|
logger.exception("VesselAPI poll error")
|
|
await asyncio.sleep(VESSELAPI_INTERVAL)
|