AI 助理
备案控制台
开发者社区 大数据与机器学习 文章 正文

python爬取m3u8实战！！

2024-11-02 37
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
智能开放搜索 OpenSearch行业算法版，1GB 20LCU 1个月
推荐场景：
搭建高质量商品搜索服务
实时计算 Flink 版，5000CU*H 3个月
推荐场景：
实时发现最热Github项目
实时数仓Hologres，5000CU*H 100GB 3个月
推荐场景：
轻松玩转一站式实时仓库
简介： 本文详细介绍了如何抓取和处理m3u8视频文件，包括从网页源代码中提取m3u8文件地址、下载m3u8文件及其对应的ts片段、处理加密的ts文件以及使用ffmpeg合并视频片段。通过多线程下载和文件路径处理，确保了高效和准确的视频抓取与合并。文中还提供了具体的Python代码示例，帮助读者理解和实现整个过程。

[TOC]

抓取m3u8视频

1、思路分析

视频url：https://www.9meiju.cc/mohuankehuan/shandianxiadibaji/1-1.html

  1. 打开网址分析当前视频是由多个片段组成还是单独一个视频 如果是一个单独视频，则找到网址，直接下载即可，如果为多个片段的视频，则需要找到片段的文件进行处理，本案例以m3u8为例
  2. 找到m3u8文件后进行下载，下载后打开文件分析是否需要秘钥，需要秘钥则根据秘钥地址进行秘钥下载，然后下载所有ts文件
  3. 合并所有视频

2、实现

分析index.m3u8

  • 通过网络查找发现有俩个m3u8文件

    url分别为

    https://new.qqaku.com/20211117/iHVkqQMI/index.m3u8

    https://new.qqaku.com/20211117/iHVkqQMI/2523kb/hls/index.m3u8

    通过分析 第一个index.m3u8请求返回的内容中包含了第二个m3u8请求的url地址

    也就是说通过第一个index.m3u8url请求返回包含第二个index.m3u8文件地址，通过拼接请求第二个index.m3u8后 返回了包含当前所有ts文件的地址内容

    现在分析出了第二个真正的index.m3u8的地址，但是第一个地址从哪里来的呢，别慌，接下来我们来查找一下第一个url是从哪里来的

    image-20220708105559202

    image-20220708105618510

  • 查找第一个index.m3u8的url地址

    打开source

    发现url存在页面源代码中的js里 知道了位置，在代码中通过正则匹配就可以获取到了

    现在我们缕一下思路，通过页面源代码可以找到第一个index.m3u8的url，通过请求返回包含第二个index.m3u8文件的url内容，进行拼接，请求第二个m3u8的url，以此返回所有的ts内容

    image-20220708110048589

3、代码实现

3.1 获取最后一个m3u8的url地址

import re
from urllib.parse import urljoin

import requests

headers = {
   "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/54.0.2840.99 Safari/537.36"}

session = requests.Session()
session.get('https://www.9meiju.cc/', headers=headers)

url = 'https://www.9meiju.cc/mohuankehuan/shandianxiadibaji/1-2.html'
response = session.get(url, headers=headers)
response.encoding = 'UTF-8'
data = response.text
# print(data)
'''
<script>
var zanpiancms_player = {"player":"\/public\/","url":"https:\/\/new.qqaku.com\/20211124\/nLwncbZW\/index.m3u8","next":"https:\/\/www.9meiju.cc\/mohuankehuan\/shandianxiadibaji\/1-3.html","name":"wjm3u8","apiurl":null,"adtime":"0","adurl":"","copyright":0,"danmu":{"status":0}};
</script>
'''
# 正则抓取上面的源代码中的m3u8的url
m3u8_uri = re.search('"url":"(.+?index.m3u8)"', data).group(1).replace('\\', '')

# 写入文件 分析当前的页面源代码
with open('99.html', 'w', encoding='UTF-8') as f:
    # 写入response.content bytes二进制类型
    f.write(response.content.decode('UTF-8'))

# 请求可以获取index.m3u8文件
response = session.get(m3u8_uri, headers=headers)
with open('m3u8_uri.text', 'w', encoding='UTF-8') as f:
    # 写入response.content bytes二进制类型
    f.write(response.content.decode('UTF-8'))
response.encoding = 'UTF-8'
data = response.text

# 拆分返回的内容获取真整的index.m3u8文件的url
url = data.split('/', 3)[-1]
print(data)
print('m3u8_uri', m3u8_uri)
print('url', url)
print(urljoin(m3u8_uri, url))

3.2 多线程下载ts文件与视频合并

import time
import requests
import os
from concurrent.futures import ThreadPoolExecutor, wait

headers = {
   
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/100.0.4896.75 Safari/537.36"
}


def down_video(url, i):
    '''
    下载ts文件
    :param url:
    :param i:
    :return:
    '''
    # print(url)
    # 下载ts文件
    resp = requests.get(url, headers=headers)
    with open(os.path.join(path, str(i)+'.ts'), mode="wb") as f3:
        f3.write(resp.content)
    print('{} 下载完成！'.format(url))


def download_all_videos(url, path):
    '''
    下载m3u8文件以及多线程下载ts文件
    :param url:
    :param path:
    :return:
    '''
    # 请求m3u8文件进行下载
    resp = requests.get(url, headers=headers)
    with open("first.m3u8", mode="w", encoding="utf-8") as f:
        f.write(resp.text)
    if not os.path.exists(path):
        os.mkdir(path)
    # 开启线程 准备下载
    pool = ThreadPoolExecutor(max_workers=50)
    # 1. 读取文件
    tasks = []
    i = 0
    with open("first.m3u8", mode="r", encoding="utf-8") as f:
        for line in f:
            # 如果不是url 则走下次循环
            if line.startswith("#"):
                continue
            print(line, i)
            # 开启线程
            tasks.append(pool.submit(down_video, line.strip(), i))
            i += 1
    print(i)
    # 统一等待
    wait(tasks)


# 处理m3u8文件中的url问题
def do_m3u8_url(path, m3u8_filename="index.m3u8"):
    # 这里还没处理key的问题
    if not os.path.exists(path):
        os.mkdir(path)
    # else:
        # shutil.rmtree(path)
        # os.mkdir(path)
    with open(m3u8_filename, mode="r", encoding="utf-8") as f:
        data = f.readlines()

    fw = open(os.path.join(path, m3u8_filename), 'w', encoding="utf-8")
    abs_path = os.getcwd()
    i = 0
    for line in data:
        # 如果不是url 则走下次循环
        if line.startswith("#"):
            # 判断处理是存在需要秘钥
            fw.write(line)
        else:
            fw.write(f'{abs_path}/{path}/{i}.ts\n')
            i += 1


def merge(filePath, filename='output'):
    '''
    进行ts文件合并 解决视频音频不同步的问题 建议使用这种
    :param filePath:
    :return:
    '''
    os.chdir(path)
    cmd = f'ffmpeg -i index.m3u8 -c copy {filename}.mp4'
    os.system(cmd)



if __name__ == '__main__':
    # 抓取99美剧闪电侠
    # ts文件存储目录
    path = 'ts'
    url = 'https://new.qqaku.com/20211124/nLwncbZW/1100kb/hls/index.m3u8'
    # 下载m3u8文件以及ts文件
    download_all_videos(url, path)
    do_m3u8_url(path)
    # 文件合并
    merge(path, 'ts2')
    print('over')

注意：当前视频合并所用的工具为ffmpeg 如需安装 查看我的另外一篇博客ffmpeg的使用

3.3 合并获取上面俩个代码段的代码

import re
from urllib.parse import urljoin
import requests
import os  # 执行cmd/控制台上的命令
from concurrent.futures import ThreadPoolExecutor, wait
from retrying import retry


def get_m3u8_url(url):
    '''
    获取页面中m3u8的url
    :param url: 电影页面的url
    :return:
    '''
    session = requests.Session()
    # 访问首页获取cookie
    session.get('https://www.9meiju.cc/', headers=headers)
    # url = 'https://www.9meiju.cc/mohuankehuan/shandianxiadibaji/1-2.html'
    response = session.get(url, headers=headers)
    response.encoding = 'UTF-8'
    data = response.text
    # print(data)
    m3u8_uri = re.search('"url":"(.+?index.m3u8)"', data).group(1).replace('\\', '')

    # 写入文件 分析当前的页面源代码
    # with open('99.html', 'w', encoding='UTF-8') as f:
        # 写入response.content bytes二进制类型
        # f.write(response.content.decode('UTF-8'))

    # 请求可以获取index.m3u8文件
    response = session.get(m3u8_uri, headers=headers)
    # with open('m3u8_uri.text', 'w', encoding='UTF-8') as f:
        # 写入response.content bytes二进制类型
        # f.write(response.content.decode('UTF-8'))
    response.encoding = 'UTF-8'
    data = response.text
    # 拆分返回的内容获取真整的index.m3u8文件的url
    # 注意 一定要strip
    url = data.split('/', 3)[-1].strip()
    print(data)
    print('m3u8_uri', m3u8_uri)
    url = urljoin(m3u8_uri, url)
    print('url', url)
    return url

@retry(stop_max_attempt_number=3)
def down_video(url, i):
    '''
    下载ts文件
    :param url:
    :param i:
    :return:
    '''
    # print(url)
    # 下载ts文件
    # try:
    resp = requests.get(url, headers=headers)
    with open(os.path.join(path, str(i)+'.ts'), mode="wb") as f3:
        f3.write(resp.content)
    assert resp.status_code == 200


def download_all_videos(url, path):
    '''
    下载m3u8文件以及多线程下载ts文件
    :param url:
    :param path:
    :return:
    '''
    # 请求m3u8文件进行下载
    resp = requests.get(url, headers=headers)
    with open("index.m3u8", mode="w", encoding="utf-8") as f:
        f.write(resp.content.decode('UTF-8'))
    if not os.path.exists(path):
        os.mkdir(path)
    # 开启线程 准备下载
    pool = ThreadPoolExecutor(max_workers=50)
    # 1. 读取文件
    tasks = []
    i = 0
    with open("index.m3u8", mode="r", encoding="utf-8") as f:
        for line in f:
            # 如果不是url 则走下次循环
            if line.startswith("#"):
                continue
            print(line, i)
            # 开启线程
            tasks.append(pool.submit(down_video, line.strip(), i))
            i += 1
    print(i)
    # 统一等待
    wait(tasks)
    # 如果阻塞可以给一个超时参数
    # wait(tasks, timeout=1800)


def do_m3u8_url(path, m3u8_filename="index.m3u8"):
    # 这里还没处理key的问题
    if not os.path.exists(path):
        os.mkdir(path)
    # else:
    # shutil.rmtree(path)
    # os.mkdir(path)
    with open(m3u8_filename, mode="r", encoding="utf-8") as f:
        data = f.readlines()

    fw = open(os.path.join(path, m3u8_filename), 'w', encoding="utf-8")
    abs_path = os.getcwd()
    i = 0
    for line in data:
        # 如果不是url 则走下次循环
        if line.startswith("#"):
            fw.write(line)
        else:
            fw.write(f'{abs_path}/{path}/{i}.ts\n')
            i += 1

def merge(path, filename='output'):
    '''
    进行ts文件合并 解决视频音频不同步的问题 建议使用这种
    :param filePath:
    :return:
    '''
    os.chdir(path)
    cmd = f'ffmpeg -i index.m3u8 -c copy {filename}.mp4'
    os.system(cmd)

if __name__ == '__main__':
    headers = {
   
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/54.0.2840.99 Safari/537.36"}

    # 电影的url 返回index.m3u8的url地址
    url = get_m3u8_url('https://www.9meiju.cc/mohuankehuan/shandianxiadibaji/1-2.html')

    # ts文件存储目录
    path = 'ts'
    # 下载m3u8文件以及ts文件
    download_all_videos(url, path)
    do_m3u8_url(path)
    # 文件合并
    merge(path, '第二集')
    print('over')

4、注意事项

4.1 说明

在获取index.m3u8文件的内容时，有的文件内容会显示...jpg/png的情况，并没显示...ts，那么遇到这种情况需要单独处理 内容如下：

image-20220708111809978

这种情况使用上面的代码就无法进行正常合并，合并后的视频无法播放

但使用ffprobe分析，发现识别为png，进而导致无法正常拼接

image-20220708112001167

在这种情况下，只需要将其中PNG文件头部分全部使用FF填充，即可处理该问题

填充后的效果如图

image-20220708112029091

4.2 使用代码进行处理

# 解析伪装成png的ts
def resolve_ts(src_path, dst_path):
    '''
    如果m3u8返回的ts文件地址为
    https://p1.eckwai.com/ufile/adsocial/7ead0935-dd4f-4d2f-b17d-dd9902f8cc77.png
    则需要下面处理后 才能进行合并
    原因在于 使用Hexeditor打开后，发现文件头被描述为了PNG
    在这种情况下，只需要将其中PNG文件头部分全部使用FF填充，即可处理该问题
    :return:
    '''
    if not os.path.exists(dst_path):
        os.mkdir(dst_path)
    file_list = sorted(os.listdir(src_path), key=lambda x: int(x.split('.')[0]))
    for i in file_list:
        origin_ts = os.path.join(src_path, i)
        resolved_ts = os.path.join(dst_path, i)
        try:
            infile = open(origin_ts, "rb")  # 打开文件
            outfile = open(resolved_ts, "wb")  # 内容输出
            data = infile.read()
            outfile.write(data)
            outfile.seek(0x00)
            outfile.write(b'\xff\xff\xff\xff')
            outfile.flush()
            infile.close()  # 文件关闭
            outfile.close()
        except:
            pass
        print('resolve ' + origin_ts + ' success')

4.3 完整代码

import shutil
import time
from urllib.parse import urljoin

import requests
import os
import re
from concurrent.futures import ThreadPoolExecutor, wait


def get_m3u8_url(url):
    '''
    获取页面中m3u8的url
    :param url: 电影页面的url
    :return:
    '''
    session = requests.Session()
    # 访问首页获取cookie
    session.get('https://www.9meiju.cc/', headers=headers)
    # url = 'https://www.9meiju.cc/mohuankehuan/shandianxiadibaji/1-2.html'
    response = session.get(url, headers=headers)
    response.encoding = 'UTF-8'
    data = response.text
    # print(data)
    m3u8_uri = re.search('"url":"(.+?index.m3u8)"', data).group(1).replace('\\', '')


    # 请求可以获取index.m3u8文件
    response = session.get(m3u8_uri, headers=headers)
    response.encoding = 'UTF-8'
    data = response.text
    # 拆分返回的内容获取真整的index.m3u8文件的url
    # 注意 一定要strip
    url = data.split('/', 3)[-1].strip()
    print(data)
    print('m3u8_uri', m3u8_uri)
    url = urljoin(m3u8_uri, url)
    print('url', url)
    return url

def down_video(url, i):
    '''
    下载ts文件
    :param url:
    :param i:
    :return:
    '''
    # print(url)
    # 下载ts文件
    resp = requests.get(url, headers=headers)
    with open(os.path.join(path, str(i)+'.ts'), mode="wb") as f3:
        f3.write(resp.content)
    # print('{} 下载完成！'.format(url))


def download_all_videos(url, path):
    '''
    下载m3u8文件以及多线程下载ts文件
    :param url:
    :param path:
    :return:
    '''
    # 请求m3u8文件进行下载
    resp = requests.get(url, headers=headers)
    with open("index.m3u8", mode="w", encoding="utf-8") as f:
        f.write(resp.content.decode('UTF-8'))
    if not os.path.exists(path):
        os.mkdir(path)
    # 开启线程 准备下载
    pool = ThreadPoolExecutor(max_workers=50)
    # 1. 读取文件
    tasks = []
    i = 0
    with open("index.m3u8", mode="r", encoding="utf-8") as f:
        for line in f:
            # 如果不是url 则走下次循环
            if line.startswith("#"):
                continue
            print(line, i)
            # 开启线程
            tasks.append(pool.submit(down_video, line.strip(), i))
            i += 1
    print(i)
    # 统一等待
    wait(tasks)



# 解析伪装成png的ts
def resolve_ts(src_path, dst_path):
    '''
    如果m3u8返回的ts文件地址为
    https://p1.eckwai.com/ufile/adsocial/7ead0935-dd4f-4d2f-b17d-dd9902f8cc77.png
    则需要下面处理后 才能进行合并
    原因在于 使用Hexeditor打开后，发现文件头被描述为了PNG
    在这种情况下，只需要将其中PNG文件头部分全部使用FF填充，即可处理该问题
    :return:
    '''
    if not os.path.exists(dst_path):
        os.mkdir(dst_path)
    file_list = sorted(os.listdir(src_path), key=lambda x: int(x.split('.')[0]))
    for i in file_list:
        origin_ts = os.path.join(src_path, i)
        resolved_ts = os.path.join(dst_path, i)
        try:
            infile = open(origin_ts, "rb")  # 打开文件
            outfile = open(resolved_ts, "wb")  # 内容输出
            data = infile.read()
            outfile.write(data)
            outfile.seek(0x00)
            outfile.write(b'\xff\xff\xff\xff')
            outfile.flush()
            infile.close()  # 文件关闭
            outfile.close()
        except:
            pass
        """
        else:
            # 删除目录
            shutil.rmtree(src_path)
            # 将副本重命名为正式文件
            os.rename(dst_path, dst_path.rstrip('2'))
        """
        print('resolve ' + origin_ts + ' success')


# 处理m3u8文件中的url问题
def do_m3u8_url(path, m3u8_filename="index.m3u8"):
    # 这里还没处理key的问题
    if not os.path.exists(path):
        os.mkdir(path)

    with open(m3u8_filename, mode="r", encoding="utf-8") as f:
        data = f.readlines()

    fw = open(os.path.join(path, m3u8_filename), 'w', encoding="utf-8")
    abs_path = os.getcwd()
    i = 0
    for line in data:
        # 如果不是url 则走下次循环
        if line.startswith("#"):
            fw.write(line)
        else:
            fw.write(f'{abs_path}/{path}/{i}.ts\n')
            i += 1

def merge(path, filename='output'):
    '''
    进行ts文件合并 解决视频音频不同步的问题 建议使用这种
    :param filePath:
    :return:
    '''
    os.chdir(path)
    cmd = f'ffmpeg -i index.m3u8 -c copy {filename}.mp4'
    os.system(cmd)


if __name__ == '__main__':
    headers = {
   
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/100.0.4896.75 Safari/537.36"
    }
    url = get_m3u8_url('https://www.9meiju.cc/mohuankehuan/shandianxiadibaji/1-20.html')
    # 抓取99美剧闪电侠
    # ts文件存储目录
    path = 'ts'
    # 下载m3u8文件以及ts文件
    download_all_videos(url, path)
    # 合并png的ts文件
    src_path = path
    dst_path = path+'2'
    resolve_ts(src_path, dst_path)
    do_m3u8_url(dst_path)
    merge(dst_path, '闪电侠')
    print('over')

5、解密处理

  • 上面我们讲的是没有经过加密的 ts 文件，这些文件下载后直接可以播放，但经过AES-128加密后的文件下载后会无法播放，所以还需要进行解密。

  • 如何判断是否需要加密？观察视频网站是否有m3u8的文件传输，下载下来并打开：

    无需解密index.m3u8文件

    #EXTM3U
#EXT-X-VERSION:3
#EXT-X-TARGETDURATION:4
#EXT-X-PLAYLIST-TYPE:VOD
#EXT-X-MEDIA-SEQUENCE:0
#EXTINF:3.086,
https://hey05.cjkypo.com/20211215/FMbNtNzz/1100kb/hls/7qs6gJc0.ts
#EXTINF:2.085,
https://hey05.cjkypo.com/20211215/FMbNtNzz/1100kb/hls/rYpHhq0I.ts
#EXTINF:2.085,
https://hey05.cjkypo.com/20211215/FMbNtNzz/1100kb/hls/bfays5sw.ts

    需要解密index.m3u8文件

    #EXT-X-VERSION:3
#EXT-X-TARGETDURATION:1
#EXT-X-PLAYLIST-TYPE:VOD
#EXT-X-MEDIA-SEQUENCE:0
#EXT-X-KEY:METHOD=AES-128,URI="/20220418/671fJxOB/2000kb/hls/key.key" # 当前路径为解密秘钥的位置  需要使用代码拼凑成完整路径 进行请求 域名+/20220418/671fJxOB/2000kb/hls/key.key
#EXTINF:1.235,
/20220418/671fJxOB/2000kb/hls/kj6uqHoP.ts  # 并且这里ts的url也要拼凑完整
#EXTINF:1.001,
/20220418/671fJxOB/2000kb/hls/ZXX8LYPa.ts
#EXTINF:1.001,
/20220418/671fJxOB/2000kb/hls/sOezpD2H.ts
#EXTINF:1.001,
...

  • 如果你的文件是加密的，那么你还需要一个key文件，Key文件下载的方法和m3u8文件类似，如下所示 key.key 就是我们需要下载的 key 文件，并注意这里 m3u8 有2个，需要使用的是像上面一样存在 ts 文件超链接的 m3u8 文件

  • 下载所有 ts 文件，将下载好的所有的 ts 文件、m3u8、key.key 放到一个文件夹中，将 m3u8 文件改名为 index.m3u8，将 key.key 改名为 key.m3u8 。更改 index.m3u8 里的 URL，变为你本地路径的 key 文件，将所有 ts 也改为你本地的路径

    文件路径

    project/

    ​ ts/

    ​ 0.ts

    ​ 1.ts

    ​ ...

    ​ index.m3u8

    ​ key.m3u8

    修改后的index.m3u8内容如下所示：

    #EXTM3U
#EXT-X-VERSION:3
#EXT-X-TARGETDURATION:1
#EXT-X-PLAYLIST-TYPE:VOD
#EXT-X-MEDIA-SEQUENCE:0
#EXT-X-KEY:METHOD=AES-128,URI="/Users/xialigang/PycharmProjects/爬虫/抓取带秘钥的电影/ts/key.m3u8"
#EXTINF:1.235,
/Users/xialigang/PycharmProjects/爬虫/抓取带秘钥的电影/ts/0.ts
#EXTINF:1.001,
/Users/xialigang/PycharmProjects/爬虫/抓取带秘钥的电影/ts/1.ts
#EXTINF:1.001,
/Users/xialigang/PycharmProjects/爬虫/抓取带秘钥的电影/ts/2.ts

    处理index.m3u8内容的代码如下所示

    import time
from urllib.parse import urljoin

import requests
import os
from concurrent.futures import ThreadPoolExecutor, wait
import re

headers = {
     
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/100.0.4896.75 Safari/537.36"
}

def down_video(url, i):
'''
下载ts文件
:param url:
:param i:
:return:
'''

  # print(url)
  # 下载ts文件
  resp = requests.get(url, headers=headers)
  with open(os.path.join(path, str(i) + '.ts'), mode="wb") as f3:
      f3.write(resp.content)
  # print('{} 下载完成！'.format(url))

def download_all_videos(path, host):
'''
下载m3u8文件以及多线程下载ts文件
:param url:
:param path:
:return:
'''
if not os.path.exists(path):
os.mkdir(path)

  # 开启线程 准备下载
  pool = ThreadPoolExecutor(max_workers=50)
  # 1. 读取文件
  tasks = []
  i = 0
  with open("index.m3u8", mode="r", encoding="utf-8") as f:
      for line in f:
          # 如果不是url 则走下次循环
          if line.startswith("#"):
              continue
          line = host + line
          print(line, i)
          # 开启线程
          tasks.append(pool.submit(down_video, line.strip(), i))
          i += 1
  # 统一等待
  wait(tasks)

处理m3u8文件中的url问题

def do_m3u8_url(url, path, m3u8_filename="index.m3u8"):

  # 这里还没处理key的问题
  if not os.path.exists(path):
      os.mkdir(path)

  with open(m3u8_filename, mode="r", encoding="utf-8") as f:
      data = f.readlines()

  fw = open(os.path.join(path, m3u8_filename), 'w')
  abs_path = os.getcwd()
  i = 0
  for line in data:
      # 如果不是url 则走下次循环
      if line.startswith("#"):
          # 判断处理是存在需要秘钥
          if line.find('URI') != -1:
              line = re.sub('(#EXT-X-KEY:METHOD=AES-128,URI=")(.*?)"', f'\\1{os.path.join(abs_path, path)}/key.m3u8"',
                            line)
              host = url.rsplit('/', 1)[0]
              # 爬取key
              download_m3u8(host + '/key.key', os.path.join(path, 'key.m3u8'))
          fw.write(line)
      else:
          fw.write(f'{abs_path}/{path}/{i}.ts\n')
          i += 1

def download_m3u8(url, m3u8_filename="index.m3u8", state=0):
print('正在下载index.m3u8文件')
resp = requests.get(url, headers=headers)
with open(m3u8_filename, mode="w", encoding="utf-8") as f:
f.write(resp.text)

def merge(filePath, filename='output'):
'''
进行ts文件合并 解决视频音频不同步的问题 建议使用这种
:param filePath:
:return:
'''
os.chdir(path)
cmd = f'ffmpeg -i index.m3u8 -c copy {filename}.mp4'
os.system(cmd)

def get_m3u8_data(first_m3u8_url):
session = requests.Session()

  # 请求第一次m3u8de url
  resp = session.get(first_m3u8_url, headers=headers)
  resp.encoding = 'UTF-8'
  data = resp.text

  # 第二次请求m3u8文件地址 返回最终包含所有ts文件的m3u8
  second_m3u8_url = urljoin(first_m3u8_url, data.split('/', 3)[-1].strip())
  resp = session.get(second_m3u8_url, headers=headers)
  with open('index.m3u8', 'wb') as f:
      f.write(resp.content)
  return second_m3u8_url

if name == 'main':

  # ts文件存储目录
  path = 'ts'
  # 带加密的ts文件的 index.m3u8  url
  url = 'https://s7.fsvod1.com/20220622/5LnZiDXn/index.m3u8'
  meu8_url = get_m3u8_data(url)
  # 下载m3u8文件以及ts文件
  host = 'https://s7.fsvod1.com'   # 主机地址  用于拼凑完整的ts路径和秘钥路径
  download_all_videos(path, host)
  do_m3u8_url(meu8_url, path)

  # 文件合并
  merge(path, '奇异博士')
  print('over')

- 这样就大功告成了！我们成功解密并使用 ffmpeg 合并了这些 ts 视频片段，实际应用场景可能和这不一样，具体网站具体分析
文章标签：
密钥管理服务
Python
数据安全/隐私保护
JavaScript
关键词：
Python实战
Python m3u8
Ly001122
目录
相关文章
土木林森
|
8天前
|
数据采集 存储 JSON
Python网络爬虫：Scrapy框架的实战应用与技巧分享
【10月更文挑战第27天】本文介绍了Python网络爬虫Scrapy框架的实战应用与技巧。首先讲解了如何创建Scrapy项目、定义爬虫、处理JSON响应、设置User-Agent和代理，以及存储爬取的数据。通过具体示例，帮助读者掌握Scrapy的核心功能和使用方法，提升数据采集效率。
土木林森
46 6
土木林森
|
8天前
|
设计模式 前端开发 数据库
Python Web开发：Django框架下的全栈开发实战
【10月更文挑战第27天】本文介绍了Django框架在Python Web开发中的应用，涵盖了Django与Flask等框架的比较、项目结构、模型、视图、模板和URL配置等内容，并展示了实际代码示例，帮助读者快速掌握Django全栈开发的核心技术。
土木林森
84 44
龙大吉
|
1天前
|
数据库 Python
异步编程不再难！Python asyncio库实战，让你的代码流畅如丝！
在编程中，随着应用复杂度的提升，对并发和异步处理的需求日益增长。Python的asyncio库通过async和await关键字，简化了异步编程，使其变得流畅高效。本文将通过实战示例，介绍异步编程的基本概念、如何使用asyncio编写异步代码以及处理多个异步任务的方法，帮助你掌握异步编程技巧，提高代码性能。
龙大吉
11 4
Star时光
|
1天前
|
机器学习/深度学习 JSON API
Python编程实战：构建一个简单的天气预报应用
Python编程实战：构建一个简单的天气预报应用
Star时光
10 1
东方睿赢
|
4天前
|
前端开发 API 开发者
Python Web开发者必看！AJAX、Fetch API实战技巧，让前后端交互如丝般顺滑！
在Web开发中，前后端的高效交互是提升用户体验的关键。本文通过一个基于Flask框架的博客系统实战案例，详细介绍了如何使用AJAX和Fetch API实现不刷新页面查看评论的功能。从后端路由设置到前端请求处理，全面展示了这两种技术的应用技巧，帮助Python Web开发者提升项目质量和开发效率。
东方睿赢
13 1
众所周知
|
4天前
|
缓存 测试技术 Apache
告别卡顿！Python性能测试实战教程，JMeter&Locust带你秒懂性能优化💡
告别卡顿！Python性能测试实战教程，JMeter&Locust带你秒懂性能优化💡
众所周知
13 1
土木林森
|
9天前
|
数据可视化 开发者 Python
Python GUI开发：Tkinter与PyQt的实战应用与对比分析
【10月更文挑战第26天】本文介绍了Python中两种常用的GUI工具包——Tkinter和PyQt。Tkinter内置于Python标准库，适合初学者快速上手，提供基本的GUI组件和方法。PyQt基于Qt库，功能强大且灵活，适用于创建复杂的GUI应用程序。通过实战示例和对比分析，帮助开发者选择合适的工具包以满足项目需求。
土木林森
42 7
土木林森
|
9天前
|
数据采集 Web App开发 前端开发
Python爬虫进阶：Selenium在动态网页抓取中的实战
【10月更文挑战第26天】动态网页抓取是网络爬虫的难点，因为数据通常通过JavaScript异步加载。Selenium通过模拟浏览器行为，可以加载和执行JavaScript，从而获取动态网页的完整内容。本文通过实战案例，介绍如何使用Selenium在Python中抓取动态网页。首先安装Selenium库和浏览器驱动，然后通过示例代码展示如何抓取英国国家美术馆的图片信息。
土木林森
30 6
东方睿赢
|
6天前
|
Linux 开发者 iOS开发
Python系统调用实战：如何在不同操作系统间游刃有余🐟
本文介绍了 Python 在跨平台开发中的强大能力，通过实际例子展示了如何使用 `os` 和 `pathlib` 模块处理文件系统操作，`subprocess` 模块执行外部命令，以及 `tkinter` 创建跨平台的图形用户界面。这些工具和模块帮助开发者轻松应对不同操作系统间的差异，专注于业务逻辑。
东方睿赢
20 2
土木林森
|
9天前
|
数据采集 前端开发 中间件
Python网络爬虫：Scrapy框架的实战应用与技巧分享
【10月更文挑战第26天】Python是一种强大的编程语言，在数据抓取和网络爬虫领域应用广泛。Scrapy作为高效灵活的爬虫框架，为开发者提供了强大的工具集。本文通过实战案例，详细解析Scrapy框架的应用与技巧，并附上示例代码。文章介绍了Scrapy的基本概念、创建项目、编写简单爬虫、高级特性和技巧等内容。
土木林森
32 4

大数据与机器学习

热门文章

最新文章

  • 1
    阿里云开源离线同步工具DataX3.0介绍
  • 2
    DataV带你回顾春节前后全国空气质量变化
  • 3
    DataV回调id——从入门到精通
  • 4
    odps是什么?
  • 5
    数据仓库介绍与实时数仓案例
  • 6
    独家专访阿里集团副总裁贾扬清：我为什么选择加入阿里巴巴？
  • 7
    【玩转数据系列六】文本分析算法实现新闻自动分类
  • 8
    大数据阿里云产品的简单介绍理解
  • 9
    10年老兵带你看尽MaxCompute大数据运算挑战与实践
  • 10
    【玩转数据系列一】人口普查统计案例
  • 1
    【赵渝强老师】Hive的分区表
    10
  • 2
    【赵渝强老师】基于ZooKeeper实现Hadoop HA
    5
  • 3
    【赵渝强老师】大数据主从架构的单点故障
    8
  • 4
    【赵渝强老师】大数据生态圈中的组件
    8
  • 5
    如何在Windows和Mac上免费将蓝光转换为MKV？
    6
  • 6
    【赵渝强老师】大数据技术的理论基础
    6
  • 7
    【赵渝强老师】Spark中的RDD
    8
  • 8
    【赵渝强老师】基于RBF的HDFS联邦架构
    5
  • 9
    【赵渝强老师】Spark Streaming中的DStream
    5
  • 10
    【赵渝强老师】Spark SQL的数据模型：DataFrame
    7

    • 相关课程

    更多
  • Python Web开发基础
  • Python开发基础入门
  • Python常用数据科学库
  • Python网络爬虫实战
  • Python完全自学手册图文教程
  • Python基础快速入门实战教程

    • 相关电子书

    更多
  • From Python Scikit-Learn to Sc
  • Data Pre-Processing in Python:
  • 双剑合璧-Python和大数据计算平台的结合

    • 相关实验场景

    更多
  • Python新手入门
  • Python入门
  • Python选择及循环结构
  • Python新手入门（Anolis OS）
  • Python网络通信程序典型应用
  • 云端Python及基本操作

    • 推荐镜像

    更多
  • python-release
  • nodejs-release
  • golang
    • 下一篇
    DataWorks售前咨询