Initial commit: Bilibili Video Downloader v1.0.0

This commit is contained in:
dvs
2026-08-28 10:12:20 +08:00
commit 7d08658757
5 changed files with 911 additions and 0 deletions
+21
View File
@@ -0,0 +1,21 @@
# Python
__pycache__/
*.py[cod]
*.egg-info/
dist/
build/
.venv/
venv/
env/
# Downloads
downloads/
videos/
*.mp4
*.flv
*.m4a
# System files
Thumbs.db
.DS_Store
desktop.ini
+21
View File
@@ -0,0 +1,21 @@
MIT License
Copyright (c) 2026 DVS (dvs-dvsxt)
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
+95
View File
@@ -0,0 +1,95 @@
# 📺 Bilibili Video Downloader
> A Python tool to **download videos from Bilibili (B站)** with WBI signature authentication, multi-quality selection, and optional ffmpeg audio+video merging.
**Bilibili Video Downloader** downloads videos from Bilibili by BV URL. It implements the WBI signature algorithm for API authentication, picks the best available MP4 quality, and supports downloading with **aria2** (fast) or pure-Python fallback, then optionally merges audio & video with **ffmpeg**.
---
## 📦 Files
| File | Description |
|------|-------------|
| `bilibili_downloader.py` | Standard version — fill in your `sessdata` cookie manually |
| `bilibili_downloader_auto.py` | Auto version — edit the `sessdata` variable inside the script |
> Both files have identical features. The difference is only how the cookie is provided.
---
## ✨ Features
| Feature | Description |
|---------|-------------|
| 🔑 **WBI Signature** | Implements Bilibili WBI signing algorithm for authenticated API calls |
| 🎞️ **Best MP4 Quality** | Automatically picks the highest available MP4 stream |
| 🧪 **URL Test** | Verifies content URL accessibility before download |
| ⚡ **aria2 Support** | Uses aria2 for fast parallel downloads (if installed) |
| 🐍 **Pure-Python Fallback** | If aria2 is unavailable, downloads directly via Python |
| 🎬 **ffmpeg Merge** | Merges separate video + audio streams into one file |
| 📄 **Info Fetch** | Retrieves video metadata (title, etc.) via BV URL |
---
## 🚀 Quick Start
### Prerequisites
- Python 3.7+
- `requests` library
- **Cookie** (`sessdata`) — required for Bilibili access
- Optional: `aria2` (fast), `ffmpeg` (audio/video merge)
### Install
```bash
pip install requests
```
### Setup Cookie
1. Log in to Bilibili in your browser
2. Open DevTools → Application → Cookies → find `sessdata`
3. Paste it into the script:
**`bilibili_downloader.py`** (fill at runtime or edit variable):
```python
sessdata = "" # <-- paste your sessdata
```
**`bilibili_downloader_auto.py`** (edit variable directly):
```python
sessdata = '填写你的Cookie' # <-- replace with your sessdata
```
### Run
```bash
python bilibili_downloader.py
# or
python bilibili_downloader_auto.py
```
Enter a Bilibili video BV URL when prompted (e.g. `BV1xx411c7mD`).
---
## 📁 Project Structure
```
Bilibili-Video-Downloader/
├── bilibili_downloader.py # Standard version
├── bilibili_downloader_auto.py # Auto version (cookie in script)
└── README.md # This document
```
---
## 📄 License
This project is licensed under the **MIT License**. See [LICENSE](LICENSE) for details.
---
## ⚠️ Disclaimer
> This tool is for **personal learning and research only**. Please respect Bilibili's Terms of Service and content creators' copyright. Do not redistribute downloaded content. Use at your own risk.
+389
View File
@@ -0,0 +1,389 @@
import requests
import time
import hashlib
import urllib.parse
import json
import os
import subprocess
from typing import Dict, List
#填写你的Cookie
sessdata = ""
class BilibiliAutoDownloader:
def __init__(self, ffmpeg_path="D:\\ff\\ffmpeg.exe"):
self.sessdata = sessdata
self.session = requests.Session()
self.ffmpeg_path = ffmpeg_path
# 设置完整的cookie和headers,模拟真实浏览器
self.session.headers.update({
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
'Referer': 'https://www.bilibili.com',
'Origin': 'https://www.bilibili.com',
'Accept': 'application/json, text/plain, */*',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
'Accept-Encoding': 'gzip, deflate, br',
'Cache-Control': 'no-cache',
'Priority': 'u=1, i'
})
if sessdata:
self.session.cookies.set('SESSDATA', sessdata, domain='.bilibili.com')
# 设置其他必要的cookies
self.session.cookies.set('buvid3', '填写你的Cookie', domain='.bilibili.com')
self.session.cookies.set('DedeUserID', '填写你的Cookie', domain='.bilibili.com')
self.session.cookies.set('bili_jct', '填写你的Cookie', domain='.bilibili.com')
self.session.cookies.set('CURRENT_FNVAL', '填写你的Cookie', domain='.bilibili.com')
def get_video_info(self, bvid: str) -> Dict:
"""获取视频信息"""
api_url = "https://api.bilibili.com/x/web-interface/view"
params = {'bvid': bvid}
response = self.session.get(api_url, params=params)
return response.json()
def _get_wbi_keys(self) -> tuple:
"""获取WBI加密密钥"""
nav_url = "https://api.bilibili.com/x/web-interface/nav"
response = self.session.get(nav_url)
data = response.json()
if data['code'] == 0:
img_url = data['data']['wbi_img']['img_url']
sub_url = data['data']['wbi_img']['sub_url']
img_key = img_url.split('/')[-1].split('.')[0]
sub_key = sub_url.split('/')[-1].split('.')[0]
return img_key, sub_key
return None, None
def _wbi_sign(self, params: Dict) -> Dict:
"""WBI签名算法"""
img_key, sub_key = self._get_wbi_keys()
if not img_key or not sub_key:
return params
mixin_key = [
46, 47, 18, 2, 53, 8, 23, 32, 15, 50, 10, 31, 58, 3, 45, 35, 27, 43, 5, 49,
33, 9, 42, 19, 29, 28, 14, 39, 12, 38, 41, 13, 37, 48, 7, 16, 24, 55, 40,
61, 26, 17, 0, 1, 60, 51, 30, 4, 22, 25, 54, 21, 56, 59, 6, 63, 57, 62, 11,
36, 20, 34, 44, 52
]
# 添加时间戳
params['wts'] = int(time.time())
# 参数排序并过滤
params = {k: ''.join(filter(str.isalnum, str(v))) for k, v in params.items()
if k not in ['sign', 'csrf'] and v != ''}
params = dict(sorted(params.items()))
# 生成待签名字符串
query = urllib.parse.urlencode(params)
# 应用mixin_key
mixed = []
for key in mixin_key:
if key < len(query):
mixed.append(query[key])
mixed_str = ''.join(mixed)[:32]
# 生成w_rid
w_rid = hashlib.md5((query + mixed_str).encode()).hexdigest()
params['w_rid'] = w_rid
return params
def get_play_url_mp4(self, bvid: str, cid: int) -> Dict:
"""获取MP4格式的播放地址(非DASH)"""
# 先获取avid
video_info = self.get_video_info(bvid)
if not video_info or video_info.get('code') != 0:
return None
avid = video_info['data']['aid']
# 构建参数 - 不使用DASH格式
params = {
'qn': 112, # 高清1080P
'fnval': 0, # 设置为0表示不使用DASH格式
'fourk': 1,
'voice_balance': 1,
'gaia_source': 'pre-load',
'isGaiaAvoided': 'true',
'avid': avid,
'bvid': bvid,
'cid': cid,
'web_location': '1315873',
}
# 应用WBI签名
signed_params = self._wbi_sign(params)
# 使用原生API端点
api_url = "https://api.bilibili.com/x/player/wbi/playurl"
response = self.session.get(api_url, params=signed_params)
return response.json()
def find_best_mp4_url(self, vajson: Dict) -> tuple:
"""找到最高质量的MP4视频URL"""
if 'data' not in vajson or 'durl' not in vajson['data']:
return None, None
durls = vajson['data']['durl']
if not durls:
return None, None
# 获取视频质量信息
quality = vajson['data'].get('quality', 0)
format_info = vajson['data'].get('format', '')
description = vajson['data'].get('accept_description', [])
print(f"🎯 找到 {len(durls)} 个MP4视频片段")
print(f"📊 视频质量: {quality} - {format_info}")
if description:
print(f"📋 可用质量: {', '.join(description)}")
# 如果有多个片段,选择第一个(通常是完整的视频)
best_durl = durls[0]
video_url = best_durl.get('url', '')
backup_urls = best_durl.get('backup_url', [])
size = best_durl.get('size', 0)
print(f"🔥 选择视频片段 - 大小: {size/(1024*1024):.2f}MB")
# 测试主URL
if self.test_url_content(video_url):
print(f"✅ 主视频URL可用")
return video_url, f"MP4-{quality}"
# 测试备用URL
for i, backup_url in enumerate(backup_urls):
if self.test_url_content(backup_url):
print(f"✅ 备用视频URL {i+1} 可用")
return backup_url, f"MP4-{quality}"
return None, None
def test_url_content(self, url: str, min_size: int = 1024) -> bool:
"""测试URL是否返回有效内容"""
try:
headers = {
'Range': 'bytes=0-8191',
'Referer': 'https://www.bilibili.com',
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
}
response = self.session.get(url, headers=headers, timeout=10, stream=True)
if response.status_code not in [200, 206]:
return False
content_length = response.headers.get('content-length')
if content_length and int(content_length) < min_size:
return False
return True
except Exception:
return False
def download_with_aria2(self, url: str, filename: str, file_type: str):
"""使用aria2多线程下载"""
print(f"🚀 开始高速下载{file_type}")
cmd = [
'aria2c',
'-x', '16',
'-s', '16',
'-k', '1M',
'--header=Referer: https://www.bilibili.com',
'--check-certificate=false',
'--continue=true',
'--max-tries=5',
'--retry-wait=3',
'-o', filename,
url
]
try:
result = subprocess.run(cmd, capture_output=True)
if result.returncode == 0:
print(f"✅ {file_type}下载完成")
return True
else:
return False
except FileNotFoundError:
return False
def download_fast_python(self, url: str, filename: str, file_type: str):
"""Python多线程下载"""
import threading
from concurrent.futures import ThreadPoolExecutor
print(f"🚀 开始多线程下载{file_type}")
def download_chunk(start_byte, end_byte, chunk_id):
headers = {
'Range': f'bytes={start_byte}-{end_byte}',
'Referer': 'https://www.bilibili.com',
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
}
try:
response = self.session.get(url, headers=headers, stream=True, timeout=30)
with open(f"{filename}.part{chunk_id}", 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
return True
except Exception:
return False
try:
response = self.session.head(url, headers={'Referer': 'https://www.bilibili.com'})
total_size = int(response.headers.get('content-length', 0))
except:
total_size = 0
if total_size == 0:
return self.download_single(url, filename, file_type)
num_threads = 8
chunk_size = total_size // num_threads
with ThreadPoolExecutor(max_workers=num_threads) as executor:
futures = []
for i in range(num_threads):
start_byte = i * chunk_size
end_byte = start_byte + chunk_size - 1 if i < num_threads - 1 else total_size - 1
futures.append(executor.submit(download_chunk, start_byte, end_byte, i))
results = [f.result() for f in futures]
if all(results):
with open(filename, 'wb') as outfile:
for i in range(num_threads):
with open(f"{filename}.part{i}", 'rb') as infile:
outfile.write(infile.read())
os.remove(f"{filename}.part{i}")
print(f"✅ {file_type}下载完成")
return True
else:
return False
def download_single(self, url: str, filename: str, file_type: str):
"""单线程下载"""
print(f"📥 开始下载{file_type}")
max_retries = 3
for attempt in range(max_retries):
try:
headers = {
'Referer': 'https://www.bilibili.com',
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
}
response = self.session.get(url, headers=headers, stream=True, timeout=30)
total_size = int(response.headers.get('content-length', 0))
downloaded = 0
with open(filename, 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
downloaded += len(chunk)
if total_size > 0:
percent = (downloaded / total_size) * 100
print(f"\r📊 下载进度: {percent:.1f}%", end='', flush=True)
if os.path.exists(filename) and os.path.getsize(filename) > 1024:
print(f"\n✅ {file_type}下载完成")
return True
else:
if os.path.exists(filename):
os.remove(filename)
except Exception:
if os.path.exists(filename):
os.remove(filename)
return False
def download_and_merge(self, bvid: str):
"""主下载函数 - 直接下载MP4格式"""
bvid = bvid.rstrip('/')
print(f"🚀 开始处理视频: {bvid}")
# 获取视频信息
video_info = self.get_video_info(bvid)
if not video_info or video_info.get('code') != 0:
print("❌ 无法获取视频信息")
return
data = video_info['data']
cid = data['cid']
title = data['title']
print(f"📺 视频标题: {title}")
print(f"🔢 视频CID: {cid}")
# 使用MP4 API获取播放信息
vajson = self.get_play_url_mp4(bvid, cid)
if not vajson or vajson.get('code') != 0:
print("❌ 无法获取MP4播放信息")
return
print("✅ 成功获取MP4播放信息")
# 找到MP4视频URL
video_url, quality_info = self.find_best_mp4_url(vajson)
if not video_url:
print("❌ 没有可用的MP4视频URL")
return
print(f"🎬 已选择MP4视频 - 质量: {quality_info}")
# 生成安全的文件名
safe_title = "".join(c for c in title if c.isalnum() or c in ('-', '_')).rstrip()
if not safe_title:
safe_title = "video"
output_file = f"{safe_title}.mp4"
# 选择下载方式
download_methods = [
('aria2', self.download_with_aria2),
('多线程', self.download_fast_python),
('单线程', self.download_single)
]
# 直接下载MP4文件
video_success = False
for method_name, method_func in download_methods:
print(f"\n尝试使用 {method_name} 下载MP4视频...")
if method_func(video_url, output_file, "MP4视频"):
video_success = True
break
if video_success:
print(f"\n🎉 下载完成!")
print(f"📹 最终文件: {output_file}")
print(f"🔧 视频格式: {quality_info}")
if os.path.exists(output_file):
file_size = os.path.getsize(output_file)
print(f"📁 文件大小: {file_size / (1024*1024):.2f} MB")
else:
print("❌ 下载失败")
# 🎯 主程序
if __name__ == "__main__":
downloader = BilibiliAutoDownloader(ffmpeg_path=r"D:\ff\ffmpeg.exe")
try:
bvid = input("📥 请输入BV号: ").strip()
if bvid:
downloader.download_and_merge(bvid)
except Exception as e:
print(f"💥 程序执行失败: {e}")
import traceback
traceback.print_exc()
+385
View File
@@ -0,0 +1,385 @@
import requests
import time
import hashlib
import urllib.parse
import json
import os
import subprocess
from typing import Dict, List
sessdata = '填写你的Cookie'
class BilibiliAutoDownloader:
def __init__(self, ffmpeg_path="D:\\ff\\ffmpeg.exe"):
self.sessdata = sessdata
self.session = requests.Session()
self.ffmpeg_path = ffmpeg_path
# 设置完整的cookie和headers,模拟真实浏览器
self.session.headers.update({
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
'Referer': 'https://www.bilibili.com',
'Origin': 'https://www.bilibili.com',
'Accept': 'application/json, text/plain, */*',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
'Accept-Encoding': 'gzip, deflate, br',
'Cache-Control': 'no-cache',
'Priority': 'u=1, i'
})
if sessdata:
self.session.cookies.set('SESSDATA', sessdata, domain='.bilibili.com')
# 设置其他必要的cookies
def get_video_info(self, bvid: str) -> Dict:
"""获取视频信息"""
api_url = "https://api.bilibili.com/x/web-interface/view"
params = {'bvid': bvid}
response = self.session.get(api_url, params=params)
return response.json()
def _get_wbi_keys(self) -> tuple:
"""获取WBI加密密钥"""
nav_url = "https://api.bilibili.com/x/web-interface/nav"
response = self.session.get(nav_url)
data = response.json()
if data['code'] == 0:
img_url = data['data']['wbi_img']['img_url']
sub_url = data['data']['wbi_img']['sub_url']
img_key = img_url.split('/')[-1].split('.')[0]
sub_key = sub_url.split('/')[-1].split('.')[0]
return img_key, sub_key
return None, None
def _wbi_sign(self, params: Dict) -> Dict:
"""WBI签名算法"""
img_key, sub_key = self._get_wbi_keys()
if not img_key or not sub_key:
return params
mixin_key = [
46, 47, 18, 2, 53, 8, 23, 32, 15, 50, 10, 31, 58, 3, 45, 35, 27, 43, 5, 49,
33, 9, 42, 19, 29, 28, 14, 39, 12, 38, 41, 13, 37, 48, 7, 16, 24, 55, 40,
61, 26, 17, 0, 1, 60, 51, 30, 4, 22, 25, 54, 21, 56, 59, 6, 63, 57, 62, 11,
36, 20, 34, 44, 52
]
# 添加时间戳
params['wts'] = int(time.time())
# 参数排序并过滤
params = {k: ''.join(filter(str.isalnum, str(v))) for k, v in params.items()
if k not in ['sign', 'csrf'] and v != ''}
params = dict(sorted(params.items()))
# 生成待签名字符串
query = urllib.parse.urlencode(params)
# 应用mixin_key
mixed = []
for key in mixin_key:
if key < len(query):
mixed.append(query[key])
mixed_str = ''.join(mixed)[:32]
# 生成w_rid
w_rid = hashlib.md5((query + mixed_str).encode()).hexdigest()
params['w_rid'] = w_rid
return params
def get_play_url_mp4(self, bvid: str, cid: int) -> Dict:
"""获取MP4格式的播放地址(非DASH)"""
# 先获取avid
video_info = self.get_video_info(bvid)
if not video_info or video_info.get('code') != 0:
return None
avid = video_info['data']['aid']
# 构建参数 - 不使用DASH格式
params = {
'qn': 112, # 高清1080P
'fnval': 0, # 设置为0表示不使用DASH格式
'fourk': 1,
'voice_balance': 1,
'gaia_source': 'pre-load',
'isGaiaAvoided': 'true',
'avid': avid,
'bvid': bvid,
'cid': cid,
'web_location': '1315873',
}
# 应用WBI签名
signed_params = self._wbi_sign(params)
# 使用原生API端点
api_url = "https://api.bilibili.com/x/player/wbi/playurl"
response = self.session.get(api_url, params=signed_params)
return response.json()
def find_best_mp4_url(self, vajson: Dict) -> tuple:
"""找到最高质量的MP4视频URL"""
if 'data' not in vajson or 'durl' not in vajson['data']:
return None, None
durls = vajson['data']['durl']
if not durls:
return None, None
# 获取视频质量信息
quality = vajson['data'].get('quality', 0)
format_info = vajson['data'].get('format', '')
description = vajson['data'].get('accept_description', [])
print(f"🎯 找到 {len(durls)} 个MP4视频片段")
print(f"📊 视频质量: {quality} - {format_info}")
if description:
print(f"📋 可用质量: {', '.join(description)}")
# 如果有多个片段,选择第一个(通常是完整的视频)
best_durl = durls[0]
video_url = best_durl.get('url', '')
backup_urls = best_durl.get('backup_url', [])
size = best_durl.get('size', 0)
print(f"🔥 选择视频片段 - 大小: {size/(1024*1024):.2f}MB")
# 测试主URL
if self.test_url_content(video_url):
print(f"✅ 主视频URL可用")
return video_url, f"MP4-{quality}"
# 测试备用URL
for i, backup_url in enumerate(backup_urls):
if self.test_url_content(backup_url):
print(f"✅ 备用视频URL {i+1} 可用")
return backup_url, f"MP4-{quality}"
return None, None
def test_url_content(self, url: str, min_size: int = 1024) -> bool:
"""测试URL是否返回有效内容"""
try:
headers = {
'Range': 'bytes=0-8191',
'Referer': 'https://www.bilibili.com',
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
}
response = self.session.get(url, headers=headers, timeout=10, stream=True)
if response.status_code not in [200, 206]:
return False
content_length = response.headers.get('content-length')
if content_length and int(content_length) < min_size:
return False
return True
except Exception:
return False
def download_with_aria2(self, url: str, filename: str, file_type: str):
"""使用aria2多线程下载"""
print(f"🚀 开始高速下载{file_type}")
cmd = [
'aria2c',
'-x', '16',
'-s', '16',
'-k', '1M',
'--header=Referer: https://www.bilibili.com',
'--check-certificate=false',
'--continue=true',
'--max-tries=5',
'--retry-wait=3',
'-o', filename,
url
]
try:
result = subprocess.run(cmd, capture_output=True)
if result.returncode == 0:
print(f"✅ {file_type}下载完成")
return True
else:
return False
except FileNotFoundError:
return False
def download_fast_python(self, url: str, filename: str, file_type: str):
"""Python多线程下载"""
import threading
from concurrent.futures import ThreadPoolExecutor
print(f"🚀 开始多线程下载{file_type}")
def download_chunk(start_byte, end_byte, chunk_id):
headers = {
'Range': f'bytes={start_byte}-{end_byte}',
'Referer': 'https://www.bilibili.com',
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
}
try:
response = self.session.get(url, headers=headers, stream=True, timeout=30)
with open(f"{filename}.part{chunk_id}", 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
return True
except Exception:
return False
try:
response = self.session.head(url, headers={'Referer': 'https://www.bilibili.com'})
total_size = int(response.headers.get('content-length', 0))
except:
total_size = 0
if total_size == 0:
return self.download_single(url, filename, file_type)
num_threads = 8
chunk_size = total_size // num_threads
with ThreadPoolExecutor(max_workers=num_threads) as executor:
futures = []
for i in range(num_threads):
start_byte = i * chunk_size
end_byte = start_byte + chunk_size - 1 if i < num_threads - 1 else total_size - 1
futures.append(executor.submit(download_chunk, start_byte, end_byte, i))
results = [f.result() for f in futures]
if all(results):
with open(filename, 'wb') as outfile:
for i in range(num_threads):
with open(f"{filename}.part{i}", 'rb') as infile:
outfile.write(infile.read())
os.remove(f"{filename}.part{i}")
print(f"✅ {file_type}下载完成")
return True
else:
return False
def download_single(self, url: str, filename: str, file_type: str):
"""单线程下载"""
print(f"📥 开始下载{file_type}")
max_retries = 3
for attempt in range(max_retries):
try:
headers = {
'Referer': 'https://www.bilibili.com',
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
}
response = self.session.get(url, headers=headers, stream=True, timeout=30)
total_size = int(response.headers.get('content-length', 0))
downloaded = 0
with open(filename, 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
downloaded += len(chunk)
if total_size > 0:
percent = (downloaded / total_size) * 100
print(f"\r📊 下载进度: {percent:.1f}%", end='', flush=True)
if os.path.exists(filename) and os.path.getsize(filename) > 1024:
print(f"\n✅ {file_type}下载完成")
return True
else:
if os.path.exists(filename):
os.remove(filename)
except Exception:
if os.path.exists(filename):
os.remove(filename)
return False
def download_and_merge(self, bvid: str):
"""主下载函数 - 直接下载MP4格式"""
bvid = bvid.rstrip('/')
print(f"🚀 开始处理视频: {bvid}")
# 获取视频信息
video_info = self.get_video_info(bvid)
if not video_info or video_info.get('code') != 0:
print("❌ 无法获取视频信息")
return
data = video_info['data']
cid = data['cid']
title = data['title']
print(f"📺 视频标题: {title}")
print(f"🔢 视频CID: {cid}")
# 使用MP4 API获取播放信息
vajson = self.get_play_url_mp4(bvid, cid)
if not vajson or vajson.get('code') != 0:
print("❌ 无法获取MP4播放信息")
return
print("✅ 成功获取MP4播放信息")
# 找到MP4视频URL
video_url, quality_info = self.find_best_mp4_url(vajson)
if not video_url:
print("❌ 没有可用的MP4视频URL")
return
print(f"🎬 已选择MP4视频 - 质量: {quality_info}")
# 生成安全的文件名
safe_title = "".join(c for c in title if c.isalnum() or c in ('-', '_')).rstrip()
if not safe_title:
safe_title = "video"
output_file = f"{safe_title}.mp4"
# 选择下载方式
download_methods = [
('aria2', self.download_with_aria2),
('多线程', self.download_fast_python),
('单线程', self.download_single)
]
# 直接下载MP4文件
video_success = False
for method_name, method_func in download_methods:
print(f"\n尝试使用 {method_name} 下载MP4视频...")
if method_func(video_url, output_file, "MP4视频"):
video_success = True
break
if video_success:
print(f"\n🎉 下载完成!")
print(f"📹 最终文件: {output_file}")
print(f"🔧 视频格式: {quality_info}")
if os.path.exists(output_file):
file_size = os.path.getsize(output_file)
print(f"📁 文件大小: {file_size / (1024*1024):.2f} MB")
else:
print("❌ 下载失败")
# 🎯 主程序
if __name__ == "__main__":
downloader = BilibiliAutoDownloader(ffmpeg_path=r"D:\ff\ffmpeg.exe")
try:
bvid = input("📥 请输入BV号: ").strip()
if bvid:
downloader.download_and_merge(bvid)
except Exception as e:
print(f"💥 程序执行失败: {e}")
import traceback
traceback.print_exc()