当前位置：首页 > news >正文

视频下载（m3u8或者其他格式的）

news 来源：原创 2024/9/20 13:48:28

m3u8_xz库

关于m3u8文件的下载网上有好多方法，但是没有一个拿来即用的，因此在这分享一个库，这个库也很简单

# 安装
pip install m3u8_xz

因为这个库是国内的网友写的，注释一看就懂，没事可以研究一下源码。

示例

from m3u8_XZ import m3u8# use m3u8 url 通过url
m3u8_url = 'https://svipsvip.ffzy-online5.com/20240721/30372_37f1038c/2000k/hls/mixed.m3u8'
obj = m3u8(m3u8_url, folder='test')
# use local file 通过本地文件
# m3u8(m3u8_file='fileName.m3u8', folder='test')
# 异步启动方式 old
# obj.run()
# 2024年5月23日 新增启动方式
obj.run(thread_num=20)

其他格式的文件下载

这个方法可以下载各种二进制文件，考虑到文件过大，内存爆掉的现象使用流的方式下载

# coding: utf-8
from pathlib import Path
from typing import Tuple
from copy import deepcopyimport requests
import urllib3urllib3.disable_warnings()def ExceptionHandler(*default):""" decorator for exception handlingParameters----------*default:the default value returned when an exception occurs"""def outer(func):def inner(*args, **kwargs):try:return func(*args, **kwargs)except BaseException as e:value = deepcopy(default)if len(value) == 0:return Noneelif len(value) == 1:return value[0]return valuereturn innerreturn outerclass StreamDownload:def __init__(self,url: str,folder: str,name: str,headers: dict = None,chunk_size: int = 1024,print_callback=None):"""StreamDownload类从url下载文件。:param url::param folder: 保存文件夹:param name: 文件名:param headers::param chunk_size::param print_callback: 打印回调函数"""self.url = urlself.folder = folderself.name = nameself.chunk_size = chunk_sizeself.headers = headers or {}self._isStop = Falseself.print_callback = print_callback@ExceptionHandler(0)def content_length(self):StreamDownload.log(f'url: {self.url}, 正在获取内容长度，请稍候...')response = requests.head(self.url, verify=False, headers=self.headers)response.raise_for_status()return int(response.headers.get('content-length'))def get_start_size(self) -> Tuple[Path, Path, int]:name = self.folder / self.namename.parent.mkdir(parents=True, exist_ok=True)temp_name = name.with_suffix('.temp')if not name.exists() and not temp_name.exists():StreamDownload.log('文件不存在，创建临时文件')temp_name.touch()if name.exists():start_size = name.stat().st_sizeStreamDownload.log('文件已存在')elif temp_name.exists():start_size = temp_name.stat().st_sizeelse:start_size = 0return name, temp_name, start_sizedef run(self):"""开始下载:return:"""name: Pathtemp_name: Pathstart_size: intresponse_length = self.content_length()if response_length == 0:returnStreamDownload.log(f'内容长度: {response_length} kb')name, temp_name, start_size = self.get_start_size()progress = start_size / self.chunk_sizeif response_length == start_size:returnheaders = self.headers.copy()headers['Range'] = f'bytes={start_size}-'response = requests.get(self.url, headers=headers, stream=True, verify=False)response.raise_for_status()with open(temp_name, 'ab') as fp:for index, content in enumerate(response.iter_content(chunk_size=self.chunk_size)):if self._isStop:breakif content:fp.write(content)loaded_num = (index + 1 + progress) * self.chunk_sizeload_count = response_lengthStreamDownload.log('\r进度: {:.2f}%'.format(loaded_num / load_count * 100),end='', flush=True)if self.print_callback:self.print_callback(loaded_num=loaded_num, load_count=load_count)temp_name.rename(name)@staticmethoddef log(*args, **kwargs):print(*args, **kwargs)def stop(self):"""停止下载:return:"""self._isStop = Trueif __name__ == '__main__':mp4_url = ('http://v16m-default.akamaized.net/d4979b2fc814c79cdce8747e082e6bc2/669fcbbc/video/tos/alisg/tos-alisg-v-0000/osPoAgeQBuamDMGMxek8GdDr0nPebyAcbpCvrg''/?a=2011&bti=MzhALjBg&ch=0&cr=0&dr=0&net=5&cd=0%7C0%7C0%7C0&br=3432&bt=1716&cs=0&ds=4&ft=XE5bCqT0mmjPD12fdGK73wU7C1JcMeF~O5&mime_type=video_mp4&qs=0&rc=NTx''lMzg2ODxlaWllOzs3O0BpM2h2aDo6Znk3ZzMzODYzNEAuLjReYl5fNV4xNV9gMzUuYSMwYzNkcjQwa3FgLS1kMC1zcw%3D%3D&vvpl=1&l=20240723090214D6FEEF98890A2308D775&btag=e000a8000')stream_download = StreamDownload(mp4_url, 'test1', 'test.mp4')stream_download.run()