HLS

即HTTP Live Streaming，将流媒体切分为若干TS片段，通过一个M3U8列表文件将TS片段批量下载实时流式播放。

#EXTM3U:第一行tag标识。#EXT-X-VERSION:版本#EXT-X-TARGETDURATION:定义每个TS的【最大】duration（持续时间）#EXT-X-ALLOW-CACHE高速缓存#EXT-X-MEDIA-SEQUENCE定义当前M3U8文件中第一个文件的序列号，每个ts文件在M3U8文件中都有固定唯一的序列号#EXT-X-DISCONTINUITY:播放器重新初始化#EXT-X-KEY加密方式（例如AES-128），URL密钥文件key的地址，IV加密向量(如果没有，就用16个0代替)#EXTINF:指定每个媒体段(ts文件)的持续时间，这个仅对其后面的TS链接有效，每两个媒体段(ts文件)间被这个tag分隔开#EXT-X-ENDLIST文件的结束

找m3u8

• 下方的{}，点击后可提高阅读性

AES加、解密

解密AES,需安装包pycryptodome
pip安装pycryptodome有时会报错，装低一点的版本解决

pip install pycryptodome
from Crypto.Cipher import AES
aes = AES.new(key, IV=b"0000000000000000", mode=AES.MODE_CBC)
decrypt_content = aes.decrypt(content)

实例

链接: link

import requests
from lxml import etree
import re
from urllib import parse
import asyncio
import aiohttp
import aiofiles
import os
#   解密AES,需安装包pycryptodome
from Crypto.Cipher import AESdef get_key(url):return get_page_source(url)
def get_page_source(url):headers = {"Referer": "http://www.wbdy.tv/","User-Agent": "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.5735.289 Safari/537.36"}resp = requests.get(url, headers=headers)# print(resp.status_code)return resp.content
async def download_one_ts(save_path, url):# for c in range(10):while True:try:c = 0# print(url)print(save_path)async with aiohttp.ClientSession() as session:async with session.get(url) as res:movie_content = await res.content.read()# 存储文件async with aiofiles.open(save_path, mode="wb") as f:await f.write(movie_content)print("download success,url:",url)# breakreturn ""except:c += 1await asyncio.sleep((c+1)*4)print("下载失败，准备重下,url:",url)return url
async def downloads_ts(save_path, ts_path):tasks = []name = 1with open(ts_path, mode='r', encoding='utf-8') as f:for line in f:if line.startswith('#'):continuetask = asyncio.create_task(download_one_ts(save_path+f"\{name}.ts", line),)tasks.append(task)# 以自然数对ts文件命名name += 1await asyncio.wait(tasks)
async def decrypt_ts(file_path, save_path, key):# 默认iv向量是16个0print(file_path)print('*'*20)print(save_path)async with aiofiles.open(file_path, mode="rb") as fr, aiofiles.open(save_path, mode="wb") as fw:content = await fr.read()aes = AES.new(key, IV=b"0000000000000000", mode=AES.MODE_CBC)decrypt_content = aes.decrypt(content)await fw.write(decrypt_content)print(f"解密成功, 文件被存放在{save_path}")
async def aio_decrypt(file_path_ts,new_file_path, key):ts_nums = len(os.listdir(file_path_ts))ts_lists = os.listdir(file_path_ts)tasks = []now_dir = os.getcwd()for name in ts_lists:file_path = os.path.join(now_dir, file_path_ts, name)new_name_path = os.path.join(now_dir, new_file_path, name)tasks.append(asyncio.create_task(decrypt_ts(file_path, new_name_path, key)))# result = await asyncio.gather(*tasks)# return resultawait asyncio.wait(tasks)
# print(res.encode("utf-8"))
def down_movie():
# 下载ts文件loop = asyncio.get_event_loop()loop.run_until_complete(downloads_ts(os.path.join(now_dir, encrypt_dir), './index.m3u8'))print(key)
# 解码ts
def decrypt_movie():loop = asyncio.get_event_loop()loop.run_until_complete(aio_decrypt('en_teacher3', 'de_teacher3', key))
def merge_ts_mp4(movie_name):current_dir = os.getcwd()ts_list = len(os.listdir('./de_teacher3'))os.chdir('./de_teacher3')lists = []big_list = []list_temp = []big = 0for i in range(ts_list):lists.append(i + 1)for i in range(len(lists)):temp = lists[i]list_temp.append(str(temp))if i != 0 and i % 50 == 0:print(list_temp)big += 1list_temp[-1] = list_temp[-1] + '.ts'name = ".ts + ".join(list_temp)print(name)os.system(f"copy /b {name} big{big}.ts")big_list.append(f"big{big}.ts")# print(f"copy /b {name} big{big}.ts")list_temp = []last_name = "+".join(big_list)os.system(f"copy /b {last_name} {movie_name}.ts")os.chdir(current_dir)
if __name__ == '__main__':# 当前文件目录now_dir = os.getcwd()# 未解密的ts文件存放目录encrypt_dir = 'en_teacher3'if not os.path.exists(encrypt_dir):os.mkdir(encrypt_dir)# 解密的ts文件存放目录decrypt_dir = 'de_teacher3'if not os.path.exists(decrypt_dir):os.mkdir(decrypt_dir)# 加密的keykey_url = 'https://hnzy.bfvvs.com/play/zbqDW1ka/enc.key'key = get_key(key_url)# 下载ts文件# down_movie()# 解码ts# decrypt_movie()# 合并ts成mp4merge_ts_mp4("电影的名字")