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

浅浅的花香味﹌ 2023-02-22 13:37 22阅读 0赞

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

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

功能  
1.使用线程池进行耗时操作  
2.可保留所有ts文件  
3.单个文件下载失败可手动下载单个文件，再通过shell命令合并  
4.如果m3u8资源支持多分辨率，可以指定速度优先(下载分辨率最小的）和画质优先（下载分辨率最大的）  
5.如果不填写视频名称，则使用随机字符串+数字的组合  
6.引入ffmpeg，增加加密m3u8文件下载功能(2020.03.15更新)

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

from tkinter import \*  
from tkinter import ttk  
import tkinter.messagebox

class M3u8Downloader:  
    def \_\_init\_\_(self, title="M3U8下载器", version=None, auth="莫近东墙"):  
        self.root = Tk()  
        self.title = title  
        self.version = version  
        self.auth = auth  
        self.root.title("%s-%s by %s" % (self.title, self.version, self.auth))  
        self.w = 350  
        self.h = 360  
        self.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"\] = 100  
        self.progress\["value"\] = 0

self.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和scrollbar  
        self.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)

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

import requests  
import Model\_http\_header

def easy\_download(url, cookie=None, header=Model\_http\_header.get\_user\_agent(), timeout=(10, 30),  
                  max\_retry\_time=3):  
    i = 1  
    while 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 None  
            return res  
        except Exception as e:  
            print(e)  
            i += 1  
    return 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

逻辑代码  
各个方法注释的挺详细的，我只提一下几个比较重要的地方：  
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 threading

m3 = 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\_increase  
    global m3  
    order\_increase = type\_  
    if type\_:  
        m3.alert("设置速度优先")  
    else:  
        m3.alert("设置画质优先")

\# 是否保存源文件  
def save\_source():  
    global save\_source\_file  
    global m3  
    if m3.cb\_status.get() == 0:  
        save\_source\_file = True  
        m3.alert("下载完成后保存源文件")  
    else:  
        save\_source\_file = False  
        m3.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 True  
    else:  
        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 res  
    return 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\_path  
    global url\_host  
    global m3  
    m3.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.text  
    else:  
        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("视频文件已加密，请等待后续版本")  
            break  
        if 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\_list  
    global url\_list  
    global exit\_flag  
    if exit\_flag:  
        return  
    response = dm.easy\_download(url)  
    if response is None:  
        download\_fail\_list.append((url, file\_name))  
        return  
    with open(file\_name, 'wb') as file:  
        file.write(response.content)  
        p = count\_file(file\_name)/len(url\_list)\*100  
        set\_progress(p)

\# 设置进度条  
def set\_progress(v):  
    global m3  
    m3.progress\["value"\] = v  
    m3.root.update()

\# 重新下载视频  
def download\_fail\_file():  
    global download\_fail\_list  
    global m3  
    if 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))  
                continue  
            with open(file\_name, 'wb') as file:  
                file.write(response.content)  
                p = count\_file(file\_name)/len(url\_list)\*100  
                set\_progress(p)

\# 合并文件  
def merge\_file(dir\_name):  
    global m3  
    com = "copy /b \\"" + dir\_name + "\\\\\*\\" \\"" + dir\_name + ".ts\\""  
    m3.alert("执行文件合并命令:%s" % com)  
    res = os.system(com)  
    if res == 0:  
        return True  
    else:  
        return False

\# 拼接下载用的参数  
def get\_download\_params(head, dir\_name):  
    global url\_list  
    i = 0  
    params = \[\]  
    while i < len(url\_list):  
        index = "%05d" % i  
        param = (\[head + url\_list\[i\], dir\_name + "\\\\" + index + ".ts"\], None)  
        params.append(param)  
        i += 1  
    return params

\# 设置线程池开始下载  
def start\_download\_in\_pool(params):  
    global m3  
    m3.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 = 0  
    for f\_path, f\_dir\_name, f\_names in os.walk(path):  
        for name in f\_names:  
            if name.endswith(".ts"):  
                file\_num += 1  
    return file\_num

\# 检查视频文件是否全部下载完成  
def check\_file(dir\_name):  
    global url\_list  
    path = dir\_name  
    file\_num = 0  
    for f\_path, f\_dir\_name, f\_names in os.walk(path):  
        for name in f\_names:  
            if name.endswith(".ts"):  
                file\_num += 1  
    return file\_num == len(url\_list)

\# 测试下载地址  
def test\_download\_url(url):  
    global m3  
    m3.alert("尝试使用%s下载视频" % url)  
    res = dm.easy\_download(url, max\_retry\_time=10)  
    return res is not None

def start(m3u8\_href, video\_name):  
    global download\_fail\_list  
    global running  
    global url\_list  
    global m3  
    global url\_path  
    global url\_host

m3.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 = False  
        return  
    \# 获取程序所在目录  
    path = os.path.dirname(os.path.realpath(sys.argv\[0\]))  
    video\_name = path+"\\\\"+video\_name  
    if 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 = False  
        return  
    \# 重新下载先前下载失败的视频  
    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 = False

def s():  
    global m3  
    if 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\_flag  
    exit\_flag = True  
    sys.exit(0)

def run():  
    global m3  
    m3 = 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()

总结  
贴出来的是我修改以后的第三个版本，日后有时间了再优化。  
打包成可执行文件的工具下载地址：链接: https://pan.baidu.com/s/1go2awUhjJgoAQpxfRMeqVw 提取码: r8jx（2020.03.15更新3.7.0版本）  
各位要注意身体啊  
原文链接：https://blog.csdn.net/muslim377287976/article/details/104340242