Github:https://github.com/xieqk/Bilibili_Spider_by_UserID
Gitee:https://gitee.com/xieqk/Bilibili_Spider_by_UserID

环境准备

  • selenium
  • bs4

安装

这里使用 conda 安装,也可使用 pip

1
conda install selenium bs4

selenium是一个操作浏览器的 Python 库,需要安装相应的浏览器驱动,如 firefox:

1
conda install gtk3 firefox -c conda-forge

此外还需要 geckodriver ,可前往 github 下载,并放置于 /usr/local/bin/

  • 也可以放置在自定义路径下(但须为环境变量能够找到的地方),如非管理员用户可放置于自己 home 目录下的 ~/bin 目录下,并将该路径添加进环境变量:
1
export PATH=~/bin${PATH:+:${PATH}}

如果需要永久将 ~/bin 路径添加进环境变量,则将上述语句添加进 ~/.bashrc 文件末尾即可(重启命令行生效,或手动输入source ~/.bashrc 在当前命令行激活)。

  • Windows 需下载对应 windows 版本并放置于环境变量能够找到的地方,或手动将 geckodriver 所在路径加入 PATH 中,并重启。

快速使用

1. 安装依赖

见上一节环境准备部分,安装对应依赖环境。

2. Clone 代码

1
2
3
4
# Github (国内访问网速不佳者可使用 Gitee)
git clone https://github.com/xieqk/Bilibili_Spider_by_UserID.git
# Gitee
git clone https://gitee.com/xieqk/Bilibili_Spider_by_UserID.git

3. 查看 B 站用户 uid

如下图所示,进入该用户主页,地址栏后面红框中的数字即为该用户的 uid

产看用户uid

4. 爬取用户视频数据

进入代码目录中,直接执行 main.py,传入 uid 参数即可:

1
python main.py --uid 362548791

爬取结果将保存于当前目录下的 json 目录,以 json 格式保存,为一个列表,内容如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
[
{
"user_name": "歪西歪小哥哥", // UP主名字
"bv": "BV1Wa4y1e7yy", // BV号
"url": "https://www.bilibili.com/video/BV1Wa4y1e7yy", // 视频链接
"title": "【新冠肺炎:全球各国+中美各省/州】累计确诊人数 & 累计死亡人数数据可视化:俄罗斯情况不容乐观", // 标题
"play": "3888", // 播放量
"duration": 796, // 总时长
"pub_date": "2020-05-16", // 发布日期
"now": "2020-11-18 15:47:28" // 当前日期
},
...
]

5. 其它参数

  • —save_dir:保存 json 结果的目录,默认为 json
  • —save_by_page:按页保存用户视频信息,默认为 False(B站用户视频页一页一般为30个视频)。
  • —time:爬取时,浏览器获取页面的等待时间,默认为 2(秒)。网络状况不佳时等待时间过短可能会导致爬取的数据不完全。
  • —detailed:进一步爬取每一个链接的详细信息(弹幕数、是否为播放列表、发布日期及时刻),默认为 False

当加入 --detailed 参数后每个 url 的爬取结果为:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
[
{
"user_name": "歪西歪小哥哥",
"bv": "BV1Wa4y1e7yy",
"url": "https://www.bilibili.com/video/BV1Wa4y1e7yy",
"title": "【新冠肺炎:全球各国+中美各省/州】累计确诊人数 & 累计死亡人数数据可视化:俄罗斯情况不容乐观",
"play": "3888",
"duration": 796,
"pub_date": "2020-05-16 02:17:16", // 发布日期精确到时分秒
"now": "2020-11-18 15:47:28",
"danmu": "85",
"type": "playlist", // 链接类型:'video'代表单个视频,'playlist'代表播放列表
"num": 4 // 分P数,如果为'video'则为1,'playlist'则为播放列表的视频集数
},
...
]

详细说明

详见 utils/bilibili_spider.pyBilibili_Spider()

1. 初始化

1
2
3
options = webdriver.FirefoxOptions()
options.add_argument('--headless')
self.browser = webdriver.Firefox(options=options)

2. 获取用户视频页数及用户名

  • 获取用户主页下视频页的第一页:
1
2
3
4
5
self.user_url = 'https://space.bilibili.com/{}'.format(uid)
page_url = self.user_url + '/video?tid=0&page={}&keyword=&order=pubdate'.format(1)
self.browser.get(page_url)
time.sleep(self.t+2*random.random())
html = BeautifulSoup(self.browser.page_source, features="html.parser")
  • 获取视频页数:找到页数所在位置,浏览器打开页面后在对应位置检查即可。

获取用户视频页数

1
2
page_number = html.find('span', attrs={'class':'be-pager-total'}).text
page_number = int(page_number.split(' ')[1])
  • 获取用户名

获取用户名

1
user_name = html.find('span', id = 'h-name').text

3. 获取每一页的视频信息

获取该页视频列表,并遍历

查看该页视频列表

1
2
3
4
5
6
7
8
9
page_url = self.user_url + '/video?tid=0&page={}&keyword=&order=pubdate'.format(idx+1)	# idx 为视频第几页
self.browser.get(page_url)
time.sleep(self.t+2*random.random())
html = BeautifulSoup(self.browser.page_source, features="html.parser")

ul_data = html.find('div', id = 'submit-video-list').find('ul', attrs= {'class': 'clearfix cube-list'})

for li in ul_data.find_all('li'):
# 获取每个视频的信息:url、标题、日期等

4. 获取每个视频信息

获取每个视频的相关信息

1
2
3
4
5
6
7
8
9
10
11
12
13
for li in ul_data.find_all('li'):
# 链接和标题
a = li.find('a', attrs = {'target':'_blank', 'class':'title'})
a_url = 'https:{}'.format(a['href'])
a_title = a.text
# 发布日期及播放数
date_str = li.find('span', attrs = {'class':'time'}).text.strip()
pub_date = self.date_convert(date_str)
now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
play = int(li.find('span', attrs = {'class':'play'}).text.strip())
# 总时长
time_str = li.find('span', attrs = {'class':'length'}).text
duration = self.time_convert(time_str)

5. 进入视频页获取信息

  • 获取视频详细数据

进入视频页查看视频详细数据:播放量、弹幕数、发布日期

1
2
3
4
5
6
7
8
9
# e.g. url = 'https://www.bilibili.com/video/BV1Wa4y1e7yy'
self.browser.get(url)
time.sleep(self.t+2*random.random())
html = BeautifulSoup(self.browser.page_source, features="html.parser")

video_data = html.find('div', id = 'viewbox_report').find_all('span')
play = int(video_data[1]['title'][4:])
danmu = int(video_data[2]['title'][7:])
date = video_data[3].text
  • 判断是否为播放列表

检查是否有 multi_page 字段即可判断是否为播放列表。

查看是否为播放列表

1
2
3
4
5
6
7
8
9
10
# 接上一段代码:“进入视频页获取视频的详细信息”
multi_page = html.find('div', id = 'multi_page')
if multi_page is not None:
url_type = 'playlist'
pages = multi_page.find('span', attrs= {'class': 'cur-page'}).text
page_total = int(pages.split('/')[-1])
else:
url_type = 'video'
page_total = 1
# 也可继续获取播放列表其它信息,如分P的标题,但如果爬取时sleep时间过短可能导致爬取失败。

完整代码

main.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
import os
import os.path as osp
import argparse

from utils.bilibili_spider import Bilibili_Spider


def main(args):
bilibili_spider = Bilibili_Spider(args.uid, args.save_dir, args.save_by_page, args.time)
bilibili_spider.get()
if args.detailed:
bilibili_spider.get_detail()


if __name__ == '__main__':
parser = argparse.ArgumentParser()
parser.add_argument('--uid', type=str, default='362548791')
parser.add_argument('--save_dir', type=str, default='json')
parser.add_argument('--save_by_page', action='store_true', default=False)
parser.add_argument('--time', type=int, default=2, help='waiting time for browser.get(url) by seconds')
parser.add_argument('--detailed', action='store_true', default=False)
args = parser.parse_args()
print(args)

main(args)

utils/bilibili_spider.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
import re
import os
import os.path as osp
import sys
import json
import time
import argparse
import datetime
from selenium import webdriver
from bs4 import BeautifulSoup
from urllib import parse as url_parse
import random

from .tools import mkdir_if_missing, write_json, read_json


class Bilibili_Spider():

def __init__(self, uid, save_dir_json='json', save_by_page=False, t=2):
self.t = t
self.uid = uid
self.user_url = 'https://space.bilibili.com/{}'.format(uid)
self.save_dir_json = save_dir_json
self.save_by_page = save_by_page
options = webdriver.FirefoxOptions()
options.add_argument('--headless')
self.browser = webdriver.Firefox(options=options)
print('spider init done.')

def close(self):
# 关闭浏览器驱动
self.browser.quit()

def time_convert(self, time_str):
time_item = time_str.split(':')
assert len(time_item) == 2, 'time format error: {}, x:x expected!'.format(time_str)
seconds = int(time_item[0])*60 + int(time_item[1])
return seconds

def date_convert(self, date_str):
date_item = date_str.split('-')
assert len(date_item) == 2 or len(date_item) == 3, 'date format error: {}, x-x or x-x-x expected!'.format(date_str)
if len(date_item) == 2:
year = datetime.datetime.now().strftime('%Y')
date_str = '{}-{:>02d}-{:>02d}'.format(year, int(date_item[0]), int(date_item[1]))
else:
date_str = '{}-{:>02d}-{:>02d}'.format(date_item[0], int(date_item[1]), int(date_item[2]))
return date_str

def get_page_num(self):
page_url = self.user_url + '/video?tid=0&page={}&keyword=&order=pubdate'.format(1)
self.browser.get(page_url)
time.sleep(self.t+2*random.random())
html = BeautifulSoup(self.browser.page_source, features="html.parser")

page_number = html.find('span', attrs={'class':'be-pager-total'}).text
user_name = html.find('span', id = 'h-name').text

return int(page_number.split(' ')[1]), user_name

def get_videos_by_page(self, idx):
# 获取第 page_idx 页的视频信息
urls_page, titles_page, plays_page, dates_page, durations_page = [], [], [], [], []
page_url = self.user_url + '/video?tid=0&page={}&keyword=&order=pubdate'.format(idx+1)
self.browser.get(page_url)
time.sleep(self.t+2*random.random())
html = BeautifulSoup(self.browser.page_source, features="html.parser")

ul_data = html.find('div', id = 'submit-video-list').find('ul', attrs= {'class': 'clearfix cube-list'})

for li in ul_data.find_all('li'):
# url & title
a = li.find('a', attrs = {'target':'_blank', 'class':'title'})
a_url = 'https:{}'.format(a['href'])
a_title = a.text
# pub_date & play
date_str = li.find('span', attrs = {'class':'time'}).text.strip()
pub_date = self.date_convert(date_str)
now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
play = int(li.find('span', attrs = {'class':'play'}).text.strip())
# duration
time_str = li.find('span', attrs = {'class':'length'}).text
duration = self.time_convert(time_str)
# append
urls_page.append(a_url)
titles_page.append(a_title)
dates_page.append((pub_date, now))
plays_page.append(play)
durations_page.append(duration)

return urls_page, titles_page, plays_page, dates_page, durations_page

def save(self, json_path, bvs, urls, titles, plays, durations, dates):
data_list = []
for i in range(len(urls)):
result = {}
result['user_name'] = self.user_name
result['bv'] = bvs[i]
result['url'] = urls[i]
result['title'] = titles[i]
result['play'] = plays[i]
result['duration'] = durations[i]
result['pub_date'] = dates[i][0]
result['now'] = dates[i][1]
data_list.append(result)

print('write json to {}'.format(json_path))
dir_name = osp.dirname(json_path)
mkdir_if_missing(dir_name)
write_json(data_list, json_path)
print('dump json file done. total {} urls. \n'.format(len(urls)))

def get(self):
# 获取该 up 主的所有基础视频信息
print('Start ... \n')
self.page_num, self.user_name = self.get_page_num()
while self.page_num == 0:
print('Failed to get user page num, poor network condition, retrying ... ')
self.page_num, self.user_name = self.get_page_num()
print('Pages Num {}, User Name: {}'.format(self.page_num, self.user_name))

bvs = []
urls = []
titles = []
plays = []
dates = []
durations = [] # by seconds

for idx in range(self.page_num):
print('>>>> page {}/{}'.format(idx+1, self.page_num))
urls_page, titles_page, plays_page, dates_page, durations_page = self.get_videos_by_page(idx)
while len(urls_page) == 0:
print('failed, try again page {}/{}'.format(idx+1, self.page_num))
urls_page, titles_page, plays_page, dates_page, durations_page = self.get_videos_by_page(idx)
bvs_page = [x.split('/')[-1] for x in urls_page]
assert len(urls_page) == len(titles_page), '{} != {}'.format(len(urls_page), len(titles_page))
assert len(urls_page) == len(plays_page), '{} != {}'.format(len(urls_page), len(titles_page))
assert len(urls_page) == len(dates_page), '{} != {}'.format(len(urls_page), len(dates_page))
assert len(urls_page) == len(durations_page), '{} != {}'.format(len(urls_page), len(durations_page))
print('result:')
print('{}_{}: '.format(self.user_name, self.uid), bvs_page, ', {} in total'.format(len(urls_page)))
sys.stdout.flush()
bvs.extend(bvs_page)
urls.extend(urls_page)
titles.extend(titles_page)
plays.extend(plays_page)
dates.extend(dates_page)
durations.extend(durations_page)
if self.save_by_page:
json_path_page = osp.join(self.save_dir_json, '{}_{}'.format(self.user_name, self.uid), 'primary', 'page_{}.json'.format(idx+1))
self.save(json_path_page, bvs_page, urls_page, titles_page, plays_page, durations_page, dates_page)

json_path = osp.join(self.save_dir_json, '{}_{}'.format(self.user_name, self.uid), 'primary', 'full.json')
self.save(json_path, bvs, urls, titles, plays, durations, dates)

def get_url(self, url):
self.browser.get(url)
time.sleep(self.t+2*random.random())
html = BeautifulSoup(self.browser.page_source, features="html.parser")

video_data = html.find('div', id = 'viewbox_report').find_all('span')
play = int(video_data[1]['title'][4:])
danmu = int(video_data[2]['title'][7:])
date = video_data[3].text

multi_page = html.find('div', id = 'multi_page')
if multi_page is not None:
url_type = 'playlist'
pages = multi_page.find('span', attrs= {'class': 'cur-page'}).text
page_total = int(pages.split('/')[-1])
else:
url_type = 'video'
page_total = 1

return play, danmu, date, url_type, page_total

def get_detail(self):
print('Start to get detailed information for each url.')
if self.save_by_page:
data = []
for idx in range(self.page_num):
json_path = osp.join(self.save_dir_json, '{}_{}'.format(self.user_name, self.uid), 'primary', 'page_{}.json'.format(idx+1))
data_page = read_json(json_path)
for j, item in enumerate(data_page):
url = item['url']
print('>>>> page {}/{}, No. {}/{}'.format(idx+1, self.page_num, j+1, len(data_page)))
play, danmu, date, url_type, page_total = self.get_url(url)
# print(play, danmu, date, url_type, page_total)
assert page_total > 0, page_total
if page_total == 1:
assert url_type == 'video', (url_type, page_total)
data_page[j]['play'] = play
data_page[j]['danmu'] = danmu
data_page[j]['pub_date'] = date
data_page[j]['type'] = url_type
data_page[j]['num'] = page_total
else:
assert url_type == 'playlist', (url_type, page_total)
data_page[j]['play'] = play
data_page[j]['danmu'] = danmu
data_page[j]['pub_date'] = date
data_page[j]['type'] = url_type
data_page[j]['num'] = page_total

json_path_save = osp.join(self.save_dir_json, '{}_{}'.format(self.user_name, self.uid), 'detailed', 'page_{}.json'.format(idx+1))
print('write json to {}'.format(json_path_save))
write_json(data_page, json_path_save)
print('dump json file done. total {} urls. \n'.format(len(data_page)))
data.extend(data_page)

json_path_save = osp.join(self.save_dir_json, '{}_{}'.format(self.user_name, self.uid), 'detailed', 'full.json')
print('write json to {}'.format(json_path_save))
write_json(data, json_path_save)
print('dump json file done. total {} urls. \n'.format(len(data)))
else:
json_path = osp.join(self.save_dir_json, '{}_{}'.format(self.user_name, self.uid), 'primary', 'full.json')
data = read_json(json_path)
for j, item in enumerate(data):
url = item['url']
print('>>>> No. {}/{}'.format(j+1, len(data)))
play, danmu, date, url_type, page_total = self.get_url(url)
assert page_total > 0, page_total
if page_total == 1:
assert url_type == 'video', (url_type, page_total)
data[j]['play'] = play
data[j]['danmu'] = danmu
data[j]['pub_date'] = date
data[j]['type'] = url_type
data[j]['num'] = page_total
else:
assert url_type == 'playlist', (url_type, page_total)
data[j]['play'] = play
data[j]['danmu'] = danmu
data[j]['pub_date'] = date
data[j]['type'] = url_type
data[j]['num'] = page_total

json_path_save = osp.join(self.save_dir_json, '{}_{}'.format(self.user_name, self.uid), 'detailed', 'full.json')
print('write json to {}'.format(json_path_save))
write_json(data, json_path_save)
print('dump json file done. total {} urls. \n'.format(len(data)))

utils/tools.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
import sys
import os
import os.path as osp
import time
import errno
import json
import warnings


def mkdir_if_missing(dirname):
"""Creates dirname if it is missing."""
if not osp.exists(dirname):
try:
os.makedirs(dirname)
except OSError as e:
if e.errno != errno.EEXIST:
raise


def check_isfile(fpath):
"""Checks if the given path is a file."""
isfile = osp.isfile(fpath)
if not isfile:
warnings.warn('No file found at "{}"'.format(fpath))
return isfile


def read_json(fpath):
"""Reads json file from a path."""
with open(fpath, 'r') as f:
obj = json.load(f)
return obj


def write_json(obj, fpath):
"""Writes to a json file."""
mkdir_if_missing(osp.dirname(fpath))
with open(fpath, 'w', encoding='utf-8') as f:
json.dump(obj, f, indent=4, separators=(',', ': '), ensure_ascii=False) # 添加中文支持