"""MusicFetch REST API. Plain HTTP behind an upstream TLS reverse proxy.""" import os from fastapi import Depends, FastAPI, Header, HTTPException, Query from fastapi.exceptions import RequestValidationError 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.exception_handler(RequestValidationError) async def _validation_exc(_req, exc: RequestValidationError): return JSONResponse(status_code=422, content={"message": "Invalid or missing request parameters."}) @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: body = {"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} # Only a parked job carries choices; keep the key out of every other body. if job.choices: body["choices"] = job.choices return body @app.post("/fetch", dependencies=[Depends(require_key)]) def fetch(q: str = Query(..., min_length=1), quality: str = Query("best"), source: str = Query("auto"), auto: bool = Query(False), limit: int = Query(10, ge=1, le=25)): """The single entry point. A URL is grabbed directly — no prompt. Free text goes through Lidarr: the artist and album land in the library, Lidarr runs an interactive search, and the job parks at `awaiting_choice` with the releases. Pick one with POST /choose. `auto=true` grabs the best release and never asks. `source=youtube` skips Lidarr; `source=lidarr` drops the YouTube fallback. """ if quality not in mf.QUALITY_CHOICES: raise HTTPException(status_code=422, detail=f"Invalid quality '{quality}'.") if mf.is_url(q): if mf._is_direct_url(q): kind, title, hits = mf.probe_url(q) syn = mf.Hit(source="youtube", kind=kind, title=title, artist="") job = jobs.create_job(hit=syn, message=actions.url_started_message(kind, title)) response = _job_public(job) done_msg = actions.playlist_done_message if kind == "playlist" else actions.url_done_message jobs.run_job( job.id, lambda: actions.perform_url_fetch(q, kind, title, hits, quality, ROOT), done_message=done_msg, fail_message="Download failed.", ) return response # Non-direct link (Spotify/Apple/...): resolve via Odesli, then run the # normal Lidarr-first pick/dispatch with the exact YouTube track fallback. try: query, hits = mf.resolve_link_hits(q, limit) except mf.OdesliError: raise HTTPException(status_code=422, detail=f"Couldn't resolve {q}. Try the direct YouTube or SoundCloud link.") if not hits: raise HTTPException(status_code=404, detail=f"No results found for '{q}'.") chosen = mf.pick(hits, query, True, False) 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) jobs.run_job( job.id, # A URL is a direct grab, so never park it for a choice. lambda: actions.perform_fetch(chosen, hits, quality, ROOT, auto=True), done_message=lambda res: actions.done_message(chosen, res), fail_message=lambda e: actions.fail_text(chosen, e), ) return response 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, limit, 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_choice_job( job.id, lambda: actions.perform_fetch(chosen, hits, quality, ROOT, auto=auto, allow_youtube=(source != "lidarr")), choice_message=lambda res: actions.choice_message(chosen, res), done_message=lambda res: actions.done_message(chosen, res), fail_message=lambda e: actions.fail_text(chosen, e), ) return response @app.post("/choose", dependencies=[Depends(require_key)]) def choose(job_id: str = Query(..., min_length=1), n: int = Query(..., ge=1)): """Send the release the client picked to the download client. `n` is the number of a choice from the same job's `choices` list.""" try: job, release = jobs.take_choice(job_id, n) except KeyError: raise HTTPException(status_code=404, detail="No such job.") except jobs.ChoiceError as e: raise HTTPException(status_code=409, detail=str(e)) prev = job.result or {} artist, album = prev.get("artist", ""), prev.get("album", "") jobs.requeue(job, actions.grab_started_message(artist, album)) response = _job_public(job) jobs.run_job( job.id, lambda: actions.perform_grab(release, artist, album), done_message=actions.grab_message, fail_message=lambda e: (str(e) if isinstance(e, actions.FetchError) and str(e) else "Lidarr refused that release."), ) 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)