mirror of
https://github.com/JefferyHcool/BiliNote.git
synced 2026-09-06 16:16:56 +08:00
pumpkinperson996
58ec74f2a2
## 问题
任意一次瞬时网络故障都会让整个笔记任务失败，即便立刻重试同一个链接就能成功。
实际遇到的报错：
ERROR: [download] Got error: HTTPSConnectionPool(
host='upos-sz-mirrorcosov.bilivideo.com', port=443): Read timed out.
同一个视频一分钟后重新提交，10.18 MiB 四秒下完。
## 原因
yt-dlp 文档里 `retries` 默认 10，但那个默认值是**命令行参数解析器**给的，
Python API 不套用。项目所有 `ydl_opts` 都没设 `retries`，于是：
# yt_dlp/downloader/http.py
for retry in RetryManager(self.params.get('retries'), ...) # -> None
# yt_dlp/utils/_utils.py
self.retries = _retries or 0 # -> 0
也就是**每次下载只尝试一次，零重试**。这不是 B 站独有的问题，
YouTube 以及所有走 yt-dlp 的路径都一样。
## 改动
- `base.py`：新增 `YDL_RETRY_OPTS`（`retries` / `fragment_retries` /
`socket_timeout`），两个 downloader 本来就从 base 导入，不额外引入模块。
- `youtube_downloader.py`（2 处）、`bilibili_downloader.py`（3 处）：
所有 `ydl_opts` 都展开该常量。只修报错的那一处会把其余四处继续留在零重试。
- 用的是 yt-dlp 自带的重试机制，没有自写重试循环。
取值偏保守（3 次而非 CLI 的 10）：笔记任务是用户在前台等的，
重试太久不如早点失败让用户重来。
## 测试
新增 `tests/test_ydl_retry_opts.py`：
- 行为：常量给出的 RetryManager 预算 > 0；并显式钉住
`RetryManager(None).retries == 0` 这个被规避的坑。
- 结构：用 AST 断言两个 downloader 里**每一个** `ydl_opts` 字面量都展开了
`YDL_RETRY_OPTS`，防止以后新增下载路径时又悄悄回到零重试。
结构用例确认过 red-green：去掉任一处展开即失败，并指出具体文件行号。
无 yt-dlp 的环境下两个行为用例自动 skip，与仓库既有测试风格一致。
352 lines
13 KiB
Python
352 lines
13 KiB
Python
import os
|
||
import json
|
||
import logging
|
||
import tempfile
|
||
from abc import ABC
|
||
from typing import Union, Optional, List
|
||
|
||
import yt_dlp
|
||
|
||
from app.downloaders.base import Downloader, DownloadQuality, QUALITY_MAP, YDL_RETRY_OPTS
|
||
from app.downloaders.bilibili_dm_patch import apply_bilibili_dm_img_patch
|
||
from app.downloaders.bilibili_subtitle import BilibiliSubtitleFetcher
|
||
from app.models.notes_model import AudioDownloadResult
|
||
from app.models.transcriber_model import TranscriptResult, TranscriptSegment
|
||
from app.utils.path_helper import get_data_dir
|
||
from app.utils.url_parser import extract_video_id
|
||
from app.services.cookie_manager import CookieConfigManager
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Inject the dm_img_* / web_location risk-control params Bilibili's wbi/playurl
|
||
# gateway now requires; without them the API path returns HTTP 412. See
|
||
# app/downloaders/bilibili_dm_patch.py for details.
|
||
apply_bilibili_dm_img_patch()
|
||
|
||
|
||
class BilibiliDownloader(Downloader, ABC):
|
||
def __init__(self):
|
||
super().__init__()
|
||
self._cookie_mgr = CookieConfigManager()
|
||
self._cookie = self._cookie_mgr.get('bilibili')
|
||
self._cookiefile = self._write_netscape_cookie_file()
|
||
|
||
def _write_netscape_cookie_file(self) -> Optional[str]:
|
||
"""将 Cookie 写入 Netscape 格式临时文件，返回文件路径（供 yt-dlp cookiefile 使用）"""
|
||
if not self._cookie:
|
||
logger.warning("B站 Cookie 未配置，下载可能失败")
|
||
return None
|
||
lines = ["# Netscape HTTP Cookie File\n"]
|
||
for pair in self._cookie.split("; "):
|
||
if "=" in pair:
|
||
key, value = pair.split("=", 1)
|
||
lines.append(f".bilibili.com\tTRUE\t/\tFALSE\t0\t{key}\t{value}\n")
|
||
tmp = tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False, encoding='utf-8')
|
||
tmp.writelines(lines)
|
||
tmp.close()
|
||
logger.info("已生成 B站 Netscape Cookie 文件: %s (条目: %d)", tmp.name, len(lines) - 1)
|
||
return tmp.name
|
||
|
||
def download(
|
||
self,
|
||
video_url: str,
|
||
output_dir: Union[str, None] = None,
|
||
quality: DownloadQuality = "fast",
|
||
need_video:Optional[bool]=False
|
||
) -> AudioDownloadResult:
|
||
if output_dir is None:
|
||
output_dir = get_data_dir()
|
||
if not output_dir:
|
||
output_dir=self.cache_data
|
||
os.makedirs(output_dir, exist_ok=True)
|
||
|
||
output_path = os.path.join(output_dir, "%(id)s.%(ext)s")
|
||
|
||
ydl_opts = {
|
||
**YDL_RETRY_OPTS,
|
||
'format': 'bestaudio[ext=m4a]/bestaudio/best',
|
||
'outtmpl': output_path,
|
||
'http_headers': {'Referer': 'https://www.bilibili.com'},
|
||
'postprocessors': [
|
||
{
|
||
'key': 'FFmpegExtractAudio',
|
||
'preferredcodec': 'mp3',
|
||
'preferredquality': '64',
|
||
}
|
||
],
|
||
'noplaylist': True,
|
||
'quiet': False,
|
||
}
|
||
if self._cookiefile:
|
||
ydl_opts['cookiefile'] = self._cookiefile
|
||
|
||
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
|
||
info = ydl.extract_info(video_url, download=True)
|
||
video_id = info.get("id")
|
||
title = info.get("title")
|
||
duration = info.get("duration", 0)
|
||
cover_url = info.get("thumbnail")
|
||
audio_path = os.path.join(output_dir, f"{video_id}.mp3")
|
||
|
||
return AudioDownloadResult(
|
||
file_path=audio_path,
|
||
title=title,
|
||
duration=duration,
|
||
cover_url=cover_url,
|
||
platform="bilibili",
|
||
video_id=video_id,
|
||
raw_info=info,
|
||
video_path=None # ❗音频下载不包含视频路径
|
||
)
|
||
|
||
def download_video(
|
||
self,
|
||
video_url: str,
|
||
output_dir: Union[str, None] = None,
|
||
) -> str:
|
||
"""
|
||
下载视频，返回视频文件路径
|
||
"""
|
||
|
||
if output_dir is None:
|
||
output_dir = get_data_dir()
|
||
os.makedirs(output_dir, exist_ok=True)
|
||
print("video_url",video_url)
|
||
video_id=extract_video_id(video_url, "bilibili")
|
||
video_path = os.path.join(output_dir, f"{video_id}.mp4")
|
||
if os.path.exists(video_path):
|
||
return video_path
|
||
|
||
# 检查是否已经存在
|
||
|
||
|
||
output_path = os.path.join(output_dir, "%(id)s.%(ext)s")
|
||
|
||
ydl_opts = {
|
||
**YDL_RETRY_OPTS,
|
||
'format': 'bv*[ext=mp4]/bestvideo+bestaudio/best',
|
||
'outtmpl': output_path,
|
||
'http_headers': {'Referer': 'https://www.bilibili.com'},
|
||
'noplaylist': True,
|
||
'quiet': False,
|
||
'merge_output_format': 'mp4', # 确保合并成 mp4
|
||
}
|
||
if self._cookiefile:
|
||
ydl_opts['cookiefile'] = self._cookiefile
|
||
|
||
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
|
||
info = ydl.extract_info(video_url, download=True)
|
||
video_id = info.get("id")
|
||
video_path = os.path.join(output_dir, f"{video_id}.mp4")
|
||
|
||
if not os.path.exists(video_path):
|
||
raise FileNotFoundError(f"视频文件未找到: {video_path}")
|
||
|
||
return video_path
|
||
|
||
def delete_video(self, video_path: str) -> str:
|
||
"""
|
||
删除视频文件
|
||
"""
|
||
if os.path.exists(video_path):
|
||
os.remove(video_path)
|
||
return f"视频文件已删除: {video_path}"
|
||
else:
|
||
return f"视频文件未找到: {video_path}"
|
||
|
||
def download_subtitles(self, video_url: str, output_dir: str = None,
|
||
langs: List[str] = None) -> Optional[TranscriptResult]:
|
||
"""
|
||
尝试获取B站视频字幕
|
||
|
||
:param video_url: 视频链接
|
||
:param output_dir: 输出路径
|
||
:param langs: 优先语言列表
|
||
:return: TranscriptResult 或 None
|
||
"""
|
||
# 1) 优先走 B 站官方 player API（直拉，无需下视频；AI 字幕需 SESSDATA cookie）
|
||
try:
|
||
result = BilibiliSubtitleFetcher().fetch_subtitles(video_url)
|
||
if result and result.segments:
|
||
return result
|
||
except Exception as e:
|
||
logger.warning(f"player API 直拉字幕异常，回退到 yt-dlp: {e}")
|
||
|
||
# 2) Fallback：原 yt-dlp 路径（更脆弱，遇到签名/Cookie 问题失败概率较高）
|
||
if output_dir is None:
|
||
output_dir = get_data_dir()
|
||
if not output_dir:
|
||
output_dir = self.cache_data
|
||
os.makedirs(output_dir, exist_ok=True)
|
||
|
||
if langs is None:
|
||
langs = ['zh-Hans', 'zh', 'zh-CN', 'ai-zh', 'en', 'en-US']
|
||
|
||
video_id = extract_video_id(video_url, "bilibili")
|
||
|
||
ydl_opts = {
|
||
**YDL_RETRY_OPTS,
|
||
'writesubtitles': True,
|
||
'writeautomaticsub': True,
|
||
'subtitleslangs': langs,
|
||
'subtitlesformat': 'srt/json3/best', # 支持多种格式
|
||
'skip_download': True,
|
||
'outtmpl': os.path.join(output_dir, f'{video_id}.%(ext)s'),
|
||
'quiet': True,
|
||
}
|
||
|
||
# 通过 CookieConfigManager 注入 B站 Cookie（Netscape cookiefile）
|
||
if self._cookiefile:
|
||
ydl_opts['cookiefile'] = self._cookiefile
|
||
ydl_opts['http_headers'] = {'Referer': 'https://www.bilibili.com'}
|
||
|
||
try:
|
||
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
|
||
info = ydl.extract_info(video_url, download=True)
|
||
|
||
# 查找下载的字幕文件
|
||
subtitles = info.get('requested_subtitles') or {}
|
||
if not subtitles:
|
||
logger.info(f"B站视频 {video_id} 没有可用字幕")
|
||
return None
|
||
|
||
# 按优先级查找字幕
|
||
detected_lang = None
|
||
sub_info = None
|
||
for lang in langs:
|
||
if lang in subtitles:
|
||
detected_lang = lang
|
||
sub_info = subtitles[lang]
|
||
break
|
||
|
||
# 如果按优先级没找到，取第一个可用的（排除弹幕）
|
||
if not detected_lang:
|
||
for lang, info_item in subtitles.items():
|
||
if lang != 'danmaku': # 排除弹幕
|
||
detected_lang = lang
|
||
sub_info = info_item
|
||
break
|
||
|
||
if not sub_info:
|
||
logger.info(f"B站视频 {video_id} 没有可用字幕（排除弹幕）")
|
||
return None
|
||
|
||
# 检查是否有内嵌数据（yt-dlp 有时直接返回字幕内容）
|
||
if 'data' in sub_info and sub_info['data']:
|
||
logger.info(f"直接从返回数据解析字幕: {detected_lang}")
|
||
return self._parse_srt_content(sub_info['data'], detected_lang)
|
||
|
||
# 查找字幕文件
|
||
ext = sub_info.get('ext', 'srt')
|
||
subtitle_file = os.path.join(output_dir, f"{video_id}.{detected_lang}.{ext}")
|
||
|
||
if not os.path.exists(subtitle_file):
|
||
logger.info(f"字幕文件不存在: {subtitle_file}")
|
||
return None
|
||
|
||
# 根据格式解析字幕文件
|
||
if ext == 'json3':
|
||
return self._parse_json3_subtitle(subtitle_file, detected_lang)
|
||
else:
|
||
with open(subtitle_file, 'r', encoding='utf-8') as f:
|
||
return self._parse_srt_content(f.read(), detected_lang)
|
||
|
||
except Exception as e:
|
||
logger.warning(f"获取B站字幕失败: {e}")
|
||
return None
|
||
|
||
def _parse_srt_content(self, srt_content: str, language: str) -> Optional[TranscriptResult]:
|
||
"""
|
||
解析 SRT 格式字幕内容
|
||
|
||
:param srt_content: SRT 字幕文本内容
|
||
:param language: 语言代码
|
||
:return: TranscriptResult
|
||
"""
|
||
import re
|
||
try:
|
||
segments = []
|
||
# SRT 格式: 序号\n时间戳\n文本\n\n
|
||
pattern = r'(\d+)\n(\d{2}:\d{2}:\d{2},\d{3})\s*-->\s*(\d{2}:\d{2}:\d{2},\d{3})\n(.*?)(?=\n\n|\n\d+\n|$)'
|
||
matches = re.findall(pattern, srt_content, re.DOTALL)
|
||
|
||
for match in matches:
|
||
idx, start_time, end_time, text = match
|
||
text = text.strip()
|
||
if not text:
|
||
continue
|
||
|
||
# 转换时间格式 00:00:00,000 -> 秒
|
||
def time_to_seconds(t):
|
||
parts = t.replace(',', '.').split(':')
|
||
return float(parts[0]) * 3600 + float(parts[1]) * 60 + float(parts[2])
|
||
|
||
segments.append(TranscriptSegment(
|
||
start=time_to_seconds(start_time),
|
||
end=time_to_seconds(end_time),
|
||
text=text
|
||
))
|
||
|
||
if not segments:
|
||
return None
|
||
|
||
full_text = ' '.join(seg.text for seg in segments)
|
||
logger.info(f"成功解析B站SRT字幕，共 {len(segments)} 段")
|
||
return TranscriptResult(
|
||
language=language,
|
||
full_text=full_text,
|
||
segments=segments,
|
||
raw={'source': 'bilibili_subtitle', 'format': 'srt'}
|
||
)
|
||
|
||
except Exception as e:
|
||
logger.warning(f"解析SRT字幕失败: {e}")
|
||
return None
|
||
|
||
def _parse_json3_subtitle(self, subtitle_file: str, language: str) -> Optional[TranscriptResult]:
|
||
"""
|
||
解析 json3 格式字幕文件
|
||
|
||
:param subtitle_file: 字幕文件路径
|
||
:param language: 语言代码
|
||
:return: TranscriptResult
|
||
"""
|
||
try:
|
||
with open(subtitle_file, 'r', encoding='utf-8') as f:
|
||
data = json.load(f)
|
||
|
||
segments = []
|
||
events = data.get('events', [])
|
||
|
||
for event in events:
|
||
# json3 格式中时间单位是毫秒
|
||
start_ms = event.get('tStartMs', 0)
|
||
duration_ms = event.get('dDurationMs', 0)
|
||
|
||
# 提取文本
|
||
segs = event.get('segs', [])
|
||
text = ''.join(seg.get('utf8', '') for seg in segs).strip()
|
||
|
||
if text: # 只添加非空文本
|
||
segments.append(TranscriptSegment(
|
||
start=start_ms / 1000.0,
|
||
end=(start_ms + duration_ms) / 1000.0,
|
||
text=text
|
||
))
|
||
|
||
if not segments:
|
||
return None
|
||
|
||
full_text = ' '.join(seg.text for seg in segments)
|
||
|
||
logger.info(f"成功解析B站字幕，共 {len(segments)} 段")
|
||
return TranscriptResult(
|
||
language=language,
|
||
full_text=full_text,
|
||
segments=segments,
|
||
raw={'source': 'bilibili_subtitle', 'file': subtitle_file}
|
||
)
|
||
|
||
except Exception as e:
|
||
logger.warning(f"解析字幕文件失败: {e}")
|
||
return None |