pdf-reader / app.pybb
Cristobal299's picture
Rename app.py to app.pybb
6668fa0 verified
Raw
History Blame Contribute Delete
23.8 kB
# -*- 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. ' ' -> ' '), 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 "<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>", "", ""
# 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("&nbsp;", " ").replace("{פ}", "").replace("{ס}", "").strip()
es_raw = _limpiar_entidades(v.get("es", "")).replace("&nbsp;", " ").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"<div class='cab'><span class='h'>{html.escape(_limpiar_entidades(d.get('heb','')))}</span><br>"
f"{html.escape(d.get('es',''))} {cap}</div>"]
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"<div class='verso'><span class='num'>{n_verso}</span>"
if he_raw:
fila += f"<span class='he'>{html.escape(he_raw)}</span>"
if es_sentido:
fila += f"<span class='es'>{html.escape(es_sentido)}</span>"
fila += "</div>"
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"<div class='pasaje'>{''.join(filas)}</div>", 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("<div id='cabecera'><div class='estrella'>✡</div>"
"<h1>Centralita Torá</h1><p>hebreo real e impecable · español fluido y natural</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("🔊 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 "<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(_limpiar_entidades(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.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)