funkwlmpv/src/fw_api.py

from src.mpv_control import player
import requests, json, time
from loguru import logger

with open('.auth.json', 'rt') as f:
    auth = json.loads(f.read())

s = requests.Session()
instance = 'fw.ponychord.rocks'
token = auth.get(instance)

if token:
    s.headers.update({
        "Authorization": "Bearer " + token
    })
    player.http_header_fields = ['Authorization: ' + 'Bearer ' + token]
else:
    s.get(f'https://{instance}/') # Get cookies from unauthorized instance for working some functionality (radios)
    player.http_header_fields = []


def select_instance(new_instance=None):
    global instance
    instance = new_instance
    with open('.auth.json', 'rt') as f:
        auth = json.loads(f.read())
    new_token = auth.get(instance)
    s.headers.update({"Authorization": None})
    player.http_header_fields = []
    if new_token:
        s.get(f'https://{instance}')
        s.headers.update({
                "Authorization": "Bearer " + new_token
        })
        player.http_header_fields = ['Authorization: ' + 'Bearer ' + new_token]


def concatinate_endpoint(endpoint):
    return 'https://' + instance + endpoint


def list_tracks(page=None, q=None, artist=None, album=None, favourites=None):
    '''This function get tracks by params'''
    params = {
            'page': page,
            'q': q,
            'artist': artist,
            'album': album,
            'favourites': favourites
    }

    r = requests.get(f'https://{instance}/api/v1/tracks', params)
    return r.json()


def list_libraries(page=None, page_size=None, q=None, scope='all'):
    params = {
            'page': page,
            'page_size': page_size,
            'q': q,
            'scope': scope,
    }
    r = requests.get(f'https://{instance}/api/v1/libraries', params)
    return r.json()


# [FunkWhale radios]
def get_radios():
    r = s.get(f'https://{instance}/api/v1/radios/radios/')
    return r.json()


def post_radio_session(requested_radio):
    r = s.post(f'https://{instance}/api/v1/radios/sessions/', json=requested_radio)
    return r.json()


def get_track_radio(radio_session):
    r = s.post(f'https://{instance}/api/v1/radios/tracks/',json=radio_session)
    try:
        r.raise_for_status()
        return r.json()
    except:
        logger.exception('Radio: get next track failed')