snowspeeder
4f8eb00c9f
Rebuild & Deploy / rebuild (push) Successful in 59s
- Stream downloads as they grow: raw best-audio (no mp3 transcode) served chunked from the stream endpoint mid-download, then with native Range once complete. Songs play within seconds of hitting Play. - Fix searched songs never starting: reconcileMedia now wires the stream URL when a track that was preparing gains one; audio_url is stable for the whole lifecycle so the player never reloads mid-playback. - Progress bars now parse yt-dlp stderr (stdout never carried [download]). - Prefer downloaded songs: /media/api/search merges ready library matches ahead of YouTube results; search_music/play_song tooling and the agent prompt tell Hope to always pick in_library=true. - Mute button turns solid red while the mic is muted.
811 lines
28 KiB
Python
811 lines
28 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 (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")
|