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)))

参考资料

1. m3u8 文件格式详解:https://www.jianshu.com/p/e97f6555a070