import json
import re
from concurrent.futures import ThreadPoolExecutor, wait, ALL_COMPLETED
import requests
from tqdm import tqdm
def get_response(html_url, stream=False):
headers = {
"referer": "https://www.bilibili.com/",
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36 Edg/108.0.1462.42"
}
response = requests.get(html_url, headers=headers, stream=stream)
return response
def get_url(html_url):
r = requests.get(html_url)
a = re.search(r"window\.__INITIAL_STATE__=(.*?)};", r.text).group(1)
json_data = json.loads(a + "}")
video_data = json_data['videoData']
avid = str(video_data['aid'])
qn = '112'
urls = []
for p in video_data['pages']:
cid = str(p['cid'])
title = p['part']
if len(video_data['pages']) == 1:
title = video_data['title']
urls.append({
'title': title,
"url": "https://api.bilibili.com/x/player/playurl?cid=" + cid + "&avid=" + avid + "&an=" + qn + "&otype=json&fourk=1"
})
return urls
def save(video_url, video_size, title):
video_res = get_response(video_url, True)
with open(r'C:\Users\NHT\Desktop\\'+title + '.mp4', 'wb') as fd:
print('开始下载文件:{}, 当前文件大小:{}KB'.format(title, video_size / 1024))
for c in tqdm(iterable=video_res.iter_content(), total=video_size, unit='b', desc=None):
fd.write(c)
print('{},文件下载成功'.format(title))
def a_single_download(info):
response = get_response(info['url'])
json_data = json.loads(response.text)
video_url = json_data['data']['durl'][0]['url']
video_size = json_data['data']['durl'][0]['size']
save(video_url, video_size, info['title'])
def concurrent_download(base_infos):
executor = ThreadPoolExecutor(max_workers=10)
futur_tasks = [executor.submit(a_single_download, info) for info in base_infos]
wait(futur_tasks, return_when=ALL_COMPLETED)
if __name__ == '__
详细解读如下,望采纳
这段代码实现了在 bilibili 网站上爬取视频并下载的功能。
它的主要流程如下:
在这段代码中,需要说明的一些点包括:
import json
import re
from concurrent.futures import ThreadPoolExecutor, wait, ALL_COMPLETED
import requests
from tqdm import tqdm
# 发送请求,获取响应
def get_response(html_url, stream=False):
# 设置请求头
headers = {
"referer": "https://www.bilibili.com/",
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36 Edg/108.0.1462.42"
}
# 发送请求,获取响应
response = requests.get(html_url, headers=headers, stream=stream)
return response
# 分析视频信息,生成视频地址列表
def get_url(html_url):
# 发送请求,获取页面 HTML
r = requests.get(html_url)
# 使用正则表达式,提取 JSON 数据
a = re.search(r"window\.__INITIAL_STATE__=(.*?)};", r.text).group(1)
# 将 JSON 数据转换为 Python 字典
json_data = json.loads(a + "}")
# 获取视频数据
video_data = json_data['videoData']
# 获取 avid 和 qn 参数
avid = str(video_data['aid'])
qn = '112'
# 创建存储视频地址的列表
urls = []
# 遍历视频分片
for p in video_data['pages']:
# 获取 cid 参数
cid = str(p['cid'])
# 获取视频分片标题
title = p['part']
# 如果该视频只有一个分片,则视频标题为整个视频的标题
if len(video_data['pages']) == 1:
title = video_data['title']
# 拼接视频地址
url = "https://api.bilibili.com/x/player/playurl?cid=" + cid + "&avid=" + avid + "&an=" + qn + "&otype=json&fourk=1"
# 将视频信息存储到字典中,并添加到列表中
urls.append({
'title': title,
"url": url
})
# 返回视频地址列表
return urls
# 保存视频
def save(video_url, video_size, title):
# 发送请求,获取视频响应
video_res = get_response(video_url, True)
# 打开文件
with open(r'C:\Users\NHT\Desktop\'+title + '.mp4', 'wb') as fd:
# 输出下载信息
print('开始下载文件:{}, 当前文件大小:{}KB'.format(title, video_size / 1024))
# 使用 tqdm 库,显示下载进度条
for c in tqdm(iterable=video_res.iter_content(), total=video_size, unit='b', desc=None):
fd.write(c)
# 输出下载成功信息
print('{},文件下载成功'.format(title))
# 单线程下载视频
def a_single_download(info):
# 获取视频详细信息
response = get_response(info['url'])
# 将 JSON 数据转换为 Python 字典
json_data = json.loads(response.text)
# 获取视频地址
video_url = json_data['data']['durl'][0]['url']
# 获取视频大小
video_size = json_data['data']['durl'][0]['size']
# 调用 save 函数,保存视频
save(video_url, video_size, info['title'])
# 多线程下载视频
def concurrent_download(base_infos):
# 创建线程池,设置最大线程数量为 10
executor= ThreadPoolExecutor(max_workers=10)
# 调用 map 函数,开启多线程进行视频下载
futur_tasks = [executor.submit(a_single_download, info) for info in base_infos]
# 等待所有线程完成,再继续执行后面的代码
wait(futur_tasks, return_when=ALL_COMPLETED)
if name == 'main':
# 视频地址
html_url = 'https://www.bilibili.com/video/BV1zE411S7V7'
# 获取视频地址列表
base_infos = get_url(html_url)
# 多线程下载视频
concurrent_download(base_infos)
Python的爬虫最简单了~核心就开头那个函数,剩下的就是对接收到的数据进行处理
标准的爬虫功能,get_response获取指定链接,save存储,download的两个函数下载,就这么简单
这是一个视频下载工具,它能够批量下载 Bilibili 视频。
代码中定义了如下几个函数:
get_response:向给定的 URL 发出 HTTP GET 请求,并返回响应。
get_url:从给定的 Bilibili 视频的 URL 中解析出该视频的信息,包括标题和下载地址。
save:下载视频并保存到指定的文件夹。
a_single_download:下载单个视频。
# 导入 json 模块,用于处理 JSON 数据。
import json
# 导入 re 模块,用于处理正则表达式。
import re
# 导入 concurrent.futures 模块,实现多线程编程。
from concurrent.futures import ThreadPoolExecutor, wait, ALL_COMPLETED
# 导入 requests 模块,方便发送 HTTP 请求。
import requests
# 导入 tqdm 模块,实现进度条。
from tqdm import tqdm
# 定义函数 get_response,用于向指定的 URL 发送 HTTP GET 请求,并返回响应。
# 此函数接受两个参数:html_url 表示要访问的 URL,stream 表示是否以流的方式获取响应。
def get_response(html_url, stream=False):
# 定义 HTTP 请求的 headers。
headers = {
"referer": "https://www.bilibili.com/",
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36 Edg/108.0.1462.42"
}
# 发送 HTTP GET 请求,并获取响应。
response = requests.get(html_url, headers=headers, stream=stream)
# 返回响应。
return response
# 定义函数 get_url,用于解析 Bilibili 视频的信息,包括标题和下载地址。
# 此函数接受一个参数:html_url 表示
# 定义函数 get_url,用于解析 Bilibili 视频的信息,包括标题和下载地址。
# 此函数接受一个参数:html_url 表示要解析的 Bilibili 视频的 URL。
def get_url(html_url):
# 发送 HTTP GET 请求,获取 Bilibili 视频的 HTML 页面。
r = requests.get(html_url)
# 使用正则表达式,从 HTML 页面中解析出视频数据。
a = re.search(r"window\.__INITIAL_STATE__=(.*?)};
# 定义函数 get_url,用于解析 Bilibili 视频的信息,包括标题和下载地址。
# 此函数接受一个参数:html_url 表示要解析的 Bilibili 视频的 URL。
def get_url(html_url):
# 发送 HTTP GET 请求,获取 Bilibili 视频的 HTML 页面。
r = requests.get(html_url)
# 使用正则表达式,从 HTML 页面中解析出视频数据。
a = re.search(r"window\.__INITIAL_STATE__=(.*?)};", r.text).group(1)
# 使用 json 模块,将视频数据解析成 Python 字典。
json_data = json.loads(a + "}")
# 从字典中获取视频的详细信息。
video_data = json_data['videoData']
# 从视频信息中获取 av 号。
avid = str(video_data['aid'])
# 设定视频质量。
qn = '112'
# 创建一个空列表,用于存储视频标题和下载地址。
urls = []
# 遍历视频中的每一个部分,并提取标题和下载地址。
for p in video_data['pages']:
# 从视频信息中获取 cid。
cid = str(p['cid'])
# 从视频信息中获取标题。
title = p['part']
# 如果视频只有一个部分,则使用整个视频的标题。
if len(video_data['pages']) == 1:
title = video_data['title']
# 根据 cid、av
# 定义函数 save,用于保存视频。
# 此函数接受三个参数:video_url 表示视频的下载地址,video_size 表示视频的大小,title 表示视频的标题。
def save(video_url, video_size, title):
# 发送 HTTP GET 请求,获取视频的二进制数据。
video_res = get_response(video_url, True)
# 使用 with 语句,创建一个文件,用于保存视频。
# 文件的保存路径为桌面,文件名为标题 + ".mp4"。
with open(r'C:\Users\NHT\Desktop\\'+title + '.mp4', 'wb') as fd:
# 打印信息,表示开始下载视频。
print('开始下载文件:{}, 当前文件大小:{}KB'.format(title, video_size / 1024))
# 使用迭代器,遍历视频的二进制数据。
# 此迭代器的每一次迭代都会返回视频的一段数据。
# 使用 tqdm 模块,显示下载进度条。
for c in tqdm(iterable=video_res.iter_content(), total=video_size, unit='b', desc=None):
# 使用文件对象的 write 方法,将每一段数据写入文件。
fd.write(c)
# 打印信息,表示视频下载完成。
print('{},文件下载成功'.format(title))
# 定义函数 a_single_download,用于下载单个视频。
# 定义函数 concurrent_download,用于并发下载多个视频。
# 此函数接受一个参数:base_infos 表示视频的信息。
# 此信息包括标题和下载地址。
def concurrent_download(base_infos):
# 使用 ThreadPoolExecutor 类,创建一个线程池。
# 最多可以同时运行 10 个线程。
executor = ThreadPoolExecutor(max_workers=10)
# 使用列表推导式,创建一个列表,其中的每一个元素都是一个线程。
# 每一个线程都会调用 a_single_download 函数,用于下载单个视频。
futur_tasks = [executor.submit(a_single_download, info) for info in base_infos]
# 等待所有线程都执行完毕,再继续后面的代码。
wait(futur_tasks, return_when=ALL_COMPLETED)
# 如果此文件被直接运行,则执行以下代码。
if __name__ == '__main__':
# 使用 input 函数,提示用户输入 Bilibili 视频的 URL。
html_url = input('请输入Bilibili视频的URL:')
# 调用 get_url 函数,获取视频的信息。
base_infos = get_url(html_url)
# 调用 concurrent_download 函数,并发下载多个视频。
concurrent_download(base_infos)
```
concurrent_download:启用多线程,批量下载视频。
最后,在 if name == 'main': 块中调用 concurrent_download 函数,传入一个包含多个 Bilibili 视频 URL 的列表,实现批量下载。
整体来说,该代码通过使用多线程技术来提高下载速度,并且使用了正则表达式和 JSON 数据处理,从而能够实现 Bilibili 视频的信息解析和下载。
python 爬虫例子及总结(详细理解注释)
如有帮助,望采纳
https://blog.csdn.net/weixin_45627194/article/details/123487327
import json
import re
from concurrent.futures import ThreadPoolExecutor, wait, ALL_COMPLETED
import requests
from tqdm import tqdm
def get_response(html_url, stream=False):
headers = {
"referer": "https://www.bilibili.com/",
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36 Edg/108.0.1462.42"
}#请求头,跳转和ua
response = requests.get(html_url, headers=headers, stream=stream)#带参数访问
return response
def get_url(html_url):
r = requests.get(html_url) #访问网站
a = re.search(r"window\.__INITIAL_STATE__=(.*?)};", r.text).group(1) #找到__INITIAL_STATE__后面的内容
json_data = json.loads(a + "}") #拼接 } 到a中
video_data = json_data['videoData'] #按字典找到videodata
avid = str(video_data['aid']) #找到字典中到aid 转化为字符串
qn = '112'
urls = []
for p in video_data['pages']: #循环字典video_data中到pages
cid = str(p['cid']) #找到每个pages字典中中到cid,转化字符串
title = p['part'] #找到每个pages字典中中到part
if len(video_data['pages']) == 1: #判断字典中pages长度为1时,取字典中title
title = video_data['title']
urls.append({
'title': title,
"url": "https://api.bilibili.com/x/player/playurl?cid=" + cid + "&avid=" + avid + "&an=" + qn + "&otype=json&fourk=1"
}) #将title和url字典格式传入列表
return urls
def save(video_url, video_size, title):
video_res = get_response(video_url, True) #访问函数,返回结果
with open(r'C:\Users\NHT\Desktop\\'+title + '.mp4', 'wb') as fd: #打开文件,并且存入
print('开始下载文件:{}, 当前文件大小:{}KB'.format(title, video_size / 1024))
for c in tqdm(iterable=video_res.iter_content(), total=video_size, unit='b', desc=None):#tqdm可以看进度条
fd.write(c)
print('{},文件下载成功'.format(title))
def a_single_download(info):
response = get_response(info['url']) #函数传惨
json_data = json.loads(response.text)
video_url = json_data['data']['durl'][0]['url']
video_size = json_data['data']['durl'][0]['size']
save(video_url, video_size, info['title'])
def concurrent_download(base_infos):
executor = ThreadPoolExecutor(max_workers=10)#开线程访问
futur_tasks = [executor.submit(a_single_download, info) for info in base_infos] #循环启动,几个线程
wait(futur_tasks, return_when=ALL_COMPLETED)#阻塞线程