Repository: streamlink/streamlink
Pull request #689 (2017-03-10 14:57:39)

plugins.camsoda Added support for camsoda.com
works only for live streams, not for vods

Fixed streamlink/streamlink#683


Patch:
diff --git a/docs/plugin_matrix.rst b/docs/plugin_matrix.rst
index 0da4fee6..2f611217 100644
--- a/docs/plugin_matrix.rst
+++ b/docs/plugin_matrix.rst
@@ -35,6 +35,7 @@ bilibili            live.bilibili.com    Yes   ?
 bliptv              blip.tv              --    Yes
 bongacams           bongacams.com        Yes   No    Only RTMP streams are available.
 btv                 btv.bg               Yes   No    Requires login, and geo-restricted to Bulgaria.
+camsoda             camsoda.com          Yes   No
 canalplus           - canalplus.fr       Yes   Yes   Streams may be geo-restricted to France.
                     - c8.fr
                     - cstar.fr
diff --git a/src/streamlink/plugins/camsoda.py b/src/streamlink/plugins/camsoda.py
new file mode 100644
index 00000000..b91f3841
--- /dev/null
+++ b/src/streamlink/plugins/camsoda.py
@@ -0,0 +1,96 @@
+import random
+import re
+
+from streamlink.plugin import Plugin
+from streamlink.plugin.api import http
+from streamlink.plugin.api import validate
+from streamlink.stream import HLSStream
+
+_url_re = re.compile(r"http(s)?://(www\.)?camsoda\.com/(?P<username>[^\"\']+)")
+
+_api_user_schema = validate.Schema(
+    {
+        "status": validate.any(int, validate.text),
+        validate.optional("user"): {
+            "online": validate.any(int, validate.text),
+            "chatstatus": validate.text,
+        }
+    }
+)
+
+_api_video_schema = validate.Schema(
+    {
+        "token": validate.text,
+        "app": validate.text,
+        "edge_servers": [validate.text],
+        "private_servers": [validate.text],
+        "mjpeg_server": validate.text,
+        "stream_name": validate.text
+    }
+)
+
+
+class Camsoda(Plugin):
+    API_URL_USER = "https://www.camsoda.com/api/v1/user/{0}"
+    API_URL_VIDEO = "https://www.camsoda.com/api/v1/video/vtoken/{0}?username=guest_{1}"
+    HLS_URL_VIDEO = "https://{server}/{app}/mp4:{stream_name}_mjpeg/playlist.m3u8?token={token}"
+
+    @classmethod
+    def can_handle_url(cls, url):
+        return _url_re.match(url)
+
+    def _stream_status(self, data_user):
+        invalid_username = data_user["status"] is False
+        if invalid_username:
+            self.logger.info("No validate username found for {0}".format(self.url))
+            return
+
+        is_offline = data_user["user"]["online"] is False
+        if is_offline:
+            self.logger.info("This stream is currently offline")
+            return
+
+        return True
+
+    def _get_api_user(self, username):
+        res = http.get(self.API_URL_USER.format(username))
+        data_user = http.json(res, schema=_api_user_schema)
+        return data_user
+
+    def _get_api_video(self, username):
+        res = http.get(self.API_URL_VIDEO.format(username, str(random.randint(1000, 99999))))
+        data_video = http.json(res, schema=_api_video_schema)
+        return data_video
+
+    def _get_hls_url(self, data_user, data_video):
+        is_edge = data_user["user"]["chatstatus"] == "online"
+        is_priv = data_user["user"]["chatstatus"] == "private"
+        if is_edge:
+            server = data_video["edge_servers"][0]
+        elif is_priv:
+            server = data_video["private_servers"][0]
+        else:
+            server = data_video["mjpeg_server"]
+
+        hls_url = self.HLS_URL_VIDEO.format(server=server, app=data_video["app"], stream_name=data_video["stream_name"], token=data_video["token"])
+        return hls_url
+
+    def _get_streams(self):
+        match = _url_re.match(self.url)
+        username = match.group("username")
+        username = username.replace("/", "")
+
+        data_user = self._get_api_user(username)
+        stream_status = self._stream_status(data_user)
+
+        if stream_status:
+            data_video = self._get_api_video(username)
+            hls_url = self._get_hls_url(data_user, data_video)
+
+            try:
+                for s in HLSStream.parse_variant_playlist(self.session, hls_url).items():
+                    yield s
+            except IOError as err:
+                self.logger.error("Error parsing stream: {0}", err)
+
+__plugin__ = Camsoda


Test patch:
diff --git a/tests/test_plugin_camsoda.py b/tests/test_plugin_camsoda.py
new file mode 100644
index 00000000..444ffbca
--- /dev/null
+++ b/tests/test_plugin_camsoda.py
@@ -0,0 +1,68 @@
+import unittest
+
+from streamlink import Streamlink
+
+try:
+    from unittest.mock import patch, Mock
+except ImportError:
+    from mock import patch, Mock
+
+from streamlink.plugins.camsoda import Camsoda
+from streamlink.stream import HLSStream
+
+
+class TestPluginCamsoda(unittest.TestCase):
+    def setUp(self):
+        self.session = Streamlink()
+        self.plugin = Camsoda("https://www.camsoda.com/stream-name")
+
+    def test_can_handle_url(self):
+        # should match
+        self.assertTrue(Camsoda.can_handle_url("https://www.camsoda.com/stream-name"))
+        self.assertTrue(Camsoda.can_handle_url("https://www.camsoda.com/streamname"))
+        self.assertTrue(Camsoda.can_handle_url("https://www.camsoda.com/username"))
+
+        # shouldn't match
+        self.assertFalse(Camsoda.can_handle_url("http://local.local/"))
+        self.assertFalse(Camsoda.can_handle_url("http://localhost.localhost/"))
+
+    def test_get_hls_url(self):
+        api_data_video = {
+            "token": "abcdefghijklmnopqrstuvwxyz123456",
+            "app": "cam",
+            "edge_servers": ["edge.server", "edge.server2"],
+            "private_servers": ["priv.server", "priv.server2"],
+            "mjpeg_server": "mjpeg.server",
+            "stream_name": "username_enc4"
+        }
+
+        values = [
+            {
+                "api_data_user": {"user": {"chatstatus": "online"}},
+                "server": "edge.server"
+            }, {
+                "api_data_user": {"user": {"chatstatus": "private"}},
+                "server": "priv.server"
+            }, {
+                "api_data_user": {"user": {"chatstatus": "foobar"}},
+                "server": "mjpeg.server"
+            }
+        ]
+        for data in values:
+            data_video = api_data_video
+            data_user = data["api_data_user"]
+            server = data["server"]
+
+            HLS_URL_VIDEO = self.plugin._get_hls_url(data_user, data_video)
+            HLS_URL_VIDEO_TEST = self.plugin.HLS_URL_VIDEO.format(server=server, app=data_video["app"], stream_name=data_video["stream_name"], token=data_video["token"])
+
+            self.assertEqual(HLS_URL_VIDEO, HLS_URL_VIDEO_TEST)
+
+    @patch('streamlink.plugins.camsoda.http')
+    @patch('streamlink.plugins.camsoda.HLSStream')
+    def test_get_streams(self, hlsstream, mock_http):
+        hlsstream.parse_variant_playlist.return_value = {"test": HLSStream(self.session, "http://test.se/stream1")}
+
+        streams = self.plugin.get_streams()
+
+        self.assertTrue("test" in streams)