Upload centralita_app.py

centralita_app.py

@@ -0,0 +1,470 @@
+# -*- coding: utf-8 -*-
+"""
+Centralita Torá — HuggingFace Space (Gradio)
+Leer hebreo + español · escuchar (Canto Gregoriano-Hebraico con Voz Divina Especial)
+"""
+import os, glob, json, tempfile, html, asyncio, subprocess, shutil, re
+import urllib.request
+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 CANTO ---
+CANTO_GREGORIANO_HEBRAICO = True
+
+VOZ_DIVINA = "es-ES-AlvaroNeural"
+VOZ_NARRADOR = "es-ES-AlvaroNeural"  # Usamos Álvaro con procesamiento acústico sagrado especial
+
+VOZ_HE = "he-IL-AvriNeural"
+
+# Perillas de entonación para Narrador
+VOZ_RATE = "-10%"
+VOZ_PITCH = "+5Hz"
+
+# Fondo musical
+FONDO = "fondo.mp3"
+FONDO_VOL = 0.18
+
+# IA de Estudio (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")
+
+
+# --- MOTOR FONÉTICO GREGORIANO-HEBRAICO ---
+def transformar_a_canto_gregoriano_hebraico(texto_sagrado: str) -> str:
+    if not texto_sagrado:
+        return ""
+
+    texto = texto_sagrado
+
+    # Melismas gregorianos en vocales tónicas
+    melismas = {
+        'á': 'a-a-a-a', 'é': 'e-e-e-e', 'í': 'i-i-i-i',
+        'ó': 'o-o-o-o', 'ú': 'u-u-u-u',
+        'Á': 'A-a-a-a', 'É': 'E-e-e-e', 'Í': 'I-i-i-i',
+        'Ó': 'O-o-o-o', 'Ú': 'U-u-u-u',
+    }
+    for vocal_con_tilde, melisma in melismas.items():
+        texto = texto.replace(vocal_con_tilde, melisma)
+
+    # Cierres litúrgicos
+    def cierre_liturgico(match):
+        vocal = match.group(1)
+        puntuacion = match.group(2)
+        return f"{vocal}-{vocal}-{vocal}{puntuacion} ... "
+
+    texto = re.sub(r'([aeiouAEIOU])([,.:;])', cierre_liturgico, texto)
+
+    # Énfasis místico
+    texto = re.sub(r'\bDios\b', 'Di-o-o-os', texto, flags=re.IGNORECASE)
+    texto = re.sub(r'\bSeñor\b', 'Se-ñoo-or', texto, flags=re.IGNORECASE)
+    texto = re.sub(r'\bIsrael\b', 'Is-ra-e-el', texto, flags=re.IGNORECASE)
+
+    return texto
+
+
+# --- PROCESAMIENTO ACÚSTICO FFMEPG ---
+def procesar_audio_narrador(ruta):
+    """Efecto catedral estándar para el narrador."""
+    if not shutil.which("ffmpeg"):
+        return ruta
+
+    filtros = [
+        "atempo=0.22",
+        "bass=g=8:f=133",
+        "treble=g=-22",
+        "vibrato=f=3.5:d=0.55",
+        "aecho=0.75:0.70:120|240:0.4|0.25"
+    ]
+    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):
+    """
+    Efecto ACÚSTICO SAGRADO Y PROFUNDO para la Voz de Dios.
+    Tono más grave, resonancia sub-bass, eco más amplio y pureza vocal.
+    """
+    if not shutil.which("ffmpeg"):
+        return ruta
+
+    filtros = [
+        "atempo=0.80",                         # Más pausado y solemne
+        "asetrate=44100*0.45",                  # Bajar tono (más grave y imponente)
+        "aresample=144000",                      # Reajustar sample rate
+        "bass=g=15:f=200",                     # Graves profundos sub-bass
+        "treble=g=-10",                        # 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):
+    """Concatena varios archivos MP3 en uno solo en orden estricto."""
+    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) ---
+def segmentar_texto_divino(texto):
+    """
+    Detecta patrones como 'Y dijo Dios: ...' o 'Dijo Dios ...'
+    y divide el texto en partes ('narrador' o 'dios').
+    """
+    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) # Ej: "Y dijo Dios: "
+        palabras_dios = match.group(2) # Ej: "Sea la luz"
+
+        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)]
+
+
+# --- GENERACIÓN DE AUDIO DINÁMICA ---
+async def generar_bloque_audio(texto, es_divino):
+    texto_tts = transformar_a_canto_gregoriano_hebraico(texto) if CANTO_GREGORIANO_HEBRAICO else texto
+    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_tts[: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 a_voz(texto, idioma):
+    if not texto or not texto.strip():
+        return None
+
+    if idioma == "he":
+        ruta = tempfile.NamedTemporaryFile(suffix=".mp3", delete=False).name
+        try:
+            com = edge_tts.Communicate(texto[:4000], VOZ_HE, rate=VOZ_RATE, pitch=VOZ_PITCH)
+            await com.save(ruta)
+            return mezclar_fondo(ruta)
+        except Exception:
+            return None
+
+    # Procesamiento para Español con Voz Divina Diferenciada
+    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
+
+
+async def leer_es(t):
+    return await a_voz(t, "es")
+
+
+async def leer_he(t):
+    return await a_voz(t, "he")
+
+
+# --- CARGA Y MANEJO DE LIBROS ---
+# Nombres de libros en español, para mostrar lindo en el desplegable
+# aunque el archivo se llame en inglés (Genesis.json, Exodus.json, etc.)
+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:
+    """
+    Convierte el formato plano que genera el traductor
+    ({"translation": "...", "missing_strong_numbers": [...]})
+    a la estructura de capítulos/versículos que usa la Centralita.
+    No tenemos separación real de capítulos en este formato, así que
+    todo el libro se muestra como Capítulo 1, con una línea = un versículo.
+    """
+    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:
+                # Formato completo ya listo (id/es/heb/capitulos)
+                lid = d.get("id") or book_id
+                libros[lid] = d
+            elif "translation" in d:
+                # Formato plano del traductor (translation + missing_strong_numbers)
+                libros[book_id] = _convertir_traduccion_plana(book_id, d)
+            else:
+                print(f"Formato no reconocido en {ruta}, se omite.")
+        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 "<div class='pasaje'><p>Aún no has subido libros a la carpeta <b>libros/</b>.</p></div>", "", ""
+    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"<div class='pasaje'><p>No hay texto para {html.escape(d.get('es',''))} {cap}.</p></div>", "", ""
+    filas = [f"<div class='cab'><span class='h'>{html.escape(d.get('heb',''))}</span><br>"
+             f"{html.escape(d.get('es',''))} {cap}</div>"]
+    plano_es, plano_he = [], []
+    for v in vs:
+        he = html.escape(v.get("he", "")); es = html.escape(v.get("es", ""))
+        fila = f"<div class='verso'><span class='num'>{v.get('n','')}</span>"
+        if he:
+            fila += f"<span class='he'>{he}</span>"; plano_he.append(v.get("he", ""))
+        if es:
+            fila += f"<span class='es'>{es}</span>"; plano_es.append(v.get("es", ""))
+        fila += "</div>"
+        filas.append(fila)
+    return f"<div class='pasaje'>{''.join(filas)}</div>", " ".join(plano_es), " ".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}")
+        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("<div id='cabecera'><div class='estrella'>✡</div>"
+            "<h1>Centralita Torá</h1><p>hebreo y español · canto Trope con voz celectial diferenciada</p></div>")
+
+    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("🔊 Canto Trope + Voz celectial", variant="secondary")
+            b_he = gr.Button("🔊 עברית (Hebreo)", 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, st_he, 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 "<p>Escribe algo para buscar.</p>"
+            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(v.get("es") or v.get("he") or "")
+                            filas.append(f"<div class='resultado'><span class='ref'>{ref}</span> — {txt}</div>")
+                            if len(filas) >= 120:
+                                return "".join(filas)
+            return "".join(filas) if filas else "<p>Sin resultados en los libros cargados.</p>"
+
+        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 or "").strip()
+            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, añade el Secret `GROQ_API_KEY`."
+        )
+
+if __name__ == "__main__":
+    demo.launch(ssr_mode=False)
+