import requests
import os
import datetime
import threading
class xiazai():
def __init__(self,url):
self.url = url
work_dir = os.getcwd()
# print(work_dir)
# 用来保存ts文件
file_dir = os.path.join(work_dir, file_tmp)
if not os.path.exists(file_dir):
os.mkdir(file_dir)
self.headers ={
user-agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.116 Safari/537.36
}
self.savefile(self.url)
def savefile(self, file_url):
r = requests.get(file_url, headers=self.headers)
# 合成带有hls的m3u8地址
if r.text.split(\n)[–1] == :
hls_mark = r.text.split(\n)[–2] # 以防\n结尾
else:
hls_mark = r.text.split(\n)[–1]
self.url_m3u8_hls = file_url.replace(index.m3u8, hls_mark)
#file_m3u8 = url_m3u8_hls.split(/)[-1]
self.duqu()
#print(url_m3u8_hls)
def duqu(self):
r = requests.get(self.url_m3u8_hls, headers=self.headers).text
text_bytes = r.split(\n)
# 筛选以.ts结尾的行
# 有些情况下可能是以其他格式的文件，比如png，下载后修改后缀即可
# ts_name = [i for i in text_string if i.endswith(.ts)]
self.ts_time = [i for i in text_bytes if i.startswith(#EXTINF)]
#self.shijian(dm_time)
#print(dm_time)
self.ts_neirong = [i for i in text_bytes if not i.startswith(#)]
self.ts_neirong.pop()
self.threads = []
self.threads.append(threading.Thread(target=self.xiazai))
self.threads.append(threading.Thread(target=self.shijian))
for t in self.threads:
# print(t)
t.start()
#self.xiazai(url_m3u8_hls)
# print(ts_neirong)
def shijian(self):
self.dm_time = 0
for i in range(len(self.ts_time)):
ts_time1 = self.ts_time[i].replace(#EXTINF:, )
ts_time2 = ts_time1.replace(,, )
self.dm_time = float(ts_time2) + self.dm_time
shichang_time = str(datetime.timedelta(seconds=self.dm_time))
print(视频时长：%s % shichang_time)
def xiazai(self):
liebiao=[]
for i in range(len(self.ts_neirong)):
hls_mark = self.url_m3u8_hls.split(/)[–1]
url_xiazai = self.url_m3u8_hls.replace(hls_mark, self.ts_neirong[i])
liebiao.append(url_xiazai)
#print(url_xiazai)
# r = requests.get(url_xiazai, headers=self.headers)
# with open(file_tmp/+ts_neirong[i], wb) as f:
# f.write(r.content)
# f.close()
x = self.bisector_list(liebiao, 10)
self.xiancheng0=x[0]
self.xiancheng1=x[1]
self.xiancheng2=x[2]
self.xiancheng3=x[3]
self.xiancheng4=x[4]
self.xiancheng5=x[5]
self.xiancheng6=x[6]
self.xiancheng7=x[7]
self.xiancheng8=x[8]
self.xiancheng9=x[9]
self.threads2 = []
self.threads2.append(threading.Thread(target=self.xiancheng_xiazai1))
self.threads2.append(threading.Thread(target=self.xiancheng_xiazai2))
self.threads2.append(threading.Thread(target=self.xiancheng_xiazai3))
self.threads2.append(threading.Thread(target=self.xiancheng_xiazai4))
self.threads2.append(threading.Thread(target=self.xiancheng_xiazai5))
self.threads2.append(threading.Thread(target=self.xiancheng_xiazai6))
self.threads2.append(threading.Thread(target=self.xiancheng_xiazai7))
self.threads2.append(threading.Thread(target=self.xiancheng_xiazai8))
self.threads2.append(threading.Thread(target=self.xiancheng_xiazai9))
self.threads2.append(threading.Thread(target=self.xiancheng_xiazai10))
for t in self.threads2:
# print(t)
t.start()
def xiancheng_xiazai1(self):
#print(self.xiancheng0)
for i in self.xiancheng0:
#print(i)
r = requests.get(i, headers=self.headers)
mingzi = i.split(/)[–1]
with open(file_tmp/+mingzi, wb) as f:
f.write(r.content)
f.close()
def xiancheng_xiazai2(self):
#print(self.xiancheng1)
for i in self.xiancheng1:
#print(i)
r = requests.get(i, headers=self.headers)
mingzi= i.split(/)[–1]
with open(file_tmp/+mingzi, wb) as f:
f.write(r.content)
f.close()
def xiancheng_xiazai3(self):
#print(self.xiancheng2)
for i in self.xiancheng2:
#print(i)
r = requests.get(i, headers=self.headers)
mingzi = i.split(/)[–1]
with open(file_tmp/+mingzi, wb) as f:
f.write(r.content)
f.close()
def xiancheng_xiazai4(self):
#print(self.xiancheng3)
for i in self.xiancheng3:
#print(i)
r = requests.get(i, headers=self.headers)
mingzi = i.split(/)[–1]
with open(file_tmp/+mingzi, wb) as f:
f.write(r.content)
f.close()
def xiancheng_xiazai5(self):
#print(self.xiancheng4)
for i in self.xiancheng4:
#print(i)
r = requests.get(i, headers=self.headers)
mingzi = i.split(/)[–1]
with open(file_tmp/+mingzi, wb) as f:
f.write(r.content)
f.close()
def xiancheng_xiazai6(self):
#print(self.xiancheng5)
for i in self.xiancheng5:
#print(i)
r = requests.get(i, headers=self.headers)
mingzi = i.split(/)[–1]
with open(file_tmp/+mingzi, wb) as f:
f.write(r.content)
f.close()
def xiancheng_xiazai7(self):
#print(self.xiancheng6)
for i in self.xiancheng6:
#print(i)
r = requests.get(i, headers=self.headers)
mingzi = i.split(/)[–1]
with open(file_tmp/+mingzi, wb) as f:
f.write(r.content)
f.close()
def xiancheng_xiazai8(self):
#print(self.xiancheng7)
for i in self.xiancheng7:
#print(i)
r = requests.get(i, headers=self.headers)
mingzi = i.split(/)[–1]
with open(file_tmp/+mingzi, wb) as f:
f.write(r.content)
f.close()
def xiancheng_xiazai9(self):
#print(self.xiancheng8)
for i in self.xiancheng8:
#print(i)
r = requests.get(i, headers=self.headers)
mingzi = i.split(/)[–1]
with open(file_tmp/+mingzi, wb) as f:
f.write(r.content)
f.close()
def xiancheng_xiazai10(self):
#print(self.xiancheng9)
for i in self.xiancheng9:
#print(i)
r = requests.get(i, headers=self.headers)
mingzi = i.split(/)[–1]
with open(file_tmp/+mingzi, wb) as f:
f.write(r.content)
f.close()
def bisector_list(self,tabulation: list, num: int):
“””
将列表平均分成几份
:param tabulation: 列表
:param num: 份数
:return: 返回一个新的列表
“””
new_list = []
列表长度大于等于份数
if len(tabulation) >= num:
remainder:列表长度除以份数，取余
remainder = len(tabulation) % num
if remainder == 0:
merchant:列表长度除以分数
merchant = int(len(tabulation) / num)
将列表平均拆分
for i in range(1, num + 1):
if i == 1:
new_list.append(tabulation[:merchant])
else:
new_list.append(tabulation[(i – 1) * merchant:i * merchant])
return new_list
else:
merchant：列表长度除以分数 取商
merchant = int(len(tabulation) // num)
remainder:列表长度除以份数，取余
remainder = int(len(tabulation) % num)
将列表平均拆分
for i in range(1, num + 1):
if i == 1:
new_list.append(tabulation[:merchant])
else:
new_list.append(tabulation[(i – 1) * merchant:i * merchant])
将剩余数据的添加前面列表中
if int(len(tabulation) – i * merchant) <= merchant:
for j in tabulation[–remainder:]:
new_list[tabulation[–remainder:].index(j)].append(j)
return new_list
else:
如果列表长度小于份数
for i in range(1, len(tabulation) + 1):
tabulation_subset = []
tabulation_subset.append(tabulation[i – 1])
new_list.append(tabulation_subset)
return new_list
if __name__==__main__:
xiazai(http://iqiyi.cdn9-okzy.com/20200907/15137_ed25d8c5/index.m3u8)
速度很慢. 40m 5分钟 不加多线程 1小时。 可能我的m3u8不行，或者我电脑不行， 多线程是机械式的。 电脑好 可以多加几条。
以上就是python 下载m3u8视频的示例代码的详细内容。
