lasuite-docs/src/backend/core/services/converter_services.py

"""Y-Provider API services."""

import logging
import typing
from base64 import b64encode

from django.conf import settings

import requests

from core.services import mime_types
from core.services.jwt_services import Audiences, JWTService

logger = logging.getLogger(__name__)


class ConversionError(Exception):
    """Base exception for conversion-related errors."""


class ValidationError(ConversionError):
    """Raised when the input validation fails."""


class ServiceUnavailableError(ConversionError):
    """Raised when the conversion service is unavailable."""


class ConverterProtocol(typing.Protocol):
    """Protocol for converter classes."""

    def convert(self, data, content_type, accept):
        """Convert content from one format to another."""


class Converter:
    """Orchestrates conversion between different formats using specialized converters."""

    docspec: ConverterProtocol
    ydoc: ConverterProtocol

    def __init__(self):
        self.docspec = DocSpecConverter()
        self.ydoc = YdocConverter()

    def convert(self, data, content_type, accept):
        """Convert input into other formats using external microservices."""

        logger.info("converting content from %s to %s", content_type, accept)

        if content_type == mime_types.DOCX and accept == mime_types.YJS:
            blocknote_data = self.docspec.convert(
                data, content_type, mime_types.BLOCKNOTE
            )
            return self.ydoc.convert(
                blocknote_data, mime_types.BLOCKNOTE, mime_types.YJS
            )

        return self.ydoc.convert(data, content_type, accept)


class DocSpecConverter:
    """Service class for DocSpec conversion-related operations."""

    def _request(self, url, data, content_type):
        """Make a request to the DocSpec API."""

        response = requests.post(
            url,
            headers={
                "Content-Type": content_type,
                "Accept": mime_types.BLOCKNOTE,
            },
            data=data,
            timeout=settings.CONVERSION_API_TIMEOUT,
            verify=settings.CONVERSION_API_SECURE,
        )
        if not response.ok:
            logger.error(
                "DocSpec API error: url=%s, status=%d, response=%s",
                url,
                response.status_code,
                response.text[:200] if response.text else "empty",
            )
        response.raise_for_status()
        return response

    def convert(self, data, content_type, accept):
        """Convert a Document to BlockNote."""
        if not data:
            raise ValidationError("Input data cannot be empty")

        if content_type != mime_types.DOCX or accept != mime_types.BLOCKNOTE:
            raise ValidationError(
                f"Conversion from {content_type} to {accept} is not supported."
            )

        try:
            return self._request(settings.DOCSPEC_API_URL, data, content_type).content
        except requests.RequestException as err:
            logger.exception("DocSpec service error: url=%s", settings.DOCSPEC_API_URL)
            raise ServiceUnavailableError(
                "Failed to connect to DocSpec conversion service",
            ) from err


class YdocConverter:
    """Service class for YDoc conversion-related operations."""

    @property
    def auth_header(self):
        """Build microservice authentication header."""
        token = JWTService().get_admin_token(audience=Audiences.Y_CONVERTER)
        return f"Bearer {token}"

    def _request(self, url, data, content_type, accept):
        """Make a request to the Y-Provider API."""
        response = requests.post(
            url,
            data=data,
            headers={
                "Authorization": self.auth_header,
                "Content-Type": content_type,
                "Accept": accept,
            },
            timeout=settings.CONVERSION_API_TIMEOUT,
            verify=settings.CONVERSION_API_SECURE,
        )
        if not response.ok:
            logger.error(
                "Y-Provider API error: url=%s, status=%d, response=%s",
                url,
                response.status_code,
                response.text[:200] if response.text else "empty",
            )
        response.raise_for_status()
        return response

    def convert(self, data, content_type=mime_types.MARKDOWN, accept=mime_types.YJS):
        """Convert a Markdown text into our internal format using an external microservice."""

        if not data:
            raise ValidationError("Input data cannot be empty")

        url = f"{settings.Y_PROVIDER_API_BASE_URL}{settings.CONVERSION_API_ENDPOINT}/"
        try:
            response = self._request(url, data, content_type, accept)
            if accept == mime_types.YJS:
                return b64encode(response.content).decode("utf-8")
            if accept in {mime_types.MARKDOWN, "text/html"}:
                return response.text
            if accept == mime_types.JSON:
                return response.json()
            raise ValidationError("Unsupported format")
        except requests.RequestException as err:
            logger.exception("Y-Provider service error: url=%s", url)
            raise ServiceUnavailableError(
                f"Failed to connect to YDoc conversion service {content_type}, {accept}",
            ) from err