python m3u8 下载

在网络视频传输中，m3u8 文件格式被广泛应用于流媒体传输。它通过一个主文件链接多个 .ts 文件，将这些文件串联成一个视频流。本教程将详细介绍如何使用 Python 下载 m3u8 文件及其所包含的 .ts 文件，并最终合成为一个完整的视频文件。

以下是详细的代码实现以及每一部分的解释。

环境准备

首先，你需要安装 requests 库来发送网络请求，并使用 ThreadPoolExecutor 进行并发下载。可以通过以下命令安装 requests：

pip install requests

代码结构与实现

下面是代码的详细实现，分为以下几个步骤：

解析 m3u8 文件：读取并解析 .m3u8 文件，提取所有 .ts 文件的链接。
下载 .ts 文件：多线程下载每个 .ts 文件，加速下载过程。
保存文件：将下载的 .ts 文件通过回调函数保存到指定位置。

def m3u8_download(base_url: str, content: str, callback):
    m3u8_content = content

    # 解析 m3u8 内容，提取 ts 文件的 URL
    urls = []
    for line in m3u8_content.splitlines():
        if line and not line.startswith("#"):  # 跳过注释行
            if line.startswith("http"):
                urls.append(line)
            else:
                urls.append(f"{base_url}/{line}")

    ts_urls = []

    # 处理子 m3u8 文件
    if urls[-1].endswith(".m3u8"):
        for i, url in enumerate(urls):
            m3u8_response = requests.get(url)
            if m3u8_response.status_code == 200:
                sub_content = m3u8_response.content.decode()
                new_contents = []
                for sub_line in sub_content.splitlines():
                    if sub_line and not sub_line.startswith("#"):
                        if sub_line.startswith("http"):
                            ts_urls.append(sub_line)
                            new_contents.append(sub_line.split("/")[-1])
                        else:
                            ts_urls.append(f"{base_url}/{sub_line}")
                            new_contents.append(sub_line)
                    else:
                        new_contents.append(sub_line)
                # 调用回调函数，将子 m3u8 文件内容保存
                callback and callback(
                    url,
                    bytes("\n".join(new_contents), encoding="utf-8"),
                    m3u8_response.headers.get("content_type"),
                )
    else:
        ts_urls = urls
        
    # 下载 ts 文件
    ts_urls = set([url.split("?")[0] for url in ts_urls if url])
    max_workers = min(10, len(ts_urls))
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        executor.map(lambda u: _download_ts(u, callback), ts_urls)


def _download_ts(ts_url, callback):
    ts_response = requests.get(ts_url)
    if ts_response.status_code == 200:
        content = ts_response.content
        callback(ts_url, content, ts_response.headers.get("content_type"))

测试代码

完整的测试代码如下，包含了回调函数实现与主函数的调用。测试时，程序会从指定 m3u8 文件 URL 获取视频数据并保存。

if __name__ == "__main__":
    # 假设 CosStore 是一个云存储实例，用于存储下载的文件
    cos_store = CosStore.get_instant(CosStoreType.PUBLIC)
    
    # 回调函数，用于存储文件
    def call(url, content, content_type):
        file_name = url.split("/")[-1].split("?")[0]
        path = f"/m3u8_test/{file_name}"
        cos_store.put_object(path, content, ContentType=content_type)
        print(f"put {path} success")

    # 获取 m3u8 文件内容
    response = requests.get("https://tianhui.xin/test.m3u8")
    cos_store.put_object(
        f"/m3u8_test/playlist_eof.m3u8",
        response.content,
        ContentType=response.headers.get("content_type"),
    )
    
    # 下载 m3u8 文件的 ts 文件
    m3u8_download(
        "https://tianhui.xin/test",
        response.content.decode(),
        call,
    )

小结

通过本文提供的代码示例，你可以轻松地将 m3u8 文件及其对应的 .ts 文件下载并保存到指定位置。这种方法尤其适用于需要批量下载流媒体视频的场景，简单高效。