#!/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 (raw best-audio files, saved by
  yt-dlp with no transcode) 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. Ready library tracks
  that match the query are returned first so Hope prefers a downloaded copy.
- Download jobs: one-at-a-time worker thread; progress is parsed from
  yt-dlp stderr and surfaced through GET /media/api/state. Playback starts
  while the file is still growing (progressive streaming), and the finished
  file stays in the library for instant replay.
- 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/). Growing downloads
are streamed by this service (chunked, no Content-Length); finished files are
served here too with native Range support for seeking. Live stream URLs are
proxied (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 asyncio
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 FileResponse, JSONResponse, 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")
# 8093, NOT 8092: 8092 is occupied by an Apache/YOURLS stack on the Unraid
# deployment host (xNAS) — a bind conflict there looks like a healthy
# deployment because the healthcheck curl (without -f) accepts any response.
MEDIA_PORT = int(os.environ.get("MEDIA_SERVER_PORT", "8093"))

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)
    # The bind mount may be inspected/cleaned from the host by a non-root
    # user (Unraid share convention) — keep the dirs world-traversable.
    try:
        os.chmod(_d, 0o777)
    except OSError:
        pass


# ── 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":
        # The stream endpoint serves the growing file during download and the
        # finished file afterwards; the URL is stable across the transition
        # so the browser never has to reload mid-playback.
        if track.get("status") in ("downloading", "ready"):
            out["audio_url"] = f"/media/api/stream/{track['id']}"
        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.]+)%")

# Audio containers yt-dlp can produce for a raw best-audio download.
_AUDIO_EXTS = {
    ".webm", ".opus", ".ogg", ".m4a", ".mp4", ".m4b", ".aac",
    ".mp3", ".mka", ".flac", ".wav",
}
_AUDIO_MIME = {
    ".webm": "audio/webm",
    ".opus": "audio/opus",
    ".ogg": "audio/ogg",
    ".m4a": "audio/mp4",
    ".mp4": "audio/mp4",
    ".m4b": "audio/mp4",
    ".aac": "audio/aac",
    ".mp3": "audio/mpeg",
    ".mka": "audio/x-matroska",
    ".flac": "audio/flac",
    ".wav": "audio/wav",
}


def _media_type_for(name: str) -> str:
    if name.endswith(".part"):
        name = name[:-5]
    return _AUDIO_MIME.get(Path(name).suffix.lower(), "audio/mpeg")


def _youtube_media_file(track_id: str) -> Path | None:
    """The on-disk audio file for a youtube track, if any.

    During an active download yt-dlp writes ``{id}.{ext}.part`` and renames it
    to ``{id}.{ext}`` on completion — return whichever currently exists (the
    partial file wins, thumbnails and other sidecars are ignored).
    """
    complete: Path | None = None
    for p in FILES_DIR.glob(f"{track_id}.*"):
        suffix = p.suffix.lower()
        if suffix == ".part":
            if p.stat().st_size > 0:
                return p
        elif suffix in _AUDIO_EXTS:
            if complete is None or p.name < complete.name:
                complete = p
    return complete


async def _wait_for_media_file(track_id: str, timeout: float = 15.0) -> Path | None:
    """Wait for yt-dlp to create the output file after a play request."""
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        path = _youtube_media_file(track_id)
        if path is not None:
            return path
        await asyncio.sleep(0.5)
    return None


async def _growing_file(track_id: str, job: dict, chunk_size: int = 64 * 1024):
    """Yield bytes from a file that yt-dlp is still writing.

    The response stays open with no Content-Length until the download
    finishes, so the browser plays the track as it grows. The open descriptor
    keeps reading the old inode across yt-dlp's .part -> final rename, and the
    path is re-resolved if the file disappears between resolution and open.
    """
    fd = -1
    pos = 0
    stall = 0
    try:
        while True:
            if fd == -1:
                path = _youtube_media_file(track_id)
                if path is not None:
                    try:
                        fd = os.open(path, os.O_RDONLY)
                    except FileNotFoundError:
                        fd = -1  # renamed between glob and open — retry
            if fd != -1:
                data = await asyncio.to_thread(os.pread, fd, chunk_size, pos)
                if data:
                    pos += len(data)
                    stall = 0
                    yield data
                    continue
            with _lock:
                active = job.get("stage") == "downloading"
            if not active:
                break
            stall += 1
            if stall > 60:  # ~30s with no new bytes — stop waiting
                break
            await asyncio.sleep(0.5)
    finally:
        if fd != -1:
            os.close(fd)


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: raw best-audio download.

    No mp3 transcode — the audio container is streamed straight into the
    library so playback can begin while the file is still growing (the stream
    endpoint serves the partial file), and the finished file stays playable
    offline. Thumbnail handling is unchanged.
    """
    cmd = _ytdlp_base_args() + [
        "-f", "bestaudio/best",
        "--newline", "--progress",
        "--print-json",
        "--write-thumbnail", "--convert-thumbnails", "jpg",
        "--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 = {}
    # Progress lines go to stderr, the metadata JSON to stdout. Drain stderr
    # in a thread so the stdout read loop can never deadlock the pipe; the
    # tail is kept for the failure detail.
    stderr_tail: list[str] = []
    proc = None

    def _drain_stderr() -> None:
        assert proc is not None and proc.stderr is not None
        for line in proc.stderr:
            stderr_tail.append(line)
            if len(stderr_tail) > 200:
                stderr_tail.pop(0)
            m = _PROGRESS_RE.search(line)
            if m:
                job["progress"] = min(99.0, float(m.group(1)))
                _persist_job(job)

    try:
        proc = subprocess.Popen(
            cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
            text=True, errors="replace",
        )
        threading.Thread(target=_drain_stderr, daemon=True).start()
        deadline = time.monotonic() + DOWNLOAD_TIMEOUT
        assert proc.stdout is not None
        for line in proc.stdout:
            if time.monotonic() > deadline:
                proc.kill()
                proc.wait(timeout=30)
                raise RuntimeError(f"Download exceeded {DOWNLOAD_TIMEOUT}s")
            line = line.strip()
            if line.startswith("{"):
                try:
                    info = json.loads(line)
                except json.JSONDecodeError:
                    pass
        proc.wait(timeout=30)
        if proc.returncode != 0:
            detail = stderr_tail[-1] if stderr_tail else f"exit code {proc.returncode}"
            raise RuntimeError(detail.strip())
        media = _youtube_media_file(track["id"])
        if media is None:
            raise RuntimeError("yt-dlp finished but no audio file 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
        track["ext"] = media.suffix.lstrip(".").lower()
        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
    paths = [
        FILES_DIR / f"{track['id']}.{ext}" for ext in
        (s.lstrip(".") for s in _AUDIO_EXTS)
    ] + [FILES_DIR / f"{track['id']}.jpg", THUMBS_DIR / f"{track['id']}.jpg"]
    for path in paths:
        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 ────────────────────────────────────────────

def _library_search(query: str, limit: int) -> list[dict]:
    """Ready library tracks whose title matches the query.

    These are merged in front of the YouTube results so Hope prefers an
    already-downloaded copy over fetching a new one.
    """
    q = query.strip().lower()
    tokens = {t for t in re.split(r"\W+", q) if len(t) > 2}
    hits: list[dict] = []
    if not q:
        return hits
    for track in _library["tracks"].values():
        if track.get("kind") != "youtube" or track.get("status") != "ready":
            continue
        title = (track.get("title") or "").strip().lower()
        if not title:
            continue
        title_tokens = {t for t in re.split(r"\W+", title) if t}
        if not (q in title or title in q or (tokens and tokens <= title_tokens)):
            continue
        hits.append({
            "video_id": track["id"],
            "track_id": track["id"],
            "title": track.get("title") or track["id"],
            "channel": track.get("channel") or "",
            "duration": track.get("duration"),
            "view_count": None,
            "in_library": True,
            "library_status": "ready",
        })
    return hits[:limit]


@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")
    try:
        limit = max(1, min(int(body.get("limit", 6)), 15))
    except (TypeError, ValueError):
        raise HTTPException(400, "limit must be an integer")
    local = _library_search(query, limit)
    remote = _run_search(query, limit)
    seen = {r["video_id"] for r in local}
    merged = local + [r for r in remote if r["video_id"] not in seen]
    return {"query": query, "results": merged[: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":
            try:
                position = max(0.0, float(body.get("position", 0)))
            except (TypeError, ValueError):
                raise HTTPException(400, "position must be a number")
            st["seek_hint"] = {"position": position, "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")
    try:
        position = float(body.get("position") or 0.0)
    except (TypeError, ValueError):
        position = 0.0
    with _lock:
        st = _room_state(room)
        st["reported"] = {
            "track_id": body.get("track_id"),
            "state": str(body.get("state") or "stopped"),
            "position": position,
            "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)
        job = next(
            (j for j in _jobs.values()
             if j["track_id"] == track_id and j["stage"] == "downloading"),
            None,
        )
    if not track:
        raise HTTPException(404, "unknown track")

    if track["kind"] == "youtube":
        # Progressive download: an active download is streamed in real time
        # (chunked, no Content-Length) so playback starts while the file
        # grows; a finished file is served directly with native Range support.
        path = _youtube_media_file(track_id)
        if path is None and job is not None:
            path = await _wait_for_media_file(track_id)
        if path is None:
            raise HTTPException(
                404, f"audio file not available (track is {track.get('status')})"
            )
        if job is not None:
            return StreamingResponse(
                _growing_file(track_id, job),
                media_type=_media_type_for(path.name),
                headers={"Cache-Control": "no-store"},
            )
        return FileResponse(path, media_type=_media_type_for(path.name))

    range_header = request.headers.get("range")
    headers = {"Range": range_header} if range_header else {}
    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")