app.py

import os
import json
import shutil
import socket
from pathlib import Path
from typing import Any, Dict, Generator, List, Optional

from fastapi.responses import FileResponse, HTMLResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
from gradio import Blocks, Server
from gradio.events import api as gradio_api

from backend.config import (
    APP_TITLE,
    MAX_PREVIEW_CHARACTERS,
    MAX_REFERENCE_AUDIO_SECONDS,
    SESSION_TTL_SECONDS,
    TEMP_ROOT,
)
from backend.epub import EpubConfig, parse_epub
from backend.export import export_audiobook
from backend.input_files import resolve_uploaded_name, resolve_uploaded_path
from backend.session_store import SessionStore
from backend.synthesis_catalog import MAGPIE_OPTIONS, SYNTHESIS_BACKENDS, SYNTHESIS_MODELS
from backend.synthesis_service import SynthesisService
from backend.types import VoiceConfig
from backend.voice_design import VOICE_DESIGN_OPTIONS
from backend.voice_presets import VOICE_PRESETS


ROOT = Path(__file__).parent
FRONTEND_DIR = ROOT / "frontend"
TEMP_ROOT.mkdir(parents=True, exist_ok=True)

app = Server()
app.title = APP_TITLE

store = SessionStore(root=TEMP_ROOT, ttl_seconds=SESSION_TTL_SECONDS)
synthesis_service = SynthesisService(session_root=TEMP_ROOT)


def _warn_about_modal_configuration() -> None:
    client = synthesis_service.modal_client
    warning = client.configuration_warning()
    if warning:
        print(f"Modal configuration warning: {warning}")
        return
    if client.is_configured():
        print(
            "Modal backend configured: "
            f"base_url={client.base_url} timeout={client.timeout_seconds}s poll_interval={client.poll_interval_seconds}s"
        )


_warn_about_modal_configuration()


def _session_root(session_id: str) -> Path:
    return store.ensure_session(session_id).root


def _book_payload(session_id: str) -> Dict[str, Any]:
    return store.load_json(session_id, "book.json")


def _selected_book(session_id: str, selected_ids: List[str]) -> Dict[str, Any]:
    book = _book_payload(session_id)
    selected = set(selected_ids)
    chapters = []
    for chapter in book["chapters"]:
        chapter = dict(chapter)
        chapter["included"] = chapter["id"] in selected
        chapters.append(chapter)
    book["chapters"] = chapters
    return book


def _voice_config_for_backend(voice_config: Dict[str, Any], session_id: str) -> Dict[str, Any]:
    config = dict(voice_config)
    sample_path = config.get("samplePath") or config.get("sample_path")
    if sample_path:
        sample_file = resolve_uploaded_path(sample_path)
        target = _session_root(session_id) / "uploads" / sample_file.name
        shutil.copyfile(sample_file, target)
        config["samplePath"] = str(target)
    return config


@app.api(name="parse_epub")
def parse_epub_api(session_id: str, epub_file: str) -> Dict[str, Any]:
    store.cleanup_expired()
    session = store.ensure_session(session_id)
    source = resolve_uploaded_path(epub_file)
    upload_name = resolve_uploaded_name(epub_file)
    if not source.exists():
        raise ValueError("Uploaded EPUB file was not found")
    target = session.root / "uploads" / upload_name
    shutil.copyfile(source, target)
    payload = parse_epub(target, config=EpubConfig())
    store.save_json(session_id, "book.json", payload)
    return payload


@app.api(name="generate_preview")
def generate_preview_api(
    session_id: str,
    chapter_id: str,
    voice_config: Dict[str, Any],
    diffusion_steps: int = 32,
    speed: float = 1.0,
) -> Dict[str, Any]:
    book = _book_payload(session_id)
    chapter = next((item for item in book["chapters"] if item["id"] == chapter_id), None)
    if chapter is None:
        raise ValueError("Chapter not found for preview")
    text = str(chapter["text"])[:MAX_PREVIEW_CHARACTERS]
    preview_path = _session_root(session_id) / "previews" / f"{chapter_id}.wav"
    voice = _voice_config_for_backend(voice_config, session_id)
    result = synthesis_service.generate_preview(
        text=text,
        output_path=preview_path,
        voice_config=VoiceConfig.from_dict(voice),
        diffusion_steps=diffusion_steps,
        speed=speed,
    )
    return {
        "url": f"/files/{session_id}/previews/{preview_path.name}",
        "duration_seconds": result["duration_seconds"],
        "backend": result["backend"],
        "model": result["model"],
    }


@app.api(name="start_render", concurrency_limit=1)
def start_render_api(
    session_id: str,
    selected_chapter_ids: List[str],
    voice_config: Dict[str, Any],
    diffusion_steps: int = 32,
    speed: float = 1.0,
) -> Generator[Dict[str, Any], None, None]:
    job = synthesis_service.get_job(session_id)
    if job.status == "running":
        raise ValueError("A render is already active for this session")

    book = _selected_book(session_id, selected_chapter_ids)
    voice = _voice_config_for_backend(voice_config, session_id)
    captured: List[Dict[str, Any]] = []
    for event in synthesis_service.render(
        session_id=session_id,
        book=book,
        chapters=book["chapters"],
        voice_config=voice,
        diffusion_steps=diffusion_steps,
        speed=speed,
    ):
        if event.get("type") == "chapter_done" and event.get("output_path"):
            filename = Path(str(event["output_path"])).name
            event = {
                **event,
                "url": f"/files/{session_id}/renders/{filename}",
            }
        captured.append(event)
        yield event

    if captured and captured[-1]["type"] == "completed":
        durations = {
            event["chapter_id"]: event["duration_seconds"]
            for event in captured
            if event["type"] == "chapter_done"
        }
        manifest = {
            "book": book,
            "chapters": [
                {
                    **chapter,
                    "duration_seconds": durations.get(chapter["id"], chapter.get("duration_seconds", 0)),
                }
                for chapter in book["chapters"]
                if chapter.get("included")
            ],
            "render_result": captured[-1],
        }
        store.save_json(session_id, "render_manifest.json", manifest)


@app.api(name="pause_render")
def pause_render_api(session_id: str) -> Dict[str, Any]:
    return synthesis_service.pause(session_id)


@app.api(name="resume_render")
def resume_render_api(session_id: str) -> Dict[str, Any]:
    return synthesis_service.resume(session_id)


@app.api(name="cancel_render")
def cancel_render_api(session_id: str) -> Dict[str, Any]:
    return synthesis_service.cancel(session_id)


@app.api(name="export_audiobook")
def export_audiobook_api(
    session_id: str,
    format: str = "m4a",
    metadata: Dict[str, str] | None = None,
    embed_markers: bool = True,
) -> Dict[str, Any]:
    manifest = store.load_json(session_id, "render_manifest.json")
    export = export_audiobook(
        session_root=_session_root(session_id),
        chapters=manifest["chapters"],
        output_format=format,
        metadata=metadata or {},
        embed_markers=embed_markers,
        cover_path=None,
    )
    export["url"] = f"/files/{session_id}/exports/{Path(export['file']).name}"
    return export


@app.get("/", response_class=HTMLResponse)
async def homepage() -> HTMLResponse:
    html = (FRONTEND_DIR / "index.html").read_text(encoding="utf-8")
    preset_script = (
        "<script>"
        f"window.__VOICE_PRESETS__ = {json.dumps(VOICE_PRESETS)};"
        f"window.__VOICE_DESIGN_OPTIONS__ = {json.dumps(VOICE_DESIGN_OPTIONS)};"
        f"window.__MAGPIE_OPTIONS__ = {json.dumps(MAGPIE_OPTIONS)};"
        f"window.__SYNTHESIS_MODELS__ = {json.dumps(SYNTHESIS_MODELS)};"
        f"window.__SYNTHESIS_BACKENDS__ = {json.dumps(SYNTHESIS_BACKENDS)};"
        "</script>"
    )
    return HTMLResponse(html.replace("</body>", f"    {preset_script}\n  </body>"))


@app.get("/health")
async def health() -> Dict[str, str]:
    return {"status": "ok"}


@app.get("/files/{session_id}/{kind}/{filename}")
async def serve_generated_file(session_id: str, kind: str, filename: str, download: bool = False):
    path = _session_root(session_id) / kind / filename
    if not path.exists():
        return JSONResponse({"error": "file not found"}, status_code=404)
    if download:
        return FileResponse(path, filename=filename)
    return FileResponse(path, content_disposition_type="inline")


app.mount("/static", StaticFiles(directory=str(FRONTEND_DIR)), name="frontend-static")


def _build_demo() -> Blocks:
    # Expose the launched Blocks at module scope so Gradio hot reload can find it.
    with Blocks(mode="server") as demo:
        for fn, api_kwargs in app._deferred_apis:
            gradio_api(fn=fn, **api_kwargs)
    return demo


demo = _build_demo()


def _port_is_available(port: int, host: str = "127.0.0.1") -> bool:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        return sock.connect_ex((host, port)) != 0


def _choose_server_port(preferred: int = 7860, attempts: int = 10) -> int:
    env_port = os.getenv("GRADIO_SERVER_PORT")
    if env_port:
        return int(env_port)
    for port in range(preferred, preferred + attempts):
        if _port_is_available(port):
            return port
    raise OSError(
        f"Cannot find empty port in range: {preferred}-{preferred + attempts - 1}. "
        "Set GRADIO_SERVER_PORT to override."
    )


def _launch_app(server_port: int) -> None:
    demo.launch(
        _app=app,
        server_name="0.0.0.0",
        server_port=server_port,
        # This app serves its own frontend and uses Gradio for the API layer only.
        # SSR adds a Node proxy that is unnecessary here and has been noisy in Spaces.
        ssr_mode=False,
    )


if __name__ == "__main__":
    server_port = _choose_server_port()
    print(f"Starting {APP_TITLE} on port {server_port}")
    _launch_app(server_port)