m3u8转mp4缓存合并工厂_使用Python下载M3U8格式视频

/  作者简介  /

本篇文章来自 MRArchive 的投稿&＃xff0c;分享了如何使用Python下载M3U8格式的视频&＃xff0c;希望对大家有所帮助&＃xff01;同时也感谢作者贡献的精彩文章。

作者信息

网站&＃xff1a;https://toodo.fun

B站&＃xff1a;https://space.bilibili.com/314191627

/  背景简介  /

M3U 是一种播放多媒体列表的文件格式&＃xff0c;它的设计初衷是为了播放音频文件&＃xff0c;比如MP3&＃xff0c;但是越来越多的软件现在用来播放视频文件列表&＃xff0c;M3U也可以指定在线流媒体音频源。很多播放器和软件都支持M3U文件格式。

m3u8 文件是 HTTP Live Streaming(缩写为 HLS) 协议的部分内容&＃xff0c;而 HLS 是一个由苹果公司提出的基于 HTTP 的流媒体网络传输协议。

HLS 是新一代流媒体传输协议&＃xff0c;其基本实现原理为将一个大的媒体文件进行分片&＃xff0c;将该分片文件资源路径记录于 m3u8 文件(即 playlist)内&＃xff0c;其中附带一些额外描述(比如该资源的多带宽信息···)用于提供给客户端。客户端依据该 m3u8 文件即可获取对应的媒体资源&＃xff0c;进行播放。

扩展M3U指令&＃xff1a;

#EXTM3U //必需&＃xff0c;表示一个扩展的m3u文件

#EXT-X-VERSION:3 //hls的协议版本号&＃xff0c;暗示媒体流的兼容性

#EXT-X-MEDIA-SEQUENCE:xx //首个分段的sequence number

#EXT-X-ALLOW-CACHE:NO //是否缓存

#EXT-X-TARGETDURATION:5 //每个视频分段最大的时长(单位秒)

#EXT-X-DISCONTINUITY //表示换编码

#EXTINF: //每个切片的时长

/  获取.m3u8文件中的视频信息  /

.m3u8文件储存了视频所在的位置信息&＃xff0c;我们可以通过发送一个Get请求来获取链接中的内容&＃xff1a;

content &＃61; requests.get(m3u8Url).text

/  拼接视频下载链接  /

我们通过得到的视频信息进行视频网址拼接&＃xff0c;.m3u8文件中的链接可以为全路径或者相对路径&＃xff0c;所以我们判断后进行拼接并存入一个列表当中&＃xff1a;

urls &＃61; []for index, video in enumerate(content.split(&＃39;\n&＃39;)): if &＃39;#EXTINF&＃39; in video: if content[index &＃43; 1][0] &＃61;&＃61; &＃39;/&＃39;: downloadLink &＃61; url.split(&＃39;//&＃39;)[0] &＃43; "//" &＃43; url.split(&＃39;//&＃39;)[1].split(&＃39;/&＃39;)[0] &＃43; content[index &＃43; 1] elif content[index &＃43; 1][:4] &＃61;&＃61; &＃39;http&＃39;: downloadLink &＃61; content[index &＃43; 1] else: downloadLink &＃61; url.replace(url.split(&＃39;/&＃39;)[-1], content[index &＃43; 1]) urls.append(downloadLink)

/  使用多线程下载视频  /

得到视频列表后&＃xff0c;我们就可以对视频进行下载了&＃xff0c;为了提高下载效率&＃xff0c;我们可以使用多线程进行下载&＃xff1a;

def download(downloadLink, name): for _ in range(10): try: req &＃61; requests.get(downloadLink, headers&＃61;headers, timeout&＃61;15).content with open(f"{name}", "wb") as f: f.write(req) f.flush() break except: if _ &＃61;&＃61; 9: print(f"{name}下载失败") else: print(f"{name}正在进行第{_}次重试")pool &＃61; ThreadPoolExecutor(max_workers&＃61;threadNum)futures &＃61; []for index, downloadLink in enumerate(urls): fileList.append(os.path.basename(downloadLink)) futures.append(pool.submit(download, downloadLink, f"{downloadPath}/{os.path.basename(downloadLink)}"))wait(futures)

/  合并视频  /

等待全部下载完成后&＃xff0c;是一个个的ts视频文件&＃xff0c;然后我们再将这些文件合并成一个视频文件&＃xff0c;此时要注意视频的顺序&＃xff0c;我们可以在我们的视频列表中依次取出进行合并

def merge_file(path, name): global fileList cmd &＃61; "copy /b " for i in fileList: if i !&＃61; fileList[-1]: cmd &＃43;&＃61; f"{i} &＃43; " else: cmd &＃43;&＃61; f"{i} {name}" os.chdir(path) with open(&＃39;combine.cmd&＃39;, &＃39;w&＃39;) as f: f.write(cmd) os.system("combine.cmd") os.system(&＃39;del /Q *.ts&＃39;) os.system(&＃39;del /Q *.cmd&＃39;)

这里我们是写了一个脚本来完成合并的任务&＃xff0c;使用命令&＃39;copy /b file1 &＃43; file2 &＃43;... &＃43; fileN newFile&＃39;进行合并&＃xff0c;并于完成后将ts小文件进行了删除。至此&＃xff0c;视频就下载完成了。

/  完整代码如下  /

import requestsimport osfrom concurrent.futures import ThreadPoolExecutor, waitimport sysfinishedNum &＃61; 0allNum &＃61; 0fileList &＃61; []headers &＃61; {&＃39;User-Agent&＃39;: &＃39;Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:62.0) Gecko/20100101 Firefox/62.0&＃39;}def download(downloadLink, name): global finishedNum global allNum for _ in range(10): try: req &＃61; requests.get(downloadLink, headers&＃61;headers, timeout&＃61;15).content with open(f"{name}", "wb") as f: f.write(req) f.flush() finishedNum &＃43;&＃61; 1 print(f"{name}下载成功, 总进度{round(finishedNum / allNum * 100, 2)}% ({finishedNum}/{allNum})") break except: if _ &＃61;&＃61; 9: print(f"{name}下载失败") else: print(f"{name}正在进行第{_}次重试")def merge_file(path, name): global fileList cmd &＃61; "copy /b " for i in fileList: if i !&＃61; fileList[-1]: cmd &＃43;&＃61; f"{i} &＃43; " else: cmd &＃43;&＃61; f"{i} {name}" os.chdir(path) with open(&＃39;combine.cmd&＃39;, &＃39;w&＃39;) as f: f.write(cmd) os.system("combine.cmd") os.system(&＃39;del /Q *.ts&＃39;) os.system(&＃39;del /Q *.cmd&＃39;)def downloader(url, name, threadNum): global allNum global fileList print("读取文件信息中...") downloadPath &＃61; &＃39;Download&＃39; if not os.path.exists(downloadPath): os.mkdir(downloadPath) # 查看是否存在 if os.path.exists(f"{downloadPath}/{name}"): print(f"视频文件已经存在&＃xff0c;如需重新下载请先删除之前的视频文件") return content &＃61; requests.get(url, headers&＃61;headers).text.split(&＃39;\n&＃39;) if "#EXTM3U" not in content[0]: raise BaseException(f"非M3U8链接") # .m3u8 跳转 for video in content: if ".m3u8" in video: if video[0] &＃61;&＃61; &＃39;/&＃39;: url &＃61; url.split(&＃39;//&＃39;)[0] &＃43; "//" &＃43; url.split(&＃39;//&＃39;)[1].split(&＃39;/&＃39;)[0] &＃43; video elif video[:4] &＃61;&＃61; &＃39;http&＃39;: url &＃61; video else: url &＃61; url.replace(url.split(&＃39;/&＃39;)[-1], video) print(url) content &＃61; requests.get(url, headers&＃61;headers).text.split(&＃39;\n&＃39;) urls &＃61; [] for index, video in enumerate(content): if &＃39;#EXTINF&＃39; in video: if content[index &＃43; 1][0] &＃61;&＃61; &＃39;/&＃39;: downloadLink &＃61; url.split(&＃39;//&＃39;)[0] &＃43; "//" &＃43; url.split(&＃39;//&＃39;)[1].split(&＃39;/&＃39;)[0] &＃43; content[index &＃43; 1] elif content[index &＃43; 1][:4] &＃61;&＃61; &＃39;http&＃39;: downloadLink &＃61; content[index &＃43; 1] else: downloadLink &＃61; url.replace(url.split(&＃39;/&＃39;)[-1], content[index &＃43; 1]) urls.append(downloadLink) allNum &＃61; len(urls) pool &＃61; ThreadPoolExecutor(max_workers&＃61;threadNum) futures &＃61; [] for index, downloadLink in enumerate(urls): fileList.append(os.path.basename(downloadLink)) futures.append(pool.submit(download, downloadLink, f"{downloadPath}/{os.path.basename(downloadLink)}")) wait(futures) print(f"运行完成") merge_file(downloadPath, name) print(f"合并完成") print(f"文件下载成功&＃xff0c;尽情享用吧")if __name__ &＃61;&＃61; &＃39;__main__&＃39;: videoUrl &＃61; str(sys.argv[1]) name &＃61; str(sys.argv[2]) threadNum &＃61; int(sys.argv[3]) downloader(videoUrl, name, threadNum)

/  打包好的Windows版软件下载  /

百度云下载地址&＃xff1a;https://pan.baidu.com/s/1ZsPb9WTmYP8VUKuR9tuoyw 提取码&＃xff1a;mxb6

蓝奏云下载地址&＃xff1a;https://lanzous.com/icnc1ve