-
-
Notifications
You must be signed in to change notification settings - Fork 1.2k
Expand file tree
/
Copy pathoneplusone.py
More file actions
127 lines (105 loc) · 4.07 KB
/
Copy pathoneplusone.py
File metadata and controls
127 lines (105 loc) · 4.07 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
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
"""
$description Ukrainian live TV channels from 1 + 1 Media group, including 1 + 1, 2 + 2, PLUSPLUS, TET and UNIAN.
$url 1plus1.video
$type live
"""
import re
from base64 import b64decode
from time import time
from urllib.parse import urljoin, urlparse
from streamlink.exceptions import NoStreamsError, PluginError
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.utils.times import fromlocaltimestamp
log = getLogger(__name__)
class OnePlusOneHLS(HLSStream):
__shortname__ = "hls-oneplusone"
def __init__(self, session, url, self_url=None, **args):
super().__init__(session, url, None, **args)
self._url = url
first_parsed = urlparse(self._url)
self._first_netloc = first_parsed.netloc
self._first_path_chunklist = first_parsed.path.split("/")[-1]
self.watch_timeout = int(first_parsed.path.split("/")[2]) - 15
self.api = OnePlusOneAPI(session, self_url)
def _next_watch_timeout(self):
when = fromlocaltimestamp(self.watch_timeout).isoformat(" ")
log.debug(f"next watch_timeout at {when}")
def open(self):
self._next_watch_timeout()
return super().open()
@property
def url(self):
if int(time()) >= self.watch_timeout:
log.debug("Reloading HLS URL")
hls_url = self.api.get_hls_url()
if not hls_url:
self.watch_timeout += 10
return self._url
parsed = urlparse(hls_url)
path_parts = parsed.path.split("/")
path_parts[-1] = self._first_path_chunklist
self.watch_timeout = int(path_parts[2]) - 15
self._next_watch_timeout()
self._url = parsed._replace(
netloc=self._first_netloc,
path="/".join(list(path_parts)),
).geturl()
return self._url
class OnePlusOneAPI:
def __init__(self, session, url):
self.session = session
self.url = url
def get_hls_url(self):
self.session.http.cookies.clear()
url_parts = self.session.http.get(
url=self.url,
schema=validate.Schema(
validate.parse_html(),
validate.xml_xpath_string(".//iframe[contains(@src,'embed')]/@src"),
),
)
if not url_parts:
raise NoStreamsError
log.trace("url_parts=%s", url_parts)
self.session.http.headers.update({"Referer": self.url})
try:
url_ovva = self.session.http.get(
url=urljoin(self.url, url_parts),
schema=validate.Schema(
validate.parse_html(),
validate.xml_xpath_string(".//script[@type='text/javascript'][contains(text(),'ovva-player')]/text()"),
str,
validate.regex(re.compile(r"ovva-player\",\"([^\"]*)\"\)")),
validate.get(1),
validate.transform(lambda x: b64decode(x).decode()),
validate.parse_json(),
{"balancer": validate.url()},
validate.get("balancer"),
),
)
except PluginError as err:
log.error(f"ovva-player: {err}")
return
log.debug(f"url_ovva={url_ovva}")
return self.session.http.get(
url=url_ovva,
schema=validate.Schema(
validate.transform(lambda x: x.split("=")),
["302", validate.url(path=validate.endswith(".m3u8"))],
validate.get(1),
),
)
@pluginmatcher(
re.compile(r"https?://1plus1\.video/(?:\w{2}/)?tvguide/[^/]+/online"),
)
class OnePlusOne(Plugin):
def _get_streams(self):
self.api = OnePlusOneAPI(self.session, self.url)
url_hls = self.api.get_hls_url()
if not url_hls:
return
return OnePlusOneHLS.parse_variant_playlist(self.session, url_hls, self_url=self.url)
__plugin__ = OnePlusOne