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

目標網址我就不說了，怕被和諧

以《叢林奇航DB》為例吧，首先找到目標網址，再F12檢查，點擊網絡，找到m3u8結尾的鏈接（找不到的話點擊刷新一下）。如下圖：

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

如果鏈接不完整的話要手動補全鏈接。

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

首先導入庫：

#!/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的鏈接地址進行訪問與獲取文本 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進行解密crypto = AES.new(key,AES.MODE_CBC, key)# 解密完成之后就可以寫入了file.write(crypto.decrypt(ts))file.close()

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

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

下面來說說aiohttp庫和asyncio庫進行協程的用法

# 第一步：得到一個url列表 urls = ['https://baidu.com','https://baidu.com','https://baidu.com'] # 第一步：用asyncio創建一個函數線程，用aiohttp來創建一個支持異步的訪問 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]# 第三步：定義一個執行模塊 if __name__ == '__main__':start = time.time()# 把第一步的列表放入函數塊里并導入到tasks列表tasks = []for url in urls:c = get_status(url)# 把函數塊放入線程task = asyncio.ensure_future(c)tasks.append(task)# 循環事件loop = asyncio.get_event_loop()loop.run_until_complete(asyncio.wait(tasks))# 循環完成后進行關閉loop.close()print('總耗時：', time.time()-start)

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

我們可以看到，調用線程已經成功，接下來就是怎么獲取和寫入了：下面對寫入進行分析：

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

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

?

到這一步，我們已經知道怎么調用協程和解密包括寫入了

假設我們這一步已經提取了所有的ts視頻文件

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

合并有調用cmd和利用ffmpeg（需要安裝）：

調用cmd和利用ffmpge進行合成：

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

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

接下來附上完整的代碼

#!/usr/bin/env python3 # -*- coding: utf-8 -*-# 導入所需的庫 import time import os import requests import re import aiohttp import asyncio from Crypto.Cipher import AES# 創建一個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] # 給文件命名，用列表進行封裝 tt = '{}' names = [tt.format(num) for num in range(10001, 10301)] # print(rst) print(names) # 把ts視頻封裝進列表里 ts_list = [] for i in rst:ts_url = 'https:'+i+'.ts'ts_list.append(ts_url)# 限制最大協程數用Semaphore，時間原來這里不講了，不懂再問 concurrency = 15 semaphore = asyncio.Semaphore(concurrency)# 用asyncio和aiphttp結合創建線程 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:# 因為是byte類型，所以用read()page_text = await response.read()return page_text, namedef parse(t):page_text = t.result()# 返回得到一個元組，轉換為列表，再對文件名和內容進行提取complex = list(page_text)na = complex[1]all = complex[0]# print(na)# print(all)# 用crypto里的AES進行解密并保存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()# 定義列表，把元素放進去并開啟協程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合并，這里創建了一個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)# 調用cmd并利用ffmpeg進行合并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'# 合并完成后再調用cmd進行ts文件刪除os.system(cmd)delete = f'del D:\\thead\\*'os.system(delete)# 整個程序運行所需的時間print('總用時：', time.time()-start)

時間原因，還有些限制協程數就不多說了，有問題可以互相交流學習。

最后看一下效果：

好，大功告成！！！！?

總結

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

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。