Python 實現 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 文件還可以通過定義不同的碼率和解析度等參數，實現適應不同網路環境和設備的自適應流媒體播放。

基礎實現

以下是使用 Python 下載 m3u8 視頻並保存為 mp4 的示例代碼：

import requests
import os

def download_m3u8_video(url, file_path):
    r = requests.get(url)
    if r.status_code != 200:
        print('m3u8視頻下載鏈接無效')
        return False

    m3u8_list = r.text.split('\n')
    m3u8_list = [i for i in m3u8_list if i and i[0] != '#']

    ts_list = []
    for ts_url in m3u8_list:
        ts_url = url.rsplit('/', 1)[0] + '/' + ts_url
        ts_list.append(ts_url)

    with open(file_path, 'wb') as f:
        for ts_url in ts_list:
            r = requests.get(ts_url)
            if r.status_code == 200:
                f.write(r.content)
    print('m3u8視頻下載完成')
    return True

def convert_ts_to_mp4(ts_file_path, mp4_file_path):
    os.system(f'ffmpeg -i {ts_file_path} -c copy {mp4_file_path}')

if __name__ == '__main__':
    url = '輸入m3u8流媒體播放列表文件下載鏈接'
    ts_file_path = '輸入ts文件保存路徑'
    mp4_file_path = '輸入mp4文件保存路徑'

    download_m3u8_video(url, ts_file_path)
    convert_ts_to_mp4(ts_file_path, mp4_file_path)

在這個示例中，download_m3u8_video 函數用於下載 m3u8 視頻，convert_ts_to_mp4 函數用於將下載的 ts 文件轉換為 mp4 文件。首先，使用 requests 庫下載 m3u8 文件，並解析出其中的 ts 文件鏈接。然後，遍歷 ts 文件鏈接列表，使用 requests 庫下載每個 ts 文件，並將其寫入到一個文件中。最後，使用 ffmpeg 工具將下載的 ts 文件轉換為 mp4 文件。需要註意的是，為了使用 ffmpeg 工具，需要在系統中安裝 ffmpeg，並將其添加到環境變數中。

實際應用中，可以根據具體情況對代碼進行調整和優化，例如增加異常處理、優化下載速度等。同時，由於 m3u8 視頻格式的特殊性，下載過程可能會較為耗時，需要耐心等待一段時間。

使用多線程來優化下載速度

為了優化下載速度，可以使用多線程或非同步 IO 的方式來下載 m3u8 視頻。以下是使用多線程下載 m3u8 視頻的示例代碼：

import requests
import os
import threading

class Downloader(threading.Thread):
    def __init__(self, url, ts_url, file_path):
        threading.Thread.__init__(self)
        self.url = url
        self.ts_url = ts_url
        self.file_path = file_path

    def run(self):
        r = requests.get(self.ts_url, stream=True)
        if r.status_code == 200:
            with open(self.file_path, 'wb') as f:
                for chunk in r.iter_content(chunk_size=1024):
                    if chunk:
                        f.write(chunk)

def download_m3u8_video(url, file_path):
    r = requests.get(url)
    if r.status_code != 200:
        print('m3u8視頻下載鏈接無效')
        return False

    m3u8_list = r.text.split('\n')
    m3u8_list = [i for i in m3u8_list if i and i[0] != '#']

    ts_list = []
    for ts_url in m3u8_list:
        ts_url = url.rsplit('/', 1)[0] + '/' + ts_url
        ts_list.append(ts_url)

    threads = []
    for i, ts_url in enumerate(ts_list):
        ts_file_path = file_path.rsplit('.', 1)[0] + f'_{i}.ts'
        thread = Downloader(url, ts_url, ts_file_path)
        thread.start()
        threads.append(thread)

    for thread in threads:
        thread.join()

    print('m3u8視頻下載完成')
    return True

def convert_ts_to_mp4(ts_file_path, mp4_file_path):
    os.system(f'ffmpeg -i {ts_file_path} -c copy {mp4_file_path}')

if __name__ == '__main__':
    url = '輸入m3u8流媒體播放列表文件下載鏈接'
    ts_file_path = '輸入ts文件保存路徑'
    mp4_file_path = '輸入mp4文件保存路徑'

    download_m3u8_video(url, ts_file_path)
    convert_ts_to_mp4(ts_file_path, mp4_file_path)

在這個示例中，定義了一個 Downloader 類，用於下載每個 ts 文件。在 Downloader 類中，使用 requests 庫的 stream 參數將下載進度分塊，每次下載 1024 個位元組，然後寫入到文件中。在 download_m3u8_video 函數中，使用多線程的方式同時下載多個 ts 文件，並等待所有線程下載完成後再將其合併成一個 mp4 文件。這樣可以大大縮短下載時間。

需要註意的是，多線程下載可能會導致網路瓶頸，從而降低下載速度。因此，在實際應用中，需要根據具體情況選擇合適的下載方式，併進行調整和優化。例如，可以使用非同步 IO、協程等技術來優化下載速度。另外，為了提高下載速度，還可以使用 CDN、負載均衡、網路加速等技術來優化下載環節。

使用非同步 IO 和協程來優化下載速度

協程（Coroutine）是一種輕量級的線程，可以在單線程中實現多個任務的併發執行，從而提高程式的效率和性能。Python 中的協程是通過 async/await 關鍵字來實現的，可以使用 asyncio 庫來進行協程編程。

非同步 IO（Asynchronous IO）是一種非阻塞式 IO 模型，可以在進行 IO 操作時不會阻塞程式的執行，從而提高程式的效率和響應速度。Python 中的非同步 IO 是通過 asyncio 庫來實現的，可以使用 async/await 關鍵字和協程來實現非同步 IO 操作。

非同步 IO 和協程的結合可以實現高效的併發編程，通過非同步 IO 可以充分利用 CPU 和網路帶寬等資源，提高程式的效率和性能；而通過協程可以在單線程中實現多個任務的併發執行，避免了線程切換的開銷，從而提高程式的響應速度和併發性能。在實際應用中，我們可以根據具體情況選擇和優化非同步 IO 和協程的使用方式，以達到最佳的效果和性能。

為了使用非同步 IO 和協程來優化下載速度，可以使用 aiohttp 和 asyncio 庫來實現。以下是使用非同步 IO 和協程下載 m3u8 視頻的示例代碼：

import aiohttp
import asyncio
import os

async def download_ts_file(ts_url, ts_file_path):
    # 防止ssl報錯：
    # aiohttp.client_exceptions.ClientConnectorCertificateError: Cannot connect to host ***.****.com:443 ssl:True
    # [SSLCertVerificationError: (1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local
    # issuer certificate (_ssl.c:1123)')]
    conn = aiohttp.TCPConnector(ssl=False)
    async with aiohttp.ClientSession(connector=conn) as session:
        async with session.get(ts_url) as response:
            if response.status != 200:
                print(f'{ts_url} 下載失敗')
                return False
            with open(ts_file_path, 'wb') as f:
                while True:
                    chunk = await response.content.read(1024)
                    if not chunk:
                        break
                    f.write(chunk)
    print(f'{ts_url} 下載完成')
    return True

async def download_m3u8_video(url, file_path):
    # 防止ssl報錯
    conn = aiohttp.TCPConnector(ssl=False)
    async with aiohttp.ClientSession(connector=conn) as session:
        async with session.get(url) as response:
            if response.status != 200:
                print('m3u8視頻下載鏈接無效')
                return False

            m3u8_text = await response.text()
            m3u8_list = m3u8_text.split('\n')
            m3u8_list = [i for i in m3u8_list if i and i[0] != '#']

            tasks = []
            for i, ts_url in enumerate(m3u8_list):
                ts_url = url.rsplit('/', 1)[0] + '/' + ts_url
                ts_file_path = file_path.rsplit('.', 1)[0] + f'_{i}.ts'
                task = asyncio.ensure_future(
                    download_ts_file(ts_url, ts_file_path))
                tasks.append(task)

            await asyncio.gather(*tasks)

    print('m3u8視頻下載完成')
    return True

def convert_ts_to_mp4(ts_file_path, mp4_file_path):
    os.system(f'ffmpeg -i {ts_file_path} -c copy {mp4_file_path}')

if __name__ == '__main__':
    url = '輸入m3u8流媒體播放列表文件下載鏈接'
    ts_file_path = '輸入ts文件保存路徑'
    mp4_file_path = '輸入mp4文件保存路徑'

    loop = asyncio.get_event_loop()
    loop.run_until_complete(download_m3u8_video(url, ts_file_path))
    convert_ts_to_mp4(ts_file_path, mp4_file_path)

在這個示例中，使用了 aiohttp 和 asyncio 庫來實現非同步 IO 和協程。定義了兩個協程函數：download_m3u8_video 和 download_ts_file。在 download_m3u8_video 函數中，使用 aiohttp 庫的 ClientSession 類非同步獲取 m3u8 文件，並解析出其中的 ts 文件鏈接。然後，使用協程和非同步 IO 的方式非同步下載每個 ts 文件，並將其寫入到本地文件中。在下載過程中，使用了非同步 IO 和協程的方式，可以充分利用網路帶寬，提高下載速度。

在 download_m3u8_video 函數中，使用了 async for 迴圈來遍歷 m3u8 文件中的 ts 文件鏈接，並創建了一個任務列表 tasks，用於存儲非同步下載的任務。然後，使用 asyncio.ensure_future 方法將每個任務添加到任務列表中。最後，使用 asyncio.gather 方法同時運行所有非同步任務，等待所有任務完成後，即可完成整個 m3u8 視頻的下載。

最後，使用 ffmpeg 工具將下載的 ts 文件轉換為 mp4 格式的視頻文件。這個步驟並不涉及非同步 IO 和協程，只是為了將下載的 ts 文件轉換為可用的視頻文件格式。

使用協程可以充分利用網路帶寬，提高下載速度。需要註意的是，在使用協程時，需要考慮到 CPU 和記憶體等資源的占用，避免出現資源耗盡或者死鎖等問題。同時，協程的使用需要掌握一定的非同步編程技巧，例如使用 async/await 關鍵字、協程調度等。因此，在實際應用中，需要根據具體情況進行調整和優化，以獲取最佳的性能和效果。

作者：飛仔FeiZai

出處：https://www.cnblogs.com/yuzhihui/p/17443299.html

聲明：歡迎任何形式的轉載，但請務必註明出處！！！