python下载m3u8视频
单线程版
import requests
import os
import re
import threading
# m3u8 url & vedio name
m3u8_url = "https://www.hyxrzs.com/20201217/Jt4nKiPm/index.m3u8"
base_url = "https://www.hyxrzs.com"
vedio_name = "超人总动员"
#创建用于合并的临时文件夹，用于存放ts文件
if not os.path.exists('merge'):
os.system('mkdir merge')
# 模拟浏览器header,防止误伤
headers = {'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/75.0.3770.100 Safari/537.36'}
# 获取m3u8文件的内容
m3u8_content = requests.get(m3u8_url,headers=headers).text
# 判断是否是最终的
is_final_m3u8 = not "m3u8" in m3u8_content
# 当 m3u8 作为主播放列表（Master Playlist）时，其内部提供的是同一份媒体资源的多份流列表资源（Variant Stream）
# 如果不是最终的m3u8,进入下一层找到最终的m3u8 （针对m3u8多分辨率适配的情况，会先有一个针对不同分辨率的m3u8索引文件）
if not is_final_m3u8:
# 解析出真正的m3u8_content
for m3u8_url in m3u8_content.split('\n'):
if "m3u8" in m3u8_url:
m3u8_url = base_url +m3u8_url
m3u8_content = requests.get(m3u8_url,headers=headers).text
break
m3u8_content_split_list = m3u8_content.split('\n')
# 判断视频是否经过AES-128加密，如果加密过则获取加密方式和加密秘钥
key = ''
for index,line in enumerate(m3u8_content_split_list):
# 判断视频是否经过AES-128加密
if "#EXT-X-KEY" in line:
#获取加密方式
method_pos = line.find("METHOD")
comma_pos = line.find(",")
method = line[method_pos:comma_pos].split('=')[1]
print("该视频经过加密，加密方式为：", method)
#获取加密密钥
uri_pos = line.find("URI")
quotation_mark_pos = line.rfind('"')
key_path = line[uri_pos:quotation_mark_pos].split('"')[1]
key_url = m3u8_url.replace("index.m3u8",key_path)
res = requests.get(key_url)
key = res.content
#从m3u8文件中解析出ts地址集
play_list = []
key = ''
for index,line in enumerate(m3u8_content_split_list):
#以下拼接方式可能会根据自己的需求进行改动
if '#EXTINF' in line:
# 如果加密，直接提取每一级的.ts文件链接地址
if 'http' in m3u8_content_split_list[index + 1]:
href = m3u8_content_split_list[index + 1]
play_list.append(href)
# 如果没有加密，直接构造出url链接
elif('ad0.ts' not in m3u8_content_split_list[index + 1]):
href = base_url + m3u8_content_split_list[index+1]
play_list.append(href)
print("m3u8文件解析成功，共解析到",len(play_list),"个ts文件")
# 封装下载当个ts文件的函数
def down_ts(index,ts_url,cryptor = False):
# 获取ts content
if not cryptor:
ts_content = requests.get(ts_url,headers=headers).content
else:
ts_content = cryptor.decrypt(requests.get(ts_url,headers=headers).content)#获取bing解密ts
# 写入文件
with open('merge/' + str(index+1) + '.ts','wb') as file:
file.write(ts_content)
print('第{}/{}个ts文件下载完成'.format(index+1,len(play_list)))
# 根据ts地址集下载ts文件
print("开始下载所有的ts文件")
if(len(key)):# 如果加密过
from Crypto.Cipher import AES
cryptor = AES.new(key, AES.MODE_CBC, key)
for index,ts_url in enumerate(play_list):
down_ts(index,ts_url,cryptor)
else: # 如果未加密
for index,ts_url in enumerate(play_list):
down_ts(index,ts_url)
print('所有ts文件都已下载完成')
# 合并ts文件为mp4 并删除下载的ts文件
merge_cmd = "cat "
for i in range(len(os.listdir("./merge"))):
merge_cmd += "merge/"+ str(i+1) +".ts "
merge_cmd += ">>vedio/"+vedio_name+".mp4"
del_cmd = 'rm merge/*.ts'
os.system(merge_cmd)#执行合并命令
os.system(del_cmd)#执行删除命令
print(vedio_name,'.mp4下载已完成')
多线程版
备注：目前该版本还存在问题(频繁下载被服务器断开连接),如过未下载完成则多次重复执行即可
import requests
import os
import re
import threading
# m3u8 url & vedio name
m3u8_url = "https://www.hyxrzs.com/20201217/Jt4nKiPm/index.m3u8"
base_url = "https://www.hyxrzs.com"
vedio_name = "超人总动员"
#使用多线程技术
thread_list = []
max_connections = 2 # 定义最大线程数
pool_sema = threading.BoundedSemaphore(max_connections) # 或使用Semaphore方法
#创建用于合并的临时文件夹，用于存放ts文件
if not os.path.exists('merge'):
os.system('mkdir merge')
# 模拟浏览器header,防止误伤
headers = {'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/75.0.3770.100 Safari/537.36'}
# 获取m3u8文件的内容
m3u8_content = requests.get(m3u8_url,headers=headers).text
# 判断是否是最终的
is_final_m3u8 = not "m3u8" in m3u8_content
# 当 m3u8 作为主播放列表（Master Playlist）时，其内部提供的是同一份媒体资源的多份流列表资源（Variant Stream）
# 如果不是最终的m3u8,进入下一层找到最终的m3u8 （针对m3u8多分辨率适配的情况，会先有一个针对不同分辨率的m3u8索引文件）
if not is_final_m3u8:
# 解析出真正的m3u8_content
for m3u8_url in m3u8_content.split('\n'):
if "m3u8" in m3u8_url:
m3u8_url = base_url +m3u8_url
m3u8_content = requests.get(m3u8_url,headers=headers).text
break
m3u8_content_split_list = m3u8_content.split('\n')
# 判断视频是否经过AES-128加密，如果加密过则获取加密方式和加密秘钥
key = ''
for index,line in enumerate(m3u8_content_split_list):
# 判断视频是否经过AES-128加密
if "#EXT-X-KEY" in line:
#获取加密方式
method_pos = line.find("METHOD")
comma_pos = line.find(",")
method = line[method_pos:comma_pos].split('=')[1]
print("该视频经过加密，加密方式为：", method)
#获取加密密钥
uri_pos = line.find("URI")
quotation_mark_pos = line.rfind('"')
key_path = line[uri_pos:quotation_mark_pos].split('"')[1]
key_url = m3u8_url.replace("index.m3u8",key_path)
res = requests.get(key_url)
key = res.content
#从m3u8文件中解析出ts地址集
play_list = []
key = ''
for index,line in enumerate(m3u8_content_split_list):
#以下拼接方式可能会根据自己的需求进行改动
if '#EXTINF' in line:
# 如果加密，直接提取每一级的.ts文件链接地址
if 'http' in m3u8_content_split_list[index + 1]:
href = m3u8_content_split_list[index + 1]
play_list.append(href)
# 如果没有加密，直接构造出url链接
elif('ad0.ts' not in m3u8_content_split_list[index + 1]):
href = base_url + m3u8_content_split_list[index+1]
play_list.append(href)
print("m3u8文件解析成功，共解析到",len(play_list),"个ts文件")
print("上次下载完成了",len(os.listdir("./merge")),"个ts文件")
# 封装下载当个ts文件的函数
def down_ts(index,ts_url,cryptor = False):
# 获取ts content
if not cryptor:
ts_content = requests.get(ts_url,headers=headers).content
else:
ts_content = cryptor.decrypt(requests.get(ts_url,headers=headers).content)#获取bing解密ts
# 写入文件
with open('merge/' + str(index+1) + '.ts','wb') as file:
file.write(ts_content)
print('第{}/{}个ts文件下载完成'.format(index+1,len(play_list)))
# 根据ts地址集下载ts文件
print("开始下载所有的ts文件")
if(len(key)):# 如果加密过
from Crypto.Cipher import AES
cryptor = AES.new(key, AES.MODE_CBC, key)
for index,ts_url in enumerate(play_list):
if os.path.exists('merge/' + str(index+1) + '.ts'):continue #跳过已经存在的ts文件
thread_list.append(threading.Thread(target=down_ts, args=(index,ts_url,cryptor)))
else: # 如果未加密
for index,ts_url in enumerate(play_list):
if os.path.exists('merge/' + str(index+1) + '.ts'):continue #跳过已经存在的ts文件
thread_list.append(threading.Thread(target=down_ts, args=(index,ts_url)))
for t in thread_list:
t.start()
for t in thread_list:
t.join() # 子线程全部加入，主线程等所有子线程运行完毕
if len(os.listdir("./merge"))==len(play_list):
print('所有ts文件都已下载完成')
# 合并ts文件为mp4 并删除下载的ts文件
merge_cmd = "cat "
for i in range(len(os.listdir("./merge"))):
merge_cmd += "merge/"+ str(i+1) +".ts "
merge_cmd += ">>vedio/"+vedio_name+".mp4"
del_cmd = 'rm merge/*.ts'
os.system(merge_cmd)#执行合并命令
os.system(del_cmd)#执行删除命令
print(vedio_name,'.mp4下载已完成')
else:
print("下载发生错误,请重新运行，下载完成度{}/{}".format(len(os.listdir("./merge")),len(play_list)))
参考资料
- m3u8 文件格式详解:https://www.jianshu.com/p/e97f6555a070
评论 (0)