api/translator_app.py

"""
# 方式1：直接运行
python api/translator_app.py
# 方式2：使用 uvicorn
uvicorn api.translator_app:app --host 0.0.0.0 --port 6006 --reload
使用说明：
Translation HTTP Service
This service provides a RESTful API for text translation using Qwen (default) or DeepL API.
The service runs on port 6006 and provides a simple translation endpoint.
API Endpoint:
    POST /translate
Request Body (JSON):
    {
        "text": "要翻译的文本",
        "target_lang": "en",  # Required: target language code (zh, en, ru, etc.)
        "source_lang": "zh",  # Optional: source language code (auto-detect if not provided)
        "model": "qwen"       # Optional: translation model ("qwen" or "deepl", default: "qwen")
    }
Response (JSON):
    {
        "text": "要翻译的文本",
        "target_lang": "en",
        "source_lang": "zh",
        "translated_text": "Text to translate",
        "status": "success"
    }
Usage Examples:
1. Translate Chinese to English:
   curl -X POST http://localhost:6006/translate \
     -H "Content-Type: application/json" \
     -d '{
       "text": "商品名称",
       "target_lang": "en",
       "source_lang": "zh"
     }'
2. Translate with auto-detection:
   curl -X POST http://localhost:6006/translate \
     -H "Content-Type: application/json" \
     -d '{
       "text": "Product name",
       "target_lang": "zh"
     }'
3. Translate using DeepL model:
   curl -X POST http://localhost:6006/translate \
     -H "Content-Type: application/json" \
     -d '{
       "text": "商品名称",
       "target_lang": "en",
       "source_lang": "zh",
       "model": "deepl"
     }'
4. Translate Russian to English:
   curl -X POST http://localhost:6006/translate \
     -H "Content-Type: application/json" \
     -d '{
       "text": "Название товара",
       "target_lang": "en",
       "source_lang": "ru"
     }'
Health Check:
    GET /health
    curl http://localhost:6006/health
Start the service:
    python api/translator_app.py
    # or
    uvicorn api.translator_app:app --host 0.0.0.0 --port 6006 --reload
"""
import os
import sys
import logging
import argparse
import uvicorn
from typing import Optional, Dict
from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
# Add parent directory to path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from query.qwen_mt_translate import Translator
from query.llm_translate import LLMTranslatorProvider
from query.deepl_provider import DeepLProvider
from config.services_config import get_translation_config
# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# Global translator instances cache (keyed by model)
_translators: Dict[str, object] = {}
def _resolve_default_model() -> str:
    """
    Resolve translator model from services.translation config first.
    Priority:
    1) TRANSLATION_MODEL env (explicit runtime override)
    2) services.translation.provider + providers.<provider>.model
    3) qwen-mt
    """
    env_model = (os.getenv("TRANSLATION_MODEL") or "").strip()
    if env_model:
        return env_model
    try:
        cfg = get_translation_config()
        provider = (cfg.provider or "").strip().lower()
        provider_cfg = cfg.get_provider_cfg() if hasattr(cfg, "get_provider_cfg") else {}
        model = (provider_cfg.get("model") or "").strip().lower() if isinstance(provider_cfg, dict) else ""
        if provider == "llm":
            return "llm"
        if provider in {"qwen-mt", "direct", "http"}:
            return model or "qwen-mt"
        if provider == "deepl":
            return "deepl"
    except Exception:
        pass
    return "qwen-mt"
def get_translator(model: str = "qwen") -> object:
    """Get or create translator instance for the specified model."""
    global _translators
    if model not in _translators:
        logger.info(f"Initializing translator with model: {model}...")
        normalized = (model or "qwen").strip().lower()
        if normalized in {"qwen", "qwen-mt", "qwen-mt-flash", "qwen-mt-flush"}:
            _translators[model] = Translator(model=normalized, use_cache=True, timeout=10)
        elif normalized == "deepl":
            _translators[model] = DeepLProvider(api_key=None, timeout=10.0)
        elif normalized == "llm":
            _translators[model] = LLMTranslatorProvider()
        else:
            raise ValueError(f"Unsupported model: {model}")
        logger.info(f"Translator initialized with model: {model}")
    return _translators[model]
# Request/Response models
class TranslationRequest(BaseModel):
    """Translation request model."""
    text: str = Field(..., description="Text to translate")
    target_lang: str = Field(..., description="Target language code (zh, en, ru, etc.)")
    source_lang: Optional[str] = Field(None, description="Source language code (optional, auto-detect if not provided)")
    model: Optional[str] = Field(None, description="Translation model: qwen-mt | deepl | llm")
    context: Optional[str] = Field(None, description="Optional translation scene or context")
    prompt: Optional[str] = Field(None, description="Optional prompt override")
    class Config:
        json_schema_extra = {
            "example": {
                "text": "商品名称",
                "target_lang": "en",
                "source_lang": "zh",
                "model": "llm",
                "context": "sku_name"
            }
        }
class TranslationResponse(BaseModel):
    """Translation response model."""
    text: str = Field(..., description="Original text")
    target_lang: str = Field(..., description="Target language code")
    source_lang: Optional[str] = Field(None, description="Source language code (detected or provided)")
    translated_text: str = Field(..., description="Translated text")
    status: str = Field(..., description="Translation status")
    model: str = Field(..., description="Translation model used")
# Create FastAPI app
app = FastAPI(
    title="Translation Service API",
    description="RESTful API for text translation using Qwen (default) or DeepL",
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc"
)
# Add CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
@app.on_event("startup")
async def startup_event():
    """Initialize translator on startup."""
    logger.info("Starting Translation Service API on port 6006")
    default_model = _resolve_default_model()
    try:
        get_translator(model=default_model)
        logger.info(f"Translation service ready with default model: {default_model}")
    except Exception as e:
        logger.error(f"Failed to initialize translator: {e}", exc_info=True)
        raise
@app.get("/health")
async def health_check():
    """Health check endpoint."""
    try:
        # 仅做轻量级本地检查，避免在健康检查中触发潜在的阻塞初始化或外部依赖
        default_model = _resolve_default_model()
        # 如果启动事件成功，默认模型通常会已经初始化到缓存中
        translator = _translators.get(default_model) or next(iter(_translators.values()), None)
        return {
            "status": "healthy",
            "service": "translation",
            "default_model": default_model,
            "available_models": list(_translators.keys()),
            "translator_initialized": translator is not None,
            "cache_enabled": bool(getattr(translator, "use_cache", False))
        }
    except Exception as e:
        logger.error(f"Health check failed: {e}")
        return JSONResponse(
            status_code=503,
            content={
                "status": "unhealthy",
                "error": str(e)
            }
        )
@app.post("/translate", response_model=TranslationResponse)
async def translate(request: TranslationRequest):
    """
    Translate text to target language.
    
    Uses a fixed prompt optimized for product SKU name translation.
    The translation is cached in Redis for performance.
    
    Supports both Qwen (default) and DeepL models via the 'model' parameter.
    """
    if not request.text or not request.text.strip():
        raise HTTPException(
            status_code=400,
            detail="Text cannot be empty"
        )
    
    if not request.target_lang:
        raise HTTPException(
            status_code=400,
            detail="target_lang is required"
        )
    
    # Validate model parameter
    model = request.model.lower() if request.model else _resolve_default_model().lower()
    if model not in ["qwen", "qwen-mt", "deepl", "llm"]:
        raise HTTPException(
            status_code=400,
            detail="Invalid model. Supported models: 'qwen-mt', 'deepl', 'llm'"
        )
    
    try:
        # Get translator instance for the specified model
        translator = get_translator(model=model)
        
        # Translate using the fixed prompt
        translated_text = translator.translate(
            text=request.text,
            target_lang=request.target_lang,
            source_lang=request.source_lang,
            context=request.context,
            prompt=request.prompt,
        )
        
        if translated_text is None:
            raise HTTPException(
                status_code=500,
                detail="Translation failed"
            )
        
        return TranslationResponse(
            text=request.text,
            target_lang=request.target_lang,
            source_lang=request.source_lang,
            translated_text=translated_text,
            status="success",
            model=str(getattr(translator, "model", model))
        )
    
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Translation error: {e}", exc_info=True)
        raise HTTPException(
            status_code=500,
            detail=f"Translation error: {str(e)}"
        )
@app.get("/")
async def root():
    """Root endpoint with API information."""
    return {
        "service": "Translation Service API",
        "version": "1.0.0",
        "status": "running",
        "endpoints": {
            "translate": "POST /translate",
            "health": "GET /health",
            "docs": "GET /docs"
        }
    }
if __name__ == "__main__":
    parser = argparse.ArgumentParser(description='Start translation API service')
    parser.add_argument('--host', default='0.0.0.0', help='Host to bind to')
    parser.add_argument('--port', type=int, default=6006, help='Port to bind to')
    parser.add_argument('--reload', action='store_true', help='Enable auto-reload')
    args = parser.parse_args()
    # Run server
    uvicorn.run(
        "api.translator_app:app",
        host=args.host,
        port=args.port,
        reload=args.reload
    )