✨ (api) getnext track api

klangbecken.py

@@ -235,6 +235,25 @@ def update_data_analyzer(playlist, fileId, ext, data):
 ]
 
 
+def next_prio_track_analyzer(data_dir):
+    data = {}
+    with open(os.path.join(data_dir, 'prio.m3u'), 'r') as f:
+        contents = f.read().strip()
+        if contents:
+            _, ext = os.path.splitext(contents)
+            changes = mutagen_tag_analyzer(
+                'prio.m3u',
+                None,
+                ext,
+                open(os.path.join(data_dir, contents), 'rb')
+            )
+            for change in changes:
+                if isinstance(change, MetadataChange):
+                    data[change.key] = change.value
+            data['next'] = True
+    return data
+
+
 ##############
 # Processors #
 ##############
@@ -464,12 +483,14 @@ def __init__(self,
                  secret,
                  upload_analyzers=DEFAULT_UPLOAD_ANALYZERS,
                  update_analyzers=DEFAULT_UPDATE_ANALYZERS,
+                 next_prio_track_info=next_prio_track_analyzer,
                  processors=DEFAULT_PROCESSORS,
                  disable_auth=False):
         self.data_dir = data_dir
         self.secret = secret
         self.upload_analyzers = upload_analyzers
         self.update_analyzers = update_analyzers
+        self.next_prio_track_info = next_prio_track_info
         self.processors = processors
         self.do_auth = not disable_auth
 
@@ -484,7 +505,8 @@ def __init__(self,
             Rule(playlist_url, methods=('POST',), endpoint='upload'),
             Rule(file_url, methods=('PUT',), endpoint='update'),
             Rule(file_url, methods=('DELETE',), endpoint='delete'),
-            Rule('/playnext/', methods=('POST',), endpoint='play_next')
+            Rule('/playnext/', methods=('POST',), endpoint='play_next'),
+            Rule('/playnext/', methods=('GET',), endpoint='get_next')
         ))
 
     def __call__(self, environ, start_response):
@@ -653,6 +675,13 @@ def on_play_next(self, request):
 
         return JSONResponse({'status': 'OK'})
 
+    def on_get_next(self, request):
+        resp = {'status': 'OK', 'next': False}
+
+        resp.update(self.next_prio_track_info(self.data_dir))
+
+        return JSONResponse(resp)
+
 
 class JSONResponse(Response):
     """

tests/playlists/single-track/prio.m3u

@@ -0,0 +1 @@
+../../audio/silence.mp3

tests/test_api.py

@@ -31,11 +31,14 @@ def setUp(self):
         self.update_analyzer = mock.Mock(return_value=['UpdateChange'])
         self.processor = mock.MagicMock()
 
+        self.next_prio_track_analyzer = mock.Mock(return_value=[])
+
         app = KlangbeckenAPI(
             'data_dir',
             'secret',
             upload_analyzers=[self.upload_analyzer],
             update_analyzers=[self.update_analyzer],
+            next_prio_track_info=self.next_prio_track_analyzer,
             processors=[self.processor],
             disable_auth=True,
         )
@@ -249,6 +252,11 @@ def testPlaynext(self, playnext_processor):
         self.assertTrue(b'invalid UTF-8 data' in resp.data)
         playnext_processor.assert_not_called()
 
+    def testGetNext(self):
+        resp = self.client.get('/playnext/')
+        self.assertEqual(resp.status_code, 200)
+        self.next_prio_track_analyzer.assert_called_with('data_dir')
+
 
 class AuthTestCase(unittest.TestCase):
     def setUp(self):
@@ -503,6 +511,24 @@ def testFFmpegAudioAnalyzer(self):
                 fs = FileStorage(f)
                 ffmpeg_audio_analyzer('music', 'id1', '.mp3', fs)
 
+    def testNextPrioTrackAnalyzer(self):
+        from klangbecken import next_prio_track_analyzer
+
+        # test empty file
+        path = os.path.join(self.current_path, 'playlists', 'empty')
+
+        data = next_prio_track_analyzer(path)
+
+        self.assertEquals(data, {})
+
+        # test file with track
+        path = os.path.join(self.current_path, 'playlists', 'single-track')
+
+        data = next_prio_track_analyzer(path)
+
+        self.assertEquals(data['artist'], 'Silence Artist')
+        self.assertEquals(data['title'], 'Silence Track')
+
 
 class ProcessorsTestCase(unittest.TestCase):
     def setUp(self):