# -*- coding: utf-8 -*- """ Centralita Torá — HuggingFace Space (Gradio) Leer hebreo real + español fluido con exégesis de Groq y audio optimizado """ import os, glob, json, tempfile, html, asyncio, subprocess, shutil, re, time import urllib.request import urllib.error import gradio as gr import edge_tts try: from gtts import gTTS except Exception: gTTS = None from gradio_theme_fenix import fenix_theme, FENIX_CSS # --- CONFIGURACIÓN DE VOZ Y AUDIO --- VOZ_DIVINA = "es-ES-AlvaroNeural" VOZ_NARRADOR = "es-ES-AlvaroNeural" # Perillas de entonación para Narrador en Español VOZ_RATE = "-6%" VOZ_PITCH = "-22Hz" # Fondo musical FONDO = "fondo.mp3" FONDO_VOL = 0.18 # IA de Estudio y Refinamiento (Groq) MODELO_ESTUDIA = "openai/gpt-oss-120b" CARPETA = "libros" GROQ_URL = "https://api.groq.com/openai/v1/chat/completions" GROQ_KEY = os.environ.get("GROQ_API_KEY") def _limpiar_entidades(s: str) -> str: """Deshace entidades HTML aunque vengan DOBLEMENTE escapadas (p.ej. '&thinsp;' -> ' '), que es lo que rompía el hebreo en pantalla mostrando literalmente ' '.""" prev = None out = s or "" for _ in range(3): if out == prev: break prev = out out = html.unescape(out) return out # --- TRADUCCIÓN CON SENTIDO (GROQ), VERSÍCULO POR VERSÍCULO --- # Groq traduce DESDE EL HEBREO (fuente segura, siempre presente) a un español # fluido y con sentido tradicional, devolviendo UNA LÍNEA POR VERSÍCULO. Así el # mismo texto con sentido se muestra alineado y se usa EXACTAMENTE en el audio. def _traducir_lote_groq(pares): """ Traduce UN LOTE pequeño de versículos (para no chocar con el límite de tokens por minuto de Groq). pares: lista de (n, hebreo, es_literal). Devuelve (dict {n: español_con_sentido}, estado). """ fallback = {n: (es or he) for (n, he, es) in pares} if not pares: return fallback, "sin_pares" if not GROQ_KEY: return fallback, "sin_key" # Bloque de entrada numerado: el hebreo es la fuente; el literal español, # si existe, va solo como ayuda. lineas_in = [] for (n, he, es) in pares: ayuda = f" (ayuda literal: {es})" if es else "" lineas_in.append(f"[{n}] {he}{ayuda}") entrada = "\n".join(lineas_in) sistema = ( "Eres un experto en traducción bíblica del hebreo y en exégesis tradicional. " "Traduces cada versículo hebreo a un español fluido, claro y natural, " "con el sentido tradicional exacto, sin perder fidelidad ni profundidad. " "Reglas de salida ESTRICTAS: devuelve SOLO las traducciones, una por línea, " "cada línea empezando por el número entre corchetes tal cual: '[N] traducción'. " "No incluyas el texto hebreo, ni títulos, ni comentarios, ni notas. " "Conserva el mismo número de versículos que recibas." ) max_tok = min(2000, max(400, len(pares) * 90)) cuerpo = { "model": MODELO_ESTUDIA, "temperature": 0.3, "max_tokens": max_tok, "messages": [ {"role": "system", "content": sistema}, {"role": "user", "content": f"Traduce con sentido cada versículo:\n\n{entrada}"}, ], } try: data = json.dumps(cuerpo).encode("utf-8") req = urllib.request.Request(GROQ_URL, data=data, method="POST") req.add_header("Content-Type", "application/json") req.add_header("Authorization", f"Bearer {GROQ_KEY}") req.add_header("User-Agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0 Safari/537.36") with urllib.request.urlopen(req, timeout=120) as r: d = json.load(r) salida = (d["choices"][0]["message"]["content"] or "").strip() except urllib.error.HTTPError as e: cuerpo_error = "" try: cuerpo_error = e.read().decode("utf-8", errors="replace")[:300] except Exception: pass motivo = f"error HTTP {e.code}: {cuerpo_error or e.reason}" print(f"Error traduciendo con Groq: {motivo}") return fallback, motivo except Exception as e: motivo = f"error: {type(e).__name__}: {e}" print(f"Error traduciendo con Groq: {motivo}") return fallback, motivo if not salida: return fallback, "respuesta vacía de Groq" # Parseamos las líneas '[N] texto' a un diccionario por versículo. res = {} for linea in salida.splitlines(): m = re.match(r"\s*\[?(\d+)\]?[\.\)\|:\-]?\s*(.+)$", linea.strip()) if m: res[int(m.group(1))] = m.group(2).strip() if not res: return fallback, "no se pudo interpretar la respuesta de Groq" # Si Groq se saltó algún versículo, se completa con el fallback. final = {n: res.get(n, fallback[n]) for (n, he, es) in pares} faltantes = sum(1 for (n, he, es) in pares if n not in res) estado = "ok" if faltantes == 0 else f"ok (Groq omitió {faltantes} versículo(s), se completó con literal)" return final, estado # Tamaño de lote: capítulos largos (p.ej. Génesis 1, 31 versículos) superan el # límite de tokens por minuto de Groq si se mandan de una vez. Se trocea en # bloques pequeños y se hacen varias llamadas, uniendo los resultados. TAMANO_LOTE_GROQ = 6 def _segundos_de_espera(motivo_error): """Si el error es un 429 con 'Please try again in X s', devuelve X (+1s de margen).""" m = re.search(r"try again in ([\d.]+)s", motivo_error) if m: try: return float(m.group(1)) + 1.0 except ValueError: pass return None def traducir_capitulo_con_groq(pares): """ pares: lista de (n, hebreo, es_literal) de TODO el capítulo. Trocea en lotes pequeños, traduce cada uno con Groq y une los resultados. Si Groq responde 429 (límite de tokens por minuto), espera el tiempo que Groq indica y reintenta ese lote (hasta 3 intentos) antes de rendirse. Devuelve (dict {n: español_con_sentido}, estado_general). """ if not pares: return {}, "sin_pares" resultado = {} lotes_ok = 0 lotes_total = 0 motivos_fallo = [] for i in range(0, len(pares), TAMANO_LOTE_GROQ): lote = pares[i:i + TAMANO_LOTE_GROQ] lotes_total += 1 parcial, estado_lote = _traducir_lote_groq(lote) intentos = 1 while "error HTTP 429" in estado_lote and intentos < 3: espera = _segundos_de_espera(estado_lote) or 8.0 time.sleep(espera) parcial, estado_lote = _traducir_lote_groq(lote) intentos += 1 resultado.update(parcial) if estado_lote == "ok" or estado_lote.startswith("ok ("): lotes_ok += 1 else: motivos_fallo.append(f"versículos {lote[0][0]}-{lote[-1][0]}: {estado_lote}") if lotes_ok == lotes_total: estado_general = "ok" elif lotes_ok == 0: estado_general = "; ".join(motivos_fallo[:3]) else: estado_general = ( f"parcial ({lotes_ok}/{lotes_total} lotes con Groq, el resto literal) — " + "; ".join(motivos_fallo[:3]) ) return resultado, estado_general # --- PROCESAMIENTO ACÚSTICO FFMPEG --- def procesar_audio_narrador(ruta): if not shutil.which("ffmpeg"): return ruta filtros = [ "atempo=0.92", "bass=g=5:f=120", "treble=g=2", "aecho=0.8:0.88:40:0.15" ] salida = ruta[:-4] + "_narrador.mp3" cmd = ["ffmpeg", "-y", "-i", ruta, "-af", ", ".join(filtros), "-ac", "2", salida] try: subprocess.run(cmd, check=True, capture_output=True, timeout=60) if os.path.getsize(salida) > 0: return salida except Exception: pass return ruta def procesar_audio_divino(ruta): if not shutil.which("ffmpeg"): return ruta filtros = [ "atempo=0.70", # Más pausado y solemne "asetrate=44100*0.55", # Bajar tono (más grave y imponente) "aresample=144000", # Reajustar sample rate "bass=g=15:f=200", # Graves profundos sub-bass "treble=g=-15", # Sonido cálido, sin aristas agudas ] salida = ruta[:-4] + "_divino.mp3" cmd = ["ffmpeg", "-y", "-i", ruta, "-af", ", ".join(filtros), "-ac", "2", salida] try: subprocess.run(cmd, check=True, capture_output=True, timeout=60) if os.path.getsize(salida) > 0: return salida except Exception: pass return ruta def concatenar_audios(lista_rutas): if not lista_rutas: return None if len(lista_rutas) == 1: return lista_rutas[0] salida_final = tempfile.NamedTemporaryFile(suffix=".mp3", delete=False).name list_file = tempfile.NamedTemporaryFile(suffix=".txt", mode="w", delete=False, encoding="utf-8") for r in lista_rutas: list_file.write(f"file '{os.path.abspath(r)}'\n") list_file.close() cmd = ["ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", list_file.name, "-c", "copy", salida_final] try: subprocess.run(cmd, check=True, capture_output=True, timeout=120) os.remove(list_file.name) return salida_final except Exception: if os.path.exists(list_file.name): os.remove(list_file.name) return lista_rutas[0] def mezclar_fondo(voz_path): if not (os.path.exists(FONDO) and shutil.which("ffmpeg")): return voz_path salida = voz_path[:-4] + "_mix.mp3" filtro = (f"[1:a]volume={FONDO_VOL}[m];" f"[0:a][m]amix=inputs=2:duration=first:normalize=0[a]") cmd = ["ffmpeg", "-y", "-i", voz_path, "-stream_loop", "-1", "-i", FONDO, "-filter_complex", filtro, "-map", "[a]", salida] try: subprocess.run(cmd, check=True, capture_output=True, timeout=120) if os.path.getsize(salida) > 0: return salida except Exception: pass return voz_path # --- SEPARACIÓN DE BLOQUES (NARRADOR vs DIOS) PARA ESPAÑOL --- def segmentar_texto_divino(texto): patron = r'((?:Y\s+dijo\s+Dios|dijo\s+Dios|Y\s+llamó\s+Dios|Y\s+bendijo\s+Dios)[^:,.–—]*[:,.–—]?\s*)([^.\n]+)' bloques = [] ultimo_idx = 0 for match in re.finditer(patron, texto, flags=re.IGNORECASE): start, end = match.span() if start > ultimo_idx: bloques.append(("narrador", texto[ultimo_idx:start])) intro_dios = match.group(1) palabras_dios = match.group(2) bloques.append(("narrador", intro_dios)) bloques.append(("dios", palabras_dios)) ultimo_idx = end if ultimo_idx < len(texto): bloques.append(("narrador", texto[ultimo_idx:])) return bloques if bloques else [("narrador", texto)] async def generar_bloque_audio(texto, es_divino): ruta = tempfile.NamedTemporaryFile(suffix=".mp3", delete=False).name voz = VOZ_DIVINA if es_divino else VOZ_NARRADOR pitch = "-5Hz" if es_divino else VOZ_PITCH rate = "-9%" if es_divino else VOZ_RATE try: com = edge_tts.Communicate(texto[:4000], voz, rate=rate, pitch=pitch) await com.save(ruta) except Exception: return None if es_divino: return procesar_audio_divino(ruta) else: return procesar_audio_narrador(ruta) async def leer_es(texto): if not texto or not texto.strip(): return None bloques = segmentar_texto_divino(texto) audios_segmentos = [] for tipo, contenido in bloques: if contenido.strip(): es_divino = (tipo == "dios") audio_seg = await generar_bloque_audio(contenido, es_divino) if audio_seg: audios_segmentos.append(audio_seg) audio_final = concatenar_audios(audios_segmentos) if audio_final: audio_final = mezclar_fondo(audio_final) return audio_final # --- REPRODUCTOR DE HEBREO LITÚRGICO REAL (TROPE) --- async def leer_he(nombre_espanol, capitulo): libro_id = id_por_nombre(nombre_espanol) if not libro_id: return None MAPEO_TORAH = { "Genesis": "01", "Exodus": "02", "Leviticus": "03", "Numbers": "04", "Deuteronomy": "05" } carpeta_audios = "audios_torah" if not os.path.exists(carpeta_audios): os.makedirs(carpeta_audios) ruta_local = os.path.join(carpeta_audios, f"{libro_id}_{capitulo}.mp3") if os.path.exists(ruta_local): return mezclar_fondo(ruta_local) if libro_id in MAPEO_TORAH: id_mechon = MAPEO_TORAH[libro_id] cap_formateado = f"{int(capitulo):02d}" url_audio = f"https://mechon-mamre.org/mp3/t{id_mechon}{cap_formateado}.mp3" try: req = urllib.request.Request( url_audio, headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"} ) with urllib.request.urlopen(req, timeout=30) as response, open(ruta_local, 'wb') as out_file: out_file.write(response.read()) return mezclar_fondo(ruta_local) except Exception as e: print(f"Error descargando el audio: {e}") return None return None # --- CARGA Y MANEJO DE LIBROS --- NOMBRES_ES = { "Genesis": "Génesis", "Exodus": "Éxodo", "Leviticus": "Levítico", "Numbers": "Números", "Deuteronomy": "Deuteronomio", "Joshua": "Josué", "Judges": "Jueces", "Ruth": "Rut", "1_Samuel": "1 Samuel", "2_Samuel": "2 Samuel", "1_Kings": "1 Reyes", "2_Kings": "2 Reyes", "1_Chronicles": "1 Crónicas", "2_Chronicles": "2 Crónicas", "Ezra": "Esdras", "Nehemiah": "Nehemías", "Esther": "Ester", "Job": "Job", "Psalms": "Salmos", "Proverbs": "Proverbios", "Ecclesiastes": "Eclesiastés", "Song_of_Songs": "Cantar de los Cantares", "Isaiah": "Isaías", "Jeremiah": "Jeremías", "Lamentations": "Lamentaciones", "Ezekiel": "Ezequiel", "Daniel": "Daniel", "Hosea": "Oseas", "Joel": "Joel", "Amos": "Amós", "Obadiah": "Abdías", "Jonah": "Jonás", "Micah": "Miqueas", "Nahum": "Nahúm", "Habakkuk": "Habacuc", "Zephaniah": "Sofonías", "Haggai": "Hageo", "Zechariah": "Zacarías", "Malachi": "Malaquías", } def _convertir_traduccion_plana(book_id: str, data: dict) -> dict: texto = data.get("translation", "") or "" lineas = [l.strip() for l in texto.split("\n") if l.strip()] versiculos = [{"n": i + 1, "he": "", "es": linea} for i, linea in enumerate(lineas)] return { "id": book_id, "es": NOMBRES_ES.get(book_id, book_id), "heb": "", "capitulos": {"1": versiculos}, } def cargar_libros(): libros = {} for ruta in sorted(glob.glob(os.path.join(CARPETA, "*.json"))): try: with open(ruta, encoding="utf-8") as f: d = json.load(f) book_id = os.path.splitext(os.path.basename(ruta))[0] if "capitulos" in d: lid = d.get("id") or book_id libros[lid] = d elif "translation" in d: libros[book_id] = _convertir_traduccion_plana(book_id, d) except Exception as e: print("No pude leer", ruta, e) return libros LIBROS = cargar_libros() ORDEN = ["Genesis", "Exodus", "Leviticus", "Numbers", "Deuteronomy"] IDS = [i for i in ORDEN if i in LIBROS] + [i for i in LIBROS if i not in ORDEN] NOMBRES = [LIBROS[i].get("es", i) for i in IDS] or ["(sube tus libros)"] def id_por_nombre(nombre): for i in IDS: if LIBROS[i].get("es", i) == nombre: return i return IDS[0] if IDS else None def versiculos(lid, cap): return LIBROS.get(lid, {}).get("capitulos", {}).get(str(cap), []) def render(nombre, cap): lid = id_por_nombre(nombre) if not lid: return "

Aún no has subido libros a la carpeta libros/.

", "", "" try: cap = max(1, int(float(cap or 1))) except Exception: cap = 1 d = LIBROS[lid] vs = versiculos(lid, cap) if not vs: return f"

No hay texto para {html.escape(d.get('es',''))} {cap}.

", "", "" # Recopilamos (n, hebreo_limpio, español_literal) por versículo. pares = [] plano_he = [] for i, v in enumerate(vs): n_verso = v.get('n', i + 1) he_raw = _limpiar_entidades(v.get("he", "")).replace(" ", " ").replace("{פ}", "").replace("{ס}", "").strip() es_raw = _limpiar_entidades(v.get("es", "")).replace(" ", " ").strip() pares.append((n_verso, he_raw, es_raw)) if he_raw: plano_he.append(he_raw) # Groq da SENTIDO a las palabras ANTES de mostrarlas: traduce el capítulo # desde el hebreo, versículo por versículo. es_por_verso, estado_groq = traducir_capitulo_con_groq(pares) if estado_groq != "ok": print(f"Traducción con sentido — estado: {estado_groq}") # Interfaz: hebreo intacto arriba y, debajo, la traducción con sentido # (sin aviso visible; el estado queda solo en los logs del Space). filas = [f"
{html.escape(_limpiar_entidades(d.get('heb','')))}
" f"{html.escape(d.get('es',''))} {cap}
"] partes_es = [] for (n_verso, he_raw, es_raw) in pares: es_sentido = (es_por_verso.get(n_verso) or es_raw or "").strip() if es_sentido: partes_es.append(es_sentido) fila = f"
{n_verso}" if he_raw: fila += f"{html.escape(he_raw)}" if es_sentido: fila += f"{html.escape(es_sentido)}" fila += "
" filas.append(fila) # Texto que se locuta = exactamente la traducción con sentido de todo el capítulo. texto_es_final = " ".join(partes_es) return f"
{''.join(filas)}
", texto_es_final, " ".join(plano_he) def estudiar(mensaje, historial, contexto): if not GROQ_KEY: return "Para el estudio, añade el Secret GROQ_API_KEY en el Space." sistema = ( "Eres un compañero de estudio de la Torá que responde en español, cálido y honesto. " "Ofreces el sentido literal (peshat), contexto histórico y lingüístico, capas de la " "tradición (midrash, Rashi cuando venga al caso) y reflexión espiritual. Vas al grano.\n\n" + contexto ) mensajes = [{"role": "system", "content": sistema}] for m in (historial or []): if isinstance(m, dict) and m.get("role") in ("user", "assistant"): mensajes.append({"role": m["role"], "content": m["content"]}) mensajes.append({"role": "user", "content": mensaje}) cuerpo = {"model": MODELO_ESTUDIA, "temperature": 0.4, "max_tokens": 1200, "messages": mensajes} try: data = json.dumps(cuerpo).encode("utf-8") req = urllib.request.Request(GROQ_URL, data=data, method="POST") req.add_header("Content-Type", "application/json") req.add_header("Authorization", f"Bearer {GROQ_KEY}") req.add_header("User-Agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0 Safari/537.36") with urllib.request.urlopen(req, timeout=90) as r: d = json.load(r) return (d["choices"][0]["message"]["content"] or "").strip() except Exception as e: return f"No pude consultar a la IA ahora mismo ({e})." # ============================ INTERFAZ GRADIO ============================ with gr.Blocks(theme=fenix_theme(), css=FENIX_CSS, title="Centralita Torá") as demo: st_es = gr.State("") st_he = gr.State("") st_nombre = gr.State(NOMBRES[0]) st_cap = gr.State(1) gr.HTML("
" "

Centralita Torá

hebreo real e impecable · español fluido y natural

") with gr.Tab("Leer"): with gr.Row(): dd = gr.Dropdown(NOMBRES, value=NOMBRES[0], label="Libro") ncap = gr.Number(value=1, precision=0, label="Capítulo", minimum=1) ver_btn = gr.Button("Ver capítulo", variant="primary") salida = gr.HTML() with gr.Row(): b_es = gr.Button("🔊 Narrador + Voz Divina (Español)", variant="secondary") b_he = gr.Button("🔊 עברית (Hebreo con Trope Real)", variant="secondary") audio = gr.Audio(label="Audio Litúrgico", autoplay=True) def ver(nombre, cap): md, es, he = render(nombre, cap) return md, es, he, nombre, cap ver_btn.click(ver, [dd, ncap], [salida, st_es, st_he, st_nombre, st_cap]) b_es.click(leer_es, st_es, audio) b_he.click(leer_he, inputs=[st_nombre, st_cap], outputs=[audio]) with gr.Tab("Buscar"): q = gr.Textbox(label="Palabra o frase (hebreo o español)", placeholder="ej. luz / אור") q_btn = gr.Button("Buscar", variant="primary") q_out = gr.HTML() def buscar(texto): t = (texto or "").strip().lower() if not t: return "

Escribe algo para buscar.

" filas = [] for lid in IDS: d = LIBROS[lid] for c, vs in d.get("capitulos", {}).items(): for v in vs: campos = " ".join(filter(None, [v.get("es"), v.get("he")])).lower() if t in campos: ref = f"{d.get('es', lid)} {c}:{v.get('n')}" txt = html.escape(_limpiar_entidades(v.get("es") or v.get("he") or "")) filas.append(f"
{ref} — {txt}
") if len(filas) >= 120: return "".join(filas) return "".join(filas) if filas else "

Sin resultados en los libros cargados.

" q_btn.click(buscar, q, q_out) with gr.Tab("Estudiar"): gr.Markdown("La IA usa el capítulo abierto en **Leer** como contexto.") chat = gr.Chatbot(type="messages", height=360) pin = gr.Textbox(placeholder="Pregunta sobre el pasaje…", label="") with gr.Row(): enviar = gr.Button("Preguntar", variant="primary") comentar = gr.Button("Comentar este capítulo", variant="secondary") def responder(mensaje, historial, es, he, nombre, cap): mensaje = (mensaje.strip() if mensaje else "") if not mensaje: return historial, "" contexto = f"Pasaje en pantalla — {nombre} {cap}:\nHebreo: {he}\nEspañol: {es}" r = estudiar(mensaje, historial, contexto) historial = (historial or []) + [ {"role": "user", "content": mensaje}, {"role": "assistant", "content": r}, ] return historial, "" enviar.click(responder, [pin, chat, st_es, st_he, st_nombre, st_cap], [chat, pin]) comentar.click( lambda h, es, he, n, c: responder("Comenta y ayúdame a estudiar este capítulo.", h, es, he, n, c), [chat, st_es, st_he, st_nombre, st_cap], [chat, pin], ) with gr.Tab("Cómo subir"): gr.Markdown( "Sube un JSON por libro a la carpeta `libros/`. Para el estudio y traducción fluida, añade el Secret `GROQ_API_KEY`." ) if __name__ == "__main__": demo.launch(ssr_mode=False)