79 lines
2.8 KiB
Python
79 lines
2.8 KiB
Python
"""MusicFetch REST API. Plain HTTP behind an upstream TLS reverse proxy."""
|
|
import os
|
|
|
|
from fastapi import Depends, FastAPI, Header, HTTPException, Query
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from . import actions, jobs, mf
|
|
|
|
API_KEY = os.environ.get("MUSICFETCH_API_KEY", "")
|
|
ROOT = os.environ.get("MUSICFETCH_ROOT", "/media/music")
|
|
|
|
app = FastAPI(title="MusicFetch API")
|
|
|
|
|
|
def require_key(x_api_key: str = Header(default="")):
|
|
if not API_KEY or x_api_key != API_KEY:
|
|
raise HTTPException(status_code=401, detail="Invalid API key.")
|
|
|
|
|
|
@app.exception_handler(HTTPException)
|
|
async def _http_exc(_req, exc: HTTPException):
|
|
# Always return a Siri-speakable {"message": ...} body.
|
|
return JSONResponse(status_code=exc.status_code, content={"message": exc.detail})
|
|
|
|
|
|
@app.get("/health")
|
|
def health():
|
|
return {"status": "ok"}
|
|
|
|
|
|
def _hit_public(hit) -> dict:
|
|
return {"source": hit.source, "kind": hit.kind, "artist": hit.artist,
|
|
"album": hit.album, "title": hit.title, "year": hit.year}
|
|
|
|
|
|
def _job_public(job) -> dict:
|
|
return {"message": job.message, "job_id": job.id, "status": job.status,
|
|
"hit": _hit_public(job.hit) if job.hit is not None else None,
|
|
"result": job.result, "error": job.error}
|
|
|
|
|
|
@app.post("/fetch", dependencies=[Depends(require_key)])
|
|
def fetch(q: str = Query(..., min_length=1),
|
|
quality: str = Query("best"),
|
|
source: str = Query("auto")):
|
|
if quality not in mf.QUALITY_CHOICES:
|
|
raise HTTPException(status_code=422, detail=f"Invalid quality '{quality}'.")
|
|
if source not in ("auto", "lidarr", "youtube"):
|
|
raise HTTPException(status_code=422, detail=f"Invalid source '{source}'.")
|
|
|
|
yt_first = source == "youtube"
|
|
hits = mf.build_combined_hits(q, 10, yt_first,
|
|
lidarr_only=(source == "lidarr"),
|
|
yt_only=(source == "youtube"))
|
|
if not hits:
|
|
raise HTTPException(status_code=404, detail=f"No results found for '{q}'.")
|
|
|
|
chosen = mf.pick(hits, q, True, yt_first)
|
|
if chosen is None:
|
|
raise HTTPException(status_code=404, detail=f"No results found for '{q}'.")
|
|
|
|
job = jobs.create_job(hit=chosen, message=actions.started_message(chosen))
|
|
response = _job_public(job) # snapshot "queued" state before background thread starts
|
|
jobs.run_job(
|
|
job.id,
|
|
lambda: actions.perform_fetch(chosen, hits, quality, ROOT),
|
|
done_message=actions.done_message(chosen),
|
|
fail_message=actions.failed_message(chosen),
|
|
)
|
|
return response
|
|
|
|
|
|
@app.get("/jobs/{job_id}", dependencies=[Depends(require_key)])
|
|
def job_status(job_id: str):
|
|
job = jobs.get_job(job_id)
|
|
if job is None:
|
|
raise HTTPException(status_code=404, detail="No such job.")
|
|
return _job_public(job)
|