Skip to content

章节梗概 (5/9)：缓存层 timeline 落盘与读回 #19

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zj1123581321 merged 2 commits into from
Jul 21, 2026
Merged

章节梗概 (5/9)：缓存层 timeline 落盘与读回 #19

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
54e0f19
接通 CapsWriter/YouTube timeline 落盘与 get_cache 读回
zj1123581321 Jul 19, 2026
4a38572
修复 timeline 接线 4 处 CI review 发现：覆盖写清旧侧车、空字幕正文从 segments 重建、无效时间不再崩侧车、…
zj1123581321 Jul 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2,324 changes: 1,201 additions & 1,123 deletions src/video_transcript_api/api/services/transcription.py
View file
Open in desktop

Large diffs are not rendered by default.

3,368 changes: 1,703 additions & 1,665 deletions src/video_transcript_api/cache/cache_manager.py
View file
Open in desktop

Large diffs are not rendered by default.

118 changes: 27 additions & 91 deletions src/video_transcript_api/downloaders/youtube.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,7 @@ def fetch_for_transcription(self, url: str, use_speaker_recognition: bool = Fals
- description: str
- platform: "youtube"
- transcript: str | None (字幕文本，已解析为纯文本)
- transcript_segments: list | None (SRT 时间戳分段；无则 None)
- audio_path: str | None (本地音频文件路径)
- need_transcription: bool (是否需要调用转录服务)

Expand Down Expand Up @@ -129,6 +130,8 @@ def fetch_for_transcription(self, url: str, use_speaker_recognition: bool = Fals
description = video_info.description if video_info else ""

transcript = None
# SRT 解析出的时间戳分段（有则落盘，无则诚实降级为 None）
transcript_segments = None
audio_path = None
need_transcription = False

Expand All @@ -147,13 +150,16 @@ def fetch_for_transcription(self, url: str, use_speaker_recognition: bool = Fals
else:
# 优先使用字幕
if result.has_transcript and not result.audio_fallback and result.transcript:
# 有字幕，下载并解析
# 有字幕，下载并解析（保留 segments 供 timeline 落盘）
srt_content = self._youtube_api_client.download_content(result.transcript.url)
transcript = YouTubeApiClient.parse_srt_to_text(srt_content)
subtitle_result = YouTubeApiClient.parse_srt_to_subtitle_result(srt_content)
transcript = subtitle_result.text
transcript_segments = subtitle_result.segments
need_transcription = False
logger.info(
f"[youtube] Transcript downloaded and parsed, "
f"length={len(transcript)} chars"
f"length={len(transcript)} chars, "
f"segments={len(transcript_segments) if transcript_segments else 0}"
)
elif result.audio and result.audio.url:
# 无字幕，下载 fallback 的音频
Expand All @@ -174,6 +180,7 @@ def fetch_for_transcription(self, url: str, use_speaker_recognition: bool = Fals
"description": description,
"platform": "youtube",
"transcript": transcript,
"transcript_segments": transcript_segments,
"audio_path": audio_path,
"need_transcription": need_transcription,
}
Expand Down Expand Up @@ -400,105 +407,34 @@ def get_video_info(self, url):

def get_subtitle(self, url):
"""
获取字幕，按优先级策略执行
获取字幕纯文本（向后兼容入口）。

优先级策略：
- 如果启用 youtube_api_server：
1. youtube_api_server（绕过本地 IP 封禁）
2. TikHub API（直接跳转，跳过本地方案）
- 如果未启用 youtube_api_server：
1. 本地 youtube-transcript-api
2. TikHub API（备用）
薄委托 get_subtitle_result()：统一走同一套优先级 / 回退策略，
只取其中的 text 字段。需要时间戳分段的调用方应直接使用
get_subtitle_result()。

参数:
url: 视频URL

返回:
str: 字幕文本，如果有的话
str | None: 字幕文本；无字幕时返回 None
"""
try:
video_id = self._extract_video_id(url)
if not video_id:
logger.warning("[字幕获取] 无法提取视频ID")
return None

# ============================================================
# 分支 A：启用了 youtube_api_server
# ============================================================
if self.use_api_server:
logger.info(
f"[字幕获取] 使用 youtube_api_server 优先策略: video_id={video_id}"
)

# 尝试通过 API Server 获取字幕
try:
transcript = self._youtube_api_client.fetch_transcript(video_id)
if transcript and transcript.strip():
logger.info(
f"[字幕获取] youtube_api_server 成功: "
f"length={len(transcript)} chars"
)
return transcript
else:
# 返回 None 或空字符串 = 视频没有字幕，不需要重试
logger.info(
f"[字幕获取] 视频没有可用字幕（已由 API Server 确认）: {video_id}"
)
return None
except Exception as api_error:
# 只有失败（异常）时才回退到 TikHub
logger.warning(
f"[字幕获取] youtube_api_server 失败: {api_error}, "
f"回退到 TikHub API（跳过本地方案）"
)
return self._get_subtitle_with_tikhub_api(url)

# ============================================================
# 分支 B：未启用 youtube_api_server，使用本地方案
# ============================================================
else:
logger.info(
f"[字幕获取] 使用本地方案: video_id={video_id}"
)

# 首先尝试使用 youtube-transcript-api
transcript = self._fetch_youtube_transcript(video_id)

if transcript and transcript.strip():
# 检查是否是IP被阻止的标记
if transcript == "IP_BLOCKED":
logger.warning(
f"[字幕获取] 本地方案 IP 被封，回退到 TikHub API: {video_id}"
)
return self._get_subtitle_with_tikhub_api(url)
elif transcript == "TRANSCRIPTS_DISABLED":
logger.info(f"[字幕获取] 视频字幕已被禁用: {video_id}")
return None
else:
logger.info(
f"[字幕获取] 本地方案成功: length={len(transcript)} chars"
)
return transcript

# 如果 youtube-transcript-api 失败，尝试使用 TikHub API
logger.info(
f"[字幕获取] 本地方案失败，回退到 TikHub API: {video_id}"
)
return self._get_subtitle_with_tikhub_api(url)

except Exception as e:
logger.exception(f"[字幕获取] 异常: {str(e)}")
return None
result = self.get_subtitle_result(url)
return result.text if result else None

def get_subtitle_result(self, url):
"""
获取字幕，保留时间戳分段信息（get_subtitle 的完整版，供后续接线使用）
获取字幕，保留时间戳分段信息（权威入口）。

优先级策略：
- 如果启用 youtube_api_server：
1. youtube_api_server（绕过本地 IP 封禁）
2. TikHub API（直接跳转，跳过本地方案）
- 如果未启用 youtube_api_server：
1. 本地 youtube-transcript-api
2. TikHub API（备用）

行为策略与 get_subtitle 完全一致（同样的优先级、同样的回退顺序），
唯一区别是成功时返回 SubtitleResult（text + 可选 segments）而不是纯
文本。get_subtitle 出于向后兼容需要保持字符串返回值不变，因此这里
单独实现一份平行的分支逻辑，而不是让 get_subtitle 反过来委托本方法
（否则 mock 了 get_subtitle 内部旧方法名的既有测试会失效）。
get_subtitle() 薄委托本方法并只返回 text，避免维护两套平行分支。

参数:
url: 视频URL
Expand Down
93 changes: 75 additions & 18 deletions src/video_transcript_api/transcriber/capswriter_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
import sys
import json
import base64
import math
import time
import asyncio
import re
Expand All @@ -25,6 +26,8 @@
# 添加项目根目录到系统路径
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from ..utils.logging import load_config
# 时间解析唯一权威：禁止在本模块另起一套 isfinite/parse 逻辑
from .segments import parse_time_to_seconds


class Config:
Expand Down Expand Up @@ -197,8 +200,25 @@ def _optimize_segment_lengths(
return final


def _finite_time_or_none(value: Any) -> Optional[float]:
"""把任意时间值规整为有限 float，否则 None。

复用 transcriber.segments.parse_time_to_seconds（唯一权威时间解析），
其内部已用 math.isfinite 拒绝 NaN/Inf；本函数再对已是 float 的路径
做一层防御，避免上游直接传入非有限值时写出 NaN/Inf 到下游 JSON。
"""
parsed = parse_time_to_seconds(value)
if parsed is None:
return None
return parsed if math.isfinite(parsed) else None


def _split_long_segment(segment: Dict[str, Any], max_len: int) -> List[Dict[str, Any]]:
"""在次级标点处分割超长句子"""
"""在次级标点处分割超长句子。

时间插值必须拒绝非有限值：start_time / duration 任一非有限时，诚实
降级为 start_time=end_time=None，文本照常切分、永不丢字。
"""
text = segment["text"]
secondary_punct = r"([，,；;])"
parts = re.split(secondary_punct, text)
Expand All @@ -208,37 +228,65 @@ def _split_long_segment(segment: Dict[str, Any], max_len: int) -> List[Dict[str,

split_segments = []
current = ""
start_time = segment["start_time"]
duration = segment["end_time"] - segment["start_time"]
orig_start = _finite_time_or_none(segment.get("start_time"))
orig_end = _finite_time_or_none(segment.get("end_time"))
# 仅当两端时间都有限且区间非倒挂时才做线性插值；否则全程 None。
# end < start 的倒挂区间（上游 duration 为负、或时间轴书写顺序颠倒）
# 不是可用时间轴：对它插值会生成递减的时间序列，比"没有时间"更误导
# 下游——与 sanitize_time_pair 同一口径，文本照常保留、时间诚实降级。
times_usable = (
orig_start is not None
and orig_end is not None
and math.isfinite(orig_end - orig_start)
and orig_end >= orig_start
)
start_time = orig_start if times_usable else None
duration = (orig_end - orig_start) if times_usable else None
total_len = segment["length"]

for part in parts:
if len(current + part) <= max_len:
current += part
else:
if current:
progress = len(current) / total_len
end_time = start_time + duration * progress
if times_usable and total_len > 0 and start_time is not None:
progress = len(current) / total_len
end_time = start_time + duration * progress
if not math.isfinite(end_time):
end_time = None
start_for_seg = None
else:
start_for_seg = round(start_time, 2)
end_time = round(end_time, 2)
start_time = end_time
else:
start_for_seg = None
end_time = None

split_segments.append(
{
"start_time": round(start_time, 2),
"end_time": round(end_time, 2),
"start_time": start_for_seg,
"end_time": end_time,
"text": current,
"length": len(current),
}
)

start_time = end_time
current = part
else:
current = part

if current:
if times_usable and start_time is not None and orig_end is not None:
final_start = round(start_time, 2) if math.isfinite(start_time) else None
final_end = round(orig_end, 2) if math.isfinite(orig_end) else None
else:
final_start = None
final_end = None
split_segments.append(
{
"start_time": round(start_time, 2),
"end_time": round(segment["end_time"], 2),
"start_time": final_start,
"end_time": final_end,
"text": current,
"length": len(current),
}
Expand Down Expand Up @@ -325,21 +373,24 @@ def _create_segments_from_capswriter(
start_token_idx = max(0, min(start_token_idx, len(timestamps) - 1))
end_token_idx = max(0, min(end_token_idx, len(timestamps) - 1))

# 提取时间
start_time = timestamps[start_token_idx]
end_time = timestamps[end_token_idx]
# 提取时间：NaN/Inf/缺失（None）等无效值经唯一权威解析统一降级为
# None（与 _split_long_segment 的诚实降级同一口径）——文本永不丢失，
# 时间宁可为 None 也不能让 round(None)/round(nan) 之类的异常或
# NaN/Inf 字面量污染整个 FunASR 兼容侧车的生成。
start_time = _finite_time_or_none(timestamps[start_token_idx])
end_time = _finite_time_or_none(timestamps[end_token_idx])

segments.append(
{
"start_time": round(start_time, 2),
"end_time": round(end_time, 2),
"start_time": round(start_time, 2) if start_time is not None else None,
"end_time": round(end_time, 2) if end_time is not None else None,
"text": sentence,
"length": len(sentence),
}
)

logger.debug(
f"句子 {idx + 1}: {sentence_len} 字符 -> tokens[{start_token_idx}:{end_token_idx}] -> {start_time:.2f}s-{end_time:.2f}s"
f"句子 {idx + 1}: {sentence_len} 字符 -> tokens[{start_token_idx}:{end_token_idx}] -> {start_time}s-{end_time}s"
)

char_offset += sentence_len
Expand Down Expand Up @@ -735,9 +786,15 @@ async def _save_results(
"error": None,
}

# 统计信息
# 统计信息：无效时间已降级为 None 的 segment（见
# _create_segments_from_capswriter / _split_long_segment
# 的诚实降级）没有可统计的时长，直接跳过——不能无条件做
# end - start 算术，否则 None - None 抛 TypeError，整个
# FunASR 兼容侧车会被 except 吞掉、跳过生成。
total_duration = sum(
seg["end_time"] - seg["start_time"] for seg in segments
seg["end_time"] - seg["start_time"]
for seg in segments
if seg["start_time"] is not None and seg["end_time"] is not None
)
avg_length = (
sum(len(seg["text"]) for seg in segments) / len(segments)
Expand Down
39 changes: 27 additions & 12 deletions tests/unit/downloaders/test_youtube_get_subtitle_result.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -231,22 +231,37 @@ def test_get_subtitle_end_to_end_unchanged_while_get_subtitle_result_carries_seg
]


def test_get_subtitle_api_server_all_numeric_srt_still_returns_none_unchanged():
"""Locks the historical str-only semantics of get_subtitle() (API Server
branch): it calls YouTubeApiClient.fetch_transcript() (str-returning,
driven by parse_srt_to_text -- a completely separate code path from
fetch_transcript_result()/parse_srt_to_subtitle_result()). For an
all-numeric-body SRT, parse_srt_to_text() has always returned "" (each
body line is mistaken for the next cue's index line), so get_subtitle()
must keep returning None exactly as before -- this fix only changes
get_subtitle_result()'s validity check, never get_subtitle()'s."""
def test_get_subtitle_api_server_all_numeric_srt_empty_text_with_segments():
"""get_subtitle() now thin-delegates to get_subtitle_result().

When the API Server path returns a SubtitleResult with empty text but
non-empty segments (all-numeric-body SRT edge case), get_subtitle_result
treats it as valid subtitle and get_subtitle returns result.text ("").
When both text and segments are empty/None, the result is treated as no
subtitle and get_subtitle returns None -- no TikHub fallback.
"""
downloader = _make_downloader()
downloader.config["youtube_api_server"] = {"enabled": True, "base_url": "http://x", "api_key": "k"}
downloader._youtube_api_client = Mock()
downloader._youtube_api_client.fetch_transcript = Mock(return_value="")
downloader._get_subtitle_with_tikhub_api = Mock()
# Empty text + no segments => no subtitle
downloader._youtube_api_client.fetch_transcript_result = Mock(
return_value=SubtitleResult(text="", segments=None)
)
downloader._get_subtitle_result_with_tikhub_api = Mock()

result = downloader.get_subtitle("https://www.youtube.com/watch?v=test")

assert result is None
assert not downloader._get_subtitle_with_tikhub_api.called
assert not downloader._get_subtitle_result_with_tikhub_api.called

# Empty text + segments => get_subtitle returns the empty text string
# (delegation returns result.text; callers that need timeline should use
# get_subtitle_result directly).
downloader._youtube_api_client.fetch_transcript_result = Mock(
return_value=SubtitleResult(
text="",
segments=[{"start_time": 1.0, "end_time": 2.0, "text": "42"}],
)
)
result_with_segments = downloader.get_subtitle("https://www.youtube.com/watch?v=test")
assert result_with_segments == ""
Loading
Loading