书接上回，b 站除了评论区出人才，弹幕也是 b 站文化富集之地，所以今天分享的是 b 站弹幕爬虫，文末同时附上源代码和 exe 工具链接。
测试了下这份代码/工具大概单个视频最多能爬到 10000 条左右的弹幕。
b 站没啥反爬的，带个 User-Agent 就能请求数据。
代码语言：javascript复制headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3970.5 Safari/537.36',
'Referer': 'https://www.bilibili.com/'
}
和评论时间不同，弹幕时间戳是距离视频开始的秒数，所以需要额外写个解析。
代码语言：javascript复制def timeFormatter(param):
minute = int(param) // 60
second = float(param) - minute * 60
return f'{str(minute).zfill(2)}:{str(second).zfill(5)}'
如果视频标题做文件名的话，也需要根据文件命名规则将视频标题处理之，也可以通过标题判断视频是否公开可见或者被删除。
代码语言：javascript复制def validateTitle(title):
re_str = r"[/\:*?"<>|]" # '/ : * ? " < > |'
new_title = re.sub(re_str, "_", title) # 替换为下划线
return new_title
请求弹幕数据主要注意下 F12 寻找弹幕的 url 地址，同时需要留意，弹幕请求的响应编码需要自适应编码。
代码语言：javascript复制def getHTML(url):
try:
response = requests.get(url=url, headers=headers,
timeout=timeout)
# 自适应编码
response.encoding = response.apparent_encoding
return response.text
# 下句作用等同于上两句
# return response.text.encode(response.encoding).decode('utf-8')
except:
print(f"reqeuset url : {url} error...")
print(traceback.format_exc())
return None
用个 for 循环遍历要爬取的视频的 bv 号，实现一次爬取多个视频的弹幕的功能。
最后构造 dataframe，边爬取边保存。
以 b 站著名百大 up 【木鱼水心】的热门视频为例
标题：《水浒传》原著影视全解读！带你看懂奇书与神剧！（P1高俅发迹） 链接：https://www.bilibili.com/video/BV16F411B7Ek
抓取的结果字段包括时刻、弹幕文本两个字段，如下图所示。
一同抓取了木鱼水心关于四大名著最热的几个视频的弹幕，关于这些结果文件的获取可以查看今天的另外一篇推送。
代码语言：javascript复制# -*- coding: utf-8 -*-
# 作者: inspurer(月小水长)
# 创建时间: 2020/10/30 23:16
# 运行环境 Python3.6
# github https://github.com/inspurer
# qq邮箱 2391527690@qq.com
# 微信公众号 月小水长(ID: inspurer)
# 文件备注信息 b 站弹幕爬虫
import requests
import re
from bs4 import BeautifulSoup
import operator
import traceback
import os
import pandas as pd
from lxml import etree
from time import sleep
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3970.5 Safari/537.36',
'Referer': 'https://www.bilibili.com/'
}
timeout = 5
def getHTML(url):
try:
response = requests.get(url=url, headers=headers,
timeout=timeout)
# 自适应编码
response.encoding = response.apparent_encoding
return response.text
# 下句作用等同于上两句
# return response.text.encode(response.encoding).decode('utf-8')
except:
print(f"reqeuset url : {url} error...")
print(traceback.format_exc())
return None
def parsePage(page):
try:
print("parsing...")
html_ = etree.HTML(page)
meta_title = html_.xpath('//meta[@name="title"]/@content')[0]
if meta_title == '视频去哪了呢？_哔哩哔哩_bilibili':
print(f'视频 404 not found')
return [], '视频 404 not found'
syntax = [':', '=']
flag = 0
keys = re.findall(r'"cid":[d]*', page)
if not keys:
keys = re.findall(r'cid=[d]*', page)
flag = 1
comments, title = {}, None
keys = [keys[1]]
for index, item in enumerate(keys):
key = item.split(syntax[flag])[1]
print(f'{index 1}/{len(keys)}: {key}')
comment_url = f'https://comment.bilibili.com/{key}.xml' # 弹幕地址
comment_text = getHTML(comment_url)
bs4 = BeautifulSoup(comment_text, "html.parser")
if not title:
title = BeautifulSoup(page, "html.parser").find('h1').get_text().strip()
for comment in bs4.find_all('d'):
time = float(comment.attrs['p'].split(',')[0])
time = timeFormatter(time)
comments[time] = comment.string
sorted_comments = sorted(comments.items(), key=operator.itemgetter(0)) # 排序
comments = dict(sorted_comments)
print("parse finish")
return comments, title
except:
print("parse error")
print(traceback.format_exc())
def validateTitle(title):
re_str = r"[/\:*?"<>|]" # '/ : * ? " < > |'
new_title = re.sub(re_str, "_", title) # 替换为下划线
return new_title
def timeFormatter(param):
minute = int(param) // 60
second = float(param) - minute * 60
return f'{str(minute).zfill(2)}:{str(second).zfill(5)}'
def main():
bvs = ['BV1mL411z7Kf', 'BV1CC4y1a7ee', 'BV1hx411e7KP', 'BV16F411B7Ek']
for bv in bvs:
url = f"https://www.bilibili.com/video/{bv}"
save_folder = "BarRage"
if not os.path.exists(save_folder):
os.mkdir(save_folder)
comments, title = parsePage(getHTML(url))
if len(comments) == 0:
continue
title = validateTitle(title)
df = pd.DataFrame({'时刻': list(comments.keys()), '弹幕文本': list(comments.values())})
df.drop_duplicates(subset=['时刻', '弹幕文本'], keep='first', inplace=True)
df.to_csv(f"{save_folder}/{title}.csv", index=False, encoding='utf-8-sig')
print(f'已经保存 {df.shape[0]} 条弹幕到 {save_folder}/{title}.csvnn')
sleep(10)
if __name__ == '__main__':
main()
