视频下载(m3u8或者其他格式的)
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 deepcopy
import requests
import urllib3
urllib3.disable_warnings()
def ExceptionHandler(*default):
""" decorator for exception handling
Parameters
----------
*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 None
elif len(value) == 1:
return value[0]
return value
return inner
return outer
class 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 = url
self.folder = folder
self.name = name
self.chunk_size = chunk_size
self.headers = headers or {}
self._isStop = False
self.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.name
name.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_size
StreamDownload.log('文件已存在')
elif temp_name.exists():
start_size = temp_name.stat().st_size
else:
start_size = 0
return name, temp_name, start_size
def run(self):
"""
开始下载
:return:
"""
name: Path
temp_name: Path
start_size: int
response_length = self.content_length()
if response_length == 0:
return
StreamDownload.log(f'内容长度: {response_length} kb')
name, temp_name, start_size = self.get_start_size()
progress = start_size / self.chunk_size
if response_length == start_size:
return
headers = 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:
break
if content:
fp.write(content)
loaded_num = (index + 1 + progress) * self.chunk_size
load_count = response_length
StreamDownload.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)
@staticmethod
def log(*args, **kwargs):
print(*args, **kwargs)
def stop(self):
"""
停止下载
:return:
"""
self._isStop = True
if __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()
原文地址:https://blog.csdn.net/weixin_54217201/article/details/140642056
免责声明:本站文章内容转载自网络资源,如本站内容侵犯了原著者的合法权益,可联系本站删除。更多内容请关注自学内容网(zxcms.com)!