使用python下载m3u8在线ts视频并解密合并

Python 使用python下载m3u8在线ts视频并解密合并

现在的在线视频都流行使用m3u8格式，加载一个ts网络地址的列表，分段加载视频。

通常，第二个才是ts文件的url列表，并说明是否进行了加密。

m3u8内容的格式比较简单。

这里在macos中使用python结合ffmpeg进行下载：

import os
import requests
from concurrent.futures import ThreadPoolExecutor, as_completed
from Cryptodome.Cipher import AES
from Cryptodome.Util.Padding import pad
from requests.adapters import HTTPAdapter


headers = {
    "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/84.0.4147.125 Safari/537.36"}
request = requests.Session()
request.mount('http://', HTTPAdapter(max_retries=3)) # 网络异常时再尝试3次
request.mount('https://', HTTPAdapter(max_retries=3))

m3u8Url = "https://video.buycar5.cn/20200902/2WeLRpAS/1000kb/hls/index.m3u8" # ts网址列表
tsKey = b"3071e5ddef9edbfb" # ts文件解密的key
tsList = [] # 按顺序保存ts，防止合并时错误

downloadPath = os.getcwd()  # 当前工作目录
downloadPath = os.path.join(downloadPath, "CSI") # 下载目录
if not os.path.exists(downloadPath):
    os.mkdir(downloadPath)  # 新建文件夹

### 下载单个ts
def downloadTs(tsUrl, tsPath):
    tsVideo = ""
    try:
        # print("执行下载：" + tsUrl) 
        tsVideo = request.get(tsUrl, headers=headers, timeout=15).content
        # print("{}大小：{}".format(tsPath, len(tsVideo)))
    except requests.exceptions.RequestException as e:
        print("下载失败：" + e)

    if tsVideo and (len(tsVideo) % 16) != 0:
        # EXT-X-KEY:METHOD=AES-128的加密方式，对应加密的key大小16byte，对应AES.MODE_CBC，
        # 如果视频大小不是16的倍数，解密时出现：ValueError(“Data must be padded to %d byte boundary in CBC mode” % self.block_size)
        tsVideo = pad(tsVideo, 16)  # 按16的倍数补齐长度

    cipher = AES.new(tsKey, AES.MODE_CBC, tsKey)  # AES 解密
    with open(tsPath, 'ab') as f:
        decryptedVideo = cipher.decrypt(tsVideo)
        f.write(decryptedVideo)
    print("下载了：" + tsPath)

### 合并文件
def merge():
    mp4Path = os.path.join(downloadPath, "tmp.ts")
    with open(mp4Path, 'wb+') as f: # 合并
        for tsPath in tsList:
            tsData = open(tsPath, 'rb').read() 
            f.write(tsData)
            f.flush() 
    
    os.chdir(downloadPath)
    os.system('ffmpeg -i tmp.ts -acodec copy -vcodec copy -f mp4 movie.mp4') # 转
    os.system('rm *.ts')
    print("合并完成：movie.mp4") 

### 下载目标视频
def download():
    m3u8Text = request.get(m3u8Url, headers=headers).text
    textLines = m3u8Text.split("\n")

    with ThreadPoolExecutor(max_workers=os.cpu_count() * 3) as taskThread: # 线程池
        taskList = []
        for index, line in enumerate(textLines):
            if "EXTINF" in line:  # EXTINF表示下一行是一段视频的url
                tsUrl = textLines[index + 1]  # 拼出ts片段的URL
                tsName = tsUrl.rsplit("/", 1)[-1]
                tsPath = os.path.join(downloadPath, tsName)
                tsList.append(tsPath)

                task = taskThread.submit(downloadTs, tsUrl, tsPath)  # 子线程
                taskList.append(task)

            if "EXT-X-DISCONTINUITY" in line:  
                # 后面的视频编码参数改变，初始化播放器。
                # 一般是影片正文结束，配合EXT-X-KEY:METHOD=NONE表示后面是未加密的广告
                break

        for future in as_completed(taskList):
            pass

    print("下载完成")
    merge()


if '__main__' == __name__:
    download()

import os

import requests

from concurrent.futures import ThreadPoolExecutor, as_completed

from Cryptodome.Cipher import AES

from Cryptodome.Util.Padding import pad

from requests.adapters import HTTPAdapter

headers = {

"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/84.0.4147.125 Safari/537.36"}

request = requests.Session()

request.mount('http://', HTTPAdapter(max_retries=3)) # 网络异常时再尝试3次

request.mount('https://', HTTPAdapter(max_retries=3))

m3u8Url = "https://video.buycar5.cn/20200902/2WeLRpAS/1000kb/hls/index.m3u8" # ts网址列表

tsKey = b"3071e5ddef9edbfb" # ts文件解密的key

tsList = [] # 按顺序保存ts，防止合并时错误

downloadPath = os.getcwd() # 当前工作目录

downloadPath = os.path.join(downloadPath, "CSI") # 下载目录

if not os.path.exists(downloadPath):

os.mkdir(downloadPath) # 新建文件夹

### 下载单个ts

def downloadTs(tsUrl, tsPath):

tsVideo = ""

try:

# print("执行下载：" + tsUrl)

tsVideo = request.get(tsUrl, headers=headers, timeout=15).content

# print("{}大小：{}".format(tsPath, len(tsVideo)))

except requests.exceptions.RequestException as e:

print("下载失败：" + e)

if tsVideo and (len(tsVideo) % 16) != 0:

# EXT-X-KEY:METHOD=AES-128的加密方式，对应加密的key大小16byte，对应AES.MODE_CBC，

# 如果视频大小不是16的倍数，解密时出现：ValueError(“Data must be padded to %d byte boundary in CBC mode” % self.block_size)

tsVideo = pad(tsVideo, 16) # 按16的倍数补齐长度

cipher = AES.new(tsKey, AES.MODE_CBC, tsKey) # AES 解密

with open(tsPath, 'ab') as f:

decryptedVideo = cipher.decrypt(tsVideo)

f.write(decryptedVideo)

print("下载了：" + tsPath)

### 合并文件

def merge():

mp4Path = os.path.join(downloadPath, "tmp.ts")

with open(mp4Path, 'wb+') as f: # 合并

for tsPath in tsList:

tsData = open(tsPath, 'rb').read()

f.write(tsData)

f.flush()

os.chdir(downloadPath)

os.system('ffmpeg -i tmp.ts -acodec copy -vcodec copy -f mp4 movie.mp4') # 转

os.system('rm *.ts')

print("合并完成：movie.mp4")

### 下载目标视频

def download():

m3u8Text = request.get(m3u8Url, headers=headers).text

textLines = m3u8Text.split("\n")

with ThreadPoolExecutor(max_workers=os.cpu_count() * 3) as taskThread: # 线程池

taskList = []

for index, line in enumerate(textLines):

if "EXTINF" in line: # EXTINF表示下一行是一段视频的url

tsUrl = textLines[index + 1] # 拼出ts片段的URL

tsName = tsUrl.rsplit("/", 1)[-1]

tsPath = os.path.join(downloadPath, tsName)

tsList.append(tsPath)

task = taskThread.submit(downloadTs, tsUrl, tsPath) # 子线程

taskList.append(task)

if "EXT-X-DISCONTINUITY" in line:

# 后面的视频编码参数改变，初始化播放器。

# 一般是影片正文结束，配合EXT-X-KEY:METHOD=NONE表示后面是未加密的广告

break

for future in as_completed(taskList):

pass

print("下载完成")

merge()

if '__main__' == __name__:

download()

恼人错误就是 ValueError: Data must be padded to 16 byte boundary in CBC mode ，解密时须要处理。

- end

声明

本文由 cuiweiyou 原创，转载请注明出处：http://www.gaohaiyan.com/2816.html

承接App定制、企业web站点、办公系统软件设计开发，外包项目，毕设

Android使用LayoutParams属性进行左右侧滑动画
Android注册、绑定（调用）远程（跨应用）服务
组件化时使用Butterknife出现错误: 元素值必须为常量表达式
物联网的几种方案