sam/mopidy-radionet
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Changes for V5 api compatibility

Browse Source
This commit is contained in:
Eric van Blokland
2024-12-24 15:59:04 +01:00
parent 7b19c20269
commit 62bf24b9b5
6 changed files with 227 additions and 303 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
mopidy_radionet/backend.py
View File

@@ -28,7 +28,6 @@ class RadioNetBackend(pykka.ThreadingActor, backend.Backend):
self.radionet.min_bitrate = int(config["radionet"]["min_bitrate"]) self.radionet.min_bitrate = int(config["radionet"]["min_bitrate"])
self.radionet.set_lang(str(config["radionet"]["language"]).strip()) self.radionet.set_lang(str(config["radionet"]["language"]).strip())
self.radionet.set_apikey(str(config["radionet"]["api_key"]))
self.radionet.set_favorites( self.radionet.set_favorites(
tuple( tuple(
file_ext.strip("'").lower() for file_ext in config["radionet"]["favorite_stations"] file_ext.strip("'").lower() for file_ext in config["radionet"]["favorite_stations"]
@@ -41,8 +40,9 @@ class RadioNetPlaybackProvider(backend.PlaybackProvider):
return True return True
def translate_uri(self, uri): def translate_uri(self, uri):
identifier = re.findall(r"^radionet:track:?([a-z0-9]+|\d+)?$", uri) identifier = re.findall(r"^radionet:track:?(.+)?$", uri)
if identifier: if identifier:
print(identifier)
return self.backend.radionet.get_stream_url(identifier[0]) return self.backend.radionet.get_stream_url(identifier[0])
return None return None

97
mopidy_radionet/library.py
View File

@@ -21,24 +21,18 @@ class RadioNetLibraryProvider(backend.LibraryProvider):
if not uri.startswith("radionet:"): if not uri.startswith("radionet:"):
return None return None
variant, identifier, sorting, page = self.parse_uri(uri) variant, identifier, page = self.parse_uri(uri)
if variant == "station" or variant == "track": if variant == "station" or variant == "track":
try:
identifier = int(identifier)
radio_data = self.backend.radionet.get_station_by_id(identifier) radio_data = self.backend.radionet.get_station_by_id(identifier)
except ValueError:
radio_data = self.backend.radionet.get_station_by_slug(identifier)
artist = Artist(name=radio_data.name) artist = Artist(name=radio_data.name)
name = "" name = ""
if radio_data.description is not None: if radio_data.description is not None:
name = radio_data.description + " / " name = radio_data.description
name = ( name = (
name name
+ radio_data.continent
+ " / "
+ radio_data.country + radio_data.country
+ " - " + " - "
+ radio_data.city + radio_data.city
@@ -64,14 +58,15 @@ class RadioNetLibraryProvider(backend.LibraryProvider):
def browse(self, uri): def browse(self, uri):
category, page, value, sorting = self.parse_uri(uri) category, page, value = self.parse_uri(uri)
if category == "root": if category == "root":
return self._browse_root() return self._browse_root()
elif category in ["favorites", "topstations", "localstations"]: elif category in ["favorites", "local"]:
return self._browse_category(category, page) return self._browse_simple_category(category, page)
elif category in ["genres", "topics", "languages", "cities", "countries"]: elif category in ["genres", "topics", "languages", "cities", "countries"]:
return self._browse_sorted_category(category, value, sorting, page) result = self._browse_category(category, value, page)
return result
else: else:
logger.debug("Unknown URI: %s", uri) logger.debug("Unknown URI: %s", uri)
return [] return []
@@ -79,7 +74,7 @@ class RadioNetLibraryProvider(backend.LibraryProvider):
def get_images(self, uris): def get_images(self, uris):
images = {} images = {}
for uri in uris: for uri in uris:
variant, identifier, sorting, page = self.parse_uri(uri) variant, identifier, page = self.parse_uri(uri)
station = self.backend.radionet.get_station_by_id(identifier) station = self.backend.radionet.get_station_by_id(identifier)
if station: if station:
images[uri] = [] images[uri] = []
@@ -95,8 +90,7 @@ class RadioNetLibraryProvider(backend.LibraryProvider):
def _browse_root(self): def _browse_root(self):
directories = [ directories = [
self.ref_directory("radionet:topstations", "Top stations"), self.ref_directory("radionet:local", "Local stations"),
self.ref_directory("radionet:localstations", "Local stations"),
self.ref_directory("radionet:genres", "Genres"), self.ref_directory("radionet:genres", "Genres"),
self.ref_directory("radionet:topics", "Topics"), self.ref_directory("radionet:topics", "Topics"),
self.ref_directory("radionet:languages", "Languages"), self.ref_directory("radionet:languages", "Languages"),
@@ -107,22 +101,22 @@ class RadioNetLibraryProvider(backend.LibraryProvider):
directories.insert(0, self.ref_directory("radionet:favorites", "Favorites")) directories.insert(0, self.ref_directory("radionet:favorites", "Favorites"))
return directories return directories
def _browse_category(self, category, page): def _browse_simple_category(self, category, page):
result = [] result = []
if category == "favorites": if category == "favorites":
items = self._get_favorites() items = self._get_favorites()
if items: if items:
for item in items: for item in items:
result.append(self.station_to_ref(item)) result.append(self.station_to_ref(item))
elif category == "topstations": # elif category == "topstations":
items = self._get_topstations() # items = self._get_topstations()
if items: # if items:
for item in items: # for item in items:
result.append(self.station_to_ref(item)) # result.append(self.station_to_ref(item))
elif not page: elif not page:
pages = self._get_category_pages(category) pages = self._get_simple_category_pages(category)
if pages == 1: if pages == 1:
items = self._get_category(category, 1) items = self._get_simple_category(category, 1)
if items: if items:
for item in items: for item in items:
result.append(self.station_to_ref(item)) result.append(self.station_to_ref(item))
@@ -141,7 +135,7 @@ class RadioNetLibraryProvider(backend.LibraryProvider):
result.append(self.station_to_ref(item)) result.append(self.station_to_ref(item))
return result return result
def _browse_sorted_category(self, category, value, sorting, page): def _browse_category(self, category, value, page):
result = [] result = []
if not value: if not value:
@@ -150,25 +144,14 @@ class RadioNetLibraryProvider(backend.LibraryProvider):
for item in items: for item in items:
result.append( result.append(
self.ref_directory( self.ref_directory(
"radionet:{0}:{1}".format(category, item["systemEnglish"]), "radionet:{0}:{1}".format(category, item["slug"]),
item["localized"], item["name"],
)
)
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: elif not page:
pages = self._get_sorted_category_pages(category, value) pages = self._get_category_pages(category, value)
if pages == 1: if pages == 1:
items = self._get_sorted_category(category, value, sorting, 1) items = self._get_category(category, value, 1)
if items: if items:
for item in items: for item in items:
result.append(self.station_to_ref(item)) result.append(self.station_to_ref(item))
@@ -176,14 +159,14 @@ class RadioNetLibraryProvider(backend.LibraryProvider):
for index in range(pages): for index in range(pages):
result.append( result.append(
self.ref_directory( self.ref_directory(
"radionet:{0}:{1}:{2}:{3}".format( "radionet:{0}:{1}:{2}".format(
category, value, sorting, str(index + 1) category, value, str(index + 1)
), ),
str(index + 1), str(index + 1),
) )
) )
else: else:
items = self._get_sorted_category(category, value, sorting, page) items = self._get_category(category, value, int(page))
if items: if items:
for item in items: for item in items:
result.append(self.station_to_ref(item)) result.append(self.station_to_ref(item))
@@ -205,19 +188,19 @@ class RadioNetLibraryProvider(backend.LibraryProvider):
return self.backend.radionet.get_countries() return self.backend.radionet.get_countries()
def _get_topstations(self): def _get_topstations(self):
return self.backend.radionet.get_category("topstations", 1) return self.backend.radionet.get_simple_category("local", 1)
def _get_sorted_category(self, category, name, sorting, page): def _get_category(self, category, name, page):
return self.backend.radionet.get_sorted_category(category, name, sorting, page) return self.backend.radionet.get_category(category, name, page)
def _get_sorted_category_pages(self, category, name): def _get_category_pages(self, category, name):
return self.backend.radionet.get_sorted_category_pages(category, name) return self.backend.radionet.get_category_pages(category, name)
def _get_category(self, category, page): def _get_simple_category(self, category, page):
return self.backend.radionet.get_category(category, page) return self.backend.radionet.get_simple_category(category, page)
def _get_category_pages(self, category): def _get_simple_category_pages(self, category):
return self.backend.radionet.get_category_pages(category) return self.backend.radionet.get_simple_category_pages(category)
def _get_favorites(self): def _get_favorites(self):
return self.backend.radionet.get_favorites() return self.backend.radionet.get_favorites()
@@ -258,22 +241,20 @@ class RadioNetLibraryProvider(backend.LibraryProvider):
category = None category = None
value = None value = None
page = None page = None
sorting = None
result = re.findall( result = re.findall(
r"^radionet:(genres|topics|languages|cities|countries)(:([^:]+)(:(rank|az)(:([0-9]+))?)?)?$", r"^radionet:(genres|topics|languages|cities|countries)(:([^:]+)(:([0-9]+))?)?$",
uri, uri,
) )
if result: if result:
category = result[0][0] category = result[0][0]
value = result[0][2] value = result[0][2]
sorting = result[0][4] page = result[0][4]
page = result[0][6]
else: else:
result = re.findall( result = re.findall(
r"^radionet:(root|favorites|topstations|localstations|station|track)(:([0-9]+))?$", r"^radionet:(root|favorites|local|station|track)(:([0-9]+))?$",
uri, uri,
) )
@@ -283,7 +264,7 @@ class RadioNetLibraryProvider(backend.LibraryProvider):
else: else:
result = re.findall( result = re.findall(
r"^radionet:(track):([^:]+)$", r"^radionet:(track|station):([^:]+)$",
uri, uri,
) )
@@ -291,4 +272,4 @@ class RadioNetLibraryProvider(backend.LibraryProvider):
category = result[0][0] category = result[0][0]
page = result[0][1] page = result[0][1]
return category, page, value, sorting return category, page, value

329
mopidy_radionet/radionet.py
View File

@@ -2,14 +2,40 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
from __future__ import unicode_literals from __future__ import unicode_literals
import json
import logging import logging
import time import time
import urllib
import requests from urllib.request import urlopen, Request
from mopidy import httpclient
import math
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
REGIONS = {
'at': 'de-AT',
'au': 'en-AU',
'br': 'pt-BR',
'ca': 'en-CA',
'co': 'es-CO',
'de': 'de-DE',
'dk': 'da-DK',
'es': 'es-ES',
'fr': 'fr-FR',
'ie': 'en-IE',
'it': 'it-IT',
'mx': 'es-MX',
'nl': 'nl-NL',
'nz': 'en-NZ',
'pl': 'pl-PL',
'pt': 'pt-PT',
'se': 'sv-SE',
'uk': 'en-GB',
'us': 'en-US',
'za': 'en-ZA',
}
class Station(object): class Station(object):
id = None id = None
@@ -28,10 +54,10 @@ class Station(object):
class RadioNetClient(object): class RadioNetClient(object):
base_url = "https://radio.net/" base_url = "https://prod.radio-api.net"
language = REGIONS['us']
user_agent = 'Radio.net - Web V5'
session = requests.Session()
api_prefix = None
min_bitrate = 96 min_bitrate = 96
max_top_stations = 100 max_top_stations = 100
station_bookmarks = None station_bookmarks = None
@@ -56,50 +82,26 @@ class RadioNetClient(object):
def __init__(self, proxy_config=None, user_agent=None): def __init__(self, proxy_config=None, user_agent=None):
super(RadioNetClient, self).__init__() 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})
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.update_prefix()
def __del__(self):
self.session.close()
def set_lang(self, lang): def set_lang(self, lang):
if lang == "en": if lang in REGIONS:
lang = "net" self.language = REGIONS[lang]
langs = ["net", "de", "at", "fr", "pt", "es", "dk", "se", "it", "pl"]
self.base_url = "https://radio.net/"
if lang in langs:
self.base_url = self.base_url.replace(".net", "." + lang)
else: else:
logging.warning("Radio.net not supported language: %s, defaulting to English", str(lang)) logging.warning("Radio.net not supported language: %s, defaulting to English", str(lang))
self.update_prefix()
def update_prefix(self):
lang = self.base_url.split(".")[-1].replace("/", "")
self.api_prefix = "https://api.radio." + lang + "/info/v2"
def set_apikey(self, api_key):
self.api_key = api_key
def do_get(self, api_suffix, url_params=None): def do_get(self, api_suffix, url_params=None):
if self.api_prefix is None: try:
return None url = self.base_url + api_suffix
if url_params is not None:
url += "?" + urllib.parse.urlencode(url_params)
req = Request(url)
req.add_header('accept-language', self.language)
req.add_header('user-agent', self.user_agent)
response = urlopen(req).read()
except Exception as err:
logging.error(f'_open_url error: {err}')
response = None
if url_params is None: return json.loads(response)
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): def get_cache(self, key):
if self.cache.get(key) is not None and self.cache[key].expired() is False: if self.cache.get(key) is not None and self.cache[key].expired() is False:
@@ -126,52 +128,25 @@ class RadioNetClient(object):
if cache is not None: if cache is not None:
return cache return cache
api_suffix = "/search/station" api_suffix = "/stations/details"
url_params = { url_params = {
"station": station_id, "stationIds": station_id,
} }
response = self.do_get(api_suffix, url_params) response = self.do_get(api_suffix, url_params)
if response.status_code != 200: if response is None or len(response) == 0:
logger.error( logger.error("Radio.net: Error on get station by id " + str(station_id))
"Radio.net: Error on get station by id "
+ str(station_id)
+ ". Error: "
+ response.text
)
return False return False
logger.debug("Radio.net: Done get top stations list") logger.debug("Radio.net: Done get top stations list")
json = response.json()
if not self.stations_by_id.get(json["id"]): json = response[0]
station = Station()
station.playable = True
else:
station = self.stations_by_id[json["id"]]
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.slug = json["subdomain"]
station.stream_url = self._get_stream_url(json["streamUrls"], self.min_bitrate)
station.image_tiny = json["logo44x44"]
station.image_small = json["logo100x100"]
station.image_medium = json["logo175x175"]
station.image_large = json["logo300x300"]
station.description = json["shortDescription"]
if json["playable"] == "PLAYABLE":
station.playable = True
self.stations_by_id[station.id] = station station = self._get_station_from_search_result(json)
self.stations_by_slug[station.slug] = station
self.set_cache("station/" + str(station.id), station, 1440) self.set_cache(cache_key, station, 1440)
self.set_cache("station/" + station.slug, station, 1440)
return station return station
def _get_station_from_search_result(self, result): def _get_station_from_search_result(self, result):
@@ -182,42 +157,56 @@ class RadioNetClient(object):
station = self.stations_by_id[result["id"]] station = self.stations_by_id[result["id"]]
station.id = result["id"] station.id = result["id"]
if result["continent"] is not None:
station.continent = result["continent"]["value"]
else:
station.continent = ""
if result["country"] is not None: if "country" in result:
station.country = result["country"]["value"] station.country = result["country"]
else: else:
station.country = "" station.country = ""
if result["city"] is not None: if "city" in result:
station.city = result["city"]["value"] station.city = result["city"]
else: else:
station.city = "" station.city = ""
if result["name"] is not None: if "name" in result:
station.name = result["name"]["value"] station.name = result["name"]
else: else:
station.name = "" station.name = ""
if result["subdomain"] is not None: if "shortDescription" in result:
station.slug = result["subdomain"]["value"] station.description = result["shortDescription"]
else:
station.slug = ""
if result["shortDescription"] is not None:
station.description = result["shortDescription"]["value"]
else: else:
station.description = "" station.description = ""
if "genres" in result:
station.genres = ", ".join(result["genres"])
else:
station.genres = ""
if "logo44x44" in result:
station.image_tiny = result["logo44x44"] station.image_tiny = result["logo44x44"]
station.image_small = result["logo100x100"] else:
station.image_tiny = ""
if "logo100x100" in result:
station.image_tiny = result["logo100x100"]
else:
station.image_tiny = ""
if "logo175x175" in result:
station.image_medium = result["logo175x175"] station.image_medium = result["logo175x175"]
else:
station.image_medium = ""
if "logo300x300" in result:
station.image_large = result["logo300x300"]
else:
station.image_large = ""
if "streams" in result:
station.stream_url = self._get_stream_url(result["streams"], self.min_bitrate)
self.stations_by_id[station.id] = station self.stations_by_id[station.id] = station
self.stations_by_slug[station.slug] = station
return station return station
def get_genres(self): def get_genres(self):
@@ -240,106 +229,100 @@ class RadioNetClient(object):
if cached is not None: if cached is not None:
return cached return cached
api_suffix = "/search/get" + key api_suffix = "/stations/tags"
response = self.do_get(api_suffix) response = self.do_get(api_suffix)
if response.status_code != 200: if response is None:
logger.error( logger.error("Radio.net: Error on get item list")
"Radio.net: Error on get item list "
+ str(api_suffix)
+ ". Error: "
+ response.text
)
return False return False
return self.set_cache(key, response.json(), 1440)
def get_sorted_category(self, category, name, sorting, page): return self.set_cache(key, self._filter_result(response, key, 0), 1440)
def _filter_result(self, data, tag, min_count_stations):
api_result = data.get(tag, [])
if api_result and min_count_stations:
# filter result by minimum count of stations
result = []
for item in api_result:
if item['count'] >= min_count_stations:
result.append(item)
return result
return api_result
def get_category(self, category, name, page=1):
results = [] results = []
for result in self._get_sorted_category(category, name, sorting, page): for result in self._get_category(category, name, page):
results.append(self._get_station_from_search_result(result)) results.append(self._get_station_from_search_result(result))
return results return results
def _get_sorted_category(self, category, name, sorting, page): def _get_category(self, category, name, page=1):
if sorting == "az": cache_key = category + "/" + name + "/" + str(page)
sorting = "STATION_NAME"
else:
sorting = "RANK"
cache_key = category + "/" + name + "/" + sorting + "/" + str(page)
cache = self.get_cache(cache_key) cache = self.get_cache(cache_key)
if cache is not None: if cache is not None:
return cache return cache
api_suffix = "/search/stationsby" + self.category_param_map[category] api_suffix = "/stations/by-tag"
url_params = { url_params = {
self.category_param_map[category]: name, "tagType": category,
"sorttype": sorting, "slug": name,
"sizeperpage": 50, "count": 50,
"pageindex": page, "offset": (page - 1) * 50,
} }
response = self.do_get(api_suffix, url_params) response = self.do_get(api_suffix, url_params)
if response.status_code != 200: if response is None:
logger.error( logger.error("Radio.net: Error on get station by " + str(category))
"Radio.net: Error on get station by "
+ str(category)
+ ". Error: "
+ response.text
)
return False return False
json = response.json() self.set_cache(category + "/" + name, int(math.ceil(response["totalCount"] / 50)), 10)
self.set_cache(category + "/" + name, int(json["numberPages"]), 10) return self.set_cache(cache_key, response["playables"], 10)
return self.set_cache(cache_key, json["categories"][0]["matches"], 10)
def get_category(self, category, page): def get_simple_category(self, category, page=1):
results = [] results = []
for result in self._get_category(category, page): for result in self._get_simple_category(category, page):
results.append(self._get_station_from_search_result(result)) results.append(self._get_station_from_search_result(result))
return results return results
def _get_category(self, category, page): def _get_simple_category(self, category, page=1):
cache_key = category + "/" + str(page) cache_key = category + "/" + str(page)
cache = self.get_cache(cache_key) cache = self.get_cache(cache_key)
if cache is not None: if cache is not None:
return cache return cache
api_suffix = "/search/" + category api_suffix = "/stations/" + category
url_params = {"sizeperpage": 50, "pageindex": page} url_params = {"count": 50, "offset": (page - 1) * 50}
response = self.do_get(api_suffix, url_params) response = self.do_get(api_suffix, url_params)
if response.status_code != 200: if response is None:
logger.error( logger.error("Radio.net: Error on get station by " + str(category))
"Radio.net: Error on get station by "
+ str(category)
+ ". Error: "
+ response.text
)
return False return False
json = response.json() self.set_cache(category, int(math.ceil(response["totalCount"] / 50)), 10)
self.set_cache(category, int(json["numberPages"]), 10) return self.set_cache(cache_key, response["playables"], 10)
return self.set_cache(cache_key, json["categories"][0]["matches"], 10)
def get_sorted_category_pages(self, category, name): def get_category_pages(self, category, name):
cache_key = category + "/" + name cache_key = category + "/" + name
cache = self.get_cache(cache_key) cache = self.get_cache(cache_key)
if cache is not None: if cache is not None:
return cache return cache
self.get_sorted_category(category, name, "rank", 1) self.get_category(category, name, 1)
return self.get_cache(cache_key) return self.get_cache(cache_key)
def get_category_pages(self, category): def get_simple_category_pages(self, category):
cache_key = category cache_key = category
cache = self.get_cache(cache_key) cache = self.get_cache(cache_key)
if cache is not None: if cache is not None:
return cache return cache
self.get_category(category, 1) self.get_simple_category(category, 1)
return self.get_cache(cache_key) return self.get_cache(cache_key)
@@ -355,29 +338,25 @@ class RadioNetClient(object):
favorite_stations = [] favorite_stations = []
for station_slug in self.favorites: for station_slug in self.favorites:
station = self.get_station_by_slug(station_slug) station = self.get_station_by_id(station_slug)
if station is False: if station is False:
api_suffix = "/search/stationsonly" api_suffix = "/stations/search"
url_params = { url_params = {
"query": station_slug, "query": station_slug,
"pageindex": 1, "count": 1,
"offset": 0,
} }
response = self.do_get(api_suffix, url_params) response = self.do_get(api_suffix, url_params)
if response.status_code != 200: if response is None:
logger.error("Radio.net: Search error " + response.text) logger.error("Radio.net: Search error")
else: else:
logger.debug("Radio.net: Done search") logger.debug("Radio.net: Done search")
json = response.json()
number_pages = int(json["numberPages"]) if "playables" in response and len(response["playables"]) > 0:
if number_pages != 0:
# take only the first match! # take only the first match!
station = self._get_station_from_search_result( station = self._get_station_from_search_result(response["playables"][0])
json["categories"][0]["matches"][0]
)
else: else:
logger.warning("Radio.net: No results for %s", station_slug) logger.warning("Radio.net: No results for %s", station_slug)
@@ -391,29 +370,30 @@ class RadioNetClient(object):
def do_search(self, query_string, page_index=1, search_results=None): def do_search(self, query_string, page_index=1, search_results=None):
api_suffix = "/search/stationsonly" api_suffix = "/stations/search"
url_params = { url_params = {
"query": query_string, "query": query_string,
"sizeperpage": 50, "count": 50,
"pageindex": page_index, "offset": (page_index - 1) * 50,
} }
logger.info(url_params)
response = self.do_get(api_suffix, url_params) response = self.do_get(api_suffix, url_params)
if response.status_code != 200: if response is None:
logger.error("Radio.net: Search error " + response.text) logger.error("Radio.net: Search error")
else: else:
logger.debug("Radio.net: Done search") logger.info("Radio.net: Done search")
if search_results is None: if search_results is None:
search_results = [] search_results = []
json = response.json() for match in response["playables"]:
for match in json["categories"][0]["matches"]:
station = self._get_station_from_search_result(match) station = self._get_station_from_search_result(match)
if station and station.playable: if station and station.playable:
search_results.append(station) search_results.append(station)
number_pages = int(json["numberPages"]) number_pages = int(math.ceil(response["totalCount"] / 50))
if number_pages >= page_index: # Search is utterly broken, don't retrieve more than 10 pages
logger.info(page_index)
if number_pages > page_index and page_index < 10:
self.do_search(query_string, page_index + 1, search_results) self.do_search(query_string, page_index + 1, search_results)
else: else:
logger.info( logger.info(
@@ -431,12 +411,13 @@ class RadioNetClient(object):
stream_url = None stream_url = None
for stream in stream_json: for stream in stream_json:
if int(stream["bitRate"]) >= bit_rate and stream["streamStatus"] == "VALID": logger.info("Radio.net: Found stream URL " + stream["url"])
stream_url = stream["streamUrl"] if ("bitRate" in stream and int(stream["bitRate"]) >= bit_rate) and stream["status"] == "VALID":
stream_url = stream["url"]
break break
if stream_url is None and len(stream_json) > 0: if stream_url is None and len(stream_json) > 0:
stream_url = stream_json[0]["streamUrl"] stream_url = stream_json[0]["url"]
return stream_url return stream_url

1
tests/conftest.py
View File

@@ -12,7 +12,6 @@ def backend_mock():
backend_mock = mock.Mock(spec=backend.RadioNetBackend) backend_mock = mock.Mock(spec=backend.RadioNetBackend)
backend_mock.radionet = RadioNetClient(proxy_config=None) backend_mock.radionet = RadioNetClient(proxy_config=None)
backend_mock.library = RadioNetLibraryProvider(backend=backend_mock) backend_mock.library = RadioNetLibraryProvider(backend=backend_mock)
backend_mock.radionet.set_apikey('test')
backend_mock.radionet.set_favorites({'lush'}) backend_mock.radionet.set_favorites({'lush'})
return backend_mock return backend_mock

69
tests/test_library.py
View File

@@ -1,12 +1,10 @@
def test_browse_root(library): def test_browse_root(library):
results = library.browse('radionet:root') results = library.browse('radionet:root')
assert 8 == len(results) assert 7 == len(results)
def test_browse_localstations(library): def test_browse_localstations(library):
results = library.browse('radionet:localstations') results = library.browse('radionet:local')
assert len(results) > 0 assert len(results) > 0
page_uri = results[0].uri if results is not None else None page_uri = results[0].uri if results is not None else None
@@ -17,28 +15,23 @@ def test_browse_localstations(library):
# assert len(results) > 0 # assert len(results) > 0
def test_browse_topstations(library):
results = library.browse('radionet:topstations')
assert len(results) > 0
def test_browse_genres(library): def test_browse_genres(library):
results = library.browse('radionet:genres') genres = library.browse('radionet:genres')
assert len(results) > 0 assert len(genres) > 0
cat_uri = results[0].uri if results is not None else None cat_uri = genres[0].uri if genres is not None else None
assert cat_uri is not None assert cat_uri is not None
results = library.browse(cat_uri) pages = library.browse(cat_uri)
assert len(results) == 2 assert len(pages) == 7
sort_uri = results[0].uri if results is not None else None page_uri = pages[0].uri if pages is not None else None
assert sort_uri is not None assert page_uri is not None
results = library.browse(sort_uri) results = library.browse(page_uri)
assert len(results) > 0 assert len(results) > 0
page_uri = results[0].uri if results is not None else None page_uri = pages[len(pages) - 1].uri if results is not None else None
assert page_uri is not None assert page_uri is not None
results = library.browse(page_uri) results = library.browse(page_uri)
@@ -47,19 +40,13 @@ def test_browse_genres(library):
def test_browse_topics(library): def test_browse_topics(library):
results = library.browse('radionet:topics') results = library.browse('radionet:topics')
assert len(results) > 0 assert len(results) > 2
cat_uri = results[0].uri if results is not None else None cat_uri = results[2].uri if results is not None else None
assert cat_uri is not None assert cat_uri is not None
results = library.browse(cat_uri) results = library.browse(cat_uri)
assert len(results) == 2 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 page_uri = results[0].uri if results is not None else None
assert page_uri is not None assert page_uri is not None
@@ -77,13 +64,7 @@ def test_browse_languages(library):
assert cat_uri is not None assert cat_uri is not None
results = library.browse(cat_uri) results = library.browse(cat_uri)
assert len(results) == 2 assert len(results) == 11
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 page_uri = results[0].uri if results is not None else None
assert page_uri is not None assert page_uri is not None
@@ -103,12 +84,6 @@ def test_browse_cities(library):
results = library.browse(cat_uri) results = library.browse(cat_uri)
assert len(results) == 2 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 page_uri = results[0].uri if results is not None else None
assert page_uri is not None assert page_uri is not None
@@ -125,13 +100,7 @@ def test_browse_countries(library):
assert cat_uri is not None assert cat_uri is not None
results = library.browse(cat_uri) results = library.browse(cat_uri)
assert len(results) == 2 assert len(results) == 4
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 page_uri = results[0].uri if results is not None else None
assert page_uri is not None assert page_uri is not None
@@ -147,13 +116,13 @@ def test_browse_favorites(library):
def test_search(library): def test_search(library):
result = library.search({'any': ['radio ram']}) result = library.search({'any': ['katze']})
assert len(result.tracks) > 0 assert len(result.tracks) > 0
old_length = len(result.tracks) old_length = len(result.tracks)
result = library.search({'any': ['radio ram']}) result = library.search({'any': ['katze']})
assert len(result.tracks) == old_length assert len(result.tracks) == old_length
@@ -169,4 +138,4 @@ def test_lookup(library):
def test_track_by_slug(library): def test_track_by_slug(library):
results = library.lookup('radionet:track:dancefm') results = library.lookup('radionet:track:dancefm')
assert 1 == len(results) assert 1 == len(results)
assert results[0].uri == 'radionet:track:2180' assert results[0].uri == 'radionet:track:dancefm'

22
tests/test_radionet.py
View File

@@ -4,39 +4,33 @@ def test_get_genres(radionet):
genres = radionet.get_genres() genres = radionet.get_genres()
assert len(genres) > 0 assert len(genres) > 0
def test_get_top_stations(radionet):
result = radionet.get_category('topstations', 1)
assert len(result) > 0
def test_get_local_stations(radionet): def test_get_local_stations(radionet):
result = radionet.get_category('localstations', 1) result = radionet.get_simple_category('local', 1)
assert len(result) > 0 assert len(result) > 0
def test_do_search(radionet): def test_do_search(radionet):
result = radionet.do_search("radio ram") result = radionet.do_search("\"radio ram\"")
assert len(result) > 0 assert len(result) > 0
assert result[0].stream_url is None assert result[0].stream_url is not None
assert radionet.get_stream_url(result[0].id) is not None assert radionet.get_stream_url(result[0].id) is not None
def test_get_favorites(radionet): def test_get_favorites(radionet):
radionet.cache = {}; radionet.cache = {}
test_favorites = ["Rock Antenne", "radio ram", "eska", "dancefm"] test_favorites = ["rockantenne", "eska", "dancefm"]
radionet.set_favorites(test_favorites) radionet.set_favorites(test_favorites)
result = radionet.get_favorites() result = radionet.get_favorites()
assert len(result) == len(test_favorites) assert len(result) == len(test_favorites)
assert result[2].name == 'Eska' assert result[1].name == 'Eska'
def test_favorites_broken_slug(radionet): def test_favorites_broken_slug(radionet):
radionet.cache = {}; radionet.cache = {}
test_favorites = ["radio 357"] test_favorites = ["radio 357"]
radionet.set_favorites(test_favorites) radionet.set_favorites(test_favorites)
result = radionet.get_favorites() result = radionet.get_favorites()
assert len(result) == 0 assert len(result) == 1