app.py

# -*- coding: utf-8 -*-
import os
import json
import tempfile
import datetime
import requests
from pathlib import Path

import gradio as gr
try:
    from PyPDF2 import PdfReader
except ImportError:
    from pdfminer.high_level import extract_text as _extract_text

    class _PdfPage:
        def __init__(self, text):
            self._text = text

        def extract_text(self):
            return self._text

    class PdfReader:
        """
        Minimal fallback PDF reader using pdfminer.
        Provides a `pages` attribute with objects that have `extract_text()`.
        """
        def __init__(self, path):
            full_text = _extract_text(path) or ""
            # Split into pages by form feed if possible, else treat as single page
            pages_text = full_text.split("\f")
            self.pages = [_PdfPage(t) for t in pages_text if t]
from pydub import AudioSegment

from gradio_theme_fenix import apply_fenix_theme, FENIX_CSS, fenix_header, fenix_footer

# ----------------------------------------------------------------------
# Configuration (read from environment)
# ----------------------------------------------------------------------
ELEVENLABS_API_KEY = os.environ.get("ELEVENLABS_API_KEY")
ELEVENLABS_VOICE_ID = os.environ.get("ELEVENLABS_VOICE_ID")  # voice for "Álvaro España"
MAX_FILE_SIZE = 10 * 1024 * 1024  # 10 MB
HISTORY_FILE = Path("history.json")
MAX_HISTORY = 5

# ----------------------------------------------------------------------
# Helper functions
# ----------------------------------------------------------------------
def load_history():
    if HISTORY_FILE.exists():
        try:
            with open(HISTORY_FILE, "r", encoding="utf-8") as f:
                return json.load(f)
        except Exception:
            return []
    return []

def save_history(entry):
    history = load_history()
    history.insert(0, entry)  # newest first
    history = history[:MAX_HISTORY]
    with open(HISTORY_FILE, "w", encoding="utf-8") as f:
        json.dump(history, f, ensure_ascii=False, indent=2)

def extract_text_from_pdf(pdf_path: str, progress=gr.Progress()):
    """
    Extracts text from a PDF file preserving reading order.
    """
    progress(0, desc="Opening PDF")
    try:
        reader = PdfReader(pdf_path)
        text_pages = []
        total_pages = len(reader.pages)
        for i, page in enumerate(reader.pages):
            text_pages.append(page.extract_text() or "")
            progress((i + 1) / total_pages, desc=f"Extracting page {i+1}/{total_pages}")
        full_text = "\n".join(text_pages).strip()
        return full_text
    except Exception as e:
        raise RuntimeError(f"Error extracting PDF: {str(e)}")

def call_elevenlabs_tts(text: str, tone: str, speed: float, progress=gr.Progress()):
    """
    Calls ElevenLabs TTS API to generate MP3 audio.
    """
    if not ELEVENLABS_API_KEY or not ELEVENLABS_VOICE_ID:
        raise RuntimeError("ElevenLabs API key or voice ID not configured.")
    url = f"https://api.elevenlabs.io/v1/text-to-speech/{ELEVENLABS_VOICE_ID}"
    headers = {
        "xi-api-key": ELEVENLABS_API_KEY,
        "Content-Type": "application/json"
    }

    # Voice settings: adjust pitch and stability for "sagrado" tone
    voice_settings = {
        "stability": 0.75,
        "similarity_boost": 0.75,
        "speed": speed
    }
    if tone == "sagrado":
        voice_settings.update({
            "stability": 0.9,
            "similarity_boost": 0.9,
            "pitch": 1.2  # higher pitch for sacred tone
        })

    payload = {
        "text": text,
        "voice_settings": voice_settings,
        "model_id": "eleven_multilingual_v2"
    }

    # Retry logic (max 2 retries)
    for attempt in range(2):
        try:
            progress(0.2, desc="Sending request to TTS")
            response = requests.post(url, headers=headers, json=payload, timeout=30)
            if response.status_code == 200:
                progress(0.8, desc="Receiving audio")
                return response.content
            else:
                raise RuntimeError(f"TTS API error {response.status_code}: {response.text}")
        except Exception as e:
            if attempt == 1:
                raise RuntimeError(f"TTS request failed after retries: {str(e)}")
            # wait a moment before retry
            progress(0.5, desc=f"Retry {attempt+1}/2")
    raise RuntimeError("Unexpected flow in TTS generation.")

def generate_audio(pdf_file, tone, speed, progress=gr.Progress()):
    """
    Main pipeline: extract text -> generate audio -> save MP3 -> update history.
    Returns path to MP3 file and a dict for history display.
    """
    if pdf_file is None:
        raise gr.Error("Please upload a PDF file.")
    if pdf_file.size > MAX_FILE_SIZE:
        raise gr.Error("File exceeds maximum size of 10 MB.")

    # Save uploaded PDF to a temporary file
    with tempfile.NamedTemporaryFile(delete=False, suffix=".pdf") as tmp_pdf:
        tmp_pdf.write(pdf_file.read())
        pdf_path = tmp_pdf.name

    # Step 1: Extract text
    progress(0.0, desc="Extracting text")
    text = extract_text_from_pdf(pdf_path, progress=progress)
    if not text:
        raise gr.Error("No readable text found in the PDF.")

    # Step 2: Generate audio via ElevenLabs
    progress(0.3, desc="Generating audio")
    audio_bytes = call_elevenlabs_tts(text, tone, speed, progress=progress)

    # Step 3: Save MP3 to temporary file
    with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as tmp_mp3:
        tmp_mp3.write(audio_bytes)
        mp3_path = tmp_mp3.name

    # Step 4: Gather metadata for history
    file_name = Path(pdf_file.name).name
    duration_sec = AudioSegment.from_file(mp3_path).duration_seconds
    entry = {
        "filename": file_name,
        "date": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "duration": f"{duration_sec:.1f}s",
        "audio_path": mp3_path
    }
    save_history(entry)

    # Clean up PDF temp file
    try:
        os.remove(pdf_path)
    except Exception:
        pass

    return mp3_path, entry

def load_history_for_ui():
    """
    Returns a list of rows for the history Dataframe.
    """
    history = load_history()
    rows = [
        [h["filename"], h["date"], h["duration"]]
        for h in history
    ]
    return rows

def play_from_history(index):
    """
    Returns the audio file path for the selected history entry.
    """
    history = load_history()
    if 0 <= index < len(history):
        return history[index]["audio_path"]
    raise gr.Error("Invalid history selection.")

# ----------------------------------------------------------------------
# Gradio Interface
# ----------------------------------------------------------------------
with gr.Blocks(css=FENIX_CSS, theme=apply_fenix_theme(), title="PDF Voice Reader") as demo:
    fenix_header("PDF Voice Reader", "Convert PDF to audio with sacred tone")

    with gr.Row():
        # Left column: main workflow
        with gr.Column(scale=3):
            pdf_input = gr.File(label="PDF file", file_types=[".pdf"], type="file")
            tone_radio = gr.Radio(
                choices=["normal", "sagrado"],
                label="Select tone",
                value="normal"
            )
            speed_slider = gr.Slider(
                minimum=0.75,
                maximum=1.25,
                step=0.25,
                label="Reading speed",
                value=1.0
            )
            generate_btn = gr.Button("Generar Audio", variant="primary")
            audio_output = gr.Audio(label="Audio result", type="filepath")
            download_btn = gr.File(label="Descargar MP3", visible=False)

            # Progress bar (hidden until used)
            progress_bar = gr.ProgressBar(visible=False)

            # Callback chain
            def on_generate(pdf_file, tone, speed):
                progress_bar.visible = True
                mp3_path, entry = generate_audio(pdf_file, tone, speed, progress=progress_bar)
                download_btn.visible = True
                download_btn.update(value=mp3_path, label="Descargar MP3")
                return mp3_path, entry

            generate_btn.click(
                fn=on_generate,
                inputs=[pdf_input, tone_radio, speed_slider],
                outputs=[audio_output, None],
                api_name=False
            )

        # Right column: history
        with gr.Column(scale=2):
            gr.Markdown("## Historial (últimos 5 PDFs)")
            history_df = gr.Dataframe(
                headers=["Archivo", "Fecha", "Duración"],
                datatype=["str", "str", "str"],
                row_count=MAX_HISTORY,
                column_count=3,
                interactive=False,
                label="Historial"
            )
            play_btn = gr.Button("Reproducir seleccionado")
            history_audio = gr.Audio(label="Audio del historial", type="filepath")

            # Load history initially
            history_df.load(fn=load_history_for_ui)

            def on_play(selected):
                if selected is None or len(selected) == 0:
                    raise gr.Error("Seleccione una fila del historial.")
                # selected is a list of row indices; take first
                idx = selected[0]
                return play_from_history(idx)

            play_btn.click(
                fn=on_play,
                inputs=[history_df.select],
                outputs=history_audio
            )

    fenix_footer()

demo.launch()