From 63da37c012e2964bc17044e2da2a8f5c0b3308e8 Mon Sep 17 00:00:00 2001 From: Eric van Blokland Date: Wed, 8 Sep 2021 00:48:20 +0200 Subject: [PATCH] Refactored library and radionet class - Changes in radio.net API (added support for configurable apikey) - Added better caching - Added support for browsing additional categories --- README.rst | 5 +- mopidy_radionet/__init__.py | 19 +- mopidy_radionet/backend.py | 38 ++-- mopidy_radionet/ext.conf | 1 + mopidy_radionet/library.py | 253 +++++++++++++++++++------ mopidy_radionet/radionet.py | 361 +++++++++++++++++++++++------------- setup.cfg | 49 +++++ tests/conftest.py | 24 +++ tests/test_library.py | 142 ++++++++++++++ tests/test_radionet.py | 58 +++--- 10 files changed, 708 insertions(+), 242 deletions(-) create mode 100644 tests/conftest.py create mode 100644 tests/test_library.py diff --git a/README.rst b/README.rst index 41525e2..ff823b8 100644 --- a/README.rst +++ b/README.rst @@ -37,6 +37,7 @@ Mopidy-RadioNet to your Mopidy configuration file:: enabled = true language = pl # or net, de, at, fr, pt, es, dk, se, it min_bitrate = 96 + api_key = valid_api_key favorite_stations = 'bbcradio1' 'bbcradio2' @@ -77,8 +78,8 @@ Mopidy-RadioNet to your Mopidy configuration file:: Project resources ================= -- `Source code `_ -- `Issue tracker `_ +- `Source code `_ +- `Issue tracker `_ Changelog diff --git a/mopidy_radionet/__init__.py b/mopidy_radionet/__init__.py index 7778a36..581dd09 100644 --- a/mopidy_radionet/__init__.py +++ b/mopidy_radionet/__init__.py @@ -5,8 +5,7 @@ import os from mopidy import config, ext - -__version__ = '0.2.2' +__version__ = "0.2.2" logger = logging.getLogger(__name__) @@ -14,21 +13,23 @@ logger = logging.getLogger(__name__) class Extension(ext.Extension): - dist_name = 'Mopidy-RadioNet' - ext_name = 'radionet' + dist_name = "Mopidy-RadioNet" + ext_name = "radionet" version = __version__ def get_default_config(self): - conf_file = os.path.join(os.path.dirname(__file__), 'ext.conf') + conf_file = os.path.join(os.path.dirname(__file__), "ext.conf") return config.read(conf_file) def get_config_schema(self): schema = super(Extension, self).get_config_schema() - schema['language'] = config.String() - schema['min_bitrate'] = config.String() - schema['favorite_stations'] = config.List() + schema["language"] = config.String() + schema["min_bitrate"] = config.String() + schema["api_key"] = config.String() + schema["favorite_stations"] = config.List() return schema def setup(self, registry): from .backend import RadioNetBackend - registry.add('backend', RadioNetBackend) + + registry.add("backend", RadioNetBackend) diff --git a/mopidy_radionet/backend.py b/mopidy_radionet/backend.py index 395fccd..95dcb85 100644 --- a/mopidy_radionet/backend.py +++ b/mopidy_radionet/backend.py @@ -1,11 +1,10 @@ from __future__ import unicode_literals +import re import time -from mopidy import backend - import pykka -import re +from mopidy import backend import mopidy_radionet @@ -19,21 +18,24 @@ class RadioNetBackend(pykka.ThreadingActor, backend.Backend): def __init__(self, config, audio): super(RadioNetBackend, self).__init__() self.radionet = RadioNetClient( - config['proxy'], - "%s/%s" % ( - mopidy_radionet.Extension.dist_name, - mopidy_radionet.__version__)) - - self.library = RadioNetLibraryProvider(backend=self) - self.playback = RadioNetPlaybackProvider( - audio=audio, backend=self + config["proxy"], + "%s/%s" + % (mopidy_radionet.Extension.dist_name, mopidy_radionet.__version__), ) - self.uri_schemes = ['radionet'] + self.library = RadioNetLibraryProvider(backend=self) + self.playback = RadioNetPlaybackProvider(audio=audio, backend=self) - self.radionet.min_bitrate = int(config['radionet']['min_bitrate']) - self.radionet.set_lang(str(config['radionet']['language'])) - self.radionet.set_favorites(tuple(file_ext.lower() for file_ext in config["radionet"]["favorite_stations"])) + self.uri_schemes = ["radionet"] + + self.radionet.min_bitrate = int(config["radionet"]["min_bitrate"]) + self.radionet.set_lang(str(config["radionet"]["language"])) + self.radionet.set_apikey(str(config["radionet"]["api_key"])) + self.radionet.set_favorites( + tuple( + file_ext.lower() for file_ext in config["radionet"]["favorite_stations"] + ) + ) def set_update_timeout(self, minutes=2): self.update_timeout = time.time() + 60 * minutes @@ -51,16 +53,16 @@ class RadioNetBackend(pykka.ThreadingActor, backend.Backend): self.radionet.get_favorites() self.set_update_timeout() -class RadioNetPlaybackProvider(backend.PlaybackProvider): +class RadioNetPlaybackProvider(backend.PlaybackProvider): def is_live(self, uri): return True def translate_uri(self, uri): - identifier = re.findall(r'^radionet:track:?([a-z0-9]+|\d+)?$', uri) + identifier = re.findall(r"^radionet:track:?([a-z0-9]+|\d+)?$", uri) if identifier: radio_data = self.backend.radionet.get_station_by_id(identifier) if radio_data: return radio_data.stream_url - return None \ No newline at end of file + return None diff --git a/mopidy_radionet/ext.conf b/mopidy_radionet/ext.conf index b7fbfb3..3b276cd 100644 --- a/mopidy_radionet/ext.conf +++ b/mopidy_radionet/ext.conf @@ -2,4 +2,5 @@ enabled = true language = pl min_bitrate = 96 +api_key = something favorite_stations = \ No newline at end of file diff --git a/mopidy_radionet/library.py b/mopidy_radionet/library.py index a2ed9e8..d62788a 100644 --- a/mopidy_radionet/library.py +++ b/mopidy_radionet/library.py @@ -10,19 +10,19 @@ logger = logging.getLogger(__name__) class RadioNetLibraryProvider(backend.LibraryProvider): - root_directory = Ref.directory(uri='radionet:root', name='Radio.net') + root_directory = Ref.directory(uri="radionet:root", name="Radio.net") def __init__(self, backend): super().__init__(backend) def lookup(self, uri): - if not uri.startswith('radionet:'): + if not uri.startswith("radionet:"): return None - variant, identifier = self.parse_uri(uri) + variant, identifier, sorting, page = self.parse_uri(uri) - if variant == 'station' or variant == 'track': + if variant == "station" or variant == "track": identifier = int(identifier) radio_data = self.backend.radionet.get_station_by_id(identifier) @@ -30,9 +30,15 @@ class RadioNetLibraryProvider(backend.LibraryProvider): album = Album( artists=[artist], - name=radio_data.description + ' / ' + radio_data.continent + - ' / ' + radio_data.country + ' - ' + radio_data.city, - uri='radionet:station:%s' % (identifier)) + name=radio_data.description + + " / " + + radio_data.continent + + " / " + + radio_data.country + + " - " + + radio_data.city, + uri="radionet:station:%s" % (identifier), + ) track = Track( artists=[artist], @@ -40,81 +46,222 @@ class RadioNetLibraryProvider(backend.LibraryProvider): name=radio_data.name, genre=radio_data.genres, comment=radio_data.description, - uri='radionet:track:%s' % (identifier)) + uri="radionet:track:%s" % (identifier), + ) return [track] return [] def browse(self, uri): - self.backend.refresh() - directories = [] - tracks = [] - variant, identifier = self.parse_uri(uri) - if variant == 'root': - if self.backend.radionet.local_stations: - directories.append( - self.ref_directory( - "radionet:category:localstations", "Local stations") - ) - if self.backend.radionet.top_stations: - directories.append( - self.ref_directory( - "radionet:category:top100", "Top 100") - ) - if self.backend.radionet.favorite_stations: - directories.append( - self.ref_directory( - "radionet:category:favorites", "Favorites") - ) - return directories - elif variant == 'category' and identifier: - if identifier == "localstations": - for station in self.backend.radionet.local_stations: - tracks.append(self.station_to_ref(station)) - if identifier == "top100": - for station in self.backend.radionet.top_stations: - tracks.append(self.station_to_ref(station)) - if identifier == "favorites": - for station in self.backend.radionet.favorite_stations: - tracks.append(self.station_to_ref(station)) - tracks.sort(key=lambda ref: ref.name) - return tracks + category, page, value, sorting = self.parse_uri(uri) + + logger.error("Uri %s", uri) + + if category == "root": + return self._browse_root() + elif category in ["favorites", "topstations", "localstations"]: + return self._browse_category(category, page) + elif category in ["genres", "topics", "languages", "cities", "countries"]: + return self._browse_sorted_category(category, value, sorting, page) else: - logger.debug('Unknown URI: %s', uri) + logger.debug("Unknown URI: %s", uri) return [] + def _browse_root(self): + directories = [ + self.ref_directory("radionet:favorites", "Favorites"), + self.ref_directory("radionet:topstations", "Top stations"), + self.ref_directory("radionet:localstations", "Local stations"), + self.ref_directory("radionet:genres", "Genres"), + self.ref_directory("radionet:topics", "Topics"), + self.ref_directory("radionet:languages", "Languages"), + self.ref_directory("radionet:cities", "Cities"), + self.ref_directory("radionet:countries", "Countries"), + ] + return directories + + def _browse_category(self, category, page): + result = [] + if category == "favorites": + items = self._get_favorites() + if items: + for item in items: + result.append( + self.ref_track( + "radionet:station:{0}".format(str(item.id)), item.name + ) + ) + elif category == "topstations": + items = self._get_topstations() + if items: + for item in items: + result.append( + self.ref_track( + "radionet:station:{0}".format(item["id"]), + item["name"]["value"], + ) + ) + elif not page: + pages = self._get_category_pages(category) + for index in range(pages): + result.append( + self.ref_directory( + "radionet:{0}:{1}".format(category, str(index + 1)), + str(index + 1), + ) + ) + else: + items = self._get_category(category, page) + if items: + for item in items: + result.append( + self.ref_track( + "radionet:station:{0}".format(item["id"]), + item["name"]["value"], + ) + ) + return result + + def _browse_sorted_category(self, category, value, sorting, page): + result = [] + + if not value: + items = self.__getattribute__("_get_{0}".format(category))() + if items: + for item in items: + result.append( + self.ref_directory( + "radionet:{0}:{1}".format(category, item["systemEnglish"]), + item["localized"], + ) + ) + elif not sorting or sorting not in ["rank", "az"]: + result.append( + self.ref_directory( + "radionet:{0}:{1}:rank".format(category, value), "By rank" + ) + ) + result.append( + self.ref_directory( + "radionet:{0}:{1}:az".format(category, value), "Alphabetical" + ) + ) + elif not page: + pages = self._get_sorted_category_pages(category, value) + for index in range(pages): + result.append( + self.ref_directory( + "radionet:{0}:{1}:{2}:{3}".format( + category, value, sorting, str(index + 1) + ), + str(index + 1), + ) + ) + else: + items = self._get_sorted_category(category, value, sorting, page) + if items: + for item in items: + result.append( + self.ref_track( + "radionet:station:{0}".format(item["id"]), + item["name"]["value"], + ) + ) + return result + + def _get_genres(self): + return self.backend.radionet.get_genres() + + def _get_topics(self): + return self.backend.radionet.get_topics() + + def _get_languages(self): + return self.backend.radionet.get_languages() + + def _get_cities(self): + return self.backend.radionet.get_cities() + + def _get_countries(self): + return self.backend.radionet.get_countries() + + def _get_topstations(self): + return self.backend.radionet.get_category("topstations", 1) + + def _get_sorted_category(self, category, name, sorting, page): + return self.backend.radionet.get_sorted_category(category, name, sorting, page) + + def _get_sorted_category_pages(self, category, name): + return self.backend.radionet.get_sorted_category_pages(category, name) + + def _get_category(self, category, page): + return self.backend.radionet.get_category(category, page) + + def _get_category_pages(self, category): + return self.backend.radionet.get_category_pages(category) + + def _get_favorites(self): + return self.backend.radionet.get_favorites() + def search(self, query=None, uris=None, exact=False): - if 'any' not in query: + if "any" not in query: return None result = [] - self.backend.radionet.do_search(' '.join(query['any'])) + self.backend.radionet.do_search(" ".join(query["any"])) for station in self.backend.radionet.search_results: result.append(self.station_to_track(station)) - return SearchResult( - tracks=result - ) + return SearchResult(tracks=result) def station_to_ref(self, station): return Ref.track( - uri='radionet:station:%s' % (station.id), + uri="radionet:station:%s" % (station.id), name=station.name, ) def station_to_track(self, station): ref = self.station_to_ref(station) - return Track(uri=ref.uri, name=ref.name, album=Album(uri=ref.uri, name=ref.name), - artists=[Artist(uri=ref.uri, name=ref.name)]) + return Track( + uri=ref.uri, + name=ref.name, + album=Album(uri=ref.uri, name=ref.name), + artists=[Artist(uri=ref.uri, name=ref.name)], + ) def ref_directory(self, uri, name): return Ref.directory(uri=uri, name=name) + def ref_track(self, uri, name): + return Ref.track(uri=uri, name=name) + def parse_uri(self, uri): - result = re.findall(r'^radionet:([a-z]+):?([a-z0-9]+|\d+)?$', uri) + category = None + value = None + page = None + sorting = None + + result = re.findall( + r"^radionet:(genres|topics|languages|cities|countries)(:([^:]+)(:(rank|az)(:([0-9]+))?)?)?$", + uri, + ) + if result: - return result[0] - return None, None + category = result[0][0] + value = result[0][2] + sorting = result[0][4] + page = result[0][6] + + else: + result = re.findall( + r"^radionet:(root|favorites|topstations|localstations|station|track)(:([0-9]+))?$", + uri, + ) + + if result: + category = result[0][0] + page = result[0][2] + + return category, page, value, sorting diff --git a/mopidy_radionet/radionet.py b/mopidy_radionet/radionet.py index d292abe..54d6404 100644 --- a/mopidy_radionet/radionet.py +++ b/mopidy_radionet/radionet.py @@ -6,9 +6,8 @@ import logging import re import time -from mopidy import httpclient - import requests +from mopidy import httpclient logger = logging.getLogger(__name__) @@ -27,214 +26,324 @@ class Station(object): class RadioNetClient(object): - base_url = 'https://radio.net/' + base_url = "https://radio.net/" session = requests.Session() api_prefix = None min_bitrate = 96 max_top_stations = 100 station_bookmarks = None + api_key = None stations_images = [] - top_stations = [] - local_stations = [] search_results = [] + favorites = [] + cache = {} + + category_param_map = { + "genres": "genre", + "topics": "topic", + "languages": "language", + "cities": "city", + "countries": "country", + } def __init__(self, proxy_config=None, user_agent=None): super(RadioNetClient, self).__init__() self.session = requests.Session() + if proxy_config is not None: proxy = httpclient.format_proxy(proxy_config) - self.session.proxies.update({'http': proxy, 'https': proxy}) + self.session.proxies.update({"http": proxy, "https": proxy}) - if user_agent is None: - user_agent = "Mopidy Radio Client" full_user_agent = httpclient.format_user_agent(user_agent) - self.session.headers.update({'user-agent': full_user_agent}) - self.session.headers.update({'cache-control': 'no-cache'}) + self.session.headers.update({"user-agent": full_user_agent}) + self.session.headers.update({"cache-control": "no-cache"}) + + self.update_prefix() + + def __del__(self): + self.session.close() def set_lang(self, lang): - langs = ['net', 'de', 'at', 'fr', 'pt', 'es', 'dk', 'se', 'it', 'pl'] + langs = ["net", "de", "at", "fr", "pt", "es", "dk", "se", "it", "pl"] if lang in langs: self.base_url = self.base_url.replace(".net", "." + lang) else: logging.error("Radio.net not supported language: %s", str(lang)) + self.update_prefix() - def flush(self): - self.top_stations = [] - self.local_stations = [] - self.search_results = [] - - def set_api_prefix(self): - if self.api_prefix is not None: - return - + def update_prefix(self): tmp_str = self.session.get(self.base_url) - lang = self.base_url.split('.')[-1].replace('/', '') + lang = self.base_url.split(".")[-1].replace("/", "") self.api_prefix = "https://api.radio." + lang + "/info/v2" - def do_get(self, api_sufix, url_params=None): - self.set_api_prefix() - response = self.session.get(self.api_prefix + api_sufix, - params=url_params) + def set_apikey(self, api_key): + self.api_key = api_key + + def do_get(self, api_suffix, url_params=None): + if self.api_prefix is None: + return None + + if url_params is None: + url_params = {} + url_params["apikey"] = self.api_key + + response = self.session.get(self.api_prefix + api_suffix, params=url_params) return response + def get_cache(self, key): + if self.cache.get(key) is not None and self.cache[key].expired() is False: + return self.cache[key].value() + return None + + def set_cache(self, key, value, expires): + self.cache[key] = CacheItem(value, expires) + return value + def get_station_by_id(self, station_id): + cache_key = "station/" + str(station_id) + cache = self.get_cache(cache_key) + if cache is not None: + return cache - api_suffix = '/search/station' + api_suffix = "/search/station" url_params = { - 'station': station_id, + "station": station_id, } response = self.do_get(api_suffix, url_params) - if response.status_code is not 200: - logger.error('Radio.net: Error on get station by id ' + - str(station_id) + ". Error: " + response.text) + if response.status_code != 200: + logger.error( + "Radio.net: Error on get station by id " + + str(station_id) + + ". Error: " + + response.text + ) return False + + logger.debug("Radio.net: Done get top stations list") + json = response.json() + + station = Station() + station.id = json["id"] + station.continent = json["continent"] + station.country = json["country"] + station.city = json["city"] + station.genres = ", ".join(json["genres"]) + station.name = json["name"] + station.stream_url = self.get_stream_url(json["streamUrls"], self.min_bitrate) + station.image = json["logo100x100"] + station.description = json["shortDescription"] + if json["playable"] == "PLAYABLE": + station.playable = True + + return self.set_cache(cache_key, station, 1440) + + def get_genres(self): + return self._get_items("genres") + + def get_topics(self): + return self._get_items("topics") + + def get_languages(self): + return self._get_items("languages") + + def get_cities(self): + return self._get_items("cities") + + def get_countries(self): + return self._get_items("countries") + + def _get_items(self, key): + cached = self.get_cache(key) + if cached is not None: + return cached + + api_suffix = "/search/get" + key + response = self.do_get(api_suffix) + if response.status_code != 200: + logger.error( + "Radio.net: Error on get item list " + + str(api_suffix) + + ". Error: " + + response.text + ) + return False + return self.set_cache(key, response.json(), 1440) + + def get_sorted_category(self, category, name, sorting, page): + + if sorting == "az": + sorting = "STATION_NAME" else: - logger.debug('Radio.net: Done get top stations list') - json = response.json() + sorting = "RANK" - station = Station() - station.id = json['id'] - station.continent = json['continent'] - station.country = json['country'] - station.city = json['city'] - station.genres = ', '.join(json["genres"]) - station.name = json['name'] - station.stream_url = self.get_stream_url( - json['streamUrls'], self.min_bitrate) - station.image = json['logo100x100'] - station.description = json['shortDescription'] - if json['playable'] == 'PLAYABLE': - station.playable = True - - return station - - def get_local_stations(self): - self.local_stations = [] - - api_suffix = '/search/localstations' + cache_key = category + "/" + name + "/" + sorting + "/" + str(page) + cache = self.get_cache(cache_key) + if cache is not None: + return cache + api_suffix = "/search/stationsby" + self.category_param_map[category] url_params = { - 'pageindex': 1, - 'sizeperpage': 100, + self.category_param_map[category]: name, + "sorttype": sorting, + "sizeperpage": 50, + "pageindex": page, } response = self.do_get(api_suffix, url_params) - if response.status_code is not 200: - logger.error('Radio.net: Get local stations error. ' + - response.text) - else: - logger.debug('Radio.net: Done get local stations list') - json = response.json() - for match in json['categories'][0]['matches']: - station = self.get_station_by_id(match['id']) - if station: - if station.playable: - self.local_stations.append(station) + if response.status_code != 200: + logger.error( + "Radio.net: Error on get station by " + + str(category) + + ". Error: " + + response.text + ) + return False - logger.info('Radio.net: Loaded ' + str(len(self.local_stations)) + - ' local stations.') + json = response.json() + self.set_cache(category + "/" + name, int(json["numberPages"]), 10) + return self.set_cache(cache_key, json["categories"][0]["matches"], 10) + + def get_category(self, category, page): + cache_key = category + "/" + str(page) + cache = self.get_cache(cache_key) + if cache is not None: + return cache + + api_suffix = "/search/" + category + url_params = {"sizeperpage": 50, "pageindex": page} + + response = self.do_get(api_suffix, url_params) + logger.error(response.text) + if response.status_code != 200: + logger.error( + "Radio.net: Error on get station by " + + str(category) + + ". Error: " + + response.text + ) + return False + + json = response.json() + self.set_cache(category, int(json["numberPages"]), 10) + return self.set_cache(cache_key, json["categories"][0]["matches"], 10) + + def get_sorted_category_pages(self, category, name): + cache_key = category + "/" + name + cache = self.get_cache(cache_key) + if cache is not None: + return cache + + self.get_sorted_category(category, name, "rank", 1) + + return self.get_cache(cache_key) + + def get_category_pages(self, category): + cache_key = category + cache = self.get_cache(cache_key) + if cache is not None: + return cache + + self.get_category(category, 1) + + return self.get_cache(cache_key) def set_favorites(self, favorites): - self.favortes = favorites + self.favorites = favorites def get_favorites(self): - self.favorite_stations = [] + logger.error(str(self.favorites)) + cache_key = "favorites" + cache = self.get_cache(cache_key) + if cache is not None: + return cache - for station in self.favortes: - api_suffix = '/search/stationsonly' + favorite_stations = [] + for station in self.favorites: + logger.error(str(station)) + api_suffix = "/search/stationsonly" url_params = { - 'query': station, - 'pageindex': 1, + "query": station, + "pageindex": 1, } - response = self.do_get(api_suffix, url_params) - if response.status_code is not 200: - logger.error('Radio.net: Search error ' + response.text) + if response.status_code != 200: + logger.error("Radio.net: Search error " + response.text) else: - logger.debug('Radio.net: Done search') + logger.debug("Radio.net: Done search") json = response.json() # take only the first match! - station = self.get_station_by_id(json['categories'][0]['matches'][0]['id']) + station = self.get_station_by_id( + json["categories"][0]["matches"][0]["id"] + ) if station and station.playable: - self.favorite_stations.append(station) + favorite_stations.append(station) - logger.info('Radio.net: Loaded ' + str(len(self.favorite_stations)) + - ' favorite stations.') + logger.info( + "Radio.net: Loaded " + str(len(favorite_stations)) + " favorite stations." + ) + return self.set_cache(cache_key, favorite_stations, 1440) - def get_top_stations(self): - self.top_stations = [] - api_suffix = '/search/topstations' + def do_search(self, query_string, page_index=1, search_results=[]): + api_suffix = "/search/stationsonly" url_params = { - 'pageindex': 1, - 'sizeperpage': 100, + "query": query_string, + "sizeperpage": 50, + "pageindex": page_index, } response = self.do_get(api_suffix, url_params) - if response.status_code is not 200: - logger.error('Radio.net: Get top stations error. ' + response.text) + if response.status_code != 200: + logger.error("Radio.net: Search error " + response.text) else: - logger.debug('Radio.net: Done get top stations list') + logger.debug("Radio.net: Done search") json = response.json() - for match in json['categories'][0]['matches']: - station = self.get_station_by_id(match['id']) - if station: - if station.playable: - self.top_stations.append(station) - - logger.info('Radio.net: Loaded ' + str(len(self.top_stations)) + - ' top stations.') - - def do_search(self, query_string, page_index=1): - if page_index == 1: - self.search_results = [] - - api_suffix = '/search/stationsonly' - url_params = { - 'query': query_string, - 'pageindex': page_index, - } - - response = self.do_get(api_suffix, url_params) - - if response.status_code is not 200: - logger.error('Radio.net: Search error ' + response.text) - else: - logger.debug('Radio.net: Done search') - json = response.json() - for match in json['categories'][0]['matches']: - station = self.get_station_by_id(match['id']) + for match in json["categories"][0]["matches"]: + station = self.get_station_by_id(match["id"]) if station and station.playable: - self.search_results.append(station) + search_results.append(station) - number_pages = int(json['numberPages']) + number_pages = int(json["numberPages"]) if number_pages >= page_index: - self.do_search(query_string, page_index + 1) + self.do_search(query_string, page_index + 1, search_results) else: - logger.info('Radio.net: Found ' + - str(len(self.search_results)) + ' stations.') + logger.info( + "Radio.net: Found " + str(len(search_results)) + " stations." + ) + return search_results def get_stream_url(self, stream_json, bit_rate): stream_url = None for stream in stream_json: - if int(stream['bitRate']) >= bit_rate and \ - stream['streamStatus'] == 'VALID': - stream_url = stream['streamUrl'] + if int(stream["bitRate"]) >= bit_rate and stream["streamStatus"] == "VALID": + stream_url = stream["streamUrl"] break if stream_url is None and len(stream_json) > 0: - stream_url = stream_json[0]['streamUrl'] + stream_url = stream_json[0]["streamUrl"] return stream_url + + +class CacheItem(object): + def __init__(self, value, expires=10): + self._value = value + self._expires = expires = time.time() + expires * 60 + + def expired(self): + return self._expires < time.time() + + def value(self): + return self._value diff --git a/setup.cfg b/setup.cfg index c4829e0..d1587db 100644 --- a/setup.cfg +++ b/setup.cfg @@ -1,3 +1,52 @@ +[metadata] +name = Mopidy-RadioNet +version = 0.2.2 +url = https://github.com/plintx/mopidy-radionet +license = Apache License, Version 2.0 +license_file = LICENSE +description = Mopidy extension for playing music from Radio.net +long_description = file: README.rst +classifiers = + Environment :: No Input/Output (Daemon) + Intended Audience :: End Users/Desktop + License :: OSI Approved :: Apache Software License + Operating System :: OS Independent + Programming Language :: Python :: 3 + Programming Language :: Python :: 3.7 + Programming Language :: Python :: 3.8 + Programming Language :: Python :: 3.9 + Topic :: Multimedia :: Sound/Audio :: Players + +[options] +zip_safe = False +include_package_data = True +packages = find: +python_requires = >= 3.7 +install_requires = + Mopidy >= 3.0.0 + Pykka >= 2.0.1 + pyspotify >= 2.0.5 + requests >= 2.20.0 + setuptools + + +[options.extras_require] +lint = + black + check-manifest + flake8 + flake8-black + flake8-bugbear + flake8-import-order + isort +test = + pytest + pytest-cov + responses +dev = + %(lint)s + %(test)s + [flake8] application-import-names = mopidy_radionet,tests exclude = .git,.tox diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..8580305 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,24 @@ +from unittest import mock + +import pytest + +from mopidy_radionet import backend +from mopidy_radionet.radionet import RadioNetClient +from mopidy_radionet.library import RadioNetLibraryProvider + +@pytest.fixture +def backend_mock(): + backend_mock = mock.Mock(spec=backend.RadioNetBackend) + backend_mock.radionet = RadioNetClient(proxy_config=None) + backend_mock.library = RadioNetLibraryProvider(backend=backend_mock) + backend_mock.radionet.set_apikey('test') + backend_mock.radionet.set_favorites({'lush'}) + return backend_mock + +@pytest.fixture +def library(backend_mock): + return backend_mock.library + +@pytest.fixture +def radionet(backend_mock): + return backend_mock.radionet diff --git a/tests/test_library.py b/tests/test_library.py new file mode 100644 index 0000000..38455e4 --- /dev/null +++ b/tests/test_library.py @@ -0,0 +1,142 @@ +from unittest import mock + + +def test_browse_root(library): + results = library.browse('radionet:root'); + assert 8 == len(results) + + +def test_browse_localstations(library): + results = library.browse('radionet:localstations'); + assert len(results) > 0 + + page_uri = results[0].uri if results is not None else None + assert page_uri is not None + + results = library.browse(page_uri) + assert len(results) > 0 + + +def test_browse_topstations(library): + results = library.browse('radionet:topstations'); + assert len(results) > 0 + + +def test_browse_genres(library): + results = library.browse('radionet:genres'); + assert len(results) > 0 + + cat_uri = results[0].uri if results is not None else None + assert cat_uri is not None + + results = library.browse(cat_uri) + assert len(results) == 2 + + sort_uri = results[0].uri if results is not None else None + assert sort_uri is not None + + results = library.browse(sort_uri) + assert len(results) > 0 + + page_uri = results[0].uri if results is not None else None + assert page_uri is not None + + results = library.browse(page_uri) + assert len(results) > 0 + + +def test_browse_topics(library): + results = library.browse('radionet:topics'); + assert len(results) > 0 + + cat_uri = results[0].uri if results is not None else None + assert cat_uri is not None + + results = library.browse(cat_uri) + assert len(results) == 2 + + sort_uri = results[0].uri if results is not None else None + assert sort_uri is not None + + results = library.browse(sort_uri) + assert len(results) > 0 + + page_uri = results[0].uri if results is not None else None + assert page_uri is not None + + results = library.browse(page_uri) + assert len(results) > 0 + + +def test_browse_languages(library): + results = library.browse('radionet:languages'); + assert len(results) > 0 + + cat_uri = results[0].uri if results is not None else None + assert cat_uri is not None + + results = library.browse(cat_uri) + assert len(results) == 2 + + sort_uri = results[0].uri if results is not None else None + assert sort_uri is not None + + results = library.browse(sort_uri) + assert len(results) > 0 + + page_uri = results[0].uri if results is not None else None + assert page_uri is not None + + results = library.browse(page_uri) + assert len(results) > 0 + + +def test_browse_cities(library): + results = library.browse('radionet:cities'); + assert len(results) > 0 + + cat_uri = results[0].uri if results is not None else None + assert cat_uri is not None + + results = library.browse(cat_uri) + assert len(results) == 2 + + sort_uri = results[0].uri if results is not None else None + assert sort_uri is not None + + results = library.browse(sort_uri) + assert len(results) > 0 + + page_uri = results[0].uri if results is not None else None + assert page_uri is not None + + results = library.browse(page_uri) + assert len(results) > 0 + + +def test_browse_countries(library): + results = library.browse('radionet:countries'); + assert len(results) > 0 + + cat_uri = results[0].uri if results is not None else None + assert cat_uri is not None + + results = library.browse(cat_uri) + assert len(results) == 2 + + sort_uri = results[0].uri if results is not None else None + assert sort_uri is not None + + results = library.browse(sort_uri) + assert len(results) > 0 + + page_uri = results[0].uri if results is not None else None + assert page_uri is not None + + results = library.browse(page_uri) + assert len(results) > 0 + + +def test_browse_favorites(library): + results = library.browse('radionet:favorites'); + assert 1 == len(results) diff --git a/tests/test_radionet.py b/tests/test_radionet.py index 36971af..48eebe1 100644 --- a/tests/test_radionet.py +++ b/tests/test_radionet.py @@ -1,38 +1,28 @@ -import unittest - -from mopidy_radionet.radionet import RadioNetClient +from unittest import mock -class RadioNetClientTest(unittest.TestCase): - - def test_get_api_key(self): - radionet = RadioNetClient() - radionet.get_api_key() - - self.assertIsNotNone(radionet.api_key) - - def test_get_top_stations(self): - radionet = RadioNetClient() - radionet.get_top_stations() - self.assertGreater(len(radionet.top_stations), 0) - - def test_get_local_stations(self): - radionet = RadioNetClient() - radionet.get_local_stations() - self.assertGreater(len(radionet.local_stations), 0) - - def test_do_search(self): - radionet = RadioNetClient() - radionet.do_search("radio ram") - self.assertGreater(len(radionet.search_results), 0) - - def test_get_favorites(self): - test_favorites = ("Rock Antenne", "radio ram") - radionet = RadioNetClient() - radionet.set_favorites(test_favorites) - radionet.get_favorites() - self.assertEqual(len(radionet.favorite_stations), len(test_favorites)) +def test_get_genres(radionet): + genres = radionet.get_genres(); + assert len(genres) > 0 -if __name__ == "__main__": - unittest.main() +def test_get_top_stations(radionet): + result = radionet.get_category('topstations', 1) + assert len(result) > 0 + + +def test_get_local_stations(radionet): + result = radionet.get_category('localstations', 1) + assert len(result) > 0 + + +def test_do_search(radionet): + result = radionet.do_search("radio ram") + assert len(result) > 0 + + +def test_get_favorites(radionet): + test_favorites = ("Rock Antenne", "radio ram") + radionet.set_favorites(test_favorites) + result = radionet.get_favorites() + assert len(result) == len(test_favorites)