GithubBackup/BiliNote
1
0
Fork 0
mirror of https://github.com/JefferyHcool/BiliNote.git synced 2026-09-07 00:27:37 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(youtube): 使用 youtube-transcript-api 优先获取字幕，有字幕时跳过音频下载

Browse Source 
- 新增 YouTubeSubtitleFetcher 模块，通过 youtube-transcript-api 获取字幕
- 重构笔记生成流程：缓存 → 平台字幕 → 按需下载 → 转写 fallback
- 有字幕时仅提取视频元信息，不下载音视频文件
- 添加 youtube-transcript-api 依赖

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
huangjianwu
2026-03-23 17:31:30 +08:00
co-authored by Claude Opus 4.6
parent 5861ef4168
commit f4801d5be7
5 changed files with 203 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/downloaders/base.py
+2 -1
View File
@@ -22,7 +22,8 @@ class Downloader(ABC):
@abstractmethod
def download(self, video_url: str, output_dir: str = None,
quality: DownloadQuality = "fast", need_video: Optional[bool] = False) -> AudioDownloadResult:
quality: DownloadQuality = "fast", need_video: Optional[bool] = False,
skip_download: bool = False) -> AudioDownloadResult:
'''
:param need_video:
backend/app/downloaders/youtube_downloader.py
+21 -113
View File
@@ -1,5 +1,4 @@
import os
import json
import logging
from abc import ABC
from typing import Union, Optional, List
@@ -7,8 +6,9 @@ from typing import Union, Optional, List
import yt_dlp
from app.downloaders.base import Downloader, DownloadQuality
from app.downloaders.youtube_subtitle import YouTubeSubtitleFetcher
from app.models.notes_model import AudioDownloadResult
from app.models.transcriber_model import TranscriptResult, TranscriptSegment
from app.models.transcriber_model import TranscriptResult
from app.utils.path_helper import get_data_dir
from app.utils.url_parser import extract_video_id
@@ -25,12 +25,13 @@ class YoutubeDownloader(Downloader, ABC):
video_url: str,
output_dir: Union[str, None] = None,
quality: DownloadQuality = "fast",
need_video:Optional[bool]=False
need_video: Optional[bool] = False,
skip_download: bool = False,
) -> AudioDownloadResult:
if output_dir is None:
output_dir = get_data_dir()
if not output_dir:
output_dir=self.cache_data
output_dir = self.cache_data
os.makedirs(output_dir, exist_ok=True)
output_path = os.path.join(output_dir, "%(id)s.%(ext)s")
@@ -42,15 +43,17 @@ class YoutubeDownloader(Downloader, ABC):
'quiet': False,
}
if skip_download:
ydl_opts['skip_download'] = True
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
info = ydl.extract_info(video_url, download=True)
info = ydl.extract_info(video_url, download=not skip_download)
video_id = info.get("id")
title = info.get("title")
duration = info.get("duration", 0)
cover_url = info.get("thumbnail")
ext = info.get("ext", "m4a") # 兜底用 m4a
ext = info.get("ext", "m4a")
audio_path = os.path.join(output_dir, f"{video_id}.{ext}")
print('os.path.join(output_dir, f"{video_id}.{ext}")',os.path.join(output_dir, f"{video_id}.{ext}"))
return AudioDownloadResult(
file_path=audio_path,
@@ -59,8 +62,8 @@ class YoutubeDownloader(Downloader, ABC):
cover_url=cover_url,
platform="youtube",
video_id=video_id,
raw_info={'tags':info.get('tags')}, #全部返回会报错
video_path=None # ❗音频下载不包含视频路径
raw_info={'tags': info.get('tags')},
video_path=None,
)
def download_video(
@@ -101,115 +104,20 @@ class YoutubeDownloader(Downloader, ABC):
def download_subtitles(self, video_url: str, output_dir: str = None,
langs: List[str] = None) -> Optional[TranscriptResult]:
"""
尝试获取YouTube视频字幕（优先人工字幕，其次自动生成）
通过 YouTube InnerTube API 直接获取字幕（优先人工字幕，其次自动生成）。
比 yt_dlp 方式更轻量，无需写临时文件到磁盘。
:param video_url: 视频链接
:param output_dir: 输出路径
:param output_dir: 未使用（保留接口兼容）
:param langs: 优先语言列表
:return: TranscriptResult 或 None
"""
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', 'zh-TW', 'en', 'en-US']
langs = ['zh-Hans', 'zh', 'zh-CN', 'zh-TW', 'en', 'en-US', 'ja']
video_id = extract_video_id(video_url, "youtube")
ydl_opts = {
'writesubtitles': True,
'writeautomaticsub': True,
'subtitleslangs': langs,
'subtitlesformat': 'json3',
'skip_download': True,
'outtmpl': os.path.join(output_dir, f'{video_id}.%(ext)s'),
'quiet': True,
}
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"YouTube视频 {video_id} 没有可用字幕")
return None
# 按优先级查找字幕文件
subtitle_file = None
detected_lang = None
for lang in langs:
if lang in subtitles:
subtitle_file = os.path.join(output_dir, f"{video_id}.{lang}.json3")
detected_lang = lang
break
# 如果按优先级没找到，取第一个可用的
if not subtitle_file:
for lang, sub_info in subtitles.items():
subtitle_file = os.path.join(output_dir, f"{video_id}.{lang}.json3")
detected_lang = lang
break
if not subtitle_file or not os.path.exists(subtitle_file):
logger.info(f"字幕文件不存在: {subtitle_file}")
return None
# 解析字幕文件
return self._parse_json3_subtitle(subtitle_file, detected_lang)
except Exception as e:
logger.warning(f"获取YouTube字幕失败: {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"成功解析YouTube字幕，共 {len(segments)} 段")
return TranscriptResult(
language=language,
full_text=full_text,
segments=segments,
raw={'source': 'youtube_subtitle', 'file': subtitle_file}
)
except Exception as e:
logger.warning(f"解析字幕文件失败: {e}")
return None
fetcher = YouTubeSubtitleFetcher()
print(
f"尝试获取字幕，video_id={video_id}, langs={langs}"
)
return fetcher.fetch_subtitles(video_id, langs)
backend/app/downloaders/youtube_subtitle.py
+98
View File
@@ -0,0 +1,98 @@
"""
通过 youtube-transcript-api 获取 YouTube 字幕。
优先人工字幕，其次自动生成字幕。不依赖 yt_dlp，无需下载任何文件。
"""
from typing import Optional, List
from youtube_transcript_api import YouTubeTranscriptApi
from app.models.transcriber_model import TranscriptResult, TranscriptSegment
from app.utils.logger import get_logger
logger = get_logger(__name__)
class YouTubeSubtitleFetcher:
"""通过 youtube-transcript-api 获取 YouTube 字幕。"""
def __init__(self):
self._api = YouTubeTranscriptApi()
def fetch_subtitles(
self,
video_id: str,
langs: Optional[List[str]] = None,
) -> Optional[TranscriptResult]:
if langs is None:
langs = ["zh-Hans", "zh", "zh-CN", "zh-TW", "en", "en-US", "ja"]
try:
# 1. 列出所有可用字幕
transcript_list = self._api.list(video_id)
available = []
for t in transcript_list:
available.append(
f"{t.language_code}({'auto' if t.is_generated else 'manual'})"
)
logger.info(f"可用字幕轨道: {', '.join(available)}")
# 2. 按优先级查找：先人工字幕，再自动字幕
transcript = None
try:
transcript = transcript_list.find_manually_created_transcript(langs)
logger.info(f"选中人工字幕: {transcript.language_code} ({transcript.language})")
except Exception:
try:
transcript = transcript_list.find_generated_transcript(langs)
logger.info(f"选中自动字幕: {transcript.language_code} ({transcript.language})")
except Exception:
# 都没匹配，取第一个可用的
for t in transcript_list:
transcript = t
source = "auto" if t.is_generated else "manual"
logger.info(f"使用首个可用字幕: {t.language_code} ({source})")
break
if not transcript:
logger.info(f"YouTube 视频 {video_id} 没有任何可用字幕")
return None
# 3. 获取字幕内容
fetched = transcript.fetch()
segments = []
for snippet in fetched:
text = snippet.get("text", "").strip() if isinstance(snippet, dict) else str(snippet).strip()
if not text:
continue
start = snippet.get("start", 0) if isinstance(snippet, dict) else 0
duration = snippet.get("duration", 0) if isinstance(snippet, dict) else 0
segments.append(TranscriptSegment(
start=float(start),
end=float(start) + float(duration),
text=text,
))
if not segments:
logger.warning(f"YouTube 字幕内容为空: {video_id}")
return None
full_text = " ".join(seg.text for seg in segments)
logger.info(f"成功获取 YouTube 字幕，共 {len(segments)} 段")
return TranscriptResult(
language=transcript.language_code,
full_text=full_text,
segments=segments,
raw={
"source": "youtube_transcript_api",
"language": transcript.language,
"language_code": transcript.language_code,
"is_generated": transcript.is_generated,
},
)
except Exception as e:
logger.warning(f"YouTube 字幕获取失败: {e}")
return None