From 7d086587576a49c6f423a3d2d8c72d15012fc84b Mon Sep 17 00:00:00 2001 From: dvs-dvsxt Date: Fri, 28 Aug 2026 10:12:20 +0800 Subject: [PATCH] Initial commit: Bilibili Video Downloader v1.0.0 --- .gitignore | 21 ++ LICENSE | 21 ++ README.md | 95 +++++++++ bilibili_downloader.py | 389 ++++++++++++++++++++++++++++++++++++ bilibili_downloader_auto.py | 385 +++++++++++++++++++++++++++++++++++ 5 files changed, 911 insertions(+) create mode 100644 .gitignore create mode 100644 LICENSE create mode 100644 README.md create mode 100644 bilibili_downloader.py create mode 100644 bilibili_downloader_auto.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..dd31224 --- /dev/null +++ b/.gitignore @@ -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 diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..fddd2dd --- /dev/null +++ b/LICENSE @@ -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. diff --git a/README.md b/README.md new file mode 100644 index 0000000..88b34e9 --- /dev/null +++ b/README.md @@ -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. diff --git a/bilibili_downloader.py b/bilibili_downloader.py new file mode 100644 index 0000000..7ecb361 --- /dev/null +++ b/bilibili_downloader.py @@ -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() diff --git a/bilibili_downloader_auto.py b/bilibili_downloader_auto.py new file mode 100644 index 0000000..731ba2e --- /dev/null +++ b/bilibili_downloader_auto.py @@ -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()