161 lines
4.1 KiB
Python
161 lines
4.1 KiB
Python
from fastapi import APIRouter, HTTPException, Query
|
|
from fastapi.responses import JSONResponse
|
|
from pydantic import BaseModel
|
|
from app import spotify
|
|
|
|
router = APIRouter(prefix="/player", tags=["player"])
|
|
|
|
|
|
class VolumeBody(BaseModel):
|
|
volume: int
|
|
|
|
|
|
class DeviceBody(BaseModel):
|
|
device_id: str
|
|
|
|
|
|
class PlayBody(BaseModel):
|
|
context_uri: str | None = None
|
|
uris: list[str] | None = None
|
|
device_id: str | None = None
|
|
|
|
|
|
def _sp():
|
|
return spotify.get_client()
|
|
|
|
|
|
@router.get("/current")
|
|
def current_track():
|
|
sp = _sp()
|
|
track = sp.current_playback()
|
|
if not track or not track.get("item"):
|
|
return {"playing": False}
|
|
|
|
item = track["item"]
|
|
artists = ", ".join(a["name"] for a in item["artists"])
|
|
album = item["album"]
|
|
image = album["images"][0]["url"] if album["images"] else ""
|
|
|
|
return {
|
|
"playing": track["is_playing"],
|
|
"track": item["name"],
|
|
"artists": artists,
|
|
"album": album["name"],
|
|
"image": image,
|
|
"progress_ms": track["progress_ms"],
|
|
"duration_ms": item["duration_ms"],
|
|
"volume": track["device"]["volume_percent"] if track.get("device") else None,
|
|
"device": track["device"]["name"] if track.get("device") else None,
|
|
"context_uri": track["context"]["uri"] if track.get("context") else None,
|
|
}
|
|
|
|
|
|
@router.get("/devices")
|
|
def get_devices():
|
|
sp = _sp()
|
|
result = sp.devices()
|
|
return result.get("devices", [])
|
|
|
|
|
|
@router.post("/play")
|
|
def play(body: PlayBody):
|
|
sp = _sp()
|
|
kwargs = {}
|
|
if body.uris:
|
|
kwargs["uris"] = body.uris
|
|
elif body.context_uri:
|
|
kwargs["context_uri"] = body.context_uri
|
|
if body.device_id:
|
|
kwargs["device_id"] = body.device_id
|
|
sp.start_playback(**kwargs)
|
|
return {"ok": True}
|
|
|
|
|
|
@router.post("/pause")
|
|
def pause():
|
|
sp = _sp()
|
|
sp.pause_playback()
|
|
return {"ok": True}
|
|
|
|
|
|
@router.post("/next")
|
|
def next_track():
|
|
sp = _sp()
|
|
sp.next_track()
|
|
return {"ok": True}
|
|
|
|
|
|
@router.post("/previous")
|
|
def previous_track():
|
|
sp = _sp()
|
|
sp.previous_track()
|
|
return {"ok": True}
|
|
|
|
|
|
@router.post("/volume")
|
|
def set_volume(body: VolumeBody):
|
|
if not 0 <= body.volume <= 100:
|
|
raise HTTPException(status_code=400, detail="El volumen debe estar entre 0 y 100")
|
|
sp = _sp()
|
|
sp.volume(body.volume)
|
|
return {"ok": True}
|
|
|
|
|
|
@router.post("/device")
|
|
def set_device(body: DeviceBody):
|
|
sp = _sp()
|
|
sp.transfer_playback(device_id=body.device_id, force_play=False)
|
|
return {"ok": True}
|
|
|
|
|
|
@router.get("/tracks")
|
|
def get_tracks(id: str = Query(...), type: str = Query(...)):
|
|
sp = _sp()
|
|
tracks = []
|
|
|
|
if type == "playlist":
|
|
results = sp.playlist_tracks(
|
|
id, limit=50,
|
|
fields="items(track(id,name,artists,duration_ms))",
|
|
)
|
|
for item in (results.get("items") or []):
|
|
t = item.get("track")
|
|
if t and t.get("id"):
|
|
tracks.append({
|
|
"id": t["id"],
|
|
"name": t["name"],
|
|
"artists": ", ".join(a["name"] for a in t["artists"]),
|
|
"duration_ms": t["duration_ms"],
|
|
})
|
|
|
|
elif type == "album":
|
|
results = sp.album_tracks(id, limit=50)
|
|
for t in (results.get("items") or []):
|
|
tracks.append({
|
|
"id": t["id"],
|
|
"name": t["name"],
|
|
"artists": ", ".join(a["name"] for a in t["artists"]),
|
|
"duration_ms": t["duration_ms"],
|
|
})
|
|
|
|
elif type == "artist":
|
|
results = sp.artist_top_tracks(id)
|
|
for t in (results.get("tracks") or []):
|
|
tracks.append({
|
|
"id": t["id"],
|
|
"name": t["name"],
|
|
"artists": ", ".join(a["name"] for a in t["artists"]),
|
|
"duration_ms": t["duration_ms"],
|
|
})
|
|
|
|
elif type == "track":
|
|
t = sp.track(id)
|
|
tracks.append({
|
|
"id": t["id"],
|
|
"name": t["name"],
|
|
"artists": ", ".join(a["name"] for a in t["artists"]),
|
|
"duration_ms": t["duration_ms"],
|
|
})
|
|
|
|
return tracks
|