"""Hardened Flask API for text-to-speech synthesis."""

from __future__ import annotations

import logging
import os
import re
import tempfile
import threading
import time
from collections import defaultdict, deque
from importlib import import_module
from importlib.util import find_spec
from pathlib import Path
from typing import TYPE_CHECKING, cast

from flask import Flask, Response, request
from flask_cors import CORS

from rex.config import _parse_int, load_config
from rex.exception_handler import wrap_entrypoint
from rex.graceful_shutdown import get_shutdown_handler
from rex.ha_bridge import create_blueprint as create_ha_blueprint
from rex.health import check_config, create_health_blueprint
from rex.http_errors import (
    BAD_REQUEST,
    INTERNAL_ERROR,
    PAYLOAD_TOO_LARGE,
    TOO_MANY_REQUESTS,
    UNAUTHORIZED,
    error_response,
)
from rex.request_logging import install_request_logging
from rex.startup_validation import check_startup_env
from rex.tts_utils import chunk_text_for_xtts

try:
    from flask_limiter import Limiter
    from flask_limiter.exceptions import RateLimitExceeded
except ImportError:

    class Limiter:  # type: ignore
        def __init__(self, *args, **kwargs):
            pass

        def limit(self, *args, **kwargs):
            def decorator(func):
                return func

            return decorator

    class RateLimitExceeded(Exception):  # type: ignore
        retry_after = None


from rex.assistant_errors import AuthenticationError, TextToSpeechError
from rex.memory_utils import (
    extract_voice_reference,
    load_all_profiles,
    load_users_map,
    resolve_user_key,
)

# ------------------------------------------------------------------------------
# App setup
# ------------------------------------------------------------------------------

app = Flask(__name__)
install_request_logging(app)
app.register_blueprint(create_health_blueprint(checks=[check_config]))

ALLOWED_ORIGINS = os.getenv(
    "REX_ALLOWED_ORIGINS",
    "http://localhost:3000,http://localhost:5000,http://127.0.0.1:3000,http://127.0.0.1:5000",
)
_CORS_ORIGINS = [origin.strip() for origin in ALLOWED_ORIGINS.split(",") if origin.strip()]

CORS(
    app,
    resources={
        r"/*": {
            "origins": _CORS_ORIGINS,
            "methods": ["GET", "POST", "OPTIONS"],
            "allow_headers": ["Content-Type", "Authorization", "X-API-Key"],
            "expose_headers": ["Retry-After"],
            "supports_credentials": False,
            "max_age": 600,
        }
    },
)

logger = logging.getLogger("rex.speak_api")

if not logger.handlers:
    logging.basicConfig(
        level=os.getenv("REX_LOG_LEVEL", "INFO").upper(),
        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    )

# ------------------------------------------------------------------------------
# Rate Limiting
# ------------------------------------------------------------------------------

TRUSTED_PROXIES = {
    ip.strip() for ip in os.getenv("REX_TRUSTED_PROXIES", "127.0.0.1,::1").split(",") if ip.strip()
}


def _rate_limit_key() -> str:
    """Return the rate-limit bucket key.

    Derived only from the validated client/proxy address — never from
    submitted credentials (API key, Authorization header, or any
    secret-derived value) so that rotating or varying a credential cannot be
    used to evade the limit, and no credential material ends up in the
    limiter's storage backend.
    """
    remote_addr = request.remote_addr or "unknown"
    if remote_addr in TRUSTED_PROXIES:
        forwarded = request.headers.get("X-Forwarded-For")
        if forwarded:
            return forwarded.split(",")[-1].strip()

    return remote_addr


_LIMITER_STORAGE_URI = (
    os.getenv("REX_SPEAK_STORAGE_URI") or os.getenv("FLASK_LIMITER_STORAGE_URI") or "memory://"
)

limiter = Limiter(
    key_func=_rate_limit_key,
    storage_uri=_LIMITER_STORAGE_URI,
    app=app,
    default_limits=[],
)

_RATE_CACHE: dict[str, deque] | None = (
    defaultdict(deque) if _LIMITER_STORAGE_URI.startswith("memory") else None
)

try:
    app.register_blueprint(create_ha_blueprint())
    logger.info("Home Assistant bridge registered at /ha/*")
except RuntimeError as _ha_err:
    logger.warning("Home Assistant bridge was not registered")

# Shopping list PWA (US-SL-004)
try:
    from rex.shopping_pwa import create_blueprint as _create_shopping_blueprint

    from rex.shopping_list import ShoppingList as _ShoppingList

    _shopping_pwa_pin = os.getenv("REX_SHOPPING_PIN") or None
    app.register_blueprint(_create_shopping_blueprint(_ShoppingList(), pin=_shopping_pwa_pin))
    logger.info(
        "Shopping list PWA registered at /shopping (pin=%s)", "set" if _shopping_pwa_pin else "none"
    )
except Exception as _e:
    logger.warning("Shopping list PWA was not registered")

# ------------------------------------------------------------------------------
# Config and Globals
# ------------------------------------------------------------------------------

USERS_MAP = load_users_map()
USER_PROFILES = load_all_profiles()
DEFAULT_USER = resolve_user_key(
    os.getenv("REX_ACTIVE_USER"), USERS_MAP, profiles=USER_PROFILES
) or (sorted(USER_PROFILES.keys())[0] if USER_PROFILES else "james")

USER_VOICES = {
    user: extract_voice_reference(profile, user_key=user) for user, profile in USER_PROFILES.items()
}
if DEFAULT_USER not in USER_VOICES:
    USER_VOICES[DEFAULT_USER] = None

_TTS_ENGINE: TTS | None = None
_tts_lock = threading.Lock()

DEFAULT_TTS_MODEL = os.getenv("REX_TTS_MODEL", "tts_models/multilingual/multi-dataset/xtts_v2")
_MODEL_PATTERN = re.compile(r"^[\w\-./]+(\/[\w\-./]+)*$")
if not _MODEL_PATTERN.match(DEFAULT_TTS_MODEL):
    logger.warning("Invalid TTS model name '%s'; using default.", DEFAULT_TTS_MODEL)
    DEFAULT_TTS_MODEL = "tts_models/multilingual/multi-dataset/xtts_v2"

API_KEY: str | None = None
RATE_LIMIT = _parse_int("REX_SPEAK_RATE_LIMIT", os.getenv("REX_SPEAK_RATE_LIMIT"), default=30)
RATE_LIMIT_WINDOW = _parse_int(
    "REX_SPEAK_RATE_WINDOW", os.getenv("REX_SPEAK_RATE_WINDOW"), default=60
)
MAX_TEXT_LENGTH = _parse_int("REX_SPEAK_MAX_CHARS", os.getenv("REX_SPEAK_MAX_CHARS"), default=800)
MAX_REQUEST_BYTES = _parse_int(
    "REX_SPEAK_MAX_REQUEST_BYTES",
    os.getenv("REX_SPEAK_MAX_REQUEST_BYTES"),
    default=65536,  # 64 KB
)
TTS_SPEED = load_config().tts_speed
_RATE_LIMIT_SPEC: str | None

# Enforce body-size limit via Flask: rejects Content-Length violations before the body
# is read and caps streamed reads at MAX_REQUEST_BYTES for requests without Content-Length.
app.config["MAX_CONTENT_LENGTH"] = MAX_REQUEST_BYTES

if RATE_LIMIT > 0 and RATE_LIMIT_WINDOW > 0:
    _UNIT_MAP = {1: "second", 60: "minute", 3600: "hour", 86400: "day"}
    unit = _UNIT_MAP.get(RATE_LIMIT_WINDOW, None)
    if unit:
        _RATE_LIMIT_SPEC = f"{RATE_LIMIT} per {unit}"
    else:
        _RATE_LIMIT_SPEC = f"{RATE_LIMIT}/{RATE_LIMIT_WINDOW} second"
else:
    _RATE_LIMIT_SPEC = None


def _typed_error_response(
    code: str,
    message: str,
    http_status: int,
    *,
    request_id: str | None = None,
) -> tuple[Response, int]:
    resp, status = error_response(
        code,
        message,
        http_status,
        request_id=request_id,
    )
    return cast(Response, resp), status


# ------------------------------------------------------------------------------
# Graceful-shutdown guard
# ------------------------------------------------------------------------------


@app.before_request
def _reject_during_shutdown() -> tuple[Response, int] | None:
    """Return 503 when a SIGTERM-triggered shutdown is in progress."""
    if get_shutdown_handler().is_shutting_down:
        resp, status = _typed_error_response("SERVICE_UNAVAILABLE", "Server is shutting down", 503)
        return resp, status
    return None


@app.before_request
def _check_request_size() -> tuple[Response, int] | None:
    """Reject requests whose Content-Length exceeds MAX_REQUEST_BYTES before reading body."""
    cl = request.content_length
    if cl is not None and cl > MAX_REQUEST_BYTES:
        resp, status = _typed_error_response(PAYLOAD_TOO_LARGE, "Request body too large", 413)
        return resp, status
    return None


# Error Handlers
# ------------------------------------------------------------------------------


# Fixed, non-sensitive messages an AuthenticationError may surface to the
# client. Any AuthenticationError text outside this allowlist is replaced
# with a generic fixed message so arbitrary exception content can never
# reach the response body.
_AUTH_ERROR_MESSAGES = frozenset({"Missing API key", "Invalid API key"})
_GENERIC_AUTH_ERROR_MESSAGE = "Authentication failed"
_GENERIC_TTS_ERROR_MESSAGE = "Speech synthesis failed"


@app.errorhandler(AuthenticationError)
def _handle_auth_error(exc: AuthenticationError) -> tuple[Response, int]:
    message = str(exc) if str(exc) in _AUTH_ERROR_MESSAGES else _GENERIC_AUTH_ERROR_MESSAGE
    resp, status = _typed_error_response(UNAUTHORIZED, message, 401)
    resp.status_code = status
    return resp, status


@app.errorhandler(TextToSpeechError)
def _handle_tts_error(exc: TextToSpeechError) -> tuple[Response, int]:
    logger.warning("Speech synthesis failed: %s", type(exc).__name__)
    resp, status = _typed_error_response(INTERNAL_ERROR, _GENERIC_TTS_ERROR_MESSAGE, 500)
    resp.status_code = status
    return resp, status


@app.errorhandler(RateLimitExceeded)
def _handle_rate_limit(exc: RateLimitExceeded) -> Response:
    resp, _ = _typed_error_response(TOO_MANY_REQUESTS, "Too many requests", 429)
    resp.status_code = 429
    if getattr(exc, "retry_after", None):
        resp.headers["Retry-After"] = str(int(exc.retry_after))
    return resp


# ------------------------------------------------------------------------------
# Helpers
# ------------------------------------------------------------------------------


def _get_tts_engine() -> TTS:
    """Return the TTS engine singleton. Must be called while holding _tts_lock."""
    global _TTS_ENGINE
    if _TTS_ENGINE is None:
        if find_spec("TTS") is None:
            raise TextToSpeechError(
                'Coqui TTS is not installed. Install with `pip install -e ".[ml]"`.'
            )
        try:
            from rex.compat import ensure_transformers_compatibility

            ensure_transformers_compatibility()
            tts_class = import_module("TTS.api").TTS
        except Exception as exc:
            raise TextToSpeechError(
                'Failed to import Coqui TTS. Install with `pip install -e ".[ml]"`.'
            ) from exc
        _TTS_ENGINE = tts_class(model_name=DEFAULT_TTS_MODEL, progress_bar=False)
    return _TTS_ENGINE


def _load_audio_dependencies():
    if find_spec("numpy") is None or find_spec("soundfile") is None:
        raise TextToSpeechError(
            'Audio dependencies missing. Install with `pip install -e ".[audio]"`.'
        )
    try:
        np = import_module("numpy")
        sf = import_module("soundfile")
    except Exception as exc:
        raise TextToSpeechError(
            'Failed to import audio dependencies. Install with `pip install -e ".[audio]"`.'
        ) from exc
    return np, sf


def _request_api_key() -> str | None:
    provided = request.headers.get("X-API-Key") or request.headers.get("Authorization")
    if not provided:
        return None
    parts = provided.split()
    return parts[-1] if parts else None


def get_api_key() -> str | None:
    """Return the configured speak-API key.

    Resolved from the credential vault (household scope) via
    ``rex.credentials.get_persisted_credential``. A configured vault entry is
    authoritative; the plaintext ``REX_SPEAK_API_KEY`` environment variable
    is consulted only as an explicit legacy/operator fallback (see
    ``rex.credentials.legacy_plaintext_fallback_enabled``) and never
    overrides a vault value.
    """
    from rex.credentials import get_persisted_credential

    return get_persisted_credential("REX_SPEAK_API_KEY") or None


def _require_api_key() -> None:
    provided = _request_api_key()
    api_key = get_api_key()
    if not api_key:
        raise AuthenticationError("Missing API key")
    if not provided:
        raise AuthenticationError("Missing API key")
    if not hmac_compare(provided, api_key):
        raise AuthenticationError("Invalid API key")


def hmac_compare(a: str, b: str) -> bool:
    if len(a) != len(b):
        return False
    result = 0
    for x, y in zip(a.encode("utf-8"), b.encode("utf-8"), strict=True):
        result |= x ^ y
    return result == 0


def _enforce_rate_limit() -> None:
    if RATE_LIMIT <= 0 or RATE_LIMIT_WINDOW <= 0:
        return
    if _RATE_CACHE is None:
        return
    now = time.monotonic()
    key = _rate_limit_key()
    bucket = _RATE_CACHE.setdefault(key, deque())
    while bucket and now - bucket[0] > RATE_LIMIT_WINDOW:
        bucket.popleft()
    if len(bucket) >= RATE_LIMIT:
        raise RateLimitExceeded("Too many requests")
    bucket.append(now)


def _resolve_speaker_wav(user_key: str) -> str | None:
    voice_path = USER_VOICES.get(user_key)
    if voice_path and Path(voice_path).is_file():
        return voice_path
    return None


def generate_speech(text: str, language: str, user_key: str) -> bytes:
    """Generate speech audio from text. Can be monkeypatched for testing."""
    speaker_wav = _resolve_speaker_wav(user_key)
    np, sf = _load_audio_dependencies()

    with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp:
        output_path = tmp.name

    # _tts_lock is acquired before initialization check and held until synthesis completes
    with _tts_lock:
        engine = _get_tts_engine()
        try:
            chunks = chunk_text_for_xtts(text, max_tokens=300)
            if not chunks:
                raise TextToSpeechError("No speech content to synthesize.")

            audio_segments = []
            sample_rate = None

            for chunk in chunks:
                with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp:
                    chunk_path = tmp.name

                try:
                    engine.tts_to_file(
                        text=chunk,
                        speaker_wav=speaker_wav,
                        language=language,
                        file_path=chunk_path,
                        speed=TTS_SPEED,
                    )
                    data, rate = sf.read(chunk_path, dtype="float32")
                    if sample_rate is None:
                        sample_rate = rate
                    audio_segments.append(data)
                finally:
                    try:
                        Path(chunk_path).unlink(missing_ok=True)
                    except PermissionError:
                        logger.debug("Skipping cleanup for locked synthesized-audio chunk")

            if not audio_segments or sample_rate is None:
                raise TextToSpeechError("XTTS produced no audio output.")

            combined_audio = np.concatenate(audio_segments, axis=0)
            sf.write(output_path, combined_audio, sample_rate)
        except TextToSpeechError:
            raise
        except Exception as exc:
            logger.warning(
                "Speech synthesis failed during voice generation: %s", type(exc).__name__
            )
            raise TextToSpeechError("Speech synthesis failed.") from exc

    try:
        with open(output_path, "rb") as audio_file:
            audio_bytes = audio_file.read()
    except Exception as exc:
        logger.warning("Failed to read synthesized audio output: %s", type(exc).__name__)
        raise TextToSpeechError("Failed to read synthesized audio output.") from exc
    finally:
        try:
            Path(output_path).unlink(missing_ok=True)
        except PermissionError:
            logger.debug("Skipping cleanup for locked synthesized-audio file")

    return audio_bytes


# ------------------------------------------------------------------------------
# Routes
# ------------------------------------------------------------------------------


@app.route("/speak", methods=["POST"])
def speak() -> Response:
    _require_api_key()
    _enforce_rate_limit()

    payload = request.get_json(silent=True) or {}
    text = payload.get("text")
    if not text or not isinstance(text, str):
        resp, status = _typed_error_response(BAD_REQUEST, "Text is required", 400)
        resp.status_code = status
        return resp
    if len(text) > MAX_TEXT_LENGTH:
        resp, status = _typed_error_response(BAD_REQUEST, "Text exceeds maximum length", 400)
        resp.status_code = status
        return resp

    language = payload.get("language") or "en"
    user_key = payload.get("user") or DEFAULT_USER

    audio_bytes = generate_speech(text, language, user_key)
    return Response(audio_bytes, mimetype="audio/wav")


# ------------------------------------------------------------------------------
# Entry point
# ------------------------------------------------------------------------------


@wrap_entrypoint
def main() -> None:
    check_startup_env()
    if not get_api_key():
        raise RuntimeError("REX_SPEAK_API_KEY must be set")
    shutdown = get_shutdown_handler()
    shutdown.install()
    app.run(host="127.0.0.1", port=int(os.getenv("REX_SPEAK_PORT") or "5005"))


if __name__ == "__main__":
    main()
if TYPE_CHECKING:  # pragma: no cover - typing only
    from TTS.api import TTS