#!/usr/bin/env python3
"""Media server for the Hope voice assistant.

A small FastAPI service that owns the permanent music library and player
coordination:

- Library: downloaded YouTube Music tracks (mp3, transcoded via yt-dlp +
  ffmpeg) and user-added live audio stream URLs, persisted in library.json.
- Search: yt-dlp ``ytsearch`` (YouTube Music pages share video ids), used by
  Hope's media MCP toolset and by the web UI search box.
- Download jobs: one-at-a-time worker thread; progress is parsed from
  yt-dlp stdout and surfaced through GET /media/api/state.
- Player state: per-room desired state (loaded track, playing/paused/stopped,
  seek hints). The browser is the actual player and reconciles against this;
  it reports real playback state back so Hope's voice tools can answer
  "what's playing".

Security posture matches the token server: bind 127.0.0.1 only, reachability
exclusively through the nginx HTTPS proxy (/media/api/). File bytes for
downloaded tracks are served as nginx static files (/media/files/, native
HTTP Range support for seeking); this service only proxies external stream
URLs (mixed-content: an http:// radio URL would be blocked by the browser,
so it is fetched server-side and re-served over HTTPS).
"""
from __future__ import annotations

import json
import os
import re
import subprocess
import sys
import threading
import time
import uuid
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from typing import Any

import httpx
import uvicorn
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse, RedirectResponse, StreamingResponse

MEDIA_DIR = Path(os.environ.get("MEDIA_DIR", "/data/media"))
FILES_DIR = MEDIA_DIR / "files"
THUMBS_DIR = MEDIA_DIR / "thumbs"
JOBS_DIR = MEDIA_DIR / "jobs"
LIBRARY_FILE = MEDIA_DIR / "library.json"
COOKIES_FILE = os.environ.get("YTDLP_COOKIES", "")
MEDIA_HOST = os.environ.get("MEDIA_SERVER_HOST", "127.0.0.1")
MEDIA_PORT = int(os.environ.get("MEDIA_SERVER_PORT", "8092"))

DOWNLOAD_TIMEOUT = int(os.environ.get("MEDIA_DOWNLOAD_TIMEOUT", "600"))
SEARCH_TIMEOUT = int(os.environ.get("MEDIA_SEARCH_TIMEOUT", "45"))

app = FastAPI(title="hope-media", docs_url=None, redoc_url=None, openapi_url=None)

_lock = threading.Lock()          # guards library + rooms below
_library: dict[str, Any] = {"tracks": {}, "version": 1}
_rooms: dict[str, dict[str, Any]] = {}
_jobs: dict[str, dict[str, Any]] = {}
_executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="media-dl")

for _d in (FILES_DIR, THUMBS_DIR, JOBS_DIR):
    _d.mkdir(parents=True, exist_ok=True)


# ── Library store ───────────────────────────────────────────────────────────

def _load_library() -> None:
    if LIBRARY_FILE.exists():
        try:
            data = json.loads(LIBRARY_FILE.read_text(encoding="utf-8"))
            tracks = data.get("tracks", {})
            if isinstance(tracks, dict):
                _library["tracks"] = tracks
                _library["version"] = int(data.get("version", 1))
        except (json.JSONDecodeError, OSError):
            # Corrupt library: keep the broken file aside, start fresh.
            try:
                LIBRARY_FILE.replace(LIBRARY_FILE.with_suffix(".json.broken"))
            except OSError:
                pass


def _save_library() -> None:
    tmp = LIBRARY_FILE.with_suffix(".json.tmp")
    tmp.write_text(json.dumps(_library), encoding="utf-8")
    tmp.replace(LIBRARY_FILE)


def _bump_version() -> int:
    _library["version"] = int(_library.get("version", 1)) + 1
    return _library["version"]


def _track_dict(track: dict) -> dict:
    out = dict(track)
    if track["kind"] == "youtube":
        if track.get("status") == "ready":
            out["audio_url"] = f"/media/files/{track['id']}.mp3"
        out["art_url"] = (
            f"/media/thumbs/{track['id']}.jpg"
            if (THUMBS_DIR / f"{track['id']}.jpg").exists()
            else None
        )
    else:
        out["audio_url"] = f"/media/api/stream/{track['id']}"
    return out


def _get_track(track_id: str) -> dict | None:
    return _library["tracks"].get(track_id)


def _upsert_track(track: dict) -> None:
    _library["tracks"][track["id"]] = track
    _bump_version()
    _save_library()


_load_library()


# ── Player state (per room) ─────────────────────────────────────────────────

def _room_state(room: str) -> dict:
    return _rooms.setdefault(
        room,
        {"track_id": None, "desired": "stopped", "reported": None,
         "seek_hint": None, "choices": []},
    )


def _room_snapshot(room: str) -> dict:
    st = _room_state(room)
    track = _get_track(st["track_id"]) if st["track_id"] else None
    pending = [
        {
            "job_id": j["id"],
            "track_id": j["track_id"],
            "stage": j["stage"],
            "progress": j["progress"],
            "error": j.get("error"),
        }
        for j in _jobs.values()
        if j["stage"] in ("queued", "downloading", "transcoding")
    ]
    return {
        "room": room,
        "track": _track_dict(track) if track else None,
        "desired": st["desired"],
        "reported": st["reported"],
        "seek_hint": st["seek_hint"],
        "choices": st["choices"],
        "pending": pending,
        "version": _library["version"],
    }


# ── yt-dlp helpers ──────────────────────────────────────────────────────────

def _ytdlp_base_args() -> list[str]:
    args = [sys.executable, "-m", "yt_dlp", "--no-playlist", "--no-warnings"]
    if COOKIES_FILE and Path(COOKIES_FILE).exists():
        args += ["--cookies", COOKIES_FILE]
    return args


_PROGRESS_RE = re.compile(r"\[download\]\s+([\d.]+)%")


def _run_search(query: str, limit: int) -> list[dict]:
    """Flat ytsearch lookup — cheap, no download, works anonymously."""
    cmd = _ytdlp_base_args() + [
        "--flat-playlist", "--dump-json",
        f"ytsearch{limit}:{query}",
    ]
    try:
        proc = subprocess.run(
            cmd, capture_output=True, text=True, timeout=SEARCH_TIMEOUT, check=False
        )
    except subprocess.TimeoutExpired:
        raise HTTPException(504, "YouTube Music search timed out")
    if proc.returncode != 0 and not proc.stdout.strip():
        detail = (proc.stderr or "").strip().splitlines()[-1:] or ["unknown error"]
        raise HTTPException(502, f"Search failed: {detail[0][:300]}")
    results = []
    for line in proc.stdout.splitlines():
        line = line.strip()
        if not line.startswith("{"):
            continue
        try:
            entry = json.loads(line)
        except json.JSONDecodeError:
            continue
        video_id = entry.get("id")
        if not video_id:
            continue
        existing = _get_track(video_id)
        results.append({
            "video_id": video_id,
            "title": entry.get("title") or video_id,
            "channel": entry.get("uploader") or entry.get("channel") or "",
            "duration": entry.get("duration"),
            "view_count": entry.get("view_count"),
            "in_library": bool(existing),
            "track_id": existing["id"] if existing else None,
            "library_status": existing["status"] if existing else None,
        })
    return results


def _job_dict(job: dict) -> dict:
    return {k: job.get(k) for k in
            ("id", "track_id", "stage", "progress", "error", "started_at")}


def _persist_job(job: dict) -> None:
    try:
        (JOBS_DIR / f"{job['id']}.json").write_text(
            json.dumps(_job_dict(job)), encoding="utf-8"
        )
    except OSError:
        pass


def _fail_job(job: dict, track: dict, message: str) -> None:
    job["stage"] = "error"
    job["error"] = message[:500]
    _persist_job(job)
    with _lock:
        track["status"] = "failed"
        track["error"] = message[:500]
        _bump_version()
        _save_library()


def _download_worker(job: dict, track: dict, video_id: str) -> None:
    """Runs in the single-worker executor: yt-dlp download + mp3 transcode."""
    target = FILES_DIR / f"{track['id']}.mp3"
    cmd = _ytdlp_base_args() + [
        "-f", "bestaudio/best",
        "-x", "--audio-format", "mp3", "--audio-quality", "192K",
        "--embed-thumbnail", "--add-metadata",
        "--write-thumbnail", "--convert-thumbnails", "jpg",
        "--newline", "--progress",
        "--print-json",
        "--retries", "3", "--fragment-retries", "3",
        "-P", str(FILES_DIR),
        "-o", f"{track['id']}.%(ext)s",
        f"https://www.youtube.com/watch?v={video_id}",
    ]
    job["stage"] = "downloading"
    info: dict = {}
    try:
        proc = subprocess.Popen(
            cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
            text=True, errors="replace",
        )
        deadline = time.monotonic() + DOWNLOAD_TIMEOUT
        assert proc.stdout is not None and proc.stderr is not None
        for line in proc.stdout:
            if time.monotonic() > deadline:
                proc.kill()
                raise RuntimeError(f"Download exceeded {DOWNLOAD_TIMEOUT}s")
            line = line.strip()
            m = _PROGRESS_RE.search(line)
            if m:
                job["progress"] = min(85.0, float(m.group(1)) * 0.85)
                _persist_job(job)
            elif line.startswith(("[ExtractAudio]", "[Metadata]", "[EmbedThumbnail]")):
                job["stage"] = "transcoding"
                job["progress"] = max(job["progress"], 86.0)
                _persist_job(job)
            elif line.startswith("{"):
                try:
                    info = json.loads(line)
                except json.JSONDecodeError:
                    pass
        proc.wait(timeout=30)
        if proc.returncode != 0:
            stderr_tail = (proc.stderr.read() or "").strip().splitlines()
            detail = stderr_tail[-1] if stderr_tail else f"exit code {proc.returncode}"
            raise RuntimeError(detail)
        if not target.exists():
            raise RuntimeError("yt-dlp finished but no mp3 was produced")
    except Exception as exc:  # noqa: BLE001 — any failure marks the track failed
        _fail_job(job, track, str(exc))
        return

    thumb_src = FILES_DIR / f"{track['id']}.jpg"
    try:
        if thumb_src.exists():
            thumb_src.replace(THUMBS_DIR / f"{track['id']}.jpg")
    except OSError:
        pass

    job["stage"] = "ready"
    job["progress"] = 100.0
    _persist_job(job)
    with _lock:
        track["status"] = "ready"
        track["error"] = None
        if info:
            if info.get("title"):
                track["title"] = info["title"]
            if info.get("uploader") or info.get("channel"):
                track["channel"] = info.get("uploader") or info.get("channel")
            if info.get("duration"):
                track["duration"] = round(float(info["duration"]))
        _bump_version()
        _save_library()


def _start_download(track: dict, video_id: str) -> dict:
    job = {
        "id": uuid.uuid4().hex[:12],
        "track_id": track["id"],
        "stage": "queued",
        "progress": 0.0,
        "error": None,
        "started_at": time.time(),
    }
    _jobs[job["id"]] = job
    _persist_job(job)
    _executor.submit(_download_worker, job, track, video_id)
    return job


# ── API: health, library ────────────────────────────────────────────────────

@app.get("/media/api/healthz")
def healthz() -> dict:
    return {"ok": True, "version": _library["version"]}


@app.get("/media/api/library")
def get_library() -> dict:
    with _lock:
        tracks = sorted(
            (_track_dict(t) for t in _library["tracks"].values()),
            key=lambda t: t.get("added_at", 0),
            reverse=True,
        )
        return {"tracks": tracks, "version": _library["version"]}


def _delete_track_files(track: dict) -> None:
    if track["kind"] != "youtube":
        return
    for path in (
        FILES_DIR / f"{track['id']}.mp3",
        FILES_DIR / f"{track['id']}.jpg",
        THUMBS_DIR / f"{track['id']}.jpg",
    ):
        try:
            path.unlink(missing_ok=True)
        except OSError:
            pass


@app.delete("/media/api/tracks/{track_id}")
def delete_track(track_id: str) -> dict:
    with _lock:
        track = _get_track(track_id)
        if not track:
            raise HTTPException(404, "unknown track")
        del _library["tracks"][track_id]
        _bump_version()
        _save_library()
        for room, st in _rooms.items():
            if st["track_id"] == track_id:
                st["track_id"] = None
                st["desired"] = "stopped"
    _delete_track_files(track)
    return {"ok": True}


# ── API: search + play + control ────────────────────────────────────────────

@app.post("/media/api/search")
async def search(request: Request) -> dict:
    body = await request.json()
    query = str(body.get("query", "")).strip()
    if not query:
        raise HTTPException(400, "query is required")
    limit = max(1, min(int(body.get("limit", 6)), 15))
    return {"query": query, "results": _run_search(query, limit)}


def _resolve_play_target(body: dict) -> dict:
    """Find or create the track a play request refers to.

    Accepts video_id (YouTube), track_id (existing library entry) or a raw
    url+name (live stream). Returns the library track dict.
    """
    video_id = str(body.get("video_id") or "").strip()
    track_id = str(body.get("track_id") or "").strip()
    url = str(body.get("url") or "").strip()

    if video_id:
        existing = _get_track(video_id)
        if existing:
            return existing
        track = {
            "id": video_id,
            "kind": "youtube",
            "video_id": video_id,
            "title": str(body.get("title") or video_id),
            "channel": str(body.get("channel") or ""),
            "duration": body.get("duration"),
            "status": "downloading",
            "error": None,
            "added_at": time.time(),
        }
        _upsert_track(track)
        _start_download(track, video_id)
        return track

    if track_id:
        track = _get_track(track_id)
        if not track:
            raise HTTPException(404, "unknown track")
        return track

    if url:
        if not re.match(r"^https?://", url):
            raise HTTPException(400, "url must be http(s)")
        track_id = "s" + uuid.uuid4().hex[:10]
        track = {
            "id": track_id,
            "kind": "stream",
            "video_id": None,
            "title": str(body.get("name") or url),
            "channel": str(body.get("channel") or ""),
            "duration": None,
            "status": "ready",
            "error": None,
            "url": url,
            "added_at": time.time(),
        }
        _upsert_track(track)
        return track

    raise HTTPException(400, "one of video_id, track_id or url is required")


@app.post("/media/api/play")
async def play(request: Request) -> dict:
    body = await request.json()
    room = str(body.get("room") or "default")
    track = _resolve_play_target(body)
    with _lock:
        st = _room_state(room)
        st["track_id"] = track["id"]
        # start_paused carries a track into a new scope without starting it —
        # the UI uses this when a conversation begins (talking pauses music).
        st["desired"] = "paused" if body.get("start_paused") else "playing"
        st["choices"] = []  # a pick resolves any open suggestion list
        st["seek_hint"] = None
        snapshot = _room_snapshot(room)
    return {"ok": True, **snapshot}


@app.post("/media/api/control")
async def control(request: Request) -> dict:
    body = await request.json()
    room = str(body.get("room") or "default")
    action = str(body.get("action") or "")
    with _lock:
        st = _room_state(room)
        if action == "pause":
            st["desired"] = "paused"
        elif action == "resume":
            if not st["track_id"]:
                raise HTTPException(409, "nothing loaded")
            st["desired"] = "playing"
        elif action == "stop":
            st["desired"] = "stopped"
        elif action == "seek":
            st["seek_hint"] = {
                "position": max(0.0, float(body.get("position", 0))),
                "ts": time.time(),
            }
        elif action == "skip":
            current = st["track_id"]
            candidates = sorted(
                (
                    t for t in _library["tracks"].values()
                    if t.get("status") == "ready" and t["id"] != current
                ),
                key=lambda t: t.get("added_at", 0),
                reverse=True,
            )
            if candidates:
                st["track_id"] = candidates[0]["id"]
                st["desired"] = "playing"
                st["seek_hint"] = None
            else:
                st["desired"] = "stopped"
        else:
            raise HTTPException(400, f"unknown action: {action}")
        snapshot = _room_snapshot(room)
    return {"ok": True, **snapshot}


@app.post("/media/api/choices")
async def choices(request: Request) -> dict:
    """Publish Hope's song suggestions so the UI can render clickable picks."""
    body = await request.json()
    room = str(body.get("room") or "default")
    raw = body.get("choices") or []
    cleaned = []
    for c in raw[:5]:
        if not isinstance(c, dict) or not c.get("video_id"):
            continue
        cleaned.append({
            "video_id": str(c["video_id"]),
            "title": str(c.get("title") or c["video_id"]),
            "channel": str(c.get("channel") or ""),
            "duration": c.get("duration"),
        })
    with _lock:
        _room_state(room)["choices"] = cleaned
    return {"ok": True, "choices": cleaned}


@app.get("/media/api/state")
def state(room: str = "default") -> dict:
    with _lock:
        return _room_snapshot(room)


@app.post("/media/api/report")
async def report(request: Request) -> dict:
    """The browser reports real playback state (source of truth for the player)."""
    body = await request.json()
    room = str(body.get("room") or "default")
    with _lock:
        st = _room_state(room)
        st["reported"] = {
            "track_id": body.get("track_id"),
            "state": str(body.get("state") or "stopped"),
            "position": float(body.get("position") or 0.0),
            "duration": body.get("duration"),
            "ts": time.time(),
        }
    return {"ok": True}


@app.post("/media/api/streams")
async def add_stream(request: Request) -> dict:
    body = await request.json()
    track = _resolve_play_target({"url": body.get("url"), "name": body.get("name")})
    return {"ok": True, "track": _track_dict(track)}


# ── API: stream proxy (external http(s) audio over same-origin HTTPS) ───────

_HOP_HEADERS = ("content-type", "content-range", "accept-ranges", "content-length")


@app.get("/media/api/stream/{track_id}")
async def stream_proxy(track_id: str, request: Request):
    with _lock:
        track = _get_track(track_id)
    if not track:
        raise HTTPException(404, "unknown track")
    if track["kind"] == "youtube":
        # Downloaded tracks are nginx static files (native Range support).
        if track.get("status") == "ready":
            return RedirectResponse(f"/media/files/{track_id}.mp3")
        raise HTTPException(409, f"track is {track.get('status')}, not ready")

    range_header = request.headers.get("range")
    headers = {"Range": range_header} if range_header else {"Range": "bytes=0-"}
    client = httpx.AsyncClient(timeout=httpx.Timeout(30, read=600), follow_redirects=True)
    try:
        upstream = await client.send(client.build_request("GET", track["url"], headers=headers), stream=True)
    except httpx.HTTPError as exc:
        await client.aclose()
        raise HTTPException(502, f"upstream stream failed: {exc}")

    if upstream.status_code >= 400:
        await upstream.aclose()
        await client.aclose()
        raise HTTPException(502, f"upstream returned {upstream.status_code}")

    out_headers = {
        name: value
        for name, value in upstream.headers.items()
        if name.lower() in _HOP_HEADERS
    }
    media_type = upstream.headers.get("content-type", "audio/mpeg")

    async def relay():
        try:
            async for chunk in upstream.aiter_bytes(64 * 1024):
                yield chunk
        finally:
            await upstream.aclose()
            await client.aclose()

    return StreamingResponse(
        relay(),
        status_code=upstream.status_code,
        media_type=media_type,
        headers=out_headers,
    )


@app.exception_handler(HTTPException)
async def http_error(_request: Request, exc: HTTPException):
    return JSONResponse({"error": exc.detail}, status_code=exc.status_code)


if __name__ == "__main__":
    uvicorn.run(app, host=MEDIA_HOST, port=MEDIA_PORT, log_level="info")