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
|
import datetime
from yt_dlp.extractor.common import InfoExtractor
from yt_dlp.utils import (
join_nonempty,
update_url_query,
traverse_obj,
)
# "hacks" as in great jank/schizo shit that works anyway
def _generate_as_live_chunks(playlist_url, start_at, end_at):
chunks = []
chunk_length = 300 # max the api allows
duration = int(end_at.timestamp() - start_at.timestamp())
cursor = 0
while cursor < duration:
chunk_length = min(chunk_length, duration - cursor)
chunk_start = start_at + datetime.timedelta(seconds=cursor)
chunk_url = update_url_query(playlist_url, {
"seek": chunk_start.timestring(),
"l": chunk_length,
})
chunks.append(chunk_url)
cursor += chunk_length
return chunks
def _playlist_from_chunks(self, chunks, src_id, headers={}):
playlist = ""
for i, chunk in enumerate(chunks):
i +=1 # for more friendly cli output, it gets reset each loop so it shouldnt effect anything
chunk_id = join_nonempty(src_id, i)
base_format = self._extract_m3u8_formats(
chunk, chunk_id, fatal=False, headers=headers,
note=f"Preparing {src_id} chunk {i}"
)
m3u8_url = traverse_obj(base_format, (..., "url",), get_all=False)
playlist += self._download_webpage(m3u8_url, chunk_id, note=f"Getting {src_id} chunk {i} fragments")
return playlist
|
