plugins.twitch: implement disable-ads parameter

- Implement TwitchHLSStream{,Worker,Writer,Reader} and TwitchM3U8Parser - Skip HLS segments between SCTE35-OUT{,-CONT} and SCTE35-IN tags - Add --twitch-disable-ads parameter (plugin specific) - Add tests
streamlink · gravyboat · Mar 29, 2019 · Mar 23, 2019 · Mar 23, 2019 · Mar 23, 2019
commit 94edeb071e28f118537649f7141eb7871398038d
diff --git a/src/streamlink/plugins/twitch.py b/src/streamlink/plugins/twitch.py
@@ -1,6 +1,8 @@
 # coding=utf-8
+import logging
 import re
 import warnings
+from collections import namedtuple
 from random import random
 
 import requests
@@ -13,13 +15,18 @@
 from streamlink.stream import (
     HTTPStream, HLSStream, FLVPlaylist, extract_flv_header_tags
 )
+from streamlink.stream.hls import HLSStreamReader, HLSStreamWriter, HLSStreamWorker
+from streamlink.stream.hls_playlist import M3U8Parser, load as load_hls_playlist
 from streamlink.utils.times import hours_minutes_seconds
 
 try:
     from itertools import izip as zip
 except ImportError:
     pass
 
+
+log = logging.getLogger(__name__)
+
 QUALITY_WEIGHTS = {
     "source": 1080,
     "1080": 1080,
@@ -128,6 +135,82 @@
 )
 
 
+Segment = namedtuple("Segment", "uri duration title key discontinuity scte35 byterange date map")
+
+
+class TwitchM3U8Parser(M3U8Parser):
+    def parse_tag_ext_x_scte35_out(self, value):
+        self.state["scte35"] = True
+
+    # unsure if this gets used by Twitch
+    def parse_tag_ext_x_scte35_out_cont(self, value):
+        self.state["scte35"] = True
+
+    def parse_tag_ext_x_scte35_in(self, value):
+        self.state["scte35"] = False
+
+    def get_segment(self, uri):
+        byterange = self.state.pop("byterange", None)
+        extinf = self.state.pop("extinf", (0, None))
+        date = self.state.pop("date", None)
+        map_ = self.state.get("map")
+        key = self.state.get("key")
+        discontinuity = self.state.pop("discontinuity", False)
+        scte35 = self.state.pop("scte35", None)
+
+        return Segment(
+            uri,
+            extinf[0],
+            extinf[1],
+            key,
+            discontinuity,
+            scte35,
+            byterange,
+            date,
+            map_
+        )
+
+
+class TwitchHLSStreamWorker(HLSStreamWorker):
+    def _reload_playlist(self, text, url):
+        return load_hls_playlist(text, url, parser=TwitchM3U8Parser)
+
+
+class TwitchHLSStreamWriter(HLSStreamWriter):
+    def __init__(self, *args, **kwargs):
+        HLSStreamWriter.__init__(self, *args, **kwargs)
+        options = self.session.plugins.get("twitch").options
+        self.disable_ads = options.get("disable-ads")
+        if self.disable_ads:
+            log.info("Will skip ad segments")
+
+    def write(self, sequence, *args, **kwargs):
+        if self.disable_ads:
+            if sequence.segment.scte35 is not None:
+                self.reader.ads = sequence.segment.scte35
+                if self.reader.ads:
+                    log.info("Will skip ads beginning with segment {0}".format(sequence.num))
+                else:
+                    log.info("Will stop skipping ads beginning with segment {0}".format(sequence.num))
+            if self.reader.ads:
+                return
+        return HLSStreamWriter.write(self, sequence, *args, **kwargs)
+
+
+class TwitchHLSStreamReader(HLSStreamReader):
+    __worker__ = TwitchHLSStreamWorker
+    __writer__ = TwitchHLSStreamWriter
+    ads = None
+
+
+class TwitchHLSStream(HLSStream):
+    def open(self):
+        reader = TwitchHLSStreamReader(self)
+        reader.open()
+
+        return reader
+
+
 class UsherService(object):
     def __init__(self, session):
         self.session = session
@@ -277,6 +360,13 @@ class Twitch(Plugin):
                        action="store_true",
                        help="""
         Do not open the stream if the target channel is hosting another channel.
+        """
+                       ),
+        PluginArgument("disable-ads",
+                       action="store_true",
+                       help="""
+        Skip embedded advertisement segments at the beginning or during a stream.
+        Will cause these segments to be missing from the stream.
         """
                        ))
 
@@ -640,9 +730,12 @@ def _get_hls_streams(self, stream_type="live"):
         try:
             # If the stream is a VOD that is still being recorded the stream should start at the
             # beginning of the recording
-            streams = HLSStream.parse_variant_playlist(self.session, url,
-                                                       start_offset=time_offset,
-                                                       force_restart=not stream_type == "live")
+            streams = TwitchHLSStream.parse_variant_playlist(
+                self.session,
+                url,
+                start_offset=time_offset,
+                force_restart=not stream_type == "live"
+            )
         except IOError as err:
             err = str(err)
             if "404 Client Error" in err or "Failed to parse playlist" in err:

diff --git a/tests/plugins/test_twitch.py b/tests/plugins/test_twitch.py
@@ -1,6 +1,17 @@
+import logging
 import unittest
+from functools import partial
 
-from streamlink.plugins.twitch import Twitch
+from streamlink.plugins.twitch import Twitch, TwitchHLSStream
+
+import requests_mock
+from tests.mock import call, patch
+
+from streamlink.session import Streamlink
+from tests.resources import text
+
+
+log = logging.getLogger(__name__)
 
 
 class TestPluginTwitch(unittest.TestCase):
@@ -21,3 +32,158 @@ def test_can_handle_url_negative(self):
         ]
         for url in should_not_match:
             self.assertFalse(Twitch.can_handle_url(url))
+
+
+class TestTwitchHLSStream(unittest.TestCase):
+    scte35_out = "#EXT-X-DISCONTINUITY\n#EXT-X-SCTE35-OUT\n"
+    scte35_out_cont = "#EXT-X-SCTE35-OUT-CONT\n"
+    scte35_in = "#EXT-X-DISCONTINUITY\n#EXT-X-SCTE35-IN\n"
+    segment = "#EXTINF:1.000,\nstream{0}.ts\n"
+
+    def getMasterPlaylist(self):
+        with text("hls/test_master.m3u8") as pl:
+            return pl.read()
+
+    def getPlaylist(self, media_sequence, items):
+        playlist = """
+#EXTM3U
+#EXT-X-VERSION:5
+#EXT-X-TARGETDURATION:1
+#EXT-X-MEDIA-SEQUENCE:{0}
+""".format(media_sequence)
+
+        for item in items:
+            if type(item) != int:
+                playlist += item
+            else:
+                playlist += self.segment.format(item)
+
+        return playlist
+
+    def start_streamlink(self, kwargs=None):
+        kwargs = kwargs or {}
+        log.info("Executing streamlink")
+        streamlink = Streamlink()
+
+        streamlink.set_option("hls-live-edge", 4)
+        streamlink.plugins.get("twitch").options.set("disable-ads", True)
+
+        masterStream = TwitchHLSStream.parse_variant_playlist(
+            streamlink,
+            "http://mocked/path/master.m3u8",
+            **kwargs
+        )
+        stream = masterStream["1080p (source)"].open()
+        data = b"".join(iter(partial(stream.read, 8192), b""))
+        stream.close()
+        log.info("End of streamlink execution")
+        return data
+
+    def mock(self, mocked, method, url, *args, **kwargs):
+        mocked[url] = method(url, *args, **kwargs)
+
+    def get_result(self, streams, playlists):
+        mocked = {}
+        with requests_mock.Mocker() as mock:
+            self.mock(mocked, mock.get, "http://mocked/path/master.m3u8", text=self.getMasterPlaylist())
+            self.mock(mocked, mock.get, "http://mocked/path/playlist.m3u8", [{"text": p} for p in playlists])
+            for i, stream in enumerate(streams):
+                self.mock(mocked, mock.get, "http://mocked/path/stream{0}.ts".format(i), content=stream)
+            return self.start_streamlink(), mocked
+
+    @patch("streamlink.plugins.twitch.log")
+    def test_hls_scte35_start_with_end(self, mock_logging):
+        streams = ["[{0}]".format(i).encode("ascii") for i in range(12)]
+        playlists = [
+            self.getPlaylist(0, [self.scte35_out, 0, 1, 2, 3]),
+            self.getPlaylist(4, [self.scte35_in, 4, 5, 6, 7]),
+            self.getPlaylist(8, [8, 9, 10, 11]) + "#EXT-X-ENDLIST\n"
+        ]
+        result, mocked = self.get_result(streams, playlists)
+
+        expected = b''.join(streams[4:12])
+        self.assertEqual(expected, result)
+        for i, _ in enumerate(streams):
+            self.assertTrue(mocked["http://mocked/path/stream{0}.ts".format(i)].called)
+        mock_logging.info.assert_has_calls([
+            call("Will skip ad segments"),
+            call("Will skip ads beginning with segment 0"),
+            call("Will stop skipping ads beginning with segment 4")
+        ])
+
+    @patch("streamlink.plugins.twitch.log")
+    def test_hls_scte35_no_start(self, mock_logging):
+        streams = ["[{0}]".format(i).encode("ascii") for i in range(8)]
+        playlists = [
+            self.getPlaylist(0, [0, 1, 2, 3]),
+            self.getPlaylist(4, [self.scte35_in, 4, 5, 6, 7]) + "#EXT-X-ENDLIST\n"
+        ]
+        result, mocked = self.get_result(streams, playlists)
+
+        expected = b''.join(streams[0:8])
+        self.assertEqual(expected, result)
+        for i, _ in enumerate(streams):
+            self.assertTrue(mocked["http://mocked/path/stream{0}.ts".format(i)].called)
+        mock_logging.info.assert_has_calls([
+            call("Will skip ad segments")
+        ])
+
+    @patch("streamlink.plugins.twitch.log")
+    def test_hls_scte35_no_start_with_cont(self, mock_logging):
+        streams = ["[{0}]".format(i).encode("ascii") for i in range(8)]
+        playlists = [
+            self.getPlaylist(0, [self.scte35_out_cont, 0, 1, 2, 3]),
+            self.getPlaylist(4, [self.scte35_in, 4, 5, 6, 7]) + "#EXT-X-ENDLIST\n"
+        ]
+        result, mocked = self.get_result(streams, playlists)
+
+        expected = b''.join(streams[4:8])
+        self.assertEqual(expected, result)
+        for i, _ in enumerate(streams):
+            self.assertTrue(mocked["http://mocked/path/stream{0}.ts".format(i)].called)
+        mock_logging.info.assert_has_calls([
+            call("Will skip ad segments"),
+            call("Will skip ads beginning with segment 0"),
+            call("Will stop skipping ads beginning with segment 4")
+        ])
+
+    @patch("streamlink.plugins.twitch.log")
+    def test_hls_scte35_no_end(self, mock_logging):
+        streams = ["[{0}]".format(i).encode("ascii") for i in range(12)]
+        playlists = [
+            self.getPlaylist(0, [0, 1, 2, 3]),
+            self.getPlaylist(4, [self.scte35_out, 4, 5, 6, 7]),
+            self.getPlaylist(8, [8, 9, 10, 11]) + "#EXT-X-ENDLIST\n"
+        ]
+        result, mocked = self.get_result(streams, playlists)
+
+        expected = b''.join(streams[0:4])
+        self.assertEqual(expected, result)
+        for i, _ in enumerate(streams):
+            self.assertTrue(mocked["http://mocked/path/stream{0}.ts".format(i)].called)
+        mock_logging.info.assert_has_calls([
+            call("Will skip ad segments"),
+            call("Will skip ads beginning with segment 4")
+        ])
+
+    @patch("streamlink.plugins.twitch.log")
+    def test_hls_scte35_in_between(self, mock_logging):
+        streams = ["[{0}]".format(i).encode("ascii") for i in range(20)]
+        playlists = [
+            self.getPlaylist(0, [0, 1, 2, 3]),
+            self.getPlaylist(4, [4, 5, self.scte35_out, 6, 7]),
+            self.getPlaylist(8, [8, 9, 10, 11]),
+            self.getPlaylist(12, [12, 13, self.scte35_in, 14, 15]),
+            self.getPlaylist(16, [16, 17, 18, 19]) + "#EXT-X-ENDLIST\n"
+        ]
+        result, mocked = self.get_result(streams, playlists)
+
+        expected = b''.join(streams[0:6]) + b''.join(streams[14:20])
+        self.assertEqual(expected, result)
+        for i, _ in enumerate(streams):
+            self.assertTrue(mocked["http://mocked/path/stream{0}.ts".format(i)].called)
+        mock_logging.info.assert_has_calls([
+            call("Will skip ad segments"),
+            call("Will skip ads beginning with segment 6"),
+            call("Will stop skipping ads beginning with segment 14")
+        ])