139 lines
4.4 KiB
Python
139 lines
4.4 KiB
Python
import aiohttp
|
|
import re
|
|
import asyncio
|
|
from botdiscord.config import get_libretranslate_url, get_languages
|
|
from botdiscord.database import get_available_languages, get_bot_languages
|
|
|
|
def load_lang_mappings(bot_type: str = None):
|
|
global LANG_MAPPING, REVERSE_MAPPING, FLAG_MAPPING, _cached_bot_type, NAME_TO_CODE
|
|
|
|
if bot_type:
|
|
_cached_bot_type = bot_type
|
|
|
|
available = get_available_languages()
|
|
|
|
if not available:
|
|
from botdiscord.config import get_languages
|
|
available = get_languages()
|
|
print(f"[DEBUG] Idiomas desde config: {available}")
|
|
|
|
all_codes = [lang["code"] for lang in available]
|
|
print(f"[DEBUG] Códigos disponibles: {all_codes}")
|
|
|
|
if _cached_bot_type:
|
|
active_codes = get_bot_languages(_cached_bot_type)
|
|
print(f"[DEBUG] Códigos activos para {_cached_bot_type}: {active_codes}")
|
|
if not active_codes:
|
|
active_codes = all_codes
|
|
else:
|
|
active_codes = all_codes
|
|
|
|
if not active_codes:
|
|
active_codes = all_codes
|
|
|
|
name_to_code = {lang["name"]: lang["code"] for lang in available if lang["code"] in active_codes}
|
|
code_to_name = {lang["code"]: lang["name"] for lang in available if lang["code"] in active_codes}
|
|
flag_dict = {lang["code"]: lang.get("flag", "") for lang in available}
|
|
|
|
print(f"[DEBUG] FLAG_MAPPING: {flag_dict}")
|
|
print(f"[DEBUG] NAME_TO_CODE: {name_to_code}")
|
|
|
|
LANG_MAPPING = code_to_name
|
|
NAME_TO_CODE = name_to_code
|
|
FLAG_MAPPING = flag_dict
|
|
REVERSE_MAPPING = code_to_name
|
|
|
|
_cached_bot_type = None
|
|
LANG_MAPPING = {}
|
|
REVERSE_MAPPING = {}
|
|
FLAG_MAPPING = {}
|
|
NAME_TO_CODE = {}
|
|
|
|
async def _do_translate_request(session, url, text, target_code):
|
|
"""Función interna para realizar una única petición de traducción."""
|
|
# Verificamos si el segmento tiene al menos una letra de cualquier idioma
|
|
if not text.strip() or not re.search(r'[a-zA-Z\u00C0-\u017F]', text):
|
|
return text
|
|
|
|
payload = {
|
|
"q": text,
|
|
"source": "auto",
|
|
"target": target_code,
|
|
"format": "html"
|
|
}
|
|
|
|
try:
|
|
async with session.post(url, json=payload, timeout=10) as resp:
|
|
if resp.status == 200:
|
|
data = await resp.json()
|
|
translated = data.get("translatedText", text)
|
|
return translated
|
|
else:
|
|
return text
|
|
except Exception:
|
|
return text
|
|
|
|
async def translate_text(text: str, target_lang: str) -> str:
|
|
url = get_libretranslate_url()
|
|
if not url:
|
|
return text
|
|
|
|
target_code = NAME_TO_CODE.get(target_lang, target_lang)
|
|
|
|
# Segmentación mejorada
|
|
segments = re.split(r'([.!?]+\s*|\n+)', text)
|
|
segments = [s for s in segments if s]
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
tasks = []
|
|
for segment in segments:
|
|
tasks.append(_do_translate_request(session, url, segment, target_code))
|
|
|
|
translated_segments = await asyncio.gather(*tasks)
|
|
|
|
return "".join(translated_segments)
|
|
|
|
def translate_text_sync(text: str, target_lang: str) -> str:
|
|
"""Versión síncrona de translate_text utilizando un hilo separado."""
|
|
if not text or not target_lang or target_lang == "es":
|
|
return text
|
|
|
|
import threading
|
|
|
|
result = []
|
|
def target():
|
|
# Creamos un nuevo bucle de eventos exclusivo para este hilo
|
|
new_loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(new_loop)
|
|
try:
|
|
res = new_loop.run_until_complete(translate_text(text, target_lang))
|
|
result.append(res)
|
|
finally:
|
|
new_loop.close()
|
|
|
|
# Ejecutamos la traducción en un hilo aparte para no interferir con el loop de FastAPI
|
|
thread = threading.Thread(target=target)
|
|
thread.start()
|
|
thread.join(timeout=15) # Esperamos máximo 15 segundos
|
|
|
|
return result[0] if result else text
|
|
|
|
def get_lang_mapping(bot_type: str = None) -> dict:
|
|
load_lang_mappings(bot_type)
|
|
return LANG_MAPPING.copy()
|
|
|
|
def get_reverse_mapping(bot_type: str = None) -> dict:
|
|
load_lang_mappings(bot_type)
|
|
return REVERSE_MAPPING.copy()
|
|
|
|
def get_flag_mapping(bot_type: str = None) -> dict:
|
|
load_lang_mappings(bot_type)
|
|
return FLAG_MAPPING.copy()
|
|
|
|
def get_name_to_code(bot_type: str = None) -> dict:
|
|
load_lang_mappings(bot_type)
|
|
return NAME_TO_CODE.copy()
|
|
|
|
def get_lang_flag(lang_code: str) -> str:
|
|
return FLAG_MAPPING.get(lang_code, "")
|