apps/api/app/lenses/base.py

from __future__ import annotations

import hashlib
import json
from abc import ABC, abstractmethod
from typing import Any

from app.storage.models import VisualAssetRecord, VisualViewRecord
from app.storage.repositories import new_id, now_ts


def canonical_hash(parameters: dict[str, Any]) -> str:
    canonical = json.dumps(parameters, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(canonical.encode()).hexdigest()


class Lens(ABC):
    name: str
    version: str = "v1"
    view_type: str

    @abstractmethod
    async def execute(self, asset: VisualAssetRecord, image_bytes: bytes, parameters: dict[str, Any]) -> tuple[dict[str, Any], float, list[dict[str, Any]]]:
        raise NotImplementedError

    async def run(self, asset: VisualAssetRecord, image_bytes: bytes, parameters: dict[str, Any]) -> VisualViewRecord:
        payload, confidence, provenance = await self.execute(asset, image_bytes, parameters)
        ts = now_ts()
        return VisualViewRecord(
            id=new_id("view"),
            asset_id=asset.id,
            sha256=asset.sha256,
            view_type=self.view_type,
            lens_name=self.name,
            lens_version=self.version,
            parameters_hash=canonical_hash(parameters),
            status="ready",
            payload_json=payload,
            confidence=confidence,
            provenance_json=provenance,
            created_at=ts,
            updated_at=ts,
        )