Python 实现 m3u8 视频下载

Posted 2023-05-30 飞仔FeiZai

Python 实现 m3u8 视频下载

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

声明：欢迎任何形式的转载，但请务必注明出处！！！

Python实现--简易视频爬虫

　　最近这几天，学习了一下python，对于爬虫比较感兴趣，就做了一个简单的爬虫项目，使用Python的库Tkinsert做了一个界面，感觉这个库使用起来还是挺方便的，这个程序中使用到了正则表达式（re模块），对爬回的网页进行匹配分析，最后通过urllib模块中urlretrieve()这个方法进行视频的下载，由于存在窗口，防止窗口在下载时候处于卡死状态，使用了threading模块，创建了一个线程去执行下载任务。以下是该程序的源码。

  1 #!/usr/bin/env python
  2 # -*- coding: utf-8 -*-
  3 #coding:utf-8
  4 # @Date    : 2016-12-28 11:18:40
  5 # @Author  : Donoy ([email protected])
  6 # @Link    : http://www.cnblogs.com/Donoy/
  7 # @Version : $Id$
  8 
  9 #多线程 爬虫 
 10 #GUI 界面
 11 
 12 import os
 13 import requests
 14 import re
 15 import threading
 16 from   Tkinter import *
 17 from   ScrolledText import ScrolledText #文本滚动框
 18 import urllib #这个模块中有下载的功能
 19 import sys
 20 #sys 模块的输出编码 格式
 21 reload(sys)
 22 sys.setdefaultencoding(‘utf-8‘)
 23 Video_Data = []
 24 
 25  
 26 def creatWnd():
 27     global root 
 28     global varl 
 29     global text
 30 
 31     #创建一个窗口
 32     root = Tk() # 窗口
 33 
 34     #窗口的标题
 35     root.title(‘DSpider‘)
 36     #文本滚动窗口
 37     text = ScrolledText(root,font=(‘微软雅黑‘))
 38     text.grid()
 39 
 40     #设置Lable
 41     varl = StringVar()
 42     lable = Label(root,font=(‘微软雅黑‘),fg=‘red‘,textvariable=varl)
 43     lable.grid()
 44     varl.set(‘Fight......‘)
 45 
 46     #设置按钮
 47     button = Button(root,text=‘开始爬取‘,font=‘黑体‘,command=begin_Thread)
 48     button.grid()
 49 
 50     #root.mainloop()     # 显示窗口
 51 
 52 def getHtmlData(url):
 53     #print html.text               # 网站的内容
 54     #print html.status_code        # 请求的返回值
 55     RequestHeader = { 
 56            ‘User-Agent‘:‘Mozilla/5.0 (Windows NT 6.3; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/55.0.2883.75 Safari/537.36‘
 57     }
 58     html  =  requests.get(url,headers=RequestHeader)
 59     #print html.text
 60     return html.text
 61 
 62 
 63 def getVideoUrl(html):
 64     #re.S 是匹配换行符 
 65     Parse   = re.compile(r‘(<div class="j-r-list-c">.*?</div>.*?</div>)‘,re.S)
 66     context = re.findall(Parse,html)
 67     Parse   = re.compile(r‘data-mp4="(.*?)"‘)
 68     for item in context:
 69         VideoUrl  = re.findall(Parse,item)
 70         if VideoUrl:
 71             Name_Parse = re.compile(r‘(<a href="/detail-.{8}.html">(.*?)</a>)‘,re.S);
 72             VideoName  = re.findall(Name_Parse,item) 
 73             for Name,url in zip(VideoName,VideoUrl): #zip这个内置函数就是将List重新整合一下
 74                 Video_Data.append([Name,url])
 75                 
 76 
 77 def begin_Thread():
 78     try:
 79         for i in range(10,12):
 80             url = ‘http://www.budejie.com/%d‘%i
 81             html = getHtmlData(url)
 82             getVideoUrl(html)    
 83     except Exception as e:
 84         raise
 85     varl.set(‘一共%s个小视频，现在开始下载......‘%(len(Video_Data)/2))
 86     thread = threading.Thread(target= myThread)
 87     thread.start()
 88 
 89 
 90 def myThread():
 91     id = 1
 92     for Data in Video_Data:
 93         text.insert(END,str(id)+‘.‘+Data[0][1]+Data[1]+‘\\n‘)
 94         urllib.urlretrieve(Data[1],‘%s.mp4‘%Data[0][1].decode(‘utf-8‘).encode(‘gbk‘))
 95         Video_Data.pop(0)
 96         id += 1
 97     varl.set(‘所有的视频都下载完成‘)    
 98         
 99 
100 
101 def main():
102     creatWnd()
103     root.mainloop()
104 
105 
106 if __name__ == ‘__main__‘:
107     main()
108     
109 
110