nickpons666/traduccion_bots
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: optimizar caché de traducción con hashes SHA256 y normalización de texto para estabilidad en producción

Browse Source
This commit is contained in:
nickpons666
2026-03-06 22:11:29 -06:00
parent ad0e80b15c
commit 6599dfcc23
4 changed files with 148 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
botdiscord/config.py
View File

@@ -34,7 +34,7 @@ def load_config(config_path: str = None) -> dict:
"port": 3306,
"user": "",
"password": "",
"name": "mi_red"
"name": ""
},
"languages": {
"enabled": [

142
botdiscord/database.py
View File

@@ -11,19 +11,42 @@ def get_connection():
db_type = get_db_type()
if db_type == "mysql":
if _connection is None or not _connection.is_connected():
db_config = get_db_config()
db_config = get_db_config()
# Si no hay conexión o no está activa, intentamos conectar/reconectar
if _connection is None:
try:
_connection = mysql.connector.connect(
host=db_config.get("host", "localhost"),
port=db_config.get("port", 3306),
user=db_config.get("user", "root"),
password=db_config.get("password", ""),
database=db_config.get("name", "mi_red")
host=db_config.get("host"),
port=db_config.get("port"),
user=db_config.get("user"),
password=db_config.get("password"),
database=db_config.get("name"),
consume_results=True,
connection_timeout=5
)
except MySQLError as e:
print(f"Error connecting to MySQL: {e}")
print(f"CRITICAL: Could not establish initial MySQL connection: {e}")
raise
# Verificar salud de la conexión existente
try:
_connection.ping(reconnect=True, attempts=3, delay=1)
except MySQLError:
try:
_connection = mysql.connector.connect(
host=db_config.get("host"),
port=db_config.get("port"),
user=db_config.get("user"),
password=db_config.get("password"),
database=db_config.get("name"),
consume_results=True,
connection_timeout=5
)
except MySQLError as e:
print(f"CRITICAL: MySQL reconnection failed: {e}")
raise
return _connection
else:
import os
@@ -107,16 +130,34 @@ def init_db():
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (message_id) REFERENCES messages(message_id) ON DELETE CASCADE,
UNIQUE KEY idx_msg_lang (message_id, target_lang)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4''')
# Tabla para traducciones de la interfaz web
cursor.execute('''CREATE TABLE IF NOT EXISTS ui_translations
(id INT AUTO_INCREMENT PRIMARY KEY,
text_hash VARCHAR(64) NOT NULL,
original_text TEXT NOT NULL,
target_lang VARCHAR(10) NOT NULL,
translated_text TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE KEY idx_ui_lang (original_text(255), target_lang)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4''')
UNIQUE KEY idx_ui_hash_lang (text_hash, target_lang)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4''')
# Migración automática: Añadir text_hash si no existe
try:
cursor.execute("SHOW COLUMNS FROM ui_translations LIKE 'text_hash'")
if not cursor.fetchone():
print("Migrating ui_translations: Adding text_hash column...")
# Primero borramos el índice viejo si existe (basado en el código anterior)
try:
cursor.execute("ALTER TABLE ui_translations DROP INDEX idx_ui_lang")
except: pass
# Añadimos la columna y el nuevo índice
cursor.execute("ALTER TABLE ui_translations ADD COLUMN text_hash VARCHAR(64) NOT NULL AFTER id")
cursor.execute("ALTER TABLE ui_translations ADD UNIQUE KEY idx_ui_hash_lang (text_hash, target_lang)")
# Limpiamos la tabla para evitar conflictos de hash con datos viejos
cursor.execute("DELETE FROM ui_translations")
except Exception as e:
print(f"Migration warning: {e}")
conn.commit()
# Tabla para administradores del panel web
cursor.execute('''CREATE TABLE IF NOT EXISTS admins
(id INT AUTO_INCREMENT PRIMARY KEY,
@@ -412,37 +453,64 @@ def get_cached_translation(message_id: int, target_lang: str) -> str:
conn.close()
return row[0] if row else None
import hashlib
def _normalize_text(text: str) -> str:
"""Normaliza el texto para que el caché sea consistente."""
if not text: return ""
# Quitar espacios en extremos, normalizar saltos de línea y quitar espacios dobles
text = text.strip().replace('\r\n', '\n')
return " ".join(text.split())
def _get_text_hash(text: str) -> str:
"""Genera un hash único para el texto normalizado."""
norm = _normalize_text(text)
return hashlib.sha256(norm.encode('utf-8')).hexdigest()
def get_ui_translation(text: str, target_lang: str) -> str:
db_type = get_db_type()
if db_type == "mysql":
conn = get_connection()
cursor = conn.cursor()
cursor.execute("SELECT translated_text FROM ui_translations WHERE original_text = %s AND target_lang = %s", (text, target_lang))
row = cursor.fetchone()
cursor.close()
return row[0] if row else None
else:
conn = get_connection()
c = conn.cursor()
c.execute("SELECT translated_text FROM ui_translations WHERE original_text = ? AND target_lang = ?", (text, target_lang))
row = c.fetchone()
conn.close()
return row[0] if row else None
text_hash = _get_text_hash(text)
try:
if db_type == "mysql":
conn = get_connection()
cursor = conn.cursor()
cursor.execute("SELECT translated_text FROM ui_translations WHERE text_hash = %s AND target_lang = %s", (text_hash, target_lang))
row = cursor.fetchone()
cursor.close()
return row[0] if row else None
else:
conn = get_connection()
c = conn.cursor()
c.execute("SELECT translated_text FROM ui_translations WHERE text_hash = ? AND target_lang = ?", (text_hash, target_lang))
row = c.fetchone()
conn.close()
return row[0] if row else None
except Exception as e:
print(f"Database error in get_ui_translation: {e}")
return None
def save_ui_translation(text: str, target_lang: str, translated_text: str):
db_type = get_db_type()
if db_type == "mysql":
conn = get_connection()
cursor = conn.cursor()
cursor.execute("INSERT INTO ui_translations (original_text, target_lang, translated_text) VALUES (%s, %s, %s) ON DUPLICATE KEY UPDATE translated_text = %s", (text, target_lang, translated_text, translated_text))
conn.commit()
cursor.close()
else:
conn = get_connection()
c = conn.cursor()
c.execute("INSERT OR REPLACE INTO ui_translations (original_text, target_lang, translated_text) VALUES (?, ?, ?)", (text, target_lang, translated_text))
conn.commit()
conn.close()
text_hash = _get_text_hash(text)
try:
if db_type == "mysql":
conn = get_connection()
cursor = conn.cursor()
query = """INSERT INTO ui_translations (text_hash, original_text, target_lang, translated_text)
VALUES (%s, %s, %s, %s)
ON DUPLICATE KEY UPDATE translated_text = %s"""
cursor.execute(query, (text_hash, text, target_lang, translated_text, translated_text))
conn.commit()
cursor.close()
else:
conn = get_connection()
c = conn.cursor()
c.execute("INSERT OR REPLACE INTO ui_translations (text_hash, original_text, target_lang, translated_text) VALUES (?, ?, ?, ?)",
(text_hash, text, target_lang, translated_text))
conn.commit()
conn.close()
except Exception as e:
print(f"Database error in save_ui_translation: {e}")
# Funciones para administradores
def get_admins():

35
botdiscord/translate.py
View File

@@ -94,24 +94,29 @@ async def translate_text(text: str, target_lang: str) -> str:
return "".join(translated_segments)
def translate_text_sync(text: str, target_lang: str) -> str:
"""Versión síncrona de translate_text para usar en filtros de Jinja2."""
"""Versión síncrona de translate_text utilizando un hilo separado."""
if not text or not target_lang or target_lang == "es":
return text
try:
import nest_asyncio
nest_asyncio.apply()
except ImportError:
# Si no está instalado, no podemos usar el bridge en FastAPI
return text
try:
loop = asyncio.get_event_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
return loop.run_until_complete(translate_text(text, target_lang))
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)