295 lines
12 KiB
Python
295 lines
12 KiB
Python
|
import requests
|
|||
|
|
import re
|
|||
|
|
import csv
|
|||
|
|
from bs4 import BeautifulSoup
|
|||
|
|
from time import sleep
|
|||
|
|
from random import uniform
|
|||
|
|
from fake_useragent import UserAgent
|
|||
|
|
import time
|
|||
|
|
|
|||
|
|
def load_targets():
|
|||
|
|
"""从targets.txt加载BV号列表"""
|
|||
|
|
try:
|
|||
|
|
with open('targets.txt', 'r', encoding='utf-8') as f:
|
|||
|
|
return [line.strip() for line in f if line.strip().startswith('BV')]
|
|||
|
|
except FileNotFoundError:
|
|||
|
|
print("未找到targets.txt文件，请创建文件并添加BV号")
|
|||
|
|
return []
|
|||
|
|
|
|||
|
|
class BiliWebCrawler:
|
|||
|
|
def __init__(self, url, cookie=None):
|
|||
|
|
self.url = url
|
|||
|
|
self.bvid = self._extract_bvid()
|
|||
|
|
self.user_agent = UserAgent()
|
|||
|
|
self.headers = {
|
|||
|
|
'User-Agent': self.user_agent.random,
|
|||
|
|
'Referer': 'https://www.bilibili.com/',
|
|||
|
|
}
|
|||
|
|
if cookie:
|
|||
|
|
self.headers['Cookie'] = cookie
|
|||
|
|
self.session = requests.Session()
|
|||
|
|
self.aid = self._get_video_id()
|
|||
|
|
|
|||
|
|
def _extract_bvid(self):
|
|||
|
|
"""从视频链接中提取 BVID"""
|
|||
|
|
match = re.search(r'(BV[A-Za-z0-9]+)', self.url)
|
|||
|
|
if match:
|
|||
|
|
return match.group(1)
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
def _get_video_id(self):
|
|||
|
|
if not self.bvid:
|
|||
|
|
return None
|
|||
|
|
url = f'https://api.bilibili.com/x/web-interface/view?bvid={self.bvid}'
|
|||
|
|
resp = self._safe_request(url)
|
|||
|
|
if resp:
|
|||
|
|
data = resp.json()
|
|||
|
|
return data.get('data', {}).get('aid')
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
def _safe_request(self, url, params=None, retry=3): # 添加 params 参数
|
|||
|
|
"""带重试机制的请求"""
|
|||
|
|
for i in range(retry):
|
|||
|
|
try:
|
|||
|
|
resp = self.session.get(url, headers=self.headers, timeout=10, params=params) # 使用 params 参数
|
|||
|
|
resp.raise_for_status()
|
|||
|
|
return resp
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"请求失败: {str(e)}, 第{i + 1}次重试...")
|
|||
|
|
sleep(uniform(1, 3))
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
def get_video_info(self):
|
|||
|
|
"""获取视频基本信息"""
|
|||
|
|
if not self.bvid:
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
# 获取视频基本信息
|
|||
|
|
url = f'https://api.bilibili.com/x/web-interface/view?bvid={self.bvid}'
|
|||
|
|
resp = self._safe_request(url)
|
|||
|
|
if not resp:
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
data = resp.json()
|
|||
|
|
video_data = data.get('data')
|
|||
|
|
if not video_data:
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
# 获取视频最高分辨率（基于dimension对象）
|
|||
|
|
max_width = 0
|
|||
|
|
max_height = 0
|
|||
|
|
for format_info in video_data.get('formats', []):
|
|||
|
|
dimension = format_info.get('dimension', {})
|
|||
|
|
width = dimension.get('width', 0)
|
|||
|
|
height = dimension.get('height', 0)
|
|||
|
|
rotate = dimension.get('rotate', 0)
|
|||
|
|
|
|||
|
|
# 处理视频旋转（当rotate=1时宽高互换）
|
|||
|
|
if rotate == 1:
|
|||
|
|
width, height = height, width
|
|||
|
|
|
|||
|
|
# 通过像素总量比较分辨率
|
|||
|
|
if (width * height) > (max_width * max_height):
|
|||
|
|
max_width = width
|
|||
|
|
max_height = height
|
|||
|
|
|
|||
|
|
# 将分辨率格式化为 "宽x高" 的字符串
|
|||
|
|
resolution_str = f"{max_width}x{max_height}" if max_width and max_height else "未知"
|
|||
|
|
|
|||
|
|
# 获取视频标签
|
|||
|
|
tag_url = f'https://api.bilibili.com/x/web-interface/view/detail/tag?bvid={self.bvid}'
|
|||
|
|
tag_resp = self._safe_request(tag_url)
|
|||
|
|
tag_data = []
|
|||
|
|
if tag_resp:
|
|||
|
|
tag_json = tag_resp.json()
|
|||
|
|
tag_data = [tag['tag_name'] for tag in tag_json.get('data', [])]
|
|||
|
|
|
|||
|
|
info = {
|
|||
|
|
'title': video_data.get('title', ''),
|
|||
|
|
'up主': video_data.get('owner', {}).get('name', ''),
|
|||
|
|
'播放量': video_data.get('stat', {}).get('view', 0),
|
|||
|
|
'弹幕量': video_data.get('stat', {}).get('danmaku', 0),
|
|||
|
|
'点赞量': video_data.get('stat', {}).get('like', 0),
|
|||
|
|
'投币量': video_data.get('stat', {}).get('coin', 0),
|
|||
|
|
'收藏量': video_data.get('stat', {}).get('favorite', 0),
|
|||
|
|
'发布时间': time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(video_data.get('pubdate', 0))),
|
|||
|
|
'分区': video_data.get('tname', ''),
|
|||
|
|
'标签': tag_data,
|
|||
|
|
'视频最高分辨率': resolution_str,
|
|||
|
|
'视频类型': video_data.get('copyright', 0),
|
|||
|
|
'视频分p数': len(video_data.get('pages', []))
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
return info
|
|||
|
|
|
|||
|
|
def get_danmaku(self):
|
|||
|
|
"""获取弹幕数据"""
|
|||
|
|
if not self.bvid:
|
|||
|
|
print("未找到 BVID，无法获取弹幕数据。")
|
|||
|
|
return []
|
|||
|
|
|
|||
|
|
url = f"https://api.bilibili.com/x/web-interface/view?bvid={self.bvid}"
|
|||
|
|
resp = self._safe_request(url)
|
|||
|
|
if not resp:
|
|||
|
|
return []
|
|||
|
|
|
|||
|
|
data = resp.json()
|
|||
|
|
cid = data.get('data', {}).get('cid')
|
|||
|
|
if not cid:
|
|||
|
|
print("未找到 cid，无法获取弹幕数据。")
|
|||
|
|
return []
|
|||
|
|
|
|||
|
|
danmaku_url = f'https://comment.bilibili.com/{cid}.xml'
|
|||
|
|
|
|||
|
|
resp = self._safe_request(danmaku_url)
|
|||
|
|
if not resp:
|
|||
|
|
return []
|
|||
|
|
|
|||
|
|
danmaku = []
|
|||
|
|
soup = BeautifulSoup(resp.content, 'lxml-xml')
|
|||
|
|
for d in soup.find_all('d'):
|
|||
|
|
attrs = d['p'].split(',')
|
|||
|
|
danmaku.append({
|
|||
|
|
'时间': float(attrs[0]),
|
|||
|
|
'模式': attrs[1],
|
|||
|
|
'颜色': f'#{int(attrs[3]):06X}',
|
|||
|
|
'弹幕内容': d.text
|
|||
|
|
})
|
|||
|
|
return danmaku
|
|||
|
|
|
|||
|
|
def get_comments(self, max_pages=1000):
|
|||
|
|
"""获取热门评论，包含二级评论"""
|
|||
|
|
if not self.aid:
|
|||
|
|
print("未找到视频 ID，无法获取评论数据。")
|
|||
|
|
return []
|
|||
|
|
comments = []
|
|||
|
|
for page in range(1, max_pages + 1):
|
|||
|
|
url = f'https://api.bilibili.com/x/v2/reply?pn={page}&type=1&oid={self.aid}&sort=2'
|
|||
|
|
try:
|
|||
|
|
response = self._safe_request(url)
|
|||
|
|
if response and response.status_code == 200:
|
|||
|
|
response.encoding = 'utf-8'
|
|||
|
|
data = response.json()
|
|||
|
|
if data and data.get('data') and data['data'].get('replies'):
|
|||
|
|
for comment in data['data']['replies']:
|
|||
|
|
comment_info = {
|
|||
|
|
'用户昵称': comment['member']['uname'],
|
|||
|
|
'评论内容': comment['content']['message'],
|
|||
|
|
'被回复用户': '',
|
|||
|
|
'评论层级': '一级评论',
|
|||
|
|
'性别': comment['member']['sex'],
|
|||
|
|
'用户当前等级': comment['member']['level_info']['current_level'],
|
|||
|
|
'点赞数量': comment['like'],
|
|||
|
|
'回复时间': time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(comment['ctime']))
|
|||
|
|
}
|
|||
|
|
comments.append(comment_info)
|
|||
|
|
replies = self.fetch_comment_replies(comment['rpid'], comment['member']['uname'])
|
|||
|
|
comments.extend(replies)
|
|||
|
|
else:
|
|||
|
|
# 当当前页没有评论时，跳出循环
|
|||
|
|
break
|
|||
|
|
except requests.RequestException as e:
|
|||
|
|
print(f"请求出错: {e}")
|
|||
|
|
break
|
|||
|
|
# 适当调整请求间隔
|
|||
|
|
sleep(1)
|
|||
|
|
return comments
|
|||
|
|
|
|||
|
|
def fetch_comment_replies(self, comment_id, parent_user_name, max_pages=1000):
|
|||
|
|
replies = []
|
|||
|
|
for page in range(1, max_pages + 1):
|
|||
|
|
url = f'https://api.bilibili.com/x/v2/reply/reply?oid={self.aid}&type=1&root={comment_id}&ps=10&pn={page}'
|
|||
|
|
try:
|
|||
|
|
response = self._safe_request(url)
|
|||
|
|
if response and response.status_code == 200:
|
|||
|
|
response.encoding = 'utf-8'
|
|||
|
|
data = response.json()
|
|||
|
|
if data and data.get('data') and data['data'].get('replies'):
|
|||
|
|
for reply in data['data']['replies']:
|
|||
|
|
reply_info = {
|
|||
|
|
'用户昵称': reply['member']['uname'],
|
|||
|
|
'评论内容': reply['content']['message'],
|
|||
|
|
'被回复用户': parent_user_name,
|
|||
|
|
'评论层级': '二级评论',
|
|||
|
|
'性别': reply['member']['sex'],
|
|||
|
|
'用户当前等级': reply['member']['level_info']['current_level'],
|
|||
|
|
'点赞数量': reply['like'],
|
|||
|
|
'回复时间': time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(reply['ctime']))
|
|||
|
|
}
|
|||
|
|
replies.append(reply_info)
|
|||
|
|
else:
|
|||
|
|
# 当当前页没有二级评论时，跳出循环
|
|||
|
|
break
|
|||
|
|
except requests.RequestException as e:
|
|||
|
|
print(f"请求二级评论出错: {e}")
|
|||
|
|
break
|
|||
|
|
# 适当调整请求间隔
|
|||
|
|
sleep(1)
|
|||
|
|
return replies
|
|||
|
|
|
|||
|
|
def _parse_count(self, text):
|
|||
|
|
"""统一处理数量文本"""
|
|||
|
|
if '万' in text:
|
|||
|
|
return int(float(text.replace('万', '')) * 10000)
|
|||
|
|
return int(text)
|
|||
|
|
|
|||
|
|
def save_to_csv(self, data, filename, mode='w'):
|
|||
|
|
"""保存数据到CSV"""
|
|||
|
|
if not data:
|
|||
|
|
return
|
|||
|
|
keys = data[0].keys()
|
|||
|
|
with open(filename, mode, newline='', encoding='utf-8-sig') as f:
|
|||
|
|
writer = csv.DictWriter(f, fieldnames=keys)
|
|||
|
|
if f.tell() == 0: # 新文件写入表头
|
|||
|
|
writer.writeheader()
|
|||
|
|
writer.writerows(data)
|
|||
|
|
|
|||
|
|
def run(self):
|
|||
|
|
"""执行完整流程"""
|
|||
|
|
print("正在获取视频基本信息...")
|
|||
|
|
video_info = self.get_video_info()
|
|||
|
|
if video_info:
|
|||
|
|
import os
|
|||
|
|
partition = video_info.get('分区', '其他')
|
|||
|
|
base_dir = os.path.join('data', partition)
|
|||
|
|
video_dir = os.path.join(base_dir, self.bvid)
|
|||
|
|
|
|||
|
|
os.makedirs(base_dir, exist_ok=True)
|
|||
|
|
os.makedirs(video_dir, exist_ok=True)
|
|||
|
|
|
|||
|
|
# 保存视频信息
|
|||
|
|
info_csv_path = os.path.join(base_dir, 'info.csv')
|
|||
|
|
self.save_to_csv([video_info], info_csv_path, mode='a')
|
|||
|
|
|
|||
|
|
play_count = video_info.get('播放量', 0)
|
|||
|
|
video_info_filename = os.path.join(video_dir, f'{self.bvid}_{play_count}_info.csv')
|
|||
|
|
self.save_to_csv([video_info], video_info_filename)
|
|||
|
|
|
|||
|
|
# 新增弹幕抓取
|
|||
|
|
print("正在抓取弹幕数据...")
|
|||
|
|
danmaku = self.get_danmaku()
|
|||
|
|
danmaku_filename = os.path.join(video_dir, f'{self.bvid}_{len(danmaku)}_danmaku.csv')
|
|||
|
|
self.save_to_csv(danmaku, danmaku_filename)
|
|||
|
|
|
|||
|
|
# 新增评论抓取
|
|||
|
|
print("正在抓取评论数据...")
|
|||
|
|
comments = self.get_comments()
|
|||
|
|
comments_filename = os.path.join(video_dir, f'{self.bvid}_{len(comments)}_comments.csv')
|
|||
|
|
self.save_to_csv(comments, comments_filename)
|
|||
|
|
|
|||
|
|
print(f"抓取完成！结果已保存到 {video_dir}/")
|
|||
|
|
else:
|
|||
|
|
print("未获取到视频信息，无法进行抓取。")
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
# 批量处理targets.txt中的BV号
|
|||
|
|
targets = load_targets()
|
|||
|
|
if not targets:
|
|||
|
|
print("未找到有效的BV号，程序退出")
|
|||
|
|
exit()
|
|||
|
|
|
|||
|
|
for bvid in targets:
|
|||
|
|
print(f"\n{'=' * 30} 开始处理 {bvid} {'=' * 30}")
|
|||
|
|
crawler = BiliWebCrawler(f"https://www.bilibili.com/video/{bvid}")
|
|||
|
|
crawler.run()
|