I.挖坑缘由

现在很多在线观看的视频为了防盗链使用了M3u8格式，想要下载的话比较麻烦，如果切分的ts文件名是递增的数字序号的还好说，但是很多是随机的字母，这种就无法通过使用迅雷的批量任务来下载了。然而网上搜到的m3u8downloader使用起来不是很满意，那个工具应该是单线程的，下载进度贼慢，而且如果有一个资源卡住了，就会一直卡在那里，另外我在开发这个下载工具时发现了很多m3u8资源指向是跨域的，不一定都在一个域名下，有可能我使用m3u8downloader时下载失败是这个原因导致的。
在被m3u8downloader折磨了一段时间后终于准备自己写一个下载器了。
先康康最终成果吧

II.功能/更新记录

使用线程池进行耗时操作
可保留所有ts文件
单个文件下载失败可手动下载单个文件，再通过shell命令合并
如果m3u8资源支持多分辨率，可以指定速度优先(下载分辨率最小的）和画质优先（下载分辨率最大的）
如果不填写视频名称，则使用随机数字的组合
引入ffmpeg，增加加密m3u8文件下载功能(2020.03.15更新)
优化地址拼接逻辑，可能新增了BUG (2020.11.11更新)
修复地址拼接逻辑的BUG(2020.11.17更新)
修复加密key处理失败的问题(2022.02.16更新)
去掉文件夹名称包含的特殊字符，修复HTTPS链接的BUG(2022.08.16更新)
修复了导致key.key文件下载地址拼接错误的BUG(2022.08.18更新)
再次修复了key.key文件下载地址拼接错误的BUG，双版本打包：-cmd.exe文件显示命令行窗内，便于定位下载\合并失败的问题，.exe文件不显示命令行窗口(2022.11.02更新)

III.代码

1.GUI

界面部分使用tkinter，虽然丑了点但是挺好用的。。
逻辑代码部分需要与GUI进行交互，显示进度、弹框等，所以把GUI封装成了一个类。这里需要注意，GUI代码部分还没有与逻辑代码绑定。

from tkinter import *
from tkinter import ttk
import tkinter.messageboxclass M3u8Downloader:def __init__(self, title="M3U8下载器", version=None, auth="莫近东墙"):self.root = Tk()self.title = titleself.version = versionself.auth = authself.root.title("%s-%s by %s" % (self.title, self.version, self.auth))self.w = 350self.h = 360self.frm = LabelFrame(self.root, width=self.w - 20, height=170, padx=10, text="设置")self.frm.place(x=10, y=5)Label(self.frm, text="m3u8地址:", font=("Lucida Grande", 11)).place(x=0, y=0)self.button_url = Entry(self.frm, width=30)self.button_url.place(x=0, y=25)Label(self.frm, text="视频名称:(无需后缀名)", font=("Lucida Grande", 11)).place(x=0, y=50)self.button_video_name = Entry(self.frm, width=30)self.button_video_name.place(x=0, y=75)self.v = IntVar()self.cb_status = IntVar()self.v.set(1)self.rb1 = Radiobutton(self.frm, text='速度优先', variable=self.v, value=1, font=("Lucida Grande", 11))self.rb2 = Radiobutton(self.frm, text='画质优先', variable=self.v, value=2, font=("Lucida Grande", 11))self.cb = Checkbutton(self.frm, text='保存源文件', variable=self.cb_status, font=("Lucida Grande", 11))self.rb1.place(x=0, y=95)self.rb2.place(x=100, y=95)self.cb.place(x=200, y=95)self.button_start = Button(self.frm, text="开始下载", width=8, font=("Lucida Grande", 11))self.button_start.place(x=230, y=15)self.button_exit = Button(self.frm, text="退出", width=8, font=("Lucida Grande", 11))self.button_exit.place(x=230, y=70)self.progress = ttk.Progressbar(self.frm, orient="horizontal", length=self.w - 40, mode="determinate")self.progress.place(x=0, y=120)self.progress["maximum"] = 100self.progress["value"] = 0self.message_frm = LabelFrame(self.root, width=self.w - 20, height=170, padx=10, text="消息")self.message_frm.place(x=10, y=180)self.scrollbar = Scrollbar(self.message_frm)self.scrollbar.pack(side='right', fill='y')self.message_v = StringVar()self.message_s = ""self.message_v.set(self.message_s)self.message = Text(self.message_frm, width=41, height='11')self.message.insert('insert', self.message_s)self.message.pack(side='left', fill='y')# 以下两行代码绑定text和scrollbarself.scrollbar.config(command=self.message.yview)self.message.config(yscrollcommand=self.scrollbar.set)self.message.config(state=DISABLED)ws, hs = self.root.winfo_screenwidth(), self.root.winfo_screenheight()self.root.geometry('%dx%d+%d+%d' % (self.w, self.h, (ws / 2) - (self.w / 2), (hs / 2) - (self.h / 2)))self.root.resizable(0, 0)# self.root.mainloop()def alert(self, m):print("%s" % m)if m:self.message.config(state=NORMAL)self.message.insert(END, m + "\n")# 确保scrollbar在底部self.message.see(END)self.message.config(state=DISABLED)self.root.update()def clear_alert(self):self.message.config(state=NORMAL)self.message.delete('1.0', 'end')self.message.config(state=DISABLED)self.root.update()def show_info(self, m):tkinter.messagebox.showinfo(self.title,  m)

2.下载工具类

这里需要注意的是，requests的超时分为两种，请求超时和读取超时，请求超时是指连接不上，读取超时是指连接上了，但是资源下载不下来（常见于下载国外的资源），timeout=(10, 30)就是设置这两种超时时间。
header=Model_http_header.get_user_agent()是我专门写了一个类用来随机设置请求头的，毕竟很多网站设置了反爬虫。。

import requests
import Model_http_headerdef easy_download(url, cookie=None, header=Model_http_header.get_user_agent(), timeout=(10, 30),max_retry_time=3):i = 1while i <= max_retry_time:try:print("连接:%s" % url)res = requests.get(url=(url.rstrip()).strip(), cookies=cookie, headers=header, timeout=timeout)if res.status_code != 200:return Nonereturn resexcept Exception as e:print(e)i += 1return None

这个就是随机设置请求头的代码，其中需要注意的是'Accept-Encoding': 'gzip, deflate'，可接受的编码格式里面我去掉了br，因为真的有网站把ts文件用br格式进行编码。但是requests默认是不支持解码br格式的。

import random"""随机设置user_agent"""
user_agent_list = ["Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.1 (KHTML, like Gecko) Chrome/22.0.1207.1 Safari/537.1","Mozilla/5.0 (X11; CrOS i686 2268.111.0) AppleWebKit/536.11 (KHTML, like Gecko) Chrome/20.0.1132.57 Safari/536.11","Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/536.6 (KHTML, like Gecko) Chrome/20.0.1092.0 Safari/536.6","Mozilla/5.0 (Windows NT 6.2) AppleWebKit/536.6 (KHTML, like Gecko) Chrome/20.0.1090.0 Safari/536.6","Mozilla/5.0 (Windows NT 6.2; WOW64) AppleWebKit/537.1 (KHTML, like Gecko) Chrome/19.77.34.5 Safari/537.1","Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/536.5 (KHTML, like Gecko) Chrome/19.0.1084.9 Safari/536.5","Mozilla/5.0 (Windows NT 6.0) AppleWebKit/536.5 (KHTML, like Gecko) Chrome/19.0.1084.36 Safari/536.5","Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/536.3 (KHTML, like Gecko) Chrome/19.0.1063.0 Safari/536.3","Mozilla/5.0 (Windows NT 5.1) AppleWebKit/536.3 (KHTML, like Gecko) Chrome/19.0.1063.0 Safari/536.3","Mozilla/5.0 (Macintosh; Intel Mac OS X 10_8_0) AppleWebKit/536.3 (KHTML, like Gecko) Chrome/19.0.1063.0 ""Safari/536.3","Mozilla/5.0 (Windows NT 6.2) AppleWebKit/536.3 (KHTML, like Gecko) Chrome/19.0.1062.0 Safari/536.3","Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/536.3 (KHTML, like Gecko) Chrome/19.0.1062.0 Safari/536.3","Mozilla/5.0 (Windows NT 6.2) AppleWebKit/536.3 (KHTML, like Gecko) Chrome/19.0.1061.1 Safari/536.3","Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/536.3 (KHTML, like Gecko) Chrome/19.0.1061.1 Safari/536.3","Mozilla/5.0 (Windows NT 6.1) AppleWebKit/536.3 (KHTML, like Gecko) Chrome/19.0.1061.1 Safari/536.3","Mozilla/5.0 (Windows NT 6.2) AppleWebKit/536.3 (KHTML, like Gecko) Chrome/19.0.1061.0 Safari/536.3","Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/535.24 (KHTML, like Gecko) Chrome/19.0.1055.1 Safari/535.24","Mozilla/5.0 (Windows NT 6.2; WOW64) AppleWebKit/535.24 (KHTML, like Gecko) Chrome/19.0.1055.1 Safari/535.24"
]def get_user_agent():header = {'Accept': 'application/json, text/javascript, */*; q=0.01','Accept-Encoding': 'gzip, deflate','content-type': 'application/json','x-requested-with': 'XMLHttpRequest','Accept-Language': 'zh-CN,zh;q=0.8','User-Agent': random.choice(user_agent_list)}return header

3.逻辑代码

各个方法注释的挺详细的，我只提一下几个比较重要的地方：
1.代码中会执行下载的耗时操作，需要另开一个线程来跑逻辑代码，不然GUI会卡住。
2.如果在GUI初始化的时候就绑定逻辑代码，就是把s()绑定到button_start这个按钮上，那么代码运行过程中show_info等方法是无法生效的，因为__init__的时候，已经把逻辑代码绑定好了，这时的m3还是None，因此只能等m3对象初始化完成以后，手动绑定按键事件。（我已经晕了）
3.获取ts下载地址是最麻烦的，首先大部分的m3u8文件里面会再嵌套一个m3u8文件，这样做原本是为了提供多分辨率资源可供选择，但是现在基本上都是用来屏蔽m3u8下载插件的。然后ts下载地址都是相对路径，但是这个相对路径有的是相对m3u8文件的，有的是相对域名的。甚至有的m3u8文件域名和嵌套的m3u8文件域名不一样。所以在正式开始下载以前只能先拿一个下载地址进行测试，测试通过了再开始下载

#!/usr/bin/python3
import Model_download as dm
import os
import sys
import shutil
import threadpool
import random
import m3u8Downloader
import threadingm3 = None
download_fail_list = []
running = False
url_list = []
order_increase = True
exit_flag = False
save_source_file = False
url_host = None
url_path = None# 设置排序模式
def order_type(type_):global order_increaseglobal m3order_increase = type_if type_:m3.alert("设置速度优先")else:m3.alert("设置画质优先")# 是否保存源文件
def save_source():global save_source_fileglobal m3if m3.cb_status.get() == 0:save_source_file = Truem3.alert("下载完成后保存源文件")else:save_source_file = Falsem3.alert("下载完成后删除源文件")# 获取域名
def get_host(url):url_param = url.split("//")return url_param[0]+"//"+url_param[1].split("/")[0]+"/"# 获取目录
def get_dir(url):host = get_host(url)url = url.replace(host, '')return ("/"+url[0:url.rfind("/")]+"/").replace("//", "/")# 获取域名+路径
def get_path(url):if url.rfind("/") != -1:return url[0:url.rfind("/")]+"/"else:return url[0:url.rfind("\\")] + "\\"# 检查地址是否正确
def check_href(m3u8_href):if m3u8_href:return Trueelse:return False# 检查文件名是否正确
def check_video_name(name):if name is None or "" == name:a = "1234567890"b = "abcdefghijklmnopqrstuvwxyz"aa = []bb = []for i in range(6):aa.append(random.choice(a))bb.append(random.choice(b))res = "".join(i + j for i, j in zip(aa, bb))return resreturn name.replace("\t", "").replace("\n", "")# 获取带宽
def get_band_width(info):info_list = info.split("\n")[0].split(",")for info in info_list:if info.startswith("BANDWIDTH"):return int(info.split("=")[1])return 0# 排序
def order_list(o_type, o_list):o_list.sort(key=get_band_width, reverse=o_type)return o_list# 获取视频下载地址
def get_ts_add(m3u8_href):global url_pathglobal url_hostglobal m3m3.alert("获取ts下载地址，m3u8地址:\n%s" % m3u8_href)url_host = get_host(m3u8_href)url_path = get_path(m3u8_href)response = dm.easy_download(m3u8_href)if response is not None:response = response.textelse:return []m3.alert("响应体:\n%s\n" % response)response_list = response.split("#")ts_add = []m3u8_href_list_new = []for res_obj in response_list:if res_obj.startswith("EXT-X-KEY"):m3.show_info("视频文件已加密，请等待后续版本")breakif res_obj.startswith("EXT-X-STREAM-INF"):# m3u8 作为主播放列表（Master Playlist），其内部提供的是同一份媒体资源的多份流列表资源（Variant Stream）# file_add = res_obj.split("\n")[1]file = res_obj.split(":")[1]m3u8_href_list_new.append(file)if res_obj.startswith("EXTINF"):# 当 m3u8 文件作为媒体播放列表（Media Playlist），其内部信息记录的是一系列媒体片段资源file = res_obj.split("\n")[1]ts_add.append(file)if len(m3u8_href_list_new) > 0:# 根据画质优先/速度优先排序m3u8_href_list_new = order_list(order_increase, m3u8_href_list_new)for info in m3u8_href_list_new:file = info.split("\n")[1]ts_add = get_ts_add(url_host + file)if len(ts_add) == 0:ts_add = get_ts_add(url_path + file)return ts_add# 下载视频并保存为文件
def download_to_file(url, file_name):global download_fail_listglobal url_listglobal exit_flagif exit_flag:returnresponse = dm.easy_download(url)if response is None:download_fail_list.append((url, file_name))returnwith open(file_name, 'wb') as file:file.write(response.content)p = count_file(file_name)/len(url_list)*100set_progress(p)# 设置进度条
def set_progress(v):global m3m3.progress["value"] = vm3.root.update()# 重新下载视频
def download_fail_file():global download_fail_listglobal m3if len(download_fail_list) > 0:for info in download_fail_list:url = info[0]file_name = info[1]m3.alert("正在尝试重新下载%s" % file_name)response = dm.easy_download(url=url, max_retry_time=50)if response is None:m3.alert("%s下载失败，请手动下载:\n%s" % (file_name, url))continuewith open(file_name, 'wb') as file:file.write(response.content)p = count_file(file_name)/len(url_list)*100set_progress(p)# 合并文件
def merge_file(dir_name):global m3com = "copy /b \"" + dir_name + "\\*\" \"" + dir_name + ".ts\""m3.alert("执行文件合并命令:%s" % com)res = os.system(com)if res == 0:return Trueelse:return False# 拼接下载用的参数
def get_download_params(head, dir_name):global url_listi = 0params = []while i < len(url_list):index = "%05d" % iparam = ([head + url_list[i], dir_name + "\\" + index + ".ts"], None)params.append(param)i += 1return params# 设置线程池开始下载
def start_download_in_pool(params):global m3m3.alert("已确认正确地址，开始下载")pool = threadpool.ThreadPool(10)thread_requests = threadpool.makeRequests(download_to_file, params)[pool.putRequest(req) for req in thread_requests]pool.wait()# 获取视频文件数量
def count_file(file_name):path = get_path(file_name)file_num = 0for f_path, f_dir_name, f_names in os.walk(path):for name in f_names:if name.endswith(".ts"):file_num += 1return file_num# 检查视频文件是否全部下载完成
def check_file(dir_name):global url_listpath = dir_namefile_num = 0for f_path, f_dir_name, f_names in os.walk(path):for name in f_names:if name.endswith(".ts"):file_num += 1return file_num == len(url_list)# 测试下载地址
def test_download_url(url):global m3m3.alert("尝试使用%s下载视频" % url)res = dm.easy_download(url, max_retry_time=10)return res is not Nonedef start(m3u8_href, video_name):global download_fail_listglobal runningglobal url_listglobal m3global url_pathglobal url_hostm3.clear_alert()set_progress(0)# 检查地址是否合法if check_href(m3u8_href) is False:m3.alert("请输入正确的m3u8地址")return# 格式化文件名video_name = check_video_name(video_name)# 任务开始标志，防止重复开启下载任务running = True# 获取所有ts视频下载地址url_list = get_ts_add(m3u8_href)if len(url_list) == 0:m3.alert("获取地址失败")# 重置任务开始标志running = Falsereturn# 获取程序所在目录path = os.path.dirname(os.path.realpath(sys.argv[0]))video_name = path+"\\"+video_nameif not os.path.exists(video_name):os.makedirs(video_name)m3.alert("总计%s个视频" % str(len(url_list)))# 拼接正确的下载地址开始下载if test_download_url(url_host+url_list[0]):params = get_download_params(head=url_host, dir_name=video_name)# 线程池开启线程下载视频start_download_in_pool(params)elif test_download_url(url_path+url_list[0]):params = get_download_params(head=url_path, dir_name=video_name)# 线程池开启线程下载视频start_download_in_pool(params)else:m3.alert("地址连接失败")running = Falsereturn# 重新下载先前下载失败的视频download_fail_file()# 检查ts文件总数是否对应if check_file(video_name):# 调用cmd方法合并视频if merge_file(video_name):if save_source_file is False:# 删除文件夹shutil.rmtree(video_name)m3.alert("下载完成")m3.show_info("下载完成")set_progress(0)else:m3.alert("视频文件合并失败,请查看消息列表")m3.show_info("视频文件合并失败,请查看消息列表")else:m3.alert("请手动下载缺失文件并合并")m3.show_info("请手动下载缺失文件并合并")# 清空下载失败视频列表download_fail_list = []# 重置任务开始标志running = Falsedef s():global m3if running is False:m3u8_href = m3.button_url.get().rstrip()video_name = m3.button_video_name.get().rstrip()# 开启线程执行耗时操作，防止GUI卡顿t = threading.Thread(target=start, args=(m3u8_href, video_name,))# 设置守护线程，进程退出不用等待子线程完成t.setDaemon(True)t.start()else:m3.show_info("任务执行中，请勿重复开启任务")def e():global exit_flagexit_flag = Truesys.exit(0)def run():global m3m3 = m3u8Downloader.M3u8Downloader(version="3.6.8")# 绑定点击事件m3.rb1.bind("<Button-1>", lambda x: order_type(True))m3.rb2.bind("<Button-1>", lambda x: order_type(False))m3.cb.bind("<Button-1>", lambda x: save_source())m3.button_start.bind("<Button-1>", lambda x: s())m3.button_exit.bind("<Button-1>", lambda x: e())# 手动加入消息队列m3.root.mainloop()if __name__ == "__main__":run()

IV.下载地址

CSDN下载

（2022.08.18更新3.7.10版本）

各位要注意身体啊

使用Python写一个m3u8多线程下载器相关推荐

转:使用Python写一个m3u8多线程下载器
转载:使用Python写一个m3u8多线程下载器可去看原文:https://blog.csdn.net/muslim377287976/article/details/104340242 文章目录 ...
用Java写一个电影自动下载器
你好! 下面是一些步骤来帮助你写一个电影自动下载器: 建立一个新的Java项目选择一个电影下载网站作为数据源, 并使用网络爬虫或API来获取电影的信息(如标题, 时长, 格式, 大小等) 使用Jav ...
【python爬虫】实现多线程下载器
写在前面为什么要多线程?单个线程不能下载吗?多线程能占满网络实现宽带的满速下载而单线程不能. 举个栗子:你的宽带是100Mb/s,理论上最大下载速度是100/8=12.5MB/s.你要下载一个843 ...
用python写一个文件管理程序下载_Python管理文件神器 os.walk
原标题:Python管理文件神器 os.walk 来自:CSDN,作者:诡途 [导语]:有没有想过用python写一个文件管理程序?听起来似乎没思路?其实是可以的,因为Python已经为你准备好了神器 ...
java爬虫写一个百度图片下载器
文章目录 img_download 1.0 看看效果吧 2.0 了解一下 "图片下载器软件" 目录结构 3.0 如何使用? 4.0 源码剖析 5.0 项目地址 6.0 写在最后的话 ...
Python写一个简洁拼写检查器
网上看到的一篇神文,利用的是朴素贝叶斯模型实现了一个简单的拼写检查器. 英文原文链接见这里,中文翻译如下 =============================================== ...
python制作一个网易音乐下载器
你只需要在代码同级目录新建一个文件夹mp3即可.代码可复制粘贴. 第一次思路如下,该效果只能一次下载单个音乐: #coding=gbk """ 描述:传参id即可下载音乐 ...
M3U8多线程下载器
一.M3U8简介 M3U8是一种用于指示多媒体播放列表的格式.这种格式通常用于流媒体播放,尤其是在直播和点播领域. M3U8文件是一个文本文件,其中包含一个或多个URL,这些URL指向实际的媒体文件或 ...
python每隔30s检查一次_用Python写一个“离线语音提示器”来提醒我们别忘记了时间...
前言本文的文字及图片来源于网络,仅供学习.交流使用,不具有任何商业用途,版权归原作者所有,如有问题请及时联系我们以作处理. 环境: Win7系统,外网未连接,主机接有返听音箱. 准备: 这里明显要用 ...
python写邮箱系统_教大家用Python写一个简单电子邮件发信器
嘛~炎热的暑假大家都在家干些啥呢?up主本人每天就是摸鱼哒!为了让这个懒懒的up每天从床上早点爬起来,我可是立了不少flag呢~那就先不多说了,直接开始正文吧. 声明一下,本文内容为原创,如果引用其他 ...

使用Python写一个m3u8多线程下载器

文章目录