M3U8流视频数据爬虫

• HLS技术介绍

  • 现在大部分视频客户端都采用HTTP Live Streaming，而不是直接播放MP4等视频文件（HLS，Apple为了提高流播效率开发的技术）。HLS技术的特点是将流媒体切分为若干【TS片段】（比如几秒一段），然后通过一个【M3U8列表文件】将这些TS片段批量下载供客户端播放器实现实时流式播放。因此，在爬取HLS的流媒体文件的思路一般是先【下载M3U8文件】并分析其中内容，然后在批量下载文件中定义的【TS片段】，最后将其【组合】成mp4文件或者直接保存TS片段。
  • 说说简单，其实在实际操作中，会遇到很多复杂的问题，例如m3u8文件下载不下来，ts片段文件被加密了，甚至加密ts片段的密钥也被加密了。

• HLS的作用

  • HLS技术目前在主流的应用产品中被应用的比较广。主要的原因在于，HLS技术比传统的流媒体技术的好处在于，视频一旦切分完成，之后的分发过程完全不需要额外使用任何专门软件，仅仅普通的Web服务器即可，这样就降低了对服务器的技术要求。
  • 另外，用TS做流媒体封装还有一个好处，就是不需要加载完整视频后播放，大大减少了首次载人的延迟，提升了用户体验。
  • 此外，HTTP Live Streaming的最大优势为自适应码率流播。客户端会根据网络状况自动选择不同码率的视频流，条件允许的情况下使用高码率，网络繁忙时使用低码率，并且自动在二者间随意切换。这对移动设备网络状况不稳定的情况下保障流畅播放非常有帮助。

• M3U8文件详解

  • 如果想要爬取HLS技术下的资源数据，首先要对M3U8的数据结构和字段定义非常了解。M3U8是一个扩展文件格式，由M3U扩展而来。那么什么事M3U呢？

  • M3U文件

  • M3U这种文件格式，本质上说不是音频视频文件，它是音频视频文件的列表文件，是纯文本文件。
  • M3U这种文件被获取后，播放软件并不是播放它，而是根据它的记录找到媒体的网络地址进行在线播放。也就是说，M3U格式的文件只是存储多媒体播放列表，并提供了一个指向其他位置的音频视频文件的索引，播放的是那些被指向的文件。
  • M3U8也是一种M3U的扩展格式（高级的M3U，所以也属于M3U）。下面我们将了解一下M3U8中定义的几个非常重要的关键字：

#EXTM3U:每个M3U文件第一行必须是这个tag标识。(简单了解)

#EXT-X-VERSION:版本，此属性可用可不用。(简单了解)

#EXT-X-TARGETDURATION:目标持续时间，是用来定义每个TS的【最大】duration（持续时间）。(简单了解)

#EXT-X-ALLOW-CACHE是否允许允许高速缓存。(简单了解)

#EXT-X-MEDIA-SEQUENCE定义当前M3U8文件中第一个文件的序列号，每个ts文件在M3U8文件中都有固定唯一的序列号。(简单了解)

#EXT-X-DISCONTINUITY:播放器重新初始化(简单了解)

#EXT-X-KEY定义加密方式，用来加密的密钥文件key的URL，加密方法（例如AES-128），以及IV加密向量。（记住）

#EXTINF:指定每个媒体段(ts文件)的持续时间，这个仅对其后面的TS链接有效，每两个媒体段(ts文件)间被这个tag分隔开。(简单了解)

#EXT-X-ENDLIST表明M3U8文件的结束。(简单了解)

• EXT-X-KEY中的密钥文件

  • 对于大多数的M3U8视频，一般是不加密的。对于一些重要的视频服务商，他们会对其视频做加密处理。M3U8视频目前的标准加密方式是使用AES-128进行加密处理。如果视频是加密的，就会在M3U8文件中出现以下信息：

#EXT-X-KEY:METHOD=AES-128,URI="https://edu.aliyun.com/hls/2452/clef/0VqtrHq9IkTfOsLqy0iC1FP9342VZm1s",IV=0x3f1c20b9dd4459d0adf972eaba85e0a2

• 其中METHOD为加密方法，标准是AES-128。

• Key是密钥文件的下载地址（密钥为16字节大小的文件，需要下载）。

• IV是加密向量（16个字节大小的16进制数），如果没有IV值则使用b"0000000000000000"填充即可。

• 注意：Key和IV是AES加密解密的必要信息，这里我们就不用深入讲解。大家只需要知道Key和IV的值会作为解密函数的参数直接调用就可以了。如果文件中没有包含#EXT-X-KEY，则媒体文件将不会被加密。

• 实战，爬取某网站的视频资源，完整代码如下：

import requests
import asyncio
import aiohttp
import aiofiles
from urllib.parse import urljoin
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 download_m3u8(m3u8_url):
    res = requests.get(m3u8_url).text
    with open('m3u8.txt','w',encoding='utf8') as f:
        f.write(res)

async def download_one(url,sem):
    async with sem:  # 使用信号量控制访问频率
        file_name = url.split("/")[-1]
        file_path = "./解密前/" + file_name
        print(file_name, "开始工作了!")
        for i in  range(10):
            try:
                async with aiohttp.ClientSession() as session:
                    async with session.get(url,headers=headers) as resp:
                        content = await resp.content.read()
                        # 写入文件
                        async with aiofiles.open(file_path,'wb') as f:
                            await f.write(content)
                print(file_name, "下载完成!")
                break
            except Exception as e:
                print(file_name, "出错了, 马上重试", e)

async def down_all_vedio():
    # 信号量, 用来控制协程的并发量
    sem = asyncio.Semaphore(100)
    tasks = []
    # 读取文件
    with open('m3u8.txt','r',encoding='utf8') as f:
        for line in f:
            if line.startswith("#"):
                continue
            line = line.strip()

            #拼接完整的ts地址
            path = 'https://s7.fsvod1.com/20220906/Jbat53Ut/1200kb/hls/'
            ts = line.split('/')[-1]
            url = urljoin(path,ts)    #拼接一下

            #创建任务
            t = asyncio.create_task(download_one(url,sem))
            tasks.append(t)
    #统一等待
    await asyncio.wait(tasks)

def get_key():
    with open("m3u8.txt", mode="r", encoding="utf-8") as f:
        file_content = f.read()  # 读取到所有内容
        obj = re.compile(r'URI="(?P<key_url>.*?)"')
        key_url = obj.search(file_content).group("key_url")
        key_url = "https://s7.fsvod1.com" + key_url
        # print(key_url)
        resp = requests.get(key_url, headers=headers)  # 发请求, 拿秘钥
        return resp.content  # 直接拿字节. 为了解密的时候. 直接丢进去就可以了.

async def desc_one(file_path,key):
    file_name = file_path.split("/")[-1]
    new_file_path = f'./解密后/{file_name}'
    # 解密
    async with aiofiles.open(file_path, mode="rb") as f1,\
            aiofiles.open(new_file_path, mode="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(new_file_path, "解密成功")

# 解密的协程逻辑
# 读M3U8文件. 拿到文件名称和路径
# 每个ts文件一个任务
# 在每个任务中. 解密即可
async def desc_all(key):
    tasks = []
    with open("m3u8.txt", mode="r", encoding="utf-8") as f:
        for line in f:
            if line.startswith("#"):
                continue
            line = line.strip()
            file_name = line.split("/")[-1]
            file_path = "./解密前/" + file_name
            # 创建任务. 去解密
            t = asyncio.create_task(desc_one(file_path, key))
            tasks.append(t)
    await asyncio.wait(tasks)

def merge():
    # 视频片段合成
    # B站视频. 不适用这个.
    # 需要一个命令
    # windows: copy /b a.ts+b.ts+c.ts xxx.mp4
    # linux/mac: cat a.ts b.ts c.ts > xxx.mp4
    # 共同的坑:
    # 1. 执行命令 太长了不行. 需要分段合并
    # 2. 执行命令的时候. 容易出现乱码. 采用popen来执行命令. 就可以避免乱码
    # 3. 你只需要关注. 是否合并成功了
    # os.system("dir")  # 会有乱码
    # r = os.popen("dir")
    # print(r.read())  # 可以暂时性的避免乱码

    # 拿到所有文件名.和正确的合并顺序
    file_list = []
    with open("m3u8.txt", mode="r", encoding="utf-8") as f:
        for line in f:
            if line.startswith("#"):
                continue
            line = line.strip()
            file_name = line.split("/")[-1]
            file_list.append(file_name)

    # 进入到文件夹内
    os.chdir("./解密后")  # 更换工作目录
    # file_list  所有文件名称

    # 分段合并
    n = 1
    temp = []  # [a.ts, b.ts, c.ts]  =?=>  a.ts+b.ts+c.ts
    for i in range(len(file_list)):
        # 每 20 个合并一次
        file_name = file_list[i]
        temp.append(file_name)
        if i != 0 and i % 20 == 0:  # 20和一次(第一次合并有21个)
            # 可以合并一次了
            cmd = f"copy /b {'+'.join(temp)} {n}.ts"
            r = os.popen(cmd)
            print(r.read())
            temp = []  # 新列表
            n = n + 1
    # 需要把剩余的ts进行合并
    cmd = f"copy /b {'+'.join(temp)} {n}.ts"
    r = os.popen(cmd)
    print(r.read())
    n = n + 1

    # 第二次大合并  1.ts + 2.ts + 3.ts xxx.mp4
    last_temp = []
    for i in range(1, n):
        last_temp.append(f"{i}.ts")
    # 最后一次合并
    cmd = f"copy /b {'+'.join(last_temp)} 美利坚邦联.mp4"
    r = os.popen(cmd)
    print(r.read())
    # 回来
    os.chdir("../")  # ../ 上层文件夹


def main():
    # 爬取视频，url如下
    # url = 'https://vip.eeussnp.com/sogou/index554315.htm?554315-0-0'
    # 1、通过抓包，分析--拿到m3u8的地址并且下载
    # m3u8_url = 'https://s7.fsvod1.com/20220906/Jbat53Ut/1200kb/hls/index.m3u8'
    # download_m3u8(m3u8_url)
    #2、下载视频，使用协程
    # event_loop = asyncio.get_event_loop()
    # event_loop.run_until_complete(down_all_vedio())
    #3、拿秘钥
    # key = get_key()
    #4、解密视频
    # event_loop = asyncio.get_event_loop()
    # event_loop.run_until_complete(desc_all(key))
    #5、合成视频
    merge()

if __name__ == '__main__':
    main()

本栏目推荐文章
• Jmeter 如何连接mysql数据库？
• SQL Server 检测是不是数字型数据的两种方法
• 如何通过Navicat把MySQL远程数据库导入到本地数据库
• 和利时MACS6通过dat文件实现任意时间范围与测点的历史数据提取与科研分析
• 深入理解spring框架：剖析多线程模式下数据库连接
• 数据可视化是如何帮助大型企业提高效率的？
• redis 数据库
• openpyxl模块--------------------------------重复数据
• Django客户端应用1向服务端应用2发送POST请求并接收解析数据
• 数据可视化是如何帮助我们高效了解数据的？

爬虫 数据 视频 M3U8 M3爬虫 数据 视频m3u8 文件 数据 视频m3u8 工具 视频m3u8 m3 格式nginx视频m3u8 m3u8 python视频m3 视频m3u8 vue hls 示例 视频m3u8 web 前端 格式 视频m3u8 流程 视频m3u8 m3 easyplayer播放器 视频m3u8