src/api/preload_api.py

"""Preload-mode API: the dropdown catalog + pre-aligned segment fetch.

Mirrors the Preload UI for programmatic consumers (e.g. the QuranCaption app):

- ``preload_recitations`` — every released recitation as a
  ``(reciter, recitation-specifics)`` tuple plus its available chapters. One
  call backs the recitation + surah dropdowns.
- ``preload_segments`` — pre-aligned segments (+ optional word timestamps) for a
  recitation / chapter / verse range. Output is byte-identical to the Preload
  segment-mode JSON download (shared ``build_segment_export`` path), plus a
  single directly-fetchable ``audio_url``.

These endpoints are **public / ungated** — Preload data is read-only catalog
content served from the inspector bucket, so they intentionally do NOT call
``enforce_beta_access`` (the Caption app and scripted callers have no OAuth
profile). Everything is a bucket read; no GPU, no session.
"""

from __future__ import annotations

import os
import threading
import time

import gradio as gr

from config import PRELOAD_AUDIO_CATALOG_TTL_S, PRELOAD_CLIP_PAD_S
from src.preload.clip_audio import plan_clip_window
from src.core.usage_logger import (
    get_user_id,
    log_error,
    mark_endpoint_entry,
    set_stage,
)
from src.pipeline.exports import build_segment_export
from src.preload import manifest_client, recitation_labels, repo_loader


# ---------------------------------------------------------------------------
# Audio URL — hand back a single directly-fetchable absolute URL.
# ---------------------------------------------------------------------------


def _space_base_url(request: gr.Request | None) -> str:
    """Public base URL of this Space (``https://<host>``), or ``""``.

    Prefers the HF-set ``SPACE_HOST`` env var (the canonical ``*.hf.space``
    host), falling back to the forwarded/host headers of the incoming request.
    """
    host = os.environ.get("SPACE_HOST")
    if host:
        return f"https://{host.rstrip('/')}"
    if request is not None:
        try:
            headers = request.headers
            h = headers.get("x-forwarded-host") or headers.get("host")
            proto = headers.get("x-forwarded-proto") or "https"
            if h:
                return f"{proto}://{h}"
        except Exception:  # noqa: BLE001 — header access is best-effort
            pass
    return ""


def _absolute_audio_url(audio_url: str, request: gr.Request | None) -> str:
    """Make a resolved chapter audio URL directly fetchable by external callers.

    ``repo_loader.resolve_chapter_audio_url`` returns the in-Space bucket-route
    path (``/preload-audio/<slug>/<ch>.mp3`` — a public, Range-capable endpoint)
    that every released delivery's audio is served through. Relative paths are
    absolutized against the Space host; an already-absolute URL is returned
    unchanged. There is no separate CDN path — audio is always one route.
    """
    if not audio_url:
        return ""
    if audio_url.startswith(("http://", "https://")):
        return audio_url
    base = _space_base_url(request)
    return f"{base}{audio_url}" if base else audio_url


def _rezero_to_clip(segments, audio_url: str, request: gr.Request | None):
    """Point audio at a range-cut clip and re-zero segment times to its start.

    The verse window is derived from the segments' own ``time_from`` /
    ``time_to`` — the only absolute fields in a segment-mode export (per-word
    ``start`` / ``end`` are already segment-relative, so they are left alone).
    The window is padded by ``PRELOAD_CLIP_PAD_S`` on each side, ``audio_url``
    gains ``?start_ms&end_ms`` for the route to cut against, and every segment's
    ``time_from`` / ``time_to`` is shifted to be relative to the clip start.

    Returns ``(clip_audio_url, clip_start_s)``. Falls back to the full chapter
    URL with ``clip_start = 0.0`` when there is no audio or no usable timing —
    callers should treat ``clip_start`` only as informational metadata.
    """
    if not audio_url or not segments:
        return _absolute_audio_url(audio_url, request), 0.0

    window = plan_clip_window(segments, PRELOAD_CLIP_PAD_S)
    if window is None:
        return _absolute_audio_url(audio_url, request), 0.0
    start_ms, end_ms, clip_start = window

    sep = "&" if "?" in audio_url else "?"
    clip_url = _absolute_audio_url(
        f"{audio_url}{sep}start_ms={start_ms}&end_ms={end_ms}", request,
    )
    return clip_url, clip_start


# ---------------------------------------------------------------------------
# Endpoint 1 — recitation catalog (drop-downs)
# ---------------------------------------------------------------------------


def preload_recitations(request: gr.Request = None):
    """Return every released recitation with its available chapters.

    Shape: ``{"recitations": [{slug, label, reciter, riwayah, style, channel,
    source, chapters: [int, ...]}, ...]}``. On catalog failure returns
    ``{"recitations": [], "error": "..."}``.
    """
    mark_endpoint_entry()
    set_stage("catalog")
    try:
        catalog = manifest_client.load_catalog()
        details = recitation_labels.build_recitation_details(catalog)
    except Exception as e:  # noqa: BLE001 — surface as an error payload
        log_error(
            error_code="catalog_load_failed",
            endpoint="preload_recitations",
            stage="catalog",
            exception=e,
            user_id=get_user_id(request) if request else "unknown",
            message=f"Preload catalog load failed: {e}",
        )
        return {"recitations": [], "error": f"Catalog load failed: {e}"}

    recitations = []
    for row in details:
        try:
            chapters = repo_loader.chapters_for_delivery(row["slug"])
        except Exception:  # noqa: BLE001 — a single bad delivery shouldn't 500 the list
            chapters = []
        recitations.append({**row, "chapters": chapters})
    return {"recitations": recitations}


# ---------------------------------------------------------------------------
# Endpoint 2 — pre-aligned segments for a recitation / chapter / verse range
# ---------------------------------------------------------------------------


def _coerce_verse(value, default: int) -> int:
    """Parse a verse bound; non-positive / unparseable falls back to ``default``."""
    try:
        iv = int(value)
    except (TypeError, ValueError):
        return default
    return iv if iv > 0 else default


def preload_segments(recitation, chapter, verse_from, verse_to,
                     include_timestamps=True, request: gr.Request = None):
    """Pre-aligned segments for a recitation chapter + verse range.

    Verse range defaults to the full chapter (1 → last verse) when
    ``verse_from`` / ``verse_to`` are missing or ≤ 0; bounds are clamped to the
    chapter and ordered. ``include_timestamps`` (default True) controls whether
    per-word timestamps are emitted. Segments mirror the Preload segment-mode
    download byte-for-byte; the envelope adds ``recitation/chapter/verse_from/
    verse_to/audio_url``. Errors return ``{"error", "segments": []}``.
    """
    mark_endpoint_entry()
    set_stage("validate")

    slug = (recitation or "").strip() if isinstance(recitation, str) else ""
    if not slug:
        log_error(
            error_code="invalid_recitation", endpoint="preload_segments",
            stage="validate", user_id=get_user_id(request) if request else "unknown",
            message="recitation (delivery slug) is required",
        )
        return {"error": "recitation is required", "segments": []}

    try:
        chapter = int(chapter)
    except (TypeError, ValueError):
        return {"error": "chapter must be an integer", "segments": []}

    surah_info = repo_loader.load_surah_info_local()
    meta = surah_info.get(str(chapter))
    if not meta:
        return {"error": f"Unknown chapter {chapter}", "segments": []}
    num_verses = int(meta.get("num_verses") or 1)

    lo = _coerce_verse(verse_from, default=1)
    hi = _coerce_verse(verse_to, default=num_verses)
    lo = max(1, min(lo, num_verses))
    hi = max(1, min(hi, num_verses))
    if lo > hi:
        lo, hi = hi, lo

    include_words = True if include_timestamps is None else bool(include_timestamps)

    set_stage("build")
    try:
        pc = repo_loader.build_segment_infos(slug, chapter, lo, hi)
    except Exception as e:  # noqa: BLE001 — surface as an error payload
        log_error(
            error_code="preload_build_failed", endpoint="preload_segments",
            stage="build", exception=e,
            user_id=get_user_id(request) if request else "unknown",
            message=f"Preload segment build failed: {e}",
            context={"recitation": slug, "chapter": chapter,
                     "verse_from": lo, "verse_to": hi},
        )
        return {"error": f"Failed to load segments: {e}", "segments": []}

    payload = build_segment_export(
        pc.all_segments, include_words=include_words, source="preload",
    )
    segments = payload.get("segments", []) if payload else []
    meta_block = (
        payload.get("_meta") if payload
        else {"view_mode": "segment", "source": "preload"}
    )

    # Range-scoped clip: audio_url points at the cut for [verse_from, verse_to]
    # and segment times are re-zeroed to the clip start, so the caller downloads
    # only the requested verses with timings that already begin at 0.
    audio_url, clip_start = _rezero_to_clip(segments, pc.audio_url, request)

    return {
        "_meta": meta_block,
        "recitation": slug,
        "chapter": chapter,
        "verse_from": lo,
        "verse_to": hi,
        "audio_url": audio_url,
        "clip_start": round(clip_start, 3),
        "segments": segments,
    }


# ---------------------------------------------------------------------------
# Endpoint 3 — audio-only catalog (full audio set, NON-published reciters)
# ---------------------------------------------------------------------------


# Assembled audio-only payload is TTL-cached: the catalog is ~2k+ deliveries and
# each chapter list is a separate sidecar read, so we build it once per window.
_AUDIO_RECITATIONS_LOCK = threading.Lock()
_AUDIO_RECITATIONS_CACHE: dict | None = None
_AUDIO_RECITATIONS_AT: float = 0.0


def _build_audio_recitations() -> dict:
    """Assemble the audio-only catalog (rows + chapters), reading sidecars in
    parallel. Raises on catalog-load failure; the caller maps it to an error."""
    catalog = manifest_client.load_catalog()
    details = recitation_labels.build_recitation_details(
        catalog,
        reciters_fn=repo_loader.audio_reciters,
        deliveries_fn=repo_loader.audio_deliveries_for_reciter,
    )
    chapters_by_slug = repo_loader.chapters_for_audio_deliveries(
        [row["slug"] for row in details]
    )
    recitations = [
        {**row, "chapters": chapters_by_slug.get(row["slug"], [])} for row in details
    ]
    return {"recitations": recitations}


def preload_audio_recitations(request: gr.Request = None):
    """Return the audio-only recitation catalog with available chapters.

    The audio-only set is every ``by_surah`` delivery that has audio on the
    bucket but is **not** in the released/segments catalog — so it never
    duplicates ``preload_recitations``. These reciters offer downloadable
    chapter audio (no reviewed segments). Same row shape as
    ``preload_recitations``; chapters come from the audio manifest sidecar (not
    ``segments.json``, which WIP reciters don't have). The assembled payload is
    TTL-cached (``PRELOAD_AUDIO_CATALOG_TTL_S``). On catalog failure returns
    ``{"recitations": [], "error": "..."}``.
    """
    global _AUDIO_RECITATIONS_CACHE, _AUDIO_RECITATIONS_AT
    mark_endpoint_entry()
    set_stage("catalog")

    now = time.monotonic()
    with _AUDIO_RECITATIONS_LOCK:
        if (
            _AUDIO_RECITATIONS_CACHE is not None
            and (now - _AUDIO_RECITATIONS_AT) < PRELOAD_AUDIO_CATALOG_TTL_S
        ):
            return _AUDIO_RECITATIONS_CACHE

    try:
        payload = _build_audio_recitations()
    except Exception as e:  # noqa: BLE001 — surface as an error payload
        log_error(
            error_code="catalog_load_failed",
            endpoint="preload_audio_recitations",
            stage="catalog",
            exception=e,
            user_id=get_user_id(request) if request else "unknown",
            message=f"Preload audio catalog load failed: {e}",
        )
        return {"recitations": [], "error": f"Catalog load failed: {e}"}

    with _AUDIO_RECITATIONS_LOCK:
        _AUDIO_RECITATIONS_CACHE = payload
        _AUDIO_RECITATIONS_AT = time.monotonic()
    return payload


# ---------------------------------------------------------------------------
# Endpoint 4 — chapter audio URL for an audio-only recitation (no segments)
# ---------------------------------------------------------------------------


def preload_audio(recitation, chapter, request: gr.Request = None):
    """Directly-fetchable full-chapter audio URL for a recitation / chapter.

    Audio-only counterpart of ``preload_segments``: no segments, no clipping —
    just the whole-chapter ``audio_url`` (bucket Space-route when the mp3 is on
    the mount, else the manifest's CDN URL). Accepts any audio-eligible delivery
    (released or audio-only). Errors return ``{"error", "audio_url": ""}``.
    """
    mark_endpoint_entry()
    set_stage("validate")

    slug = (recitation or "").strip() if isinstance(recitation, str) else ""
    if not slug:
        log_error(
            error_code="invalid_recitation", endpoint="preload_audio",
            stage="validate", user_id=get_user_id(request) if request else "unknown",
            message="recitation (delivery slug) is required",
        )
        return {"error": "recitation is required", "audio_url": ""}

    try:
        chapter = int(chapter)
    except (TypeError, ValueError):
        return {"error": "chapter must be an integer", "audio_url": ""}

    if slug not in repo_loader._list_audio_delivery_slugs():
        return {"error": f"Unknown or unavailable recitation {slug}", "audio_url": ""}

    set_stage("build")
    try:
        url = repo_loader.resolve_chapter_audio_url(slug, chapter)
    except Exception as e:  # noqa: BLE001 — surface as an error payload
        log_error(
            error_code="preload_audio_failed", endpoint="preload_audio",
            stage="build", exception=e,
            user_id=get_user_id(request) if request else "unknown",
            message=f"Preload audio resolve failed: {e}",
            context={"recitation": slug, "chapter": chapter},
        )
        return {"error": f"Failed to resolve audio: {e}", "audio_url": ""}

    audio_url = _absolute_audio_url(url, request)
    if not audio_url:
        return {"error": "No audio available for this recitation/chapter", "audio_url": ""}

    return {"recitation": slug, "chapter": chapter, "audio_url": audio_url}