花了幾天時間搞m3u8里的ts視頻，還有多線程協(xié)程的處理問題。看了大量的回答，發(fā)現(xiàn)大多數(shù)人講的都是個大概，具體的怎么用、什么原理沒講。今天就來帶大家講解怎么爬取m3u8里的加密ts視頻。

目標(biāo)網(wǎng)址我就不說了，怕被和諧

以《叢林奇航DB》為例吧，首先找到目標(biāo)網(wǎng)址，再F12檢查，點(diǎn)擊網(wǎng)絡(luò)，找到m3u8結(jié)尾的鏈接（找不到的話點(diǎn)擊刷新一下）。如下圖：

點(diǎn)擊預(yù)覽后會看到一連串的ts鏈接和以AES-128加密的key鏈接。如下圖：

如果鏈接不完整的話要手動補(bǔ)全鏈接。

看到這里的話已經(jīng)所找到了視頻資源，接下來寫代碼爬取了。

首先導(dǎo)入庫：

#!/usr/bin/env python3 # -*- coding: utf-8 -*-import time import os import requests import re import aiohttp import asyncio from Crypto.Cipher import AES

?crypto庫的安裝請前往crypto安裝 - 百度文庫自行了解。

再來說說AES的一個解密ts視頻原理，這里以爬取一部ts為例：

from Crypto.Cipher import AES# 提取key和ts的鏈接地址進(jìn)行訪問與獲取文本 key_url = 'https://pps.shanshanku.com/20211127/g8V4A0hE/1000kb/hls/key.key' ts_url = 'https://pps.shanshanku.com/20211127/g8V4A0hE/1000kb/hls/z8WfPVdF.ts' key = requests.get(key_url).content ts = requests.get(ts_url).contentwith open('./date/video.ts', 'wb') as file:# 用Crupto庫里的AES進(jìn)行解密crypto = AES.new(key,AES.MODE_CBC, key)# 解密完成之后就可以寫入了file.write(crypto.decrypt(ts))file.close()

得到視頻。解密成功。沒解密直接寫入是無法播放的。

?解密原理已經(jīng)了解，接下來就是如何爬取多個ts視頻了。

下面來說說aiohttp庫和asyncio庫進(jìn)行協(xié)程的用法

# 第一步：得到一個url列表 urls = ['https://baidu.com','https://baidu.com','https://baidu.com'] # 第一步：用asyncio創(chuàng)建一個函數(shù)線程，用aiohttp來創(chuàng)建一個支持異步的訪問 async def get_status(url):# 類似session = aiohttp.ClientSession,with 前面要加async,獲取東西前用awaitasync with aiohttp.ClientSession() as session:async with await session.get(url) as response:# 如果是<str>類型的就用text().<byte>類型的就用read()text = await response.text()return text[:100]# 第三步：定義一個執(zhí)行模塊 if __name__ == '__main__':start = time.time()# 把第一步的列表放入函數(shù)塊里并導(dǎo)入到tasks列表tasks = []for url in urls:c = get_status(url)# 把函數(shù)塊放入線程task = asyncio.ensure_future(c)tasks.append(task)# 循環(huán)事件loop = asyncio.get_event_loop()loop.run_until_complete(asyncio.wait(tasks))# 循環(huán)完成后進(jìn)行關(guān)閉loop.close()print('總耗時：', time.time()-start)

運(yùn)行后查看三個線程同時啟動到完成后的總用時：

我們可以看到，調(diào)用線程已經(jīng)成功，接下來就是怎么獲取和寫入了：下面對寫入進(jìn)行分析：

...... # 如果是<str>類型的就用text().<byte>類型的就用read()text = await response.text()return text[:100]# 這里只截取了一小部分 # 增加一個函數(shù)塊# 形參規(guī)定設(shè)為t，對線程運(yùn)行返回的結(jié)果進(jìn)行接收 def parsel(t):tlt = t.result()# 直接打印tlt返回的是一個元組類型,一個線程對應(yīng)一個元組，這里有三個線程print(tlt)# 如果元組里有多個數(shù)據(jù)時，需要轉(zhuǎn)換為列表才能提取lis = list(tlt)with open('./date/http.txt', 'a') as file:# 對獲取的text[:100]進(jìn)行寫入file.write(str(tlt))file.close()......c = get_status(url)# 把函數(shù)塊放入線程task = asyncio.ensure_future(c)# 把parsel函數(shù)添加進(jìn)去。。。。。。。。。。。。。。。。。。。。task.add_done_callback(parsel)tasks.append(task)

?我們來看看打印效果和寫入效果：

?

到這一步，我們已經(jīng)知道怎么調(diào)用協(xié)程和解密包括寫入了

假設(shè)我們這一步已經(jīng)提取了所有的ts視頻文件

接下來就是對ts視頻進(jìn)行合并與刪除源文件了

合并有調(diào)用cmd和利用ffmpeg（需要安裝）：

調(diào)用cmd和利用ffmpge進(jìn)行合成：

# 導(dǎo)入所需模塊 import os# 用ffmpeg進(jìn)行ts視頻合成 cmd = 'ffmpeg -f concat -safe 0 -i complex.txt(合成路徑與命名) -c copy output.mp4（輸出路徑與命名）'# 調(diào)用cmd os.system(cmd)

這樣我們就完成了ts視頻的合成。

接下來附上完整的代碼

#!/usr/bin/env python3 # -*- coding: utf-8 -*-# 導(dǎo)入所需的庫 import time import os import requests import re import aiohttp import asyncio from Crypto.Cipher import AES# 創(chuàng)建一個URL列表 m3u8_URL = 'https://pps.sd-play.com/20220424/RN3p7Bj5/1200kb/hls/index.m3u8' resp = requests.get(url=m3u8_URL).text# 這里只提取300個ts視頻 rst = re.findall('https:(.*?).ts', resp)[:300] # name = re.findall('1200kb/hls/(.*?).ts',resp)[:5] # 給文件命名，用列表進(jìn)行封裝 tt = '{}' names = [tt.format(num) for num in range(10001, 10301)] # print(rst) print(names) # 把ts視頻封裝進(jìn)列表里 ts_list = [] for i in rst:ts_url = 'https:'+i+'.ts'ts_list.append(ts_url)# 限制最大協(xié)程數(shù)用Semaphore，時間原來這里不講了，不懂再問 concurrency = 15 semaphore = asyncio.Semaphore(concurrency)# 用asyncio和aiphttp結(jié)合創(chuàng)建線程 async def get_request(url, name):async with semaphore:# conn = aiohttp.TCPConnector(limit=15)async with aiohttp.ClientSession() as session:async with await session.get(url) as response:# 因?yàn)槭莃yte類型，所以用read()page_text = await response.read()return page_text, namedef parse(t):page_text = t.result()# 返回得到一個元組，轉(zhuǎn)換為列表，再對文件名和內(nèi)容進(jìn)行提取complex = list(page_text)na = complex[1]all = complex[0]# print(na)# print(all)# 用crypto里的AES進(jìn)行解密并保存key = b'b7d463938dcfabff'path = 'D:\\thead\\'with open(f'{path}{na}.ts', 'ab') as file:cryptor = AES.new(key, AES.MODE_CBC, key)file.write(cryptor.decrypt(all))print('下載完成')if __name__ == '__main__':start = time.time()# 定義列表，把元素放進(jìn)去并開啟協(xié)程tasks = []for url,name in zip(ts_list,names):c = get_request(url, name)task = asyncio.ensure_future(c)task.add_done_callback(parse)tasks.append(task)loop = asyncio.get_event_loop()try:loop.run_until_complete(asyncio.wait(tasks))except:loop.close()# 為方便ffmpeg合并，這里創(chuàng)建了一個txt文件列表里面包含每個ts的名稱。for j in range(10001, 10301):with open('D:\\thead\\list.txt', 'a') as file:file.write(f'file {j}.ts' + '\n')file.close()time.sleep(1)# 調(diào)用cmd并利用ffmpeg進(jìn)行合并cmd = f'ffmpeg -f concat -safe 0 -i D:\\thead\\list.txt -c copy D:\\thead\\theads\\new01.mp4'# cmd = f'copy/b D:\\thead\\* D:\\thead\\theads\\new01.ts'# 合并完成后再調(diào)用cmd進(jìn)行ts文件刪除os.system(cmd)delete = f'del D:\\thead\\*'os.system(delete)# 整個程序運(yùn)行所需的時間print('總用時：', time.time()-start)

時間原因，還有些限制協(xié)程數(shù)就不多說了，有問題可以互相交流學(xué)習(xí)。

最后看一下效果：

好，大功告成！！！！?

總結(jié)

以上是生活随笔為你收集整理的一文看懂，python抓取m3u8里ts加密视频及合成、多线程、写入的问题的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。