【自用】Python爬虫学习（六）：通过m3u8文件下载ts文件并合并为.mp4文件

Python爬虫 学习（六）

下载视频（简单版）的步骤介绍
- 第一步：在网页上找到.m3u8文件
- 第二步：通过.m3u8文件下载对应的.ts视频文件
- 第三步：依据.m3u8文件合并.ts文件为一个.mp4文件
下载视频（复杂版）

下载视频（简单版）的步骤介绍

"""
<video src='视频.mp4"></video:>
一般的视频网站是怎么做的？
用户上传->转码（把视频做处理，2K,1080,标清)->切片处理（把单个的文件进行拆分，形成众多的.ts文件）需要一个文件记录：1.视频播放顺序，2.视频存放的路径，这个文件就是m3u
m3u以utf-8编码存储就是m3u8文件，本质就是一个文本文件。
M3U8 txt json =>文本想要抓取一个视频：
1.找到 m3u8（各种手段)
2.通过 m3u8下载到 ts文件（这里先不管.ts是否被加密）
3.通过各种手段（不仅是编程手段）把ts文件合并为一个mp4文件
"""

第一步：在网页上找到.m3u8文件

（这里假设网页对应的.m3u8文件没有进行加密、隐藏等处理，真实的.m3u8文件下载链接直接就在页面源码中）

python">import requests
import re# 第一步，下载m3u8文件
headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127.0.0.0 Safari/537.36 Edg/127.0.0.0"
}# 测试网址，已经不能正常打开
url = "https://www.91kanju.com/vod-play/54812-1-1.html"  
# 用来提取找到m3u8的url地址的预加载正则表达式，需要根据具体网页情况编写合适的表达式
obj = re.compile(r"url: '(?P<url>.*?)',", re.S)  resp = requests.get(url)
m3u8_url = obj.search(resp.text).group("url")  # 拿到m3u8的地址# print(m3u8_url)
resp.close()# 下载m3u8文件
resp2 = requests.get(m3u8_url, headers=headers)
with open("video.m3u8", mode="wb") as f:f.write(resp2.content)resp2.close()
print("下载完毕")

第二步：通过.m3u8文件下载对应的.ts视频文件

上一步的网址不能打开，得不到对应的.m3u8文件，可以直接用下面的.m3u8文件进行测试。
测试链接：https://upyun.luckly-mjw.cn/Assets/media-source/example/media/index.m3u8
打开下载好的.m3u8文件如下所示，不带#的行就是.ts视频文件的下载地址，只需要对其进行发送请求就能下载得到对应的.ts视频文件。
注意：这里不同的.m3u8比较特殊，直接就是完整的下载链接，大部分只有部分文件名，需要根据网页通过一些手段找到对应的网站域名或者网址前缀。
在这里插入图片描述

python"># 第二步，解析m3u8文件
with open("data_file/index.m3u8", mode="r", encoding="utf-8") as f:for line in f:line = line.strip()  # 先去掉空格，空白，换行符if line.startswith("#"):  # 如果以#开头，跳过这一行continue# print(line)ts_name = line.split('/')[-1]  # test-1.tsresp3 = requests.get(line)f = open(f"data_file/video/{ts_name}", mode="wb")f.write(resp3.content)resp3.close()print(f"{ts_name}，下载成功！")

第三步：依据.m3u8文件合并.ts文件为一个.mp4文件

注意：
1、.ts文件的名称要与# .m3u8文件记录的名称一样才可以利用下面的代码进行合并。
2、这个合并代码仅适用于Windows系统，合并二进制文件使用的是copy命令，对于mac系统应该使用cat命令，具体细节请“百度”。

python">import os
import subprocessdef merge_ts_to_mp4(m3u8_file, ts_folder_path, merge_video_name):# 检查m3u8文件路径是否存在if not os.path.isfile(m3u8_file):print(f"错误：m3u8文件 '{m3u8_file}' 不存在！")return# 检查ts文件夹路径是否存在if not os.path.isdir(ts_folder_path):print(f"错误：TS文件夹 '{ts_folder_path}' 不存在！")returnlst = []try:with open(m3u8_file, mode='r', encoding='utf-8') as f:for line in f:if line.startswith('#'):continueline = line.strip()  # 去掉空格和换行ts_name = line.split('/')[-1]  # 提取文件名ts_path = os.path.join(ts_folder_path, ts_name)  # 构建完整路径# 检查每个ts文件是否存在if os.path.isfile(ts_path):lst.append(ts_path)else:print(f"警告：TS文件 '{ts_path}' 不存在，将被跳过。")if not lst:print("没有有效的TS文件可供合并。")returntemp_output_path = os.path.join(ts_folder_path, 'temp_output.ts')  # 临时文件路径total_ts_files = len(lst)print("开始合并视频文件...")for index, ts_file in enumerate(lst):command = f'copy /b "{temp_output_path}" + "{ts_file}" "{temp_output_path}"' if os.path.exists(temp_output_path) else f'copy /b "{ts_file}" "{temp_output_path}"'result = subprocess.run(command, shell=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)if result.returncode == 0:percentage = (index + 1) * 100 / total_ts_filesprint(f"{percentage:.2f}% - 已合并: {index + 1}/{total_ts_files}")else:print(f"合并失败：{ts_file}，跳过该文件。")# 最终重命名final_output_path = os.path.join(ts_folder_path, merge_video_name)os.rename(temp_output_path, final_output_path)  # 重命名临时文件为最终输出文件名# 完成合并后提示用户print(f"所有.ts文件已合并到：\n '{final_output_path}'。")except Exception as e:print(f"发生异常：{e}")if __name__ == '__main__':# .m3u8文件路径m3u8_file = r'D:\User_Data\Documents\PycharmProjects\NewFile\data_file\index.m3u8'# 从.m3u8文件下载的.ts文件的目录路径，该目录下放置下载的众多.ts文件ts_folder_path = r'D:\User_Data\Documents\PycharmProjects\NewFile\data_file\video'# 最终合并的视频名称，放在与.ts文件相同的目录下merge_video_name = 'ts视频合并.mp4'merge_ts_to_mp4(m3u8_file, ts_folder_path, merge_video_name)

运行结果如下所示：
在这里插入图片描述

下载视频（复杂版）

思路：

拿到主页面的页面源代码，找到iframe
从iframe的页面源代码中拿到m3u8文件
下载第一层m3u8文件 -->下载第二层m3u8文件（真实的视频存放路径）
下载视频
下载miyao，进行jiemi操作
合并所有ts文件为一个mp4文件

注意：演示网址的视频没有jiemi，只是yincang了真实的.m3u8下载地址，不需要jiemi，对于需要jiemi的也有代码演示，代码不可直接运行，部分存在问题，重点是理清思路，实际需要具体情况具体分析。

"""
网页播放地址：https://www.555dy16.com/vodplay/128103-7-2/
iframe对应的网页地址：https://www.555dy16.com/player/?url=https://vip.kuaikan-cdn4.com/20240808/s1aDNcWE/index.m3u8&dianshiju&next=https://www.555dy16.com/vodplay/128103-7-3/
iframe里面的m3u8地址：https://vip.kuaikan-cdn4.com/20240808/s1aDNcWE/index.m3u8
抓包里面的m3u8地址：
第一个https://vip.kuaikan-cdn4.com/20240808/s1aDNcWE/index.m3u8预览：#EXTM3U#EXT-X-STREAM-INF:PROGRAM-ID=1,BANDWIDTH=933000,RESOLUTION=1280x720/20240808/s1aDNcWE/933kb/hls/index.m3u8
第二个https://vip.kuaikan-cdn4.com/20240808/s1aDNcWE/933kb/hls/index.m3u8预览：#EXTM3U#EXT-X-VERSION:3#EXT-X-TARGETDURATION:2#EXT-X-PLAYLIST-TYPE:VOD#EXT-X-MEDIA-SEQUENCE:0#EXTINF:1,/20240808/s1aDNcWE/933kb/hls/8IfxeFcu.ts#EXTINF:1,/20240808/s1aDNcWE/933kb/hls/d4F8NT9f.ts#EXTINF:1,/20240808/s1aDNcWE/933kb/hls/W8uFMNJv.ts……"""

python">import re
import os
import requests
import asyncio
import aiohttp
import aiofiles
import subprocess
from bs4 import BeautifulSoup
from Crypto.Cipher import AES# 代码想要正常运行需要对一些位置进行适当修改，切勿直接运行！
# 对某网页的视频进行下载headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127.0.0.0 Safari/537.36 Edg/127.0.0.0"
}def get_iframe_src(url):# resp = requests.get(url, headers=headers)# resp.encoding = 'utf-8'# print(resp.text)# main_page = BeautifulSoup(resp.text, "html.parser")# iframe_src = main_page.find('iframe').get('src')# 网页原因不能找到iframe，先直接给定iframe_src = 'https://www.555dy16.com/player/?url=https://vip.kuaikan-cdn4.com/20240808/s1aDNcWE/index.m3u8&dianshiju&next=https://www.555dy16.com/vodplay/128103-7-3/'print(iframe_src)return iframe_srcdef get_first_m3u8_url(url):resp = requests.get(url, headers=headers)resp.encoding = 'utf-8'# print(resp.text)resp_html = resp.textobj = re.compile(r'"url": "(?P<m3u8_url>.*?)"', re.S)m3u8_url = obj.search(resp_html).group('m3u8_url')print(m3u8_url)return m3u8_urldef download_m3u8_file(url, file_name):resp = requests.get(url, headers=headers)with open('data_file/' + file_name, mode='wb') as f:f.write(resp.content)print(f"'{file_name}' 下载成功！")async def download_ts(ts_url, ts_name, session):async with session.get(ts_url) as resp:async with aiofiles.open(f'data_file/video_ts/{ts_name}', mode='wb') as f:await f.write(await resp.content.read())  # 下载到的内容写入到文件print(f"{ts_name} 下载完成！")async def aio_download(up_url):tasks = []async with aiohttp.ClientSession() as session:  # 提前准备好sessionasync with aiofiles.open('data_file/' + "second_m3u8.txt", mode='r', encoding='utf-8') as f:async for line in f:if line.startswith('#'):  # 可能pycharm提示高亮，实际运行没有问题，不必理会continueline = line.strip()  # 去掉没用的空格和换行# /20240808/s1aDNcWE/933kb/hls/8IfxeFcu.tsts_name = line.split('/')[-1]# 8IfxeFcu.ts# 拼接得到真正的ts下载路径ts_url = up_url + line# https://vip.kuaikan-cdn4.com/20240808/s1aDNcWE/933kb/hls/8IfxeFcu.tstask = asyncio.create_task(download_ts(ts_url, ts_name, session))  # 创建任务tasks.append(task)await asyncio.wait(tasks)  # 等待任务结束def merge_ts_to_mp4():# mac: cat 1.ts 2.ts 3.ts > xxx mp4# windows: copy /b 1.ts+2.ts+3.ts xxx.mp4lst = []with open('data_file/' + "second_m3u8.txt", mode='r', encoding='utf-8') as f:for line in f:# line = await line  # 确保获取的是字符串if line.startswith('#'):continueline = line.strip()  # 去掉没用的空格和换行# /20240808/s1aDNcWE/933kb/hls/8IfxeFcu.tsts_name = line.split('/')[-1]# 8IfxeFcu.tsts_path = 'data_file/video_ts/' + ts_name  # 构建完整路径# data_file/video_ts/8IfxeFcu.tslst.append(ts_path)# Windows系统使用copy命令temp_output_path = os.path.join('data_file/video_ts', 'temp_output.ts')  # 临时文件路径total_ts_files = len(lst)  # 所有的ts文件数量print("开始合并.ts视频文件...")for index, ts_file in enumerate(lst):# 使用加号连接文件名，Windows系统使用copy命令command = f'copy /b "{temp_output_path}" + "{ts_file}" "{temp_output_path}"' if os.path.exists(temp_output_path) else f'copy /b "{ts_file}" "{temp_output_path}"'result = subprocess.run(command, shell=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)if result.returncode == 0:percentage = (index + 1) * 100 / total_ts_filesprint(f"{percentage:.2f}% - 已合并: {index + 1}/{total_ts_files}")else:print(f"合并失败：{ts_file}，跳过该文件。")# 最终重命名final_output_path = os.path.join('data_file/video_ts', 'movies.mp4')os.rename(temp_output_path, final_output_path)  # 重命名临时文件为最终输出文件名def main(url):# 1.拿到主页面的页面源代码，找到iframe对应的urliframe_src = get_iframe_src(url)# https://www.555dy16.com/player/?url=https://vip.kuaikan-cdn4.com/20240808/s1aDNcWE/index.m3u8&dianshiju&next=https://www.555dy16.com/vodplay/128103-7-3/# 2.拿到第一层的m3u8文件的下载地址，看具体情况对拿到的地址进行拼接处理first_m3u8_url = get_first_m3u8_url(iframe_src)# https://vip.kuaikan-cdn4.com/20240808/s1aDNcWE/index.m3u8m3u8_domain = first_m3u8_url.split('.com')[0] + '.com'# 3.1.下载第一层m3u8文件first_txt_filename = "first_m3u8.txt"download_m3u8_file(first_m3u8_url, first_txt_filename)# 3.2.下载第二层m3u8文件with open('data_file/' + first_txt_filename, mode='r', encoding='utf-8') as f:for line in f:if line.startswith('#'):continueelse:# 去掉空白或者换行符line = line.strip()  # /20240808/s1aDNcWE/933kb/hls/index.m3u8# 拼接第二层m3u8的下载地址second_m3u8_url = m3u8_domain + line# https://vip.kuaikan-cdn4.com/20240808/s1aDNcWE/933kb/hls/index.m3u8print(second_m3u8_url)# 下载第二层m3u8文件second_txt_filename = "second_m3u8.txt"download_m3u8_file(second_m3u8_url, second_txt_filename)# 4.下载视频ts_domain_url = 'https://vip.kuaikan-cdn4.com'# 异步协程asyncio.run(aio_download(ts_domain_url))# =======================这一部分看网站m3u8文件具体情况，是否需要解密==========================# 关注.m3u8文件是否包含这一行：#EXT-X-KEY:METHOD=AES-128,URI="Key.Key"，# 有代表不能直接对下载的.ts文件进行合并，合并前需要对.ts文件进行解密，对解密后的.ts文件进行合并# 5.1 拿到秘钥  (后面内容仅做示范，代码不可运行，需要具体情况具体分析，为了方便理顺流程这部分函数与代码直接写在一起)def get_key(url):resp = requests.get(url)# print(resp.text)  # c5878c26baaaac8c，会得到诸如注释类似格式的文本return resp.textkey_url = 'https://vip.kuaikan-cdn4.com/……/key.key'  # 要从m3u8文件里去获取key = get_key(key_url)# 5.2 解密    (要对下载的每一个.ts文件进行解密，需要使用异步协程提高效率)async def dec_ts(ts_name, key):aes = AES.new(key=key, IV=b"0000000000000000", mode=AES.MODE_CBC)async with aiofiles.open(f'data_file/video_ts/{ts_name}', mode="rb") as f1, \aiofiles.open(f'data_file/video_ts/temp_{ts_name}', mode="wb") as f2:bs = await f1.read()  # 从源文件读取内容await f2.write(aes.decrypt(bs))  # 把解密好的内容写入文件print(f'temp_{ts_name} 处理完毕！')async def aio_dec(key):# 解密tasks = []async with aiofiles.open('data_file/' + "second_m3u8.txt", mode='r', encoding='utf-8') as f:async for line in f:# line = await line  # 确保获取的是字符串if line.startswith('#'):continueline = line.strip()  # 去掉没用的空格和换行# /20240808/s1aDNcWE/933kb/hls/8IfxeFcu.tsts_name = line.split('/')[-1]# 8IfxeFcu.ts# 开始创建异步任务task = asyncio.create_task(dec_ts(ts_name, key))tasks.append(task)await asyncio.wait(tasks)  # 等待任务结束passasyncio.run(aio_dec(key))# ======================================================================================# 6.合并ts文件merge_ts_to_mp4()  # 合并ts文件为mp4文件# 主程序
if __name__ == '__main__':url = 'https://www.555dy16.com/vodplay/128103-7-2/'main(url)print("所有文件下载完毕！")