instalive.py

importconcurrent.futures
importre
importthreading
importtime
importxml.etree.ElementTreeasET
fromdatetimeimportdatetime
frompathlibimportPath
fromshutilimportcopy2, copyfileobj, get_terminal_size
fromsubprocessimportrun
fromurllib.parseimporturljoin

fromtqdmimporttqdm
fromutilimportrequests_retry_session, get_webname, td_format


TOLERANCE=0.2


defparse_iso8601_duration(duration):
'''Parse ISO8601 duration to seconds.
 it sometimes uses float for seconds part, so we need to handle that.'''
pattern=r'P(?:(?P<days>\d+)D)?(?:T(?:(?P<hours>\d+)H)?(?:(?P<minutes>\d+)M)?(?:(?P<seconds>[.\d]+)S)?)?'
match=re.match(pattern, duration)
ifnotmatch:
returnNone

days=int(match.group('days') or0)
hours=int(match.group('hours') or0)
minutes=int(match.group('minutes') or0)
seconds=int(float(match.group('seconds') or0))

# Convert everything to seconds
total_seconds= (days*86400) + (hours*3600) + (minutes*60) +seconds
returntotal_seconds


defprint_full_width(s):
'''Print a string to full width of the terminal.'''
width=get_terminal_size().columns
print('\r'+s.ljust(width), end='')


defconcat(files, output, verbose=False):
'''Concatenate files into one file.'''
output=Path(output)
ifnotoutput.parent.exists():
output.parent.mkdir(parents=True)
out=output.open('wb')
forfintqdm(files, ncols=100):
f=Path(f)
ifverbose:
print(f'Merging {f.name}...')
fi=f.open('rb')
copyfileobj(fi, out)
fi.close()
out.close()


classInstaliveDownloader:
def__init__(self, url, save_path, debug=False, quality=None):
self.session=requests_retry_session()
self.url=url
ifsave_pathisNone:
mpd_name=get_webname(url).split('.')[0]
save_path=f'instalive_{mpd_name}'
self.save_path=Path(save_path)
ifnotself.save_path.exists():
self.save_path.mkdir(parents=True)
assertself.save_path.is_dir(), f'{self.save_path} is not a directory!'
self.debug=debug
ifdebug:
print('Debug mode enabled. Only download 30 segments.')
self.fetch_mpd()
self.parse_mpd(quality)

def_download(self, url, save_path=None, filename=None, skip_existing=True):
ifsave_path:
f=Path(save_path) /get_webname(url)
eliffilename:
f=Path(filename)
ifskip_existingandf.exists() andf.stat().st_size>0:
return'Exists'
f.parent.mkdir(parents=True, exist_ok=True)
withself.session.get(url) asr:
ifr.status_code==200:
withf.open('wb') asf:
f.write(r.content)
returnr.status_code

defsave_mpd(self):
print('Save mpd to local file...')
mpd_lines=self.mpd_text.splitlines(keepends=True)
comment=f'<!-- {self.url} -->\n'
ifmpd_linesandmpd_lines[0].strip().startswith('<?xml'):
mpd_lines.insert(1, comment)
else:
mpd_lines.insert(0, comment)
 (self.save_path/'mpd.mpd').write_text(''.join(mpd_lines), encoding='utf-8')

deffetch_mpd(self, mute=False):
ifnotmute:
print('Fetch mpd...')
self.mpd_text=self.session.get(self.url).text
self.mpd=ET.fromstring(self.mpd_text)
returnself.mpd

defparse_mpd(self, quality=None):
print('Parse mpd...')
mpd=self.mpd
# availabilityStartTime="2024-05-06T01:21:00-07:00"
# availabilityEndTime="2024-05-06T01:29:00-07:00"

deftd_format_with_direction(td):
'''a wrapper around td_format to show both positive and negative time deltas'''
iftd.total_seconds() <0:
returntd_format(-td) +" ago"
else:
return'in '+td_format(td)
current_time=datetime.now().astimezone()
availability_start_time=datetime.strptime(mpd.attrib['availabilityStartTime'], '%Y-%m-%dT%H:%M:%S%z')
availability_end_time=datetime.strptime(mpd.attrib['availabilityEndTime'], '%Y-%m-%dT%H:%M:%S%z')

print(f'Availability start time: {availability_start_time} ({td_format_with_direction(availability_start_time-current_time)})')
print(f'Availability end time: {availability_end_time} ({td_format_with_direction(availability_end_time-current_time)})')

# process video representations
period=mpd[0]
video_adaptation_set=period[0]
# make a simple list of video representations because we can't directly sort xml elements
videos= [
 {
'width': int(x.attrib['width']),
'height': int(x.attrib['height']),
'frame_rate': float(x.attrib['frameRate']),
'bandwidth': float(x.attrib['bandwidth']),
'idx': idx,
'id': x.attrib['id']
 } foridx, xinenumerate(video_adaptation_set)
 ]

print('Video representations:')
forvinvideos:
print(f'[{v["idx"]}] {v["id"]}{v["width"]}x{v["height"]}, {v["frame_rate"]}fps, {v["bandwidth"]/1024:.1f} kbps')

ifnotqualityorquality=='highest':
videos.sort(reverse=True, key=lambdax: (x['width'], x['height'], x['frame_rate'], x['bandwidth']))
self.video_index=videos[0]['idx']
self.video_id=videos[0]['id']
print(f'Use the highest one ([{self.video_index}] {self.video_id}).')
else:
forvinvideos:
# quality could be either "dash-lp-pst-v" or "pst"
ifquality==v['id'] orf'{str(quality)}-v'inv['id']:
self.video_index=v['idx']
self.video_id=v['id']
print(f'Use the specified one ([{self.video_index}] {self.video_id}).')
break
else:
raiseException(f'Quality {quality} not found.')

self.save_path_video=self.save_path/self.video_id

video_representation=video_adaptation_set[self.video_index]
video_segment_template=video_representation[0]
self.timescale=video_segment_template.attrib['timescale'] # not used for now
self.video_init=urljoin(self.url, video_segment_template.attrib['initialization'])
self.video_url_template=urljoin(self.url, video_segment_template.attrib['media']).replace('$Time$', '{}')
timeline=video_segment_template[0]
# find the last segment's t
self.last_t=int(timeline[-1].attrib['t'])
print(f'Last video segment t={self.last_t}')
# find the best interval for iterating heuristically
d_list= [int(x.attrib['d']) forxintimeline]
print('Intervals between segments:', ', '.join(str(d) fordind_list))
# get most common interval
self.interval=max(set(d_list), key=d_list.count)
print(f'Heuristically set interval to {self.interval}')

# process audio representations
audio_adaptation_set=period[1]
audios= [
 {
'id': x.attrib['id'],
'bandwidth': float(x.attrib['bandwidth']),
'sampling_rate': int(x.attrib['audioSamplingRate']),
'idx': idx
 } foridx, xinenumerate(audio_adaptation_set)
 ]
print('Audio representations:')
forainaudios:
print(f'[{a["idx"]}] {a["id"]}, {a["bandwidth"]/1024:.1f} kbps, {a["sampling_rate"]} Hz')

audios.sort(reverse=True, key=lambdax: (x['bandwidth'], x['sampling_rate']))
self.audio_index=audios[0]['idx']
self.audio_id=audios[0]['id']
print(f'Use the best one ([{self.audio_index}] {self.audio_id}).')
self.save_path_audio=self.save_path/self.audio_id

audio_representation=audio_adaptation_set[self.audio_index]
audio_segment_template=audio_representation[0]

self.audio_init=urljoin(self.url, audio_segment_template.attrib['initialization'])
self.audio_url_template=urljoin(self.url, audio_segment_template.attrib['media']).replace('$Time$', '{}')

defdownload_init(self):
print('Download init segments...')
r=self._download(self.video_init, save_path=self.save_path_video)
assertrin [200, 'Exists']
r=self._download(self.audio_init, save_path=self.save_path_audio)
assertrin [200, 'Exists']

def_get_segments(self):
mpd=self.fetch_mpd(mute=True) # fetch a new mpd to get the latest segments
video_representation=mpd[0][0][self.video_index]
assertvideo_representation.attrib['id'] ==self.video_id
timeline=video_representation[0][0]
segments= [int(timeline[i].attrib['t']) foriinrange(len(timeline))]
returnsegments, mpd

defmanually_set(self, last_t=None):
iflast_tisnotNone:
self.last_t=int(last_t)
print(f'Last time set to {self.last_t}')

deffetch_video_by_id(self, id):
url=self.video_url_template.format(id)
status=self._download(url, save_path=self.save_path_video)
returnstatus

deffetch_audio_by_id(self, id):
url=self.audio_url_template.format(id)
status=self._download(url, save_path=self.save_path_audio)
returnstatus

defquick_iterate(self, ids):
# make sure ids are larger than 0
ids= [idforidinidsifid>0]
print(f'\nUse multi-threading to check {len(ids)} IDs starting from {ids[0]}...')
withconcurrent.futures.ThreadPoolExecutor(max_workers=20) asex:
futures= {ex.submit(self.fetch_video_by_id, id): idforidinids}
try:
forfutureinconcurrent.futures.as_completed(futures):
iffuture.cancelled():
continue
id=futures[future]
status=future.result()
ifstatusin ['Exists', 200]:
# cancel all other futures
forfinfutures:
f.cancel()
returnid, status
# TODO: does not work since we call it in a thread.
exceptKeyboardInterrupt:
print('\nInterrupted. Cancel all futures...')
forfinfutures:
f.cancel()
raiseKeyboardInterrupt
returnNone, None

defdownload_live(self):
downloaded=set()

mpd=self.mpd
# check if mpd is dynamic
ifmpd.attrib.get('type') !='dynamic':
print('This mpd is not dynamic. Stop monitoring live.')
return
# use minimumUpdatePeriod if available. otherwise, use timeShiftBufferDepth/2-1 as the interval.
# make sure it is at least 2s.
fetch_interval=parse_iso8601_duration(mpd.attrib.get('minimumUpdatePeriod', 'PT0S')) \
orparse_iso8601_duration(mpd.attrib.get('timeShiftBufferDepth', 'PT0S')) //2-1
fetch_interval=max(fetch_interval, 2)

MAX_IDLE_COUNT=20

unchanged_mpd_count=0
whileTrue:
new_segments, mpd=self._get_segments()
ifall(idindownloadedforidinnew_segments):
unchanged_mpd_count+=1
ifunchanged_mpd_count>=MAX_IDLE_COUNT:
ifmpd.attrib.get('type') =='dynamic':
print(f'No new segments found in the last {MAX_IDLE_COUNT} checks. But the mpd is still dynamic. Continue monitoring...')
else:
print('All segments are downloaded. Stop.')
return
else:
undownloaded= [idforidinnew_segmentsifidnotindownloaded]
print(f'{len(undownloaded)} new segments found. Downloading...')
foridinundownloaded:
# singe-thread should be enough for live stream
self.fetch_video_by_id(id)
self.fetch_audio_by_id(id)
downloaded.add(id)

time.sleep(fetch_interval)

defdownload_video(self, forward=False):
count=0
known_intervals= {
self.interval: 0,
self.interval-1: 0,
self.interval+1: 0,
 }
id_guesses= [self.last_t]
prev_id=None
sign=1ifforwardelse-1

defsurrounding(x):
start_id=x-int(self.interval* (0.5+TOLERANCE)) *sign
end_id=x+int(self.interval* (1.0+TOLERANCE) +1) *sign
ids=list(range(start_id, end_id, -1ifstart_id>end_idelse1))
ids.sort(key=lambdak: abs(k-x))
returnids

whileTrue:
valid_id=None
# firstly, we try to find if existing local file that is close to the guesses[0].
# which is defined as +/- TOLERANCE (default: 20%) of the interval.
# this is a "narrower" window than surrounding() function.
# notice that we don't need to try all the guesses, they will be checked in the
# next step.
ids=list(range(id_guesses[0] -int(self.interval*TOLERANCE), id_guesses[0] +int(self.interval*TOLERANCE) +1))
ids.sort(key=lambdak: abs(k-id_guesses[0]))
# only try ids that are before the prev_id (if backward) or after the prev_id (if forward).
ifprev_id:
ids= [idforidinidsif (id-prev_id) *sign>0]
forcandidateinids:
url=self.video_url_template.format(candidate)
f=self.save_path_video/get_webname(url)
iff.exists() andf.stat().st_size>0:
valid_id=candidate
print_full_width(f'Segment {valid_id}: {f} already exists. Skip.')
break
# if not found locally, we try to fetch the segment by id from the all
# (last_valid_id - potential_interval) pools.
# The order matters: we try the most common interval first.
# this is single-threaded, because it is more time-consuming trying to close all the threads.
ifnotvalid_id:
forcandidateinid_guesses:
status=self.fetch_video_by_id(candidate)
print_full_width(f'Segment {candidate}: HTTP {status}')
ifstatusin [200, 'Exists']:
valid_id=candidate
break
# If still not found, we iterate around the guesses[0] to find the next segment.
# the range is defined as 50%+TOLERANCE behind of x and 100%+TOLERANCE ahead of x,
# where x is the guesses[0].
# e.g. for 2000 interval, x = last_id - 2000, the range would be
# last_id - 800 to last_id - 4400, sorted by distance to last_id - 2000.
# notice that it covers up to the range of next next id, this way we ensure we still
# continue the downloading instead of stopping too early (despite missing a segment).
ifnotvalid_id:
ids= [idforidinsurrounding(id_guesses[0]) ifidnotinid_guesses]
ifprev_id:
ids= [idforidinidsif (id-prev_id) *sign>0]
valid_id, status=self.quick_iterate(ids)
# if still not, we assume we downloaded them all and stop.
ifnotvalid_id:
print("\nFailed to find next segment. Assume we downloaded all. Stop.")
break
print_full_width(f'Segment {valid_id}: HTTP {status}')
# at this point, we should have a valid_id.
assertvalid_id
# add new interval to known_intervals
ifprev_id:
new_interval=abs(valid_id-prev_id)
assertnew_interval>0# this should not happen.
# do not add new interval if it is too different from the current interval.
# for example, if the nominal interval is 2000, we should only add ones that are
# less than 3000. Otherwise we enables the possibility of skipping segments.
# But it is still allowed to have such large interval so we don't stop downloading
# in the middle just because of one missing segment.
ifnew_interval>=self.interval*1.6:
pass
# add interval to known_intervals (if not already), and increase the count.
else:
known_intervals.setdefault(new_interval, 0)
known_intervals[new_interval] +=1

# print(f'[Debug] known_intervals: {known_intervals}')
count+=1
ifself.debugandcount==30:
break
prev_id=valid_id
# sort known_intervals by appearance, so we try the most common interval first.
known_intervals=dict(sorted(known_intervals.items(), key=lambdax: x[1], reverse=True))
id_guesses= [valid_id+interval*signforintervalinknown_intervals]

defcheck(self):
print('Check if there is any missing video segment...')
files=list(self.save_path_video.iterdir())
# filename format: 17981336783244063_0-1297029.m4v
ids= [int(m[1]) forfinfilesif (m:=re.search(r'\d+_0-(\d+)', f.name))]
ids.sort()
# calculate difference between each id
diffs= [ids[i+1] -ids[i] foriinrange(len(ids)-1)]

diff_count= {}
fordiffindiffs:
diff_count[diff] =diff_count.get(diff, 0) +1
print(f'Count of diff values between each segment:', diff_count)
max_idx=diffs.index(max(diffs))
print(f'Largest ID diff between each segment: {max(diffs)}, at {ids[max_idx]} -> {ids[max_idx+1]}')
min_idx=diffs.index(min(diffs))
print(f'Smallest ID diff between each segment: {min(diffs)}, at {ids[min_idx]} -> {ids[min_idx+1]}')

# Check if there are any missing segments, defined as gaps larger than 1.2 times the interval.
ifmax(diffs) >self.interval* (1+TOLERANCE):
print(f'It is likely that the video is not fully downloaded!!')
return (ids[max_idx], ids[max_idx+1])
else:
print('No missing segment found.')


defdownload_audio(self):
print('Downloading audio segments...')
files=list(self.save_path_video.iterdir())
withconcurrent.futures.ThreadPoolExecutor(max_workers=20) asex:
futures= []
count=0
forfinfiles:
ifnotf.is_file():
continue
iff.suffix!='.m4v':
continue
id=re.search(r'_0-(init|\d+)\.m4v', f.name)[1]
url=self.audio_url_template.format(id)
futures.append(ex.submit(self._download, url, save_path=self.save_path_audio))
for_inconcurrent.futures.as_completed(futures):
count+=1
print(f'Finished {count}/{len(futures)} ', end='\r')

defmerge(self):
defget_key(f):
'''Make sure init segment is always at the beginning.'''
if'-init'inf.name:
return0
returnint(re.search(r'\d+_0-(\d+)', f.name)[1])

video_file=self.save_path/'video.m4v'
files=list(self.save_path_video.iterdir())
print(f'Find {len(files)} video segments. Merging...')
files.sort(key=get_key)
concat(files, video_file)

audio_file=self.save_path/'audio.m4a'
files2=list(self.save_path_audio.iterdir())
print(f'Find {len(files2)} audio segments. Merging...')
files2.sort(key=get_key)
concat(files2, audio_file)

print(f'Merging video and audio using FFMPEG...')
run(['ffmpeg', '-loglevel', 'error', '-stats', '-i', video_file, '-i', audio_file, '-c', 'copy', self.save_path/'merged.mp4'])

defimport_segments(self, path):
'''Import segments downloaded via N_m3u8DL-RE.'''
path=Path(path)
forpinpath.iterdir():
ifp.is_dir() and (p/'_init.mp4').exists():
if'avc'inp.name:
template=self.video_url_template
save_path=self.save_path_video
else:
template=self.audio_url_template
save_path=self.save_path_audio
forfinp.iterdir():
iff.stem.isdigit():
id=int(f.stem)
new_filename=get_webname(template.format(id))
newf=save_path/new_filename
ifnewf.exists():
ifnewf.stat().st_size==f.stat().st_size:
print(f'{id}: {newf} already exists. Skip.')
else:
raiseException(f'{id}: {newf} already exists but size is different.')
else:
print(f'{id}: Copy {f} to {newf}')
copy2(f, newf)


defmain(url, save_path, time, debug, action, quality):
downloader=InstaliveDownloader(url=url, save_path=save_path, debug=debug, quality=quality)
iftimeisnotNone:
downloader.manually_set(time)
ifaction.startswith('import'):
import_path=action.partition(':')[2]
downloader.import_segments(import_path)
return
ifaction=='info':
fork, vindownloader.__dict__.items():
ifk=='mpd_text': # too long.
v=v[:10] +'...'+v[-10:]
print(f'{k}: {v}')
return
try:
ifaction=='all':
downloader.save_mpd()
downloader.download_init()
deftask1():
downloader.download_video()
downloader.download_audio()
t1=threading.Thread(target=task1)
t2=threading.Thread(target=downloader.download_live)
t1.start()
t2.start()
t1.join()
t2.join()
downloader.check()
downloader.merge()
elifaction=='live':
downloader.save_mpd()
downloader.download_live()
elifaction=='video':
downloader.save_mpd()
downloader.download_init()
downloader.download_video()
elifaction=='audio':
downloader.download_audio()
elifaction=='merge':
downloader.merge()
elifaction=='check':
downloader.check()
elifaction=='manual':
ifargs.range:
start, end=map(int, args.range.split('-'))
print(f'manually set range to from {start} to {end} (inclusive)')
else:
worst=downloader.check()
ifnotworst:
print('No bad interval found. Stop.')
return
start=worst[0] +1
end=worst[1] -1
print(f'Automatically check the largest interval, between {start} and {end} (inclusive)')
ids=list(range(start, end+1)) ifstart<endelselist(range(start, end-1, -1))
id, status=downloader.quick_iterate(ids)
ifnotid:
print('No segment found in this range.')
else:
print(f'Found segment {id} with status {status}')

exceptKeyboardInterrupt:
print('\nInterrupted by user. Stop.')


if__name__=='__main__':
importargparse

parser=argparse.ArgumentParser(
description='Available actions:\n'
' all - Download both video and audio (including live and backtracking), and then merge them (default)\n'
' live - Download the live stream only (no backtracking)\n'
' video - Download video only\n'
' audio - Download audio only\n'
' merge - Merge downloaded video and audio\n'
' check - Check the downloaded segments to make sure there are no missing segments\n'
' manual - Manually check missing segments at the largest gap (or use --range to assign a range) (for debugging only)\n'
' info - Display downloader object info\n'
' import:<path> - Import segments downloaded via N_m3u8DL-RE from a given path',
formatter_class=argparse.RawTextHelpFormatter
 )

parser.add_argument("url", help="url of mpd")
parser.add_argument("--action", '-a', default='all', help="action to perform (default: all)")
parser.add_argument("--dir", "-d", help="save path (default: instalive_{mpd_id})")
parser.add_argument("--debug", action='store_true', help="debug mode")
parser.add_argument("--quality", "-q", default='pst', help="manually assign video quality by quality name.\n"
"(default: \"pst\": which is the \"original\" (?) and has the best bitrate,\n"
"but not necessarily the highest resolution.\n"
"Pass empty string or \"highest\" to use the highest resolution one.)")
parser.add_argument("--time", "-t", help="for debugging only; manually assign last t (default: auto)")
parser.add_argument('--range', help='for debugging only; manually assign iteration range (start,end) for manual action')

args=parser.parse_args()

main(args.url, args.dir, args.time, args.debug, args.action, args.quality)
importconcurrent.futures
importre
importthreading
importtime
importxml.etree.ElementTreeasET
fromdatetimeimportdatetime
frompathlibimportPath
fromshutilimportcopy2, copyfileobj, get_terminal_size
fromsubprocessimportrun
fromurllib.parseimporturljoin

fromtqdmimporttqdm
fromutilimportrequests_retry_session, get_webname, td_format


TOLERANCE=0.2


defparse_iso8601_duration(duration):
'''Parse ISO8601 duration to seconds.
 it sometimes uses float for seconds part, so we need to handle that.'''
pattern=r'P(?:(?P<days>\d+)D)?(?:T(?:(?P<hours>\d+)H)?(?:(?P<minutes>\d+)M)?(?:(?P<seconds>[.\d]+)S)?)?'
match=re.match(pattern, duration)
ifnotmatch:
returnNone

days=int(match.group('days') or0)
hours=int(match.group('hours') or0)
minutes=int(match.group('minutes') or0)
seconds=int(float(match.group('seconds') or0))

# Convert everything to seconds
total_seconds= (days*86400) + (hours*3600) + (minutes*60) +seconds
returntotal_seconds


defprint_full_width(s):
'''Print a string to full width of the terminal.'''
width=get_terminal_size().columns
print('\r'+s.ljust(width), end='')


defconcat(files, output, verbose=False):
'''Concatenate files into one file.'''
output=Path(output)
ifnotoutput.parent.exists():
output.parent.mkdir(parents=True)
out=output.open('wb')
forfintqdm(files, ncols=100):
f=Path(f)
ifverbose:
print(f'Merging {f.name}...')
fi=f.open('rb')
copyfileobj(fi, out)
fi.close()
out.close()


classInstaliveDownloader:
def__init__(self, url, save_path, debug=False, quality=None):
self.session=requests_retry_session()
self.url=url
ifsave_pathisNone:
mpd_name=get_webname(url).split('.')[0]
save_path=f'instalive_{mpd_name}'
self.save_path=Path(save_path)
ifnotself.save_path.exists():
self.save_path.mkdir(parents=True)
assertself.save_path.is_dir(), f'{self.save_path} is not a directory!'
self.debug=debug
ifdebug:
print('Debug mode enabled. Only download 30 segments.')
self.fetch_mpd()
self.parse_mpd(quality)

def_download(self, url, save_path=None, filename=None, skip_existing=True):
ifsave_path:
f=Path(save_path) /get_webname(url)
eliffilename:
f=Path(filename)
ifskip_existingandf.exists() andf.stat().st_size>0:
return'Exists'
f.parent.mkdir(parents=True, exist_ok=True)
withself.session.get(url) asr:
ifr.status_code==200:
withf.open('wb') asf:
f.write(r.content)
returnr.status_code

defsave_mpd(self):
print('Save mpd to local file...')
mpd_lines=self.mpd_text.splitlines(keepends=True)
comment=f'<!-- {self.url} -->\n'
ifmpd_linesandmpd_lines[0].strip().startswith('<?xml'):
mpd_lines.insert(1, comment)
else:
mpd_lines.insert(0, comment)
 (self.save_path/'mpd.mpd').write_text(''.join(mpd_lines), encoding='utf-8')

deffetch_mpd(self, mute=False):
ifnotmute:
print('Fetch mpd...')
self.mpd_text=self.session.get(self.url).text
self.mpd=ET.fromstring(self.mpd_text)
returnself.mpd

defparse_mpd(self, quality=None):
print('Parse mpd...')
mpd=self.mpd
# availabilityStartTime="2024-05-06T01:21:00-07:00"
# availabilityEndTime="2024-05-06T01:29:00-07:00"

deftd_format_with_direction(td):
'''a wrapper around td_format to show both positive and negative time deltas'''
iftd.total_seconds() <0:
returntd_format(-td) +" ago"
else:
return'in '+td_format(td)
current_time=datetime.now().astimezone()
availability_start_time=datetime.strptime(mpd.attrib['availabilityStartTime'], '%Y-%m-%dT%H:%M:%S%z')
availability_end_time=datetime.strptime(mpd.attrib['availabilityEndTime'], '%Y-%m-%dT%H:%M:%S%z')

print(f'Availability start time: {availability_start_time} ({td_format_with_direction(availability_start_time-current_time)})')
print(f'Availability end time: {availability_end_time} ({td_format_with_direction(availability_end_time-current_time)})')

# process video representations
period=mpd[0]
video_adaptation_set=period[0]
# make a simple list of video representations because we can't directly sort xml elements
videos= [
 {
'width': int(x.attrib['width']),
'height': int(x.attrib['height']),
'frame_rate': float(x.attrib['frameRate']),
'bandwidth': float(x.attrib['bandwidth']),
'idx': idx,
'id': x.attrib['id']
 } foridx, xinenumerate(video_adaptation_set)
 ]

print('Video representations:')
forvinvideos:
print(f'[{v["idx"]}] {v["id"]}{v["width"]}x{v["height"]}, {v["frame_rate"]}fps, {v["bandwidth"]/1024:.1f} kbps')

ifnotqualityorquality=='highest':
videos.sort(reverse=True, key=lambdax: (x['width'], x['height'], x['frame_rate'], x['bandwidth']))
self.video_index=videos[0]['idx']
self.video_id=videos[0]['id']
print(f'Use the highest one ([{self.video_index}] {self.video_id}).')
else:
forvinvideos:
# quality could be either "dash-lp-pst-v" or "pst"
ifquality==v['id'] orf'{str(quality)}-v'inv['id']:
self.video_index=v['idx']
self.video_id=v['id']
print(f'Use the specified one ([{self.video_index}] {self.video_id}).')
break
else:
raiseException(f'Quality {quality} not found.')

self.save_path_video=self.save_path/self.video_id

video_representation=video_adaptation_set[self.video_index]
video_segment_template=video_representation[0]
self.timescale=video_segment_template.attrib['timescale'] # not used for now
self.video_init=urljoin(self.url, video_segment_template.attrib['initialization'])
self.video_url_template=urljoin(self.url, video_segment_template.attrib['media']).replace('$Time$', '{}')
timeline=video_segment_template[0]
# find the last segment's t
self.last_t=int(timeline[-1].attrib['t'])
print(f'Last video segment t={self.last_t}')
# find the best interval for iterating heuristically
d_list= [int(x.attrib['d']) forxintimeline]
print('Intervals between segments:', ', '.join(str(d) fordind_list))
# get most common interval
self.interval=max(set(d_list), key=d_list.count)
print(f'Heuristically set interval to {self.interval}')

# process audio representations
audio_adaptation_set=period[1]
audios= [
 {
'id': x.attrib['id'],
'bandwidth': float(x.attrib['bandwidth']),
'sampling_rate': int(x.attrib['audioSamplingRate']),
'idx': idx
 } foridx, xinenumerate(audio_adaptation_set)
 ]
print('Audio representations:')
forainaudios:
print(f'[{a["idx"]}] {a["id"]}, {a["bandwidth"]/1024:.1f} kbps, {a["sampling_rate"]} Hz')

audios.sort(reverse=True, key=lambdax: (x['bandwidth'], x['sampling_rate']))
self.audio_index=audios[0]['idx']
self.audio_id=audios[0]['id']
print(f'Use the best one ([{self.audio_index}] {self.audio_id}).')
self.save_path_audio=self.save_path/self.audio_id

audio_representation=audio_adaptation_set[self.audio_index]
audio_segment_template=audio_representation[0]

self.audio_init=urljoin(self.url, audio_segment_template.attrib['initialization'])
self.audio_url_template=urljoin(self.url, audio_segment_template.attrib['media']).replace('$Time$', '{}')

defdownload_init(self):
print('Download init segments...')
r=self._download(self.video_init, save_path=self.save_path_video)
assertrin [200, 'Exists']
r=self._download(self.audio_init, save_path=self.save_path_audio)
assertrin [200, 'Exists']

def_get_segments(self):
mpd=self.fetch_mpd(mute=True) # fetch a new mpd to get the latest segments
video_representation=mpd[0][0][self.video_index]
assertvideo_representation.attrib['id'] ==self.video_id
timeline=video_representation[0][0]
segments= [int(timeline[i].attrib['t']) foriinrange(len(timeline))]
returnsegments, mpd

defmanually_set(self, last_t=None):
iflast_tisnotNone:
self.last_t=int(last_t)
print(f'Last time set to {self.last_t}')

deffetch_video_by_id(self, id):
url=self.video_url_template.format(id)
status=self._download(url, save_path=self.save_path_video)
returnstatus

deffetch_audio_by_id(self, id):
url=self.audio_url_template.format(id)
status=self._download(url, save_path=self.save_path_audio)
returnstatus

defquick_iterate(self, ids):
# make sure ids are larger than 0
ids= [idforidinidsifid>0]
print(f'\nUse multi-threading to check {len(ids)} IDs starting from {ids[0]}...')
withconcurrent.futures.ThreadPoolExecutor(max_workers=20) asex:
futures= {ex.submit(self.fetch_video_by_id, id): idforidinids}
try:
forfutureinconcurrent.futures.as_completed(futures):
iffuture.cancelled():
continue
id=futures[future]
status=future.result()
ifstatusin ['Exists', 200]:
# cancel all other futures
forfinfutures:
f.cancel()
returnid, status
# TODO: does not work since we call it in a thread.
exceptKeyboardInterrupt:
print('\nInterrupted. Cancel all futures...')
forfinfutures:
f.cancel()
raiseKeyboardInterrupt
returnNone, None

defdownload_live(self):
downloaded=set()

mpd=self.mpd
# check if mpd is dynamic
ifmpd.attrib.get('type') !='dynamic':
print('This mpd is not dynamic. Stop monitoring live.')
return
# use minimumUpdatePeriod if available. otherwise, use timeShiftBufferDepth/2-1 as the interval.
# make sure it is at least 2s.
fetch_interval=parse_iso8601_duration(mpd.attrib.get('minimumUpdatePeriod', 'PT0S')) \
orparse_iso8601_duration(mpd.attrib.get('timeShiftBufferDepth', 'PT0S')) //2-1
fetch_interval=max(fetch_interval, 2)

MAX_IDLE_COUNT=20

unchanged_mpd_count=0
whileTrue:
new_segments, mpd=self._get_segments()
ifall(idindownloadedforidinnew_segments):
unchanged_mpd_count+=1
ifunchanged_mpd_count>=MAX_IDLE_COUNT:
ifmpd.attrib.get('type') =='dynamic':
print(f'No new segments found in the last {MAX_IDLE_COUNT} checks. But the mpd is still dynamic. Continue monitoring...')
else:
print('All segments are downloaded. Stop.')
return
else:
undownloaded= [idforidinnew_segmentsifidnotindownloaded]
print(f'{len(undownloaded)} new segments found. Downloading...')
foridinundownloaded:
# singe-thread should be enough for live stream
self.fetch_video_by_id(id)
self.fetch_audio_by_id(id)
downloaded.add(id)

time.sleep(fetch_interval)

defdownload_video(self, forward=False):
count=0
known_intervals= {
self.interval: 0,
self.interval-1: 0,
self.interval+1: 0,
 }
id_guesses= [self.last_t]
prev_id=None
sign=1ifforwardelse-1

defsurrounding(x):
start_id=x-int(self.interval* (0.5+TOLERANCE)) *sign
end_id=x+int(self.interval* (1.0+TOLERANCE) +1) *sign
ids=list(range(start_id, end_id, -1ifstart_id>end_idelse1))
ids.sort(key=lambdak: abs(k-x))
returnids

whileTrue:
valid_id=None
# firstly, we try to find if existing local file that is close to the guesses[0].
# which is defined as +/- TOLERANCE (default: 20%) of the interval.
# this is a "narrower" window than surrounding() function.
# notice that we don't need to try all the guesses, they will be checked in the
# next step.
ids=list(range(id_guesses[0] -int(self.interval*TOLERANCE), id_guesses[0] +int(self.interval*TOLERANCE) +1))
ids.sort(key=lambdak: abs(k-id_guesses[0]))
# only try ids that are before the prev_id (if backward) or after the prev_id (if forward).
ifprev_id:
ids= [idforidinidsif (id-prev_id) *sign>0]
forcandidateinids:
url=self.video_url_template.format(candidate)
f=self.save_path_video/get_webname(url)
iff.exists() andf.stat().st_size>0:
valid_id=candidate
print_full_width(f'Segment {valid_id}: {f} already exists. Skip.')
break
# if not found locally, we try to fetch the segment by id from the all
# (last_valid_id - potential_interval) pools.
# The order matters: we try the most common interval first.
# this is single-threaded, because it is more time-consuming trying to close all the threads.
ifnotvalid_id:
forcandidateinid_guesses:
status=self.fetch_video_by_id(candidate)
print_full_width(f'Segment {candidate}: HTTP {status}')
ifstatusin [200, 'Exists']:
valid_id=candidate
break
# If still not found, we iterate around the guesses[0] to find the next segment.
# the range is defined as 50%+TOLERANCE behind of x and 100%+TOLERANCE ahead of x,
# where x is the guesses[0].
# e.g. for 2000 interval, x = last_id - 2000, the range would be
# last_id - 800 to last_id - 4400, sorted by distance to last_id - 2000.
# notice that it covers up to the range of next next id, this way we ensure we still
# continue the downloading instead of stopping too early (despite missing a segment).
ifnotvalid_id:
ids= [idforidinsurrounding(id_guesses[0]) ifidnotinid_guesses]
ifprev_id:
ids= [idforidinidsif (id-prev_id) *sign>0]
valid_id, status=self.quick_iterate(ids)
# if still not, we assume we downloaded them all and stop.
ifnotvalid_id:
print("\nFailed to find next segment. Assume we downloaded all. Stop.")
break
print_full_width(f'Segment {valid_id}: HTTP {status}')
# at this point, we should have a valid_id.
assertvalid_id
# add new interval to known_intervals
ifprev_id:
new_interval=abs(valid_id-prev_id)
assertnew_interval>0# this should not happen.
# do not add new interval if it is too different from the current interval.
# for example, if the nominal interval is 2000, we should only add ones that are
# less than 3000. Otherwise we enables the possibility of skipping segments.
# But it is still allowed to have such large interval so we don't stop downloading
# in the middle just because of one missing segment.
ifnew_interval>=self.interval*1.6:
pass
# add interval to known_intervals (if not already), and increase the count.
else:
known_intervals.setdefault(new_interval, 0)
known_intervals[new_interval] +=1

# print(f'[Debug] known_intervals: {known_intervals}')
count+=1
ifself.debugandcount==30:
break
prev_id=valid_id
# sort known_intervals by appearance, so we try the most common interval first.
known_intervals=dict(sorted(known_intervals.items(), key=lambdax: x[1], reverse=True))
id_guesses= [valid_id+interval*signforintervalinknown_intervals]

defcheck(self):
print('Check if there is any missing video segment...')
files=list(self.save_path_video.iterdir())
# filename format: 17981336783244063_0-1297029.m4v
ids= [int(m[1]) forfinfilesif (m:=re.search(r'\d+_0-(\d+)', f.name))]
ids.sort()
# calculate difference between each id
diffs= [ids[i+1] -ids[i] foriinrange(len(ids)-1)]

diff_count= {}
fordiffindiffs:
diff_count[diff] =diff_count.get(diff, 0) +1
print(f'Count of diff values between each segment:', diff_count)
max_idx=diffs.index(max(diffs))
print(f'Largest ID diff between each segment: {max(diffs)}, at {ids[max_idx]} -> {ids[max_idx+1]}')
min_idx=diffs.index(min(diffs))
print(f'Smallest ID diff between each segment: {min(diffs)}, at {ids[min_idx]} -> {ids[min_idx+1]}')

# Check if there are any missing segments, defined as gaps larger than 1.2 times the interval.
ifmax(diffs) >self.interval* (1+TOLERANCE):
print(f'It is likely that the video is not fully downloaded!!')
return (ids[max_idx], ids[max_idx+1])
else:
print('No missing segment found.')


defdownload_audio(self):
print('Downloading audio segments...')
files=list(self.save_path_video.iterdir())
withconcurrent.futures.ThreadPoolExecutor(max_workers=20) asex:
futures= []
count=0
forfinfiles:
ifnotf.is_file():
continue
iff.suffix!='.m4v':
continue
id=re.search(r'_0-(init|\d+)\.m4v', f.name)[1]
url=self.audio_url_template.format(id)
futures.append(ex.submit(self._download, url, save_path=self.save_path_audio))
for_inconcurrent.futures.as_completed(futures):
count+=1
print(f'Finished {count}/{len(futures)} ', end='\r')

defmerge(self):
defget_key(f):
'''Make sure init segment is always at the beginning.'''
if'-init'inf.name:
return0
returnint(re.search(r'\d+_0-(\d+)', f.name)[1])

video_file=self.save_path/'video.m4v'
files=list(self.save_path_video.iterdir())
print(f'Find {len(files)} video segments. Merging...')
files.sort(key=get_key)
concat(files, video_file)

audio_file=self.save_path/'audio.m4a'
files2=list(self.save_path_audio.iterdir())
print(f'Find {len(files2)} audio segments. Merging...')
files2.sort(key=get_key)
concat(files2, audio_file)

print(f'Merging video and audio using FFMPEG...')
run(['ffmpeg', '-loglevel', 'error', '-stats', '-i', video_file, '-i', audio_file, '-c', 'copy', self.save_path/'merged.mp4'])

defimport_segments(self, path):
'''Import segments downloaded via N_m3u8DL-RE.'''
path=Path(path)
forpinpath.iterdir():
ifp.is_dir() and (p/'_init.mp4').exists():
if'avc'inp.name:
template=self.video_url_template
save_path=self.save_path_video
else:
template=self.audio_url_template
save_path=self.save_path_audio
forfinp.iterdir():
iff.stem.isdigit():
id=int(f.stem)
new_filename=get_webname(template.format(id))
newf=save_path/new_filename
ifnewf.exists():
ifnewf.stat().st_size==f.stat().st_size:
print(f'{id}: {newf} already exists. Skip.')
else:
raiseException(f'{id}: {newf} already exists but size is different.')
else:
print(f'{id}: Copy {f} to {newf}')
copy2(f, newf)


defmain(url, save_path, time, debug, action, quality):
downloader=InstaliveDownloader(url=url, save_path=save_path, debug=debug, quality=quality)
iftimeisnotNone:
downloader.manually_set(time)
ifaction.startswith('import'):
import_path=action.partition(':')[2]
downloader.import_segments(import_path)
return
ifaction=='info':
fork, vindownloader.__dict__.items():
ifk=='mpd_text': # too long.
v=v[:10] +'...'+v[-10:]
print(f'{k}: {v}')
return
try:
ifaction=='all':
downloader.save_mpd()
downloader.download_init()
deftask1():
downloader.download_video()
downloader.download_audio()
t1=threading.Thread(target=task1)
t2=threading.Thread(target=downloader.download_live)
t1.start()
t2.start()
t1.join()
t2.join()
downloader.check()
downloader.merge()
elifaction=='live':
downloader.save_mpd()
downloader.download_live()
elifaction=='video':
downloader.save_mpd()
downloader.download_init()
downloader.download_video()
elifaction=='audio':
downloader.download_audio()
elifaction=='merge':
downloader.merge()
elifaction=='check':
downloader.check()
elifaction=='manual':
ifargs.range:
start, end=map(int, args.range.split('-'))
print(f'manually set range to from {start} to {end} (inclusive)')
else:
worst=downloader.check()
ifnotworst:
print('No bad interval found. Stop.')
return
start=worst[0] +1
end=worst[1] -1
print(f'Automatically check the largest interval, between {start} and {end} (inclusive)')
ids=list(range(start, end+1)) ifstart<endelselist(range(start, end-1, -1))
id, status=downloader.quick_iterate(ids)
ifnotid:
print('No segment found in this range.')
else:
print(f'Found segment {id} with status {status}')

exceptKeyboardInterrupt:
print('\nInterrupted by user. Stop.')


if__name__=='__main__':
importargparse

parser=argparse.ArgumentParser(
description='Available actions:\n'
' all - Download both video and audio (including live and backtracking), and then merge them (default)\n'
' live - Download the live stream only (no backtracking)\n'
' video - Download video only\n'
' audio - Download audio only\n'
' merge - Merge downloaded video and audio\n'
' check - Check the downloaded segments to make sure there are no missing segments\n'
' manual - Manually check missing segments at the largest gap (or use --range to assign a range) (for debugging only)\n'
' info - Display downloader object info\n'
' import:<path> - Import segments downloaded via N_m3u8DL-RE from a given path',
formatter_class=argparse.RawTextHelpFormatter
 )

parser.add_argument("url", help="url of mpd")
parser.add_argument("--action", '-a', default='all', help="action to perform (default: all)")
parser.add_argument("--dir", "-d", help="save path (default: instalive_{mpd_id})")
parser.add_argument("--debug", action='store_true', help="debug mode")
parser.add_argument("--quality", "-q", default='pst', help="manually assign video quality by quality name.\n"
"(default: \"pst\": which is the \"original\" (?) and has the best bitrate,\n"
"but not necessarily the highest resolution.\n"
"Pass empty string or \"highest\" to use the highest resolution one.)")
parser.add_argument("--time", "-t", help="for debugging only; manually assign last t (default: auto)")
parser.add_argument('--range', help='for debugging only; manually assign iteration range (start,end) for manual action')

args=parser.parse_args()

main(args.url, args.dir, args.time, args.debug, args.action, args.quality)
importconcurrent.futures
importre
importthreading
importtime
importxml.etree.ElementTreeasET
fromdatetimeimportdatetime
frompathlibimportPath
fromshutilimportcopy2, copyfileobj, get_terminal_size
fromsubprocessimportrun
fromurllib.parseimporturljoin

fromtqdmimporttqdm
fromutilimportrequests_retry_session, get_webname, td_format


TOLERANCE=0.2


defparse_iso8601_duration(duration):
'''Parse ISO8601 duration to seconds.
 it sometimes uses float for seconds part, so we need to handle that.'''
pattern=r'P(?:(?P<days>\d+)D)?(?:T(?:(?P<hours>\d+)H)?(?:(?P<minutes>\d+)M)?(?:(?P<seconds>[.\d]+)S)?)?'
match=re.match(pattern, duration)
ifnotmatch:
returnNone

days=int(match.group('days') or0)
hours=int(match.group('hours') or0)
minutes=int(match.group('minutes') or0)
seconds=int(float(match.group('seconds') or0))

# Convert everything to seconds
total_seconds= (days*86400) + (hours*3600) + (minutes*60) +seconds
returntotal_seconds


defprint_full_width(s):
'''Print a string to full width of the terminal.'''
width=get_terminal_size().columns
print('\r'+s.ljust(width), end='')


defconcat(files, output, verbose=False):
'''Concatenate files into one file.'''
output=Path(output)
ifnotoutput.parent.exists():
output.parent.mkdir(parents=True)
out=output.open('wb')
forfintqdm(files, ncols=100):
f=Path(f)
ifverbose:
print(f'Merging {f.name}...')
fi=f.open('rb')
copyfileobj(fi, out)
fi.close()
out.close()


classInstaliveDownloader:
def__init__(self, url, save_path, debug=False, quality=None):
self.session=requests_retry_session()
self.url=url
ifsave_pathisNone:
mpd_name=get_webname(url).split('.')[0]
save_path=f'instalive_{mpd_name}'
self.save_path=Path(save_path)
ifnotself.save_path.exists():
self.save_path.mkdir(parents=True)
assertself.save_path.is_dir(), f'{self.save_path} is not a directory!'
self.debug=debug
ifdebug:
print('Debug mode enabled. Only download 30 segments.')
self.fetch_mpd()
self.parse_mpd(quality)

def_download(self, url, save_path=None, filename=None, skip_existing=True):
ifsave_path:
f=Path(save_path) /get_webname(url)
eliffilename:
f=Path(filename)
ifskip_existingandf.exists() andf.stat().st_size>0:
return'Exists'
f.parent.mkdir(parents=True, exist_ok=True)
withself.session.get(url) asr:
ifr.status_code==200:
withf.open('wb') asf:
f.write(r.content)
returnr.status_code

defsave_mpd(self):
print('Save mpd to local file...')
mpd_lines=self.mpd_text.splitlines(keepends=True)
comment=f'<!-- {self.url} -->\n'
ifmpd_linesandmpd_lines[0].strip().startswith('<?xml'):
mpd_lines.insert(1, comment)
else:
mpd_lines.insert(0, comment)
 (self.save_path/'mpd.mpd').write_text(''.join(mpd_lines), encoding='utf-8')

deffetch_mpd(self, mute=False):
ifnotmute:
print('Fetch mpd...')
self.mpd_text=self.session.get(self.url).text
self.mpd=ET.fromstring(self.mpd_text)
returnself.mpd

defparse_mpd(self, quality=None):
print('Parse mpd...')
mpd=self.mpd
# availabilityStartTime="2024-05-06T01:21:00-07:00"
# availabilityEndTime="2024-05-06T01:29:00-07:00"

deftd_format_with_direction(td):
'''a wrapper around td_format to show both positive and negative time deltas'''
iftd.total_seconds() <0:
returntd_format(-td) +" ago"
else:
return'in '+td_format(td)
current_time=datetime.now().astimezone()
availability_start_time=datetime.strptime(mpd.attrib['availabilityStartTime'], '%Y-%m-%dT%H:%M:%S%z')
availability_end_time=datetime.strptime(mpd.attrib['availabilityEndTime'], '%Y-%m-%dT%H:%M:%S%z')

print(f'Availability start time: {availability_start_time} ({td_format_with_direction(availability_start_time-current_time)})')
print(f'Availability end time: {availability_end_time} ({td_format_with_direction(availability_end_time-current_time)})')

# process video representations
period=mpd[0]
video_adaptation_set=period[0]
# make a simple list of video representations because we can't directly sort xml elements
videos= [
 {
'width': int(x.attrib['width']),
'height': int(x.attrib['height']),
'frame_rate': float(x.attrib['frameRate']),
'bandwidth': float(x.attrib['bandwidth']),
'idx': idx,
'id': x.attrib['id']
 } foridx, xinenumerate(video_adaptation_set)
 ]

print('Video representations:')
forvinvideos:
print(f'[{v["idx"]}] {v["id"]}{v["width"]}x{v["height"]}, {v["frame_rate"]}fps, {v["bandwidth"]/1024:.1f} kbps')

ifnotqualityorquality=='highest':
videos.sort(reverse=True, key=lambdax: (x['width'], x['height'], x['frame_rate'], x['bandwidth']))
self.video_index=videos[0]['idx']
self.video_id=videos[0]['id']
print(f'Use the highest one ([{self.video_index}] {self.video_id}).')
else:
forvinvideos:
# quality could be either "dash-lp-pst-v" or "pst"
ifquality==v['id'] orf'{str(quality)}-v'inv['id']:
self.video_index=v['idx']
self.video_id=v['id']
print(f'Use the specified one ([{self.video_index}] {self.video_id}).')
break
else:
raiseException(f'Quality {quality} not found.')

self.save_path_video=self.save_path/self.video_id

video_representation=video_adaptation_set[self.video_index]
video_segment_template=video_representation[0]
self.timescale=video_segment_template.attrib['timescale'] # not used for now
self.video_init=urljoin(self.url, video_segment_template.attrib['initialization'])
self.video_url_template=urljoin(self.url, video_segment_template.attrib['media']).replace('$Time$', '{}')
timeline=video_segment_template[0]
# find the last segment's t
self.last_t=int(timeline[-1].attrib['t'])
print(f'Last video segment t={self.last_t}')
# find the best interval for iterating heuristically
d_list= [int(x.attrib['d']) forxintimeline]
print('Intervals between segments:', ', '.join(str(d) fordind_list))
# get most common interval
self.interval=max(set(d_list), key=d_list.count)
print(f'Heuristically set interval to {self.interval}')

# process audio representations
audio_adaptation_set=period[1]
audios= [
 {
'id': x.attrib['id'],
'bandwidth': float(x.attrib['bandwidth']),
'sampling_rate': int(x.attrib['audioSamplingRate']),
'idx': idx
 } foridx, xinenumerate(audio_adaptation_set)
 ]
print('Audio representations:')
forainaudios:
print(f'[{a["idx"]}] {a["id"]}, {a["bandwidth"]/1024:.1f} kbps, {a["sampling_rate"]} Hz')

audios.sort(reverse=True, key=lambdax: (x['bandwidth'], x['sampling_rate']))
self.audio_index=audios[0]['idx']
self.audio_id=audios[0]['id']
print(f'Use the best one ([{self.audio_index}] {self.audio_id}).')
self.save_path_audio=self.save_path/self.audio_id

audio_representation=audio_adaptation_set[self.audio_index]
audio_segment_template=audio_representation[0]

self.audio_init=urljoin(self.url, audio_segment_template.attrib['initialization'])
self.audio_url_template=urljoin(self.url, audio_segment_template.attrib['media']).replace('$Time$', '{}')

defdownload_init(self):
print('Download init segments...')
r=self._download(self.video_init, save_path=self.save_path_video)
assertrin [200, 'Exists']
r=self._download(self.audio_init, save_path=self.save_path_audio)
assertrin [200, 'Exists']

def_get_segments(self):
mpd=self.fetch_mpd(mute=True) # fetch a new mpd to get the latest segments
video_representation=mpd[0][0][self.video_index]
assertvideo_representation.attrib['id'] ==self.video_id
timeline=video_representation[0][0]
segments= [int(timeline[i].attrib['t']) foriinrange(len(timeline))]
returnsegments, mpd

defmanually_set(self, last_t=None):
iflast_tisnotNone:
self.last_t=int(last_t)
print(f'Last time set to {self.last_t}')

deffetch_video_by_id(self, id):
url=self.video_url_template.format(id)
status=self._download(url, save_path=self.save_path_video)
returnstatus

deffetch_audio_by_id(self, id):
url=self.audio_url_template.format(id)
status=self._download(url, save_path=self.save_path_audio)
returnstatus

defquick_iterate(self, ids):
# make sure ids are larger than 0
ids= [idforidinidsifid>0]
print(f'\nUse multi-threading to check {len(ids)} IDs starting from {ids[0]}...')
withconcurrent.futures.ThreadPoolExecutor(max_workers=20) asex:
futures= {ex.submit(self.fetch_video_by_id, id): idforidinids}
try:
forfutureinconcurrent.futures.as_completed(futures):
iffuture.cancelled():
continue
id=futures[future]
status=future.result()
ifstatusin ['Exists', 200]:
# cancel all other futures
forfinfutures:
f.cancel()
returnid, status
# TODO: does not work since we call it in a thread.
exceptKeyboardInterrupt:
print('\nInterrupted. Cancel all futures...')
forfinfutures:
f.cancel()
raiseKeyboardInterrupt
returnNone, None

defdownload_live(self):
downloaded=set()

mpd=self.mpd
# check if mpd is dynamic
ifmpd.attrib.get('type') !='dynamic':
print('This mpd is not dynamic. Stop monitoring live.')
return
# use minimumUpdatePeriod if available. otherwise, use timeShiftBufferDepth/2-1 as the interval.
# make sure it is at least 2s.
fetch_interval=parse_iso8601_duration(mpd.attrib.get('minimumUpdatePeriod', 'PT0S')) \
orparse_iso8601_duration(mpd.attrib.get('timeShiftBufferDepth', 'PT0S')) //2-1
fetch_interval=max(fetch_interval, 2)

MAX_IDLE_COUNT=20

unchanged_mpd_count=0
whileTrue:
new_segments, mpd=self._get_segments()
ifall(idindownloadedforidinnew_segments):
unchanged_mpd_count+=1
ifunchanged_mpd_count>=MAX_IDLE_COUNT:
ifmpd.attrib.get('type') =='dynamic':
print(f'No new segments found in the last {MAX_IDLE_COUNT} checks. But the mpd is still dynamic. Continue monitoring...')
else:
print('All segments are downloaded. Stop.')
return
else:
undownloaded= [idforidinnew_segmentsifidnotindownloaded]
print(f'{len(undownloaded)} new segments found. Downloading...')
foridinundownloaded:
# singe-thread should be enough for live stream
self.fetch_video_by_id(id)
self.fetch_audio_by_id(id)
downloaded.add(id)

time.sleep(fetch_interval)

defdownload_video(self, forward=False):
count=0
known_intervals= {
self.interval: 0,
self.interval-1: 0,
self.interval+1: 0,
 }
id_guesses= [self.last_t]
prev_id=None
sign=1ifforwardelse-1

defsurrounding(x):
start_id=x-int(self.interval* (0.5+TOLERANCE)) *sign
end_id=x+int(self.interval* (1.0+TOLERANCE) +1) *sign
ids=list(range(start_id, end_id, -1ifstart_id>end_idelse1))
ids.sort(key=lambdak: abs(k-x))
returnids

whileTrue:
valid_id=None
# firstly, we try to find if existing local file that is close to the guesses[0].
# which is defined as +/- TOLERANCE (default: 20%) of the interval.
# this is a "narrower" window than surrounding() function.
# notice that we don't need to try all the guesses, they will be checked in the
# next step.
ids=list(range(id_guesses[0] -int(self.interval*TOLERANCE), id_guesses[0] +int(self.interval*TOLERANCE) +1))
ids.sort(key=lambdak: abs(k-id_guesses[0]))
# only try ids that are before the prev_id (if backward) or after the prev_id (if forward).
ifprev_id:
ids= [idforidinidsif (id-prev_id) *sign>0]
forcandidateinids:
url=self.video_url_template.format(candidate)
f=self.save_path_video/get_webname(url)
iff.exists() andf.stat().st_size>0:
valid_id=candidate
print_full_width(f'Segment {valid_id}: {f} already exists. Skip.')
break
# if not found locally, we try to fetch the segment by id from the all
# (last_valid_id - potential_interval) pools.
# The order matters: we try the most common interval first.
# this is single-threaded, because it is more time-consuming trying to close all the threads.
ifnotvalid_id:
forcandidateinid_guesses:
status=self.fetch_video_by_id(candidate)
print_full_width(f'Segment {candidate}: HTTP {status}')
ifstatusin [200, 'Exists']:
valid_id=candidate
break
# If still not found, we iterate around the guesses[0] to find the next segment.
# the range is defined as 50%+TOLERANCE behind of x and 100%+TOLERANCE ahead of x,
# where x is the guesses[0].
# e.g. for 2000 interval, x = last_id - 2000, the range would be
# last_id - 800 to last_id - 4400, sorted by distance to last_id - 2000.
# notice that it covers up to the range of next next id, this way we ensure we still
# continue the downloading instead of stopping too early (despite missing a segment).
ifnotvalid_id:
ids= [idforidinsurrounding(id_guesses[0]) ifidnotinid_guesses]
ifprev_id:
ids= [idforidinidsif (id-prev_id) *sign>0]
valid_id, status=self.quick_iterate(ids)
# if still not, we assume we downloaded them all and stop.
ifnotvalid_id:
print("\nFailed to find next segment. Assume we downloaded all. Stop.")
break
print_full_width(f'Segment {valid_id}: HTTP {status}')
# at this point, we should have a valid_id.
assertvalid_id
# add new interval to known_intervals
ifprev_id:
new_interval=abs(valid_id-prev_id)
assertnew_interval>0# this should not happen.
# do not add new interval if it is too different from the current interval.
# for example, if the nominal interval is 2000, we should only add ones that are
# less than 3000. Otherwise we enables the possibility of skipping segments.
# But it is still allowed to have such large interval so we don't stop downloading
# in the middle just because of one missing segment.
ifnew_interval>=self.interval*1.6:
pass
# add interval to known_intervals (if not already), and increase the count.
else:
known_intervals.setdefault(new_interval, 0)
known_intervals[new_interval] +=1

# print(f'[Debug] known_intervals: {known_intervals}')
count+=1
ifself.debugandcount==30:
break
prev_id=valid_id
# sort known_intervals by appearance, so we try the most common interval first.
known_intervals=dict(sorted(known_intervals.items(), key=lambdax: x[1], reverse=True))
id_guesses= [valid_id+interval*signforintervalinknown_intervals]

defcheck(self):
print('Check if there is any missing video segment...')
files=list(self.save_path_video.iterdir())
# filename format: 17981336783244063_0-1297029.m4v
ids= [int(m[1]) forfinfilesif (m:=re.search(r'\d+_0-(\d+)', f.name))]
ids.sort()
# calculate difference between each id
diffs= [ids[i+1] -ids[i] foriinrange(len(ids)-1)]

diff_count= {}
fordiffindiffs:
diff_count[diff] =diff_count.get(diff, 0) +1
print(f'Count of diff values between each segment:', diff_count)
max_idx=diffs.index(max(diffs))
print(f'Largest ID diff between each segment: {max(diffs)}, at {ids[max_idx]} -> {ids[max_idx+1]}')
min_idx=diffs.index(min(diffs))
print(f'Smallest ID diff between each segment: {min(diffs)}, at {ids[min_idx]} -> {ids[min_idx+1]}')

# Check if there are any missing segments, defined as gaps larger than 1.2 times the interval.
ifmax(diffs) >self.interval* (1+TOLERANCE):
print(f'It is likely that the video is not fully downloaded!!')
return (ids[max_idx], ids[max_idx+1])
else:
print('No missing segment found.')


defdownload_audio(self):
print('Downloading audio segments...')
files=list(self.save_path_video.iterdir())
withconcurrent.futures.ThreadPoolExecutor(max_workers=20) asex:
futures= []
count=0
forfinfiles:
ifnotf.is_file():
continue
iff.suffix!='.m4v':
continue
id=re.search(r'_0-(init|\d+)\.m4v', f.name)[1]
url=self.audio_url_template.format(id)
futures.append(ex.submit(self._download, url, save_path=self.save_path_audio))
for_inconcurrent.futures.as_completed(futures):
count+=1
print(f'Finished {count}/{len(futures)} ', end='\r')

defmerge(self):
defget_key(f):
'''Make sure init segment is always at the beginning.'''
if'-init'inf.name:
return0
returnint(re.search(r'\d+_0-(\d+)', f.name)[1])

video_file=self.save_path/'video.m4v'
files=list(self.save_path_video.iterdir())
print(f'Find {len(files)} video segments. Merging...')
files.sort(key=get_key)
concat(files, video_file)

audio_file=self.save_path/'audio.m4a'
files2=list(self.save_path_audio.iterdir())
print(f'Find {len(files2)} audio segments. Merging...')
files2.sort(key=get_key)
concat(files2, audio_file)

print(f'Merging video and audio using FFMPEG...')
run(['ffmpeg', '-loglevel', 'error', '-stats', '-i', video_file, '-i', audio_file, '-c', 'copy', self.save_path/'merged.mp4'])

defimport_segments(self, path):
'''Import segments downloaded via N_m3u8DL-RE.'''
path=Path(path)
forpinpath.iterdir():
ifp.is_dir() and (p/'_init.mp4').exists():
if'avc'inp.name:
template=self.video_url_template
save_path=self.save_path_video
else:
template=self.audio_url_template
save_path=self.save_path_audio
forfinp.iterdir():
iff.stem.isdigit():
id=int(f.stem)
new_filename=get_webname(template.format(id))
newf=save_path/new_filename
ifnewf.exists():
ifnewf.stat().st_size==f.stat().st_size:
print(f'{id}: {newf} already exists. Skip.')
else:
raiseException(f'{id}: {newf} already exists but size is different.')
else:
print(f'{id}: Copy {f} to {newf}')
copy2(f, newf)


defmain(url, save_path, time, debug, action, quality):
downloader=InstaliveDownloader(url=url, save_path=save_path, debug=debug, quality=quality)
iftimeisnotNone:
downloader.manually_set(time)
ifaction.startswith('import'):
import_path=action.partition(':')[2]
downloader.import_segments(import_path)
return
ifaction=='info':
fork, vindownloader.__dict__.items():
ifk=='mpd_text': # too long.
v=v[:10] +'...'+v[-10:]
print(f'{k}: {v}')
return
try:
ifaction=='all':
downloader.save_mpd()
downloader.download_init()
deftask1():
downloader.download_video()
downloader.download_audio()
t1=threading.Thread(target=task1)
t2=threading.Thread(target=downloader.download_live)
t1.start()
t2.start()
t1.join()
t2.join()
downloader.check()
downloader.merge()
elifaction=='live':
downloader.save_mpd()
downloader.download_live()
elifaction=='video':
downloader.save_mpd()
downloader.download_init()
downloader.download_video()
elifaction=='audio':
downloader.download_audio()
elifaction=='merge':
downloader.merge()
elifaction=='check':
downloader.check()
elifaction=='manual':
ifargs.range:
start, end=map(int, args.range.split('-'))
print(f'manually set range to from {start} to {end} (inclusive)')
else:
worst=downloader.check()
ifnotworst:
print('No bad interval found. Stop.')
return
start=worst[0] +1
end=worst[1] -1
print(f'Automatically check the largest interval, between {start} and {end} (inclusive)')
ids=list(range(start, end+1)) ifstart<endelselist(range(start, end-1, -1))
id, status=downloader.quick_iterate(ids)
ifnotid:
print('No segment found in this range.')
else:
print(f'Found segment {id} with status {status}')

exceptKeyboardInterrupt:
print('\nInterrupted by user. Stop.')


if__name__=='__main__':
importargparse

parser=argparse.ArgumentParser(
description='Available actions:\n'
' all - Download both video and audio (including live and backtracking), and then merge them (default)\n'
' live - Download the live stream only (no backtracking)\n'
' video - Download video only\n'
' audio - Download audio only\n'
' merge - Merge downloaded video and audio\n'
' check - Check the downloaded segments to make sure there are no missing segments\n'
' manual - Manually check missing segments at the largest gap (or use --range to assign a range) (for debugging only)\n'
' info - Display downloader object info\n'
' import:<path> - Import segments downloaded via N_m3u8DL-RE from a given path',
formatter_class=argparse.RawTextHelpFormatter
 )

parser.add_argument("url", help="url of mpd")
parser.add_argument("--action", '-a', default='all', help="action to perform (default: all)")
parser.add_argument("--dir", "-d", help="save path (default: instalive_{mpd_id})")
parser.add_argument("--debug", action='store_true', help="debug mode")
parser.add_argument("--quality", "-q", default='pst', help="manually assign video quality by quality name.\n"
"(default: \"pst\": which is the \"original\" (?) and has the best bitrate,\n"
"but not necessarily the highest resolution.\n"
"Pass empty string or \"highest\" to use the highest resolution one.)")
parser.add_argument("--time", "-t", help="for debugging only; manually assign last t (default: auto)")
parser.add_argument('--range', help='for debugging only; manually assign iteration range (start,end) for manual action')

args=parser.parse_args()

main(args.url, args.dir, args.time, args.debug, args.action, args.quality)
importconcurrent.futures
importre
importthreading
importtime
importxml.etree.ElementTreeasET
fromdatetimeimportdatetime
frompathlibimportPath
fromshutilimportcopy2, copyfileobj, get_terminal_size
fromsubprocessimportrun
fromurllib.parseimporturljoin

fromtqdmimporttqdm
fromutilimportrequests_retry_session, get_webname, td_format


TOLERANCE=0.2


defparse_iso8601_duration(duration):
'''Parse ISO8601 duration to seconds.
 it sometimes uses float for seconds part, so we need to handle that.'''
pattern=r'P(?:(?P<days>\d+)D)?(?:T(?:(?P<hours>\d+)H)?(?:(?P<minutes>\d+)M)?(?:(?P<seconds>[.\d]+)S)?)?'
match=re.match(pattern, duration)
ifnotmatch:
returnNone

days=int(match.group('days') or0)
hours=int(match.group('hours') or0)
minutes=int(match.group('minutes') or0)
seconds=int(float(match.group('seconds') or0))

# Convert everything to seconds
total_seconds= (days*86400) + (hours*3600) + (minutes*60) +seconds
returntotal_seconds


defprint_full_width(s):
'''Print a string to full width of the terminal.'''
width=get_terminal_size().columns
print('\r'+s.ljust(width), end='')


defconcat(files, output, verbose=False):
'''Concatenate files into one file.'''
output=Path(output)
ifnotoutput.parent.exists():
output.parent.mkdir(parents=True)
out=output.open('wb')
forfintqdm(files, ncols=100):
f=Path(f)
ifverbose:
print(f'Merging {f.name}...')
fi=f.open('rb')
copyfileobj(fi, out)
fi.close()
out.close()


classInstaliveDownloader:
def__init__(self, url, save_path, debug=False, quality=None):
self.session=requests_retry_session()
self.url=url
ifsave_pathisNone:
mpd_name=get_webname(url).split('.')[0]
save_path=f'instalive_{mpd_name}'
self.save_path=Path(save_path)
ifnotself.save_path.exists():
self.save_path.mkdir(parents=True)
assertself.save_path.is_dir(), f'{self.save_path} is not a directory!'
self.debug=debug
ifdebug:
print('Debug mode enabled. Only download 30 segments.')
self.fetch_mpd()
self.parse_mpd(quality)

def_download(self, url, save_path=None, filename=None, skip_existing=True):
ifsave_path:
f=Path(save_path) /get_webname(url)
eliffilename:
f=Path(filename)
ifskip_existingandf.exists() andf.stat().st_size>0:
return'Exists'
f.parent.mkdir(parents=True, exist_ok=True)
withself.session.get(url) asr:
ifr.status_code==200:
withf.open('wb') asf:
f.write(r.content)
returnr.status_code

defsave_mpd(self):
print('Save mpd to local file...')
mpd_lines=self.mpd_text.splitlines(keepends=True)
comment=f'<!-- {self.url} -->\n'
ifmpd_linesandmpd_lines[0].strip().startswith('<?xml'):
mpd_lines.insert(1, comment)
else:
mpd_lines.insert(0, comment)
 (self.save_path/'mpd.mpd').write_text(''.join(mpd_lines), encoding='utf-8')

deffetch_mpd(self, mute=False):
ifnotmute:
print('Fetch mpd...')
self.mpd_text=self.session.get(self.url).text
self.mpd=ET.fromstring(self.mpd_text)
returnself.mpd

defparse_mpd(self, quality=None):
print('Parse mpd...')
mpd=self.mpd
# availabilityStartTime="2024-05-06T01:21:00-07:00"
# availabilityEndTime="2024-05-06T01:29:00-07:00"

deftd_format_with_direction(td):
'''a wrapper around td_format to show both positive and negative time deltas'''
iftd.total_seconds() <0:
returntd_format(-td) +" ago"
else:
return'in '+td_format(td)
current_time=datetime.now().astimezone()
availability_start_time=datetime.strptime(mpd.attrib['availabilityStartTime'], '%Y-%m-%dT%H:%M:%S%z')
availability_end_time=datetime.strptime(mpd.attrib['availabilityEndTime'], '%Y-%m-%dT%H:%M:%S%z')

print(f'Availability start time: {availability_start_time} ({td_format_with_direction(availability_start_time-current_time)})')
print(f'Availability end time: {availability_end_time} ({td_format_with_direction(availability_end_time-current_time)})')

# process video representations
period=mpd[0]
video_adaptation_set=period[0]
# make a simple list of video representations because we can't directly sort xml elements
videos= [
 {
'width': int(x.attrib['width']),
'height': int(x.attrib['height']),
'frame_rate': float(x.attrib['frameRate']),
'bandwidth': float(x.attrib['bandwidth']),
'idx': idx,
'id': x.attrib['id']
 } foridx, xinenumerate(video_adaptation_set)
 ]

print('Video representations:')
forvinvideos:
print(f'[{v["idx"]}] {v["id"]}{v["width"]}x{v["height"]}, {v["frame_rate"]}fps, {v["bandwidth"]/1024:.1f} kbps')

ifnotqualityorquality=='highest':
videos.sort(reverse=True, key=lambdax: (x['width'], x['height'], x['frame_rate'], x['bandwidth']))
self.video_index=videos[0]['idx']
self.video_id=videos[0]['id']
print(f'Use the highest one ([{self.video_index}] {self.video_id}).')
else:
forvinvideos:
# quality could be either "dash-lp-pst-v" or "pst"
ifquality==v['id'] orf'{str(quality)}-v'inv['id']:
self.video_index=v['idx']
self.video_id=v['id']
print(f'Use the specified one ([{self.video_index}] {self.video_id}).')
break
else:
raiseException(f'Quality {quality} not found.')

self.save_path_video=self.save_path/self.video_id

video_representation=video_adaptation_set[self.video_index]
video_segment_template=video_representation[0]
self.timescale=video_segment_template.attrib['timescale'] # not used for now
self.video_init=urljoin(self.url, video_segment_template.attrib['initialization'])
self.video_url_template=urljoin(self.url, video_segment_template.attrib['media']).replace('$Time$', '{}')
timeline=video_segment_template[0]
# find the last segment's t
self.last_t=int(timeline[-1].attrib['t'])
print(f'Last video segment t={self.last_t}')
# find the best interval for iterating heuristically
d_list= [int(x.attrib['d']) forxintimeline]
print('Intervals between segments:', ', '.join(str(d) fordind_list))
# get most common interval
self.interval=max(set(d_list), key=d_list.count)
print(f'Heuristically set interval to {self.interval}')

# process audio representations
audio_adaptation_set=period[1]
audios= [
 {
'id': x.attrib['id'],
'bandwidth': float(x.attrib['bandwidth']),
'sampling_rate': int(x.attrib['audioSamplingRate']),
'idx': idx
 } foridx, xinenumerate(audio_adaptation_set)
 ]
print('Audio representations:')
forainaudios:
print(f'[{a["idx"]}] {a["id"]}, {a["bandwidth"]/1024:.1f} kbps, {a["sampling_rate"]} Hz')

audios.sort(reverse=True, key=lambdax: (x['bandwidth'], x['sampling_rate']))
self.audio_index=audios[0]['idx']
self.audio_id=audios[0]['id']
print(f'Use the best one ([{self.audio_index}] {self.audio_id}).')
self.save_path_audio=self.save_path/self.audio_id

audio_representation=audio_adaptation_set[self.audio_index]
audio_segment_template=audio_representation[0]

self.audio_init=urljoin(self.url, audio_segment_template.attrib['initialization'])
self.audio_url_template=urljoin(self.url, audio_segment_template.attrib['media']).replace('$Time$', '{}')

defdownload_init(self):
print('Download init segments...')
r=self._download(self.video_init, save_path=self.save_path_video)
assertrin [200, 'Exists']
r=self._download(self.audio_init, save_path=self.save_path_audio)
assertrin [200, 'Exists']

def_get_segments(self):
mpd=self.fetch_mpd(mute=True) # fetch a new mpd to get the latest segments
video_representation=mpd[0][0][self.video_index]
assertvideo_representation.attrib['id'] ==self.video_id
timeline=video_representation[0][0]
segments= [int(timeline[i].attrib['t']) foriinrange(len(timeline))]
returnsegments, mpd

defmanually_set(self, last_t=None):
iflast_tisnotNone:
self.last_t=int(last_t)
print(f'Last time set to {self.last_t}')

deffetch_video_by_id(self, id):
url=self.video_url_template.format(id)
status=self._download(url, save_path=self.save_path_video)
returnstatus

deffetch_audio_by_id(self, id):
url=self.audio_url_template.format(id)
status=self._download(url, save_path=self.save_path_audio)
returnstatus

defquick_iterate(self, ids):
# make sure ids are larger than 0
ids= [idforidinidsifid>0]
print(f'\nUse multi-threading to check {len(ids)} IDs starting from {ids[0]}...')
withconcurrent.futures.ThreadPoolExecutor(max_workers=20) asex:
futures= {ex.submit(self.fetch_video_by_id, id): idforidinids}
try:
forfutureinconcurrent.futures.as_completed(futures):
iffuture.cancelled():
continue
id=futures[future]
status=future.result()
ifstatusin ['Exists', 200]:
# cancel all other futures
forfinfutures:
f.cancel()
returnid, status
# TODO: does not work since we call it in a thread.
exceptKeyboardInterrupt:
print('\nInterrupted. Cancel all futures...')
forfinfutures:
f.cancel()
raiseKeyboardInterrupt
returnNone, None

defdownload_live(self):
downloaded=set()

mpd=self.mpd
# check if mpd is dynamic
ifmpd.attrib.get('type') !='dynamic':
print('This mpd is not dynamic. Stop monitoring live.')
return
# use minimumUpdatePeriod if available. otherwise, use timeShiftBufferDepth/2-1 as the interval.
# make sure it is at least 2s.
fetch_interval=parse_iso8601_duration(mpd.attrib.get('minimumUpdatePeriod', 'PT0S')) \
orparse_iso8601_duration(mpd.attrib.get('timeShiftBufferDepth', 'PT0S')) //2-1
fetch_interval=max(fetch_interval, 2)

MAX_IDLE_COUNT=20

unchanged_mpd_count=0
whileTrue:
new_segments, mpd=self._get_segments()
ifall(idindownloadedforidinnew_segments):
unchanged_mpd_count+=1
ifunchanged_mpd_count>=MAX_IDLE_COUNT:
ifmpd.attrib.get('type') =='dynamic':
print(f'No new segments found in the last {MAX_IDLE_COUNT} checks. But the mpd is still dynamic. Continue monitoring...')
else:
print('All segments are downloaded. Stop.')
return
else:
undownloaded= [idforidinnew_segmentsifidnotindownloaded]
print(f'{len(undownloaded)} new segments found. Downloading...')
foridinundownloaded:
# singe-thread should be enough for live stream
self.fetch_video_by_id(id)
self.fetch_audio_by_id(id)
downloaded.add(id)

time.sleep(fetch_interval)

defdownload_video(self, forward=False):
count=0
known_intervals= {
self.interval: 0,
self.interval-1: 0,
self.interval+1: 0,
 }
id_guesses= [self.last_t]
prev_id=None
sign=1ifforwardelse-1

defsurrounding(x):
start_id=x-int(self.interval* (0.5+TOLERANCE)) *sign
end_id=x+int(self.interval* (1.0+TOLERANCE) +1) *sign
ids=list(range(start_id, end_id, -1ifstart_id>end_idelse1))
ids.sort(key=lambdak: abs(k-x))
returnids

whileTrue:
valid_id=None
# firstly, we try to find if existing local file that is close to the guesses[0].
# which is defined as +/- TOLERANCE (default: 20%) of the interval.
# this is a "narrower" window than surrounding() function.
# notice that we don't need to try all the guesses, they will be checked in the
# next step.
ids=list(range(id_guesses[0] -int(self.interval*TOLERANCE), id_guesses[0] +int(self.interval*TOLERANCE) +1))
ids.sort(key=lambdak: abs(k-id_guesses[0]))
# only try ids that are before the prev_id (if backward) or after the prev_id (if forward).
ifprev_id:
ids= [idforidinidsif (id-prev_id) *sign>0]
forcandidateinids:
url=self.video_url_template.format(candidate)
f=self.save_path_video/get_webname(url)
iff.exists() andf.stat().st_size>0:
valid_id=candidate
print_full_width(f'Segment {valid_id}: {f} already exists. Skip.')
break
# if not found locally, we try to fetch the segment by id from the all
# (last_valid_id - potential_interval) pools.
# The order matters: we try the most common interval first.
# this is single-threaded, because it is more time-consuming trying to close all the threads.
ifnotvalid_id:
forcandidateinid_guesses:
status=self.fetch_video_by_id(candidate)
print_full_width(f'Segment {candidate}: HTTP {status}')
ifstatusin [200, 'Exists']:
valid_id=candidate
break
# If still not found, we iterate around the guesses[0] to find the next segment.
# the range is defined as 50%+TOLERANCE behind of x and 100%+TOLERANCE ahead of x,
# where x is the guesses[0].
# e.g. for 2000 interval, x = last_id - 2000, the range would be
# last_id - 800 to last_id - 4400, sorted by distance to last_id - 2000.
# notice that it covers up to the range of next next id, this way we ensure we still
# continue the downloading instead of stopping too early (despite missing a segment).
ifnotvalid_id:
ids= [idforidinsurrounding(id_guesses[0]) ifidnotinid_guesses]
ifprev_id:
ids= [idforidinidsif (id-prev_id) *sign>0]
valid_id, status=self.quick_iterate(ids)
# if still not, we assume we downloaded them all and stop.
ifnotvalid_id:
print("\nFailed to find next segment. Assume we downloaded all. Stop.")
break
print_full_width(f'Segment {valid_id}: HTTP {status}')
# at this point, we should have a valid_id.
assertvalid_id
# add new interval to known_intervals
ifprev_id:
new_interval=abs(valid_id-prev_id)
assertnew_interval>0# this should not happen.
# do not add new interval if it is too different from the current interval.
# for example, if the nominal interval is 2000, we should only add ones that are
# less than 3000. Otherwise we enables the possibility of skipping segments.
# But it is still allowed to have such large interval so we don't stop downloading
# in the middle just because of one missing segment.
ifnew_interval>=self.interval*1.6:
pass
# add interval to known_intervals (if not already), and increase the count.
else:
known_intervals.setdefault(new_interval, 0)
known_intervals[new_interval] +=1

# print(f'[Debug] known_intervals: {known_intervals}')
count+=1
ifself.debugandcount==30:
break
prev_id=valid_id
# sort known_intervals by appearance, so we try the most common interval first.
known_intervals=dict(sorted(known_intervals.items(), key=lambdax: x[1], reverse=True))
id_guesses= [valid_id+interval*signforintervalinknown_intervals]

defcheck(self):
print('Check if there is any missing video segment...')
files=list(self.save_path_video.iterdir())
# filename format: 17981336783244063_0-1297029.m4v
ids= [int(m[1]) forfinfilesif (m:=re.search(r'\d+_0-(\d+)', f.name))]
ids.sort()
# calculate difference between each id
diffs= [ids[i+1] -ids[i] foriinrange(len(ids)-1)]

diff_count= {}
fordiffindiffs:
diff_count[diff] =diff_count.get(diff, 0) +1
print(f'Count of diff values between each segment:', diff_count)
max_idx=diffs.index(max(diffs))
print(f'Largest ID diff between each segment: {max(diffs)}, at {ids[max_idx]} -> {ids[max_idx+1]}')
min_idx=diffs.index(min(diffs))
print(f'Smallest ID diff between each segment: {min(diffs)}, at {ids[min_idx]} -> {ids[min_idx+1]}')

# Check if there are any missing segments, defined as gaps larger than 1.2 times the interval.
ifmax(diffs) >self.interval* (1+TOLERANCE):
print(f'It is likely that the video is not fully downloaded!!')
return (ids[max_idx], ids[max_idx+1])
else:
print('No missing segment found.')


defdownload_audio(self):
print('Downloading audio segments...')
files=list(self.save_path_video.iterdir())
withconcurrent.futures.ThreadPoolExecutor(max_workers=20) asex:
futures= []
count=0
forfinfiles:
ifnotf.is_file():
continue
iff.suffix!='.m4v':
continue
id=re.search(r'_0-(init|\d+)\.m4v', f.name)[1]
url=self.audio_url_template.format(id)
futures.append(ex.submit(self._download, url, save_path=self.save_path_audio))
for_inconcurrent.futures.as_completed(futures):
count+=1
print(f'Finished {count}/{len(futures)} ', end='\r')

defmerge(self):
defget_key(f):
'''Make sure init segment is always at the beginning.'''
if'-init'inf.name:
return0
returnint(re.search(r'\d+_0-(\d+)', f.name)[1])

video_file=self.save_path/'video.m4v'
files=list(self.save_path_video.iterdir())
print(f'Find {len(files)} video segments. Merging...')
files.sort(key=get_key)
concat(files, video_file)

audio_file=self.save_path/'audio.m4a'
files2=list(self.save_path_audio.iterdir())
print(f'Find {len(files2)} audio segments. Merging...')
files2.sort(key=get_key)
concat(files2, audio_file)

print(f'Merging video and audio using FFMPEG...')
run(['ffmpeg', '-loglevel', 'error', '-stats', '-i', video_file, '-i', audio_file, '-c', 'copy', self.save_path/'merged.mp4'])

defimport_segments(self, path):
'''Import segments downloaded via N_m3u8DL-RE.'''
path=Path(path)
forpinpath.iterdir():
ifp.is_dir() and (p/'_init.mp4').exists():
if'avc'inp.name:
template=self.video_url_template
save_path=self.save_path_video
else:
template=self.audio_url_template
save_path=self.save_path_audio
forfinp.iterdir():
iff.stem.isdigit():
id=int(f.stem)
new_filename=get_webname(template.format(id))
newf=save_path/new_filename
ifnewf.exists():
ifnewf.stat().st_size==f.stat().st_size:
print(f'{id}: {newf} already exists. Skip.')
else:
raiseException(f'{id}: {newf} already exists but size is different.')
else:
print(f'{id}: Copy {f} to {newf}')
copy2(f, newf)


defmain(url, save_path, time, debug, action, quality):
downloader=InstaliveDownloader(url=url, save_path=save_path, debug=debug, quality=quality)
iftimeisnotNone:
downloader.manually_set(time)
ifaction.startswith('import'):
import_path=action.partition(':')[2]
downloader.import_segments(import_path)
return
ifaction=='info':
fork, vindownloader.__dict__.items():
ifk=='mpd_text': # too long.
v=v[:10] +'...'+v[-10:]
print(f'{k}: {v}')
return
try:
ifaction=='all':
downloader.save_mpd()
downloader.download_init()
deftask1():
downloader.download_video()
downloader.download_audio()
t1=threading.Thread(target=task1)
t2=threading.Thread(target=downloader.download_live)
t1.start()
t2.start()
t1.join()
t2.join()
downloader.check()
downloader.merge()
elifaction=='live':
downloader.save_mpd()
downloader.download_live()
elifaction=='video':
downloader.save_mpd()
downloader.download_init()
downloader.download_video()
elifaction=='audio':
downloader.download_audio()
elifaction=='merge':
downloader.merge()
elifaction=='check':
downloader.check()
elifaction=='manual':
ifargs.range:
start, end=map(int, args.range.split('-'))
print(f'manually set range to from {start} to {end} (inclusive)')
else:
worst=downloader.check()
ifnotworst:
print('No bad interval found. Stop.')
return
start=worst[0] +1
end=worst[1] -1
print(f'Automatically check the largest interval, between {start} and {end} (inclusive)')
ids=list(range(start, end+1)) ifstart<endelselist(range(start, end-1, -1))
id, status=downloader.quick_iterate(ids)
ifnotid:
print('No segment found in this range.')
else:
print(f'Found segment {id} with status {status}')

exceptKeyboardInterrupt:
print('\nInterrupted by user. Stop.')


if__name__=='__main__':
importargparse

parser=argparse.ArgumentParser(
description='Available actions:\n'
' all - Download both video and audio (including live and backtracking), and then merge them (default)\n'
' live - Download the live stream only (no backtracking)\n'
' video - Download video only\n'
' audio - Download audio only\n'
' merge - Merge downloaded video and audio\n'
' check - Check the downloaded segments to make sure there are no missing segments\n'
' manual - Manually check missing segments at the largest gap (or use --range to assign a range) (for debugging only)\n'
' info - Display downloader object info\n'
' import:<path> - Import segments downloaded via N_m3u8DL-RE from a given path',
formatter_class=argparse.RawTextHelpFormatter
 )

parser.add_argument("url", help="url of mpd")
parser.add_argument("--action", '-a', default='all', help="action to perform (default: all)")
parser.add_argument("--dir", "-d", help="save path (default: instalive_{mpd_id})")
parser.add_argument("--debug", action='store_true', help="debug mode")
parser.add_argument("--quality", "-q", default='pst', help="manually assign video quality by quality name.\n"
"(default: \"pst\": which is the \"original\" (?) and has the best bitrate,\n"
"but not necessarily the highest resolution.\n"
"Pass empty string or \"highest\" to use the highest resolution one.)")
parser.add_argument("--time", "-t", help="for debugging only; manually assign last t (default: auto)")
parser.add_argument('--range', help='for debugging only; manually assign iteration range (start,end) for manual action')

args=parser.parse_args()

main(args.url, args.dir, args.time, args.debug, args.action, args.quality)
importconcurrent.futures
importre
importthreading
importtime
importxml.etree.ElementTreeasET
fromdatetimeimportdatetime
frompathlibimportPath
fromshutilimportcopy2, copyfileobj, get_terminal_size
fromsubprocessimportrun
fromurllib.parseimporturljoin

fromtqdmimporttqdm
fromutilimportrequests_retry_session, get_webname, td_format


TOLERANCE=0.2


defparse_iso8601_duration(duration):
'''Parse ISO8601 duration to seconds.
 it sometimes uses float for seconds part, so we need to handle that.'''
pattern=r'P(?:(?P<days>\d+)D)?(?:T(?:(?P<hours>\d+)H)?(?:(?P<minutes>\d+)M)?(?:(?P<seconds>[.\d]+)S)?)?'
match=re.match(pattern, duration)
ifnotmatch:
returnNone

days=int(match.group('days') or0)
hours=int(match.group('hours') or0)
minutes=int(match.group('minutes') or0)
seconds=int(float(match.group('seconds') or0))

# Convert everything to seconds
total_seconds= (days*86400) + (hours*3600) + (minutes*60) +seconds
returntotal_seconds


defprint_full_width(s):
'''Print a string to full width of the terminal.'''
width=get_terminal_size().columns
print('\r'+s.ljust(width), end='')


defconcat(files, output, verbose=False):
'''Concatenate files into one file.'''
output=Path(output)
ifnotoutput.parent.exists():
output.parent.mkdir(parents=True)
out=output.open('wb')
forfintqdm(files, ncols=100):
f=Path(f)
ifverbose:
print(f'Merging {f.name}...')
fi=f.open('rb')
copyfileobj(fi, out)
fi.close()
out.close()


classInstaliveDownloader:
def__init__(self, url, save_path, debug=False, quality=None):
self.session=requests_retry_session()
self.url=url
ifsave_pathisNone:
mpd_name=get_webname(url).split('.')[0]
save_path=f'instalive_{mpd_name}'
self.save_path=Path(save_path)
ifnotself.save_path.exists():
self.save_path.mkdir(parents=True)
assertself.save_path.is_dir(), f'{self.save_path} is not a directory!'
self.debug=debug
ifdebug:
print('Debug mode enabled. Only download 30 segments.')
self.fetch_mpd()
self.parse_mpd(quality)

def_download(self, url, save_path=None, filename=None, skip_existing=True):
ifsave_path:
f=Path(save_path) /get_webname(url)
eliffilename:
f=Path(filename)
ifskip_existingandf.exists() andf.stat().st_size>0:
return'Exists'
f.parent.mkdir(parents=True, exist_ok=True)
withself.session.get(url) asr:
ifr.status_code==200:
withf.open('wb') asf:
f.write(r.content)
returnr.status_code

defsave_mpd(self):
print('Save mpd to local file...')
mpd_lines=self.mpd_text.splitlines(keepends=True)
comment=f'<!-- {self.url} -->\n'
ifmpd_linesandmpd_lines[0].strip().startswith('<?xml'):
mpd_lines.insert(1, comment)
else:
mpd_lines.insert(0, comment)
 (self.save_path/'mpd.mpd').write_text(''.join(mpd_lines), encoding='utf-8')

deffetch_mpd(self, mute=False):
ifnotmute:
print('Fetch mpd...')
self.mpd_text=self.session.get(self.url).text
self.mpd=ET.fromstring(self.mpd_text)
returnself.mpd

defparse_mpd(self, quality=None):
print('Parse mpd...')
mpd=self.mpd
# availabilityStartTime="2024-05-06T01:21:00-07:00"
# availabilityEndTime="2024-05-06T01:29:00-07:00"

deftd_format_with_direction(td):
'''a wrapper around td_format to show both positive and negative time deltas'''
iftd.total_seconds() <0:
returntd_format(-td) +" ago"
else:
return'in '+td_format(td)
current_time=datetime.now().astimezone()
availability_start_time=datetime.strptime(mpd.attrib['availabilityStartTime'], '%Y-%m-%dT%H:%M:%S%z')
availability_end_time=datetime.strptime(mpd.attrib['availabilityEndTime'], '%Y-%m-%dT%H:%M:%S%z')

print(f'Availability start time: {availability_start_time} ({td_format_with_direction(availability_start_time-current_time)})')
print(f'Availability end time: {availability_end_time} ({td_format_with_direction(availability_end_time-current_time)})')

# process video representations
period=mpd[0]
video_adaptation_set=period[0]
# make a simple list of video representations because we can't directly sort xml elements
videos= [
 {
'width': int(x.attrib['width']),
'height': int(x.attrib['height']),
'frame_rate': float(x.attrib['frameRate']),
'bandwidth': float(x.attrib['bandwidth']),
'idx': idx,
'id': x.attrib['id']
 } foridx, xinenumerate(video_adaptation_set)
 ]

print('Video representations:')
forvinvideos:
print(f'[{v["idx"]}] {v["id"]}{v["width"]}x{v["height"]}, {v["frame_rate"]}fps, {v["bandwidth"]/1024:.1f} kbps')

ifnotqualityorquality=='highest':
videos.sort(reverse=True, key=lambdax: (x['width'], x['height'], x['frame_rate'], x['bandwidth']))
self.video_index=videos[0]['idx']
self.video_id=videos[0]['id']
print(f'Use the highest one ([{self.video_index}] {self.video_id}).')
else:
forvinvideos:
# quality could be either "dash-lp-pst-v" or "pst"
ifquality==v['id'] orf'{str(quality)}-v'inv['id']:
self.video_index=v['idx']
self.video_id=v['id']
print(f'Use the specified one ([{self.video_index}] {self.video_id}).')
break
else:
raiseException(f'Quality {quality} not found.')

self.save_path_video=self.save_path/self.video_id

video_representation=video_adaptation_set[self.video_index]
video_segment_template=video_representation[0]
self.timescale=video_segment_template.attrib['timescale'] # not used for now
self.video_init=urljoin(self.url, video_segment_template.attrib['initialization'])
self.video_url_template=urljoin(self.url, video_segment_template.attrib['media']).replace('$Time$', '{}')
timeline=video_segment_template[0]
# find the last segment's t
self.last_t=int(timeline[-1].attrib['t'])
print(f'Last video segment t={self.last_t}')
# find the best interval for iterating heuristically
d_list= [int(x.attrib['d']) forxintimeline]
print('Intervals between segments:', ', '.join(str(d) fordind_list))
# get most common interval
self.interval=max(set(d_list), key=d_list.count)
print(f'Heuristically set interval to {self.interval}')

# process audio representations
audio_adaptation_set=period[1]
audios= [
 {
'id': x.attrib['id'],
'bandwidth': float(x.attrib['bandwidth']),
'sampling_rate': int(x.attrib['audioSamplingRate']),
'idx': idx
 } foridx, xinenumerate(audio_adaptation_set)
 ]
print('Audio representations:')
forainaudios:
print(f'[{a["idx"]}] {a["id"]}, {a["bandwidth"]/1024:.1f} kbps, {a["sampling_rate"]} Hz')

audios.sort(reverse=True, key=lambdax: (x['bandwidth'], x['sampling_rate']))
self.audio_index=audios[0]['idx']
self.audio_id=audios[0]['id']
print(f'Use the best one ([{self.audio_index}] {self.audio_id}).')
self.save_path_audio=self.save_path/self.audio_id

audio_representation=audio_adaptation_set[self.audio_index]
audio_segment_template=audio_representation[0]

self.audio_init=urljoin(self.url, audio_segment_template.attrib['initialization'])
self.audio_url_template=urljoin(self.url, audio_segment_template.attrib['media']).replace('$Time$', '{}')

defdownload_init(self):
print('Download init segments...')
r=self._download(self.video_init, save_path=self.save_path_video)
assertrin [200, 'Exists']
r=self._download(self.audio_init, save_path=self.save_path_audio)
assertrin [200, 'Exists']

def_get_segments(self):
mpd=self.fetch_mpd(mute=True) # fetch a new mpd to get the latest segments
video_representation=mpd[0][0][self.video_index]
assertvideo_representation.attrib['id'] ==self.video_id
timeline=video_representation[0][0]
segments= [int(timeline[i].attrib['t']) foriinrange(len(timeline))]
returnsegments, mpd

defmanually_set(self, last_t=None):
iflast_tisnotNone:
self.last_t=int(last_t)
print(f'Last time set to {self.last_t}')

deffetch_video_by_id(self, id):
url=self.video_url_template.format(id)
status=self._download(url, save_path=self.save_path_video)
returnstatus

deffetch_audio_by_id(self, id):
url=self.audio_url_template.format(id)
status=self._download(url, save_path=self.save_path_audio)
returnstatus

defquick_iterate(self, ids):
# make sure ids are larger than 0
ids= [idforidinidsifid>0]
print(f'\nUse multi-threading to check {len(ids)} IDs starting from {ids[0]}...')
withconcurrent.futures.ThreadPoolExecutor(max_workers=20) asex:
futures= {ex.submit(self.fetch_video_by_id, id): idforidinids}
try:
forfutureinconcurrent.futures.as_completed(futures):
iffuture.cancelled():
continue
id=futures[future]
status=future.result()
ifstatusin ['Exists', 200]:
# cancel all other futures
forfinfutures:
f.cancel()
returnid, status
# TODO: does not work since we call it in a thread.
exceptKeyboardInterrupt:
print('\nInterrupted. Cancel all futures...')
forfinfutures:
f.cancel()
raiseKeyboardInterrupt
returnNone, None

defdownload_live(self):
downloaded=set()

mpd=self.mpd
# check if mpd is dynamic
ifmpd.attrib.get('type') !='dynamic':
print('This mpd is not dynamic. Stop monitoring live.')
return
# use minimumUpdatePeriod if available. otherwise, use timeShiftBufferDepth/2-1 as the interval.
# make sure it is at least 2s.
fetch_interval=parse_iso8601_duration(mpd.attrib.get('minimumUpdatePeriod', 'PT0S')) \
orparse_iso8601_duration(mpd.attrib.get('timeShiftBufferDepth', 'PT0S')) //2-1
fetch_interval=max(fetch_interval, 2)

MAX_IDLE_COUNT=20

unchanged_mpd_count=0
whileTrue:
new_segments, mpd=self._get_segments()
ifall(idindownloadedforidinnew_segments):
unchanged_mpd_count+=1
ifunchanged_mpd_count>=MAX_IDLE_COUNT:
ifmpd.attrib.get('type') =='dynamic':
print(f'No new segments found in the last {MAX_IDLE_COUNT} checks. But the mpd is still dynamic. Continue monitoring...')
else:
print('All segments are downloaded. Stop.')
return
else:
undownloaded= [idforidinnew_segmentsifidnotindownloaded]
print(f'{len(undownloaded)} new segments found. Downloading...')
foridinundownloaded:
# singe-thread should be enough for live stream
self.fetch_video_by_id(id)
self.fetch_audio_by_id(id)
downloaded.add(id)

time.sleep(fetch_interval)

defdownload_video(self, forward=False):
count=0
known_intervals= {
self.interval: 0,
self.interval-1: 0,
self.interval+1: 0,
 }
id_guesses= [self.last_t]
prev_id=None
sign=1ifforwardelse-1

defsurrounding(x):
start_id=x-int(self.interval* (0.5+TOLERANCE)) *sign
end_id=x+int(self.interval* (1.0+TOLERANCE) +1) *sign
ids=list(range(start_id, end_id, -1ifstart_id>end_idelse1))
ids.sort(key=lambdak: abs(k-x))
returnids

whileTrue:
valid_id=None
# firstly, we try to find if existing local file that is close to the guesses[0].
# which is defined as +/- TOLERANCE (default: 20%) of the interval.
# this is a "narrower" window than surrounding() function.
# notice that we don't need to try all the guesses, they will be checked in the
# next step.
ids=list(range(id_guesses[0] -int(self.interval*TOLERANCE), id_guesses[0] +int(self.interval*TOLERANCE) +1))
ids.sort(key=lambdak: abs(k-id_guesses[0]))
# only try ids that are before the prev_id (if backward) or after the prev_id (if forward).
ifprev_id:
ids= [idforidinidsif (id-prev_id) *sign>0]
forcandidateinids:
url=self.video_url_template.format(candidate)
f=self.save_path_video/get_webname(url)
iff.exists() andf.stat().st_size>0:
valid_id=candidate
print_full_width(f'Segment {valid_id}: {f} already exists. Skip.')
break
# if not found locally, we try to fetch the segment by id from the all
# (last_valid_id - potential_interval) pools.
# The order matters: we try the most common interval first.
# this is single-threaded, because it is more time-consuming trying to close all the threads.
ifnotvalid_id:
forcandidateinid_guesses:
status=self.fetch_video_by_id(candidate)
print_full_width(f'Segment {candidate}: HTTP {status}')
ifstatusin [200, 'Exists']:
valid_id=candidate
break
# If still not found, we iterate around the guesses[0] to find the next segment.
# the range is defined as 50%+TOLERANCE behind of x and 100%+TOLERANCE ahead of x,
# where x is the guesses[0].
# e.g. for 2000 interval, x = last_id - 2000, the range would be
# last_id - 800 to last_id - 4400, sorted by distance to last_id - 2000.
# notice that it covers up to the range of next next id, this way we ensure we still
# continue the downloading instead of stopping too early (despite missing a segment).
ifnotvalid_id:
ids= [idforidinsurrounding(id_guesses[0]) ifidnotinid_guesses]
ifprev_id:
ids= [idforidinidsif (id-prev_id) *sign>0]
valid_id, status=self.quick_iterate(ids)
# if still not, we assume we downloaded them all and stop.
ifnotvalid_id:
print("\nFailed to find next segment. Assume we downloaded all. Stop.")
break
print_full_width(f'Segment {valid_id}: HTTP {status}')
# at this point, we should have a valid_id.
assertvalid_id
# add new interval to known_intervals
ifprev_id:
new_interval=abs(valid_id-prev_id)
assertnew_interval>0# this should not happen.
# do not add new interval if it is too different from the current interval.
# for example, if the nominal interval is 2000, we should only add ones that are
# less than 3000. Otherwise we enables the possibility of skipping segments.
# But it is still allowed to have such large interval so we don't stop downloading
# in the middle just because of one missing segment.
ifnew_interval>=self.interval*1.6:
pass
# add interval to known_intervals (if not already), and increase the count.
else:
known_intervals.setdefault(new_interval, 0)
known_intervals[new_interval] +=1

# print(f'[Debug] known_intervals: {known_intervals}')
count+=1
ifself.debugandcount==30:
break
prev_id=valid_id
# sort known_intervals by appearance, so we try the most common interval first.
known_intervals=dict(sorted(known_intervals.items(), key=lambdax: x[1], reverse=True))
id_guesses= [valid_id+interval*signforintervalinknown_intervals]

defcheck(self):
print('Check if there is any missing video segment...')
files=list(self.save_path_video.iterdir())
# filename format: 17981336783244063_0-1297029.m4v
ids= [int(m[1]) forfinfilesif (m:=re.search(r'\d+_0-(\d+)', f.name))]
ids.sort()
# calculate difference between each id
diffs= [ids[i+1] -ids[i] foriinrange(len(ids)-1)]

diff_count= {}
fordiffindiffs:
diff_count[diff] =diff_count.get(diff, 0) +1
print(f'Count of diff values between each segment:', diff_count)
max_idx=diffs.index(max(diffs))
print(f'Largest ID diff between each segment: {max(diffs)}, at {ids[max_idx]} -> {ids[max_idx+1]}')
min_idx=diffs.index(min(diffs))
print(f'Smallest ID diff between each segment: {min(diffs)}, at {ids[min_idx]} -> {ids[min_idx+1]}')

# Check if there are any missing segments, defined as gaps larger than 1.2 times the interval.
ifmax(diffs) >self.interval* (1+TOLERANCE):
print(f'It is likely that the video is not fully downloaded!!')
return (ids[max_idx], ids[max_idx+1])
else:
print('No missing segment found.')


defdownload_audio(self):
print('Downloading audio segments...')
files=list(self.save_path_video.iterdir())
withconcurrent.futures.ThreadPoolExecutor(max_workers=20) asex:
futures= []
count=0
forfinfiles:
ifnotf.is_file():
continue
iff.suffix!='.m4v':
continue
id=re.search(r'_0-(init|\d+)\.m4v', f.name)[1]
url=self.audio_url_template.format(id)
futures.append(ex.submit(self._download, url, save_path=self.save_path_audio))
for_inconcurrent.futures.as_completed(futures):
count+=1
print(f'Finished {count}/{len(futures)} ', end='\r')

defmerge(self):
defget_key(f):
'''Make sure init segment is always at the beginning.'''
if'-init'inf.name:
return0
returnint(re.search(r'\d+_0-(\d+)', f.name)[1])

video_file=self.save_path/'video.m4v'
files=list(self.save_path_video.iterdir())
print(f'Find {len(files)} video segments. Merging...')
files.sort(key=get_key)
concat(files, video_file)

audio_file=self.save_path/'audio.m4a'
files2=list(self.save_path_audio.iterdir())
print(f'Find {len(files2)} audio segments. Merging...')
files2.sort(key=get_key)
concat(files2, audio_file)

print(f'Merging video and audio using FFMPEG...')
run(['ffmpeg', '-loglevel', 'error', '-stats', '-i', video_file, '-i', audio_file, '-c', 'copy', self.save_path/'merged.mp4'])

defimport_segments(self, path):
'''Import segments downloaded via N_m3u8DL-RE.'''
path=Path(path)
forpinpath.iterdir():
ifp.is_dir() and (p/'_init.mp4').exists():
if'avc'inp.name:
template=self.video_url_template
save_path=self.save_path_video
else:
template=self.audio_url_template
save_path=self.save_path_audio
forfinp.iterdir():
iff.stem.isdigit():
id=int(f.stem)
new_filename=get_webname(template.format(id))
newf=save_path/new_filename
ifnewf.exists():
ifnewf.stat().st_size==f.stat().st_size:
print(f'{id}: {newf} already exists. Skip.')
else:
raiseException(f'{id}: {newf} already exists but size is different.')
else:
print(f'{id}: Copy {f} to {newf}')
copy2(f, newf)


defmain(url, save_path, time, debug, action, quality):
downloader=InstaliveDownloader(url=url, save_path=save_path, debug=debug, quality=quality)
iftimeisnotNone:
downloader.manually_set(time)
ifaction.startswith('import'):
import_path=action.partition(':')[2]
downloader.import_segments(import_path)
return
ifaction=='info':
fork, vindownloader.__dict__.items():
ifk=='mpd_text': # too long.
v=v[:10] +'...'+v[-10:]
print(f'{k}: {v}')
return
try:
ifaction=='all':
downloader.save_mpd()
downloader.download_init()
deftask1():
downloader.download_video()
downloader.download_audio()
t1=threading.Thread(target=task1)
t2=threading.Thread(target=downloader.download_live)
t1.start()
t2.start()
t1.join()
t2.join()
downloader.check()
downloader.merge()
elifaction=='live':
downloader.save_mpd()
downloader.download_live()
elifaction=='video':
downloader.save_mpd()
downloader.download_init()
downloader.download_video()
elifaction=='audio':
downloader.download_audio()
elifaction=='merge':
downloader.merge()
elifaction=='check':
downloader.check()
elifaction=='manual':
ifargs.range:
start, end=map(int, args.range.split('-'))
print(f'manually set range to from {start} to {end} (inclusive)')
else:
worst=downloader.check()
ifnotworst:
print('No bad interval found. Stop.')
return
start=worst[0] +1
end=worst[1] -1
print(f'Automatically check the largest interval, between {start} and {end} (inclusive)')
ids=list(range(start, end+1)) ifstart<endelselist(range(start, end-1, -1))
id, status=downloader.quick_iterate(ids)
ifnotid:
print('No segment found in this range.')
else:
print(f'Found segment {id} with status {status}')

exceptKeyboardInterrupt:
print('\nInterrupted by user. Stop.')


if__name__=='__main__':
importargparse

parser=argparse.ArgumentParser(
description='Available actions:\n'
' all - Download both video and audio (including live and backtracking), and then merge them (default)\n'
' live - Download the live stream only (no backtracking)\n'
' video - Download video only\n'
' audio - Download audio only\n'
' merge - Merge downloaded video and audio\n'
' check - Check the downloaded segments to make sure there are no missing segments\n'
' manual - Manually check missing segments at the largest gap (or use --range to assign a range) (for debugging only)\n'
' info - Display downloader object info\n'
' import:<path> - Import segments downloaded via N_m3u8DL-RE from a given path',
formatter_class=argparse.RawTextHelpFormatter
 )

parser.add_argument("url", help="url of mpd")
parser.add_argument("--action", '-a', default='all', help="action to perform (default: all)")
parser.add_argument("--dir", "-d", help="save path (default: instalive_{mpd_id})")
parser.add_argument("--debug", action='store_true', help="debug mode")
parser.add_argument("--quality", "-q", default='pst', help="manually assign video quality by quality name.\n"
"(default: \"pst\": which is the \"original\" (?) and has the best bitrate,\n"
"but not necessarily the highest resolution.\n"
"Pass empty string or \"highest\" to use the highest resolution one.)")
parser.add_argument("--time", "-t", help="for debugging only; manually assign last t (default: auto)")
parser.add_argument('--range', help='for debugging only; manually assign iteration range (start,end) for manual action')

args=parser.parse_args()

main(args.url, args.dir, args.time, args.debug, args.action, args.quality)
importconcurrent.futures
importre
importthreading
importtime
importxml.etree.ElementTreeasET
fromdatetimeimportdatetime
frompathlibimportPath
fromshutilimportcopy2, copyfileobj, get_terminal_size
fromsubprocessimportrun
fromurllib.parseimporturljoin

fromtqdmimporttqdm
fromutilimportrequests_retry_session, get_webname, td_format


TOLERANCE=0.2


defparse_iso8601_duration(duration):
'''Parse ISO8601 duration to seconds.
 it sometimes uses float for seconds part, so we need to handle that.'''
pattern=r'P(?:(?P<days>\d+)D)?(?:T(?:(?P<hours>\d+)H)?(?:(?P<minutes>\d+)M)?(?:(?P<seconds>[.\d]+)S)?)?'
match=re.match(pattern, duration)
ifnotmatch:
returnNone

days=int(match.group('days') or0)
hours=int(match.group('hours') or0)
minutes=int(match.group('minutes') or0)
seconds=int(float(match.group('seconds') or0))

# Convert everything to seconds
total_seconds= (days*86400) + (hours*3600) + (minutes*60) +seconds
returntotal_seconds


defprint_full_width(s):
'''Print a string to full width of the terminal.'''
width=get_terminal_size().columns
print('\r'+s.ljust(width), end='')


defconcat(files, output, verbose=False):
'''Concatenate files into one file.'''
output=Path(output)
ifnotoutput.parent.exists():
output.parent.mkdir(parents=True)
out=output.open('wb')
forfintqdm(files, ncols=100):
f=Path(f)
ifverbose:
print(f'Merging {f.name}...')
fi=f.open('rb')
copyfileobj(fi, out)
fi.close()
out.close()


classInstaliveDownloader:
def__init__(self, url, save_path, debug=False, quality=None):
self.session=requests_retry_session()
self.url=url
ifsave_pathisNone:
mpd_name=get_webname(url).split('.')[0]
save_path=f'instalive_{mpd_name}'
self.save_path=Path(save_path)
ifnotself.save_path.exists():
self.save_path.mkdir(parents=True)
assertself.save_path.is_dir(), f'{self.save_path} is not a directory!'
self.debug=debug
ifdebug:
print('Debug mode enabled. Only download 30 segments.')
self.fetch_mpd()
self.parse_mpd(quality)

def_download(self, url, save_path=None, filename=None, skip_existing=True):
ifsave_path:
f=Path(save_path) /get_webname(url)
eliffilename:
f=Path(filename)
ifskip_existingandf.exists() andf.stat().st_size>0:
return'Exists'
f.parent.mkdir(parents=True, exist_ok=True)
withself.session.get(url) asr:
ifr.status_code==200:
withf.open('wb') asf:
f.write(r.content)
returnr.status_code

defsave_mpd(self):
print('Save mpd to local file...')
mpd_lines=self.mpd_text.splitlines(keepends=True)
comment=f'<!-- {self.url} -->\n'
ifmpd_linesandmpd_lines[0].strip().startswith('<?xml'):
mpd_lines.insert(1, comment)
else:
mpd_lines.insert(0, comment)
 (self.save_path/'mpd.mpd').write_text(''.join(mpd_lines), encoding='utf-8')

deffetch_mpd(self, mute=False):
ifnotmute:
print('Fetch mpd...')
self.mpd_text=self.session.get(self.url).text
self.mpd=ET.fromstring(self.mpd_text)
returnself.mpd

defparse_mpd(self, quality=None):
print('Parse mpd...')
mpd=self.mpd
# availabilityStartTime="2024-05-06T01:21:00-07:00"
# availabilityEndTime="2024-05-06T01:29:00-07:00"

deftd_format_with_direction(td):
'''a wrapper around td_format to show both positive and negative time deltas'''
iftd.total_seconds() <0:
returntd_format(-td) +" ago"
else:
return'in '+td_format(td)
current_time=datetime.now().astimezone()
availability_start_time=datetime.strptime(mpd.attrib['availabilityStartTime'], '%Y-%m-%dT%H:%M:%S%z')
availability_end_time=datetime.strptime(mpd.attrib['availabilityEndTime'], '%Y-%m-%dT%H:%M:%S%z')

print(f'Availability start time: {availability_start_time} ({td_format_with_direction(availability_start_time-current_time)})')
print(f'Availability end time: {availability_end_time} ({td_format_with_direction(availability_end_time-current_time)})')

# process video representations
period=mpd[0]
video_adaptation_set=period[0]
# make a simple list of video representations because we can't directly sort xml elements
videos= [
 {
'width': int(x.attrib['width']),
'height': int(x.attrib['height']),
'frame_rate': float(x.attrib['frameRate']),
'bandwidth': float(x.attrib['bandwidth']),
'idx': idx,
'id': x.attrib['id']
 } foridx, xinenumerate(video_adaptation_set)
 ]

print('Video representations:')
forvinvideos:
print(f'[{v["idx"]}] {v["id"]}{v["width"]}x{v["height"]}, {v["frame_rate"]}fps, {v["bandwidth"]/1024:.1f} kbps')

ifnotqualityorquality=='highest':
videos.sort(reverse=True, key=lambdax: (x['width'], x['height'], x['frame_rate'], x['bandwidth']))
self.video_index=videos[0]['idx']
self.video_id=videos[0]['id']
print(f'Use the highest one ([{self.video_index}] {self.video_id}).')
else:
forvinvideos:
# quality could be either "dash-lp-pst-v" or "pst"
ifquality==v['id'] orf'{str(quality)}-v'inv['id']:
self.video_index=v['idx']
self.video_id=v['id']
print(f'Use the specified one ([{self.video_index}] {self.video_id}).')
break
else:
raiseException(f'Quality {quality} not found.')

self.save_path_video=self.save_path/self.video_id

video_representation=video_adaptation_set[self.video_index]
video_segment_template=video_representation[0]
self.timescale=video_segment_template.attrib['timescale'] # not used for now
self.video_init=urljoin(self.url, video_segment_template.attrib['initialization'])
self.video_url_template=urljoin(self.url, video_segment_template.attrib['media']).replace('$Time$', '{}')
timeline=video_segment_template[0]
# find the last segment's t
self.last_t=int(timeline[-1].attrib['t'])
print(f'Last video segment t={self.last_t}')
# find the best interval for iterating heuristically
d_list= [int(x.attrib['d']) forxintimeline]
print('Intervals between segments:', ', '.join(str(d) fordind_list))
# get most common interval
self.interval=max(set(d_list), key=d_list.count)
print(f'Heuristically set interval to {self.interval}')

# process audio representations
audio_adaptation_set=period[1]
audios= [
 {
'id': x.attrib['id'],
'bandwidth': float(x.attrib['bandwidth']),
'sampling_rate': int(x.attrib['audioSamplingRate']),
'idx': idx
 } foridx, xinenumerate(audio_adaptation_set)
 ]
print('Audio representations:')
forainaudios:
print(f'[{a["idx"]}] {a["id"]}, {a["bandwidth"]/1024:.1f} kbps, {a["sampling_rate"]} Hz')

audios.sort(reverse=True, key=lambdax: (x['bandwidth'], x['sampling_rate']))
self.audio_index=audios[0]['idx']
self.audio_id=audios[0]['id']
print(f'Use the best one ([{self.audio_index}] {self.audio_id}).')
self.save_path_audio=self.save_path/self.audio_id

audio_representation=audio_adaptation_set[self.audio_index]
audio_segment_template=audio_representation[0]

self.audio_init=urljoin(self.url, audio_segment_template.attrib['initialization'])
self.audio_url_template=urljoin(self.url, audio_segment_template.attrib['media']).replace('$Time$', '{}')

defdownload_init(self):
print('Download init segments...')
r=self._download(self.video_init, save_path=self.save_path_video)
assertrin [200, 'Exists']
r=self._download(self.audio_init, save_path=self.save_path_audio)
assertrin [200, 'Exists']

def_get_segments(self):
mpd=self.fetch_mpd(mute=True) # fetch a new mpd to get the latest segments
video_representation=mpd[0][0][self.video_index]
assertvideo_representation.attrib['id'] ==self.video_id
timeline=video_representation[0][0]
segments= [int(timeline[i].attrib['t']) foriinrange(len(timeline))]
returnsegments, mpd

defmanually_set(self, last_t=None):
iflast_tisnotNone:
self.last_t=int(last_t)
print(f'Last time set to {self.last_t}')

deffetch_video_by_id(self, id):
url=self.video_url_template.format(id)
status=self._download(url, save_path=self.save_path_video)
returnstatus

deffetch_audio_by_id(self, id):
url=self.audio_url_template.format(id)
status=self._download(url, save_path=self.save_path_audio)
returnstatus

defquick_iterate(self, ids):
# make sure ids are larger than 0
ids= [idforidinidsifid>0]
print(f'\nUse multi-threading to check {len(ids)} IDs starting from {ids[0]}...')
withconcurrent.futures.ThreadPoolExecutor(max_workers=20) asex:
futures= {ex.submit(self.fetch_video_by_id, id): idforidinids}
try:
forfutureinconcurrent.futures.as_completed(futures):
iffuture.cancelled():
continue
id=futures[future]
status=future.result()
ifstatusin ['Exists', 200]:
# cancel all other futures
forfinfutures:
f.cancel()
returnid, status
# TODO: does not work since we call it in a thread.
exceptKeyboardInterrupt:
print('\nInterrupted. Cancel all futures...')
forfinfutures:
f.cancel()
raiseKeyboardInterrupt
returnNone, None

defdownload_live(self):
downloaded=set()

mpd=self.mpd
# check if mpd is dynamic
ifmpd.attrib.get('type') !='dynamic':
print('This mpd is not dynamic. Stop monitoring live.')
return
# use minimumUpdatePeriod if available. otherwise, use timeShiftBufferDepth/2-1 as the interval.
# make sure it is at least 2s.
fetch_interval=parse_iso8601_duration(mpd.attrib.get('minimumUpdatePeriod', 'PT0S')) \
orparse_iso8601_duration(mpd.attrib.get('timeShiftBufferDepth', 'PT0S')) //2-1
fetch_interval=max(fetch_interval, 2)

MAX_IDLE_COUNT=20

unchanged_mpd_count=0
whileTrue:
new_segments, mpd=self._get_segments()
ifall(idindownloadedforidinnew_segments):
unchanged_mpd_count+=1
ifunchanged_mpd_count>=MAX_IDLE_COUNT:
ifmpd.attrib.get('type') =='dynamic':
print(f'No new segments found in the last {MAX_IDLE_COUNT} checks. But the mpd is still dynamic. Continue monitoring...')
else:
print('All segments are downloaded. Stop.')
return
else:
undownloaded= [idforidinnew_segmentsifidnotindownloaded]
print(f'{len(undownloaded)} new segments found. Downloading...')
foridinundownloaded:
# singe-thread should be enough for live stream
self.fetch_video_by_id(id)
self.fetch_audio_by_id(id)
downloaded.add(id)

time.sleep(fetch_interval)

defdownload_video(self, forward=False):
count=0
known_intervals= {
self.interval: 0,
self.interval-1: 0,
self.interval+1: 0,
 }
id_guesses= [self.last_t]
prev_id=None
sign=1ifforwardelse-1

defsurrounding(x):
start_id=x-int(self.interval* (0.5+TOLERANCE)) *sign
end_id=x+int(self.interval* (1.0+TOLERANCE) +1) *sign
ids=list(range(start_id, end_id, -1ifstart_id>end_idelse1))
ids.sort(key=lambdak: abs(k-x))
returnids

whileTrue:
valid_id=None
# firstly, we try to find if existing local file that is close to the guesses[0].
# which is defined as +/- TOLERANCE (default: 20%) of the interval.
# this is a "narrower" window than surrounding() function.
# notice that we don't need to try all the guesses, they will be checked in the
# next step.
ids=list(range(id_guesses[0] -int(self.interval*TOLERANCE), id_guesses[0] +int(self.interval*TOLERANCE) +1))
ids.sort(key=lambdak: abs(k-id_guesses[0]))
# only try ids that are before the prev_id (if backward) or after the prev_id (if forward).
ifprev_id:
ids= [idforidinidsif (id-prev_id) *sign>0]
forcandidateinids:
url=self.video_url_template.format(candidate)
f=self.save_path_video/get_webname(url)
iff.exists() andf.stat().st_size>0:
valid_id=candidate
print_full_width(f'Segment {valid_id}: {f} already exists. Skip.')
break
# if not found locally, we try to fetch the segment by id from the all
# (last_valid_id - potential_interval) pools.
# The order matters: we try the most common interval first.
# this is single-threaded, because it is more time-consuming trying to close all the threads.
ifnotvalid_id:
forcandidateinid_guesses:
status=self.fetch_video_by_id(candidate)
print_full_width(f'Segment {candidate}: HTTP {status}')
ifstatusin [200, 'Exists']:
valid_id=candidate
break
# If still not found, we iterate around the guesses[0] to find the next segment.
# the range is defined as 50%+TOLERANCE behind of x and 100%+TOLERANCE ahead of x,
# where x is the guesses[0].
# e.g. for 2000 interval, x = last_id - 2000, the range would be
# last_id - 800 to last_id - 4400, sorted by distance to last_id - 2000.
# notice that it covers up to the range of next next id, this way we ensure we still
# continue the downloading instead of stopping too early (despite missing a segment).
ifnotvalid_id:
ids= [idforidinsurrounding(id_guesses[0]) ifidnotinid_guesses]
ifprev_id:
ids= [idforidinidsif (id-prev_id) *sign>0]
valid_id, status=self.quick_iterate(ids)
# if still not, we assume we downloaded them all and stop.
ifnotvalid_id:
print("\nFailed to find next segment. Assume we downloaded all. Stop.")
break
print_full_width(f'Segment {valid_id}: HTTP {status}')
# at this point, we should have a valid_id.
assertvalid_id
# add new interval to known_intervals
ifprev_id:
new_interval=abs(valid_id-prev_id)
assertnew_interval>0# this should not happen.
# do not add new interval if it is too different from the current interval.
# for example, if the nominal interval is 2000, we should only add ones that are
# less than 3000. Otherwise we enables the possibility of skipping segments.
# But it is still allowed to have such large interval so we don't stop downloading
# in the middle just because of one missing segment.
ifnew_interval>=self.interval*1.6:
pass
# add interval to known_intervals (if not already), and increase the count.
else:
known_intervals.setdefault(new_interval, 0)
known_intervals[new_interval] +=1

# print(f'[Debug] known_intervals: {known_intervals}')
count+=1
ifself.debugandcount==30:
break
prev_id=valid_id
# sort known_intervals by appearance, so we try the most common interval first.
known_intervals=dict(sorted(known_intervals.items(), key=lambdax: x[1], reverse=True))
id_guesses= [valid_id+interval*signforintervalinknown_intervals]

defcheck(self):
print('Check if there is any missing video segment...')
files=list(self.save_path_video.iterdir())
# filename format: 17981336783244063_0-1297029.m4v
ids= [int(m[1]) forfinfilesif (m:=re.search(r'\d+_0-(\d+)', f.name))]
ids.sort()
# calculate difference between each id
diffs= [ids[i+1] -ids[i] foriinrange(len(ids)-1)]

diff_count= {}
fordiffindiffs:
diff_count[diff] =diff_count.get(diff, 0) +1
print(f'Count of diff values between each segment:', diff_count)
max_idx=diffs.index(max(diffs))
print(f'Largest ID diff between each segment: {max(diffs)}, at {ids[max_idx]} -> {ids[max_idx+1]}')
min_idx=diffs.index(min(diffs))
print(f'Smallest ID diff between each segment: {min(diffs)}, at {ids[min_idx]} -> {ids[min_idx+1]}')

# Check if there are any missing segments, defined as gaps larger than 1.2 times the interval.
ifmax(diffs) >self.interval* (1+TOLERANCE):
print(f'It is likely that the video is not fully downloaded!!')
return (ids[max_idx], ids[max_idx+1])
else:
print('No missing segment found.')


defdownload_audio(self):
print('Downloading audio segments...')
files=list(self.save_path_video.iterdir())
withconcurrent.futures.ThreadPoolExecutor(max_workers=20) asex:
futures= []
count=0
forfinfiles:
ifnotf.is_file():
continue
iff.suffix!='.m4v':
continue
id=re.search(r'_0-(init|\d+)\.m4v', f.name)[1]
url=self.audio_url_template.format(id)
futures.append(ex.submit(self._download, url, save_path=self.save_path_audio))
for_inconcurrent.futures.as_completed(futures):
count+=1
print(f'Finished {count}/{len(futures)} ', end='\r')

defmerge(self):
defget_key(f):
'''Make sure init segment is always at the beginning.'''
if'-init'inf.name:
return0
returnint(re.search(r'\d+_0-(\d+)', f.name)[1])

video_file=self.save_path/'video.m4v'
files=list(self.save_path_video.iterdir())
print(f'Find {len(files)} video segments. Merging...')
files.sort(key=get_key)
concat(files, video_file)

audio_file=self.save_path/'audio.m4a'
files2=list(self.save_path_audio.iterdir())
print(f'Find {len(files2)} audio segments. Merging...')
files2.sort(key=get_key)
concat(files2, audio_file)

print(f'Merging video and audio using FFMPEG...')
run(['ffmpeg', '-loglevel', 'error', '-stats', '-i', video_file, '-i', audio_file, '-c', 'copy', self.save_path/'merged.mp4'])

defimport_segments(self, path):
'''Import segments downloaded via N_m3u8DL-RE.'''
path=Path(path)
forpinpath.iterdir():
ifp.is_dir() and (p/'_init.mp4').exists():
if'avc'inp.name:
template=self.video_url_template
save_path=self.save_path_video
else:
template=self.audio_url_template
save_path=self.save_path_audio
forfinp.iterdir():
iff.stem.isdigit():
id=int(f.stem)
new_filename=get_webname(template.format(id))
newf=save_path/new_filename
ifnewf.exists():
ifnewf.stat().st_size==f.stat().st_size:
print(f'{id}: {newf} already exists. Skip.')
else:
raiseException(f'{id}: {newf} already exists but size is different.')
else:
print(f'{id}: Copy {f} to {newf}')
copy2(f, newf)


defmain(url, save_path, time, debug, action, quality):
downloader=InstaliveDownloader(url=url, save_path=save_path, debug=debug, quality=quality)
iftimeisnotNone:
downloader.manually_set(time)
ifaction.startswith('import'):
import_path=action.partition(':')[2]
downloader.import_segments(import_path)
return
ifaction=='info':
fork, vindownloader.__dict__.items():
ifk=='mpd_text': # too long.
v=v[:10] +'...'+v[-10:]
print(f'{k}: {v}')
return
try:
ifaction=='all':
downloader.save_mpd()
downloader.download_init()
deftask1():
downloader.download_video()
downloader.download_audio()
t1=threading.Thread(target=task1)
t2=threading.Thread(target=downloader.download_live)
t1.start()
t2.start()
t1.join()
t2.join()
downloader.check()
downloader.merge()
elifaction=='live':
downloader.save_mpd()
downloader.download_live()
elifaction=='video':
downloader.save_mpd()
downloader.download_init()
downloader.download_video()
elifaction=='audio':
downloader.download_audio()
elifaction=='merge':
downloader.merge()
elifaction=='check':
downloader.check()
elifaction=='manual':
ifargs.range:
start, end=map(int, args.range.split('-'))
print(f'manually set range to from {start} to {end} (inclusive)')
else:
worst=downloader.check()
ifnotworst:
print('No bad interval found. Stop.')
return
start=worst[0] +1
end=worst[1] -1
print(f'Automatically check the largest interval, between {start} and {end} (inclusive)')
ids=list(range(start, end+1)) ifstart<endelselist(range(start, end-1, -1))
id, status=downloader.quick_iterate(ids)
ifnotid:
print('No segment found in this range.')
else:
print(f'Found segment {id} with status {status}')

exceptKeyboardInterrupt:
print('\nInterrupted by user. Stop.')


if__name__=='__main__':
importargparse

parser=argparse.ArgumentParser(
description='Available actions:\n'
' all - Download both video and audio (including live and backtracking), and then merge them (default)\n'
' live - Download the live stream only (no backtracking)\n'
' video - Download video only\n'
' audio - Download audio only\n'
' merge - Merge downloaded video and audio\n'
' check - Check the downloaded segments to make sure there are no missing segments\n'
' manual - Manually check missing segments at the largest gap (or use --range to assign a range) (for debugging only)\n'
' info - Display downloader object info\n'
' import:<path> - Import segments downloaded via N_m3u8DL-RE from a given path',
formatter_class=argparse.RawTextHelpFormatter
 )

parser.add_argument("url", help="url of mpd")
parser.add_argument("--action", '-a', default='all', help="action to perform (default: all)")
parser.add_argument("--dir", "-d", help="save path (default: instalive_{mpd_id})")
parser.add_argument("--debug", action='store_true', help="debug mode")
parser.add_argument("--quality", "-q", default='pst', help="manually assign video quality by quality name.\n"
"(default: \"pst\": which is the \"original\" (?) and has the best bitrate,\n"
"but not necessarily the highest resolution.\n"
"Pass empty string or \"highest\" to use the highest resolution one.)")
parser.add_argument("--time", "-t", help="for debugging only; manually assign last t (default: auto)")
parser.add_argument('--range', help='for debugging only; manually assign iteration range (start,end) for manual action')

args=parser.parse_args()

main(args.url, args.dir, args.time, args.debug, args.action, args.quality)
importconcurrent.futures
importre
importthreading
importtime
importxml.etree.ElementTreeasET
fromdatetimeimportdatetime
frompathlibimportPath
fromshutilimportcopy2, copyfileobj, get_terminal_size
fromsubprocessimportrun
fromurllib.parseimporturljoin

fromtqdmimporttqdm
fromutilimportrequests_retry_session, get_webname, td_format


TOLERANCE=0.2


defparse_iso8601_duration(duration):
'''Parse ISO8601 duration to seconds.
 it sometimes uses float for seconds part, so we need to handle that.'''
pattern=r'P(?:(?P<days>\d+)D)?(?:T(?:(?P<hours>\d+)H)?(?:(?P<minutes>\d+)M)?(?:(?P<seconds>[.\d]+)S)?)?'
match=re.match(pattern, duration)
ifnotmatch:
returnNone

days=int(match.group('days') or0)
hours=int(match.group('hours') or0)
minutes=int(match.group('minutes') or0)
seconds=int(float(match.group('seconds') or0))

# Convert everything to seconds
total_seconds= (days*86400) + (hours*3600) + (minutes*60) +seconds
returntotal_seconds


defprint_full_width(s):
'''Print a string to full width of the terminal.'''
width=get_terminal_size().columns
print('\r'+s.ljust(width), end='')


defconcat(files, output, verbose=False):
'''Concatenate files into one file.'''
output=Path(output)
ifnotoutput.parent.exists():
output.parent.mkdir(parents=True)
out=output.open('wb')
forfintqdm(files, ncols=100):
f=Path(f)
ifverbose:
print(f'Merging {f.name}...')
fi=f.open('rb')
copyfileobj(fi, out)
fi.close()
out.close()


classInstaliveDownloader:
def__init__(self, url, save_path, debug=False, quality=None):
self.session=requests_retry_session()
self.url=url
ifsave_pathisNone:
mpd_name=get_webname(url).split('.')[0]
save_path=f'instalive_{mpd_name}'
self.save_path=Path(save_path)
ifnotself.save_path.exists():
self.save_path.mkdir(parents=True)
assertself.save_path.is_dir(), f'{self.save_path} is not a directory!'
self.debug=debug
ifdebug:
print('Debug mode enabled. Only download 30 segments.')
self.fetch_mpd()
self.parse_mpd(quality)

def_download(self, url, save_path=None, filename=None, skip_existing=True):
ifsave_path:
f=Path(save_path) /get_webname(url)
eliffilename:
f=Path(filename)
ifskip_existingandf.exists() andf.stat().st_size>0:
return'Exists'
f.parent.mkdir(parents=True, exist_ok=True)
withself.session.get(url) asr:
ifr.status_code==200:
withf.open('wb') asf:
f.write(r.content)
returnr.status_code

defsave_mpd(self):
print('Save mpd to local file...')
mpd_lines=self.mpd_text.splitlines(keepends=True)
comment=f'<!-- {self.url} -->\n'
ifmpd_linesandmpd_lines[0].strip().startswith('<?xml'):
mpd_lines.insert(1, comment)
else:
mpd_lines.insert(0, comment)
 (self.save_path/'mpd.mpd').write_text(''.join(mpd_lines), encoding='utf-8')

deffetch_mpd(self, mute=False):
ifnotmute:
print('Fetch mpd...')
self.mpd_text=self.session.get(self.url).text
self.mpd=ET.fromstring(self.mpd_text)
returnself.mpd

defparse_mpd(self, quality=None):
print('Parse mpd...')
mpd=self.mpd
# availabilityStartTime="2024-05-06T01:21:00-07:00"
# availabilityEndTime="2024-05-06T01:29:00-07:00"

deftd_format_with_direction(td):
'''a wrapper around td_format to show both positive and negative time deltas'''
iftd.total_seconds() <0:
returntd_format(-td) +" ago"
else:
return'in '+td_format(td)
current_time=datetime.now().astimezone()
availability_start_time=datetime.strptime(mpd.attrib['availabilityStartTime'], '%Y-%m-%dT%H:%M:%S%z')
availability_end_time=datetime.strptime(mpd.attrib['availabilityEndTime'], '%Y-%m-%dT%H:%M:%S%z')

print(f'Availability start time: {availability_start_time} ({td_format_with_direction(availability_start_time-current_time)})')
print(f'Availability end time: {availability_end_time} ({td_format_with_direction(availability_end_time-current_time)})')

# process video representations
period=mpd[0]
video_adaptation_set=period[0]
# make a simple list of video representations because we can't directly sort xml elements
videos= [
 {
'width': int(x.attrib['width']),
'height': int(x.attrib['height']),
'frame_rate': float(x.attrib['frameRate']),
'bandwidth': float(x.attrib['bandwidth']),
'idx': idx,
'id': x.attrib['id']
 } foridx, xinenumerate(video_adaptation_set)
 ]

print('Video representations:')
forvinvideos:
print(f'[{v["idx"]}] {v["id"]}{v["width"]}x{v["height"]}, {v["frame_rate"]}fps, {v["bandwidth"]/1024:.1f} kbps')

ifnotqualityorquality=='highest':
videos.sort(reverse=True, key=lambdax: (x['width'], x['height'], x['frame_rate'], x['bandwidth']))
self.video_index=videos[0]['idx']
self.video_id=videos[0]['id']
print(f'Use the highest one ([{self.video_index}] {self.video_id}).')
else:
forvinvideos:
# quality could be either "dash-lp-pst-v" or "pst"
ifquality==v['id'] orf'{str(quality)}-v'inv['id']:
self.video_index=v['idx']
self.video_id=v['id']
print(f'Use the specified one ([{self.video_index}] {self.video_id}).')
break
else:
raiseException(f'Quality {quality} not found.')

self.save_path_video=self.save_path/self.video_id

video_representation=video_adaptation_set[self.video_index]
video_segment_template=video_representation[0]
self.timescale=video_segment_template.attrib['timescale'] # not used for now
self.video_init=urljoin(self.url, video_segment_template.attrib['initialization'])
self.video_url_template=urljoin(self.url, video_segment_template.attrib['media']).replace('$Time$', '{}')
timeline=video_segment_template[0]
# find the last segment's t
self.last_t=int(timeline[-1].attrib['t'])
print(f'Last video segment t={self.last_t}')
# find the best interval for iterating heuristically
d_list= [int(x.attrib['d']) forxintimeline]
print('Intervals between segments:', ', '.join(str(d) fordind_list))
# get most common interval
self.interval=max(set(d_list), key=d_list.count)
print(f'Heuristically set interval to {self.interval}')

# process audio representations
audio_adaptation_set=period[1]
audios= [
 {
'id': x.attrib['id'],
'bandwidth': float(x.attrib['bandwidth']),
'sampling_rate': int(x.attrib['audioSamplingRate']),
'idx': idx
 } foridx, xinenumerate(audio_adaptation_set)
 ]
print('Audio representations:')
forainaudios:
print(f'[{a["idx"]}] {a["id"]}, {a["bandwidth"]/1024:.1f} kbps, {a["sampling_rate"]} Hz')

audios.sort(reverse=True, key=lambdax: (x['bandwidth'], x['sampling_rate']))
self.audio_index=audios[0]['idx']
self.audio_id=audios[0]['id']
print(f'Use the best one ([{self.audio_index}] {self.audio_id}).')
self.save_path_audio=self.save_path/self.audio_id

audio_representation=audio_adaptation_set[self.audio_index]
audio_segment_template=audio_representation[0]

self.audio_init=urljoin(self.url, audio_segment_template.attrib['initialization'])
self.audio_url_template=urljoin(self.url, audio_segment_template.attrib['media']).replace('$Time$', '{}')

defdownload_init(self):
print('Download init segments...')
r=self._download(self.video_init, save_path=self.save_path_video)
assertrin [200, 'Exists']
r=self._download(self.audio_init, save_path=self.save_path_audio)
assertrin [200, 'Exists']

def_get_segments(self):
mpd=self.fetch_mpd(mute=True) # fetch a new mpd to get the latest segments
video_representation=mpd[0][0][self.video_index]
assertvideo_representation.attrib['id'] ==self.video_id
timeline=video_representation[0][0]
segments= [int(timeline[i].attrib['t']) foriinrange(len(timeline))]
returnsegments, mpd

defmanually_set(self, last_t=None):
iflast_tisnotNone:
self.last_t=int(last_t)
print(f'Last time set to {self.last_t}')

deffetch_video_by_id(self, id):
url=self.video_url_template.format(id)
status=self._download(url, save_path=self.save_path_video)
returnstatus

deffetch_audio_by_id(self, id):
url=self.audio_url_template.format(id)
status=self._download(url, save_path=self.save_path_audio)
returnstatus

defquick_iterate(self, ids):
# make sure ids are larger than 0
ids= [idforidinidsifid>0]
print(f'\nUse multi-threading to check {len(ids)} IDs starting from {ids[0]}...')
withconcurrent.futures.ThreadPoolExecutor(max_workers=20) asex:
futures= {ex.submit(self.fetch_video_by_id, id): idforidinids}
try:
forfutureinconcurrent.futures.as_completed(futures):
iffuture.cancelled():
continue
id=futures[future]
status=future.result()
ifstatusin ['Exists', 200]:
# cancel all other futures
forfinfutures:
f.cancel()
returnid, status
# TODO: does not work since we call it in a thread.
exceptKeyboardInterrupt:
print('\nInterrupted. Cancel all futures...')
forfinfutures:
f.cancel()
raiseKeyboardInterrupt
returnNone, None

defdownload_live(self):
downloaded=set()

mpd=self.mpd
# check if mpd is dynamic
ifmpd.attrib.get('type') !='dynamic':
print('This mpd is not dynamic. Stop monitoring live.')
return
# use minimumUpdatePeriod if available. otherwise, use timeShiftBufferDepth/2-1 as the interval.
# make sure it is at least 2s.
fetch_interval=parse_iso8601_duration(mpd.attrib.get('minimumUpdatePeriod', 'PT0S')) \
orparse_iso8601_duration(mpd.attrib.get('timeShiftBufferDepth', 'PT0S')) //2-1
fetch_interval=max(fetch_interval, 2)

MAX_IDLE_COUNT=20

unchanged_mpd_count=0
whileTrue:
new_segments, mpd=self._get_segments()
ifall(idindownloadedforidinnew_segments):
unchanged_mpd_count+=1
ifunchanged_mpd_count>=MAX_IDLE_COUNT:
ifmpd.attrib.get('type') =='dynamic':
print(f'No new segments found in the last {MAX_IDLE_COUNT} checks. But the mpd is still dynamic. Continue monitoring...')
else:
print('All segments are downloaded. Stop.')
return
else:
undownloaded= [idforidinnew_segmentsifidnotindownloaded]
print(f'{len(undownloaded)} new segments found. Downloading...')
foridinundownloaded:
# singe-thread should be enough for live stream
self.fetch_video_by_id(id)
self.fetch_audio_by_id(id)
downloaded.add(id)

time.sleep(fetch_interval)

defdownload_video(self, forward=False):
count=0
known_intervals= {
self.interval: 0,
self.interval-1: 0,
self.interval+1: 0,
 }
id_guesses= [self.last_t]
prev_id=None
sign=1ifforwardelse-1

defsurrounding(x):
start_id=x-int(self.interval* (0.5+TOLERANCE)) *sign
end_id=x+int(self.interval* (1.0+TOLERANCE) +1) *sign
ids=list(range(start_id, end_id, -1ifstart_id>end_idelse1))
ids.sort(key=lambdak: abs(k-x))
returnids

whileTrue:
valid_id=None
# firstly, we try to find if existing local file that is close to the guesses[0].
# which is defined as +/- TOLERANCE (default: 20%) of the interval.
# this is a "narrower" window than surrounding() function.
# notice that we don't need to try all the guesses, they will be checked in the
# next step.
ids=list(range(id_guesses[0] -int(self.interval*TOLERANCE), id_guesses[0] +int(self.interval*TOLERANCE) +1))
ids.sort(key=lambdak: abs(k-id_guesses[0]))
# only try ids that are before the prev_id (if backward) or after the prev_id (if forward).
ifprev_id:
ids= [idforidinidsif (id-prev_id) *sign>0]
forcandidateinids:
url=self.video_url_template.format(candidate)
f=self.save_path_video/get_webname(url)
iff.exists() andf.stat().st_size>0:
valid_id=candidate
print_full_width(f'Segment {valid_id}: {f} already exists. Skip.')
break
# if not found locally, we try to fetch the segment by id from the all
# (last_valid_id - potential_interval) pools.
# The order matters: we try the most common interval first.
# this is single-threaded, because it is more time-consuming trying to close all the threads.
ifnotvalid_id:
forcandidateinid_guesses:
status=self.fetch_video_by_id(candidate)
print_full_width(f'Segment {candidate}: HTTP {status}')
ifstatusin [200, 'Exists']:
valid_id=candidate
break
# If still not found, we iterate around the guesses[0] to find the next segment.
# the range is defined as 50%+TOLERANCE behind of x and 100%+TOLERANCE ahead of x,
# where x is the guesses[0].
# e.g. for 2000 interval, x = last_id - 2000, the range would be
# last_id - 800 to last_id - 4400, sorted by distance to last_id - 2000.
# notice that it covers up to the range of next next id, this way we ensure we still
# continue the downloading instead of stopping too early (despite missing a segment).
ifnotvalid_id:
ids= [idforidinsurrounding(id_guesses[0]) ifidnotinid_guesses]
ifprev_id:
ids= [idforidinidsif (id-prev_id) *sign>0]
valid_id, status=self.quick_iterate(ids)
# if still not, we assume we downloaded them all and stop.
ifnotvalid_id:
print("\nFailed to find next segment. Assume we downloaded all. Stop.")
break
print_full_width(f'Segment {valid_id}: HTTP {status}')
# at this point, we should have a valid_id.
assertvalid_id
# add new interval to known_intervals
ifprev_id:
new_interval=abs(valid_id-prev_id)
assertnew_interval>0# this should not happen.
# do not add new interval if it is too different from the current interval.
# for example, if the nominal interval is 2000, we should only add ones that are
# less than 3000. Otherwise we enables the possibility of skipping segments.
# But it is still allowed to have such large interval so we don't stop downloading
# in the middle just because of one missing segment.
ifnew_interval>=self.interval*1.6:
pass
# add interval to known_intervals (if not already), and increase the count.
else:
known_intervals.setdefault(new_interval, 0)
known_intervals[new_interval] +=1

# print(f'[Debug] known_intervals: {known_intervals}')
count+=1
ifself.debugandcount==30:
break
prev_id=valid_id
# sort known_intervals by appearance, so we try the most common interval first.
known_intervals=dict(sorted(known_intervals.items(), key=lambdax: x[1], reverse=True))
id_guesses= [valid_id+interval*signforintervalinknown_intervals]

defcheck(self):
print('Check if there is any missing video segment...')
files=list(self.save_path_video.iterdir())
# filename format: 17981336783244063_0-1297029.m4v
ids= [int(m[1]) forfinfilesif (m:=re.search(r'\d+_0-(\d+)', f.name))]
ids.sort()
# calculate difference between each id
diffs= [ids[i+1] -ids[i] foriinrange(len(ids)-1)]

diff_count= {}
fordiffindiffs:
diff_count[diff] =diff_count.get(diff, 0) +1
print(f'Count of diff values between each segment:', diff_count)
max_idx=diffs.index(max(diffs))
print(f'Largest ID diff between each segment: {max(diffs)}, at {ids[max_idx]} -> {ids[max_idx+1]}')
min_idx=diffs.index(min(diffs))
print(f'Smallest ID diff between each segment: {min(diffs)}, at {ids[min_idx]} -> {ids[min_idx+1]}')

# Check if there are any missing segments, defined as gaps larger than 1.2 times the interval.
ifmax(diffs) >self.interval* (1+TOLERANCE):
print(f'It is likely that the video is not fully downloaded!!')
return (ids[max_idx], ids[max_idx+1])
else:
print('No missing segment found.')


defdownload_audio(self):
print('Downloading audio segments...')
files=list(self.save_path_video.iterdir())
withconcurrent.futures.ThreadPoolExecutor(max_workers=20) asex:
futures= []
count=0
forfinfiles:
ifnotf.is_file():
continue
iff.suffix!='.m4v':
continue
id=re.search(r'_0-(init|\d+)\.m4v', f.name)[1]
url=self.audio_url_template.format(id)
futures.append(ex.submit(self._download, url, save_path=self.save_path_audio))
for_inconcurrent.futures.as_completed(futures):
count+=1
print(f'Finished {count}/{len(futures)} ', end='\r')

defmerge(self):
defget_key(f):
'''Make sure init segment is always at the beginning.'''
if'-init'inf.name:
return0
returnint(re.search(r'\d+_0-(\d+)', f.name)[1])

video_file=self.save_path/'video.m4v'
files=list(self.save_path_video.iterdir())
print(f'Find {len(files)} video segments. Merging...')
files.sort(key=get_key)
concat(files, video_file)

audio_file=self.save_path/'audio.m4a'
files2=list(self.save_path_audio.iterdir())
print(f'Find {len(files2)} audio segments. Merging...')
files2.sort(key=get_key)
concat(files2, audio_file)

print(f'Merging video and audio using FFMPEG...')
run(['ffmpeg', '-loglevel', 'error', '-stats', '-i', video_file, '-i', audio_file, '-c', 'copy', self.save_path/'merged.mp4'])

defimport_segments(self, path):
'''Import segments downloaded via N_m3u8DL-RE.'''
path=Path(path)
forpinpath.iterdir():
ifp.is_dir() and (p/'_init.mp4').exists():
if'avc'inp.name:
template=self.video_url_template
save_path=self.save_path_video
else:
template=self.audio_url_template
save_path=self.save_path_audio
forfinp.iterdir():
iff.stem.isdigit():
id=int(f.stem)
new_filename=get_webname(template.format(id))
newf=save_path/new_filename
ifnewf.exists():
ifnewf.stat().st_size==f.stat().st_size:
print(f'{id}: {newf} already exists. Skip.')
else:
raiseException(f'{id}: {newf} already exists but size is different.')
else:
print(f'{id}: Copy {f} to {newf}')
copy2(f, newf)


defmain(url, save_path, time, debug, action, quality):
downloader=InstaliveDownloader(url=url, save_path=save_path, debug=debug, quality=quality)
iftimeisnotNone:
downloader.manually_set(time)
ifaction.startswith('import'):
import_path=action.partition(':')[2]
downloader.import_segments(import_path)
return
ifaction=='info':
fork, vindownloader.__dict__.items():
ifk=='mpd_text': # too long.
v=v[:10] +'...'+v[-10:]
print(f'{k}: {v}')
return
try:
ifaction=='all':
downloader.save_mpd()
downloader.download_init()
deftask1():
downloader.download_video()
downloader.download_audio()
t1=threading.Thread(target=task1)
t2=threading.Thread(target=downloader.download_live)
t1.start()
t2.start()
t1.join()
t2.join()
downloader.check()
downloader.merge()
elifaction=='live':
downloader.save_mpd()
downloader.download_live()
elifaction=='video':
downloader.save_mpd()
downloader.download_init()
downloader.download_video()
elifaction=='audio':
downloader.download_audio()
elifaction=='merge':
downloader.merge()
elifaction=='check':
downloader.check()
elifaction=='manual':
ifargs.range:
start, end=map(int, args.range.split('-'))
print(f'manually set range to from {start} to {end} (inclusive)')
else:
worst=downloader.check()
ifnotworst:
print('No bad interval found. Stop.')
return
start=worst[0] +1
end=worst[1] -1
print(f'Automatically check the largest interval, between {start} and {end} (inclusive)')
ids=list(range(start, end+1)) ifstart<endelselist(range(start, end-1, -1))
id, status=downloader.quick_iterate(ids)
ifnotid:
print('No segment found in this range.')
else:
print(f'Found segment {id} with status {status}')

exceptKeyboardInterrupt:
print('\nInterrupted by user. Stop.')


if__name__=='__main__':
importargparse

parser=argparse.ArgumentParser(
description='Available actions:\n'
' all - Download both video and audio (including live and backtracking), and then merge them (default)\n'
' live - Download the live stream only (no backtracking)\n'
' video - Download video only\n'
' audio - Download audio only\n'
' merge - Merge downloaded video and audio\n'
' check - Check the downloaded segments to make sure there are no missing segments\n'
' manual - Manually check missing segments at the largest gap (or use --range to assign a range) (for debugging only)\n'
' info - Display downloader object info\n'
' import:<path> - Import segments downloaded via N_m3u8DL-RE from a given path',
formatter_class=argparse.RawTextHelpFormatter
 )

parser.add_argument("url", help="url of mpd")
parser.add_argument("--action", '-a', default='all', help="action to perform (default: all)")
parser.add_argument("--dir", "-d", help="save path (default: instalive_{mpd_id})")
parser.add_argument("--debug", action='store_true', help="debug mode")
parser.add_argument("--quality", "-q", default='pst', help="manually assign video quality by quality name.\n"
"(default: \"pst\": which is the \"original\" (?) and has the best bitrate,\n"
"but not necessarily the highest resolution.\n"
"Pass empty string or \"highest\" to use the highest resolution one.)")
parser.add_argument("--time", "-t", help="for debugging only; manually assign last t (default: auto)")
parser.add_argument('--range', help='for debugging only; manually assign iteration range (start,end) for manual action')

args=parser.parse_args()

main(args.url, args.dir, args.time, args.debug, args.action, args.quality)
importconcurrent.futures
importre
importthreading
importtime
importxml.etree.ElementTreeasET
fromdatetimeimportdatetime
frompathlibimportPath
fromshutilimportcopy2, copyfileobj, get_terminal_size
fromsubprocessimportrun
fromurllib.parseimporturljoin

fromtqdmimporttqdm
fromutilimportrequests_retry_session, get_webname, td_format


TOLERANCE=0.2


defparse_iso8601_duration(duration):
'''Parse ISO8601 duration to seconds.
 it sometimes uses float for seconds part, so we need to handle that.'''
pattern=r'P(?:(?P<days>\d+)D)?(?:T(?:(?P<hours>\d+)H)?(?:(?P<minutes>\d+)M)?(?:(?P<seconds>[.\d]+)S)?)?'
match=re.match(pattern, duration)
ifnotmatch:
returnNone

days=int(match.group('days') or0)
hours=int(match.group('hours') or0)
minutes=int(match.group('minutes') or0)
seconds=int(float(match.group('seconds') or0))

# Convert everything to seconds
total_seconds= (days*86400) + (hours*3600) + (minutes*60) +seconds
returntotal_seconds


defprint_full_width(s):
'''Print a string to full width of the terminal.'''
width=get_terminal_size().columns
print('\r'+s.ljust(width), end='')


defconcat(files, output, verbose=False):
'''Concatenate files into one file.'''
output=Path(output)
ifnotoutput.parent.exists():
output.parent.mkdir(parents=True)
out=output.open('wb')
forfintqdm(files, ncols=100):
f=Path(f)
ifverbose:
print(f'Merging {f.name}...')
fi=f.open('rb')
copyfileobj(fi, out)
fi.close()
out.close()


classInstaliveDownloader:
def__init__(self, url, save_path, debug=False, quality=None):
self.session=requests_retry_session()
self.url=url
ifsave_pathisNone:
mpd_name=get_webname(url).split('.')[0]
save_path=f'instalive_{mpd_name}'
self.save_path=Path(save_path)
ifnotself.save_path.exists():
self.save_path.mkdir(parents=True)
assertself.save_path.is_dir(), f'{self.save_path} is not a directory!'
self.debug=debug
ifdebug:
print('Debug mode enabled. Only download 30 segments.')
self.fetch_mpd()
self.parse_mpd(quality)

def_download(self, url, save_path=None, filename=None, skip_existing=True):
ifsave_path:
f=Path(save_path) /get_webname(url)
eliffilename:
f=Path(filename)
ifskip_existingandf.exists() andf.stat().st_size>0:
return'Exists'
f.parent.mkdir(parents=True, exist_ok=True)
withself.session.get(url) asr:
ifr.status_code==200:
withf.open('wb') asf:
f.write(r.content)
returnr.status_code

defsave_mpd(self):
print('Save mpd to local file...')
mpd_lines=self.mpd_text.splitlines(keepends=True)
comment=f'<!-- {self.url} -->\n'
ifmpd_linesandmpd_lines[0].strip().startswith('<?xml'):
mpd_lines.insert(1, comment)
else:
mpd_lines.insert(0, comment)
 (self.save_path/'mpd.mpd').write_text(''.join(mpd_lines), encoding='utf-8')

deffetch_mpd(self, mute=False):
ifnotmute:
print('Fetch mpd...')
self.mpd_text=self.session.get(self.url).text
self.mpd=ET.fromstring(self.mpd_text)
returnself.mpd

defparse_mpd(self, quality=None):
print('Parse mpd...')
mpd=self.mpd
# availabilityStartTime="2024-05-06T01:21:00-07:00"
# availabilityEndTime="2024-05-06T01:29:00-07:00"

deftd_format_with_direction(td):
'''a wrapper around td_format to show both positive and negative time deltas'''
iftd.total_seconds() <0:
returntd_format(-td) +" ago"
else:
return'in '+td_format(td)
current_time=datetime.now().astimezone()
availability_start_time=datetime.strptime(mpd.attrib['availabilityStartTime'], '%Y-%m-%dT%H:%M:%S%z')
availability_end_time=datetime.strptime(mpd.attrib['availabilityEndTime'], '%Y-%m-%dT%H:%M:%S%z')

print(f'Availability start time: {availability_start_time} ({td_format_with_direction(availability_start_time-current_time)})')
print(f'Availability end time: {availability_end_time} ({td_format_with_direction(availability_end_time-current_time)})')

# process video representations
period=mpd[0]
video_adaptation_set=period[0]
# make a simple list of video representations because we can't directly sort xml elements
videos= [
 {
'width': int(x.attrib['width']),
'height': int(x.attrib['height']),
'frame_rate': float(x.attrib['frameRate']),
'bandwidth': float(x.attrib['bandwidth']),
'idx': idx,
'id': x.attrib['id']
 } foridx, xinenumerate(video_adaptation_set)
 ]

print('Video representations:')
forvinvideos:
print(f'[{v["idx"]}] {v["id"]}{v["width"]}x{v["height"]}, {v["frame_rate"]}fps, {v["bandwidth"]/1024:.1f} kbps')

ifnotqualityorquality=='highest':
videos.sort(reverse=True, key=lambdax: (x['width'], x['height'], x['frame_rate'], x['bandwidth']))
self.video_index=videos[0]['idx']
self.video_id=videos[0]['id']
print(f'Use the highest one ([{self.video_index}] {self.video_id}).')
else:
forvinvideos:
# quality could be either "dash-lp-pst-v" or "pst"
ifquality==v['id'] orf'{str(quality)}-v'inv['id']:
self.video_index=v['idx']
self.video_id=v['id']
print(f'Use the specified one ([{self.video_index}] {self.video_id}).')
break
else:
raiseException(f'Quality {quality} not found.')

self.save_path_video=self.save_path/self.video_id

video_representation=video_adaptation_set[self.video_index]
video_segment_template=video_representation[0]
self.timescale=video_segment_template.attrib['timescale'] # not used for now
self.video_init=urljoin(self.url, video_segment_template.attrib['initialization'])
self.video_url_template=urljoin(self.url, video_segment_template.attrib['media']).replace('$Time$', '{}')
timeline=video_segment_template[0]
# find the last segment's t
self.last_t=int(timeline[-1].attrib['t'])
print(f'Last video segment t={self.last_t}')
# find the best interval for iterating heuristically
d_list= [int(x.attrib['d']) forxintimeline]
print('Intervals between segments:', ', '.join(str(d) fordind_list))
# get most common interval
self.interval=max(set(d_list), key=d_list.count)
print(f'Heuristically set interval to {self.interval}')

# process audio representations
audio_adaptation_set=period[1]
audios= [
 {
'id': x.attrib['id'],
'bandwidth': float(x.attrib['bandwidth']),
'sampling_rate': int(x.attrib['audioSamplingRate']),
'idx': idx
 } foridx, xinenumerate(audio_adaptation_set)
 ]
print('Audio representations:')
forainaudios:
print(f'[{a["idx"]}] {a["id"]}, {a["bandwidth"]/1024:.1f} kbps, {a["sampling_rate"]} Hz')

audios.sort(reverse=True, key=lambdax: (x['bandwidth'], x['sampling_rate']))
self.audio_index=audios[0]['idx']
self.audio_id=audios[0]['id']
print(f'Use the best one ([{self.audio_index}] {self.audio_id}).')
self.save_path_audio=self.save_path/self.audio_id

audio_representation=audio_adaptation_set[self.audio_index]
audio_segment_template=audio_representation[0]

self.audio_init=urljoin(self.url, audio_segment_template.attrib['initialization'])
self.audio_url_template=urljoin(self.url, audio_segment_template.attrib['media']).replace('$Time$', '{}')

defdownload_init(self):
print('Download init segments...')
r=self._download(self.video_init, save_path=self.save_path_video)
assertrin [200, 'Exists']
r=self._download(self.audio_init, save_path=self.save_path_audio)
assertrin [200, 'Exists']

def_get_segments(self):
mpd=self.fetch_mpd(mute=True) # fetch a new mpd to get the latest segments
video_representation=mpd[0][0][self.video_index]
assertvideo_representation.attrib['id'] ==self.video_id
timeline=video_representation[0][0]
segments= [int(timeline[i].attrib['t']) foriinrange(len(timeline))]
returnsegments, mpd

defmanually_set(self, last_t=None):
iflast_tisnotNone:
self.last_t=int(last_t)
print(f'Last time set to {self.last_t}')

deffetch_video_by_id(self, id):
url=self.video_url_template.format(id)
status=self._download(url, save_path=self.save_path_video)
returnstatus

deffetch_audio_by_id(self, id):
url=self.audio_url_template.format(id)
status=self._download(url, save_path=self.save_path_audio)
returnstatus

defquick_iterate(self, ids):
# make sure ids are larger than 0
ids= [idforidinidsifid>0]
print(f'\nUse multi-threading to check {len(ids)} IDs starting from {ids[0]}...')
withconcurrent.futures.ThreadPoolExecutor(max_workers=20) asex:
futures= {ex.submit(self.fetch_video_by_id, id): idforidinids}
try:
forfutureinconcurrent.futures.as_completed(futures):
iffuture.cancelled():
continue
id=futures[future]
status=future.result()
ifstatusin ['Exists', 200]:
# cancel all other futures
forfinfutures:
f.cancel()
returnid, status
# TODO: does not work since we call it in a thread.
exceptKeyboardInterrupt:
print('\nInterrupted. Cancel all futures...')
forfinfutures:
f.cancel()
raiseKeyboardInterrupt
returnNone, None

defdownload_live(self):
downloaded=set()

mpd=self.mpd
# check if mpd is dynamic
ifmpd.attrib.get('type') !='dynamic':
print('This mpd is not dynamic. Stop monitoring live.')
return
# use minimumUpdatePeriod if available. otherwise, use timeShiftBufferDepth/2-1 as the interval.
# make sure it is at least 2s.
fetch_interval=parse_iso8601_duration(mpd.attrib.get('minimumUpdatePeriod', 'PT0S')) \
orparse_iso8601_duration(mpd.attrib.get('timeShiftBufferDepth', 'PT0S')) //2-1
fetch_interval=max(fetch_interval, 2)

MAX_IDLE_COUNT=20

unchanged_mpd_count=0
whileTrue:
new_segments, mpd=self._get_segments()
ifall(idindownloadedforidinnew_segments):
unchanged_mpd_count+=1
ifunchanged_mpd_count>=MAX_IDLE_COUNT:
ifmpd.attrib.get('type') =='dynamic':
print(f'No new segments found in the last {MAX_IDLE_COUNT} checks. But the mpd is still dynamic. Continue monitoring...')
else:
print('All segments are downloaded. Stop.')
return
else:
undownloaded= [idforidinnew_segmentsifidnotindownloaded]
print(f'{len(undownloaded)} new segments found. Downloading...')
foridinundownloaded:
# singe-thread should be enough for live stream
self.fetch_video_by_id(id)
self.fetch_audio_by_id(id)
downloaded.add(id)

time.sleep(fetch_interval)

defdownload_video(self, forward=False):
count=0
known_intervals= {
self.interval: 0,
self.interval-1: 0,
self.interval+1: 0,
 }
id_guesses= [self.last_t]
prev_id=None
sign=1ifforwardelse-1

defsurrounding(x):
start_id=x-int(self.interval* (0.5+TOLERANCE)) *sign
end_id=x+int(self.interval* (1.0+TOLERANCE) +1) *sign
ids=list(range(start_id, end_id, -1ifstart_id>end_idelse1))
ids.sort(key=lambdak: abs(k-x))
returnids

whileTrue:
valid_id=None
# firstly, we try to find if existing local file that is close to the guesses[0].
# which is defined as +/- TOLERANCE (default: 20%) of the interval.
# this is a "narrower" window than surrounding() function.
# notice that we don't need to try all the guesses, they will be checked in the
# next step.
ids=list(range(id_guesses[0] -int(self.interval*TOLERANCE), id_guesses[0] +int(self.interval*TOLERANCE) +1))
ids.sort(key=lambdak: abs(k-id_guesses[0]))
# only try ids that are before the prev_id (if backward) or after the prev_id (if forward).
ifprev_id:
ids= [idforidinidsif (id-prev_id) *sign>0]
forcandidateinids:
url=self.video_url_template.format(candidate)
f=self.save_path_video/get_webname(url)
iff.exists() andf.stat().st_size>0:
valid_id=candidate
print_full_width(f'Segment {valid_id}: {f} already exists. Skip.')
break
# if not found locally, we try to fetch the segment by id from the all
# (last_valid_id - potential_interval) pools.
# The order matters: we try the most common interval first.
# this is single-threaded, because it is more time-consuming trying to close all the threads.
ifnotvalid_id:
forcandidateinid_guesses:
status=self.fetch_video_by_id(candidate)
print_full_width(f'Segment {candidate}: HTTP {status}')
ifstatusin [200, 'Exists']:
valid_id=candidate
break
# If still not found, we iterate around the guesses[0] to find the next segment.
# the range is defined as 50%+TOLERANCE behind of x and 100%+TOLERANCE ahead of x,
# where x is the guesses[0].
# e.g. for 2000 interval, x = last_id - 2000, the range would be
# last_id - 800 to last_id - 4400, sorted by distance to last_id - 2000.
# notice that it covers up to the range of next next id, this way we ensure we still
# continue the downloading instead of stopping too early (despite missing a segment).
ifnotvalid_id:
ids= [idforidinsurrounding(id_guesses[0]) ifidnotinid_guesses]
ifprev_id:
ids= [idforidinidsif (id-prev_id) *sign>0]
valid_id, status=self.quick_iterate(ids)
# if still not, we assume we downloaded them all and stop.
ifnotvalid_id:
print("\nFailed to find next segment. Assume we downloaded all. Stop.")
break
print_full_width(f'Segment {valid_id}: HTTP {status}')
# at this point, we should have a valid_id.
assertvalid_id
# add new interval to known_intervals
ifprev_id:
new_interval=abs(valid_id-prev_id)
assertnew_interval>0# this should not happen.
# do not add new interval if it is too different from the current interval.
# for example, if the nominal interval is 2000, we should only add ones that are
# less than 3000. Otherwise we enables the possibility of skipping segments.
# But it is still allowed to have such large interval so we don't stop downloading
# in the middle just because of one missing segment.
ifnew_interval>=self.interval*1.6:
pass
# add interval to known_intervals (if not already), and increase the count.
else:
known_intervals.setdefault(new_interval, 0)
known_intervals[new_interval] +=1

# print(f'[Debug] known_intervals: {known_intervals}')
count+=1
ifself.debugandcount==30:
break
prev_id=valid_id
# sort known_intervals by appearance, so we try the most common interval first.
known_intervals=dict(sorted(known_intervals.items(), key=lambdax: x[1], reverse=True))
id_guesses= [valid_id+interval*signforintervalinknown_intervals]

defcheck(self):
print('Check if there is any missing video segment...')
files=list(self.save_path_video.iterdir())
# filename format: 17981336783244063_0-1297029.m4v
ids= [int(m[1]) forfinfilesif (m:=re.search(r'\d+_0-(\d+)', f.name))]
ids.sort()
# calculate difference between each id
diffs= [ids[i+1] -ids[i] foriinrange(len(ids)-1)]

diff_count= {}
fordiffindiffs:
diff_count[diff] =diff_count.get(diff, 0) +1
print(f'Count of diff values between each segment:', diff_count)
max_idx=diffs.index(max(diffs))
print(f'Largest ID diff between each segment: {max(diffs)}, at {ids[max_idx]} -> {ids[max_idx+1]}')
min_idx=diffs.index(min(diffs))
print(f'Smallest ID diff between each segment: {min(diffs)}, at {ids[min_idx]} -> {ids[min_idx+1]}')

# Check if there are any missing segments, defined as gaps larger than 1.2 times the interval.
ifmax(diffs) >self.interval* (1+TOLERANCE):
print(f'It is likely that the video is not fully downloaded!!')
return (ids[max_idx], ids[max_idx+1])
else:
print('No missing segment found.')


defdownload_audio(self):
print('Downloading audio segments...')
files=list(self.save_path_video.iterdir())
withconcurrent.futures.ThreadPoolExecutor(max_workers=20) asex:
futures= []
count=0
forfinfiles:
ifnotf.is_file():
continue
iff.suffix!='.m4v':
continue
id=re.search(r'_0-(init|\d+)\.m4v', f.name)[1]
url=self.audio_url_template.format(id)
futures.append(ex.submit(self._download, url, save_path=self.save_path_audio))
for_inconcurrent.futures.as_completed(futures):
count+=1
print(f'Finished {count}/{len(futures)} ', end='\r')

defmerge(self):
defget_key(f):
'''Make sure init segment is always at the beginning.'''
if'-init'inf.name:
return0
returnint(re.search(r'\d+_0-(\d+)', f.name)[1])

video_file=self.save_path/'video.m4v'
files=list(self.save_path_video.iterdir())
print(f'Find {len(files)} video segments. Merging...')
files.sort(key=get_key)
concat(files, video_file)

audio_file=self.save_path/'audio.m4a'
files2=list(self.save_path_audio.iterdir())
print(f'Find {len(files2)} audio segments. Merging...')
files2.sort(key=get_key)
concat(files2, audio_file)

print(f'Merging video and audio using FFMPEG...')
run(['ffmpeg', '-loglevel', 'error', '-stats', '-i', video_file, '-i', audio_file, '-c', 'copy', self.save_path/'merged.mp4'])

defimport_segments(self, path):
'''Import segments downloaded via N_m3u8DL-RE.'''
path=Path(path)
forpinpath.iterdir():
ifp.is_dir() and (p/'_init.mp4').exists():
if'avc'inp.name:
template=self.video_url_template
save_path=self.save_path_video
else:
template=self.audio_url_template
save_path=self.save_path_audio
forfinp.iterdir():
iff.stem.isdigit():
id=int(f.stem)
new_filename=get_webname(template.format(id))
newf=save_path/new_filename
ifnewf.exists():
ifnewf.stat().st_size==f.stat().st_size:
print(f'{id}: {newf} already exists. Skip.')
else:
raiseException(f'{id}: {newf} already exists but size is different.')
else:
print(f'{id}: Copy {f} to {newf}')
copy2(f, newf)


defmain(url, save_path, time, debug, action, quality):
downloader=InstaliveDownloader(url=url, save_path=save_path, debug=debug, quality=quality)
iftimeisnotNone:
downloader.manually_set(time)
ifaction.startswith('import'):
import_path=action.partition(':')[2]
downloader.import_segments(import_path)
return
ifaction=='info':
fork, vindownloader.__dict__.items():
ifk=='mpd_text': # too long.
v=v[:10] +'...'+v[-10:]
print(f'{k}: {v}')
return
try:
ifaction=='all':
downloader.save_mpd()
downloader.download_init()
deftask1():
downloader.download_video()
downloader.download_audio()
t1=threading.Thread(target=task1)
t2=threading.Thread(target=downloader.download_live)
t1.start()
t2.start()
t1.join()
t2.join()
downloader.check()
downloader.merge()
elifaction=='live':
downloader.save_mpd()
downloader.download_live()
elifaction=='video':
downloader.save_mpd()
downloader.download_init()
downloader.download_video()
elifaction=='audio':
downloader.download_audio()
elifaction=='merge':
downloader.merge()
elifaction=='check':
downloader.check()
elifaction=='manual':
ifargs.range:
start, end=map(int, args.range.split('-'))
print(f'manually set range to from {start} to {end} (inclusive)')
else:
worst=downloader.check()
ifnotworst:
print('No bad interval found. Stop.')
return
start=worst[0] +1
end=worst[1] -1
print(f'Automatically check the largest interval, between {start} and {end} (inclusive)')
ids=list(range(start, end+1)) ifstart<endelselist(range(start, end-1, -1))
id, status=downloader.quick_iterate(ids)
ifnotid:
print('No segment found in this range.')
else:
print(f'Found segment {id} with status {status}')

exceptKeyboardInterrupt:
print('\nInterrupted by user. Stop.')


if__name__=='__main__':
importargparse

parser=argparse.ArgumentParser(
description='Available actions:\n'
' all - Download both video and audio (including live and backtracking), and then merge them (default)\n'
' live - Download the live stream only (no backtracking)\n'
' video - Download video only\n'
' audio - Download audio only\n'
' merge - Merge downloaded video and audio\n'
' check - Check the downloaded segments to make sure there are no missing segments\n'
' manual - Manually check missing segments at the largest gap (or use --range to assign a range) (for debugging only)\n'
' info - Display downloader object info\n'
' import:<path> - Import segments downloaded via N_m3u8DL-RE from a given path',
formatter_class=argparse.RawTextHelpFormatter
 )

parser.add_argument("url", help="url of mpd")
parser.add_argument("--action", '-a', default='all', help="action to perform (default: all)")
parser.add_argument("--dir", "-d", help="save path (default: instalive_{mpd_id})")
parser.add_argument("--debug", action='store_true', help="debug mode")
parser.add_argument("--quality", "-q", default='pst', help="manually assign video quality by quality name.\n"
"(default: \"pst\": which is the \"original\" (?) and has the best bitrate,\n"
"but not necessarily the highest resolution.\n"
"Pass empty string or \"highest\" to use the highest resolution one.)")
parser.add_argument("--time", "-t", help="for debugging only; manually assign last t (default: auto)")
parser.add_argument('--range', help='for debugging only; manually assign iteration range (start,end) for manual action')

args=parser.parse_args()

main(args.url, args.dir, args.time, args.debug, args.action, args.quality)