ps: 本文前提是取得了m3u8_url，同时该url没有进行任何加密。
单线程下载
import requestsimport 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 aiohttpimport asyncioimport os# from subprocess import rundef 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。