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("所有文件下载完毕!")



















