snowspeeder
ce8fd1e046
Rebuild & Deploy / rebuild (push) Successful in 1m0s
The Unraid deployment's /media/api requests were answered by an Apache/YOURLS stack squatting on 8092, surfacing as UI 503s; the CI healthcheck (curl without -f) counted that 503 as healthy, so run #15 deployed green against a broken media server. - media server port 8092 -> 8093 everywhere (server, MCP, agent, nginx, compose, CI, .env.example, docs, tests). - healthchecks use curl -f in compose and rebuild.yml so HTTP error statuses fail deployment. - tests default to the Unraid deployment (VOICE_WEB_URL= https://192.168.86.2:9178); token goes through nginx (loopback :8091 is unreachable off-host); bind-mount-coupled tests skip on remote targets. AGENTS.md documents the convention and both failure modes.
643 lines
22 KiB
Python
643 lines
22 KiB
Python
#!/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 tempfile
|
|
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")
|
|
# 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":
|
|
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 = {}
|
|
# stderr goes to a temp file: reading it from a PIPE while the stdout
|
|
# loop blocks could deadlock if yt-dlp ever fills the stderr buffer.
|
|
with tempfile.TemporaryFile(mode="w+", encoding="utf-8", errors="replace") as errfile:
|
|
try:
|
|
proc = subprocess.Popen(
|
|
cmd, stdout=subprocess.PIPE, stderr=errfile,
|
|
text=True, errors="replace",
|
|
)
|
|
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()
|
|
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:
|
|
errfile.seek(0)
|
|
stderr_tail = (errfile.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")
|
|
try:
|
|
limit = max(1, min(int(body.get("limit", 6)), 15))
|
|
except (TypeError, ValueError):
|
|
raise HTTPException(400, "limit must be an integer")
|
|
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":
|
|
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)
|
|
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 {}
|
|
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")
|