feitian-project/pachong
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test

Browse Source
This commit is contained in:
cmk_B
2026-06-29 16:09:05 +08:00
parent a7ed2b0d86
commit 06ff51d8f1
3 changed files with 366 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
练习/003_B站视频下载/BliBli.py
+349
View File
@@ -0,0 +1,349 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# ------------------------------------
# Time : 2025/8/10
# File : BliBli.py
# 脚本作用 : 获取B站视频
# 项目名称 :
# ------------------------------------
import logging
import os
import re
import shutil
import time
from concurrent.futures import ThreadPoolExecutor
import requests
from moviepy import AudioFileClip, VideoFileClip
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
# -------------------------- 配置参数（可自定义） --------------------------
# 自定义存储路径（默认在当前目录下的download文件夹，可修改为绝对路径如"D:/B站视频"）
STORAGE_PATH = "E:/download/B站视频"
# 日志输出级别（INFO=基本信息，WARNING=仅警告和错误，ERROR=仅错误）
LOG_LEVEL = logging.WARNING
# 最大线程数（避免请求过于频繁）
MAX_THREADS = 3
# 最大重试次数
MAX_RETRY = 5
# -------------------------------------------------------------------------
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[logging.FileHandler("bilibili.log"), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)
# 全局请求头
headers = {
'referer': 'https://www.bilibili.com/',
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36',
}
def get_response(url, retry_times=3, timeout=15):
"""获取HTTP响应，包含重试机制和错误处理"""
session = requests.Session()
retries = Retry(
total=retry_times,
backoff_factor=0.5,
status_forcelist=[500, 502, 503, 504, 429]
)
session.mount('http://', HTTPAdapter(max_retries=retries))
session.mount('https://', HTTPAdapter(max_retries=retries))
try:
response = session.get(url, headers=headers, timeout=timeout)
response.raise_for_status()
return response
except requests.exceptions.RequestException as e:
logger.error(f"请求失败: {url}（错误: {str(e)}）")
return None
def get_video_urls_advanced(bv_id, cid, page=1):
"""增强版视频链接获取函数，支持新旧格式自动切换"""
try:
# 优先尝试新版DASH格式（视频音频分离）
url = f'https://api.bilibili.com/x/player/playurl?bvid={bv_id}&cid={cid}&qn=112&type=&otype=json'
response = get_response(url)
if not response:
logger.warning(f"尝试旧版格式获取链接: {bv_id}")
return get_old_format_urls(bv_id, cid, page)
data = response.json()
if data.get('code') != 0:
logger.warning(f"新版格式错误，尝试旧版: {data.get('message')}")
return get_old_format_urls(bv_id, cid, page)
dash_data = data.get('data', {}).get('dash', {})
video_urls = dash_data.get('video', [])
audio_urls = dash_data.get('audio', [])
if video_urls and audio_urls:
video_url = video_urls[0].get('backupUrl', [video_urls[0].get('url', '')])[0]
audio_url = audio_urls[0].get('backupUrl', [audio_urls[0].get('url', '')])[0]
return video_url, audio_url
# DASH格式获取失败，尝试旧版格式
#logger.warning(f"新版格式解析失败，切换旧版格式")
return get_old_format_urls(bv_id, cid, page)
except Exception as e:
logger.error(f"链接获取异常，切换旧版: {str(e)}")
return get_old_format_urls(bv_id, cid, page)
def get_old_format_urls(bv_id, cid, page=1):
"""获取旧版FLV格式视频链接（视频音频合并）"""
try:
url = f'https://api.bilibili.com/x/player/playurl?bvid={bv_id}&cid={cid}&qn=112&type=flv&otype=json'
response = get_response(url)
if not response:
logger.error(f"旧版格式请求失败: {bv_id}")
return None, None
data = response.json()
if data.get('code') != 0:
logger.error(f"旧版格式API错误: {data.get('message')}")
return None, None
durl = data.get('data', {}).get('durl', [{}])[0].get('url', '')
if not durl:
logger.error(f"旧版格式链接为空: {bv_id}")
return None, None
return durl, None # 旧版无独立音频链接
except Exception as e:
logger.error(f"旧版格式处理异常: {str(e)}")
return None, None
def download_video_audio_and_combine_compatible(video_url, audio_url, video_path, audio_path, output_path):
"""兼容新旧格式的下载合并函数"""
try:
# 下载视频
print(f"正在下载视频: {os.path.basename(output_path)}") # 控制台显示进度
video_resp = get_response(video_url)
if not video_resp:
logger.error(f"视频下载失败: {os.path.basename(output_path)}")
return False
with open(video_path, 'wb') as f:
f.write(video_resp.content)
# 处理旧版格式(无独立音频)
if not audio_url:
shutil.copy2(video_path, output_path)
print(f"下载完成: {os.path.basename(output_path)}")
return True
# 下载音频(新版格式)
audio_resp = get_response(audio_url)
if not audio_resp:
logger.warning(f"音频下载失败，仅保存视频: {os.path.basename(output_path)}")
shutil.copy2(video_path, output_path)
return True
with open(audio_path, 'wb') as f:
f.write(audio_resp.content)
# 合并视频和音频
try:
video = VideoFileClip(video_path)
audio = AudioFileClip(audio_path)
final_video = video.set_audio(audio)
final_video.write_videofile(output_path, codec="libx264", audio_codec="aac")
print(f"合并完成: {os.path.basename(output_path)}")
return True
except Exception as e:
logger.warning(f"合并失败，保存原始视频: {str(e)}")
shutil.copy2(video_path, output_path)
return True
finally:
# 清理临时文件
if os.path.exists(audio_path):
os.remove(audio_path)
except Exception as e:
logger.error(f"处理失败: {str(e)}")
return False
finally:
if os.path.exists(video_path):
os.remove(video_path)
def add_copyright_headers():
"""添加版权敏感内容所需的请求头（可选）"""
global headers
headers.update({
'Referer': 'https://www.bilibili.com/cheese/',
'Origin': 'https://www.bilibili.com',
# 如需下载会员内容，可添加Cookie（格式: 'Cookie': 'SESSDATA=xxx; bili_jct=xxx;'）
# 'Cookie': 'your_cookie_here'
})
def create_directory(path):
"""创建目录（支持自定义存储路径）"""
dir_path = os.path.dirname(path)
try:
os.makedirs(dir_path, exist_ok=True)
return True
except Exception as e:
logger.error(f"目录创建失败: {dir_path}（错误: {str(e)}）")
return False
def download_part_video_advanced(part_info, main_title):
"""增强版分P下载函数"""
bv_id = part_info['bv']
page = part_info['page']
part_title = part_info['title']
cid = part_info.get('cid')
# 生成文件路径（基于自定义存储路径）
safe_title = re.sub(r'[\\/:*?"<>|]', '_', part_title)
main_dir = os.path.join(STORAGE_PATH, re.sub(r'[\\/:*?"<>|]', '_', main_title))
video_path = os.path.join(main_dir, f'{safe_title}_video.mp4') # 临时视频文件
audio_path = os.path.join(main_dir, f'{safe_title}_audio.mp3') # 临时音频文件
output_path = os.path.join(main_dir, f'{safe_title}.mp4') # 最终输出路径
# 创建目录（包括临时文件夹）
if not create_directory(video_path) or not create_directory(output_path):
logger.error(f"跳过下载: {part_title}（目录创建失败）")
return False
# 重试机制
for retry in range(MAX_RETRY):
wait_time = 2 ** retry # 指数退避（1s, 2s, 4s...）
if retry > 0:
print(f"重试 {retry}/{MAX_RETRY}（等待{wait_time}秒）: {part_title}")
time.sleep(wait_time)
try:
# 获取视频链接
video_url, audio_url = get_video_urls_advanced(bv_id, cid, page)
if not video_url:
continue
# 针对版权敏感内容添加额外头
if "会员" in part_title or "加密" in part_title:
add_copyright_headers()
# 下载并合并
if download_video_audio_and_combine_compatible(video_url, audio_url, video_path, audio_path, output_path):
return True
except Exception as e:
logger.error(f"下载异常: {str(e)}")
continue
logger.error(f"下载失败: {part_title}（已达最大重试次数）")
return False
def parse_video_info(bv_id):
"""解析视频基本信息和分P信息"""
base_url = f'https://www.bilibili.com/video/{bv_id}'
response = get_response(base_url)
if not response:
logger.error(f"无法访问视频页面: {base_url}")
return None, None
html = response.text
try:
# 提取视频标题
title_match = re.search(r'<h1 data-title="(.*?)" title="', html)
if not title_match:
logger.error("无法提取视频标题")
return None, None
main_title = title_match.group(1).strip()
# 通过API获取分P信息
try:
api_url = f'https://api.bilibili.com/x/player/pagelist?bvid={bv_id}&jsonp=jsonp'
api_resp = get_response(api_url)
if not api_resp:
raise Exception("分P API请求失败")
api_data = api_resp.json()
if api_data.get('code') != 0:
raise Exception(f"API错误: {api_data.get('message')}")
parts = api_data.get('data', [])
if not parts:
# 单P视频
cid = re.search(r'"cid":(\d+),', html).group(1)
return [{'title': main_title, 'bv': bv_id, 'page': 1, 'cid': cid}], main_title
# 多P视频
part_list = []
for i, part in enumerate(parts, 1):
part_list.append({
'title': part.get('part', f'第{i}集'),
'bv': bv_id,
'page': part.get('page', i),
'cid': part.get('cid')
})
return part_list, main_title
except Exception as e:
logger.warning(f"API解析失败，尝试HTML解析: {str(e)}")
# HTML备用解析
part_pattern = r'"cid":(\d+),"page":(\d+),"part":"(.*?)","duration"'
parts = re.findall(part_pattern, html)
if not parts:
cid = re.search(r'"cid":(\d+),', html).group(1)
return [{'title': main_title, 'bv': bv_id, 'page': 1, 'cid': cid}], main_title
part_list = []
for cid, page, title in parts:
part_list.append({
'title': title.strip(),
'bv': bv_id,
'page': int(page),
'cid': cid
})
return part_list, main_title
except Exception as e:
logger.error(f"视频信息解析失败: {str(e)}")
return None, None
def download_bilibili_video(bv_id):
"""下载B站视频主函数"""
print(f"开始处理视频: {bv_id}")
part_list, main_title = parse_video_info(bv_id)
if not part_list or not main_title:
logger.error("视频信息解析失败，退出")
return False
print(f"视频标题: {main_title}（共{len(part_list)}个分P）")
print(f"文件将保存至: {os.path.abspath(os.path.join(STORAGE_PATH, main_title))}\n")
# 多线程下载
with ThreadPoolExecutor(max_workers=MAX_THREADS) as executor:
futures = [executor.submit(download_part_video_advanced, part, main_title) for part in part_list]
# 等待所有任务完成
completed = 0
for future in futures:
if future.result():
completed += 1
# 下载结果统计
success_rate = completed / len(part_list) * 100
print(f"\n下载结束: 共{len(part_list)}个分P，成功{completed}个（成功率: {success_rate:.1f}%）")
print(f"文件保存位置: {os.path.abspath(os.path.join(STORAGE_PATH, main_title))}")
return success_rate >= 50
if __name__ == '__main__':
# 示例：替换为需要下载的BV号
bv_id = 'BV1xe4y1s7Uq' # 可修改为目标视频的BV号
download_bilibili_video(bv_id)
练习/003_B站视频下载/README.md
+12
View File
@@ -0,0 +1,12 @@
# tools
#### 介绍
个人脚本存放
#### 软件架构
软件架构说明
#### 使用说明
获取bilibili上面免费视频
练习/003_B站视频下载/bilibili.log
+5
View File
@@ -0,0 +1,5 @@
2025-08-10 19:20:08,907 - WARNING - 新版格式解析失败，切换旧版格式
2025-08-10 19:20:08,928 - WARNING - 新版格式解析失败，切换旧版格式
2025-08-10 19:20:08,930 - WARNING - 新版格式解析失败，切换旧版格式
2025-08-10 19:20:16,658 - WARNING - 新版格式解析失败，切换旧版格式
2025-08-10 19:20:16,677 - WARNING - 新版格式解析失败，切换旧版格式