app.py

"""Validated HTTP service for consented OpenVoice synthesis."""

from __future__ import annotations

import logging
import os
import tempfile
import time
import uuid
from pathlib import Path
from typing import Any

from flask import Flask, jsonify, request, send_from_directory

from backend import OpenVoiceBackend, VoiceBackend

LOGGER = logging.getLogger(__name__)


def _error(message: str, status: int):
    return jsonify({"error": message}), status


def create_app(
    backend: VoiceBackend | None = None,
    *,
    output_dir: str | Path | None = None,
) -> Flask:
    app = Flask(__name__)
    app.config["BACKEND"] = backend or OpenVoiceBackend()
    app.config["OUTPUT_DIR"] = Path(
        output_dir or os.getenv("OPENVOICE_OUTPUT_DIR", "generated_audio")
    ).resolve()
    app.config["REFERENCE_AUDIO"] = os.getenv("OPENVOICE_REFERENCE_AUDIO", "")
    app.config["MAX_TEXT_CHARS"] = int(os.getenv("MAX_TEXT_CHARS", "2000"))
    app.config["MAX_CONTENT_LENGTH"] = int(os.getenv("MAX_UPLOAD_MB", "20")) * 1024 * 1024
    app.config["OUTPUT_DIR"].mkdir(parents=True, exist_ok=True)

    @app.get("/health")
    def health():
        selected: VoiceBackend = app.config["BACKEND"]
        return jsonify(
            {
                "status": "ok",
                "backend": selected.name,
                "model_loaded": selected.loaded,
                "supported_languages": list(selected.supported_languages),
            }
        )

    def read_request() -> tuple[dict[str, Any] | None, Any]:
        if request.mimetype == "application/json":
            payload = request.get_json(silent=True)
            if not isinstance(payload, dict):
                return None, _error("a JSON object is required", 400)
            return payload, None
        return request.form.to_dict(), None

    def run_synthesis(*, legacy: bool = False):
        payload, problem = read_request()
        if problem is not None:
            return problem
        payload = payload or {}
        text = payload.get("text")
        language = str(payload.get("language") or payload.get("lang") or "JP").upper()
        if not isinstance(text, str) or not text.strip():
            return _error("'text' is required", 400)
        if len(text) > app.config["MAX_TEXT_CHARS"]:
            return _error("'text' is too long", 400)

        try:
            speed = float(payload.get("speed", 1.0))
        except (TypeError, ValueError):
            return _error("'speed' must be numeric", 400)
        if not 0.5 <= speed <= 2.0:
            return _error("'speed' must be between 0.5 and 2.0", 400)

        selected: VoiceBackend = app.config["BACKEND"]
        if language not in selected.supported_languages:
            return _error(f"unsupported language: {language}", 400)

        uploaded = request.files.get("reference_audio")
        temporary_reference: Path | None = None
        if uploaded is not None and uploaded.filename:
            suffix = Path(uploaded.filename).suffix.lower()
            if suffix not in {".wav", ".mp3", ".m4a", ".flac", ".ogg"}:
                return _error("unsupported reference-audio format", 400)
            with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as temporary:
                uploaded.save(temporary)
                temporary_reference = Path(temporary.name)
            reference = temporary_reference
        else:
            configured = app.config["REFERENCE_AUDIO"]
            if not configured:
                return _error(
                    "upload 'reference_audio' or configure OPENVOICE_REFERENCE_AUDIO",
                    400,
                )
            reference = Path(configured)

        requested_name = payload.get("file_path") if legacy else None
        if requested_name:
            filename = Path(str(requested_name)).name
            if filename != requested_name or Path(filename).suffix.lower() != ".wav":
                return _error("'file_path' must be a plain .wav filename", 400)
        else:
            filename = f"{uuid.uuid4().hex}.wav"
        destination = app.config["OUTPUT_DIR"] / filename
        started = time.perf_counter()
        try:
            selected.synthesize(text.strip(), language, reference, destination, speed)
        except ValueError as exc:
            return _error(str(exc), 400)
        except RuntimeError as exc:
            LOGGER.warning("voice backend is unavailable: %s", exc)
            return _error(str(exc), 503)
        except Exception:
            LOGGER.exception("voice synthesis failed")
            return _error("voice synthesis failed", 500)
        finally:
            if temporary_reference is not None:
                temporary_reference.unlink(missing_ok=True)

        body = {
            "audio_path": filename,
            "audio_url": f"/audio/{filename}",
            "language": language,
            "elapsed_seconds": round(time.perf_counter() - started, 3),
        }
        return jsonify(body), 200

    @app.post("/synthesize")
    def synthesize():
        return run_synthesis()

    @app.get("/audio/<path:filename>")
    def audio(filename: str):
        if Path(filename).name != filename or Path(filename).suffix.lower() != ".wav":
            return _error("invalid audio filename", 400)
        return send_from_directory(app.config["OUTPUT_DIR"], filename)

    app.add_url_rule(
        "/get_openvoice",
        endpoint="legacy_openvoice",
        view_func=lambda: run_synthesis(legacy=True),
        methods=["POST"],
    )
    app.add_url_rule(
        "/get_openvoice_batch",
        endpoint="legacy_openvoice_batch",
        view_func=lambda: run_synthesis(legacy=True),
        methods=["POST"],
    )
    app.add_url_rule(
        "/get_openvoice_batch_2",
        endpoint="legacy_openvoice_batch_2",
        view_func=lambda: run_synthesis(legacy=True),
        methods=["POST"],
    )
    return app


app = create_app()


if __name__ == "__main__":
    logging.basicConfig(level=os.getenv("LOG_LEVEL", "INFO"))
    app.run(host="0.0.0.0", port=int(os.getenv("PORT", "5002")))