赞
踩
原文链接:https://blog.xieqiaokang.com/posts/36033.html
Github:https://github.com/xieqk/Bilibili_Spider_by_UserID
Gitee:https://gitee.com/xieqk/Bilibili_Spider_by_UserID
这里使用 conda 安装,也可使用 pip
conda install selenium bs4
selenium是一个操作浏览器的 Python 库,需要安装相应的浏览器驱动,如 firefox:
conda install gtk3 firefox -c conda-forge
此外还需要 geckodriver ,可前往 github 下载,并放置于 /usr/local/bin/:
- 也可以放置在自定义路径下(但须为环境变量能够找到的地方),如非管理员用户可放置于自己 home 目录下的
~/bin目录下,并将该路径添加进环境变量:export PATH=~/bin${PATH:+:${PATH}}
- 1
如果需要永久将
~/bin路径添加进环境变量,则将上述语句添加进~/.bashrc文件末尾即可(重启命令行生效,或手动输入source ~/.bashrc在当前命令行激活)。
- Windows 需下载对应 windows 版本并放置于环境变量能够找到的地方,或手动将
geckodriver所在路径加入PATH中,并重启。
见上一节环境准备部分,安装对应依赖环境。
# 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
如下图所示,进入该用户主页,地址栏后面红框中的数字即为该用户的 uid。

进入代码目录中,直接执行 main.py,传入 uid 参数即可:
python main.py --uid 362548791
爬取结果将保存于当前目录下的 json 目录,以 json 格式保存,为一个列表,内容如下:
[
{
"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" // 当前日期
},
...
]
json。False(B站用户视频页一页一般为30个视频)。2(秒)。网络状况不佳时等待时间过短可能会导致爬取的数据不完全。False。当加入 --detailed 参数后每个 url 的爬取结果为:
[ { "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.py 的 Bilibili_Spider()。
options = webdriver.FirefoxOptions()
options.add_argument('--headless')
self.browser = webdriver.Firefox(options=options)
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")

page_number = html.find('span', attrs={'class':'be-pager-total'}).text
page_number = int(page_number.split(' ')[1])

user_name = html.find('span', id = 'h-name').text
获取该页视频列表,并遍历

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、标题、日期等

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)

# 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 字段即可判断是否为播放列表。

# 接上一段代码:“进入视频页获取视频的详细信息”
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时间过短可能导致爬取失败。
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)
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)))
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) # 添加中文支持
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。