app.py

"""Main app file for the translation service"""
from contextlib import asynccontextmanager
from typing import List, Tuple

import asyncio
import httpx
from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse, FileResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel

from translate import translate_batch_async

class TranslateRequest(BaseModel):
    sourceLangCode: str
    targetLangCode: str
    textArray: List[str]

active_translation_requests: List[Tuple[str, str, List[str], asyncio.Future]] = [
]  # a list to store translation requests


async def process_translation(
    src_lang: str, tgt_lang: str, textArr: List[str]
) -> List[str]:
    future = asyncio.get_running_loop().create_future()
    active_translation_requests.append(
        (src_lang, tgt_lang, textArr, future))
    result = await future
    return result

async def batch_translate():
    """Background task that processes queued translation requests."""
    global active_translation_requests

    while True:
        if active_translation_requests:
            requestsProcessingNow = len(active_translation_requests)
            results = await translate_batch_async(
                [src for src, _, _, _ in active_translation_requests],
                [tgt for _, tgt, _, _ in active_translation_requests],
                [text for _, _, text, _ in active_translation_requests]
            )

            for i, (_, _, _, future) in enumerate(active_translation_requests[:requestsProcessingNow]):
                future.set_result(results[i])
            # remove the processed tasks:
            active_translation_requests = active_translation_requests[requestsProcessingNow:]
        else:
            await asyncio.sleep(0.1)


@asynccontextmanager
async def lifespan(app_instance: FastAPI):
    """Lifespan context manager for startup and shutdown events."""
    # Startup: Launch the batch translation background task
    asyncio.create_task(batch_translate())
    yield
    # Shutdown: Clean up (if needed in the future)


# Define a FastAPI app with lifespan
app = FastAPI(lifespan=lifespan)

@app.exception_handler(Exception)
async def exception_handler(request, exc):
    """Handle exceptions"""
    error_data = {
        'status': 'failure',
        'error': 'An unexpected error occurred'
    }
    return JSONResponse(content=error_data, status_code=500)


# Serve static files
app.mount("/static", StaticFiles(directory="static"), name="static")

@app.get("/")
def index():
    """Serve UI page"""
    return FileResponse("templates/index.html")

@app.get("/favicon.svg")
def favicon():
    "Serve favicon"
    return FileResponse("static/favicon.svg", media_type="image/svg+xml")

@app.post("/api/translate")
async def api_translate(data: TranslateRequest):
    """
    Endpoint to handle batch translation requests.
    """
    if data.targetLangCode == 'bn_NG':
        # Bini endpoint uses en_XX for English source instead of eng_Latn
        if data.sourceLangCode == 'eng_Latn':
            data.sourceLangCode = 'en_XX'
            
        # Route Bini requests to the external endpoint
        try:
            async with httpx.AsyncClient() as client:
                response = await client.post(
                    "https://dalaone-bini.hf.space/api/translate",
                    json=data.dict(),
                    timeout=30.0
                )
                response.raise_for_status()
                return response.json()
        except httpx.HTTPError as e:
            raise HTTPException(status_code=502, detail=f"Error proxying request to Bini endpoint: {str(e)}")

    try:
        task = asyncio.create_task(process_translation(
            data.sourceLangCode,
            data.targetLangCode,
            data.textArray
        ))
        
        result = await task
        
        return {
            "status": "success",
            "data": {
                "translationArray": result
            }
        }
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))