Decouple 'My Stations' feature from server module

Also add in check whether to show the respective link or just
a disclaimer on the landing page.
This commit is contained in:
milaq 2019-08-14 16:01:30 +02:00
parent 92756a7144
commit 1316bb65d0
2 changed files with 80 additions and 40 deletions

61
ycast/my_stations.py Normal file
View file

@ -0,0 +1,61 @@
import logging
import yaml
import ycast.vtuner as vtuner
ID_PREFIX = "MY_"
config_file = 'my_stations.yml'
class Station:
def __init__(self, name, url, category):
self.id = ID_PREFIX + '000000' # TODO: generate meaningful ID
self.name = name
self.url = url
self.tag = category
def to_vtuner(self):
return vtuner.Station(self.id, self.name, self.tag, self.url, None, self.tag, None, None, None, None)
def set_config(config):
global config_file
if config:
config_file = config
if get_stations():
return True
else:
return False
def get_stations():
try:
with open(config_file, 'r') as f:
my_stations = yaml.safe_load(f)
except FileNotFoundError:
logging.error("Station configuration '%s' not found", config_file)
return None
except yaml.YAMLError as e:
logging.error("Station configuration format error: %s", e)
return None
return my_stations
def get_categories():
my_stations_yaml = get_stations()
categories = []
if my_stations_yaml:
for category in my_stations_yaml:
categories.append(category)
return categories
def get_stations_by_category(category):
my_stations_yaml = get_stations()
stations = []
if my_stations_yaml and category in my_stations_yaml:
for station_name in my_stations_yaml[category]:
stations.append(Station(station_name, my_stations_yaml[category][station_name], category))
return stations

View file

@ -5,42 +5,32 @@ from flask import Flask, request, url_for
import ycast.vtuner as vtuner import ycast.vtuner as vtuner
import ycast.radiobrowser as radiobrowser import ycast.radiobrowser as radiobrowser
import ycast.my_stations as my_stations
PATH_ROOT = 'ycast' PATH_ROOT = 'ycast'
PATH_CUSTOM_STATIONS = 'my_stations' PATH_MY_STATIONS = 'my_stations'
PATH_RADIOBROWSER = 'radiobrowser' PATH_RADIOBROWSER = 'radiobrowser'
PATH_RADIOBROWSER_COUNTRY = 'country' PATH_RADIOBROWSER_COUNTRY = 'country'
PATH_RADIOBROWSER_GENRE = 'genre' PATH_RADIOBROWSER_GENRE = 'genre'
PATH_RADIOBROWSER_POPULAR = 'popular' PATH_RADIOBROWSER_POPULAR = 'popular'
PATH_RADIOBROWSER_SEARCH = 'search' PATH_RADIOBROWSER_SEARCH = 'search'
my_stations = {} my_stations_enabled = False
app = Flask(__name__) app = Flask(__name__)
def run(config, address='0.0.0.0', port=8010): def run(config, address='0.0.0.0', port=8010):
try: try:
get_stations(config) check_my_stations_feature(config)
app.run(host=address, port=port) app.run(host=address, port=port)
except PermissionError: except PermissionError:
logging.error("No permission to create socket. Are you trying to use ports below 1024 without elevated rights?") logging.error("No permission to create socket. Are you trying to use ports below 1024 without elevated rights?")
def get_stations(config): def check_my_stations_feature(config):
global my_stations global my_stations_enabled
if not config: my_stations_enabled = my_stations.set_config(config)
logging.warning("If you want to use the 'My Stations' feature, please supply a valid station configuration")
return
try:
with open(config, 'r') as f:
my_stations = yaml.safe_load(f)
except FileNotFoundError:
logging.error("Station configuration '%s' not found", config)
return
except yaml.YAMLError as e:
logging.error("Config error: %s", e)
return
def get_directories_page(subdir, directories, requestargs): def get_directories_page(subdir, directories, requestargs):
@ -112,36 +102,25 @@ def landing(path):
return vtuner.get_init_token() return vtuner.get_init_token()
page = vtuner.Page() page = vtuner.Page()
page.add(vtuner.Directory('Radiobrowser', url_for('radiobrowser_landing', _external=True))) page.add(vtuner.Directory('Radiobrowser', url_for('radiobrowser_landing', _external=True)))
page.add(vtuner.Directory('My Stations', url_for('custom_stations_landing', _external=True))) if my_stations_enabled:
page.add(vtuner.Directory('My Stations', url_for('my_stations_landing', _external=True)))
else:
page.add(vtuner.Display("'My Stations' feature not configured."))
return page.to_string() return page.to_string()
@app.route('/' + PATH_ROOT + '/' + PATH_CUSTOM_STATIONS + '/') @app.route('/' + PATH_ROOT + '/' + PATH_MY_STATIONS + '/')
def custom_stations_landing(): def my_stations_landing():
page = vtuner.Page() page = vtuner.Page()
page.add(vtuner.Previous(url_for("landing", _external=True))) page.add(vtuner.Previous(url_for("landing", _external=True)))
if not my_stations: directories = my_stations.get_categories()
page.add(vtuner.Display("No stations found")) return get_directories_page('my_stations_category', directories, request.args).to_string()
else:
for category in sorted(my_stations, key=str.lower):
directory = vtuner.Directory(category, url_for('custom_stations_category',
_external=True, category=category))
page.add(directory)
return page.to_string()
@app.route('/' + PATH_ROOT + '/' + PATH_CUSTOM_STATIONS + '/<category>') @app.route('/' + PATH_ROOT + '/' + PATH_MY_STATIONS + '/<directory>')
def custom_stations_category(category): def my_stations_category(directory):
page = vtuner.Page() stations = my_stations.get_stations_by_category(directory)
page.add(vtuner.Previous(url_for('custom_stations_landing', _external=True))) return get_stations_page(stations, request.args).to_string()
if category not in my_stations:
page.add(vtuner.Display("Category '" + category + "' not found"))
else:
for station in sorted(my_stations[category], key=str.lower):
station = vtuner.Station(None, station, None, my_stations[category][station],
None, None, None, None, None, None)
page.add(station)
return page.to_string()
@app.route('/' + PATH_ROOT + '/' + PATH_RADIOBROWSER + '/') @app.route('/' + PATH_ROOT + '/' + PATH_RADIOBROWSER + '/')