Python爬虫添加详细注释


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 网站上爬取视频并下载的功能。

它的主要流程如下:

  • 首先调用 get_response 函数,传入一个视频的 URL,以获取该视频的详细信息。
  • 接着,调用 get_url 函数,分析获取到的视频信息,并生成一个字典列表,每个字典包含该视频分片的标题和下载地址。
  • 然后,调用 concurrent_download 函数,遍历每个字典,并开启多线程进行视频下载。
  • 在下载过程中,调用 a_single_download 函数,通过传入字典中的下载地址,再次获取该视频的详细信息,并调用 save 函数将视频保存到本地。

在这段代码中,需要说明的一些点包括:

  • get_response 函数中的 headers 参数是为了伪装浏览器,避免被 bilibili 网站反爬虫检测到。
  • get_response 函数中的 stream 参数是为了避免在下载大型视频时,将视频整个读入内存,造成内存溢出。
  • concurrent_download 函数中使用了线程池,可以同时开启多个线程进行视频下载,提高下载效率。

下面是对代码中每个函数进行详细注释:
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)#阻塞线程