diff --git a/CHANGELOG.md b/CHANGELOG.md index 358494a6..727ab769 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ ### Fixed +- 章节笔记：章节目录产物现在记录实际锚定的数据来源（缓存结构化对话 / 原始转写分片），笔记按同一来源精确复现。此前目录标成「原始分片」却实际锚在缓存对话上，笔记必然指纹校验失败且无法生成（生产事故：BV1fcHD6cEw3）。标签声明的来源已不存在时直接报错，不再默默换成另一份数据。 - 章节目录：模型按主题弧线而非时间序返回章节时（每章时间锚点正确），章节层不再整体失败，改为按时间排序归一化（排序改变顺序时记入日志，含错位数与 LIS 统计）。此前此类输出会导致整个视频无章节目录（生产事故：BV1fcHD6cEw3）。 - 章节生成失败原因现在落盘到任务状态的 llm_status.json（chapters_error 字段），不再只出现在服务日志里；补跑成功后旧失败原因自动清除。 diff --git a/src/video_transcript_api/api/services/llm_ops.py b/src/video_transcript_api/api/services/llm_ops.py index c6d709a5..ad91049e 100644 --- a/src/video_transcript_api/api/services/llm_ops.py +++ b/src/video_transcript_api/api/services/llm_ops.py @@ -732,6 +732,7 @@ def _handle_llm_task(llm_task: dict): speaker_count_hint=llm_task.get("cached_speaker_count"), skip_chapters=skip_chapters_for_coordinator, timeline_segments=timeline_segments_seed, + timeline_segments_kind=chapters_seed_kind, ) # 适配返回格式 diff --git a/src/video_transcript_api/llm/coordinator.py b/src/video_transcript_api/llm/coordinator.py index 9b7356bd..1072c6c2 100644 --- a/src/video_transcript_api/llm/coordinator.py +++ b/src/video_transcript_api/llm/coordinator.py @@ -141,6 +141,7 @@ def process( contradiction_scan: Optional[bool] = None, skip_chapters: bool = False, timeline_segments: Optional[List[Dict]] = None, + timeline_segments_kind: Optional[str] = None, ) -> Dict: """处理文本（统一入口） @@ -173,6 +174,10 @@ def process( timeline_segments: 章节输入（dialogs 或原始 segments 列表）。由 llm_ops 按输入梯度解析后传入，保持 llm/ 包不依赖 cache 读盘。 None/空时 processor 会诚实返回 SKIPPED_NO_TIMELINE。 + timeline_segments_kind: timeline_segments 的真实来源标签（调用方 + 已解析的事实，如 ``cached_dialogs`` / ``segments``）。标签会原样 + 写进 llm_chapters.json 的 source.kind，notes reader 据此精确复现 + 同一份锚点数据，缺失才默认按原始 segments 形态处理。 Returns: 处理结果字典: @@ -314,7 +319,14 @@ def process( else: logger.info("Step 3/3: Chapters generation") chapters_input = timeline_segments - chapters_source_kind = "segments" if timeline_segments else "none" + # notes_anchor_kind_truth: source.kind 必须是「实际喂给 chapters 的 + # 锚点数据」的真实来源，而不是「入参长什么样」。调用方 llm_ops 按输入 + # 梯度解析种子时已经知道它是缓存 dialogs 还是原始 segments，这里直接 + # 采用调用方的事实；本轮 structured dialogs 覆盖种子时改写为 dialogs。 + # 两者都是真实锚点，notes reader 才能据此精确复现（#227）。 + chapters_source_kind = ( + (timeline_segments_kind or "segments") if timeline_segments else "none" + ) structured_for_chapters = calibration_result.get("structured_data") if isinstance(structured_for_chapters, dict): dialogs = structured_for_chapters.get("dialogs") diff --git a/src/video_transcript_api/llm/processors/notes_processor.py b/src/video_transcript_api/llm/processors/notes_processor.py index a4db9a67..66a7c034 100644 --- a/src/video_transcript_api/llm/processors/notes_processor.py +++ b/src/video_transcript_api/llm/processors/notes_processor.py @@ -117,28 +117,39 @@ def load_notes_source_segments( ``dialogs``/``cached_dialogs`` sources use ``llm_processed.json``. A ``segments`` source uses the shared ``load_segments`` adapter, preserving the plain-text path's normalized list and original index order. + + A declared ``source_kind`` is reproduced exactly: if the cache no longer + holds that anchor, the lookup fails as ``(None, None)`` instead of silently + substituting the other source, which would make notes hash a different + anchor than chapters did (#227). Only a product carrying no label at all + falls back to probing ``dialogs`` then ``segments``. """ cache_path = Path(cache_dir) normalized_kind = (source_kind or "").lower() - wants_dialogs = normalized_kind in { + + processed = _read_json_file(cache_path / "llm_processed.json") + dialogs = processed.get("dialogs") if processed else None + dialogs_source = dialogs if isinstance(dialogs, list) and dialogs else None + + if normalized_kind in { "dialogs", "cached_dialogs", "structured", "plain_structured", - } - wants_segments = normalized_kind in {"segments", "timeline"} + }: + return (dialogs_source, "dialogs") if dialogs_source is not None else (None, None) - processed = _read_json_file(cache_path / "llm_processed.json") - dialogs = processed.get("dialogs") if processed else None - if (wants_dialogs or not wants_segments) and isinstance(dialogs, list) and dialogs: - return dialogs, "dialogs" + if normalized_kind in {"segments", "timeline"}: + segments = load_segments(cache_path) + if isinstance(segments, list) and segments: + return segments, "segments" + return None, None + if dialogs_source is not None: + return dialogs_source, "dialogs" segments = load_segments(cache_path) if isinstance(segments, list) and segments: return segments, "segments" - - if not wants_segments and isinstance(dialogs, list) and dialogs: - return dialogs, "dialogs" return None, None diff --git a/tests/unit/test_notes_anchor_source.py b/tests/unit/test_notes_anchor_source.py new file mode 100644 index 00000000..019f5f58 --- /dev/null +++ b/tests/unit/test_notes_anchor_source.py @@ -0,0 +1,398 @@ +"""Chapter anchor source truth: the writer label must reproduce the reader's load. + +Production shape reproduced here (#227): a cache directory holding BOTH the +structured ``llm_processed.json`` dialogs and the raw ``transcript_funasr.json`` +segments, where a chapters-only re-generation anchors on the cached dialogs. + +Locked invariant: the anchor fingerprint stored in a ``chapters_status=GENERATED`` +product equals the fingerprint the notes reader recomputes from that same cache +directory under the repaired resolution rule. A label recording only "what the +seed input looked like" sends the reader to the other source -- exactly the +production failure (product 518 anchors / reader 556 segments). +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any, Dict, List +from unittest.mock import MagicMock + +import pytest + +from video_transcript_api.api.services import llm_ops +from video_transcript_api.llm.coordinator import LLMCoordinator +from video_transcript_api.llm.core.config import LLMConfig +from video_transcript_api.llm.processors.chapters_processor import ChaptersProcessor +from video_transcript_api.llm.processors.notes_processor import ( + NotesProcessor, + compute_notes_anchor_fingerprint, + load_notes_source_segments, +) +from video_transcript_api.transcriber.segments import load_segments +from video_transcript_api.utils.llm_status import NotesStatus + +# Dialog text must clear ChaptersProcessor.min_chapters_threshold (set low in +# _config below); the two sources deliberately differ in length and text so a +# mislabeled anchor cannot reproduce the other source's fingerprint. +def _hhmmss(total_seconds: int) -> str: + return "%02d:%02d:%02d" % ( + total_seconds // 3600, + (total_seconds % 3600) // 60, + total_seconds % 60, + ) + + +DIALOGS: List[Dict[str, Any]] = [ + { + "speaker": "Speaker1", + "start_time": _hhmmss(index * 60), + "end_time": _hhmmss(index * 60 + 45), + "text": f"cached dialog line {index} " + "dialog padding words. " * 12, + } + for index in range(12) +] + +SEGMENTS: List[Dict[str, Any]] = [ + { + "start_time": float(index * 8), + "end_time": float(index * 8 + 7), + "text": f"raw funasr segment {index} " + "raw padding words. " * 9, + "speaker": f"spk{index % 2}", + } + for index in range(17) +] + + +class FakeNotesClient: + """Return scripted chapter notes without any network access.""" + + def __init__(self, responses): + self.responses = list(responses) + self.calls: List[Dict[str, Any]] = [] + + def call(self, **kwargs): + self.calls.append(kwargs) + return self.responses.pop(0) + + +class FakeChaptersClient: + """Answer the chapters LLM call with a fixed, valid chapter outline.""" + + def __init__(self, chapters): + self.chapters = chapters + self.calls: List[Dict[str, Any]] = [] + + def call(self, **kwargs): + self.calls.append(kwargs) + response = MagicMock() + response.structured_output = {"chapters": self.chapters} + return response + + +def _notes_config() -> LLMConfig: + return LLMConfig( + api_key="test-key", + base_url="http://localhost", + calibrate_model="calibrate-model", + summary_model="summary-model", + notes_model="notes-model", + notes_concurrency=1, + ) + + +def _coordinator_config() -> Dict[str, Any]: + return { + "llm": { + "api_key": "test-key", + "base_url": "http://test.invalid", + "calibrate_model": "calibrate-model", + "summary_model": "summary-model", + "min_summary_threshold": 1, + "min_chapters_threshold": 1, + "structured_calibration": {"min_chunk_length": 1}, + } + } + + +def _write_production_cache(cache_dir: Path) -> None: + """Write the two coexisting anchor sources exactly as production does.""" + (cache_dir / "llm_processed.json").write_text( + json.dumps({"dialogs": DIALOGS}, ensure_ascii=False), encoding="utf-8" + ) + (cache_dir / "transcript_funasr.json").write_text( + json.dumps({"segments": SEGMENTS}, ensure_ascii=False), encoding="utf-8" + ) + + +class _FileBackedCacheManager: + """Minimal get_cache that reads the real cache files, as production does.""" + + def __init__(self, cache_dir: Path): + self.cache_dir = cache_dir + + def get_cache(self, platform, media_id, use_speaker_recognition=False): + processed = json.loads( + (self.cache_dir / "llm_processed.json").read_text(encoding="utf-8") + ) + return {"llm_processed": processed} + + +CHAPTER_OUTLINE = [ + {"title": "Opening", "gist": "intro", "start_seg": 0}, + {"title": "Middle", "gist": "core", "start_seg": 5}, + {"title": "Closing", "gist": "outro", "start_seg": 10}, +] + + +def _run_chapters_round( + coordinator: LLMCoordinator, + *, + seed_segments, + seed_kind: str, +) -> Dict[str, Any]: + """Run the real coordinator chapters stage with calibration suppressed. + + Calibration suppression is the production trigger: the re-generation ran + chapters only, so ``structured_data`` is absent and the anchor is whatever + seed llm_ops resolved from cache. + """ + coordinator._route_to_calibration_processor = MagicMock( + return_value={ + "calibrated_text": "cached calibrated text", + "key_info": {}, + "stats": {"calibration_status": "disabled"}, + "structured_data": None, + } + ) + return coordinator.process( + content="", + title="Anchor source video", + skip_summary=True, + skip_calibration=True, + timeline_segments=seed_segments, + timeline_segments_kind=seed_kind, + ) + + +def _write_chapters_product(cache_dir: Path, result: Dict[str, Any]) -> Dict[str, Any]: + """Persist llm_chapters.json with the same field mapping llm_ops uses.""" + stats = result["stats"] + payload = { + "format_version": "v1", + "source": { + "kind": stats["chapters_source_kind"], + "segment_count": stats["chapters_segment_count"], + "fingerprint": stats["chapters_fingerprint"], + "generated_at": "2026-10-10T15:12:30+00:00", + }, + "chapters": stats["chapters"], + } + (cache_dir / "llm_chapters.json").write_text( + json.dumps(payload, ensure_ascii=False), encoding="utf-8" + ) + return payload + + +def _resolve_production_seed(monkeypatch, cache_dir: Path): + """Run llm_ops' real input gradient against the two coexisting sources.""" + monkeypatch.setattr(llm_ops, "cache_manager", _FileBackedCacheManager(cache_dir)) + return llm_ops._resolve_chapters_timeline_segments( + llm_task={}, + platform="bilibili", + media_id="BV1fcHD6cEw3", + use_speaker_recognition=False, + ) + + +def _produce_chapters_product(monkeypatch, cache_dir: Path) -> Dict[str, Any]: + """Full chapters-only round on a production-shaped cache; returns the product.""" + _write_production_cache(cache_dir) + seed, seed_kind = _resolve_production_seed(monkeypatch, cache_dir) + + coordinator = LLMCoordinator(_coordinator_config(), str(cache_dir)) + coordinator.chapters_processor = ChaptersProcessor( + llm_client=FakeChaptersClient(CHAPTER_OUTLINE), + config=LLMConfig.from_dict(_coordinator_config()), + ) + result = _run_chapters_round( + coordinator, seed_segments=seed, seed_kind=seed_kind + ) + return _write_chapters_product(cache_dir, result) + + +class TestCachedDialogsSeedAnchorsNotesReader: + """Production scenario: cached dialogs seed, both sources coexist.""" + + def test_seed_resolution_prefers_cached_dialogs(self, monkeypatch, tmp_path): + _write_production_cache(tmp_path) + seed, seed_kind = _resolve_production_seed(monkeypatch, tmp_path) + + assert seed_kind == "cached_dialogs" + assert seed == DIALOGS + + def test_product_label_reloads_the_same_anchor_and_notes_succeed( + self, monkeypatch, tmp_path + ): + payload = _produce_chapters_product(monkeypatch, tmp_path) + source = payload["source"] + + # The writer must name the source it actually anchored on, and the two + # available sources must be distinguishable (otherwise this test would + # pass no matter which one the reader picked). + assert source["kind"] == "cached_dialogs" + assert source["segment_count"] == len(DIALOGS) + assert len(load_segments(tmp_path)) != len(DIALOGS) + assert compute_notes_anchor_fingerprint(load_segments(tmp_path)) != ( + source["fingerprint"] + ) + + anchor, anchor_kind = load_notes_source_segments(tmp_path, source["kind"]) + assert anchor_kind == "dialogs" + assert anchor == DIALOGS + assert compute_notes_anchor_fingerprint(anchor) == source["fingerprint"] + + client = FakeNotesClient(["notes one", "notes two", "notes three"]) + notes_result = NotesProcessor(client, _notes_config()).process( + cache_dir=tmp_path + ) + + assert notes_result.status is NotesStatus.GENERATED, notes_result.error + assert notes_result.text + assert notes_result.fingerprint == source["fingerprint"] + assert len(client.calls) == 3 + + def test_handle_llm_task_passes_the_resolved_seed_kind(self, monkeypatch): + """The seed kind resolved by llm_ops must reach the coordinator.""" + captured: Dict[str, Any] = {} + + mock_cache_manager = MagicMock() + mock_cache_manager.get_cache.return_value = {"llm_processed": {"dialogs": DIALOGS}} + mock_cache_manager.media_lock.return_value.__enter__ = MagicMock(return_value=None) + mock_cache_manager.media_lock.return_value.__exit__ = MagicMock(return_value=False) + mock_cache_manager.invalidate_llm_status.return_value = {} + mock_cache_manager.get_task_by_id.return_value = {"view_token": "tok"} + for _writer in ("save_llm_result", "save_llm_status", "update_task_status"): + getattr(mock_cache_manager, _writer).return_value = True + + mock_coordinator = MagicMock() + mock_coordinator.process = lambda **kwargs: captured.update(kwargs) or { + "calibrated_text": "calibrated", + "summary_text": "summary", + "stats": {}, + "models_used": {}, + } + + router = MagicMock() + monkeypatch.setattr(llm_ops, "llm_task_queue", MagicMock()) + monkeypatch.setattr(llm_ops, "cache_manager", mock_cache_manager) + monkeypatch.setattr(llm_ops, "llm_coordinator", mock_coordinator) + monkeypatch.setattr(llm_ops, "get_notification_router", lambda: router) + monkeypatch.setattr(llm_ops, "normalize_processing_options", lambda _options: { + "calibrate": True, + "summarize": True, + "chapters": True, + "infer_speaker_names": False, + }) + + llm_ops._handle_llm_task( + { + "task_id": "task-anchor", + "url": "https://example.com/v/1", + "platform": "bilibili", + "media_id": "BV1fcHD6cEw3", + "video_title": "Anchor source video", + "author": "author", + "description": "desc", + "transcript": "transcript", + "use_speaker_recognition": False, + "transcription_data": None, + "wechat_webhook": None, + "notification_webhooks": {}, + "is_generic": False, + "calibrate_only": False, + "processing_options": { + "calibrate": False, + "summarize": False, + "chapters": True, + "infer_speaker_names": False, + }, + } + ) + + assert captured["timeline_segments"] == DIALOGS + assert captured["timeline_segments_kind"] == "cached_dialogs" + + +class TestSegmentsAnchorIsNotRegressed: + """The #228 path (chapters anchored on raw segments) still resolves.""" + + def test_segments_label_reads_the_transcript_anchor(self, tmp_path): + (tmp_path / "llm_processed.json").write_text( + json.dumps({"dialogs": DIALOGS}, ensure_ascii=False), encoding="utf-8" + ) + (tmp_path / "transcript_funasr.json").write_text( + json.dumps({"segments": SEGMENTS}, ensure_ascii=False), encoding="utf-8" + ) + + anchor, anchor_kind = load_notes_source_segments(tmp_path, "segments") + + assert anchor_kind == "segments" + assert [entry["text"] for entry in anchor] == [ + entry["text"] for entry in SEGMENTS + ] + assert compute_notes_anchor_fingerprint(anchor) == ( + compute_notes_anchor_fingerprint(load_segments(tmp_path)) + ) + + +class TestDeclaredKindFailsSafe: + """A declared kind must never be silently substituted (fail-safe only).""" + + def test_dialogs_label_without_dialogs_file_returns_nothing(self, tmp_path): + (tmp_path / "transcript_funasr.json").write_text( + json.dumps({"segments": SEGMENTS}, ensure_ascii=False), encoding="utf-8" + ) + + assert load_notes_source_segments(tmp_path, "dialogs") == (None, None) + + def test_segments_label_without_transcript_file_returns_nothing(self, tmp_path): + (tmp_path / "llm_processed.json").write_text( + json.dumps({"dialogs": DIALOGS}, ensure_ascii=False), encoding="utf-8" + ) + + assert load_notes_source_segments(tmp_path, "segments") == (None, None) + + def test_notes_fails_loudly_when_declared_anchor_is_gone(self, tmp_path): + (tmp_path / "transcript_funasr.json").write_text( + json.dumps({"segments": SEGMENTS}, ensure_ascii=False), encoding="utf-8" + ) + anchor_fingerprint = compute_notes_anchor_fingerprint(SEGMENTS) + (tmp_path / "llm_chapters.json").write_text( + json.dumps( + { + "format_version": "v1", + "source": { + "kind": "dialogs", + "segment_count": len(DIALOGS), + "fingerprint": anchor_fingerprint, + }, + "chapters": [ + { + "title": "Only", + "gist": "gone", + "start_seg": 0, + "end_seg": 0, + } + ], + }, + ensure_ascii=False, + ), + encoding="utf-8", + ) + + client = FakeNotesClient(["must not be called"]) + result = NotesProcessor(client, _notes_config()).process(cache_dir=tmp_path) + + assert result.status is NotesStatus.FAILED + assert client.calls == [] \ No newline at end of file