Create app.py

app.py

@@ -0,0 +1,114 @@
+# main.py
+from fastapi import FastAPI, HTTPException, Depends
+from pydantic import BaseModel
+from typing import Optional, Dict
+import redis
+import hashlib
+import json
+import torch
+from transformers import AutoModelForSeq2SeqLM, AutoTokenizer
+import asyncio
+from concurrent.futures import ThreadPoolExecutor
+from contextlib import asynccontextmanager
+
+# Configuration
+CACHE_TTL = 3600  # 1 hour default
+REDIS_URL = "redis://localhost:6379"
+DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
+
+# Global model and tokenizer
+model = None
+tokenizer = None
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    # Load model on startup
+    global model, tokenizer
+    model_name = "Helsinki-NLP/opus-mt-mul-en"  # مدل چندزبانه مثال
+    print(f"Loading model on {DEVICE}...")
+    tokenizer = AutoTokenizer.from_pretrained(model_name)
+    model = AutoModelForSeq2SeqLM.from_pretrained(model_name).to(DEVICE)
+    print("Model loaded successfully")
+    yield
+    # Cleanup on shutdown
+    if model:
+        del model
+    if tokenizer:
+        del tokenizer
+
+app = FastAPI(lifespan=lifespan)
+redis_client = redis.Redis.from_url(REDIS_URL, decode_responses=True)
+executor = ThreadPoolExecutor(max_workers=4)
+
+class TranslationRequest(BaseModel):
+    text: str
+    source_lang: str
+    target_lang: str
+
+class TranslationResponse(BaseModel):
+    translated_text: str
+    from_cache: bool
+    character_count: int
+
+def generate_cache_key(text: str, source_lang: str, target_lang: str) -> str:
+    """Generate unique cache key"""
+    key_str = f"{text}_{source_lang}_{target_lang}"
+    return hashlib.md5(key_str.encode()).hexdigest()
+
+def translate_text(text: str, source_lang: str, target_lang: str) -> str:
+    """Perform translation using Hugging Face model"""
+    # Prepare text for translation based on model requirements
+    if source_lang != "en":
+        text = f">>{target_lang}<< {text}"
+    
+    inputs = tokenizer(text, return_tensors="pt", padding=True, truncation=True).to(DEVICE)
+    
+    with torch.no_grad():
+        outputs = model.generate(**inputs, max_length=512)
+    
+    translated_text = tokenizer.decode(outputs[0], skip_special_tokens=True)
+    return translated_text
+
+@app.post("/translate", response_model=TranslationResponse)
+async def translate(request: TranslationRequest):
+    # Check cache first
+    cache_key = generate_cache_key(request.text, request.source_lang, request.target_lang)
+    cached_result = redis_client.get(cache_key)
+    
+    if cached_result:
+        return TranslationResponse(
+            translated_text=cached_result,
+            from_cache=True,
+            character_count=len(request.text)
+        )
+    
+    # Perform translation
+    try:
+        # Run translation in thread pool to avoid blocking
+        translated_text = await asyncio.get_event_loop().run_in_executor(
+            executor, 
+            translate_text, 
+            request.text, 
+            request.source_lang, 
+            request.target_lang
+        )
+        
+        # Cache the result
+        redis_client.setex(cache_key, CACHE_TTL, translated_text)
+        
+        return TranslationResponse(
+            translated_text=translated_text,
+            from_cache=False,
+            character_count=len(request.text)
+        )
+        
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Translation error: {str(e)}")
+
+@app.get("/health")
+async def health_check():
+    return {"status": "healthy", "device": DEVICE}
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8000)