一、完整代碼

完整代碼如下：

import requests
from multiprocessing import Pool
import re
import os
from tqdm import tqdm
from Crypto.Cipher import AES


# 創(chuàng)建臨時文件夾
dirs = 'ts_list_need_to_merge/'
os.makedirs(dirs, exist_ok=True)

headers = {
        'Accept': '*/*',
        'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
        'Connection': 'keep-alive',
        'Origin': 'http://www.kpd510.me',
        'Referer': 'http://www.kpd510.me/',
        'Sec-Fetch-Dest': 'empty',
        'Sec-Fetch-Mode': 'cors',
        'Sec-Fetch-Site': 'cross-site',
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/116.0.0.0 Safari/537.36 Edg/116.0.1938.69',
        'sec-ch-ua': '"Chromium";v="116", "Not)A;Brand";v="24", "Microsoft Edge";v="116"',
        'sec-ch-ua-mobile': '?0',
        'sec-ch-ua-platform': '"Windows"',
    }



def parse_m3u8_text(m3u8_text):
    m3u8_text = m3u8_text.split()
    encode_info = [line for line in m3u8_text if line.startswith('#EXT-X-KEY:')][0]
    pattern = r"#EXT-X-KEY:METHOD=(.*),URI=\"(.*)\""  
    
    ## 獲得加密method 和 key.key的url
    match = re.search(pattern, encode_info)
    if match:
        method = match.group(1)
        key_url = match.group(2)
    else:
        raise '解析失敗'
    
    ## 獲得ts文件url
    ts_list = [line for line in m3u8_text if line.endswith('ts')]
    return method, key_url, ts_list

def decrypt_content_and_save_file(filename, content, decrypter):
    with open(filename, mode='wb') as f:
        f.write(decrypter.decrypt(content))


def merge_ts_to_mp4(filename, ts_file_list):
    with open(filename, mode='ab') as f1:
        for ts_file in ts_file_list:
            with open(ts_file, mode='rb') as f2:
                f1.write(f2.read())
    print(filename, '完成！')


def process_one_url(ts_url, key):
    decrypter = AES.new(key, AES.MODE_CBC)
    filename = dirs + os.path.split(ts_url)[-1]
    content = requests.get(ts_url, headers=headers).content
    decrypt_content_and_save_file(filename, content, decrypter)
    return filename


def download_method_1(ts_list, key):
    # 普通次序一個一個下載，耗時11分鐘
    ts_file_list = []
    for ts_url in tqdm(ts_list):
        filename = process_one_url(ts_url=ts_url, key=key)
        ts_file_list.append(filename)
    return ts_file_list


def download_method_2(ts_list, key, processes_nums=2):
	# 多進(jìn)程下載， 耗時1分鐘
    class CallBack:
        def __init__(self, nums) -> None:
            self.pbar = tqdm(total=nums)
            self.filenames = []
        
        def callback(self, filename):
            self.pbar.update(1)
            self.filenames.append(filename)
    
    callback = CallBack(len(ts_list))
    pool = Pool(processes=processes_nums)
    for ts_url in ts_list:
        pool.apply_async(process_one_url, (ts_url, key), error_callback=print, callback=callback.callback)
    pool.close()
    pool.join()
    callback.pbar.close()
    return [dirs + os.path.split(ts_url)[-1] for ts_url in ts_list]


if __name__ == "__main__":
    m3u8_url = 'https://play.bo262626.com/20231108/xV1bY9Cn/700kb/hls/index.m3u8'
    response = requests.get(m3u8_url, headers=headers)
    m3u8 = response.text
    method, key_url, ts_list = parse_m3u8_text(m3u8)
    
    key_url = 'https://play.bo262626.com' + key_url
    ts_list = ['https://play.bo262626.com' + item for item in ts_list]
    
    key = requests.get(key_url, headers=headers).content
    ts_file_list = download_method_2(ts_list, key=key, processes_nums=10)
    
    merge_ts_to_mp4('test.mp4', ts_file_list)

二、視頻分析

1. 認(rèn)識m3u8文件

m3u8的結(jié)構(gòu)詳細(xì)分析可以看這個鏈接m3u8 文件格式詳解 - 簡書 (jianshu.com)，這里我們只簡要介紹一下；

相信無論多小白都應(yīng)該知道如何打開開發(fā)者模型解析得到下面的結(jié)果；

要注意的是，只有預(yù)覽里面包含了ts信息的才算是我們需要的m3u8文件；大家可以看到左側(cè)有兩個index.m3u8，其中一個是沒有ts信息的，所以我們直接忽略掉；現(xiàn)在我們先下載來，再來具體分析一下m3u8文件以及里面的內(nèi)容分別表達(dá)什么意思；

下載代碼如下：

import requests
import re

headers = {
    'Accept': '*/*',
    'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
    'Connection': 'keep-alive',
    'Origin': 'http://www.kpd510.me',
    'Referer': 'http://www.kpd510.me/',
    'Sec-Fetch-Dest': 'empty',
    'Sec-Fetch-Mode': 'cors',
    'Sec-Fetch-Site': 'cross-site',
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/116.0.0.0 Safari/537.36 Edg/116.0.1938.69',
    'sec-ch-ua': '"Chromium";v="116", "Not)A;Brand";v="24", "Microsoft Edge";v="116"',
    'sec-ch-ua-mobile': '?0',
    'sec-ch-ua-platform': '"Windows"',
}

response = requests.get('https://play.bo262626.com/20231108/xV1bY9Cn/700kb/hls/index.m3u8', headers=headers)

m3u8 = response.text

m3u8文件的實(shí)質(zhì)是一個視頻的url列表，其中ts是計(jì)算器可以直接播放的視頻格式文件，但是直接下載是可能被加了密的文件，我們需要m3u8文件內(nèi)容信息進(jìn)行解密；

我們可以這樣理解，m3u8把一個完整的mp4視頻切割成了很多的小塊，每一個小塊在m3u8都是ts文件格式，并在m3u8中采取了加密的措施，至于為什么要加密，這里就不多介紹；

在一般的視頻爬取中，我們只需要考慮兩個部分，一個是EXT-X-KEY，一個是ts；

其中EXT-X-KEY包含了ts的加密方式，ts包含了ts文件的下載地址；

在紅色部分也就是EXT-X-KEY部分，我們可以從METHOD中獲取到采取的加密方式是AES-128，同時看到URI的地址/20231126/10VkaJks/700kb/hls/key.key，這也就是AES加密密匙的地址：key.key，接下來我們寫一個文件來對m3u8文件解析，目的是提取出紅色部分和藍(lán)色部分；

代碼如下：

def parse_m3u8_text(m3u8_text):
    m3u8_text = m3u8_text.split()
    encode_info = [line for line in m3u8_text if line.startswith('#EXT-X-KEY:')][0]
    pattern = r"#EXT-X-KEY:METHOD=(.*),URI=\"(.*)\""  
    
    ## 獲得加密method 和 key.key的url
    match = re.search(pattern, encode_info)
    if match:
        method = match.group(1)
        key_url = match.group(2)
    else:
        raise '解析失敗'
    
    ## 獲得ts文件url
    ts_list = [line for line in m3u8_text if line.endswith('ts')]
    return method, key_url, ts_list

## 在這里我們直接把m3u8文本丟進(jìn)去就可以獲得
## method, key_url, ts_list
method, key_url, ts_list = parse_m3u8_text(m3u8)
## method = 'AES-128'
## key_url = '/20231108/xV1bY9Cn/700kb/hls/key.key'
## ts_list = ['...ts', '...ts', ...]

2. 獲取密鑰，構(gòu)建解密器

因?yàn)闃?gòu)建解密器我們需要密鑰，而密鑰存儲在key.key中，首先我們需要解析key_url獲取密鑰；

在這里可以明顯的看到key_url = '/20231108/xV1bY9Cn/700kb/hls/key.key'這不是一個完整的url，我們在這里加上獲取m3u8請求的主域名便好；

代碼如下：

headers = {
    'Accept': '*/*',
    'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
    'Connection': 'keep-alive',
    'Origin': 'http://www.kpd510.me',
    'Referer': 'http://www.kpd510.me/',
    'Sec-Fetch-Dest': 'empty',
    'Sec-Fetch-Mode': 'cors',
    'Sec-Fetch-Site': 'cross-site',
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/116.0.0.0 Safari/537.36 Edg/116.0.1938.69',
    'sec-ch-ua': '"Chromium";v="116", "Not)A;Brand";v="24", "Microsoft Edge";v="116"',
    'sec-ch-ua-mobile': '?0',
    'sec-ch-ua-platform': '"Windows"',
}

key_url = 'https://play.bo262626.com' + key_url
key = requests.get(key_url, headers=headers).content
# 這里我們得到key = b'388d590fabfeabcf' 是一個二進(jìn)制結(jié)果

得到了密鑰，再結(jié)合加密方式AES-128，我們就可以構(gòu)建一個解密器，構(gòu)建解密器代碼如下：

from Crypto.Cipher import AES
## 這里網(wǎng)絡(luò)爬取視頻一般是MODE_CBC模式
decrypter = AES.new(key, AES.MODE_CBC)

這里要提起的是網(wǎng)絡(luò)上的m3u8文件采取的加密一般是AES.MODE_CBC模式，在后續(xù)操作中如果這里出問題就換MODE一個一個試就好；

3. 下載ts文件

由于有許多的ts文件，我們有三種方法，第一是簡單的requests請求一個一個下，這也是最費(fèi)時的一種；第二個是多進(jìn)程或者多線程的方式下載；第三個是采用協(xié)程的方式；接下來我們一個個實(shí)現(xiàn)；

在開始之間，ts_list存在同樣的問題，就是需要重構(gòu)url，這里代碼如下：

ts_list = ['https://play.bo262626.com' + item for item in ts_list]

# 這里得到：
# ['https://play.bo262626.com/20231108/xV1bY9Cn/700kb/hls/o3jSJ9mc.ts',
#  'https://play.bo262626.com/20231108/xV1bY9Cn/700kb/hls/GNHDlClJ.ts',
#  'https://play.bo262626.com/20231108/xV1bY9Cn/700kb/hls/zKym5c6V.ts',
#  'https://play.bo262626.com/20231108/xV1bY9Cn/700kb/hls/4ll4NQH3.ts',
#  'https://play.bo262626.com/20231108/xV1bY9Cn/700kb/hls/RwUOniSQ.ts' ...]

再測試一下解密器是否可以：

import os
from tqdm import tqdm

dirs = 'ts_list_need_to_merge/'
os.makedirs(dirs, exist_ok=True)


def decrypt_content_and_save_file(filename, content, decrypter):
    with open(filename, mode='wb') as f:
        f.write(decrypter.decrypt(content))


def process_one_url(ts_url, key):
    decrypter = AES.new(key, AES.MODE_CBC)
    filename = dirs + os.path.split(ts_url)[-1]
    content = requests.get(ts_url, headers=headers).content
    decrypt_content_and_save_file(filename, content, decrypter)
    return filename

process_one_url(ts_list[0], key)

## 打開視頻看是否能打開
## 如果能打開說明解密沒問題

直接requests: 代碼如下

def download_method_1(ts_list, key):
    # 這里弄一個filename_list 方便后續(xù)合并ts到mp4
    ts_file_list = []
    for ts_url in tqdm(ts_list):
        filename = process_one_url(ts_url=ts_url, key=key)
        ts_file_list.append(filename)
    return ts_file_list

# 下載測試
ts_file_list = download_method_1(ts_list, key)

實(shí)現(xiàn)挺慢的，不合理；

多進(jìn)程: 代碼如下

def download_method_2(ts_list, key, processes_nums=2):
    class CallBack:
        def __init__(self, nums) -> None:
            self.pbar = tqdm(total=nums)
            self.filenames = []
        def callback(self, filename):
            self.pbar.update(1)
            self.filenames.append(filename)

    callback = CallBack(len(ts_list))
    pool = Pool(processes=processes_nums)
    for ts_url in ts_list:
        pool.apply_async(process_one_url, (ts_url, key), error_callback=print, callback=callback.callback)
    pool.close()
    pool.join()
    callback.pbar.close()
    return [dirs + os.path.split(ts_url)[-1] for ts_url in ts_list]

ts_file_list = download_method_2(ts_list, key=key, processes_nums=10)

爬取巨快，1分鐘下載20多分鐘的視頻；

4. 合并ts文件為mp4

在完成前面的步驟后，直接ab的方式把所有的文件按順序加入就好；

def merge_ts_to_mp4(filename, ts_file_list):
    with open(filename, mode='ab') as f1:
        for ts_file in ts_file_list:
            with open(ts_file, mode='rb') as f2:
                f1.write(f2.read())

merge_ts_to_mp4('test.mp4', ts_file_list)

后續(xù)如果需要刪除'ts_list_need_to_merge/'這個臨時文件夾里面的所有內(nèi)容，直接運(yùn)行下面代碼

import send2trash

send2trash.send2trash('ts_list_need_to_merge/') # send2trash.send2trash(dirs)

三、總結(jié)

別在圖書館測試這段代碼！文章來源地址http://www.zghlxwxcb.cn/news/detail-829331.html

到了這里，關(guān)于python 爬蟲 m3u8 視頻文件 加密解密 整合mp4的文章就介紹完了。如果您還想了解更多內(nèi)容，請?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！