390 lines
14 KiB
Python
390 lines
14 KiB
Python
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()
|