python爬虫获取B站视频评论
代码使用deepseek生成点击查看代码import requestsimport timeimport reimport csvfrom datetime import datetimeimport osdef get_video_info(bvid): """获取视频信息（包含oid）""" url = f"https://api.bilibili.com/x/web-interface/view?bvid={bvid}" headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36", "Referer": "https://www.bilibili.com/" } try: response = requests.get(url, headers=headers, timeout=10) response.raise_for_status() data = response.json() if data.get('code') == 0: return data['data']['aid'], data['data']['title'] raise Exception(f"API错误: {data.get('message')}") except Exception as e: print(f"获取视频信息失败: {str(e)}") return None, Nonedef get_comments(oid, page=1): """获取单页评论 - 使用新版API""" url = "https://api.bilibili.com/x/v2/reply/wbi/main" params = { "next": page, "type": 1, "oid": oid, "mode": 3,# 排序模式: 3-最新, 2-最热 "plat": 1, "web_location": 1315875, "wts": int(time.time()) } headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36", "Referer": f"https://www.bilibili.com/video/BV1xx411c7BF", "Origin": "https://www.bilibili.com" } try: response = requests.get(url, params=params, headers=headers, timeout=15) response.raise_for_status() return response.json() except Exception as e: print(f"获取评论失败(第{page}页): {str(e)}") return Nonedef safe_get(data, keys, default=None): """安全获取嵌套字典值""" for key in keys: if isinstance(data, dict) and key in data: data = data else: return default return datadef parse_comments(comment_data): """解析评论数据 - 增强健壮性""" comments = [] try: # 处理一级评论 replies = safe_get(comment_data, ['data', 'replies']) or [] for reply in replies: try: # 使用安全获取方法处理可能缺失的字段 user_name = safe_get(reply, ['member', 'uname'], '未知用户') content = safe_get(reply, ['content', 'message'], '') ctime = safe_get(reply, ['ctime'], int(time.time())) like_count = safe_get(reply, ['like'], 0) comments.append({ "user": user_name, "content": content, "time": datetime.fromtimestamp(ctime).strftime("%Y-%m-%d %H:%M:%S"), "like": like_count, "level": 1 }) # 处理二级评论（楼中楼） sub_replies = safe_get(reply, ['replies']) or [] for sub_reply in sub_replies: try: sub_user = safe_get(sub_reply, ['member', 'uname'], '未知用户') sub_content = safe_get(sub_reply, ['content', 'message'], '') sub_ctime = safe_get(sub_reply, ['ctime'], int(time.time())) sub_like = safe_get(sub_reply, ['like'], 0) comments.append({ "user": sub_user, "content": sub_content, "time": datetime.fromtimestamp(sub_ctime).strftime("%Y-%m-%d %H:%M:%S"), "like": sub_like, "level": 2 }) except Exception as e: print(f"解析二级评论时出错: {str(e)}") continue except Exception as e: print(f"解析一级评论时出错: {str(e)}") continue # 处理置顶评论 - 使用更安全的方法 top_data = safe_get(comment_data, ['data', 'top', 'reply']) if top_data: try: top_user = safe_get(top_data, ['member', 'uname'], '未知用户') top_content = safe_get(top_data, ['content', 'message'], '') top_ctime = safe_get(top_data, ['ctime'], int(time.time())) top_like = safe_get(top_data, ['like'], 0) comments.insert(0, { "user": top_user, "content": "[置顶] " + top_content, "time": datetime.fromtimestamp(top_ctime).strftime("%Y-%m-%d %H:%M:%S"), "like": top_like, "level": 0 }) except Exception as e: print(f"解析置顶评论时出错: {str(e)}") return comments except Exception as e: print(f"解析评论数据时出错: {str(e)}") return commentsdef save_comments(comments, filename="bilibili_comments.csv"): """保存评论到CSV文件 - 修复路径问题""" try: # 如果文件名包含路径，确保目录存在 if os.path.dirname(filename) and not os.path.exists(os.path.dirname(filename)): os.makedirs(os.path.dirname(filename), exist_ok=True) with open(filename, 'w', encoding='utf-8-sig', newline='') as f: writer = csv.writer(f) writer.writerow(["用户", "评论内容", "时间", "点赞数", "评论层级"]) for comment in comments: # 清理内容中的换行符 clean_content = comment['content'].replace('\n', ' ').replace('\r', '') writer.writerow([ comment['user'], clean_content, comment['time'], comment['like'], comment['level'] ]) print(f"评论已保存到: {filename}") return True except Exception as e: print(f"保存文件失败: {str(e)}") return Falsedef main(): print("=" * 50) print("B站(Bilibili)视频评论下载工具 v2.1") print("=" * 50) # 获取用户输入 video_url = input("请输入B站视频链接: ").strip() # 从URL中提取BV号 bvid_match = re.search(r"(BV{10})", video_url) if not bvid_match: print("错误: 无法从URL中提取BV号，请确认链接格式正确") return bvid = bvid_match.group(0) print(f"提取到视频BV号: {bvid}") # 获取视频信息 oid, title = get_video_info(bvid) if not oid: print("获取视频信息失败，程序终止") return print(f"视频标题: {title}") print(f"视频OID: {oid}") # 获取评论 all_comments = [] page = 1 max_retries = 3 max_pages = 50 comment_count = 0 print("开始获取评论...") while page
页:
[1]