Python 爬取 M3U8 视频文件

发布于：2025-07-25 ⋅ 阅读:(16) ⋅ 点赞:(0)

Python 爬虫入门
Python 爬虫开发
Python 爬虫工具 BeautifulSoup
Python 爬取 M3U8 视频文件

文章目录


参考：
https://www.zhihu.com/question/614024449/answer/3569588562
https://docs.pingcode.com/ask/1139281.html
https://docs.pingcode.com/ask/1140886.html
https://xie.infoq.cn/article/9024aa2ec17c3b5932b3c192b

爬虫工具：
https://zhuanlan.zhihu.com/p/653351722

m3u8文件链接：
https://www.jspoo.com/wz/4389.html
http://tonkiang.us/?iptv=%E7%94%B5%E5%BD%B1&l=49e96f36c2
https://v.cctv.com/?spm=C28340.PALq7Iyxwdh3.E2XVQsMhlk44.5

1. m3u8 文件概述

m3u8 是一种基于文本的媒体播放列表文件格式，通常用于指定流媒体播放器播放在线媒体流。它是一个简单的文本文件，其中包含多个由 URI 引用的媒体资源文件的 URL。m3u8 文件通常包含多个 ts 文件的链接，这些 ts 文件是实际的视频和音频数据文件，通常是通过 HTTP 协议传输。

ts 文件是一种流媒体传输格式，是 MPEG-2 传输流（MPEG-2 Transport Stream）的缩写。ts 文件通常用于存储视频、音频和字幕等媒体数据，是流媒体传输的基本单位。在 m3u8 文件中，ts 文件通常是通过 URI 引用的方式来指定的，播放器会根据 m3u8 文件中的 ts 文件链接，依次请求并下载 ts 文件，然后将其组合成完整的视频流进行播放。

因此，m3u8 文件和 ts 文件在流媒体播放领域密切相关，m3u8 文件是流媒体的播放列表，而 ts 文件是实际的媒体数据文件。m3u8 文件中包含了多个 ts 文件的链接，播放器会根据 m3u8 文件中的 ts 文件链接，依次请求并下载 ts 文件，然后将其组合成完整的视频流进行播放。这种方式可以充分利用网络带宽，提高流媒体的播放效率和质量。同时，m3u8 文件还可以通过定义不同的码率和分辨率等参数，实现适应不同网络环境和设备的自适应流媒体播放。

m3u8 文件格式参考：m3u8 文件格式详解

2. 库文件介绍

2.1. 使用逻辑

要使用 Python 库爬取 m3u8 文件，可以使用 requests、m3u8、ffmpeg 等库，分别负责网络请求、解析 m3u8 文件、下载和合并视频片段。
首先，通过 requests 库获取 m3u8 文件内容；
其次，使用 m3u8 库解析出所有的 .ts 视频片段链接；
最后，利用 ffmpeg 库将这些视频片段合并成一个完整的视频文件。

2.2. 库文件安装方法

安装 python 库

pip install requests m3u8

安装 ffmpeg

# For Ubuntu/Debian-based systems
sudo apt-get install ffmpeg

# For macOS using Homebrew
brew install ffmpeg

3. 爬取视频文件

3.1. 获取 m3u8 文件内容

首先，我们需要获取 m3u8 文件的内容。可以通过 requests 库发送 HTTP 请求来获取 m3u8 文件。

import requests

def get_m3u8_content(url):
    response = requests.get(url)
    response.raise_for_status()  # 检查请求是否成功
    return response.text

m3u8_url = 'http://example.com/path/to/your.m3u8'
m3u8_content = get_m3u8_content(m3u8_url)
print(m3u8_content)

3.2. 解析 m3u8 文件

使用 m3u8 库来解析 m3u8 文件，提取出所有 .ts 视频片段的 URL。

import m3u8

def parse_m3u8(content):
	# 解析 m3u8 文件
	# 将 m3u8 文件的内容解析为一个 m3u8 对象
    m3u8_obj = m3u8.loads(content)
    # 获取 .ts 文件 uri 列表
    ts_urls = [segment.uri for segment in m3u8_obj.segments]
    return ts_urls

ts_urls = parse_m3u8(m3u8_content)
print(ts_urls)

3.3. 下载视频片段

我们需要下载所有的.ts视频片段，并将它们保存到本地磁盘。

import os
def download_ts_segments(ts_urls, save_dir='videos'):
    # 判断文件夹是否存在，无则创建
    if not os.path.exists(save_dir):
        os.makedirs(save_dir)

    # 循环下载 .ts 文件，并保存
    for i, url in enumerate(ts_urls):
        response = requests.get(url)
        response.raise_for_status()
        
        # 保存文件
        ts_path = os.path.join(save_dir, f'segment_{i}.ts')
        with open(ts_path, 'wb') as f:
            f.write(response.content)
        print(f'Downloaded {ts_path}')

download_ts_segments(ts_urls)

3.4. 合并视频片段

下载所有视频片段后，使用 ffmpeg 工具将它们合并成一个完整的视频文件。

import subprocess
def merge_ts_segments(save_dir='videos', output_file='output.mp4'):
    # 创建 ffmpeg 参数
    ts_files = [os.path.join(save_dir, f'segment_{i}.ts') for i in range(len(ts_urls))]
    ts_file_list = os.path.join(save_dir, 'file_list.txt')
    with open(ts_file_list, 'w') as f:
        # 将 .ts 文件绝对路径存放在 file_list.txt 文件中。当作 ffmpeg 的输入参数。
        for ts_file in ts_files:
            f.write(f"file '{ts_file}'\n")

    subprocess.run(['ffmpeg', '-f', 'concat', '-SAFe', '0', '-i', ts_file_list, '-c', 'copy', output_file])

merge_ts_segments()

3.5. 完整实例

import requests

def get_m3u8_content(url):
    response = requests.get(url)
    response.raise_for_status()  # 检查请求是否成功
    return response.text

m3u8_url = 'http://example.com/path/to/your.m3u8'
m3u8_content = get_m3u8_content(m3u8_url)
print(m3u8_content)


import m3u8

def parse_m3u8(content):
	# 解析 m3u8 文件
    m3u8_obj = m3u8.loads(content)
    # 获取 .ts 文件 uri 列表
    ts_urls = [segment.uri for segment in m3u8_obj.segments]
    return ts_urls

ts_urls = parse_m3u8(m3u8_content)
print(ts_urls)


import os

def download_ts_segments(ts_urls, save_dir='videos'):
    # 判断文件夹是否存在，无则创建
    if not os.path.exists(save_dir):
        os.makedirs(save_dir)

    try:
        # 循环下载 .ts 文件，并保存
        for i, url in enumerate(ts_urls):
            response = requests.get(url)
            response.raise_for_status()

            # 保存文件
            ts_path = os.path.join(save_dir, f'segment_{i}.ts')
            with open(ts_path, 'wb') as f:
                f.write(response.content)
            print(f'Downloaded {ts_path}')

    except requests.RequestException as e:
        print(f'Error downloading {url}: {e}')

download_ts_segments(ts_urls)


import subprocess

def merge_ts_segments(save_dir='videos', output_file='output.mp4'):
    # 创建 ffmpeg 参数
    ts_files = [os.path.join(save_dir, f'segment_{i}.ts') for i in range(len(ts_urls))]
    ts_file_list = os.path.join(save_dir, 'file_list.txt')
    with open(ts_file_list, 'w') as f:
        # 将 .ts 文件绝对路径存放在 file_list.txt 文件中。当作 ffmpeg 的输入参数。
        for ts_file in ts_files:
            f.write(f"file '{ts_file}'\n")

    subprocess.run(['ffmpeg', '-f', 'concat', '-SAFe', '0', '-i', ts_file_list, '-c', 'copy', output_file])

merge_ts_segments()

4. 并发下载视频文件

4.1. 多线程（线程池）

import os
import requests
from concurrent.futures import ThreadPoolExecutor

def download_ts_segment(url, save_path):
    try:
        response = requests.get(url)
        response.raise_for_status()

        with open(save_path, 'wb') as f:
            f.write(response.content)
        print(f'Downloaded {save_path}')
    except requests.RequestException as e:
        print(f'Error downloading {url}: {e}')

def download_ts_segments_concurrently(ts_urls, save_dir='videos'):
    if not os.path.exists(save_dir):
        os.makedirs(save_dir)

    with ThreadPoolExecutor(max_workers=5) as executor:
        for i, url in enumerate(ts_urls):
            ts_path = os.path.join(save_dir, f'segment_{i}.ts')
            executor.submit(download_ts_segment, url, ts_path)

download_ts_segments_concurrently(ts_urls)

完成代码

from urllib.parse import urlparse, urlunparse, urljoin, urlsplit
import requests
import m3u8
import os
from concurrent.futures import ThreadPoolExecutor

#########################################################################
################ 1. 下载 m3u8 
#########################################################################
def get_m3u8_content(url):
    response = requests.get(url)
    response.raise_for_status()  # 检查请求是否成功
    return response.text

#########################################################################
################ 2. 解析 m3u8 
#########################################################################
def parse_master_m3u8(m3u8_content):
	# 解析 m3u8 文件
    # 将 m3u8 文件的内容解析为一个 m3u8 对象
    m3u8_obj = m3u8.loads(m3u8_content)

    # # print(m3u8_obj.playlists)
    # for variant in m3u8_obj.playlists:
    #     quality = variant.stream_info
    #     # print(quality)
    #     print(variant.uri)

    ts_urls = [variant.uri for variant in m3u8_obj.playlists]
    return ts_urls

def parse_media_m3u8(m3u8_content):
	# 解析 m3u8 文件
    # 将 m3u8 文件的内容解析为一个 m3u8 对象
    m3u8_obj = m3u8.loads(m3u8_content)

    # for segment in m3u8_obj.segments:
    #     # print(segment.uri)
    #     print(segment.uri)
    #     # print(segment.dumps())
    
    # 获取 .ts 文件 uri 列表
    ts_urls = [segment.uri for segment in m3u8_obj.segments]
    return ts_urls

#########################################################################
################ 3. 下载 切片文件 
#########################################################################
def absolute_url(base_url: str, ts_url: str):
    # print(base_url)
    if ts_url.startswith("http") or ts_url.startswith("https"):
        print(ts_url)
        return ts_url

    # 方法一
    parsed_url = urlparse(base_url)
    # print(parsed_url.scheme)    # 输出协议：https
    # print(parsed_url.netloc)    # 输出域名：www.example.com
    # print(parsed_url.path)      # 输出路径：/path
    # print(parsed_url.params)    # 
    # print(parsed_url.query)     # 输出查询参数：param1=value1&param2=value2
    # print(parsed_url.fragment)  # 片段标识符: 

    # 会自动去除最后一个元素
    absolute_path = urljoin(parsed_url.path, ts_url)
    # print(absolute_path)

    # 使用 urlunparse 组装 URL
    absolute_url = urlunparse((parsed_url.scheme, parsed_url.netloc, absolute_path, parsed_url.params, parsed_url.query, parsed_url.fragment))

    # # 方法二, 但是不适合有参数的 url
    # # # segment_url = base_url.rsplit('/', 1)[0] + '/' + "0.ts"
    # absolute_url = base_url.rsplit('/', 1)[0] + '/' + file_name

    print(absolute_url)
    return absolute_url

def download_ts_segment(ts_url, ts_file):
    try:
        response = requests.get(ts_url)
        response.raise_for_status()

        with open(ts_file, 'wb') as f:
            f.write(response.content)
        print(f'Downloaded {ts_file}')
    except requests.RequestException as e:
        print(f'Error downloading {ts_url}: {e}')

def download_ts_segments_concurrently(base_url, ts_urls, save_dir='videos'):
    # 判断保存文件夹 路径是否存在。无则创建
    if not os.path.exists(save_dir):
        os.makedirs(save_dir)

    # 启动线程池，现在视频文件
    with ThreadPoolExecutor(max_workers=5) as executor:
        for idx, url in enumerate(ts_urls):
            ts_url = absolute_url(base_url, url)
            ts_file = os.path.join(save_dir, f'segment_{idx}.ts')
            executor.submit(download_ts_segment, ts_url, ts_file)

if __name__ == "__main__":
    m3u8_url = "https://dh5qqwx01.v.cntv.cn/asp/h5e/hls/2000/0303000a/3/default/c8031e40ba9d4e03be14e20786ade622/2000.m3u8"
    # ts_url = "https://dh5qqwx01.v.cntv.cn/asp/h5e/hls/1200/0303000a/3/default/c8031e40ba9d4e03be14e20786ade622/0.ts"
    m3u8_content = ""
    # absolute_url(m3u8_url, "http://000.ts")

    # # 1. 下载 m3u8 文件
    # m3u8_content = get_m3u8_content(m3u8_url)
    # print(m3u8_content)

    # # 2. 解析 m3u8 文件
    # # with open("video_master.m3u8", "r") as file:
    # #     m3u8_content = file.read()
    # # ts_urls = parse_master_m3u8(m3u8_content)
    # with open("video_media.m3u8", "r") as file:
    #     m3u8_content = file.read()
    ts_urls = parse_media_m3u8(m3u8_content)
    print(ts_urls)

    # 3. 下载切片文件
    download_ts_segments_concurrently(m3u8_url, ts_urls)

4.2. 多线程

https://www.zhihu.com/question/614024449/answer/3569588562

4.3. 协程 aiohttp

https://www.zhihu.com/question/614024449/answer/3569588562

5. .ts 文件播放

在获取到 .ts 文件链接后，可以使用 requests 库下载这些文件，并使用视频播放库（如 OpenCV 或 VLC ）进行播放。
https://docs.pingcode.com/ask/1140886.html

网站公告

今日签到

点亮在社区的每一天
去签到

热门文章

最新发布