爬虫下载B站视频简单步伐（仅供学习）

请输入视频链接：https://www.bilibili.com/video/BV1owFSeREoh （示例所在）                                 步伐表现结果如下：              
下载进度: 100.00%
下载完成
视频已生存到: ./video.mp4
核心功能

1. 视频信息剖析

• 主动辨认B站视频格式（新版DASH/旧版FLV）
  

• 提取最高画质视频流所在
  

• 支持分段视频检测
  

2.下载

• 多线程分块下载（1MB/块）
  

• 主动重试机制（最多5次）
  

• 及时下载进度表现
  

• 支持署理和反爬计谋
  

- 通过正则提取B站视频核心数据

• 包罗分辨率、码率、分段所在等关键信息   re.search(r'<script>window\.__playinfo__=(.*?)</script>', html)
  

注意：

• 仅下载已得到观看权限的内容
  

• 本代码仅供学习利用，请服从B站用户协媾和版权法规
  

2. import re
3. import json
4. import requests
5. from urllib.parse import urlparse
6. import subprocess
7. import time
9. class BilibiliDownloader:
10.     def __init__(self, url):
11.         self.url = url
12.         self.headers = {
13.             'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
14.             'Referer': 'https://www.bilibili.com/'
15.         }
16.    
17.     def get_video_info(self):
18.         # 获取视频页面源码
19.         response = requests.get(self.url, headers=self.headers)
20.         html = response.text
21.         
22.         # 提取JSON格式的视频信息
23.         match = re.search(r'<script>window\.__playinfo__=(.*?)</script>', html)
24.         if match:
25.             return json.loads(match.group(1))
26.         else:
27.             raise Exception("无法提取视频信息")
29.     def download_video(self, save_path='./video.mp4'):
30.         # 获取视频下载链接
31.         video_info = self.get_video_info()
32.         
33.         # 处理不同视频格式（优先dash格式）
34.         try:
35.             if video_info['data'].get('dash'):
36.                 # Dash格式视频流
37.                 video_list = video_info['data']['dash']['video']
38.                 best_quality = max(video_list, key=lambda x: x['bandwidth'])
39.                 video_url = best_quality['baseUrl']
40.             else:  
41.                 # 旧版FLV格式
42.                 video_list = video_info['data']['durl']
43.                 best_quality = max(video_list, key=lambda x: x['size'])
44.                 video_url = best_quality['url']
45.                
46.                 # 处理分段视频（需要合并多个flv）
47.                 if len(video_list) > 1:
48.                     print("检测到分段视频，需要合并多个文件...")
49.                     return self._download_flv_segments(video_list, save_path)
51.         except KeyError as e:
52.             raise Exception(f"不支持的视频格式: {str(e)}")
53.         
54.         # 使用requests流式下载（带重试机制）
55.         max_retries = 5
56.         timeout = 15  # 设置超时时间（秒）
57.         chunk_size = 1024 * 1024  # 1MB chunks
58.         
59.         for attempt in range(max_retries):
60.             try:
61.                 with requests.get(video_url, headers=self.headers,
62.                                 stream=True, timeout=timeout) as r:
63.                     r.raise_for_status()
64.                     total_size = int(r.headers.get('content-length', 0))
65.                     
66.                     with open(save_path, 'wb') as f:
67.                         downloaded = 0
68.                         for chunk in r.iter_content(chunk_size=chunk_size):
69.                             if chunk:  # 过滤保持连接的空白chunk
70.                                 f.write(chunk)
71.                                 downloaded += len(chunk)
72.                                 print(f"\r下载进度: {downloaded/total_size:.2%}", end='')
73.                     print("\n下载完成")
74.                     break
75.                     
76.             except (requests.exceptions.Timeout,
77.                   requests.exceptions.ConnectionError) as e:
78.                 print(f"\n第 {attempt+1} 次尝试失败: {str(e)}")
79.                 if attempt == max_retries - 1:
80.                     raise Exception("多次尝试下载失败，请检查网络连接")
81.                 time.sleep(2 ** attempt)  # 指数退避
82.         
83.         # 合并音视频（需要ffmpeg）
84.         self._merge_audio_video(save_path)
85.         print(f"视频已保存到: {save_path}")
87.     def _download_flv_segments(self, segments, save_path):
88.         """下载并合并FLV分段视频"""
89.         temp_files = []
90.         for i, seg in enumerate(segments):
91.             part_url = seg['url']
92.             part_file = f"{save_path}.part{i}"
93.             temp_files.append(part_file)
94.             
95.             with requests.get(part_url, headers=self.headers, stream=True) as r:
96.                 with open(part_file, 'wb') as f:
97.                     for chunk in r.iter_content(chunk_size=8192):
98.                         f.write(chunk)
99.         self._merge_files(temp_files, save_path)
101.   
103. if __name__ == "__main__":
104.     video_url = input("请输入视频链接：")
105.     downloader = BilibiliDownloader(video_url)
106.     downloader.download_video()

复制代码