Spaces:
Paused
Paused
File size: 4,305 Bytes
1c2ad4f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 | from datetime import datetime, timezone
from fastapi import APIRouter, Request
from groq import Groq, RateLimitError
from pydantic import BaseModel
from src.api.utils import (
check_user_rate_limit,
get_groq_keys,
track_key_usage,
user_request_tracker,
verify_user_license,
)
from src.common.logger import logger
from src.common.utils import response_error, response_success
router = APIRouter()
class TranslationRequest(BaseModel):
email: str
license_key: str
texts: list[str]
source_language: str = "auto"
target_language: str = "en"
custom_prompt: str = None
@router.post("/translate")
async def translate_texts(req: TranslationRequest, request: Request):
MODEL = "openai/gpt-oss-120b"
_, error = verify_user_license(req.email, req.license_key)
if error:
return response_error(error, f"License verification failed: {error}", 403)
if not check_user_rate_limit(req.email):
remaining = user_request_tracker[req.email]["reset_at"] - datetime.now(
timezone.utc
)
return response_error(
"USER_RATE_LIMIT",
f"Rate limit exceeded. Try again in {int(remaining.total_seconds())} seconds",
429,
)
if not req.texts:
return response_error("INVALID_INPUT", "texts array cannot be empty", 400)
api_keys = get_groq_keys()
if not api_keys:
return response_error("NO_API_KEYS", "No Groq API keys configured", 503)
logger.info(f"Translation request received: {req.email}, {len(req.texts)} texts")
base_instruction = (
"You are a professional translator. First, correct any typos or grammatical errors in the input, "
"then translate it naturally, accurately, and contextually. Do not explain unless asked."
)
if req.custom_prompt:
base_instruction += f" {req.custom_prompt}"
numbered_texts = "\n".join(f"{i+1}. {t}" for i, t in enumerate(req.texts))
if req.source_language == "auto":
prompt = (
f"{base_instruction}\n\n"
f"Translate the following texts to {req.target_language}.\n"
f"Return exactly {len(req.texts)} lines, numbered.\n\n"
f"{numbered_texts}"
)
else:
prompt = (
f"{base_instruction}\n\n"
f"Translate from {req.source_language} to {req.target_language}.\n"
f"Return exactly {len(req.texts)} lines, numbered.\n\n"
f"{numbered_texts}"
)
last_error = None
max_retries = 3
for retry in range(max_retries):
for i, api_key in enumerate(api_keys):
try:
client = Groq(api_key=api_key)
completion = client.chat.completions.create(
model=MODEL,
messages=[
{"role": "system", "content": base_instruction},
{"role": "user", "content": prompt},
],
temperature=1,
top_p=1,
max_completion_tokens=8192,
)
track_key_usage(api_key)
output = completion.choices[0].message.content.strip()
lines = [
l.split(". ", 1)[1] if ". " in l else l
for l in output.split("\n")
if l.strip()
]
if len(lines) != len(req.texts):
if retry < max_retries - 1:
prompt += (
f"\n\nIMPORTANT: Return EXACTLY {len(req.texts)} items."
)
continue
return response_success(
{
"original_texts": req.texts,
"translated_texts": lines,
"source_language": req.source_language,
"target_language": req.target_language,
"model": MODEL,
}
)
except RateLimitError:
continue
except Exception as e:
last_error = str(e)
continue
return response_error("TRANSLATION_FAILED", f"All keys failed: {last_error}", 500)
|