plugins.garena - Added new plugin garena

docs/plugin_matrix.rst

@@ -92,6 +92,7 @@ foxtr               fox.com.tr           Yes   No
 funimationnow       - funimation.com     --    Yes
                     - funimationnow.uk
 furstream           furstre.am           Yes   No
+garena              garena.live          Yes   --
 gomexp              gomexp.com           Yes   No
 goodgame            goodgame.ru          Yes   No    Only HLS streams are available.
 gulli               replay.gulli.fr      Yes   Yes   Streams may be geo-restricted to France.

src/streamlink/plugins/garena.py

@@ -0,0 +1,69 @@
+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"https?\:\/\/garena\.live\/(?:(?P<channel_id>\d+)|(?P<alias>\w+))")
+
+
+class Garena(Plugin):
+    API_INFO = "https://garena.live/api/channel_info_get"
+    API_STREAM = "https://garena.live/api/channel_stream_get"
+
+    _info_schema = validate.Schema(
+        {
+            "reply": validate.any({
+                "channel_id": int,
+            }, None),
+            "result": validate.text
+        }
+    )
+    _stream_schema = validate.Schema(
+        {
+            "reply": validate.any({
+                "streams": [
+                    {
+                        "url": validate.text,
+                        "resolution": int,
+                        "bitrate": int,
+                        "format": int
+                    }
+                ]
+            }, None),
+            "result": validate.text
+        }
+    )
+
+    @classmethod
+    def can_handle_url(self, url):
+        return _url_re.match(url)
+
+    def _post_api(self, api, payload, schema):
+        res = http.post(api, json=payload)
+        data = http.json(res, schema=schema)
+
+        if data["result"] == "success":
+            post_data = data["reply"]
+            return post_data
+
+    def _get_streams(self):
+        match = _url_re.match(self.url)
+        if match.group("alias"):
+            payload = {"alias": match.group("alias")}
+            info_data = self._post_api(self.API_INFO, payload, self._info_schema)
+            channel_id = info_data["channel_id"]
+        elif match.group("channel_id"):
+            channel_id = int(match.group("channel_id"))
+
+        if channel_id:
+            payload = {"channel_id": channel_id}
+            stream_data = self._post_api(self.API_STREAM, payload, self._stream_schema)
+            for stream in stream_data["streams"]:
+                n = "{0}p".format(stream["resolution"])
+                if stream["format"] == 3:
+                    s = HLSStream(self.session, stream["url"])
+                    yield n, s
+
+__plugin__ = Garena

tests/test_plugin_garena.py

@@ -0,0 +1,86 @@
+import json
+import unittest
+
+from streamlink import Streamlink
+
+try:
+    from unittest.mock import patch, Mock
+except ImportError:
+    from mock import patch, Mock
+
+from streamlink.plugins.garena import Garena
+
+
+class TestPluginGarena(unittest.TestCase):
+    def setUp(self):
+        self.session = Streamlink()
+
+    def test_can_handle_url(self):
+        # should match
+        self.assertTrue(Garena.can_handle_url("https://garena.live/LOLTW"))
+        self.assertTrue(Garena.can_handle_url("https://garena.live/358220"))
+
+        # shouldn't match
+        self.assertFalse(Garena.can_handle_url("http://local.local/"))
+        self.assertFalse(Garena.can_handle_url("http://localhost.localhost/"))
+
+    @patch('streamlink.plugins.garena.http')
+    def test_post_api_info(self, mock_http):
+        API_INFO = Garena.API_INFO
+        schema = Garena._info_schema
+
+        api_data = {
+            "reply": {
+                "channel_id": 358220,
+            },
+            "result": "success"
+        }
+
+        api_resp = Mock()
+        api_resp.text = json.dumps(api_data)
+        mock_http.post.return_value = api_resp
+        mock_http.json.return_value = api_data
+
+        payload = {"alias": "LOLTW"}
+
+        plugin = Garena("https://garena.live/LOLTW")
+
+        info_data = plugin._post_api(API_INFO, payload, schema)
+
+        self.assertEqual(info_data["channel_id"], 358220)
+
+        mock_http.post.assert_called_with(API_INFO, json=dict(alias="LOLTW"))
+
+    @patch('streamlink.plugins.garena.http')
+    def test_post_api_stream(self, mock_http):
+        API_STREAM = Garena.API_STREAM
+        schema = Garena._stream_schema
+
+        api_data = {
+            "reply": {
+                "streams": [
+                    {
+                        "url": "https://test.se/stream1",
+                        "bitrate": 0,
+                        "resolution": 1080,
+                        "format": 3
+                    },
+                ]
+            },
+            "result": "success"
+        }
+
+        api_resp = Mock()
+        api_resp.text = json.dumps(api_data)
+        mock_http.post.return_value = api_resp
+        mock_http.json.return_value = api_data
+
+        payload = {"channel_id": 358220}
+
+        plugin = Garena("https://garena.live/358220")
+
+        stream_data = plugin._post_api(API_STREAM, payload, schema)
+
+        self.assertEqual(stream_data["streams"], api_data["reply"]["streams"])
+
+        mock_http.post.assert_called_with(API_STREAM, json=dict(channel_id=358220))