Bilibili数据分析 B站爬虫|Bilibili 视频下载器

需求分析

本项目的目的是实现一个 Bilibili 视频下载器,能够根据给定的 Bilibili 视频 ID(bvid)列表,自动下载对应的视频和音频文件。

功能介绍

  1. 读取包含 Bilibili 视频 ID 的 CSV 文件。
  2. 通过 Bilibili API 获取视频和音频的 URL。
  3. 下载视频和音频文件,并将它们保存在本地文件夹中。
  4. 记录成功下载和下载失败的视频 ID。

代码讲解

导入必要的库

import re
import os
import requests
import json
import time
from lxml import etree
from tqdm import tqdm
import pandas as pd
import random

初始化变量和请求头

fold_path = '/Volumes/SSD/Data'
ERROR_CID_PATH = "/Volumes/SSD/Data/Video/error_bvids.csv"
SUCCESS_CID_PATH = "/Volumes/SSD/Data/Video/successful_bvids.csv"

user_agents = [
    # ...
]

headers = {
    # ...
}

bilibili_url = 'https://www.bilibili.com/video/'
videolist_url = 'https://api.bilibili.com/x/web-interface/web/channel/featured/list?'

这部分代码初始化了一些全局变量,如用于存储视频数据的文件夹路径、错误和成功的视频 ID 文件路径。同时,我们定义了请求头和 Bilibili 的 URL。

读取 CSV 文件

video_list = pd.read_csv('/Volumes/SSD/Data/getVideoinfo/getVideoinfo_byhot_1.csv')
bvid_list = video_list['bvid'].values.tolist()
cid_list = video_list['cid'].values.tolist()

这部分代码读取了包含 Bilibili 视频 ID 的 CSV 文件,并将其中的 bvid 和 cid 分别保存到两个列表中。

下载视频和音频

在以下代码中,我们遍历 bvid_list 中的每个 bvid,获取相应视频页面的 HTML 内容,并从中提取视频和音频的 URL。然后,我们将视频和音频文件分别保存到本地文件夹中。

for bvid in tqdm(bvid_list):
    # ...
    page_url = bilibili_url + bvid
    # ...
    video_content = json.loads(temp)
    # ...
    print(video_url)
    print(audio_url)
    # ...
    with open(vfilename, 'wb') as f:
        # ...
    with open(afilename, 'wb') as f:
        # ...

记录成功和失败的视频 ID

我们将成功下载的视频 ID 记录到 SUCCESS_CID_PATH 指定的文件中,将下载失败的视频 ID 记录到 ERROR_CID_PATH 指定的文件中。

with open(SUCCESS_CID_PATH, "a") as success_file:
    # ...
with open(ERROR_CID_PATH, "a") as error_file:
    # ...

使用方法

  1. 将包含 Bilibili 视频 ID 的 CSV 文件放到指定的文件夹中。
  2. 运行脚本,等待下载完成。

注意:本脚本仅用于个人学习和研究,请勿用于非法用途,尊重 Bilibili 平台的版权规定。

注意事项

  1. 确保你的 Python 环境中已安装 lxml, requests, pandastqdm 库。
  2. 请根据需要更改脚本中的文件路径。
  3. 如果遇到下载速度慢或者失败的情况,可以尝试更换请求头中的 User-Agent 或 cookie 信息。

代码优化建议

  1. 使用函数和类对代码进行模块化,提高代码的可读性和可维护性。
  2. 增加异常处理,以避免程序因网络或其他原因而意外中断。
  3. 可以尝试使用多线程或多进程,以提高下载速度。

总结:本文档详细介绍了一个 Bilibili 视频下载器的需求分析、功能介绍和代码讲解。通过阅读本文档,你可以了解如何实现一个简单的 Bilibili 视频下载器,并根据需要进行扩展和优化。

完整代码(按需进行参考)

import re
import os
import requests
import json
import time
from lxml import etree
from tqdm import tqdm
import pandas as pd
import random

# 定义文件夹路径和错误信息文件路径
fold_path = '/Volumes/SSD/Data'
ERROR_CID_PATH = "/Volumes/SSD/Data/Video/error_bvids.csv"
SUCCESS_CID_PATH = "/Volumes/SSD/Data/Video/successful_bvids.csv"

# 设置请求头
user_agents = [
    # ...(省略部分 User-Agent,可自行添加)
]

headers = {
    'User-Agent': random.choice(user_agents),
    'Referer': 'https://www.bilibili.com/',
    # 添加其他请求头信息,例如 Cookie
    # ...
}

# 定义 Bilibili 视频页面 URL 和生活区视频列表 API URL
bilibili_url = 'https://www.bilibili.com/video/'
videolist_url = 'https://api.bilibili.com/x/web-interface/web/channel/featured/list?'

# 读取 CSV 文件中的 bvid 和 cid
video_list = pd.read_csv('/Volumes/SSD/Data/getVideoinfo/getVideoinfo_byhot_1.csv')
bvid_list = video_list['bvid'].values.tolist()
cid_list = video_list['cid'].values.tolist()

# 遍历 bvid 列表,下载每个视频
for bvid in tqdm(bvid_list):
    bvid_path = f"/Volumes/SSD/Data/Video/{bvid}"

    # 如果已经下载过这个视频,跳过
    if os.path.exists(bvid_path):
        print(f"File for bvid {bvid} already exists. Skipping...")
        with open(SUCCESS_CID_PATH, "a") as success_file:
            success_file.write(f"{bvid}\n")
        continue

    # 请求视频页面
    page_url = bilibili_url + bvid
    response = requests.get(page_url, headers=headers)
    page_content = response.text
    page_tree = etree.HTML(page_content)

    # 提取视频和音频的 URL
    video_content = page_tree.xpath('/html/head/script[5]/text()')
    pattern = r'\<script\>window\.__playinfo__=(.*?)\</script\>'
    no_video_pattern = r'视频去哪了呢?'

    # 如果无法获取视频,记录错误信息并跳过
    if len(re.findall(no_video_pattern, page_content)) != 0:
        with open(ERROR_CID_PATH, "a") as error_file:
            error_file.write(f"{bvid}\n")
            print(f"Cannot get video for bvid {bvid}. Skipping...")
        continue

    try:
        temp = re.findall(pattern, page_content)[0]
    except:
        with open(ERROR_CID_PATH, "a") as error_file:
            error_file.write(f"{bvid}\n")
            print(f"Cannot get video for bvid {bvid}. Skipping...")
        continue

    video_content = json.loads(temp)

    # 提取 Dash 格式的视频和音频 URL
    if video_content['data'] is not None:
        if 'dash' in video_content['data'].keys():
            for item in video_content['data']['dash']['video']:
                if 'baseUrl' in item.keys():
                    video_url = item['baseUrl']
                    continue
            for item in video_content['data']['dash']['audio']:
                if 'baseUrl' in item.keys():
                    audio_url = item['baseUrl']
                    continue

    print(video_url)
    print(audio_    url)

    # 创建视频文件夹
    videofold = fold_path + '/Video/' + str(bvid)

    if not os.path.exists(videofold):
        print(1)
        os.mkdir(videofold)

    # 处理视频标题,移除特殊字符
    title = bvid
    title = title.replace('/', '')
    title = title.replace(' ', '')
    title = title.replace('|', '')
    title = title.replace(':', '')
    title = title.replace(':', '')

    # 定义视频和音频文件名
    vfilename = videofold + '/' + title + '.mp4'
    afilename = videofold + '/' + title + '.mp3'

    # 更新请求头,添加 Origin 和 Referer
    header = headers
    header['Origin'] = 'https://www.bilibili.com'
    header['Referer'] = page_url

    # 下载视频
    with open(vfilename, 'wb') as f:
        with requests.get(url=video_url, headers=header, stream=True) as r:
            for chunk in r.iter_content(chunk_size=512):
                if chunk:
                    f.write(chunk)

    # 下载音频
    with open(afilename, 'wb') as f:
        f.write(requests.get(url=audio_url, headers=header).content)
        with open(SUCCESS_CID_PATH, "a") as success_file:
            success_file.write(f"{bvid}\n")
        continue

    # 添加下载间隔,减轻对服务器的压力
    # time.sleep(3)

Related Posts