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
宣告：歡迎任何形式的轉載，但請務必注明出處！！！
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/553820.html
標籤：其他
上一篇：一文詳解 Sa-Token 中的 SaSession 物件
下一篇：返回列表