1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
|
# -*- coding: utf8 -*-
from io import BytesIO
from streamlink import NoStreamsError
from streamlink.plugins import Plugin
from streamlink.options import Options
from streamlink.stream import *
from streamlink.plugin.api.support_plugin import testplugin_support
class TestStream(Stream):
__shortname__ = "test"
def open(self):
return BytesIO(b'x' * 8192 * 2)
class TestPlugin(Plugin):
options = Options({
"a_option": "default"
})
@classmethod
def can_handle_url(self, url):
return "test.se" in url
def get_title(self):
return "Test Title"
def get_author(self):
return u"Tѥst Āuƭhǿr"
def get_category(self):
return None
def _get_streams(self):
if "empty" in self.url:
return
if "UnsortableStreamNames" in self.url:
def gen():
for i in range(3):
yield "vod", HTTPStream(self.session, "http://test.se/stream")
return gen()
if "NoStreamsError" in self.url:
raise NoStreamsError(self.url)
streams = {}
streams["test"] = TestStream(self.session)
streams["rtmp"] = RTMPStream(self.session, dict(rtmp="rtmp://test.se"))
streams["hls"] = HLSStream(self.session, "http://test.se/playlist.m3u8")
streams["http"] = HTTPStream(self.session, "http://test.se/stream")
streams["akamaihd"] = AkamaiHDStream(self.session, "http://test.se/stream")
streams["240p"] = HTTPStream(self.session, "http://test.se/stream")
streams["360p"] = HTTPStream(self.session, "http://test.se/stream")
streams["1080p"] = HTTPStream(self.session, "http://test.se/stream")
streams["350k"] = HTTPStream(self.session, "http://test.se/stream")
streams["800k"] = HTTPStream(self.session, "http://test.se/stream")
streams["1500k"] = HTTPStream(self.session, "http://test.se/stream")
streams["3000k"] = HTTPStream(self.session, "http://test.se/stream")
streams["480p"] = [HTTPStream(self.session, "http://test.se/stream"),
RTMPStream(self.session, dict(rtmp="rtmp://test.se"))]
streams.update(testplugin_support.get_streams(self.session))
return streams
__plugin__ = TestPlugin
|
