import requests import os import hashlib import random import string BASE = os.environ["SUBSONIC_URL"] + "/rest" def _auth_params(): salt = "".join(random.choices(string.ascii_letters + string.digits, k=6)) token = hashlib.md5((os.environ["SUBSONIC_PASS"] + salt).encode()).hexdigest() return { "u": os.environ["SUBSONIC_USER"], "t": token, "s": salt, "v": "1.16.1", "c": os.environ["SUBSONIC_CLIENT"], "f": "json", } def _req(endpoint, **params): p = _auth_params() p.update(params) r = requests.get(f"{BASE}/{endpoint}.view", params=p) r.raise_for_status() return r.json()["subsonic-response"] def search_song(query): r = _req("search3", query=query, songCount=1) songs = r.get("searchResult3", {}).get("song", []) return songs[0]["id"] if songs else None def create_playlist(name): r = _req("createPlaylist", name=name) return r["playlist"]["id"] def add_to_playlist(pid, song_ids): for sid in song_ids: _req("updatePlaylist", playlistId=pid, songIdToAdd=sid) def get_playlists(): r = _req("getPlaylists") return r["playlists"]["playlist"] def get_playlist_by_name(name): for p in get_playlists(): if p["name"] == name: return p return None def get_playlist_songs(pid): r = _req("getPlaylist", id=pid) entries = r["playlist"].get("entry", []) return {e["id"] for e in entries}