Source code for luna_api.app.handlers.matcher_handler

""" Matcher handler. """
from typing import Literal, Optional

from app.handlers.base_handler import PythonMatcherServiceBaseHandler
from app.handlers.schemas import schemas
from classes.token import Permissions
from crutches_on_wheels.errors.errors import Error
from crutches_on_wheels.errors.exception import VLException


class MatcherFacesHandler(PythonMatcherServiceBaseHandler):
    """
    Matcher handler allows to submit tasks to a service that searches for faces similar to a given reference by
    matching them. See `spec_matcher <_static/api.html#tag/matchFaces>`_.

    Resource: "/{api_version}/matcher/faces"
    """

    allowedMethods = ("POST",)

    @staticmethod
    def getUnallowedMatchingSource(
        inputJson: dict, permissions: Permissions
    ) -> Optional[Literal["face", "event", "attribute"]]:
        """
        Get unallowed matching source if exists
        Args:
            inputJson: input json from request
            permissions: token permissions
        Returns:
            unallowed matching source
        """
        faceCondition = "matching" in permissions.face
        eventCondition = "matching" in permissions.event
        attrCondition = "matching" in permissions.attribute

        candidateOrigins = {candidate["filters"]["origin"] for candidate in inputJson["candidates"]}
        if "faces" in candidateOrigins and not faceCondition:
            return "face"
        if "events" in candidateOrigins and not eventCondition:
            return "event"
        if "attributes" in candidateOrigins and not attrCondition:
            return "attribute"

        referenceOrigins = {reference["type"] for reference in inputJson["references"]}
        if ("face" in referenceOrigins or "face_external_id" in referenceOrigins) and not faceCondition:
            return "face"
        if (
            "event" in referenceOrigins
            or "event_external_id" in referenceOrigins
            or "event_track_id" in referenceOrigins
        ) and not eventCondition:
            return "event"
        if "attribute" in referenceOrigins and not attrCondition:
            return "attribute"

    def checkTokenPermissions(self) -> None:
        """
        Description see :func:`~BaseRequestHandler.checkTokenPermissions`.
        """
        if (permissions := self.request.credentials.permissions) is not None:
            self.validateJson(self.request.json, schemas.FACE_MATCHING_SCHEMA)
            if (
                self.request.method == "POST"
                and (restrictedSource := self.getUnallowedMatchingSource(self.request.json, permissions)) is not None
            ):
                raise VLException(Error.ForbiddenByToken.format("matching", restrictedSource), 403, False)


class RawMatcherHandler(PythonMatcherServiceBaseHandler):
    """
    Match each candidate descriptor with each reference descriptor.
    See `spec matcher raw <_static/api.html#tag/matchingRaw>`_.

    Resource: "/{api_version}/matcher/raw"
    """

    allowedMethods = ("POST",)

    def checkTokenPermissions(self) -> None:
        """
        Description see :func:`~BaseRequestHandler.checkTokenPermissions`.
        """


class MatcherBodiesHandler(PythonMatcherServiceBaseHandler):
    """
    Matcher handler allows to submit tasks to a service that searches for faces similar to a given reference by
    matching them. See `spec matcher <_static/api.html#tag/matchBodies>`_.

    Resource: "/{api_version}/matcher/bodies"
    """

    allowedMethods = ("POST",)

    @staticmethod
    def getUnallowedMatchingSource(inputJson: dict, permissions: Permissions) -> Optional[Literal["event"]]:
        """
        Get unallowed matching source if exists
        Args:
            inputJson: input json from request
            permissions: token permissions
        Returns:
            unallowed matching source
        """
        eventCondition = "matching" in permissions.event

        candidateOrigins = {candidate["filters"]["origin"] for candidate in inputJson["candidates"]}
        if "events" in candidateOrigins and not eventCondition:
            return "event"

        referenceOrigins = {reference["type"] for reference in inputJson["references"]}
        if (
            "event" in referenceOrigins
            or "event_external_id" in referenceOrigins
            or "event_track_id" in referenceOrigins
        ) and not eventCondition:
            return "event"

    def checkTokenPermissions(self) -> None:
        """
        Description see :func:`~BaseRequestHandler.checkTokenPermissions`.
        """
        if (permissions := self.request.credentials.permissions) is not None:
            self.validateJson(self.request.json, schemas.BODY_MATCHING_SCHEMA)
            if (
                self.request.method == "POST"
                and (restrictedSource := self.getUnallowedMatchingSource(self.request.json, permissions)) is not None
            ):
                raise VLException(Error.ForbiddenByToken.format("matching", restrictedSource), 403, False)