-
-
Notifications
You must be signed in to change notification settings - Fork 1.2k
Expand file tree
/
Copy pathkugou.py
More file actions
102 lines (88 loc) · 2.91 KB
/
kugou.py
File metadata and controls
102 lines (88 loc) · 2.91 KB
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
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
"""
$description Artist and fan live streaming for live video game broadcasts and individual live streams.
$url fanxing.kugou.com
$type live
"""
import re
import time
from streamlink.logger import getLogger
from streamlink.plugin import Plugin, pluginmatcher
from streamlink.plugin.api import validate
from streamlink.stream.hls import HLSStream
from streamlink.stream.http import HTTPStream
log = getLogger(__name__)
@pluginmatcher(
re.compile(r"https?://fanxing\.kugou\.com/(?P<room_id>\d+)"),
)
class Kugou(Plugin):
_roomid_re = re.compile(r"roomId:\s*'(\d+)'")
_room_stream_list_schema = validate.Schema(
{
"data": validate.any(
None,
{
"httpflv": validate.url(),
},
),
},
validate.get("httpflv_room_stream_list_schema"),
)
_stream_hv_schema = validate.Schema(
validate.any(
None,
[
{
"httpshls": [validate.url()],
"httpsflv": [validate.url()],
},
],
),
)
_stream_data_schema = validate.Schema({
"msg": str,
"code": int,
"data": {
"status": int,
"vertical": _stream_hv_schema,
"horizontal": _stream_hv_schema,
"roomId": int,
},
})
def _get_streams(self):
res = self.session.http.get(self.url)
m = self._roomid_re.search(res.text)
if m:
room_id = m.group(1)
else:
room_id = self.match.group("room_id")
res = self.session.http.get(
"https://fx2.service.kugou.com/video/pc/live/pull/v3/streamaddr",
params={
"ch": "fx",
"version": "1.0",
# 1=rtmp, 2=httpflv, 3=hls, 5=httpsflv, 6=httpshls
"streamType": "1-2-5-6",
"ua": "fx-flash",
"kugouId": "0",
"roomId": room_id,
"_": int(time.time()),
},
)
stream_data_json = self.session.http.json(res, schema=self._stream_data_schema)
log.trace("%r", stream_data_json)
if stream_data_json["code"] != 0 or stream_data_json["data"]["status"] != 1:
return
h = stream_data_json["data"]["horizontal"]
v = stream_data_json["data"]["vertical"]
stream_data = h[0] if h else v[0]
if stream_data.get("httpshls"):
for hls_url in stream_data["httpshls"]:
s = HLSStream.parse_variant_playlist(self.session, hls_url)
if not s:
yield "live", HLSStream(self.session, hls_url)
else:
yield from s.items()
if stream_data.get("httpsflv"):
for http_url in stream_data["httpsflv"]:
yield "live", HTTPStream(self.session, http_url)
__plugin__ = Kugou