一键下载 M3U8 并转换为 MP4升级版

之前的下载 M3U8程序，有很多问题，为此做了一些升级，分享给大家。

增加了存在播放列表的情况处理
播放列表路径和ts路径错误问题
多线程问题
对于电视剧多文件下载的处理

这里从网上找了一部的链接，可以参考这个网站https://www.zuida001.com/


import os
import urllib3
import requests
import subprocess
import m3u8
from urllib import parse
from tenacity import retry, wait_random, stop_after_attempt
import gevent
from gevent.threadpool import ThreadPoolurllib3.disable_warnings()class M3u8Downloader:def __init__(self, pool_size=10):self.pool = ThreadPool(pool_size)@retry(stop=stop_after_attempt(3), wait=wait_random(2, 5))def request(self, url):"""发送请求"""try:res = requests.get(url, verify=False, timeout=5)return resexcept Exception as e:print(url, e)raise edef download_segment(self, url, file):"""下载ts文件"""if os.path.exists(file):return res = self.request(url)with open(file, "wb")as f:f.write(res.content)def convert_mp4(self, path, output, key):if not os.path.exists(output):# 使用FFmpeg将所有.ts文件合并为一个MP4文件    ffmpeg -allowed_extensions ALL -i index.m3u8 -c copy xxx.mp4if key:subprocess.call(['ffmpeg', '-allowed_extensions', 'ALL', '-i', 'local.m3u8', '-c', 'copy', output], cwd=path)else:subprocess.call(['ffmpeg', '-i', 'local.m3u8', '-c', 'copy', output], cwd=path)def download_m3u8(self, url, path):"""下载M3U8文件，有些存在播放列表，默认选择第一个"""m3u8_file_name = os.path.join(path, "index.m3u8")res = self.request(url)with open(m3u8_file_name, "w", encoding="utf-8")as f:f.write(res.text)# 解析M3U8文件m3u8_obj = m3u8.loads(res.text)# 如果存在清晰度列表，请求解析清晰度列表m3u8_playlist = []for playlist in m3u8_obj.playlists:uri = parse.urljoin(url, playlist.uri)bandwidth = playlist.stream_info.bandwidthresolution= playlist.stream_info.resolutionm3u8_file_name = os.path.join(path, f"{'x'.join([str(i) for i in resolution])}_{bandwidth}.m3u8")res = self.request(uri)with open(m3u8_file_name, "w", encoding="utf-8")as f:f.write(res.text)cur_m3u8 = m3u8.loads(res.text)cur_m3u8.uri = urim3u8_playlist.append(cur_m3u8)if m3u8_playlist:# 播放列表默认选择第一个return m3u8_playlist[0]else:return m3u8_objdef download(self, url, path, output):"""下载单个m3u8主程序url: m3u8链接urlpath: 单个m3u8目录output: 转换输出文件路径名"""# 创建目录segment_dir = os.path.join(path, "index")if not os.path.exists(segment_dir):os.makedirs(segment_dir)output_dir = os.path.dirname(output)if not os.path.exists(output_dir):os.makedirs(output_dir)# 解析M3U8文件m3u8_obj = self.download_m3u8(url, path)# 下载key文件 for item in m3u8_obj.keys:if item:key_url = parse.urljoin(url, item.uri)key_file_name = os.path.join(path, key_url.split("/")[-1])res = self.request(key_url)with open(key_file_name, "w", encoding="utf-8")as f:f.write(res.text)segments = []# 生成新的本地M3U8文件内容for index, segment in enumerate(m3u8_obj.segments):uri = parse.urljoin(m3u8_obj.uri, segment.uri)segments.append(uri)# 有些ts文件名过长，对其以序号重新命名segment.uri = f"index/{index}.{uri.split('.')[-1]}"# 保存M3U8文件local_file_name = os.path.join(path, "local.m3u8")with open(local_file_name, 'w') as f:f.write(m3u8_obj.dumps())# 下载ts文件for index, url in enumerate(segments):file = os.path.join(segment_dir, f"{index}.{url.split('.')[-1]}")self.pool.spawn(self.download_segment, url, file)gevent.wait()# ts文件下载完成, 转换成mp4文件if len(segments) == len(os.listdir(segment_dir)):self.convert_mp4(path, output, key=[item.uri for item in m3u8_obj.keys if item])if __name__ == "__main__":cur_path = os.path.abspath(os.path.dirname(__file__))data = [{"name": "凶劫601航班第01集", "url": "https://v4.mstopq.com/202404/11/4cSTt8dMgB7/video/index.m3u8"},{"name": "凶劫601航班第02集", "url": "https://v4.mstopq.com/202404/11/JtNttP8HfS7/video/index.m3u8"},{"name": "凶劫601航班第03集", "url": "https://v4.mstopq.com/202404/11/b10TUD8C4T7/video/index.m3u8"},{"name": "凶劫601航班第04集", "url": "https://v4.mstopq.com/202404/11/eWLuKWqFBL7/video/index.m3u8"},{"name": "凶劫601航班第05集", "url": "https://v4.mstopq.com/202404/11/Sj3pwtX0hN7/video/index.m3u8"},{"name": "凶劫601航班第06集", "url": "https://v4.mstopq.com/202404/11/i62BN8wMfc7/video/index.m3u8"},]for item in data:url = item["url"]name = item["name"]path = os.path.join(cur_path, "凶劫601航班_tmp", f"{name}") output = os.path.join(cur_path, "凶劫601航班", f"{name}.mp4")# 因为是多线程下载，可能存在某个线程下载失败的情况， 如果发现下载不完整，没有输出文件，可以尝试重新运行，已经下载过的不会再次下载。download = M3u8Downloader(pool_size=20)download.download(url, path, output)

查看原文：一键下载 M3U8 并转换为 MP4升级版

关注公众号 "字节航海家" 及时获取最新内容