[Python] 纯文本查看 复制代码
import os
import aiofiles
import aiohttp
import requests
import re
from urllib.parse import urljoin
import hashlib
import binascii
from Crypto.Cipher import AES
from Crypto.Util.Padding import unpad
import base64
import time
headers = {
}
# 加密解密相关函数
def aes_ecb_decrypt(key, ciphertext_base64):
"""
使用AES-ECB模式解密数据。
参数:
key: 解密密钥，必须是16、24或32字节长。
ciphertext_base64: Base64编码的密文字符串。
返回:
解密后的明文字符串。
"""
# 确保密钥长度符合AES要求
if len(key) not in [16, 24, 32]:
raise ValueError("Key length must be 16, 24, or 32 bytes")
# 将Base64编码的密文转换为字节串
ciphertext = base64.b64decode(ciphertext_base64)
# 初始化AES-ECB解密器
cipher = AES.new(key, AES.MODE_ECB)
# 执行解密
decrypted_padded = cipher.decrypt(ciphertext)
# 移除PKCS7填充
decrypted_text = unpad(decrypted_padded, AES.block_size)
return decrypted_text
def aes_cbc_decrypt(key, ciphertext, iv):
"""
使用AES-CBC模式解密数据。
参数:
key: 解密密钥，必须是16、24或32字节长。
ciphertext: 密文字节串。
iv: 初始化向量（IV），字节串形式。
返回:
解密后的明文字符串。
"""
# 确保密钥长度符合AES要求
if len(key) not in [16, 24, 32]:
raise ValueError("Key length must be 16, 24, or 32 bytes")
# 使用CBC模式初始化解密器
cipher = AES.new(key, AES.MODE_CBC, iv)
# 执行解密
decrypted_padded = cipher.decrypt(ciphertext)
# 移除PKCS7填充
decrypted_text = unpad(decrypted_padded, AES.block_size)
return decrypted_text
# MD5加密函数
def md5_encrypt(input_string):
"""
对输入字符串进行MD5加密。
参数:
input_string: 需要加密的字符串。
返回:
MD5加密后的16进制字符串表示。
"""
# 创建一个md5哈希对象
md5_hash = hashlib.md5()
# 更新哈希对象，使它包含要加密的数据
md5_hash.update(input_string.encode('utf-8'))
# 获取16进制形式的哈希值
encrypted_string = md5_hash.hexdigest()
return encrypted_string[:16]
def get_signs(m3u8_url):
"""
从给定的m3u8_url中提取签名信息和ts文件列表。
参数:
m3u8_url (str): m3u8视频链接地址。
返回:
tuple: 包含ts文件列表、解密密钥和IV向量的元组。
"""
# 获取m3u8文件内容并保存到本地文件
resp = requests.get(m3u8_url, headers=headers)
with open('test.m3u8', 'w', encoding='utf-8') as f:
f.write(resp.text)
# 读取本地m3u8文件，提取ts文件链接和加密信息
ts_lst = []
with open('test.m3u8', 'r', encoding='utf-8') as f:
for line in f.readlines():
if line.startswith('#'):
if 'key' in line:
key_url = re.findall(r'URI="(.*?)"', line)[0]
key = re.findall(r'_keys/(.*?)"', line)[0]
iv = re.findall(r'IV=(.*?)\n', line)[0]
iv = binascii.unhexlify(iv.replace('0x', "")).hex()[:16].encode('utf-8')
continue
else:
ts_lst.append(urljoin(m3u8_url, line))
# 构建获取签名的URL并请求签名信息
sign_url = key_url + '/signs'
resp_signs = requests.get(sign_url, headers=headers)
nonce = resp_signs.json()['nonce']
md5_result = md5_encrypt(nonce + key)
# 构建带签名的URL并请求解密密钥
new_signs_url = key_url + '?nonce=' + nonce + '&sign=' + md5_result
resp_new_signs = requests.get(new_signs_url, headers=headers)
id_key = resp_new_signs.json()
data_base64 = id_key['key']
ts_key = aes_ecb_decrypt(md5_result.encode('utf-8'), data_base64)
return ts_lst, ts_key, iv
# 合并TS文件函数
def merge_ts(path):
# 获取当前工作目录，用于后续返回原始目录
current_dir = os.getcwd()
# 将当前工作目录设置为新的目标目录
new_dir = os.chdir(path)
# 列出新目录中的所有文件和目录
all_entries = os.listdir(new_dir)
# 构建合并所有文件为一个MP4文件的命令
# 使用'copy /b'命令将所有.ts文件合并为一个.all.mp4文件
cmd = 'copy /b ' + '+'.join(all_entries) + ' ' + 'result.mp4'
# 执行命令
r = os.system(cmd)
# 检查命令执行结果，如果返回值为0，则表示合并成功
if r == 0:
print('合并成功')
else:
print('合并失败')
# 删除所有.ts文件
r = os.system('del *.ts')
# 返回原始工作目录
os.chdir(current_dir)
# 删除所有.m3u8文件
r = os.system('del *.m3u8')
# 异步下载TS文件函数
async def download_ts(ts, ts_key, iv, path):
# 通过拼接路径和时间戳后缀来构造文件名
file_name = path + '/' + ts.split('-')[-1]
# 尝试下载文件，最多重试10次
for i in range(10):
try:
# 开始下载文件，打印文件名以供跟踪
print('开始下载', file_name)
# 使用aiohttp创建一个客户端会话，用于异步HTTP请求
async with aiohttp.ClientSession() as session:
# 发起GET请求来下载文件
async with session.get(ts) as resp:
# 读取响应内容
ts_data = await resp.content.read()
# 使用AES解密响应内容
ts_data = aes_cbc_decrypt(ts_key, ts_data, iv)
# 使用aiofiles打开文件以异步写入解密后的数据
async with aiofiles.open(file_name, 'wb') as f:
await f.write(ts_data)
# 文件下载完成，打印文件名以供确认
print('下载完成', file_name)
# 成功下载后跳出循环
break
except Exception as e:
# 下载过程中出现异常，打印文件名和异常信息，然后尝试重新下载
print(file_name, '开始重新下载。。', e)
continue
# 主函数入口
async def run_main():
while True:
# 获取M3U8文件列表
m3u8_lst = get_m3u8_url()
for m3u8_url in m3u8_lst:
# 以当前时间戳创建唯一目录名，用于存储下载的TS文件
path = str(int(time.time())) + 'ts'
# 检查目录是否存在，不存在则创建
if not os.path.exists(path):
os.makedirs(path)
# 解析M3U8文件，获取TS文件列表、加密密钥和IV
# get_signs(m3u8_url)
ts_lst, ts_key, iv = get_signs(m3u8_url)
# 创建异步任务下载前10个TS文件
tasks = [asyncio.create_task(download_ts(ts, ts_key, iv, path)) for ts in ts_lst]
# 同步执行所有异步任务
await asyncio.gather(*tasks)
# 合并下载的TS文件为完整的视频文件
merge_ts(path)
def get_m3u8_url():
"""
获取m3u8视频链接列表。
通过用户输入的视频地址或m3u8地址，解析出m3u8链接列表。
如果用户输入的是m3u8地址，直接返回该地址列表。
如果用户输入的是视频地址，通过地址中的activityId获取相应的json文件，从中解析出m3u8链接。
"""
# 初始化m3u8链接列表
lst_m3u8 = []
# 获取用户输入的视频或m3u8地址
video_url = input('请输入视频地址或者m3u8地址(输入多个m3u8地址以,分开)：')
# 如果用户输入包含m3u8，则认为是m3u8地址列表，直接返回
if 'm3u8' in video_url:
lst_m3u8 = video_url.split(',')
return lst_m3u8
# 从视频地址中提取activityId
id = video_url.split('activityId=')[1].split('&')[0]
# 构造请求json文件的url
json_url = f'https://s-file-2.ykt.cbern.com.cn/zxx/ndrv2/national_lesson/resources/details/{id}.json'
# 发送请求，获取json数据
resp = requests.get(json_url)
json_data = resp.json()
# 从json数据中提取m3u8链接
resource = json_data['relations']["national_course_resource"]
# 遍历资源列表，每5个资源提取一个m3u8链接
for i in range(len(resource)):
if i % 5 == 0:
m3u8_url = resource[i]['ti_items'][0]['ti_storages'][0]
lst_m3u8.append(m3u8_url)
return lst_m3u8
if __name__ == '__main__':
print('------------------仅做练习，请勿用于商业用途------------------')
print('------------------仅做练习，请勿用于非法用途------------------')
print('------------------仅做练习，请勿滥用造成不良后果自负------------')
asyncio.run(run_main())
[复制链接]
发表于 2024-6-24 22:44
|
发表于 2024-6-24 23:07
