f2f4c7a908
We now get the playable station URL directly from the Radiobrowser API. This fixes the issue with playlists in the 'url' attribute. Does not work if played station tracking would be disabled. It _could_ work but we would need to create an additional API request for every single listed station, even the ones not enqueued for playing. This would then be way slower and put extra strain on the Radiobrowser API.
273 lines
10 KiB
Python
273 lines
10 KiB
Python
import logging
|
|
|
|
from flask import Flask, request, url_for, redirect, abort, make_response
|
|
|
|
import ycast.vtuner as vtuner
|
|
import ycast.radiobrowser as radiobrowser
|
|
import ycast.my_stations as my_stations
|
|
import ycast.generic as generic
|
|
import ycast.station_icons as station_icons
|
|
|
|
|
|
PATH_ROOT = 'ycast'
|
|
PATH_PLAY = 'play'
|
|
PATH_STATION = 'station'
|
|
PATH_SEARCH = 'search'
|
|
PATH_ICON = 'icon'
|
|
PATH_MY_STATIONS = 'my_stations'
|
|
PATH_RADIOBROWSER = 'radiobrowser'
|
|
PATH_RADIOBROWSER_COUNTRY = 'country'
|
|
PATH_RADIOBROWSER_LANGUAGE = 'language'
|
|
PATH_RADIOBROWSER_GENRE = 'genre'
|
|
PATH_RADIOBROWSER_POPULAR = 'popular'
|
|
|
|
my_stations_enabled = False
|
|
app = Flask(__name__)
|
|
|
|
|
|
def run(config, address='0.0.0.0', port=8010):
|
|
try:
|
|
check_my_stations_feature(config)
|
|
app.run(host=address, port=port)
|
|
except PermissionError:
|
|
logging.error("No permission to create socket. Are you trying to use ports below 1024 without elevated rights?")
|
|
|
|
|
|
def check_my_stations_feature(config):
|
|
global my_stations_enabled
|
|
my_stations_enabled = my_stations.set_config(config)
|
|
|
|
|
|
def get_directories_page(subdir, directories, request):
|
|
page = vtuner.Page()
|
|
if len(directories) == 0:
|
|
page.add(vtuner.Display("No entries found"))
|
|
page.set_count(1)
|
|
return page
|
|
for directory in get_paged_elements(directories, request.args):
|
|
vtuner_directory = vtuner.Directory(directory.name, url_for(subdir, _external=True, directory=directory.name),
|
|
directory.item_count)
|
|
page.add(vtuner_directory)
|
|
page.set_count(len(directories))
|
|
return page
|
|
|
|
|
|
def get_stations_page(stations, request, tracked=True):
|
|
page = vtuner.Page()
|
|
if len(stations) == 0:
|
|
page.add(vtuner.Display("No stations found"))
|
|
page.set_count(1)
|
|
return page
|
|
for station in get_paged_elements(stations, request.args):
|
|
vtuner_station = station.to_vtuner()
|
|
if tracked:
|
|
vtuner_station.set_trackurl(request.host_url + PATH_ROOT + '/' + PATH_PLAY + '?id=' + vtuner_station.uid)
|
|
vtuner_station.icon = request.host_url + PATH_ROOT + '/' + PATH_ICON + '?id=' + vtuner_station.uid
|
|
page.add(vtuner_station)
|
|
page.set_count(len(stations))
|
|
return page
|
|
|
|
|
|
def get_paged_elements(items, requestargs):
|
|
if requestargs.get('startitems'):
|
|
offset = int(requestargs.get('startitems')) - 1
|
|
elif requestargs.get('start'):
|
|
offset = int(requestargs.get('start')) - 1
|
|
else:
|
|
offset = 0
|
|
if offset > len(items):
|
|
logging.warning("Paging offset larger than item count")
|
|
return []
|
|
if requestargs.get('enditems'):
|
|
limit = int(requestargs.get('enditems'))
|
|
elif requestargs.get('start') and requestargs.get('howmany'):
|
|
limit = int(requestargs.get('start')) - 1 + int(requestargs.get('howmany'))
|
|
else:
|
|
limit = len(items)
|
|
if limit < offset:
|
|
logging.warning("Paging limit smaller than offset")
|
|
return []
|
|
if limit > len(items):
|
|
limit = len(items)
|
|
return items[offset:limit]
|
|
|
|
|
|
def get_station_by_id(stationid):
|
|
station_id_prefix = generic.get_stationid_prefix(stationid)
|
|
if station_id_prefix == my_stations.ID_PREFIX:
|
|
return my_stations.get_station_by_id(generic.get_stationid_without_prefix(stationid))
|
|
elif station_id_prefix == radiobrowser.ID_PREFIX:
|
|
station = radiobrowser.get_station_by_id(generic.get_stationid_without_prefix(stationid))
|
|
station.get_playable_url()
|
|
return station
|
|
return None
|
|
|
|
|
|
@app.route('/setupapp/<path:path>')
|
|
def upstream(path):
|
|
if request.args.get('token') == '0':
|
|
return vtuner.get_init_token()
|
|
if request.args.get('search'):
|
|
return station_search()
|
|
if 'statxml.asp' in path and request.args.get('id'):
|
|
return get_station_info()
|
|
if 'loginXML.asp' in path:
|
|
return redirect(url_for('landing', _external=True), code=302)
|
|
logging.error("Unhandled upstream query (/setupapp/%s)", path)
|
|
abort(404)
|
|
|
|
|
|
@app.route('/', defaults={'path': ''})
|
|
@app.route('/' + PATH_ROOT + '/', defaults={'path': ''})
|
|
def landing(path):
|
|
page = vtuner.Page()
|
|
page.add(vtuner.Directory('Radiobrowser', url_for('radiobrowser_landing', _external=True), 4))
|
|
if my_stations_enabled:
|
|
page.add(vtuner.Directory('My Stations', url_for('my_stations_landing', _external=True),
|
|
len(my_stations.get_category_directories())))
|
|
else:
|
|
page.add(vtuner.Display("'My Stations' feature not configured."))
|
|
page.set_count(1)
|
|
return page.to_string()
|
|
|
|
|
|
@app.route('/' + PATH_ROOT + '/' + PATH_MY_STATIONS + '/')
|
|
def my_stations_landing():
|
|
directories = my_stations.get_category_directories()
|
|
return get_directories_page('my_stations_category', directories, request).to_string()
|
|
|
|
|
|
@app.route('/' + PATH_ROOT + '/' + PATH_MY_STATIONS + '/<directory>')
|
|
def my_stations_category(directory):
|
|
stations = my_stations.get_stations_by_category(directory)
|
|
return get_stations_page(stations, request).to_string()
|
|
|
|
|
|
@app.route('/' + PATH_ROOT + '/' + PATH_RADIOBROWSER + '/')
|
|
def radiobrowser_landing():
|
|
page = vtuner.Page()
|
|
page.add(vtuner.Directory('Genres', url_for('radiobrowser_genres', _external=True),
|
|
len(radiobrowser.get_genre_directories())))
|
|
page.add(vtuner.Directory('Countries', url_for('radiobrowser_countries', _external=True),
|
|
len(radiobrowser.get_country_directories())))
|
|
page.add(vtuner.Directory('Languages', url_for('radiobrowser_languages', _external=True),
|
|
len(radiobrowser.get_language_directories())))
|
|
page.add(vtuner.Directory('Most Popular', url_for('radiobrowser_popular', _external=True),
|
|
len(radiobrowser.get_stations_by_votes())))
|
|
page.set_count(4)
|
|
return page.to_string()
|
|
|
|
|
|
@app.route('/' + PATH_ROOT + '/' + PATH_RADIOBROWSER + '/' + PATH_RADIOBROWSER_COUNTRY + '/')
|
|
def radiobrowser_countries():
|
|
directories = radiobrowser.get_country_directories()
|
|
return get_directories_page('radiobrowser_country_stations', directories, request).to_string()
|
|
|
|
|
|
@app.route('/' + PATH_ROOT + '/' + PATH_RADIOBROWSER + '/' + PATH_RADIOBROWSER_COUNTRY + '/<directory>')
|
|
def radiobrowser_country_stations(directory):
|
|
stations = radiobrowser.get_stations_by_country(directory)
|
|
return get_stations_page(stations, request).to_string()
|
|
|
|
|
|
@app.route('/' + PATH_ROOT + '/' + PATH_RADIOBROWSER + '/' + PATH_RADIOBROWSER_LANGUAGE + '/')
|
|
def radiobrowser_languages():
|
|
directories = radiobrowser.get_language_directories()
|
|
return get_directories_page('radiobrowser_language_stations', directories, request).to_string()
|
|
|
|
|
|
@app.route('/' + PATH_ROOT + '/' + PATH_RADIOBROWSER + '/' + PATH_RADIOBROWSER_LANGUAGE + '/<directory>')
|
|
def radiobrowser_language_stations(directory):
|
|
stations = radiobrowser.get_stations_by_language(directory)
|
|
return get_stations_page(stations, request).to_string()
|
|
|
|
|
|
@app.route('/' + PATH_ROOT + '/' + PATH_RADIOBROWSER + '/' + PATH_RADIOBROWSER_GENRE + '/')
|
|
def radiobrowser_genres():
|
|
directories = radiobrowser.get_genre_directories()
|
|
return get_directories_page('radiobrowser_genre_stations', directories, request).to_string()
|
|
|
|
|
|
@app.route('/' + PATH_ROOT + '/' + PATH_RADIOBROWSER + '/' + PATH_RADIOBROWSER_GENRE + '/<directory>')
|
|
def radiobrowser_genre_stations(directory):
|
|
stations = radiobrowser.get_stations_by_genre(directory)
|
|
return get_stations_page(stations, request).to_string()
|
|
|
|
|
|
@app.route('/' + PATH_ROOT + '/' + PATH_RADIOBROWSER + '/' + PATH_RADIOBROWSER_POPULAR + '/')
|
|
def radiobrowser_popular():
|
|
stations = radiobrowser.get_stations_by_votes()
|
|
return get_stations_page(stations, request).to_string()
|
|
|
|
|
|
@app.route('/' + PATH_ROOT + '/' + PATH_SEARCH + '/')
|
|
def station_search():
|
|
query = request.args.get('search')
|
|
if not query or len(query) < 3:
|
|
page = vtuner.Page()
|
|
page.add(vtuner.Display("Search query too short"))
|
|
page.set_count(1)
|
|
return page.to_string()
|
|
else:
|
|
# TODO: we also need to include 'my station' elements
|
|
stations = radiobrowser.search(query)
|
|
return get_stations_page(stations, request).to_string()
|
|
|
|
|
|
@app.route('/' + PATH_ROOT + '/' + PATH_PLAY)
|
|
def get_stream_url():
|
|
stationid = request.args.get('id')
|
|
if not stationid:
|
|
logging.error("Stream URL without station ID requested")
|
|
abort(400)
|
|
station = get_station_by_id(stationid)
|
|
if not station:
|
|
logging.error("Could not get station with id '%s'", stationid)
|
|
abort(404)
|
|
logging.debug("Station with ID '%s' requested", station.id)
|
|
return redirect(station.url, code=302)
|
|
|
|
|
|
@app.route('/' + PATH_ROOT + '/' + PATH_STATION)
|
|
def get_station_info(tracked=True):
|
|
stationid = request.args.get('id')
|
|
if not stationid:
|
|
logging.error("Station info without station ID requested")
|
|
abort(400)
|
|
station = get_station_by_id(stationid)
|
|
if not station:
|
|
logging.error("Could not get station with id '%s'", stationid)
|
|
page = vtuner.Page()
|
|
page.add(vtuner.Display("Station not found"))
|
|
page.set_count(1)
|
|
return page.to_string()
|
|
vtuner_station = station.to_vtuner()
|
|
if tracked:
|
|
vtuner_station.set_trackurl(request.host_url + PATH_ROOT + '/' + PATH_PLAY + '?id=' + vtuner_station.uid)
|
|
vtuner_station.icon = request.host_url + PATH_ROOT + '/' + PATH_ICON + '?id=' + vtuner_station.uid
|
|
page = vtuner.Page()
|
|
page.add(vtuner_station)
|
|
page.set_count(1)
|
|
return page.to_string()
|
|
|
|
|
|
@app.route('/' + PATH_ROOT + '/' + PATH_ICON)
|
|
def get_station_icon():
|
|
stationid = request.args.get('id')
|
|
if not stationid:
|
|
logging.error("Station icon without station ID requested")
|
|
abort(400)
|
|
station = get_station_by_id(stationid)
|
|
if not station:
|
|
logging.error("Could not get station with id '%s'", stationid)
|
|
abort(404)
|
|
if not hasattr(station, 'icon') or not station.icon:
|
|
logging.warning("No icon information found for station with id '%s'", stationid)
|
|
abort(404)
|
|
station_icon = station_icons.get_icon(station)
|
|
if not station_icon:
|
|
logging.error("Could not get station icon for station with id '%s'", stationid)
|
|
abort(404)
|
|
response = make_response(station_icon)
|
|
response.headers.set('Content-Type', 'image/jpeg')
|
|
return response
|