digital_reception_ra/apps/video-ingest/video_ingest/api.py

"""HTTP клиенты к face-service и apps/api для минимального video-ingest."""
from __future__ import annotations

import os
import base64
import logging
from datetime import datetime
from typing import Any

import requests

logger = logging.getLogger(__name__)


class FaceServiceClient:
    def __init__(self, base_url: str | None = None) -> None:
        self.base_url = (base_url or os.getenv("FACE_SERVICE_URL", "http://localhost:8001")).rstrip("/")

    def embed(self, jpeg_bytes: bytes) -> tuple[list[float], float] | None:
        """Возвращает (embedding, quality) или None если лицо не найдено."""
        b64 = base64.b64encode(jpeg_bytes).decode("ascii")
        r = requests.post(f"{self.base_url}/embed", json={"frame": b64}, timeout=10)
        r.raise_for_status()
        data = r.json()
        if data is None:
            return None
        return data["embedding"], data["quality"]

    def save_track_embedding(
        self,
        jpeg_bytes: bytes,
        track_id: str,
        camera_id: str,
        captured_at: datetime,
    ) -> dict[str, Any] | None:
        b64 = base64.b64encode(jpeg_bytes).decode("ascii")
        r = requests.post(
            f"{self.base_url}/track-embeddings",
            json={
                "frame": b64,
                "track_id": track_id,
                "camera_id": camera_id,
                "captured_at": captured_at.isoformat(),
            },
            timeout=10,
        )
        r.raise_for_status()
        return r.json()


class ReceptionApiClient:
    def __init__(self, base_url: str | None = None) -> None:
        self.base_url = (base_url or os.getenv("API_BASE_URL", "http://localhost:4000")).rstrip("/")

    def create_track(self, camera_name: str, first_seen_at: datetime) -> dict[str, Any]:
        r = requests.post(
            f"{self.base_url}/ingest/tracks",
            json={"cameraName": camera_name, "firstSeenAt": first_seen_at.isoformat()},
            timeout=10,
        )
        r.raise_for_status()
        return r.json()

    def add_event(
        self,
        track_id: str,
        event_type: str,
        camera_name: str,
        occurred_at: datetime,
        zone_code: str | None = None,
    ) -> dict[str, Any]:
        body = {
            "trackId": track_id,
            "type": event_type,
            "cameraName": camera_name,
            "occurredAt": occurred_at.isoformat(),
        }
        if zone_code:
            body["zoneCode"] = zone_code
        r = requests.post(f"{self.base_url}/ingest/track-events", json=body, timeout=10)
        r.raise_for_status()
        return r.json()