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文件下载链接直接就在页面源码中)
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比较特殊,直接就是完整的下载链接,大部分只有部分文件名,需要根据网页通过一些手段找到对应的网站域名或者网址前缀。
# 第二步,解析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.ts
resp3 = 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命令,具体细节请“百度”。
import os
import subprocess
def 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}' 不存在!")
return
lst = []
try:
with open(m3u8_file, mode='r', encoding='utf-8') as f:
for line in f:
if line.startswith('#'):
continue
line = 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文件可供合并。")
return
temp_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_files
print(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
……
"""
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_src
def get_first_m3u8_url(url):
resp = requests.get(url, headers=headers)
resp.encoding = 'utf-8'
# print(resp.text)
resp_html = resp.text
obj = re.compile(r'"url": "(?P<m3u8_url>.*?)"', re.S)
m3u8_url = obj.search(resp_html).group('m3u8_url')
print(m3u8_url)
return m3u8_url
def 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: # 提前准备好session
async with aiofiles.open('data_file/' + "second_m3u8.txt", mode='r', encoding='utf-8') as f:
async for line in f:
if line.startswith('#'): # 可能pycharm提示高亮,实际运行没有问题,不必理会
continue
line = line.strip() # 去掉没用的空格和换行
# /20240808/s1aDNcWE/933kb/hls/8IfxeFcu.ts
ts_name = line.split('/')[-1]
# 8IfxeFcu.ts
# 拼接得到真正的ts下载路径
ts_url = up_url + line
# https://vip.kuaikan-cdn4.com/20240808/s1aDNcWE/933kb/hls/8IfxeFcu.ts
task = 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.mp4
lst = []
with open('data_file/' + "second_m3u8.txt", mode='r', encoding='utf-8') as f:
for line in f:
# line = await line # 确保获取的是字符串
if line.startswith('#'):
continue
line = line.strip() # 去掉没用的空格和换行
# /20240808/s1aDNcWE/933kb/hls/8IfxeFcu.ts
ts_name = line.split('/')[-1]
# 8IfxeFcu.ts
ts_path = 'data_file/video_ts/' + ts_name # 构建完整路径
# data_file/video_ts/8IfxeFcu.ts
lst.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_files
print(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对应的url
iframe_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.m3u8
m3u8_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('#'):
continue
else:
# 去掉空白或者换行符
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.m3u8
print(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.text
key_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('#'):
continue
line = line.strip() # 去掉没用的空格和换行
# /20240808/s1aDNcWE/933kb/hls/8IfxeFcu.ts
ts_name = line.split('/')[-1]
# 8IfxeFcu.ts
# 开始创建异步任务
task = asyncio.create_task(dec_ts(ts_name, key))
tasks.append(task)
await asyncio.wait(tasks) # 等待任务结束
pass
asyncio.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("所有文件下载完毕!")