【python】M3U8下载器脚本
脚本目标：
- 输入M3U8文件的链接，得到视频
2.使用异步操作，这样可以快很多，不加锁，因为懒得写，而且影响不大
已知条件：
1.m3u8文件其实就是一个记录了ts文件下载链接的工具文件，每个ts文件就是视频的一部分，把所有ts文件下载下来，合并就可以得到完整的视频
脚本思路：
1.创建一个文件夹，用来存放下载好的m3u8文件和下载好的ts文件
2.下载并打开m3u8文件，根据m3u8文件下载ts文件，这边设计了两种情况，a.ts的下载链接是完整的 b.ts的下载链接是需要拼接的
3.根据m3u8文件自动校验文件是否下载完整
4.由于很多时候ts文件的命名是没有规律的，所以再次打开m3u8文件，根据里面的顺序，以追加的形式写入到一个新的ts文件里
代码实现：
先创建好文件夹，这边使用了相对路径
def init():
if os.path.exists("./temp_data"):
return
else:
os.mkdir("./temp_data")
得到m3u8下载链接，获得m3u8文件名，这边假设是https://xxxxxxx126.net/nos/hls/2019/03/13/1214418271_9xxxxxxx32465d1f4c8_sd.m3u8，那么就设置”1214418271_9xxxxxxx32465d1f4c8_sd.m3u8″为文件名
url =str(input("输入m3u8文件url >"))
name = url.rsplit("/")[-1]
下载m3u8文件
def m3u8_files_download(url,name): #下载m3u8文件
resp = requests.get(url)
with open(f"temp_data/{name}.txt",mode="wb") as f:
f.write(resp.content)
resp.close()
给出第一个ts的下载链接，用户自己判断一下是需要拼接的，还是无需拼接的完整url
def get_type(name):
with open(f"temp_data/{name}.txt","r") as f:
for line in f:
if line.startswith("#"):
continue
else:
print("内容为:",line)
print("选择模式： 1.直接下载型 2.拼接型")
choice = input(">")
return str(choice)
写一个启动器，根据不同的选择，创建不同的任务，创建的任务为异步任务
async def starter(choice,name):
tasks=[]
async with aiohttp.ClientSession() as session:
if choice =="1":
with open(f"/temp_data/{name}.txt","r") as f:
for line in f:
if line.startswith("#"):
continue
else:
download_url = line.strip()
line = line.split("/")
file_name = str(line[-1]).strip() # 得下载的ts文件名
task = download_ts(file_name,download_url,session)
tasks.append(task)
print("文件下载中.....")
await asyncio.wait(tasks) # 等待任务执行结束
print("文件下载完成")
if choice=="2":
url = str(input("输入拼接的url>"))
with open(f"temp_data/{name}.txt","r") as f:
for line in f:
if line.startswith("#"):
continue
else:
line = line.strip()
file_name = line # 得下载的ts文件名
download_url = url+line
task = download_ts(file_name,download_url,session)
tasks.append(task)
print("文件下载中.....")
await asyncio.wait(tasks) # 等待任务执行结束
print("文件下载完成")
下载ts文件，用aiohttp来代理requests
async def aio_download_ts(download_url,line_name,session):
async with session.get(download_url,headers=header) as resp:
async with aiofiles.open(f"temp_data/{line_name}",mode="wb") as f:
await f.write(await resp.content.read())
print(f"文件{line_name}下载完成！！")
校验文件的完整性：依据m3u8文件，判断文件是否存在
def verification(name):
files=[]
with open(f"temp_data/{name}.txt","r") as f:
for line in f:
if line.startswith("#"):
continue
else:
line=line.strip()
if os.path.exists(f"temp_data/{line}"):
continue
else:
files.append(line)
print("以下文件缺失，请手动查看:",files)
合并文件，实现的方式时创建一个ts文件，依据m3u8文件里的文件顺序，依次将二进制文件写入到新的ts文件里
def merge_ts(file_name):
new_name = str(input("输入合并后的文件名>"))
with open(f"./{new_name}.ts", "ab+") as f:
with open(f"temp_data/{file_name}.txt","r") as f2:
for line in f2:
if line.startswith("#"):
continue
else:
line = line.strip().split("/")[-1].strip()
ts_name = line
try:
with open(f"temp_data/{ts_name}","rb") as f3:
f.write(f3.read())
except:
continue
最后再写一个主函数，执行这一切
def main():
init()
url =str(input("输入m3u8文件url >"))
name = url.rsplit("/")[-1]
m3u8_files_download(url,name)#下载m3u8文件
choice=get_type(name)
asyncio.run(starter(choice,name))
print("校验文件完整性")
verification(name)
print("是否合并文件？ Y/N")
if str(input(">"))=="Y":
merge_ts(name)
else:
print("结束")
最终功能代码
import aiohttp
import aiofiles
import asyncio
import requests
import os
header = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/96.0.4664.93 Safari/537.36"}
def merge_ts(file_name):
new_name = str(input("输入合并后的文件名>"))
with open(f"./{new_name}.ts", "ab+") as f:
with open(f"temp_data/{file_name}.txt","r") as f2:
for line in f2:
if line.startswith("#"):
continue
else:
line = line.strip().split("/")[-1].strip()
ts_name = line
try:
with open(f"temp_data/{ts_name}","rb") as f3:
f.write(f3.read())
except:
continue
async def aio_download_ts(download_url,line_name,session):
async with session.get(download_url,headers=header) as resp:
async with aiofiles.open(f"temp_data/{line_name}",mode="wb") as f:
await f.write(await resp.content.read())
print(f"文件{line_name}下载完成！！")
def m3u8_files_download(url,name): #下载m3u8文件
resp = requests.get(url)
with open(f"temp_data/{name}.txt",mode="wb") as f:
f.write(resp.content)
resp.close()
def get_type(name):
with open(f"temp_data/{name}.txt","r") as f:
for line in f:
if line.startswith("#"):
continue
else:
print("内容为:",line)
print("选择模式： 1.直接下载型 2.拼接型")
choice = input(">")
return str(choice)
def init():
if os.path.exists("./temp_data"):
return
else:
os.mkdir("./temp_data")
def verification(name):
files=[]
with open(f"temp_data/{name}.txt","r") as f:
for line in f:
if line.startswith("#"):
continue
else:
line=line.strip()
if os.path.exists(f"temp_data/{line}"):
continue
else:
files.append(line)
print("以下文件缺失，请手动查看:",files)
async def download_ts(file_name,download_url,session):
async with session.get(download_url,headers=header) as resp:
async with aiofiles.open(f"temp_data/{file_name}",mode="wb") as f:
await f.write(await resp.content.read())
async def starter(choice,name):
tasks=[]
async with aiohttp.ClientSession() as session:
if choice =="1":
with open(f"/temp_data/{name}.txt","r") as f:
for line in f:
if line.startswith("#"):
continue
else:
download_url = line.strip()
line = line.split("/")
file_name = str(line[-1]).strip() # 得下载的ts文件名
task = download_ts(file_name,download_url,session)
tasks.append(task)
print("文件下载中.....")
await asyncio.wait(tasks) # 等待任务执行结束
print("文件下载完成")
if choice=="2":
url = str(input("输入拼接的url>"))
with open(f"temp_data/{name}.txt","r") as f:
for line in f:
if line.startswith("#"):
continue
else:
line = line.strip()
file_name = line # 得下载的ts文件名
download_url = url+line
task = download_ts(file_name,download_url,session)
tasks.append(task)
print("文件下载中.....")
await asyncio.wait(tasks) # 等待人物执行结束
print("文件下载完成")
def main():
init()
url =str(input("输入m3u8文件url >"))
name = url.rsplit("/")[-1]
m3u8_files_download(url,name)#下载m3u8文件
choice=get_type(name)
asyncio.run(starter(choice,name))
print("校验文件完整性")
verification(name)
print("是否合并文件？ Y/N")
if str(input(">"))=="Y":
merge_ts(name)
else:
print("结束")
main()
使用自欺欺人术，直接把ts文件后缀改成MP4，看着舒服点。
实现效果
视频打开能正常观看，脚本完成
后记：关于脚本的使用
理论上把aiohttp，aiofiles，asyncio三个库安装好，复制粘贴应该就可以直接用，也可以把一些需要手工提供的量，在脚本中写死，以在不同的爬虫中使用。
ENDING……….
Original: https://www.cnblogs.com/water-wells/p/15797812.html
Author: 水风井
Title: 【python】M3U8下载器脚本
原创文章受到原创版权保护。转载请注明出处：https://www.johngo689.com/603131/
转载文章受原作者版权保护。转载请注明原作者出处！