main.py

"""Language Detection API"""

import logging
import time
from os import environ
from typing import Annotated

from dotenv import load_dotenv
from litestar import Litestar, get, post
from litestar.config.cors import CORSConfig
from litestar.datastructures import State
from litestar.openapi import OpenAPIConfig
from litestar.params import Parameter
from litestar.response import Redirect
from litestar.status_codes import HTTP_200_OK

from app.http_exceptions import EmptyTextException
from app.language_utils import get_all_supported_languages
from app.schemas import DetectRequest, DetectResponse, LanguageInfo
from detectors.lingua_detector import LinguaDetector
from detectors.mediapipe_detector import MediaPipeDetector

# Setup logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

load_dotenv()

# Initialize detectors lazily
_mediapipe_detector = None
_lingua_detector = None


def get_mediapipe_detector() -> MediaPipeDetector:
    """Get or initialize MediaPipe detector"""
    global _mediapipe_detector
    if _mediapipe_detector is None:
        model_path = environ.get(
            "MEDIAPIPE_MODEL_PATH", "/app/models/language_detector.tflite"
        )
        print(f"Initializing MediaPipe detector with model: {model_path}")
        _mediapipe_detector = MediaPipeDetector(model_path)
        print("✓ MediaPipe detector initialized")
    return _mediapipe_detector


def get_lingua_detector() -> LinguaDetector:
    """Get or initialize Lingua detector"""
    global _lingua_detector
    if _lingua_detector is None:
        print("Initializing Lingua detector...")
        _lingua_detector = LinguaDetector()
        print("✓ Lingua detector initialized")
    return _lingua_detector


@get("/health", summary="Health check endpoint", include_in_schema=False)
async def health() -> dict:
    """Health check"""
    return {"status": "ok"}


@post(
    "/detect",
    summary="Detect language of text",
    description="Detect the language of provided text using MediaPipe (default) or Lingua detector",
    status_code=HTTP_200_OK,
)
async def detect(data: DetectRequest) -> DetectResponse:
    """Detect language of text"""
    start_time = time.time()

    if not data.text or not data.text.strip():
        raise EmptyTextException()

    # Truncate text for logging (max 50 chars)
    text_preview = data.text[:50] + "..." if len(data.text) > 50 else data.text

    if data.method == "mediapipe":
        detector = get_mediapipe_detector()
        result = detector.detect(data.text)
        response = DetectResponse(
            language=result.language, confidence=result.confidence, method="mediapipe"
        )
    elif data.method == "lingua":
        detector = get_lingua_detector()
        result = detector.detect(data.text)
        response = DetectResponse(
            language=result.language, confidence=result.confidence, method="lingua"
        )
    else:
        raise EmptyTextException()

    # Calculate execution time
    execution_time = (time.time() - start_time) * 1000  # Convert to milliseconds

    # Log request info
    confidence_str = f"{response.confidence:.2f}" if response.confidence is not None else "None"
    logger.info(
        f"[{execution_time:.0f}ms] input=\"{text_preview}\" -> {response.language}-{confidence_str},method={response.method}"
    )

    return response


@get(["/", "/docs"], include_in_schema=False)
async def docs() -> Redirect:
    """Redirect to API documentation"""
    return Redirect("/schema")


@get(
    "/detect/languages",
    summary="Get supported languages",
    description="Get list of all supported languages with their detection methods",
    status_code=HTTP_200_OK,
)
async def get_languages() -> list[LanguageInfo]:
    """Get list of all supported languages"""
    langs = get_all_supported_languages()
    return [LanguageInfo(**lang) for lang in langs]


app = Litestar(
    [health, detect, docs, get_languages],
    openapi_config=OpenAPIConfig(
        title="Language Detection API",
        version="1.0.0",
        root_schema_site="swagger",
        description="Detect language of text using MediaPipe or Lingua",
    ),
    cors_config=CORSConfig(),
)