Python3，多线程爬完B站UP主的视频弹幕及评论，我飘了~ ~ ~

2023-11-09

利用线程爬取B站Up主弹幕及评论

1、爬取视频弹幕信息
2、爬取视频评论信息
3、整合代码，线程提速~

1、爬取视频弹幕信息

爬取视频的弹幕信息，同样要借助于我们的接口工具，这里我们使用Charles，获取到的弹幕的url地址

弹幕url

https://api.bilibili.com/x/v1/dm/list.so?oid={cid}

视频url

https://api.bilibili.com/x/player/pagelist?aid={aid}&jsonp=jsonp

至于怎么获取，我想大佬们都应该会的~ ~ 这里就不演示了，
接下来，老规矩，直接上代码：

get_video_barrage.py

# -*- coding: utf-8 -*-
"""
@ auth : carl_DJ
@ time : 2020-8-24
"""

def get_video_barrage(aid,uid,d):
	#获取视频弹幕信息
	cid = d['data']
	#弹幕地址，即运用到Charles抓取的接口地址
	barrage_url = 'https://api.bilibili.com/x/v1/dm/list.so?oid={cid}'
	#设置超时等待设置3秒
	r = get_http_session().get(barrage_url,timeout=3)
	#弹幕保存路径，及格式为 xml, 
	#uid类型要是 str的，不能是int类型，否则报错
	uid_dir_path = os.path.join(dir_path,str(uid))
	if not os.path.exists(uid_dir_path):
		os.makedirs(uid_dir_path)
		#设置弹幕信息保存路径
	barrage_path = os.path.join(uid_dir_path,f'barrage_{aid}.xml')
	#设置字符类型
	r.encoding = 'utf-8'
	content = t.text
	#保存弹幕信息
	save_file(barrage_path,content)
	print(f'video id:{aid} barrage save success !')

def get_up_video_info(name,id,filepath):
	#获取UP主的视频信息
	#读取文件信息
	res = read_json(filepath)
	#读取html页面中的vlist信息
	vlist = res['data']['list']['vlist']
	#循环读取vlist里面的内容
	for v in vlist:
		aid = v['aid']
		url = 'https://api.bilibili.com/x/player/pagelist?aid={aid}&jsonp=jsonp'
		player = get_http_session().get(url,timeout=3)
		#转换成json格式
		player = player.json()
		data = player['data']
		#如果没有数据，则直接返回
		if not data:
			return
		#循环读取data的内容
		for d in data:
			try:
				#这里直接调用上面写好的 get_video_barrage方法
				get_video_barrage(aid,uid,d)
			except Exception as e:
				#如果出错，则输出到get_up_video_info.log这个日志文件中
				log(traceback.format_exc(), 'error', 'get_up_video_info.log')
				error_str = f'name: [{name}], uid: [{uid}]'
				log(error_str,'error','get_up_video_info.log')

这里说明一下：
我直接写的是获取视频弹幕的方法！

2、爬取视频评论信息

关乎于获取评论的信息代码，和获取视频弹幕的方法一样
我就直接上代码，不做过多的啰嗦！

get_video_comment.py

# -*- coding: utf-8 -*-
"""
@ auth : carl_DJ
@ time : 2020-8-24
"""

def get_video_comment(uid,aid):
	#获取视频评论的信息
	comment_url = f'视频评论url'
	r = get_http_session().get(comment_url,timeout=3)
	#如果返回的状态码是200，则执行以下动作
	if r.status_code == 200:
		#同样这里uid要设置成str类型
		uid_dir_path = os.path.join(dir_path,str(uid))
		#如果没有地址，则创建
		if not os.path.exists(uid_dir_path):
			os.makedirs(uid_dir_path)
		common_path = os.path.join(uid_dir_path,f'comment_{aid}.json')
		#格式为json，缩进4个，内容不进行编码
		content = json.dumps(r.json(),indent=4,ensure_ascii= False)
		#保存
		save_file(common_path,content)
		#打印结果
		print(f'video id:{aid} comment save success')

代码到这里，就完成，没有什么难度，
那么接下来，我们就要使用线程，进行提速了

3、整合代码，线程提速~

同样，在爬取B站UP主的信息是，如果对单一视频进行爬取，还好一些
如果同时对100+UP主进行视频弹幕，评论进行爬取，那爬取量可海了去了~
而且还有可能在爬取一小半的时候，就被B站给反爬了，这是不是就尴尬了~
为了解决这种问题，我们必须要提速…
把省下来的时间，看看漫画，看看小视频，不香吗！！

说正就整，老规矩，上代码：

# -*- coding: utf-8 -*-
"""
@ auth : carl_DJ
@ time : 2020-8-24
"""


import os
import json
import requests
import traceback
import time

from threading import Thread
from concurrent.futures import ThreadPoolExecutor
from queue import Queue

#创建线程池： 10 个
executor = ThreadPoolExecutor(10)
queue = Queue()

#存放数据的根目录
dir_path = os.path.join('Bili_UP')

def get_http_session(pool_connections = 2,pool_maxsize=10,max_retries = 3)
    """
    http的连接池
    :param pool_connection: 连接池数
    :param poll_maxsize: 最大连接池数
    :param max_retries: 最大重试数
    :return:
    """
	session = requests.session()
	#适配器,传入3个参数
	adapter = requests.adapters.HTTPAdapter(pool_connections=pool_connections, pool_maxsize=pool_maxsize, max_retries=max_retries)
    session.mount('http://',adapter)
	session.mount('https://',adapter)
	return session

def save_file(filepath,content):
	'''
	由于这里保存的代码在《[Pyhotn3，爬取B站up主的信息！》已经写过过，
	这就就不在重复定义，不太熟练的，可以点击传送。
	'''
	# 保存文件
	pass   
def log(content,level,filepath):
	#创建错误日志级别
	pass

def make_dir(name):
	#创建dir
    pass
    
def read_json(filepath):
    """
    :param filepath: 读取文件
    :return:
    """
    #这里如果不写encoding='utf-8'，转义会有问题
    with open(filepath,'r',encoding='UTF-8') as f:
        res = f.read()
        #将读取的文件转换成json格式，
        return json.loads(res)

def get_up_base_info(name, uid):
	#获取UP主的基本信息
	try:
		url = f'UP主的url地址'
		#设置超时等待设置3秒
		r = get_http_session().get(url, timeout=100)
		#如果返回的状态码是200，则执行以下动作
		if r.status.code ==200:
			up_dir = make_dir(name)
			filepath = os.path.join(up_dir,f'{uid}_base_info.json')
			content = json.dumps(r.json(), indent=4, ensure_ascii=False)
			#保存
			save_file(filepath, content)
			#打印信息
			print(f'{name} up主信息保存成功')
			#将信息推到队列中
			global queue
			queue.put((name,uid,filepath))
		else:
			fail_str = f'name: [{name}], uid: [{uid}], url: [{url}]'
			log(fail_str,'fail','base_info_fail.log')
	except Exception as e:
		log(traceback.format_exc(),'error','base_info_error.log')
		error_str = f'name: [{name}], uid: [{uid}]'
		log(error_str, 'error', 'base_info_error.log')

def get_video_barrage(aid,uid,d):
	#获取视频弹幕信息
	cid = d['data']
	#弹幕地址，即运用到Charles抓取的接口地址
	barrage_url = 'https://api.bilibili.com/x/v1/dm/list.so?oid={cid}'
	#设置超时等待设置3秒
	r = get_http_session().get(barrage_url,timeout=3)
	#弹幕保存路径，及格式为 xml, 
	#uid类型要是 str的，不能是int类型，否则报错
	uid_dir_path = os.path.join(dir_path,str(uid))
	if not os.path.exists(uid_dir_path):
		os.makedirs(uid_dir_path)
		#设置弹幕信息保存路径
	barrage_path = os.path.join(uid_dir_path,f'barrage_{aid}.xml')
	#设置字符类型
	r.encoding = 'utf-8'
	content = t.text
	#保存弹幕信息
	save_file(barrage_path,content)
	print(f'video id:{aid} barrage save success !')

def get_up_video_info(name,id,filepath):
	#获取UP主的视频信息
	#读取文件信息
	res = read_json(filepath)
	#读取html页面中的vlist信息
	vlist = res['data']['list']['vlist']
	#循环读取vlist里面的内容
	for v in vlist:
		aid = v['aid']
		url = 'https://api.bilibili.com/x/player/pagelist?aid={aid}&jsonp=jsonp'
		player = get_http_session().get(url,timeout=3)
		#转换成json格式
		player = player.json()
		data = player['data']
		#如果没有数据，则直接返回
		if not data:
			return
		#循环读取data的内容
		for d in data:
			try:
				#这里直接调用上面写好的 get_video_barrage方法
				get_video_barrage(aid,uid,d)
			except Exception as e:
				#如果出错，则输出到get_up_video_info.log这个日志文件中
				log(traceback.format_exc(), 'error', 'get_up_video_info.log')
				error_str = f'name: [{name}], uid: [{uid}]'
				log(error_str,'error','get_up_video_info.log')

def get_video_comment(uid,aid):
	#获取视频评论的信息
	comment_url = f'视频评论url'
	r = get_http_session().get(comment_url,timeout=3)
	#如果返回的状态码是200，则执行以下动作
	if r.status_code == 200:
		#同样这里uid要设置成str类型
		uid_dir_path = os.path.join(dir_path,str(uid))
		#如果没有地址，则创建
		if not os.path.exists(uid_dir_path):
			os.makedirs(uid_dir_path)
		common_path = os.path.join(uid_dir_path,f'comment_{aid}.json')
		#格式为json，缩进4个，内容不进行编码
		content = json.dumps(r.json(),indent=4,ensure_ascii= False)
		#保存
		save_file(common_path,content)
		#打印结果
		print(f'video id:{aid} comment save success')

def base_info_task(power_json):
	#设定启动方法,启动获取UP主的信息
	for d in power_json:
        uid = d['uid']
        name = d['name']
        #通过线程池去执行
        executor.submit(get_up_base_info,name,uid)
        
def get_video_task(power_json):
	#设定启动方法,启动获取UP主的视频和弹幕的信息
	#设定最大为10
	with ThreadPoolExecutor(max_workers = 10) as executor:
		#做循环，
		while True:
			global queue:
			name,uid,filepath = queue.get()
			executor.submit(get_up_video_info,name,uid,filepath)
			queue.task_done()
			#设置等待时间，防止被反爬
			time.sleep(2)

def mian():
	#读取up主的基本信息，
	power_json = read_json('UP_base.json')
	#启动线程
	Thread(target = base_info_task,args = (power_json, )).start()
	Thread(target=video_info_task).start()

if __name__ == '__main__':
    main()

注：
1.这里省略了make_dir，log等方法，
→① 是因为在《Pyhotn3，爬取B站up主的信息！》写过了
→②这篇主要是多线程爬取UP主的视频弹幕及评论，写的太多，容易造成混乱；
2.如果只想获取某一UP主的弹幕信息，可以把这里面的代码抽离出来，做单独爬取即可；
3.由于我们爬取的内容，是公开信息，所以，在爬取的过程中，B站并没有设置太多障碍！
4.如果要爬取的内容过多，小鱼还是建议以下几点：
→①设置User-Agent
→②设置 IP代理池
→③找别人给你爬

要是被封号，你就当吃瓜群众好了，O(∩_∩)O~

最后的最后

欢迎更多爬友来交流
同时，也给小鱼提更多更好的建议~~
让我们一起快乐，一起爬！！

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)