funkwlmpv/src/fw_api.py

197 lines
5.2 KiB
Python
Raw Normal View History

2022-11-08 17:58:27 +00:00
from src.mpv_control import set_http_header
2022-11-01 10:15:28 +00:00
import requests, json, time
2022-11-09 10:11:36 +00:00
import os
2022-11-01 10:15:28 +00:00
from loguru import logger
2022-11-09 10:11:36 +00:00
auth_file = '.auth.json'
if os.path.exists(auth_file):
with open('.auth.json', 'rt') as f:
auth = json.loads(f.read())
else:
# The default umask is 0o22 which turns off write permission of group and others
os.umask(0)
descriptor = os.open(
path=auth_file,
flags=(
os.O_WRONLY # access mode: write only
| os.O_CREAT # create if not exists
| os.O_TRUNC # truncate the file to zero
),
mode=0o600)
with open(descriptor, 'wt') as f:
f.write('{}')
auth = {}
2022-11-01 10:15:28 +00:00
s = requests.Session()
2022-11-01 10:15:28 +00:00
instance = 'fw.ponychord.rocks'
token = auth.get(instance)
if token:
s.headers.update({
"Authorization": "Bearer " + token,
"Accept-encoding": 'gzip'
})
2022-11-08 17:58:27 +00:00
set_http_header(['Authorization: ' + 'Bearer ' + token])
else:
s.headers.update({"Accept-encoding": 'gzip'})
s.get(f'https://{instance}/') # Get cookies from unauthorized instance for working some functionality (radios)
2022-11-08 17:58:27 +00:00
set_http_header()
def select_instance(new_instance=None):
global instance
instance = new_instance
2022-11-09 10:11:36 +00:00
with open(auth_file, 'rt') as f:
auth = json.loads(f.read())
new_token = auth.get(instance)
s.headers.update({"Authorization": None,
"Accept-encoding": 'gzip'})
2022-11-08 17:58:27 +00:00
set_http_header()
if new_token:
s.get(f'https://{instance}')
s.headers.update({
"Authorization": "Bearer " + new_token,
"Accept-encoding": 'gzip'
})
player.http_header_fields = ['Authorization: ' + 'Bearer ' + new_token]
2022-11-08 17:58:27 +00:00
set_http_header(['Authorization: ' + 'Bearer ' + token])
2022-11-01 10:15:28 +00:00
def concatinate_endpoint(endpoint):
return 'https://' + instance + endpoint
2022-11-08 17:58:27 +00:00
@logger.catch
def get_tracks(page=None, q=None, artist=None, album=None, favourites=None, include_channels=None, pg=None):
'''This function get tracks by params'''
params = {
'page': page,
'q': q,
'artist': artist,
'album': album,
'favourites': favourites,
'include_channels': include_channels
}
2022-11-08 17:58:27 +00:00
if pg:
r = s.get(pg)
else:
r = s.get(f'https://{instance}/api/v1/tracks', params=params)
return r.json()
2022-11-08 17:58:27 +00:00
@logger.catch
2022-11-09 11:26:23 +00:00
def get_artists(page=None, q=None, artist=None, album=None, favourites=None, refresh=False, pg=None):
2022-11-08 17:58:27 +00:00
'''This function get artists by params'''
params = {
'page': page,
'q': q,
'artist': artist,
'album': album,
2022-11-09 11:26:23 +00:00
'favourites': favourites,
'refresh': refresh
2022-11-08 17:58:27 +00:00
}
if pg:
r = s.get(pg)
else:
r = s.get(f'https://{instance}/api/v1/artists', params=params)
return r.json()
2022-11-08 17:58:27 +00:00
@logger.catch
2022-11-09 11:26:23 +00:00
def get_albums(page=None, q=None, artist=None, include_channels=None, refresh=False, pg=None):
2022-11-08 17:58:27 +00:00
'''This function get artists by params'''
params = {
'page': page,
'q': q,
'artist': artist,
2022-11-09 11:26:23 +00:00
'include_channels': include_channels,
'refresh': refresh
2022-11-08 17:58:27 +00:00
}
if pg:
r = s.get(pg)
else:
r = s.get(f'https://{instance}/api/v1/albums', params=params)
return r.json()
@logger.catch
def get_channels(page=None, q=None, tag=None, pg=None):
params = {
'page': page,
'q': q,
'tag': tag
}
if pg:
r = s.get(pg)
else:
r = s.get(f'https://{instance}/api/v1/channels', params=params)
return r.json()
@logger.catch
def list_libraries(page=None, page_size=None, q=None, scope='all', pg=None):
2022-11-08 00:29:44 +00:00
params = {
'page': page,
'page_size': page_size,
'q': q,
'scope': scope,
}
if pg:
r = s.get(pg)
else:
r = s.get(f'https://{instance}/api/v1/libraries', params=params)
2022-11-08 00:29:44 +00:00
return r.json()
@logger.catch
def federate_search_by_url(object):
params = {
'object': object
}
r = s.post(f'https://{instance}/api/v1/federation/fetches', json=params)
return r.json()
2022-11-08 19:23:19 +00:00
@logger.catch
def favorite_track(track_id):
r = s.post(f'https://{instance}/api/v1/favorites/tracks', json={'track': int(track_id)})
r.raise_for_status()
return r.json
@logger.catch
def unfavorite_track(track_id):
r = s.post(f'https://{instance}/api/v1/favorites/tracks/delete', json={'track': int(track_id)})
r.raise_for_status()
return r.json
@logger.catch
def hide_content(content):
'''This function hide content (write permission)'''
r = s.post(f'https://{instance}/api/v1/moderation/content-filters/', json=content)
r.raise_for_status()
return r.json
2022-11-01 10:15:28 +00:00
# [FunkWhale radios]
def get_radios():
r = s.get(f'https://{instance}/api/v1/radios/radios/')
2022-11-01 10:15:28 +00:00
return r.json()
def post_radio_session(requested_radio):
r = s.post(f'https://{instance}/api/v1/radios/sessions/', json=requested_radio)
2022-11-01 10:15:28 +00:00
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')
2022-11-01 10:15:28 +00:00