1.说明
m3u8是一种传输数据的方式，比如说一集20分钟的完整视频被分割成一千多段一两秒的小视频，客户端播放的时候是感觉是连续，但如果你要下载这集视频，那就要把一千多个小视频全都下载然后自己拼接成一个完整视频。拼接的话很简单，像格式工厂等很多软件都可以轻松完成，但要一个一个下载视频分段确实麻烦，所以我打算使用Python开启多线程下载，每个视频使用一个线程，只要你的网速够快，几秒钟下载一集视频没什么问题。
2.实现思路
2.1.m3u8文件。m3u8一般是以m3u8结尾的文件，如果是浏览器，可以按一下F12打开DevTools进行抓包获取m3u8的完整链接，下载之后提取所有视频分段的uri，为了方便操作，我们可以使用m3u8库。
2.2.加密解密。有些m3u8是加密的，但会在文件里给出秘钥的url，请求一下即可得到秘钥，秘钥一般是一个数字字母组成的字符串。一般加密算法是AES-128，我们需要借助pycryptodome库对已加密的视频进行解密操作。
2.3.视频合并。Windows系统自带的copy命令也可以合并，但是经过我测试，发现合并之后的视频可能会混乱，所以如果视频比较少，可以借助格式工厂等软件合并，如果较多，可以使用Python操作FFmpeg
2.4.限频问题。因为很多网站都会限频，也就是说，同时发起的请求个数不能超过一定值，否则服务器不会正常响应数据，所以我们可能需要限制一下并发执行的线程数，使用Python自带的BoundedSemaphore就行
2.5.显示进度条。如果只是简单的打印当前进度，感觉不够美观，我们可以借助tqdm等库实现进度条的显示
3.代码实现
此脚本需要用到的第三方库
pip install requests
pip install fake_useragent
pip install m3u8
pip install pycryptodome
pip install tqdm
参考代码
import logging
import os.path
import sys
import time
from datetime import datetime
from threading import Thread, BoundedSemaphore
import requests
from Crypto.Cipher import AES
from fake_useragent import UserAgent
import m3u8
from tqdm import tqdm
class M3U8Loader:
def __init__(self, uri, base_url, segments):
self.uri = uri
self.base_url = base_url
self.segments = segments
@classmethod
def load(cls, uri, base_url=None):
if uri.startswith("http"):
res = requests.get(uri, headers={"User-Agent": get_user_agent()})
if res.status_code != 200:
raise Exception(f"load u3u8 failed when download file, uri: {uri}")
text = res.text.encode().decode('unicode_escape')
segments = text.split("\n")
else:
with open(uri, encoding="utf-8") as f:
segments = f.read()
segments = segments.encode().decode('unicode_escape')
segments = segments.split("\n")
if not base_url:
for line in segments:
if line.startswith("http"):
base_url = os.path.split(line.split("?")[0])[0]
break
if not base_url and uri.startswith("http"):
base_url = str(uri).split("?")[0].rsplit("/", maxsplit=1)[0]
segments = [s.strip() for s in segments]
return M3U8Loader(uri, base_url, segments)
def decode_video(video_stream, key, iv):
if iv and iv and str(iv).startswith("0x") and int(iv, 16):
aes = AES.new(bytes(key, encoding='utf8'), AES.MODE_CBC, bytes(iv, encoding='utf8'))
else:
aes = AES.new(bytes(key, encoding='utf8'), AES.MODE_CBC, bytes(key, encoding='utf8'))
return aes.decrypt(video_stream)
def get_datetime_num():
return datetime.strftime(datetime.now(), "%Y%m%d%H%M%S")
def get_user_agent():
return UserAgent(path="./utils/fake_useragent_0.1.11.json").random
class M3U8Downloader:
def __init__(self, m3u8_url, base_url, save_dir, video_folder, headers, if_random_ug, merge_name, ffmpeg_path,
sp_count, if_tqdm):
self.tqdm = None
self.if_tqdm = if_tqdm
self.m3u8_url = m3u8_url
self.base_url = base_url if base_url else ""
self.to_download_url = list()
self.download_failed_dict = dict()
self.key_method = None
self.key_iv = None
self.key_str = None
self.current_file_path = os.path.dirname(os.path.abspath(__file__))
self.save_dir = save_dir if save_dir else os.path.join(self.current_file_path, "m3u8_download")
self.video_folder = video_folder if video_folder else get_datetime_num()
if not os.path.isabs(ffmpeg_path):
ffmpeg_path = os.path.join(self.current_file_path, ffmpeg_path)
self.headers = headers if isinstance(headers, dict) else dict()
self.if_random_ug = if_random_ug if isinstance(if_random_ug, bool) else True
self.ffmpeg_path = ffmpeg_path
self.merge_name = merge_name if merge_name else "merge.ts"
self.file_type = ".ts"
self.semaphore = BoundedSemaphore(sp_count) if sp_count else None
self.logger = self.get_logger()
self.normalize_m3u8_file(self.m3u8_url)
self.normalize_base_url()
self.logger.info(f"init info m3u8_url: {self.m3u8_url}")
self.logger.info(f"init info base_url: {self.base_url}")
self.logger.info(f"init info if_random_ug: {self.if_random_ug}")
self.logger.info(f"init info headers: {self.headers}")
self.logger.info(f"init info save_dir: {self.save_dir}")
self.logger.info(f"init info video_folder: {self.video_folder}")
self.logger.info(f"init info current_file_path: {self.current_file_path}")
self.logger.info(f"init info ffmpeg_path: {self.ffmpeg_path}")
self.logger.info(f"init info merge_name: {self.merge_name}")
def __del__(self):
if self.tqdm:
self.tqdm.close()
def get_headers(self):
headers = self.headers
if self.if_random_ug:
headers.update({"User-Agent": get_user_agent()})
return headers
def get_logger(self):
logger = logging.getLogger("M3U8Downloader")
logger.setLevel(logging.INFO)
formatter = logging.Formatter("%(asctime)s-%(filename)s-line:%(lineno)d-%(levelname)s-%(process)s: %(message)s")
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
console_handler.setFormatter(formatter)
if not os.path.exists(self.save_dir):
os.mkdir(self.save_dir)
file_handler = logging.FileHandler(os.path.join(self.save_dir, "m3u8_download.log"), encoding="utf-8")
file_handler.setLevel(logging.INFO)
file_handler.setFormatter(formatter)
logger.addHandler(console_handler)
logger.addHandler(file_handler)
return logger
def get_m3u8_info(self):
m3u8_obj = m3u8.load(self.m3u8_url, timeout=10, headers=self.get_headers())
keys = m3u8_obj.keys
if keys and keys[-1]:
key_alg = keys[-1].method
if key_alg != "AES-128":
raise Exception(f"matched key but algorithm ({key_alg}) is not AES-128")
self.key_method = key_alg
self.key_iv = keys[-1].iv
self.get_key(self.normalize_url(keys[-1].absolute_uri))
self.to_download_url = [self.normalize_url(segment.uri) for segment in m3u8_obj.segments]
self.to_download_url = [d_url for d_url in self.to_download_url if d_url]
if not self.to_download_url:
loader_obj = M3U8Loader.load(self.m3u8_url, self.base_url)
self.to_download_url = [self.normalize_url(segment) for segment in loader_obj.segments]
self.to_download_url = [d_url for d_url in self.to_download_url if d_url]
self.logger.info(f"to_download_url: {len(self.to_download_url)} {self.to_download_url[:5]}, ...")
self.tqdm = tqdm(total=len(self.to_download_url), desc="download progress") if self.if_tqdm else None
if self.to_download_url:
self.file_type = os.path.splitext(self.to_download_url[0].split("?")[0])[1]
def get_key(self, key_url):
self.logger.info(f"key_url: {key_url}")
res = requests.get(key_url, headers=self.get_headers(), timeout=10)
self.key_str = res.text
if not self.key_str:
raise Exception("get key error, key: {}".format(self.key_str))
self.logger.info(f"get_key key_str: {self.key_str}")
def test_download(self, d_url):
self.logger.info(f"test download url: {d_url}")
try:
res = requests.get(d_url, timeout=30, headers=self.get_headers(), stream=True)
return True if res.status_code < 300 else False
except Exception as e:
self.logger.error(f"test_download meet error: {e}")
return False
def download_video(self, number, url):
if self.semaphore:
self.semaphore.acquire()
trt_times = 10
res_content = None
while trt_times > 0:
try:
res = requests.get(url, timeout=10, stream=True)
if res.status_code == 200:
res_content = res.content
break
except Exception as e:
self.logger.error(f"download failed, will try again: url:{url} ,error:{e}")
res_content = None
trt_times -= 1
time.sleep(1)
if res_content:
if self.key_str:
res_content = decode_video(res_content, self.key_str, self.key_iv)
path = os.path.join(self.save_dir, self.video_folder, "{0:0>8}".format(number) + str(self.file_type))
with open(path, "wb+") as f:
f.write(res_content)
if self.tqdm:
self.tqdm.update(1)
else:
self.logger.warning(f"download video failed, number:{number},url:{url}")
self.download_failed_dict.update({number: url})
if self.semaphore:
self.semaphore.release()
def merge_videos(self):
if os.name != "nt":
self.logger.warning(f"current system {os.name} is not Windows, can't merge.")
return
self.logger.info("start merge")
path = self.save_dir
if os.path.isabs(path):
path = self.save_dir + os.sep + self.video_folder
else:
path = self.current_file_path + os.sep + os.path.basename(self.save_dir) + os.sep + self.video_folder
if not os.path.exists(path):
self.logger.warning(f"merge_videos canceled, the path({path}) is not exist")
return
self.logger.info(f"ffmpeg path: {self.ffmpeg_path}")
all_ts_files = os.listdir(path)
all_ts_files = [ts for ts in all_ts_files if ts.startswith("0") and ts.endswith(self.file_type)]
if not all_ts_files:
self.logger.warning(f"there is no {self.file_type} file need to merge")
return
all_ts_files.sort(key=lambda x: x)
self.logger.info(f"ffmpeg path: {self.ffmpeg_path}")
if self.ffmpeg_path and os.path.exists(self.ffmpeg_path):
self.logger.info(f"use ffmpeg to merge")
with open(path + os.sep + "merge_file_list.txt", "w") as f:
for file in all_ts_files:
f.write("file " + "'" + path + os.sep + file + "'" + "\n")
cmd = "{} -f concat -safe 0 -i {} -c copy {}".format(
self.ffmpeg_path, path + os.sep + 'merge_file_list.txt', path + os.sep + self.merge_name)
self.logger.info(f"merge cmd: {cmd}")
res = os.system(cmd)
if res:
self.logger.error("merge failed")
else:
self.logger.info("merge success")
else:
self.logger.warning(f"ffmpeg not exist, will merge by python")
try:
with open(path + os.sep + self.merge_name, "wb+") as f:
for ts_file in all_ts_files:
with open(path + os.sep + ts_file, "rb+") as t:
f.write(t.read())
except Exception as e:
self.logger.error(f"merge failed: {e}")
else:
self.logger.info("merge success")
def mkdir(self):
if not os.path.exists(self.save_dir):
os.mkdir(self.save_dir)
self.logger.info(f"make save_dir({self.save_dir}) success.")
video_folder = os.path.join(self.save_dir, self.video_folder)
if not os.path.exists(video_folder):
os.mkdir(video_folder)
self.logger.info(f"make video_folder({video_folder}) success.")
def normalize_url(self, raw_url):
raw_url = raw_url.strip()
if raw_url.startswith("#"):
return
if raw_url and raw_url.startswith("http") and any(
[raw_url.split("?")[0].endswith(".ts"), raw_url.split("?")[0].endswith(".key")]):
return raw_url
if raw_url and not str(raw_url).startswith("http"):
last_find_str = ""
for i in range(1, len(raw_url) + 1):
start_str = raw_url[:i]
if self.base_url.rfind(start_str) == -1:
break
else:
last_find_str = start_str
sep = "" if self.base_url.endswith("/") or raw_url.startswith("/") else "/"
if len(last_find_str) > 2 and self.base_url.endswith(last_find_str):
raw_url = f"{self.base_url}{sep}{raw_url.replace(last_find_str, '')}"
else:
raw_url = f"{self.base_url}{sep}{raw_url}"
return raw_url
def normalize_m3u8_file(self, path):
if not os.path.exists(path):
return
with open(path, "r", encoding="utf-8") as f:
contents = f.read()
contents = contents.encode().decode('unicode_escape')
print()
with open(path, 'w', encoding='utf-8') as w:
w.write(contents.replace("'", "").replace('"', ''))
self.logger.info(f"normalize m3u8 file success, path: {path}")
return contents
def normalize_base_url(self):
if self.base_url and self.base_url.startswith('http'):
return
base_url = M3U8Loader.load(self.m3u8_url).base_url
if base_url:
self.base_url = base_url
else:
raise Exception("automatically identify base_url failed, please fill in manually")
def run(self):
start_time = time.time()
self.get_m3u8_info()
if not self.to_download_url:
self.logger.warning("there is no url to download, self.to_download_url is empty, please check url")
return
self.logger.info(f"self.key_str: {self.key_str}")
self.logger.info(f"self.key_method: {self.key_method}")
if not self.test_download(self.to_download_url[0]):
self.logger.warning(f"test download failed, pls check whether the url is valid ({self.to_download_url[0]})")
return
self.mkdir()
threads = [Thread(target=self.download_video, args=(idx, url)) for idx, url in enumerate(self.to_download_url)]
for t in threads:
t.start()
for t in threads:
t.join()
self.logger.info(f"all download finish, spent time: {time.time() - start_time:.2f} second")
self.logger.info(f"total video count: {len(self.to_download_url)}")
self.logger.info(f"download_failed_dict: {self.download_failed_dict}")
if self.download_failed_dict:
self.logger.warning(f"{len(self.download_failed_dict)} video file download failed.")
raise Exception(f"{len(self.download_failed_dict)} video file download failed.")
if self.ffmpeg_path:
self.merge_videos()
if self.tqdm:
self.tqdm.close()
if __name__ == '__main__':
url = "https://test/index.m3u8"
if len(sys.argv) > 1 and str(sys.argv[1]).startswith("http"):
url = sys.argv[1]
if not url:
raise Exception("missing download url")
params_dict = {
"m3u8_url": url,
"base_url": "",
"save_dir": "",
"video_folder": "",
"headers": {
},
"if_random_ug": True,
"ffmpeg_path": "./utils/ffmpeg.exe",
"merge_name": "",
"sp_count": 2,
"if_tqdm": True,
}
downloader = M3U8Downloader(**params_dict)
downloader.run()
该代码已上传
GitHub链接：https://github.com/panmeibing/python_downloader
本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)
