首先，解析m3u8文件，将其密钥和所有视频地址存储下来。

import os
# 创建电影名称文件夹
def makedir_video(video_name: str):
    if not os.path.exists(video_name):
        os.mkdir(video_name)
# 获取所有的视频地址
def get_videos_url(video_name, m3u8_url: str):
    res = requests.get(m3u8_url, headers=headers)
    # 保存m3u8文件
    m3u8_filename = m3u8_url.split(r'/')[-1]
    with open(os.path.join(video_name, m3u8_filename), 'w', encoding='GBK') as f:
        f.write(res.text)
    all_videos_url = []
    for i in res.text.strip().split('\n'):
        # 判断视频是否加密，如果加密则将秘钥存储下来文件名为：enc.key，如果未加密则不存储任何文件
        if 'AES' in i and 'KEY' in i:
            key_uri = re.findall('URI="(.*?)"', i)[0]
            if "http" not in key_uri:
                key_uri = r'/'.join(m3u8_url.split(r'/')[:-1]) + '/' + key_uri
            key = requests.get(key_uri, headers=headers).content
            with open(os.path.join(video_name, 'enc.key'), 'wb') as f:
                f.write(key)
        if i.startswith('#') or len(i) == 0:
            continue
        all_videos_url.append(i)
    print(all_videos_url)
    return all_videos_url
if __name__ == '__main__':
    url = 'https://www.66s.cc/e/DownSys/play/?classid=4&id=20778&pathid1=0&bf=0'  # 66v电影网视频地址
    # 获取m3u8地址和电影名称
    # get_m3u8_url(url)
    m3u8_url, video_name = get_m3u8_url(url)
    # # # # 创建电影名称文件夹
    makedir_video(video_name)
    # # # # 根据m3u8地址，获取所有的视频地址，并判断是否加密
    all_videos = get_videos_url(video_name, m3u8_url)
运行完毕后再对应电影的文件夹下会生成index.m3u8文件和enc.key文件
 
现在我们已经有了所有的视频地址，接下来只需要根据视频地址将所有的视频下载下来即可。
 
由于总共视频片段较多，如果使用传统的代码运行方式进行运行，将会十分耗时，故在这里我们采用协程异步操作来进行访问和下载。
 
由于使用协程会对服务器产生较大的访问压力，所有在这里我们引入信号量去控制访问协程的并发数，减轻服务器访问压力。
 
其次，由于这里的请求采用的是异步操作，所有很有可能出现中间部分视频片段第一次下载失败的情况，在这里我们处理的方式是，先让其重复访问10次，如果10次还没有访问到视频资源，则将其记录到对应的文件中，便于后续对其进行进一步分析，为什么没有访问到资源？
 
import asyncio
import aiofiles
import aiohttp
async def download_video(filepath, video_url, sem):
    async with sem:  # 使用信号量控制访问频率
        for i in range(10):
            try:
                video_name = video_url.split(r'/')[-1]
                async with aiohttp.ClientSession() as session:
                    async with session.get(video_url, headers=headers) as res:
                        content = await res.content.read()
                        async with aiofiles.open(os.path.join(filepath, video_name), 'wb') as f:
                            await f.write(content)
                break
            except Exception as e:
                if i == 9:
                    with open(filepath+'_Error.txt', 'a', encoding='utf-8') as f:
                        f.write(video_url+'\n')
                        print(f'----- {video_name}下载失败，请求次数达到上限（{i+1}次），已写入文件{filepath+"_Error.txt"} -----')
                    break
                print(f'----- {video_name}下载失败，正在重试{i} -----')
                print(e)
async def download_all_videos(sem_num, filepath, all_videos_url):
    # 信号量, 用来控制协程的并发量
    sem = asyncio.Semaphore(sem_num)  # 极个别电影需要控制在5左右
    tasks = []
    for video_url in all_videos_url:
        tasks.append(asyncio.create_task(download_video(filepath, video_url, sem)))
    await asyncio.wait(tasks)
if __name__ == '__main__':
    url = 'https://www.66s.cc/e/DownSys/play/?classid=4&id=20778&pathid1=0&bf=0'  # 66v电影网视频地址
    # 获取m3u8地址和电影名称
    # get_m3u8_url(url)
    m3u8_url, video_name = get_m3u8_url(url)
    # # # # 创建电影名称文件夹
    makedir_video(video_name)
    # # # # 根据m3u8地址，获取所有的视频地址，并判断是否加密
    all_videos = get_videos_url(video_name, m3u8_url)
    # # # # 根据视频地址下载视频
    sem_num = 100  # 用于控制并发量
    event_loop = asyncio.get_event_loop()
    event_loop.run_until_complete(download_all_videos(sem_num, video_name, all_videos))
可以看到数据已经下载成功。




    
 
并且是有部分片段第一次是无法获取到数据的，并且越往后走这种情况出现得越多。我们也可以在请求过程中加入timeout参数，控制代码一定时间内无响应的话就中断请求，重新请求，加快效率。否则最后几个包可能需要耗费大量的时间才能抓下来。
 
但是大部分经过二次访问都可以获取到视频片段。
 
例如这里的Hda3SBaf.ts片段，第一次没有获取到，但是第二次获取到了，在文件夹中也能搜索到。
 
跟我们前面分析的一样，这里的视频文件是无法打开的。
 
而且前面我们已经将其对应的密钥文存储到了enc.key文件中。直接使用该文件对视频进行解密即可。
 
由于解密过程是需要先将视频读取进来，解密完成后还要将其再次存储下来，同样需要大量的IO操作，故也使用的是协程异步的方式去操作。
 
解密使用的是AES算法，要调用AES进行解密需要先下载pycryptodome，下载完毕后即可使用from Crypto.Cipher import AES方式进行调用。
 
这里为了适配没有加密的视频，做了一个判断，是否有加密文件存在，如果有才是加密的视频。如果没有则直接跳过此步骤。
 
from Crypto.Cipher import AES  # pip install pycryptodome
async def parse_video(video_file, video_name, new_video_name, key):
    print(os.path.join(video_name, video_file))
    print(os.path.join(new_video_name, video_file))
    async with aiofiles.open(os.path.join(video_name, video_file), 'rb') as f1, aiofiles.open(os.path.join(new_video_name, video_file), 'wb') as f2:
        content = await f1.read()
        # aes解密视频
        aes = AES.new(key=key, mode=AES.MODE_CBC, IV=b'0000000000000000')
        new_content = aes.decrypt(content)
        new_content = PKCS7_unpad(new_content)
        await f2.write(new_content)
        print(f'------ {video_file}解密成功 ------')
async def parse_all_videos(video_name):
    all_file = os.listdir(video_name)
    new_video_name = video_name + '_parse'
    if 'enc.key' in all_file:
        video_files = [i for i in all_file if i.endswith('ts')]
        makedir_video(new_video_name)
        print('------ 开始解密视频 ------')
        # 读取秘钥
        with open(os.path.join(video_name, 'enc.key'), 'rb') as f:
            key = f.read()
        # 创建协程任务
        tasks = []
        for video_file in video_files:
            tasks.append(asyncio.create_task(parse_video(video_file, video_name, new_video_name, key)))
        await asyncio.wait(tasks)
        print('------ 视频解密完成 ------')
    else:
        os.rename(video_name, new_video_name)
        print('------ 视频无加密 ------')
if __name__ == '__main__':
    url = 'https://www.66s.cc/e/DownSys/play/?classid=4&id=20778&pathid1=0&bf=0'  # 66v电影网视频地址
    # 获取m3u8地址和电影名称
    # get_m3u8_url(url)
    m3u8_url, video_name = get_m3u8_url(url)
    # # # # 创建电影名称文件夹
    makedir_video(video_name)
    # # # # 根据m3u8地址，获取所有的视频地址，并判断是否加密
    all_videos = get_videos_url(video_name, m3u8_url)
    # # # # 根据视频地址下载视频
    sem_num = 100  # 用于控制并发量
    event_loop = asyncio.get_event_loop()
    event_loop.run_until_complete(download_all_videos(sem_num, video_name, all_videos))
    # # # # 解密所有视频
    event_loop = asyncio.get_event_loop()
    event_loop.run_until_complete(parse_all_videos(video_name))
解密成功，在统计目录下会生成一个parse文件夹




    
 
视频合并可以采用windows或者Linux中自带的命令进行完成。
 
 
windows 自带合成命令: copy /b a.ts+b.ts full.mp4
 
 
linux/mac 命令: cat a.ts b.ts c.ts > xxx.mp4
 
 
并且视频的合并应该按照播放顺序进行合并，即m3u8文件的视频顺序。
 
def merge(video_name):
    new_video_name = video_name + '_parse'
    # 读取m3u8文件，获取文件正确顺序
    m3u8_file = [i for i in os.listdir(video_name) if i.endswith('m3u8')][0]    with open(os.path.join(video_name, m3u8_file)) as f:
        video_sort = [i.split('/')[-1].strip() for i in f.readlines() if not i.startswith("#") and len(i) > 0]
    n = 1
    # 切换工作目录
    os.chdir(new_video_name)
    tmp = []
    for i in range(len(video_sort)):
        tmp.append(video_sort[i])
        if i != 0 and i % 20 == 0:
            # 每20个视频合并一次
            cmd = f"copy /b {'+'.join(tmp)} {n}_copy.ts"
            os.system(cmd)
            tmp = []  # 新列表
            n = n + 1
    # 有剩余时将剩下的也合并完
    cmd = f"copy /b {'+'.join(tmp)} {n}_copy.ts"
    os.system(cmd)
    n = n + 1
    # 第二次大合并  1_copy.ts + 2_copy.ts + 3_copy.ts xxx.mp4
    last_temp = []
    for i in range(1, n):
        last_temp.append(f"{i}_copy.ts")
    # 最后一次合并
    cmd = f"copy /b {'+'.join(last_temp)} {video_name}.mp4"
    os.system(cmd)
if __name__ == '__main__':
    url = 'https://www.66s.cc/e/DownSys/play/?classid=4&id=20778&pathid1=0&bf=0'  # 66v电影网视频地址
    # 获取m3u8地址和电影名称
    # get_m3u8_url(url)
    m3u8_url, video_name = get_m3u8_url(url)
    # # # # 创建电影名称文件夹
    makedir_video(video_name)
    # # # # 根据m3u8地址，获取所有的视频地址，并判断是否加密
    all_videos = get_videos_url(video_name, m3u8_url)
    # # # # 根据视频地址下载视频
    sem_num = 100  # 用于控制并发量
    event_loop = asyncio.get_event_loop()
    event_loop.run_until_complete(download_all_videos(sem_num, video_name, all_videos))
    # # # # 解密所有视频
    event_loop = asyncio.get_event_loop()
    event_loop.run_until_complete(parse_all_videos(video_name))
    # # # 合并视频
    merge(video_name)
打开文件夹可以看到文件夹内有一个MP4文件。即为完整的视频资源。
 
打开后也可以直接播放
 
由于网吧电影网站已经关闭，经过查询将案例目标修改为6v电影  https://www.66s.cc/
1. 获取m3u8文件地址
2. 根据m3u8文件，获取所有分段视频
3. 观察视频是否经过加密，如果经过加密则需要进行解密
4. 合并所有解密视频
import asyncio
import aiofiles
import aiohttp
import requests
from lxml import etree
import re
import os
from Crypto.Cipher import AES  # pip install pycryptodome
# 设置请求头
headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/111.0.0.0 Safari/537.36',
def get_m3u8_url(url: str):
    这里m3u8地址有多种情况
    1. iframe直接链接到m3u8文件
    2. 通过一个中间包
    3. 通过两个中间包
    res = requests.get(url, headers=headers)
    res_html = etree.HTML(res.text)
    # 获取电影名称
    video_name = res_html.xpath('//title/text()')[0].split('-')[0].strip()
    # 获取iframe地址
    iframe_url = str(res_html.xpath('//iframe/@src')[0])
    # iframe直接链接到m3u8文件
    if iframe_url.endswith('m3u8'):
        return iframe_url, video_name
    # 通过一个中间包，获取m3u8文件地址
    iframe_res = requests.get(iframe_url, headers=headers)
    m3u8_url = re.findall("url: '(.*?)',", iframe_res.text, re.S)
    if len(m3u8_url) == 0:
        m3u8_url = iframe_url.rsplit('/', maxsplit=2)[0] + re.findall('"url":"(.*?)"', iframe_res.text, re.S)[0]
    else:
        m3u8_url = m3u8_url[0]
    # print(m3u8_url)
    return m3u8_url, video_name
def makedir_video(video_name: str):
    if not os.path.exists(video_name):
        os.mkdir(video_name)
def get_videos_url(video_name, m3u8_url: str):
    res = requests.get(m3u8_url, headers=headers)
    # 保存m3u8文件
    m3u8_filename = m3u8_url.split(r'/')[-1]
    with open(os.path.join(video_name, m3u8_filename), 'w', encoding='GBK') as f:
        f.write(res.text)
    all_videos_url = []
    for i in res.text.strip().split('\n'):
        # 判断视频是否加密，如果加密则将秘钥存储下来文件名为：enc.key，如果未加密则不存储任何文件
        if 'AES' in i and 'KEY' in i:
            key_uri = re.findall('URI="(.*?)"', i)[0]
            if "http" not in key_uri:
                key_uri = r'/'.join(m3u8_url.split(r'/')[:-1]) + '/' + key_uri
            key = requests.get(key_uri, headers=headers).content
            with open(os.path.join(video_name, 'enc.key'), 'wb') as f:
                f.write(key)
        if i.startswith('#') or len(i) == 0:
            continue
        all_videos_url.append(i)
        # print(all_videos_url)
    return all_videos_url
async def download_video(filepath, video_url, sem):
    async with sem:  # 使用信号量控制访问频率
        for i in range(10):
            try:
                video_name = video_url.split(r'/')[-1]
                async with aiohttp.ClientSession() as session:
                    async with session.get(video_url, headers=headers, timeout=10) as res:
                        content = await res.content.read()
                        async with aiofiles.open(os.path.join(filepath, video_name), 'wb') as f:
                            await f.write(content)
                break
            except Exception as e:
                if i == 9:
                    with open(filepath+'_Error.txt', 'a', encoding='utf-8') as f:
                        f.write(video_url+'\n')
                        print(f'----- {video_name}下载失败，请求次数达到上限（{i+1}次），已写入文件{filepath+"_Error.txt"} -----')
                    break
                print(f'----- {video_name}下载失败，正在重试{i} -----')
                print(e)
async def download_all_videos(sem_num, filepath, all_videos_url):
    # 信号量, 用来控制协程的并发量
    sem = asyncio.Semaphore(sem_num)  # 极个别电影需要控制在5左右
    tasks = []
    for video_url in all_videos_url:
        tasks.append(asyncio.create_task(download_video(filepath, video_url, sem)))
    await asyncio.wait(tasks)
async def parse_video(video_file, video_name, new_video_name, key):
    print(os.path.join(video_name, video_file))
    print(os.path.join(new_video_name, video_file))
    async with aiofiles.open(os.path.join(video_name, video_file), 'rb') as f1, aiofiles.open(os.path.join(new_video_name, video_file), 'wb') as f2:
        content = await f1.read()
        aes = AES.new(key=key, mode=AES.MODE_CBC, IV=b'0000000000000000')
        new_content = aes.decrypt(content)
        await f2.write(new_content)
        print(f'------ {video_file}解密成功 ------')
async def parse_all_videos(video_name):
    all_file = os.listdir(video_name)
    new_video_name = video_name + '_parse'
    if 'enc.key' in all_file:
        video_files = [i for i in all_file if i.endswith('ts')]
        makedir_video(new_video_name)
        print('------ 开始解密视频 ------')
        # 读取秘钥
        with open(os.path.join(video_name, 'enc.key'), 'rb') as f:
            key = f.read()
        # 创建协程任务
        tasks = []
        for video_file in video_files:
            tasks.append(asyncio.create_task(parse_video(video_file, video_name, new_video_name, key)))
        await asyncio.wait(tasks)
        print('------ 视频解密完成 ------')
    else:
        os.rename(video_name, new_video_name)
        print('------ 视频无加密 ------')
def merge(video_name):
    new_video_name = video_name + '_parse'
    # 读取m3u8文件，获取文件正确顺序
    m3u8_file = [i for i in os.listdir(video_name) if i.endswith('m3u8')][0]
    with open(os.path.join(video_name, m3u8_file)) as f:
        video_sort = [i.split('/')[-1].strip() for i in f.readlines() if not i.startswith("#") and len(i) > 0]
    n = 1
    # 切换工作目录
    os.chdir(new_video_name)
    tmp = []
    for i in range(len(video_sort)):
        tmp.append(video_sort[i])
        if i != 0 and i % 20 == 0:
            # 每20个视频合并一次
            cmd = f"copy /b {'+'.join(tmp)} {n}_copy.ts"
            os.system(cmd)
            tmp = []  # 新列表
            n = n + 1
    # 有剩余时将剩下的也合并完
    cmd = f"copy /b {'+'.join(tmp)} {n}_copy.ts"
    os.system(cmd)
    n = n + 1
    # 第二次大合并  1_copy.ts + 2_copy.ts + 3_copy.ts xxx.mp4
    last_temp = []
    for i in range(1, n):
        last_temp.append(f"{i}_copy.ts")
    # 最后一次合并
    cmd = f"copy /b {'+'.join(last_temp)} {video_name}.mp4"
    os.system(cmd)
if __name__ == '__main__':
    url = 'https://www.66s.cc/e/DownSys/play/?classid=4&id=20778&pathid1=0&bf=0'  # 66v电影网视频地址
    # 获取m3u8地址和电影名称
    # get_m3u8_url(url)
    m3u8_url, video_name = get_m3u8_url(url)
    # 创建电影名称文件夹
    makedir_video(video_name)
    # 根据m3u8地址，获取所有的视频地址，并判断是否加密
    all_videos = get_videos_url(video_name, m3u8_url)
    # 根据视频地址下载视频
    sem_num = 1000  # 用于控制并发量
    event_loop = asyncio.get_event_loop()
    event_loop.run_until_complete(download_all_videos(sem_num, video_name, all_videos))
    # 解密所有视频
    event_loop = asyncio.get_event_loop()
    event_loop.run_until_complete(parse_all_videos(video_name))
    # 合并视频
    # windows 自带合成命令: copy /b a.ts+b.ts full.mp4
    # linux/mac 命令: cat a.ts b.ts c.ts > xxx.mp4
    merge(video_name)
				
情景：女友买的学习视频将在一个月后到期（到期后下载在本地也无法看），让我帮忙把视频下载下来，之前抓过m3u8文件下载过视频切片合成后是一个完整视频，以为这次的任务非常简单~
然鹅，查看一下app信息，已经被加固处理（伪加固）
已经加固了，暂时不考虑脱壳编译
于是开始抓包，我的安卓手机没有root，在抓取某课app时由于 检测到代理导致某课app里面没网络，之前在玩安卓逆向的时候偶然发现
部分APP可以放在容器中，通过抓取容器获得运行APP的抓包数据
也就是用把 xx 安装在 VirtualXposed.
				
这段时间为了下载网页视频（网课、电影等），接触到了 m3u8 文件。在折腾了一段时间之后，猫君越发喜爱上这个格式的文件。今儿特意将关于 m3u8 格式视频链接获取方式、下载方法，以及相关的进阶操作整理成文。若诸位看官有什么更妙的使用方案，不妨分享一二。
恐怕有些朋友对 m3u8 文件还不甚了解，先作个科普吧——
m3u8 是 HLS 协议的部分内容，而 HLS 又是由苹果公司提出的基于 HTTP 的流媒体网络传输协议。其实现的基本原理是将一个大的媒体文件进行分片，而分片文件资源的路径记录于 m3u8 文
				
超给力的网站资源分享给你们，娱乐、生活、学习应有尽有，赶紧来看看吧！
千千音乐：http://music.taihe.com/songlist千千音乐是中国音乐门户之一,为你提供海量正版高品质音乐,权威的音乐榜单,快速的新歌速递,契合你的主题电台,人性化的歌曲搜索,让你更快地找到喜爱的音乐。
6V电影网：http://www.hao6v.com/每×××集互联网最新电影和电视剧,为使用迅雷软件的...
				
《Python网络爬虫技术案例教程》PPT课件(共10单元)七单元爬取APP和PC客户端数据.pdf《Python网络爬虫技术案例教程》PPT课件(共10单元)七单元爬取APP和PC客户端数据.pdf《Python网络爬虫技术案例教程》PPT课件(共10单元)七单元爬取APP和PC客户端数据.pdf《Python网络爬虫技术案例教程》PPT课件(共10单元)七单元爬取APP和PC客户端数据.pdf《Python网络爬虫技术案例教程》PPT课件(共10单元)七单元爬取APP和PC客户端数据.pdf《Python网络爬虫技术案例教程》PPT课件(共10单元)七单元爬取APP和PC客户端数据.pdf
				
python爬虫相关：
由于很多网站上的视频只提供在线观看，没有下载入口，故有必要进行网络爬虫获取视频资源。
利用requests获取网页源代码中的m3u8链接，对链接进行逐步解析，获取ts列表，下载所有ts文件，将其合并生成mp4文件。做到对视频的爬取。
同名博文相关代码。
				
不知道大家平常喜不喜欢待在宿舍一个人看电影？
作为一个高龄屌丝，电影对我来说是必不可少的。平常无聊时自己一个人待在宿舍看看电影，看看书。
（人闲下来就会胡思乱想，不能让寂寞侵蚀自己的内心）
其实还是喜欢和朋友一起去看电影的，更有气氛，有感觉点。可惜几个朋友都和我一样是钢铁男，极其尴尬(⊙o⊙)！
哈哈哈。。。
所以呢，我都是自己在网上找些电影资源来消遣时间，像电影天堂、6V电影网等等（你们有什...
对于爬虫反爬机制m3u8视频流的解决方案，有以下几种：
1. 使用代理：使用代理服务器可以隐藏真实IP地址，从而避免被网站识别为爬虫。同时，可以使用多个代理轮流访问目标网站，以降低被封禁的风险。
2. 随机UA：修改User-Agent可以模拟不同的浏览器访问目标网站，从而避免被识别为爬虫。可以使用随机UA的方式来增加访问的随机性。
3. 模拟人类行为：通过模拟人类的浏览行为，如随机点击、滑动、停留等，可以更好地隐藏爬虫的特征，以降低被封禁的风险。
4. 解密m3u8链接：有些网站会对m3u8视频流进行加密，需要先解密才能进行下载。可以使用相关工具对加密的链接进行解密。
希望这些解决方案能够帮助您解决爬虫反爬机制m3u8视频流的问题。