关于python:python下载M3U8

ps: 本文前提是取得了 m3u8_url，同时该 url 没有进行任何加密。

单线程下载

import requests
import os
# from subprocess import run


# 创立相应的文件夹
if not os.path.exists('download'):
    os.mkdir('download')

if not os.path.exists('download/m3u8'):
    os.mkdir('download/m3u8')

if not os.path.exists('download/temp'):
    os.mkdir('download/temp')


def download(medianame, m3u8_url):
    headers = {
        'user-agent':
        'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.116 Safari/537.36'
    }
    url = '/'.join(m3u8_url.split('/')[:-1]) + '/'
    with open('download/m3u8/' + medianame + '.m3u8', 'wb') as f:
        r = requests.get(m3u8_url, headers=headers)
        f.write(r.content)

    line = []
    for i in open('download/m3u8/' + medianame + '.m3u8', 'r'):
        if '#' in i:
            continue
        line.append(i.replace('\r', '').replace('\n',''))

    print('开始下载!')
    with open('download/' + medianame + '.ts', 'wb') as f:
        for i in tqdm(line):
            # print(url+i)
            r = requests.get(url + i)
            f.write(r.content)
    # print('开始转码, 请期待~')
    # run([
    #     'ffmpeg/bin/ffmpeg.exe', '-i', 'download/' + medianame + '.ts', '-c:v',
    #     'copy', '-c:a', 'copy', 'download/' + medianame + '.mp4'
    # ])
    # os.remove('download/' + medianame + '.ts')
    os.remove('download/m3u8/' + medianame + '.m3u8')

多线程下载

import aiohttp
import asyncio
import os
# from subprocess import run


def combine(temp, file):
    with open(file, 'wb') as f:
        list = os.listdir(temp)
        list.sort()
        for i in list:
            with open(os.path.join(temp, i), 'rb') as t:
                f.write(t.read())


async def req(client, url, filename, sem):
    async with sem:
        response = await client.get(url)
        with open(filename, 'wb') as f:
            f.write(await response.content.read())


async def download_async(medianame, m3u8_url, sem_num):
    if int(sem_num) in range(2, 13):
        sem = asyncio.Semaphore(int(sem_num))
        print('以 %s 进行下载' % sem_num)
    else:
        print('输出不合格, 以 6 进行下载')
        sem = asyncio.Semaphore(6)
    async with aiohttp.ClientSession(connector=aiohttp.TCPConnector(ssl=False)) as session:
        headers = {
            'user-agent':
            'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.116 Safari/537.36'
        }
        url = '/'.join(m3u8_url.split('/')[:-1]) + '/'
        with open('download/m3u8/' + medianame + '.m3u8', 'wb') as f:
            r = requests.get(m3u8_url, headers=headers)
            f.write(r.content)
        urls = []
        for i in open('download/m3u8/' + medianame + '.m3u8', 'r'):
            if '#' in i:
                continue
            urls.append(i.replace('\r', '').replace('\n',''))
        if not os.path.exists('download/temp/' + medianame):
            os.mkdir('download/temp/' + medianame)
        task_list = []
        for ts_name in urls:
            filename = 'download/temp/' + medianame + '/' + ts_name
            task = asyncio.create_task(req(session, url + ts_name, filename, sem))
            task_list.append(task)
        responses = [await f for f in tqdm(asyncio.as_completed(task_list),
                                  total=len(task_list))
        ]
        combine('download/temp/' + medianame, 'download/' + medianame + '.ts')
        # run([
        #     'ffmpeg/bin/ffmpeg.exe', '-i', 'download/' + medianame + '.ts', '-c:v',
        #     'copy', '-c:a', 'copy', 'download/' + medianame + '.mp4'
        # ])
        # shutil.rmtree('download/temp/' + medianame)
        # os.remove('download/' + medianame + '.ts')
        os.remove('download/m3u8/' + medianame + '.m3u8')
        

loop = asyncio.get_event_loop()
loop.run_until_complete(download_async(mediaName, m3u8_url, sem_num))

转码成 mp4

勾销掉正文的代码，并批改 ffmpeg 门路至你的 ffmpeg 门路，即可主动转化成 mp4。