stream: Extract first track from playlists

docs/changelog.rst

@@ -85,6 +85,12 @@ MPD frontend
 - Track data now include the ``Last-Modified`` field if set on the track model.
   (Fixes: :issue:`1218`, PR: :issue:`1219`)
 
+Stream backend
+--------------
+
+- Move stream playlist parsing from GStreamer to the stream backend. (Fixes:
+  :issue:`671`)
+
 Local backend
 -------------
 

mopidy/stream/actor.py

@@ -3,12 +3,16 @@ from __future__ import absolute_import, unicode_literals
 import fnmatch
 import logging
 import re
+import time
 import urlparse
 
 import pykka
 
-from mopidy import audio as audio_lib, backend, exceptions
+import requests
+
+from mopidy import audio as audio_lib, backend, exceptions, httpclient, stream
 from mopidy.audio import scan, utils
+from mopidy.internal import playlists
 from mopidy.models import Track
 
 logger = logging.getLogger(__name__)
@@ -19,18 +23,19 @@ class StreamBackend(pykka.ThreadingActor, backend.Backend):
     def __init__(self, config, audio):
         super(StreamBackend, self).__init__()
 
+        self._scanner = scan.Scanner(
+            timeout=config['stream']['timeout'],
+            proxy_config=config['proxy'])
+
         self.library = StreamLibraryProvider(
             backend=self, blacklist=config['stream']['metadata_blacklist'])
-        self.playback = backend.PlaybackProvider(audio=audio, backend=self)
+        self.playback = StreamPlaybackProvider(
+            audio=audio, backend=self, config=config)
         self.playlists = None
 
         self.uri_schemes = audio_lib.supported_uri_schemes(
             config['stream']['protocols'])
 
-        self._scanner = scan.Scanner(
-            timeout=config['stream']['timeout'],
-            proxy_config=config['proxy'])
-
 
 class StreamLibraryProvider(backend.LibraryProvider):
 
@@ -57,3 +62,78 @@ class StreamLibraryProvider(backend.LibraryProvider):
             track = Track(uri=uri)
 
         return [track]
+
+
+class StreamPlaybackProvider(backend.PlaybackProvider):
+
+    def __init__(self, audio, backend, config):
+        super(StreamPlaybackProvider, self).__init__(audio, backend)
+        self._config = config
+        self._scanner = backend._scanner
+
+    def translate_uri(self, uri):
+        try:
+            scan_result = self._scanner.scan(uri)
+        except exceptions.ScannerError as e:
+            logger.warning(
+                'Problem scanning URI %s: %s', uri, e)
+            return None
+
+        if not (scan_result.mime.startswith('text/') or
+                scan_result.mime.startswith('application/')):
+            return uri
+
+        content = self._download(uri)
+        if content is None:
+            return None
+
+        tracks = list(playlists.parse(content))
+        if tracks:
+            # TODO Test streams and return first that seems to be playable
+            return tracks[0]
+
+    def _download(self, uri):
+        timeout = self._config['stream']['timeout'] / 1000.0
+
+        session = get_requests_session(
+            proxy_config=self._config['proxy'],
+            user_agent='%s/%s' % (
+                stream.Extension.dist_name, stream.Extension.version))
+
+        try:
+            response = session.get(
+                uri, stream=True, timeout=timeout)
+        except requests.exceptions.Timeout:
+            logger.warning(
+                'Download of stream playlist (%s) failed due to connection '
+                'timeout after %.3fs', uri, timeout)
+            return None
+
+        deadline = time.time() + timeout
+        content = b''
+        for chunk in response.iter_content(4096):
+            content += chunk
+            if time.time() > deadline:
+                logger.warning(
+                    'Download of stream playlist (%s) failed due to download '
+                    'taking more than %.3fs', uri, timeout)
+                return None
+
+        if not response.ok:
+            logger.warning(
+                'Problem downloading stream playlist %s: %s',
+                uri, response.reason)
+            return None
+
+        return content
+
+
+def get_requests_session(proxy_config, user_agent):
+    proxy = httpclient.format_proxy(proxy_config)
+    full_user_agent = httpclient.format_user_agent(user_agent)
+
+    session = requests.Session()
+    session.proxies.update({'http': proxy, 'https': proxy})
+    session.headers.update({'user-agent': full_user_agent})
+
+    return session

tests/stream/test_playback.py

@@ -0,0 +1,145 @@
+from __future__ import absolute_import, unicode_literals
+
+import mock
+
+import pytest
+
+import requests
+
+import responses
+
+from mopidy import exceptions
+from mopidy.audio import scan
+from mopidy.stream import actor
+
+
+TIMEOUT = 1000
+URI = 'http://example.com/listen.m3u'
+BODY = """
+#EXTM3U
+http://example.com/stream.mp3
+http://foo.bar/baz
+""".strip()
+
+
+@pytest.fixture
+def config():
+    return {
+        'proxy': {},
+        'stream': {
+            'timeout': TIMEOUT,
+        },
+    }
+
+
+@pytest.fixture
+def audio():
+    return mock.Mock()
+
+
+@pytest.fixture
+def scanner():
+    scanner = mock.Mock(spec=scan.Scanner)
+    scanner.scan.return_value.mime = 'text/foo'
+    return scanner
+
+
+@pytest.fixture
+def backend(scanner):
+    backend = mock.Mock()
+    backend.uri_schemes = ['file']
+    backend._scanner = scanner
+    return backend
+
+
+@pytest.fixture
+def provider(audio, backend, config):
+    return actor.StreamPlaybackProvider(audio, backend, config)
+
+
+@responses.activate
+def test_translate_uri_of_audio_stream_returns_same_uri(
+        scanner, provider):
+
+    scanner.scan.return_value.mime = 'audio/ogg'
+
+    result = provider.translate_uri(URI)
+
+    scanner.scan.assert_called_once_with(URI)
+    assert result == URI
+
+
+@responses.activate
+def test_translate_uri_of_playlist_returns_first_uri_in_list(
+        scanner, provider):
+
+    responses.add(
+        responses.GET, URI, body=BODY, content_type='audio/x-mpegurl')
+
+    result = provider.translate_uri(URI)
+
+    scanner.scan.assert_called_once_with(URI)
+    assert result == 'http://example.com/stream.mp3'
+    assert responses.calls[0].request.headers['User-Agent'].startswith(
+        'Mopidy-Stream/')
+
+
+@responses.activate
+def test_translate_uri_of_playlist_with_xml_mimetype(scanner, provider):
+    scanner.scan.return_value.mime = 'application/xspf+xml'
+    responses.add(
+        responses.GET, URI, body=BODY, content_type='application/xspf+xml')
+
+    result = provider.translate_uri(URI)
+
+    scanner.scan.assert_called_once_with(URI)
+    assert result == 'http://example.com/stream.mp3'
+
+
+def test_translate_uri_when_scanner_fails(scanner, provider, caplog):
+    scanner.scan.side_effect = exceptions.ScannerError('foo failed')
+
+    result = provider.translate_uri('bar')
+
+    assert result is None
+    assert 'Problem scanning URI bar: foo failed' in caplog.text()
+
+
+@responses.activate
+def test_translate_uri_when_playlist_download_fails(provider, caplog):
+    responses.add(responses.GET, URI, body=BODY, status=500)
+
+    result = provider.translate_uri(URI)
+
+    assert result is None
+    assert 'Problem downloading stream playlist' in caplog.text()
+
+
+def test_translate_uri_times_out_if_connection_times_out(provider, caplog):
+    with mock.patch.object(actor.requests, 'Session') as session_mock:
+        get_mock = session_mock.return_value.get
+        get_mock.side_effect = requests.exceptions.Timeout
+
+        result = provider.translate_uri(URI)
+
+    get_mock.assert_called_once_with(URI, timeout=1.0, stream=True)
+    assert result is None
+    assert (
+        'Download of stream playlist (%s) failed due to connection '
+        'timeout after 1.000s' % URI in caplog.text())
+
+
+@responses.activate
+def test_translate_uri_times_out_if_download_is_slow(provider, caplog):
+    responses.add(
+        responses.GET, URI, body=BODY, content_type='audio/x-mpegurl')
+
+    with mock.patch.object(actor, 'time') as time_mock:
+        time_mock.time.side_effect = [0, TIMEOUT + 1]
+
+        result = provider.translate_uri(URI)
+
+    assert result is None
+    assert (
+        'Download of stream playlist (%s) failed due to download taking '
+        'more than 1.000s' % URI in caplog.text())