From 5570732a51f58dda79c6df7a4fa5fda92c3fdb2b Mon Sep 17 00:00:00 2001 From: zj1123581321 Date: Sun, 19 Jul 2026 09:47:09 -0400 Subject: [PATCH 1/3] =?UTF-8?q?T8(S1):=20=E7=A1=AE=E5=AE=9A=E6=80=A7?= =?UTF-8?q?=E6=AE=B5=E8=90=BD=E5=8C=96=E5=B7=A5=E5=85=B7=20paragraphize?= =?UTF-8?q?=EF=BC=88=E9=95=BF=E5=BA=A6=E9=A2=84=E7=AE=97+=E6=8E=88?= =?UTF-8?q?=E6=9D=83=E6=96=AD=E7=82=B9=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../transcriber/paragraphize.py | 273 +++++++++++ tests/unit/test_paragraphize.py | 425 ++++++++++++++++++ 2 files changed, 698 insertions(+) create mode 100644 src/video_transcript_api/transcriber/paragraphize.py create mode 100644 tests/unit/test_paragraphize.py diff --git a/src/video_transcript_api/transcriber/paragraphize.py b/src/video_transcript_api/transcriber/paragraphize.py new file mode 100644 index 00000000..64571ed5 --- /dev/null +++ b/src/video_transcript_api/transcriber/paragraphize.py @@ -0,0 +1,273 @@ +"""确定性段落化（deterministic paragraphization, v1）。 + +把连续的小段（ASR/字幕 segments）拼成阅读段落。**只选边界、不改文本**： +每个输出段落都是连续输入成员的并集，断点只落在成员边界上，全文覆盖 +不重不漏。零 LLM 依赖、零 I/O、同输入必同输出。 + +规格来源：docs/sessions/260719-0513-chapters/TASKS.md T8 +「确定性段落化算法规格（v1）」。核心语义——**长度只是预算不是闸刀**： +累计长度达到 target 后开始寻找授权断点，授权点才真的断；一路无授权 +点时由 hard_max 逗号级兜底、2*hard_max 终端硬断保证收敛。 +""" + +import math +from typing import Any, Dict, List, Optional + +from ..utils.logging import setup_logger + +logger = setup_logger("paragraphize") + +# 句末标点：CJK `。！？…` + ASCII `.!?`。 +_SENTENCE_END_CHARS = frozenset("。！？….!?") +# 逗号级标点（hard_max 兜底时才获得授权）：`，；：、` + `,;:`。 +_COMMA_LEVEL_CHARS = frozenset("，；：、,;:") +# 标点后允许出现的收尾引号/括号：判断句末/逗号收尾时先剥离这层字符， +# 兼容 `。"` `！」` `."` `")` 这类写法。 +_TRAILING_CLOSERS = frozenset("\"'”’」』）)]}》") + + +def _is_cjk_char(ch: str) -> bool: + """判断字符是否属于 CJK 语境（拼接时两侧任一则为 CJK，不加空格）。 + + 覆盖 CJK 统一表意文字（含扩展 A/B+ 与兼容表意）、CJK 标点（`。，` 等 + 本身也按 CJK 语境处理——`。"` 后不该出现空格）、全角形式、日文假名、 + 韩文音节。 + """ + cp = ord(ch) + return ( + 0x4E00 <= cp <= 0x9FFF + or 0x3400 <= cp <= 0x4DBF + or 0xF900 <= cp <= 0xFAFF + or 0x3000 <= cp <= 0x303F + or 0xFF00 <= cp <= 0xFFEF + or 0x3040 <= cp <= 0x30FF + or 0xAC00 <= cp <= 0xD7AF + or 0x20000 <= cp <= 0x2FA1F + ) + + +def _ends_with_char_in(text: str, chars: frozenset) -> bool: + """判断文本是否以指定标点集合中的字符收尾（兼容收尾引号/括号）。 + + 先剥离尾部空白，再循环剥离收尾引号/括号（`_TRAILING_CLOSERS`）， + 最后看剩下的末字符是否落在目标集合里。 + """ + stripped = text.rstrip() + while stripped and stripped[-1] in _TRAILING_CLOSERS: + stripped = stripped[:-1].rstrip() + return bool(stripped) and stripped[-1] in chars + + +def _smart_join(parts: List[str]) -> str: + """CJK 感知拼接：左文本以 CJK 结尾或右文本以 CJK 开头时不加空格， + 否则加一个空格。成员原文逐字保留（不改文本），只在边界插入 0/1 个空格。 + """ + if not parts: + return "" + out = parts[0] + for part in parts[1:]: + left = out.rstrip() + right = part.lstrip() + if left and right and not _is_cjk_char(left[-1]) and not _is_cjk_char(right[0]): + out += " " + out += part + return out + + +def _coerce_seconds(value: Any) -> Optional[float]: + """把时间字段收敛为 float 秒或 None。 + + 输入契约允许 float 秒或 None；任何其他形态（字符串、bool、非有限值） + 一律按不可用处理（None），让停顿授权永不因脏数据误触发。 + """ + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + seconds = float(value) + return seconds if math.isfinite(seconds) else None + + +def _coerce_number(value: Any) -> Optional[float]: + """把 duration 字段收敛为有限数值或 None（用于"全员可数值化则求和"）。""" + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + number = float(value) + return number if math.isfinite(number) else None + + +def _boundary_level( + left: Dict[str, Any], right: Dict[str, Any], pause_threshold_seconds: float +) -> int: + """评估相邻两成员之间边界的授权级别。 + + Returns: + 2 = 强授权（左成员以句末标点收尾，或两侧时间均可用且 + `right.start - left.end >= pause_threshold`）； + 1 = 弱授权（左成员以逗号级标点收尾，仅 hard_max 兜底时生效）； + 0 = 无授权。 + """ + if _ends_with_char_in(left["text"], _SENTENCE_END_CHARS): + return 2 + left_end = left["end_time"] + right_start = right["start_time"] + if ( + left_end is not None + and right_start is not None + and right_start - left_end >= pause_threshold_seconds + ): + return 2 + if _ends_with_char_in(left["text"], _COMMA_LEVEL_CHARS): + return 1 + return 0 + + +def _build_paragraph(members: List[Dict[str, Any]]) -> Dict[str, Any]: + """把一组连续成员组装成一个段落 dict。 + + - `text`：smart join 拼接（不改原文，只插入 0/1 个空格）； + - `start_time` = 首成员 start、`end_time` = 末成员 end（None 原样保留， + 绝不编造）； + - `original_text`：仅当全组成员都携带时拼接输出； + - `duration`：仅当全员可数值化时求和输出。 + """ + paragraph: Dict[str, Any] = { + "text": _smart_join([m["text"] for m in members]), + "start_time": members[0]["start_time"], + "end_time": members[-1]["end_time"], + } + if all(m["original_text"] is not None for m in members): + paragraph["original_text"] = _smart_join([m["original_text"] for m in members]) + durations = [_coerce_number(m["duration"]) for m in members] + if all(d is not None for d in durations): + paragraph["duration"] = sum(d for d in durations if d is not None) + return paragraph + + +def paragraphize_segments( + segments: List[Dict[str, Any]], + *, + target_chars: int = 300, + hard_max_chars: int = 600, + pause_threshold_seconds: float = 2.0, +) -> List[Dict[str, Any]]: + """把连续 segments 确定性拼成阅读段落（只选边界、不改文本）。 + + 前提（调用方保证）：每条 segment 必有 `"text": str` 且非空； + `"start_time"`/`"end_time"` 为 float 秒或 None（缺失/其他类型按 None + 处理）；`"original_text"`（str）与 `"duration"`（数值）可选。 + + 规则（cur_len = 当前段落累计字符数，只算 text 长度，不算拼接空格）： + 1. 句末授权：cur_len >= target 且左成员以句末标点收尾 → 断； + 2. 停顿授权：cur_len >= target 且两侧时间均可用、gap >= + pause_threshold → 断（无需句末标点）； + 3. target 之前任何授权点都不提前断； + 4. 硬上限兜底：cur_len >= hard_max 且组内尚无断点 → 逗号级也获得 + 授权，断点取 hard_max 之前最后一个授权点（句末/停顿/逗号均可， + 回溯不再受 target 下限约束——宁可短也不腰斩语义），取不到则取 + hard_max 之后第一个授权点； + 5. 终端规则：到 2*hard_max 仍无任何授权点 → 在保持组长度 + <= 2*hard_max 的最后一个成员边界硬断并记 warning（保证收敛）； + 6. 病理兜底：单成员 len(text) > hard_max → 自成一段并记 warning， + 不影响相邻段； + 7. 时间缺失：任一侧时间为 None → 该边界停顿信号不可用，仅用标点。 + + Args: + segments: 输入 segment dict 列表（见上方前提）。 + target_chars: 长度预算（达到后开始寻找授权断点）。 + hard_max_chars: 硬上限（达到后逗号级授权、触发回溯选点）。 + pause_threshold_seconds: 停顿授权阈值（秒）。 + + Returns: + 段落 dict 列表，每段 `{"text", "start_time", "end_time"}` 必有， + `original_text`/`duration` 按全员规则条件性输出。段落是连续成员 + 的并集，覆盖全部输入、不重不漏。 + """ + # 归一化为内部记录：时间收敛为 float|None，可选字段统一成显式 None， + # 后续组装与授权判断都不必再防御脏形态。 + records: List[Dict[str, Any]] = [] + for seg in segments: + original_text = seg.get("original_text") + records.append( + { + "text": seg["text"], + "start_time": _coerce_seconds(seg.get("start_time")), + "end_time": _coerce_seconds(seg.get("end_time")), + "original_text": original_text if isinstance(original_text, str) else None, + "duration": seg.get("duration"), + } + ) + + paragraphs: List[Dict[str, Any]] = [] + n = len(records) + group_start = 0 # 当前组首成员下标（含） + cur_len = 0 # 当前组累计字符数（只算 text，不含拼接空格） + # 组内已见的授权边界：(右成员下标, 级别 2/1)。全部记录（含 target 之前 + # 的），供 hard_max 回溯取"最后一个授权点"。 + boundaries: List[tuple] = [] + + i = 0 + while i < n: + rec = records[i] + text_len = len(rec["text"]) + + # 规则 6（病理兜底）：单成员超 hard_max → 自成一段，相邻组照常输出。 + if text_len > hard_max_chars: + if cur_len > 0: + paragraphs.append(_build_paragraph(records[group_start:i])) + paragraphs.append(_build_paragraph(records[i : i + 1])) + logger.warning( + f"paragraphize: single segment length {text_len} exceeds " + f"hard_max_chars {hard_max_chars}, kept as its own paragraph" + ) + i += 1 + group_start = i + cur_len = 0 + boundaries = [] + continue + + if cur_len > 0: + # 组非空：评估边界 (i-1, i) 是否断开。回溯断点后组变短，授权 + # 状态随之变化，因此用循环重判直到"不断"为止。 + while True: + level = _boundary_level(records[i - 1], rec, pause_threshold_seconds) + brk: Optional[int] = None + force_break = False + if cur_len >= target_chars and level == 2: + # 规则 1/2：句末/停顿授权（target 之后才生效）。 + brk = i + elif cur_len >= hard_max_chars: + if boundaries: + # 规则 4 前半：回溯到 hard_max 之前最后一个授权点。 + brk = boundaries[-1][0] + elif level == 1: + # 规则 4 后半：hard_max 之后第一个逗号级授权点。 + brk = i + elif cur_len + text_len > 2 * hard_max_chars: + # 规则 5（终端）：无任何授权点且再加就超 2*hard_max。 + brk = i + force_break = True + if brk is None: + # 不断：把当前边界授权级别登记进组内记录，供后续回溯。 + if level: + boundaries.append((i, level)) + break + if force_break: + logger.warning( + f"paragraphize: no authorized breakpoint before " + f"2x hard_max_chars ({2 * hard_max_chars}), force break " + f"at segment boundary (group_chars={cur_len})" + ) + paragraphs.append(_build_paragraph(records[group_start:brk])) + group_start = brk + cur_len = sum(len(records[k]["text"]) for k in range(brk, i)) + boundaries = [(r, lv) for (r, lv) in boundaries if r > brk] + if brk == i: + # 断在当前边界：新组为空，直接落成员 i。 + break + + cur_len += text_len + i += 1 + + if group_start < n: + paragraphs.append(_build_paragraph(records[group_start:n])) + + return paragraphs diff --git a/tests/unit/test_paragraphize.py b/tests/unit/test_paragraphize.py new file mode 100644 index 00000000..e890308b --- /dev/null +++ b/tests/unit/test_paragraphize.py @@ -0,0 +1,425 @@ +"""Unit tests for the deterministic paragraphization tool. + +Covers video_transcript_api.transcriber.paragraphize.paragraphize_segments: +a pure function (no LLM, no I/O, deterministic) that only SELECTS break +boundaries between consecutive segments and never rewrites any text. + +Spec: docs/sessions/260719-0513-chapters/TASKS.md T8 "deterministic +paragraphization v1" -- length is a budget, not a gate; breaks only happen +at authorized points (sentence-end / pause / comma-level fallback / +terminal force-break / pathological single-segment passthrough). + +All console output in this file must be English only (no emoji, no +Chinese), per project convention. +""" + +import pytest +from loguru import logger + +from video_transcript_api.transcriber.paragraphize import paragraphize_segments + + +# --------------------------------------------------------------------------- +# helpers +# --------------------------------------------------------------------------- + +def _seg(text, start=None, end=None, **extra): + """Build one input segment dict (float seconds or None for times).""" + seg = {"text": text, "start_time": start, "end_time": end} + seg.update(extra) + return seg + + +def _member_char_total(paragraph_text): + """Total characters ignoring join spaces -- for coverage assertions.""" + return len(paragraph_text.replace(" ", "")) + + +@pytest.fixture +def warnings_sink(): + """Capture loguru WARNING+ messages emitted during a test.""" + messages = [] + sink_id = logger.add( + lambda m: messages.append(str(m.record["message"])), level="WARNING" + ) + try: + yield messages + finally: + logger.remove(sink_id) + + +# --------------------------------------------------------------------------- +# Chinese prose with sentence-end punctuation +# --------------------------------------------------------------------------- + +class TestChineseSentenceEndBreaks: + def test_breaks_after_sentence_end_once_target_reached(self): + # 5 segments x 30 chars each, all ending with the CJK full stop. + segs = [_seg("文" * 29 + "。") for _ in range(5)] + result = paragraphize_segments(segs, target_chars=100, hard_max_chars=200) + # cur_len hits 120 (>= 100) at the 4th segment boundary -> break there; + # the 5th segment is flushed as the trailing paragraph. + assert len(result) == 2 + assert _member_char_total(result[0]["text"]) == 120 + assert _member_char_total(result[1]["text"]) == 30 + + def test_every_non_terminal_paragraph_ends_with_sentence_punctuation(self): + # Property: in a pure-punctuation scenario every paragraph except the + # trailing one must end with sentence-end punctuation. + segs = [_seg("文" * 29 + "。") for _ in range(9)] + result = paragraphize_segments(segs, target_chars=100, hard_max_chars=200) + assert len(result) >= 2 + for paragraph in result[:-1]: + assert paragraph["text"].rstrip()[-1] in "。！？….!?" + + def test_no_early_break_before_target(self): + # A run of ten 10-char sentences must NOT be split into ten tiny + # paragraphs: authorization points are inert while cur_len < target. + segs = [_seg("短" * 9 + "。") for _ in range(10)] # total 100 chars + result = paragraphize_segments(segs, target_chars=300, hard_max_chars=600) + assert len(result) == 1 + assert _member_char_total(result[0]["text"]) == 100 + + +# --------------------------------------------------------------------------- +# pause authorization +# --------------------------------------------------------------------------- + +class TestPauseAuthorization: + def test_pause_at_or_above_threshold_breaks_without_punctuation(self): + segs = [ + _seg("字" * 30, 0.0, 10.0), + _seg("字" * 30, 13.0, 23.0), # gap 3.0 >= 2.0, but cur_len 30 < 50 + _seg("字" * 30, 26.0, 36.0), # gap 3.0 >= 2.0, cur_len 60 >= 50 -> break + ] + result = paragraphize_segments( + segs, target_chars=50, hard_max_chars=1000, pause_threshold_seconds=2.0 + ) + assert len(result) == 2 + assert _member_char_total(result[0]["text"]) == 60 + assert _member_char_total(result[1]["text"]) == 30 + + def test_pause_below_threshold_does_not_break(self): + segs = [ + _seg("字" * 30, 0.0, 10.0), + _seg("字" * 30, 10.5, 20.0), # gap 0.5 < 2.0 + _seg("字" * 30, 20.5, 30.0), + ] + result = paragraphize_segments( + segs, target_chars=50, hard_max_chars=1000, pause_threshold_seconds=2.0 + ) + assert len(result) == 1 + + def test_pause_exactly_at_threshold_breaks(self): + segs = [ + _seg("字" * 30, 0.0, 10.0), + _seg("字" * 30, 10.0, 20.0), + _seg("字" * 30, 22.0, 32.0), # gap exactly 2.0 -> authorized + ] + result = paragraphize_segments( + segs, target_chars=50, hard_max_chars=1000, pause_threshold_seconds=2.0 + ) + assert len(result) == 2 + + +# --------------------------------------------------------------------------- +# missing time +# --------------------------------------------------------------------------- + +class TestMissingTime: + def test_punctuation_mode_still_works_with_none_times(self): + segs = [_seg("文" * 14 + "。") for _ in range(3)] # 15 chars each + result = paragraphize_segments(segs, target_chars=20, hard_max_chars=1000) + assert len(result) == 2 + assert result[0]["start_time"] is None + assert result[0]["end_time"] is None + + def test_pause_never_fires_when_time_is_none(self): + # No punctuation, no usable time -> no authorization at all; with a + # large hard_max the whole input stays one paragraph. + segs = [_seg("字" * 15) for _ in range(3)] + result = paragraphize_segments(segs, target_chars=20, hard_max_chars=1000) + assert len(result) == 1 + + def test_one_sided_none_time_disables_pause(self): + # Right side start_time None -> pause signal unavailable even though + # the left side has a complete interval. + segs = [ + _seg("字" * 30, 0.0, 10.0), + _seg("字" * 30, None, None), + _seg("字" * 30, None, None), + ] + result = paragraphize_segments(segs, target_chars=50, hard_max_chars=1000) + assert len(result) == 1 + + +# --------------------------------------------------------------------------- +# hard_max fallback (comma-level authorization) +# --------------------------------------------------------------------------- + +class TestHardMaxFallback: + def test_breaks_at_last_comma_before_hard_max(self): + # 6 segments x 20 chars; only segment 4 (index 3) ends with a comma. + segs = [ + _seg("文" * 20), + _seg("文" * 20), + _seg("文" * 20), + _seg("文" * 19 + "，"), + _seg("文" * 20), + _seg("文" * 20), + ] + result = paragraphize_segments(segs, target_chars=50, hard_max_chars=100) + # At cur_len == 100 with no strong point, the break falls back to the + # last authorized point before hard_max: the comma after segment 4. + assert len(result) == 2 + assert _member_char_total(result[0]["text"]) == 80 + assert result[0]["text"].rstrip().endswith("，") + assert _member_char_total(result[1]["text"]) == 40 + + def test_breaks_at_first_comma_after_hard_max_when_none_before(self): + # 7 segments x 20 chars; the only comma is at segment 6 (index 5), + # i.e. cum 120 > hard_max 100 -- no authorized point exists before + # hard_max, so the first comma-level point after it wins. + segs = [_seg("文" * 20) for _ in range(7)] + segs[5] = _seg("文" * 19 + "，") + result = paragraphize_segments(segs, target_chars=50, hard_max_chars=100) + assert len(result) == 2 + assert _member_char_total(result[0]["text"]) == 120 + assert result[0]["text"].rstrip().endswith("，") + + def test_ascii_comma_also_authorized_at_hard_max(self): + segs = [ + _seg("w" * 20), + _seg("w" * 20), + _seg("w" * 19 + ","), + _seg("w" * 20), + _seg("w" * 20), + ] + result = paragraphize_segments(segs, target_chars=50, hard_max_chars=80) + # cur_len reaches 80 at segment 4 (index 3); last comma point is at + # cum 60 -> first paragraph covers 3 segments. + assert _member_char_total(result[0]["text"]) == 60 + assert result[0]["text"].rstrip().endswith(",") + + +# --------------------------------------------------------------------------- +# terminal force-break rule (no punctuation, no pause) +# --------------------------------------------------------------------------- + +class TestTerminalForceBreak: + def test_english_cue_stream_force_breaks_before_2x_hard_max(self, warnings_sink): + # 5 segments x 60 chars, no punctuation, no time -> no authorization + # point ever exists. Groups must stay <= 2 * hard_max = 200. + segs = [_seg("w" * 60) for _ in range(5)] + result = paragraphize_segments(segs, target_chars=50, hard_max_chars=100) + assert len(result) == 2 + for paragraph in result: + assert _member_char_total(paragraph["text"]) <= 200 + # Coverage: nothing lost, nothing duplicated. + total = sum(_member_char_total(p["text"]) for p in result) + assert total == 300 + assert any("force" in m.lower() for m in warnings_sink) + + def test_force_break_never_splits_a_member(self, warnings_sink): + segs = [_seg("w" * 60) for _ in range(7)] + result = paragraphize_segments(segs, target_chars=50, hard_max_chars=100) + # Every paragraph text length must be a multiple of 60 plus join + # spaces -- members are never cut mid-text. + for paragraph in result: + assert _member_char_total(paragraph["text"]) % 60 == 0 + + +# --------------------------------------------------------------------------- +# pathological single member exceeding hard_max +# --------------------------------------------------------------------------- + +class TestPathologicalOversizeMember: + def test_oversize_member_stands_alone_with_warning(self, warnings_sink): + segs = [ + _seg("短" * 19 + "。"), + _seg("长" * 150), + _seg("尾" * 20), + ] + result = paragraphize_segments(segs, target_chars=50, hard_max_chars=100) + assert len(result) == 3 + assert result[0]["text"] == "短" * 19 + "。" + assert result[1]["text"] == "长" * 150 + assert result[2]["text"] == "尾" * 20 + assert any("exceeds" in m.lower() for m in warnings_sink) + + def test_oversize_member_does_not_merge_neighbors(self, warnings_sink): + # The oversize member must not drag its neighbors into its paragraph, + # and neighbors must not be glued to each other across it. + segs = [_seg("甲" * 30), _seg("长" * 150), _seg("乙" * 30)] + result = paragraphize_segments(segs, target_chars=300, hard_max_chars=100) + assert [p["text"] for p in result] == ["甲" * 30, "长" * 150, "乙" * 30] + + +# --------------------------------------------------------------------------- +# ASCII punctuation and closer-suffix handling +# --------------------------------------------------------------------------- + +class TestAsciiPunctuation: + def test_breaks_after_ascii_period(self): + segs = [_seg("Hello world."), _seg("Next one."), _seg("Third here.")] + result = paragraphize_segments(segs, target_chars=10, hard_max_chars=1000) + assert len(result) == 2 + assert result[0]["text"] == "Hello world." + assert result[1]["text"] == "Next one. Third here." + + def test_period_followed_by_closing_quote_is_sentence_end(self): + segs = [_seg('He said "wow."'), _seg("She left early today.")] + result = paragraphize_segments(segs, target_chars=10, hard_max_chars=1000) + assert len(result) == 2 + assert result[0]["text"] == 'He said "wow."' + + def test_period_followed_by_closing_paren_is_sentence_end(self): + segs = [_seg("It works (really)."), _seg("Move on now please.")] + result = paragraphize_segments(segs, target_chars=10, hard_max_chars=1000) + assert len(result) == 2 + + def test_cjk_exclamation_followed_by_bracket_is_sentence_end(self): + segs = [_seg("她喊「够了！」"), _seg("全场安静下来没人再说话。")] + result = paragraphize_segments(segs, target_chars=5, hard_max_chars=1000) + assert len(result) == 2 + assert result[0]["text"] == "她喊「够了！」" + + +# --------------------------------------------------------------------------- +# smart join (CJK-aware spacing) +# --------------------------------------------------------------------------- + +class TestSmartJoin: + def test_chinese_segments_join_without_space(self): + segs = [_seg("你好吗。"), _seg("我很好。")] + result = paragraphize_segments(segs, target_chars=1000) + assert result[0]["text"] == "你好吗。我很好。" + + def test_english_segments_join_with_single_space(self): + segs = [_seg("hello"), _seg("world")] + result = paragraphize_segments(segs, target_chars=1000) + assert result[0]["text"] == "hello world" + + def test_cjk_left_edge_suppresses_space(self): + segs = [_seg("你好"), _seg("world")] + result = paragraphize_segments(segs, target_chars=1000) + assert result[0]["text"] == "你好world" + + def test_cjk_right_edge_suppresses_space(self): + segs = [_seg("hello"), _seg("世界")] + result = paragraphize_segments(segs, target_chars=1000) + assert result[0]["text"] == "hello世界" + + def test_no_space_after_chinese_full_stop(self): + # The classic defect this guards against: a space inserted after the + # CJK full stop inside a Chinese paragraph. + segs = [_seg("这是第一句。"), _seg("这是第二句。"), _seg("这是第三句。")] + result = paragraphize_segments(segs, target_chars=1000) + assert result[0]["text"] == "这是第一句。这是第二句。这是第三句。" + assert "。 " not in result[0]["text"] + + +# --------------------------------------------------------------------------- +# output contract: original_text / duration / start / end +# --------------------------------------------------------------------------- + +class TestOutputContract: + def test_original_text_joined_when_all_members_have_it(self): + segs = [ + _seg("校准一。", 1.0, 2.0, original_text="原始一，", duration=1.0), + _seg("校准二。", 2.0, 3.5, original_text="原始二，", duration=1.5), + ] + result = paragraphize_segments(segs, target_chars=1000) + assert len(result) == 1 + paragraph = result[0] + assert paragraph["text"] == "校准一。校准二。" + assert paragraph["original_text"] == "原始一，原始二，" + assert paragraph["start_time"] == 1.0 + assert paragraph["end_time"] == 3.5 + assert paragraph["duration"] == 2.5 + + def test_original_text_omitted_when_any_member_lacks_it(self): + segs = [ + _seg("校准一。", 1.0, 2.0, original_text="原始一，", duration=1.0), + _seg("校准二。", 2.0, 3.5), # no original_text, no duration + ] + result = paragraphize_segments(segs, target_chars=1000) + assert "original_text" not in result[0] + assert "duration" not in result[0] + + def test_start_end_taken_from_first_and_last_member(self): + segs = [ + _seg("文" * 29 + "。", 5.0, 6.0), + _seg("文" * 29 + "。", 6.0, 7.0), + _seg("文" * 29 + "。", 7.0, 9.5), + ] + result = paragraphize_segments(segs, target_chars=1000) + assert result[0]["start_time"] == 5.0 + assert result[0]["end_time"] == 9.5 + + def test_none_times_preserved_not_fabricated(self): + segs = [ + _seg("文" * 29 + "。", None, 6.0), + _seg("文" * 29 + "。", 6.0, None), + ] + result = paragraphize_segments(segs, target_chars=1000) + assert result[0]["start_time"] is None + assert result[0]["end_time"] is None + + def test_per_paragraph_times_follow_member_span_after_break(self): + segs = [ + _seg("文" * 29 + "。", 0.0, 1.0), + _seg("文" * 29 + "。", 1.0, 2.0), + _seg("文" * 29 + "。", 2.0, 3.0), + _seg("文" * 29 + "。", 3.0, 4.0), + ] + result = paragraphize_segments(segs, target_chars=100, hard_max_chars=1000) + # cur_len reaches 90 (< target) at the last internal boundary, so the + # run stays a single paragraph of 120 chars -- target is a budget, + # not a gate; the trailing flush keeps whatever accumulated. + assert len(result) == 1 + for paragraph in result: + assert paragraph["start_time"] is not None + assert paragraph["end_time"] is not None + assert paragraph["end_time"] > paragraph["start_time"] + + +# --------------------------------------------------------------------------- +# determinism and edge cases +# --------------------------------------------------------------------------- + +class TestDeterminismAndEdges: + def test_same_input_same_output(self): + segs = [ + _seg("文" * 29 + "。", 0.0, 1.0, original_text="原" * 30, duration=1.0), + _seg("文" * 20, 1.5, 2.0), + _seg("hello world again.", 2.0, 3.0), + _seg("字" * 70, 3.0, 4.0), + _seg("文" * 19 + "，", 4.0, 5.0), + _seg("w" * 60, None, None), + ] + kwargs = dict(target_chars=50, hard_max_chars=100, pause_threshold_seconds=2.0) + assert paragraphize_segments(segs, **kwargs) == paragraphize_segments(segs, **kwargs) + + def test_empty_input_returns_empty_list(self): + assert paragraphize_segments([]) == [] + + def test_single_segment_passthrough(self): + result = paragraphize_segments([_seg("唯一的一段。", 1.0, 2.0)]) + assert result == [{"text": "唯一的一段。", "start_time": 1.0, "end_time": 2.0}] + + def test_full_coverage_no_loss_no_duplication(self): + # Invariant: paragraphs partition the input -- concatenated member + # text (ignoring join spaces) equals concatenated input text. + segs = [ + _seg("文" * 29 + "。"), + _seg("文" * 20), + _seg("hello."), + _seg("文" * 19 + "，"), + _seg("文" * 150), + _seg("world"), + ] + result = paragraphize_segments(segs, target_chars=50, hard_max_chars=100) + produced = "".join(p["text"].replace(" ", "") for p in result) + expected = "".join(s["text"] for s in segs) + assert produced == expected From 72e938a0f4cdf0edbbeb8fc83c3a9bdb0f6d954a Mon Sep 17 00:00:00 2001 From: zj1123581321 Date: Sun, 19 Jul 2026 12:23:31 -0400 Subject: [PATCH 2/3] =?UTF-8?q?T8(S3):=20SpeakerAwareProcessor=20has=5Fspe?= =?UTF-8?q?aker=20=E6=A8=A1=E5=BC=8F=EF=BC=88=E6=97=A0=E8=AF=B4=E8=AF=9D?= =?UTF-8?q?=E4=BA=BA=E9=80=90=E6=AE=B5=E6=A0=A1=E5=AF=B9=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - has_speaker=None 自动判定（任一原始段可解析出 speaker 即为 True）， False 走无说话人逐段校对：SpeakerInferencer.infer 整步跳过（零 LLM 调用），key_info 在校准开启时仍提取 - 不做同说话人合并；speaker/speaker_id 与 None 时间保留缺省，不落 "unknown"/"00:00:00" - 落盘前确定性段落化（消费 normalize 前 float 秒时间快照）， llm_processed dialogs/章节输入/渲染锚点共用同一列表 - prompts 新增无说话人变体（行格式去 [speaker]），DialogSegmenter 支持按模式覆盖 preferred/max 块长 - 新增 19 个单测锁定契约 --- .../llm/processors/speaker_aware_processor.py | 281 ++++++++-- .../llm/prompts/__init__.py | 99 ++++ .../llm/segmenters/dialog_segmenter.py | 23 +- tests/unit/test_speaker_aware_no_speaker.py | 486 ++++++++++++++++++ 4 files changed, 831 insertions(+), 58 deletions(-) create mode 100644 tests/unit/test_speaker_aware_no_speaker.py diff --git a/src/video_transcript_api/llm/processors/speaker_aware_processor.py b/src/video_transcript_api/llm/processors/speaker_aware_processor.py index 0a1c0659..3715b29a 100644 --- a/src/video_transcript_api/llm/processors/speaker_aware_processor.py +++ b/src/video_transcript_api/llm/processors/speaker_aware_processor.py @@ -19,9 +19,12 @@ from ..prompts import ( STRUCTURED_CALIBRATE_SYSTEM_PROMPT, STRUCTURED_CALIBRATE_SYSTEM_PROMPT_EN, + STRUCTURED_CALIBRATE_NO_SPEAKER_SYSTEM_PROMPT, + STRUCTURED_CALIBRATE_NO_SPEAKER_SYSTEM_PROMPT_EN, build_structured_calibrate_user_prompt, ) from ..schemas import CALIBRATION_RESULT_SCHEMA +from ...transcriber.paragraphize import paragraphize_segments from ..utils.language_detector import detect_language logger = setup_logger(__name__) @@ -65,6 +68,7 @@ def process( selected_models: Optional[Dict] = None, skip_calibration: bool = False, infer_speaker_names: bool = True, + has_speaker: Optional[bool] = None, ) -> Dict: """处理有说话人文本 @@ -79,12 +83,27 @@ def process( skip_calibration: 是否跳过分块 LLM 校对调用（processing_options.calibrate=False 时为 True）。说话人推断、说话人映射、对话规范化合并这些步骤仍会执行—— 它们属于"转录"交付物（谁在说话）而非"校对"（文字是否准确），跳过的只是 - 逐块把文本喂给 LLM 做文字校正/纠错的那一步。 + 逐块把文本喂给 LLM 做文字校正/纠错的那一步。 + has_speaker: 输入是否携带说话人标签。None（默认）时自动判定—— + 任一原始输入段能解析出说话人标签即为 True（混合输入维持现状 + 行为）；False 时走「无说话人逐段校对」模式：跳过说话人推断、 + 不做同说话人合并、speaker/时间缺省保留缺省，最终产物是确定性 + 段落化后的段落。 Returns: 处理结果字典 """ - base_dialogs = self._coerce_dialogs(dialogs) + # has_speaker 自动判定（None 时）：任一原始输入段能解析出说话人标签 + # 即视为有说话人（混合输入维持现状 has_speaker=True 行为）。必须在 + # _coerce_dialogs 之前判定——coerce 会把缺省说话人规范掉。 + if has_speaker is None: + has_speaker = any( + isinstance(d, dict) + and SpeakerInferencer.resolve_dialog_speaker(d) is not None + for d in dialogs or [] + ) + + base_dialogs = self._coerce_dialogs(dialogs, has_speaker=has_speaker) total_length = sum(len(d.get("text", "")) for d in base_dialogs) logger.info( f"Start processing speaker-aware text: {title}, dialog count: {len(base_dialogs)}, " @@ -98,7 +117,11 @@ def process( # Key info only feeds calibration and speaker-name inference prompts. # Skipping both must not make a hidden LLM call. - if skip_calibration and not infer_speaker_names: + # 无说话人模式推广：说话人推断整步跳过时，key_info 的唯一剩余消费者 + # 是校准——need = not skip_calibration or (has_speaker and + # infer_speaker_names)；has_speaker=True 时与原条件完全等价。 + need_key_info = (not skip_calibration) or (has_speaker and infer_speaker_names) + if not need_key_info: key_info = KeyInfo([], [], [], [], [], [], []) else: key_info = self.key_info_extractor.extract( @@ -116,21 +139,33 @@ def process( # 在这份输入上与原来的列表推导式行为完全等价，但与 # 读侧（transcription.py 的分层缓存预检）共用同一个实现，不再因两处各自手写悄悄漂移再次拉出 # 经典读/写指纹不一致 bug。 - speakers = SpeakerInferencer.extract_speaker_labels(base_dialogs) - # 细化 stage=speaker_inference，仅覆盖本次说话人推断调用的审计标签， - # 退出 with 块后自动恢复为外层的 calibration - with set_context(stage="speaker_inference"): - speaker_inference_result = self.speaker_inferencer.infer( - speakers=speakers, - dialogs=base_dialogs, - title=title, - author=author, - description=description, - key_info=key_info, - platform=platform, - media_id=media_id, - allow_llm=infer_speaker_names, - ) + if has_speaker: + speakers = SpeakerInferencer.extract_speaker_labels(base_dialogs) + # 细化 stage=speaker_inference，仅覆盖本次说话人推断调用的审计标签， + # 退出 with 块后自动恢复为外层的 calibration + with set_context(stage="speaker_inference"): + speaker_inference_result = self.speaker_inferencer.infer( + speakers=speakers, + dialogs=base_dialogs, + title=title, + author=author, + description=description, + key_info=key_info, + platform=platform, + media_id=media_id, + allow_llm=infer_speaker_names, + ) + else: + # 无说话人模式：说话人推断整步跳过（零 LLM 调用）。 + # identity_fallback source 让下游 llm_ops 的刷新判断按既有语义 + # 安全跳过；meta 放说明性内容保持可观测。 + speaker_inference_result = { + "mapping": {}, + "meta": { + "note": "input has no speaker labels; speaker inference skipped" + }, + "source": "identity_fallback", + } speaker_mapping = speaker_inference_result.get("mapping", {}) speaker_inference_meta = speaker_inference_result.get("meta", {}) # 本地 codex review 第 6 轮 G4：随 meta 一起把本轮映射的真实 @@ -140,8 +175,22 @@ def process( speaker_inference_source = speaker_inference_result.get("source") # 结构化标准化（应用映射 + 合并连续同说话人 + 时间字段规范化） + # 无说话人模式：normalize 前从 coerce 输出快照 float 秒时间轴。 + # normalize 会把时间 int() 截断成 HH:MM:SS 字符串，±1s 误差会污染 + # 段落化的停顿授权判定；不合并时 coerced/normalized/calibrated 三列表 + # 按下标 1:1 对齐（coerce 会丢空文本条，快照在 coerce 后做）。 + time_snapshot = None + if not has_speaker: + time_snapshot = [ + ( + self._parse_time_value(d.get("start_time")), + self._parse_time_value(d.get("end_time")), + ) + for d in base_dialogs + ] + normalized_dialogs = self._normalize_and_merge_dialogs( - base_dialogs, speaker_mapping + base_dialogs, speaker_mapping, has_speaker=has_speaker ) if skip_calibration: @@ -170,7 +219,17 @@ def process( ) else: # 步骤2: 分段 - chunks = self.segmenter.segment(normalized_dialogs) + # 无说话人模式用 plain 独立分块参数（段落比对话长，预算放大）； + # has_speaker=True 路径沿用 __init__ 的 segmenter，行为不变。 + if has_speaker: + segmenter = self.segmenter + else: + segmenter = DialogSegmenter( + self.config, + preferred_chunk_length=self.config.plain_structured_preferred_chunk_length, + max_chunk_length=self.config.plain_structured_max_chunk_length, + ) + chunks = segmenter.segment(normalized_dialogs) logger.debug(f"Dialogs segmented: {len(chunks)} chunks") # 步骤3: 分段校对（每段独立验证） @@ -182,7 +241,8 @@ def process( title=title, description=description, selected_models=selected_models, - language=detected_language, + language=detected_language, + has_speaker=has_speaker, ) # 合并校对结果（不再进行整体验证） @@ -191,8 +251,20 @@ def process( calibrated_dialogs.extend(chunk) # 构建文本用于统计 - original_text = self._build_text_from_dialogs(normalized_dialogs) - calibrated_text = self._build_text_from_dialogs(calibrated_dialogs) + # 无说话人模式：确定性段落化（钉死在构造 structured_data 返回之前， + # 校准/skip 两条分支共用）——llm_processed.json dialogs、章节输入、 + # 渲染锚点三者是同一个列表。 + if not has_speaker: + calibrated_dialogs = self._paragraphize_no_speaker_dialogs( + calibrated_dialogs, time_snapshot + ) + + original_text = self._build_text_from_dialogs( + normalized_dialogs, has_speaker=has_speaker + ) + calibrated_text = self._build_text_from_dialogs( + calibrated_dialogs, has_speaker=has_speaker + ) logger.info( f"Speaker-aware text processing completed: " @@ -209,7 +281,9 @@ def process( "stats": { "original_length": len(original_text), "calibrated_length": len(calibrated_text), - "dialog_count": len(normalized_dialogs), + "dialog_count": ( + len(normalized_dialogs) if has_speaker else len(calibrated_dialogs) + ), "chunk_count": len(chunks), "calibration_stats": calibration_stats, "speaker_inference": speaker_inference_meta, @@ -217,8 +291,11 @@ def process( } } - def _coerce_dialogs(self, dialogs: List[Dict]) -> List[Dict]: - """将原始对话列表规范化为最小可用格式（speaker/text/start/end/duration）""" + def _coerce_dialogs(self, dialogs: List[Dict], has_speaker: bool = True) -> List[Dict]: + """将原始对话列表规范化为最小可用格式（speaker/text/start/end/duration）。 + + has_speaker=False（无说话人模式）时缺省说话人保留 None，不塞 "unknown"。 + """ coerced = [] for dialog in dialogs or []: if not isinstance(dialog, dict): @@ -240,7 +317,14 @@ def _coerce_dialogs(self, dialogs: List[Dict]) -> List[Dict]: coerced.append( { - "speaker": str(speaker) if speaker is not None else "unknown", + # 无说话人模式保留缺省（None），不塞 "unknown"。注意 + # d.get("speaker", "unknown") 的默认值只在键缺失时生效， + # 键存在但值为 None 时不触发——所以这里必须显式写 None。 + "speaker": ( + str(speaker) + if speaker is not None + else ("unknown" if has_speaker else None) + ), "text": str(text), "start_time": dialog.get("start_time", dialog.get("start")), "end_time": dialog.get("end_time", dialog.get("end")), @@ -251,10 +335,19 @@ def _coerce_dialogs(self, dialogs: List[Dict]) -> List[Dict]: return coerced def _normalize_and_merge_dialogs( - self, dialogs: List[Dict], speaker_mapping: Dict[str, str] + self, dialogs: List[Dict], speaker_mapping: Dict[str, str], has_speaker: bool = True ) -> List[Dict]: - """应用说话人映射、规范化时间字段并合并连续同说话人对话""" - normalized = [] + """应用说话人映射、规范化时间字段并合并连续同说话人对话（has_speaker=False 时仅逐条规范化，不合并）""" + normalized = [] + if not has_speaker: + for dialog in dialogs: + normalized_dialog, _, _ = self._normalize_dialog( + dialog, speaker_mapping, has_speaker=False + ) + if normalized_dialog: + normalized.append(normalized_dialog) + return normalized + current = None current_start_seconds = None current_end_seconds = None @@ -293,10 +386,10 @@ def _normalize_and_merge_dialogs( return normalized def _normalize_dialog( - self, dialog: Dict, speaker_mapping: Dict[str, str] + self, dialog: Dict, speaker_mapping: Dict[str, str], has_speaker: bool = True ) -> tuple: """规范化单条对话，返回(对话, start_seconds, end_seconds)""" - raw_speaker = dialog.get("speaker", "unknown") + raw_speaker = dialog.get("speaker", "unknown" if has_speaker else None) speaker = raw_speaker text = dialog.get("text", "") if not text: @@ -315,12 +408,12 @@ def _normalize_dialog( start_time = ( self._format_timestamp(start_seconds) if start_seconds is not None - else (str(start_raw) if start_raw else "00:00:00") + else (str(start_raw) if start_raw else ("00:00:00" if has_speaker else None)) ) end_time = ( self._format_timestamp(end_seconds) if end_seconds is not None - else (str(end_raw) if end_raw else start_time) + else (str(end_raw) if end_raw else (start_time if has_speaker else None)) ) if start_seconds is not None and end_seconds is not None: @@ -390,7 +483,8 @@ def _calibrate_chunks( title: str, description: str, selected_models: Optional[Dict], - language: str = "zh", + language: str = "zh", + has_speaker: bool = True, ) -> List[List[Dict]]: """校对分块对话（并发处理，每块独立验证） @@ -402,7 +496,8 @@ def _calibrate_chunks( title: 视频标题 description: 描述 selected_models: 选定的模型 - language: 检测到的语言（"zh" 或 "en"） + language: 检测到的语言（"zh" 或 "en"） + has_speaker: 是否含说话人（选择 prompt 变体与行格式/合并兜底） Returns: (calibrated_chunks, calibration_stats) 元组: @@ -432,11 +527,18 @@ def _calibrate_chunks( except Exception as e: logger.warning(f"Failed to load terminology DB: {e}") - # 根据语言选择 system prompt - structured_system_prompt = ( - STRUCTURED_CALIBRATE_SYSTEM_PROMPT_EN if language == "en" - else STRUCTURED_CALIBRATE_SYSTEM_PROMPT - ) + # 根据语言与说话人模式选择 system prompt（无说话人变体的行格式描述 + # 与 _format_chunk_for_prompt 的 has_speaker=False 输出一致） + if has_speaker: + structured_system_prompt = ( + STRUCTURED_CALIBRATE_SYSTEM_PROMPT_EN if language == "en" + else STRUCTURED_CALIBRATE_SYSTEM_PROMPT + ) + else: + structured_system_prompt = ( + STRUCTURED_CALIBRATE_NO_SPEAKER_SYSTEM_PROMPT_EN if language == "en" + else STRUCTURED_CALIBRATE_NO_SPEAKER_SYSTEM_PROMPT + ) calibrated_chunks = [None] * len(chunks) # 跟踪每个 chunk 的校准状态: "success" | "partial" | "fallback" | "failed" @@ -496,7 +598,7 @@ def calibrate_single_chunk(index: int, chunk: List[Dict]): return try: # 构建 prompt（每行带 [id] 锚点） - chunk_text = self._format_chunk_for_prompt(chunk, speaker_mapping) + chunk_text = self._format_chunk_for_prompt(chunk, speaker_mapping, has_speaker) user_prompt = build_structured_calibrate_user_prompt( dialogs_text=chunk_text, @@ -523,7 +625,7 @@ def calibrate_single_chunk(index: int, chunk: List[Dict]): # 按 id 查表合并：结构永不失败，缺号自动保留原文 merged_dialogs, counts = self._apply_corrections_by_id( - corrections, chunk + corrections, chunk, has_speaker ) coverage = (n - counts["kept_original"]) / n if n else 1.0 @@ -702,7 +804,7 @@ def calibrate_single_chunk(index: int, chunk: List[Dict]): return calibrated_chunks, calibration_stats - def _format_chunk_for_prompt(self, chunk: List[Dict], speaker_mapping: Dict[str, str]) -> str: + def _format_chunk_for_prompt(self, chunk: List[Dict], speaker_mapping: Dict[str, str], has_speaker: bool = True) -> str: """格式化对话块为 prompt 文本 Args: @@ -725,11 +827,16 @@ def _format_chunk_for_prompt(self, chunk: List[Dict], speaker_mapping: Dict[str, # [id] 为 ID 锚点（chunk 内 0 基下标），LLM 必须按此 id 回传修正； # 时间戳仅作上下文，不要求回传。 time_tag = f"[{start_time}]" if start_time else "" - parts.append(f"[{idx}]{time_tag}[{speaker}]: {text}") + if has_speaker: + parts.append(f"[{idx}]{time_tag}[{speaker}]: {text}") + else: + # 无说话人行格式变体：去掉 speaker 括号，避免 f-string 遇 + # None 输出字面 [None] 污染 prompt；time_tag 为空时为 [{idx}]: {text} + parts.append(f"[{idx}]{time_tag}: {text}") return "\n".join(parts) - def _build_text_from_dialogs(self, dialogs: List[Dict]) -> str: + def _build_text_from_dialogs(self, dialogs: List[Dict], has_speaker: bool = True) -> str: """从对话列表构建纯文本 Args: @@ -742,7 +849,11 @@ def _build_text_from_dialogs(self, dialogs: List[Dict]) -> str: for dialog in dialogs: speaker = dialog.get("speaker", "") text = dialog.get("text", "") - parts.append(f"{speaker}：{text}") + if has_speaker: + parts.append(f"{speaker}：{text}") + else: + # 无 speaker 变体只输出文本，不输出 `speaker：` 前缀 + parts.append(text) return "\n\n".join(parts) @@ -776,13 +887,13 @@ def _valid_correction_text(text: Any) -> bool: stripped = text.strip() if not stripped: return False - # 拒绝模型把输入行格式 `[0][S0]: ...` 原样抄回 - if re.match(r"^\[\d+\]\[", stripped): + # 拒绝模型把输入行格式原样抄回：`[0][S0]: ...`（有说话人）与 `[0]: ...`（无说话人变体） + if re.match(r"^\[\d+\](\[|:)", stripped): return False return True def _apply_corrections_by_id( - self, corrections: List[Dict], original_chunk: List[Dict] + self, corrections: List[Dict], original_chunk: List[Dict], has_speaker: bool = True ) -> tuple: """按 id 锚点合并校对结果。 @@ -837,10 +948,10 @@ def _apply_corrections_by_id( counts["kept_original"] += 1 merged_dialogs.append( { - "start_time": original.get("start_time", "00:00:00"), - "end_time": original.get("end_time", "00:00:00"), + "start_time": original.get("start_time", "00:00:00" if has_speaker else None), + "end_time": original.get("end_time", "00:00:00" if has_speaker else None), "duration": original.get("duration", 0), - "speaker": original.get("speaker", "unknown"), + "speaker": original.get("speaker", "unknown" if has_speaker else None), # speaker_id 是原始说话人标签（由 _normalize_dialog 构建）， # 之前这里重建 dialog dict 时漏掉了它——本地 # codex review 第 5 轮 F4：llm_ops.py:: @@ -870,4 +981,66 @@ def _apply_structured_fallback( if fallback_strategy == "best_quality" and best_candidate: return best_candidate - return original_chunk + return original_chunk + + def _paragraphize_no_speaker_dialogs( + self, calibrated_dialogs: List[Dict], time_snapshot: Optional[List[tuple]] + ) -> List[Dict]: + """无说话人模式：把逐段（校准后）结果确定性拼成阅读段落。 + + 段落化只选边界、不动文本。输入文本用校准后 text，时间用 normalize 前 + 快照的 float 秒（不合并时 coerced/normalized/calibrated 按下标 1:1 + 对齐）；条数不一致——如 DialogSegmenter 拆分超长单条——时回退解析 + 条目自身的 HH:MM:SS，停顿授权精度降级但不中断。输出段落格式化回 + HH:MM:SS（None 保留，不编造），不带 speaker/speaker_id 键，与 + FunASR 产物同构，直接作为最终 dialogs 落盘 llm_processed.json。 + """ + use_snapshot = time_snapshot is not None and len(time_snapshot) == len( + calibrated_dialogs + ) + segments = [] + for idx, dialog in enumerate(calibrated_dialogs): + if use_snapshot: + start_seconds, end_seconds = time_snapshot[idx] + else: + start_seconds = self._parse_time_value(dialog.get("start_time")) + end_seconds = self._parse_time_value(dialog.get("end_time")) + segments.append( + { + "text": dialog.get("text", ""), + "start_time": start_seconds, + "end_time": end_seconds, + "original_text": dialog.get("original_text"), + "duration": dialog.get("duration"), + } + ) + + paragraphs = paragraphize_segments( + segments, + target_chars=self.config.paragraphization_target_chars, + hard_max_chars=self.config.paragraphization_hard_max_chars, + pause_threshold_seconds=self.config.paragraphization_pause_threshold_seconds, + ) + + final_dialogs = [] + for paragraph in paragraphs: + dialog = { + "start_time": ( + self._format_timestamp(paragraph["start_time"]) + if paragraph["start_time"] is not None + else None + ), + "end_time": ( + self._format_timestamp(paragraph["end_time"]) + if paragraph["end_time"] is not None + else None + ), + "text": paragraph["text"], + } + # duration / original_text 由 paragraphize 输出透传（有则带） + if "duration" in paragraph: + dialog["duration"] = paragraph["duration"] + if "original_text" in paragraph: + dialog["original_text"] = paragraph["original_text"] + final_dialogs.append(dialog) + return final_dialogs diff --git a/src/video_transcript_api/llm/prompts/__init__.py b/src/video_transcript_api/llm/prompts/__init__.py index 083bede6..b2fa06c0 100644 --- a/src/video_transcript_api/llm/prompts/__init__.py +++ b/src/video_transcript_api/llm/prompts/__init__.py @@ -426,6 +426,105 @@ def build_summary_user_prompt( Return ONLY the proofread JSON, without any additional explanations or comments.""" +# ============================================================ +# 结构化校对任务 Prompt 模板（无说话人变体） +# ============================================================ +# 与 STRUCTURED_CALIBRATE 系列同契约（{id,text} 输出、禁止合并/拆分/增删/重排）， +# 仅行格式描述去掉 [说话人] 括号，与 SpeakerAwareProcessor._format_chunk_for_prompt +# 的 has_speaker=False 行生成保持一致。 + +STRUCTURED_CALIBRATE_NO_SPEAKER_SYSTEM_PROMPT = """你是专业的中文文本校对助手。你的任务是对语音转录（ASR）文本进行逐段校对。 + +## 核心要求（必须遵守） + +输入每一行格式为 `[id][HH:MM:SS]: 内容`（该段无时间戳时为 `[id]: 内容`），其中 `[id]` 是该段的编号（锚点）。 + +1. **按 id 锚点返回**：对每一段输入，返回一个 `{id, text}` 对象，`id` 与输入行的 `[id]` 完全一致 +2. **每段都要返回**：即使某段无需修改，也要原样返回该 id 的 text；不要遗漏任何 id +3. **只改文本**：`text` 只包含校对后的内容，**不要**包含 id 或时间戳 +4. **禁止合并、拆分、增删、重排**：一个 id 对应一段，不得把多段并入一个 text，也不得拆成多个 + +## 专有名词修正（最高优先级） + +用户会提供一份**关键信息列表**，其中包含本次内容涉及的正确人名、品牌、术语等。 +ASR 转录经常将这些专有名词错误识别为同音字或近音字。你必须： +1. **逐一检查**关键信息列表中的每个词，在转录文本中寻找可能的 ASR 误识别 +2. 将 ASR 误识别的文字**替换为关键信息列表中的正确拼写** +3. 常见的 ASR 专有名词错误模式： + - 英文品牌被拆成无意义的中文字/拼音（如 "哥斯 r 哦" → "Cursor"、"欧普斯" → "Opus"） + - 中文名词被识别为同音字（如 "海陆" → "海螺"、"金业酒店" → "今夜酒店"） + - 英文缩写被错误拼写（如 "IR" → "ARR"、"PMFF" → "PMF"） + - 英文单词被拆成多个片段（如 "mini max" → "MiniMax"） + +## 校对规则 + +1. **只能在单个段落内部进行修改**，不得跨段落操作 +2. **最小改动原则**：只做必要的校对与通顺化；若有多种改法，选择更接近原句的版本 +3. 修正明显的错别字和语法错误，优先修正"读起来明显不通顺"的表达 +4. 调整标点符号的使用，确保其正确性和一致性 +5. 如有必要，可以**适度调整语序**以提高可读性，但不要改变原意或强调重点 +6. 可以**删减少量无意义的口头禅/口误/重复词**（如"那个…就是…就…"），但不要删掉关键信息或语气词 +7. 可对明显口误做纠正（如把"任一/套生/压力面"等修成通顺表达） +8. 保留原文中的口语化表达和说话者的语气特点，避免过度书面化 +9. **保留语义与事实**：不要改变人称、否定/肯定、因果关系、时间顺序、数字/金额/数量等核心信息 +10. **不确定时偏保守**：宁可轻微修整，也不要重写或引入新含义 +11. **重点修正 ASR 常见错误**（不改变原意）： + - 叠字与重复：例如"遇遇不到/薪薪十多万万/约约咖啡馆"等 + - 错分词或错字：例如"a 本大学/任一/身出市场/套生大学"等明显不通顺表达 + - 英文字母或符号误入导致的断裂（如 a 本 → 一本） + - 同音词误识别导致的歧义（结合上下文与关键信息列表修正） +12. **保持完整性**：不要新增或删减实质内容，不要压缩或概括 +13. 不要解释或评论文本内容 + +## 输出格式 + +必须输出 JSON 格式，包含 corrections 数组，每项为 `{"id": 整数, "text": "校对后文本"}`。 +- `id` 与输入行的 `[id]` 一一对应，必须覆盖每一个输入 id +- `text` 只放校对后的内容，不要带 id、时间戳或行格式标签 + +只返回校对后的JSON，不要包含任何其他解释或评论。""" + +STRUCTURED_CALIBRATE_NO_SPEAKER_SYSTEM_PROMPT_EN = """You are a professional English text proofreading assistant. Your task is to proofread speech-to-text (ASR) transcription text segment by segment. + +## Core Requirements (MUST follow) + +Each input line has the format `[id][HH:MM:SS]: content` (or `[id]: content` when the segment has no timestamp), where `[id]` is the anchor index of that segment. + +1. **Return by id anchor**: for every input segment, return one `{id, text}` object whose `id` exactly matches the input line's `[id]` +2. **Return every segment**: even if a segment needs no change, return its text unchanged under its id; do NOT omit any id +3. **Only edit text**: `text` contains ONLY the proofread content, NOT the id or timestamp +4. **Do NOT merge, split, add, remove, or reorder segments**: one id maps to exactly one segment + +## Proofreading Rules + +1. **Only modify within a single segment**, do not operate across segments +2. **Minimal changes principle**: only make necessary corrections; when multiple options exist, choose the one closest to the original +3. Fix obvious typos and grammar errors, prioritizing expressions that are clearly unnatural +4. Adjust punctuation for correctness and consistency +5. If necessary, **slightly adjust word order** to improve readability, but do not change the meaning +6. May **remove minor filler words/speech errors/repetitions** (e.g., "um", "like like", "you know you know"), but do not remove key information +7. Correct obvious speech errors and ASR misrecognitions (e.g., homophones, run-on words, incorrectly split words) +8. Preserve colloquial expressions and the speaker's tone/style, avoid over-formalizing +9. **Preserve semantics and facts**: do not change person, negation/affirmation, causality, temporal order, numbers/amounts +10. **When uncertain, be conservative**: prefer minor polishing over rewriting or introducing new meaning +11. **Focus on common ASR errors** (without changing meaning): + - Homophones: e.g., "their/there/they're", "your/you're", "its/it's" + - Run-on or incorrectly split words + - Missing or incorrect punctuation from speech-to-text + - Misrecognized proper nouns (correct using reference information) +12. **Maintain completeness**: do not add or remove substantive content, do not compress or summarize +13. Do not explain or comment on the text content +14. **Do NOT translate**: Keep ALL content in English. Do NOT translate to any other language. + +## Output Format + +Must output JSON format containing a corrections array, each item being `{"id": integer, "text": "proofread text"}`. +- `id` maps one-to-one to the input line's `[id]`, and MUST cover every input id +- `text` holds ONLY the proofread content, without id, timestamp, or line-format tags + +Return ONLY the proofread JSON, without any additional explanations or comments.""" + + def build_structured_calibrate_user_prompt( input_data: dict = None, dialogs_text: str = None, diff --git a/src/video_transcript_api/llm/segmenters/dialog_segmenter.py b/src/video_transcript_api/llm/segmenters/dialog_segmenter.py index 3273451a..b5a267f1 100644 --- a/src/video_transcript_api/llm/segmenters/dialog_segmenter.py +++ b/src/video_transcript_api/llm/segmenters/dialog_segmenter.py @@ -4,7 +4,7 @@ """ import re -from typing import List, Dict, Any +from typing import List, Dict, Any, Optional from ...utils.logging import setup_logger from ..core.config import LLMConfig @@ -15,16 +15,31 @@ class DialogSegmenter: """有说话人文本分段器""" - def __init__(self, config: LLMConfig): + def __init__( + self, + config: LLMConfig, + preferred_chunk_length: Optional[int] = None, + max_chunk_length: Optional[int] = None, + ): """初始化对话分段器 Args: config: LLM 配置 + preferred_chunk_length: 理想块长覆盖（None 时取 + config.preferred_chunk_length）。无说话人（plain 源）模式 + 用独立分块参数覆盖，has_speaker=True 路径不传、行为不变。 + max_chunk_length: 最大块长覆盖（None 时取 config.max_chunk_length） """ self.config = config self.min_chunk_length = config.min_chunk_length - self.max_chunk_length = config.max_chunk_length - self.preferred_chunk_length = config.preferred_chunk_length + self.max_chunk_length = ( + max_chunk_length if max_chunk_length is not None else config.max_chunk_length + ) + self.preferred_chunk_length = ( + preferred_chunk_length + if preferred_chunk_length is not None + else config.preferred_chunk_length + ) def segment(self, dialogs: List[Dict]) -> List[List[Dict]]: """对对话列表进行智能分块 diff --git a/tests/unit/test_speaker_aware_no_speaker.py b/tests/unit/test_speaker_aware_no_speaker.py new file mode 100644 index 00000000..02c2d336 --- /dev/null +++ b/tests/unit/test_speaker_aware_no_speaker.py @@ -0,0 +1,486 @@ +"""T8: SpeakerAwareProcessor has_speaker=False（无说话人逐段校对）模式测试。 + +无 speaker 的 segments（plain 源：CapsWriter/YouTube 字幕）走「无说话人逐段 +校对」：结构与 FunASR 路径同构（id 锚点映射、禁止合并/拆分/重排），但 + +- SpeakerInferencer.infer 整步跳过（零 LLM 调用）； +- 不做「连续同说话人合并」（两边 speaker_id 都是 None 会被误判成同一人， + 全文塌缩成一条）； +- speaker/speaker_id 保留缺省（不落 "unknown"）、None 时间保留 None + （不落 "00:00:00"）； +- 最终 dialogs 是确定性段落化后的段落（无 speaker 键），可直接落盘 + llm_processed.json 并支撑 dlg-{i} 锚点渲染。 + +规格来源：docs/sessions/260719-0513-chapters/TASKS.md T8。 +""" + +import json +import re +from unittest.mock import MagicMock + +import pytest + +from video_transcript_api.llm.core.config import LLMConfig +from video_transcript_api.llm.core.key_info_extractor import KeyInfo +from video_transcript_api.llm.processors.speaker_aware_processor import ( + SpeakerAwareProcessor, +) +from video_transcript_api.llm.prompts import ( + STRUCTURED_CALIBRATE_NO_SPEAKER_SYSTEM_PROMPT, + STRUCTURED_CALIBRATE_SYSTEM_PROMPT, +) +from video_transcript_api.utils.llm_status import CalibrationStatus + + +def _make_config(**overrides): + """真实 LLMConfig（非 MagicMock），DialogSegmenter/段落化的算术才能跑。 + + 默认把段落化参数调小，方便用短文本构造断点场景。 + """ + kwargs = dict( + api_key="k", + base_url="http://test", + calibrate_model="test-model", + summary_model="test-model", + paragraphization_target_chars=10, + paragraphization_hard_max_chars=100, + paragraphization_pause_threshold_seconds=2.0, + ) + kwargs.update(overrides) + return LLMConfig(**kwargs) + + +def _empty_key_info(): + return KeyInfo( + names=[], + places=[], + technical_terms=[], + brands=[], + abbreviations=[], + foreign_terms=[], + other_entities=[], + ) + + +def _calibrate_side_effect(**kwargs): + """按 user_prompt 里的 [id] 行生成全覆盖 corrections（合法非空文本）。 + + coverage=1.0 → 每个 chunk 干净成功，不走重试/降级，产物结构完全由 + _apply_corrections_by_id 决定。 + """ + user_prompt = kwargs["user_prompt"] + corrections = [] + for line in user_prompt.splitlines(): + m = re.match(r"^\[(\d+)\]", line.strip()) + if m: + corrections.append( + {"id": int(m.group(1)), "text": f"校准文本{m.group(1)}。"} + ) + response = MagicMock() + response.structured_output = {"corrections": corrections} + return response + + +def _make_processor(config=None, llm_client=None, speaker_inferencer=None): + """组装 processor：llm_client/key_info_extractor/speaker_inferencer/ + quality_validator 全部 mock（参照 test_speaker_aware_processor_stats.py）。""" + if config is None: + config = _make_config() + if llm_client is None: + llm_client = MagicMock() + llm_client.call = MagicMock(side_effect=_calibrate_side_effect) + key_info_extractor = MagicMock() + key_info_extractor.extract = MagicMock(return_value=_empty_key_info()) + if speaker_inferencer is None: + speaker_inferencer = MagicMock() + speaker_inferencer.infer = MagicMock( + return_value={"mapping": {}, "meta": {}, "source": "identity_fallback"} + ) + quality_validator = MagicMock() + processor = SpeakerAwareProcessor( + config=config, + llm_client=llm_client, + key_info_extractor=key_info_extractor, + speaker_inferencer=speaker_inferencer, + quality_validator=quality_validator, + ) + return processor, llm_client, key_info_extractor, speaker_inferencer + + +def _prompt_id_lines(user_prompt): + """提取 user_prompt 里以 [id] 开头的数据行。""" + return [ + line + for line in user_prompt.splitlines() + if re.match(r"^\[\d+\]", line.strip()) + ] + + +# ---------------------------------------------------------------------- +# 1. has_speaker 自动判定 +# ---------------------------------------------------------------------- + + +class TestHasSpeakerAutoDetection: + def test_no_speaker_input_detected_as_false(self): + """全无 speaker 的输入自动走无说话人模式：infer 零调用。""" + processor, llm_client, _, speaker_inferencer = _make_processor() + dialogs = [ + {"text": "第一句话。", "start_time": 0.0, "end_time": 1.0}, + {"text": "第二句话。", "start_time": 1.0, "end_time": 2.0}, + ] + + processor.process(dialogs=dialogs, title="Test", skip_calibration=True) + + speaker_inferencer.infer.assert_not_called() + + def test_mixed_input_detected_as_true(self): + """混合输入（部分有 speaker）维持现状 has_speaker=True 路径： + infer 被调用、产物保留 speaker 键、不段落化。""" + processor, _, _, speaker_inferencer = _make_processor() + speaker_inferencer.infer.return_value = { + "mapping": {"S1": "Alice"}, + "meta": {}, + "source": "llm", + } + dialogs = [ + {"speaker": "S1", "text": "有说话人的一段。", "start_time": 0.0, "end_time": 1.0}, + {"text": "缺说话人的一段。", "start_time": 1.0, "end_time": 2.0}, + ] + + result = processor.process(dialogs=dialogs, title="Test", skip_calibration=True) + + speaker_inferencer.infer.assert_called_once() + out_dialogs = result["structured_data"]["dialogs"] + # 现状路径：每条都有 speaker 键（缺省段按现状塞 "unknown"），且不段落化 + assert len(out_dialogs) == 2 + assert all("speaker" in d for d in out_dialogs) + assert out_dialogs[0]["speaker"] == "Alice" + + def test_explicit_has_speaker_overrides_auto_detection(self): + """显式传 has_speaker=True 覆盖自动判定：全无 speaker 的输入也走 + 现状路径（coerce 塞 "unknown"、按同说话人合并塌缩成一条）。""" + processor, _, _, speaker_inferencer = _make_processor() + dialogs = [ + {"text": "第一段。", "start_time": 0.0, "end_time": 1.0}, + {"text": "第二段。", "start_time": 1.0, "end_time": 2.0}, + ] + + result = processor.process( + dialogs=dialogs, title="Test", skip_calibration=True, has_speaker=True + ) + + speaker_inferencer.infer.assert_called_once() + out_dialogs = result["structured_data"]["dialogs"] + # 现状合并行为：speaker_id 都是 "unknown" → 塌缩成一条 + assert len(out_dialogs) == 1 + assert out_dialogs[0]["speaker"] == "unknown" + + +# ---------------------------------------------------------------------- +# 2. 序列化硬断言（落盘契约） +# ---------------------------------------------------------------------- + + +class TestNoSpeakerSerialization: + def test_serialized_structured_data_has_no_unknown_no_speaker_keys(self): + """json.dumps(structured_data) 全文不含 "unknown"；dialogs 无 + speaker/speaker_id 键；保留 start/end/text 字段。""" + processor, _, _, _ = _make_processor() + dialogs = [ + {"text": "第一段内容。", "start_time": 0.0, "end_time": 5.0}, + {"text": "第二段内容。", "start_time": 5.0, "end_time": 10.0}, + ] + + result = processor.process(dialogs=dialogs, title="Test") + + serialized = json.dumps(result["structured_data"], ensure_ascii=False) + assert "unknown" not in serialized + for dialog in result["structured_data"]["dialogs"]: + assert "speaker" not in dialog + assert "speaker_id" not in dialog + assert "start_time" in dialog + assert "end_time" in dialog + assert "text" in dialog + + def test_calibrated_text_has_no_speaker_prefix(self): + """calibrated_text 无 "None："/"unknown：" 前缀，只是段落文本的 + \\n\\n 连接。""" + processor, _, _, _ = _make_processor() + dialogs = [ + {"text": "第一段内容。", "start_time": 0.0, "end_time": 5.0}, + {"text": "第二段内容。", "start_time": 5.0, "end_time": 10.0}, + ] + + result = processor.process(dialogs=dialogs, title="Test") + + calibrated_text = result["calibrated_text"] + assert "None：" not in calibrated_text + assert "unknown：" not in calibrated_text + paragraphs = result["structured_data"]["dialogs"] + assert calibrated_text == "\n\n".join(p["text"] for p in paragraphs) + + def test_stats_dialog_count_is_paragraph_count(self): + """无说话人模式 stats.dialog_count = 段落数（与落盘 dialogs 一致）。""" + processor, _, _, _ = _make_processor() + dialogs = [ + {"text": "第一段内容。", "start_time": 0.0, "end_time": 5.0}, + {"text": "第二段内容。", "start_time": 5.0, "end_time": 10.0}, + ] + + result = processor.process(dialogs=dialogs, title="Test") + + assert result["stats"]["dialog_count"] == len( + result["structured_data"]["dialogs"] + ) + + +# ---------------------------------------------------------------------- +# 3. infer 零调用 / key_info 提取规则 +# ---------------------------------------------------------------------- + + +class TestInferAndKeyInfo: + def test_infer_not_called_and_key_info_extracted_when_calibrating(self): + """校准开启：SpeakerInferencer.infer 零调用，key_info 仍提取 + （喂养校对 prompt，与 speaker 无关）。""" + processor, _, key_info_extractor, speaker_inferencer = _make_processor() + dialogs = [{"text": "内容。", "start_time": 0.0, "end_time": 1.0}] + + result = processor.process(dialogs=dialogs, title="Test") + + speaker_inferencer.infer.assert_not_called() + key_info_extractor.extract.assert_called_once() + # identity_fallback source 保持下游 llm_ops 刷新判断安全 + assert result["stats"]["speaker_inference_source"] == "identity_fallback" + assert result["structured_data"]["speaker_mapping"] == {} + + def test_key_info_not_extracted_when_skip_calibration(self): + """skip_calibration=True 且无 speaker（推断步骤也不存在）时， + key_info 无消费者，整步不提取（无隐藏 LLM 调用）。""" + processor, llm_client, key_info_extractor, speaker_inferencer = _make_processor() + dialogs = [{"text": "内容。", "start_time": 0.0, "end_time": 1.0}] + + processor.process(dialogs=dialogs, title="Test", skip_calibration=True) + + speaker_inferencer.infer.assert_not_called() + key_info_extractor.extract.assert_not_called() + llm_client.call.assert_not_called() + + +# ---------------------------------------------------------------------- +# 4. 不合并：校准前保持原始 segments 粒度 +# ---------------------------------------------------------------------- + + +class TestNoMergingBeforeCalibration: + def test_dialogs_not_collapsed_into_one_before_calibration(self): + """无 speaker 的多条输入在校准 prompt 里必须各占一行（[0]/[1]/[2]）。 + 若 _normalize_and_merge_dialogs 按 speaker_id==None 合并，全文会 + 塌缩成一条，prompt 里只剩 [0] 一行。""" + processor, llm_client, _, _ = _make_processor() + dialogs = [ + {"text": "第一段。", "start_time": 0.0, "end_time": 1.0}, + {"text": "第二段。", "start_time": 1.0, "end_time": 2.0}, + {"text": "第三段。", "start_time": 2.0, "end_time": 3.0}, + ] + + processor.process(dialogs=dialogs, title="Test") + + llm_client.call.assert_called_once() + user_prompt = llm_client.call.call_args.kwargs["user_prompt"] + id_lines = _prompt_id_lines(user_prompt) + assert len(id_lines) == 3 + assert id_lines[0].startswith("[0]") + assert id_lines[1].startswith("[1]") + assert id_lines[2].startswith("[2]") + + +# ---------------------------------------------------------------------- +# 5. None 时间保留 None +# ---------------------------------------------------------------------- + + +class TestNoneTimesPreserved: + def test_none_times_stay_none_no_zero_fallback(self): + """无时间输入：段落 start_time/end_time 为 None，序列化产物里 + 不出现 "00:00:00" 兜底标签。""" + processor, _, _, _ = _make_processor() + dialogs = [ + {"text": "第一段内容。"}, + {"text": "第二段内容。"}, + ] + + result = processor.process(dialogs=dialogs, title="Test", skip_calibration=True) + + out_dialogs = result["structured_data"]["dialogs"] + assert len(out_dialogs) >= 1 + for dialog in out_dialogs: + assert dialog["start_time"] is None + assert dialog["end_time"] is None + serialized = json.dumps(result["structured_data"], ensure_ascii=False) + assert "00:00:00" not in serialized + + +# ---------------------------------------------------------------------- +# 6. 确定性段落化集成 +# ---------------------------------------------------------------------- + + +class TestParagraphizationIntegration: + def test_output_dialogs_are_paragraphs_broken_at_authorized_points(self): + """输出 dialogs 是段落：长度到预算后在句末授权点断段（不腰斩句子）。""" + processor, _, _, _ = _make_processor() + dialogs = [ + {"text": "啊啊啊啊啊", "start_time": 0.0, "end_time": 5.0}, + {"text": "啊啊啊啊啊啊。", "start_time": 5.5, "end_time": 10.0}, + {"text": "呜呜呜呜呜", "start_time": 10.5, "end_time": 15.0}, + {"text": "呜呜呜呜呜呜。", "start_time": 15.5, "end_time": 20.0}, + ] + + result = processor.process(dialogs=dialogs, title="Test", skip_calibration=True) + + out = result["structured_data"]["dialogs"] + # 4 条输入 → 2 段：断点落在 seg1/seg3（句末标点）之后 + assert len(out) == 2 + assert out[0]["text"] == "啊啊啊啊啊" + "啊啊啊啊啊啊。" + assert out[1]["text"] == "呜呜呜呜呜" + "呜呜呜呜呜呜。" + # 段落时间 = 首成员 start / 末成员 end，格式化 HH:MM:SS + assert out[0]["start_time"] == "00:00:00" + assert out[0]["end_time"] == "00:00:10" + assert out[1]["start_time"] == "00:00:10" + assert out[1]["end_time"] == "00:00:20" + + def test_pause_authorization_consumes_float_second_snapshot(self): + """停顿授权消费 normalize 前的 float 秒快照，不被 HH:MM:SS 截断 + 误差带偏：gap=1.5s（12.1-10.6）< 2.0 不断；gap=2.5s（13.1-10.6） + >= 2.0 断。若误用截断值（10 vs 12），前者会被误判成 gap=2.0 断开。""" + processor, _, _, _ = _make_processor() + + base = [ + {"text": "AAAAAAAAAA", "start_time": 0.0, "end_time": 10.6}, + {"text": "BBBBBBBBBB", "start_time": 12.1, "end_time": 20.0}, + ] + result = processor.process(dialogs=base, title="Test", skip_calibration=True) + assert len(result["structured_data"]["dialogs"]) == 1 + + wider = [ + {"text": "AAAAAAAAAA", "start_time": 0.0, "end_time": 10.6}, + {"text": "BBBBBBBBBB", "start_time": 13.1, "end_time": 20.0}, + ] + result = processor.process(dialogs=wider, title="Test", skip_calibration=True) + assert len(result["structured_data"]["dialogs"]) == 2 + + def test_calibrated_branch_also_paragraphizes(self): + """校准分支同样段落化：段落 text 来自校准后文本，并携带 + original_text（原文拼接）与 duration。""" + processor, llm_client, _, _ = _make_processor() + dialogs = [ + {"text": "第一段内容。", "start_time": 0.0, "end_time": 5.0, "duration": 5.0}, + {"text": "第二段内容。", "start_time": 5.0, "end_time": 10.0, "duration": 5.0}, + ] + + result = processor.process(dialogs=dialogs, title="Test") + + llm_client.call.assert_called_once() + out = result["structured_data"]["dialogs"] + assert len(out) == 1 # 总长度未到 target=10 之后的授权点之外，合成一段 + assert out[0]["text"] == "校准文本0。校准文本1。" + assert out[0]["original_text"] == "第一段内容。第二段内容。" + assert out[0]["duration"] == 10.0 + + +# ---------------------------------------------------------------------- +# 7. skip_calibration 分支 +# ---------------------------------------------------------------------- + + +class TestSkipCalibration: + def test_skip_calibration_still_paragraphizes_with_zero_llm_calls(self): + """skip_calibration=True：仍做确定性段落化、零 chunk LLM 调用、 + calibration_status=DISABLED（与 FunASR calibrate=false 形态一致）。""" + processor, llm_client, _, _ = _make_processor() + dialogs = [ + {"text": "啊啊啊啊啊", "start_time": 0.0, "end_time": 5.0}, + {"text": "啊啊啊啊啊啊。", "start_time": 5.5, "end_time": 10.0}, + {"text": "呜呜呜呜呜", "start_time": 10.5, "end_time": 15.0}, + {"text": "呜呜呜呜呜呜。", "start_time": 15.5, "end_time": 20.0}, + ] + + result = processor.process(dialogs=dialogs, title="Test", skip_calibration=True) + + llm_client.call.assert_not_called() + assert ( + result["stats"]["calibration_stats"]["calibration_status"] + == CalibrationStatus.DISABLED + ) + out = result["structured_data"]["dialogs"] + assert len(out) == 2 + # 未校准：段落 text 即原文拼接 + assert out[0]["text"] == "啊啊啊啊啊" + "啊啊啊啊啊啊。" + + +# ---------------------------------------------------------------------- +# 8. prompt 行格式与 echo 拒绝 +# ---------------------------------------------------------------------- + + +class TestPromptFormat: + def test_format_chunk_no_speaker_line_format(self): + """无说话人行格式：[{idx}]{time_tag}: {text}（去掉 speaker 括号； + time_tag 为空时为 [{idx}]: {text}），f-string 不得输出字面 [None]。""" + processor, _, _, _ = _make_processor() + chunk = [ + {"text": "你好", "start_time": "00:00:01"}, + {"text": "世界", "start_time": None}, + ] + + out = processor._format_chunk_for_prompt(chunk, {}, has_speaker=False) + + assert out == "[0][00:00:01]: 你好\n[1]: 世界" + + def test_format_chunk_with_speaker_line_format_unchanged(self): + """has_speaker=True 行格式现状不变：[{idx}]{time_tag}[{speaker}]: {text}。""" + processor, _, _, _ = _make_processor() + chunk = [{"text": "你好", "start_time": "00:00:01", "speaker": "S0"}] + + out = processor._format_chunk_for_prompt(chunk, {}, has_speaker=True) + + assert out == "[0][00:00:01][S0]: 你好" + + def test_calibrate_uses_no_speaker_system_prompt(self): + """无说话人模式校准选用 NO_SPEAKER system prompt（行格式描述与 + 实际行生成一致）。""" + processor, llm_client, _, _ = _make_processor() + dialogs = [{"text": "内容。", "start_time": 0.0, "end_time": 1.0}] + + processor.process(dialogs=dialogs, title="Test") + + system_prompt = llm_client.call.call_args.kwargs["system_prompt"] + assert system_prompt == STRUCTURED_CALIBRATE_NO_SPEAKER_SYSTEM_PROMPT + assert system_prompt != STRUCTURED_CALIBRATE_SYSTEM_PROMPT + + def test_echo_rejection_covers_no_speaker_line_format(self): + """echo 拒绝正则同时覆盖两种行格式的回吐： + `[0]: fake text`（无 speaker 新格式）与 + `[0][00:00:01][S0]: fake`（有 speaker 旧格式）。""" + assert SpeakerAwareProcessor._valid_correction_text("[0]: fake text") is False + assert ( + SpeakerAwareProcessor._valid_correction_text("[0][00:00:01][S0]: fake") + is False + ) + assert SpeakerAwareProcessor._valid_correction_text("正常校对文本") is True + + def test_build_text_no_speaker_variant(self): + """无 speaker 变体只输出 text（\\n\\n 连接），不输出 `speaker：` 前缀。""" + processor, _, _, _ = _make_processor() + + out = processor._build_text_from_dialogs( + [{"text": "第一段"}, {"text": "第二段"}], has_speaker=False + ) + + assert out == "第一段\n\n第二段" + + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) From 03dadb5b710829bbfcc1e55e4bed7b7dcfe4dc2f Mon Sep 17 00:00:00 2001 From: zj1123581321 Date: Sun, 19 Jul 2026 12:24:24 -0400 Subject: [PATCH 3/3] =?UTF-8?q?docs:=20T8=20S3=20=E6=94=B6=E5=B7=A5?= =?UTF-8?q?=E4=BA=A4=E6=8E=A5=EF=BC=88HANDOFF=EF=BC=89=E4=B8=8E=20review?= =?UTF-8?q?=20=E6=97=A5=E5=BF=97=E5=BB=BA=E6=A1=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 上个实施 session 因 provider 2MB 消息上限报废（compaction 亦超限失败）， 状态以本 HANDOFF + git 历史为准恢复。 --- .../260719-1155-t8-implementation/HANDOFF.md | 52 +++++++++++++++++++ .../REVIEW-LOG.md | 17 ++++++ 2 files changed, 69 insertions(+) create mode 100644 docs/sessions/260719-1155-t8-implementation/HANDOFF.md create mode 100644 docs/sessions/260719-1155-t8-implementation/REVIEW-LOG.md diff --git a/docs/sessions/260719-1155-t8-implementation/HANDOFF.md b/docs/sessions/260719-1155-t8-implementation/HANDOFF.md new file mode 100644 index 00000000..4b7b921c --- /dev/null +++ b/docs/sessions/260719-1155-t8-implementation/HANDOFF.md @@ -0,0 +1,52 @@ +# T8 实施 session —— 交接（HANDOFF） + +> 创建：2026-07-19（第二个实施 session 开头） +> 任务：T8 阶段二逐段校对推广 v1（确定性段落化），规格唯一来源：`docs/sessions/260719-0513-chapters/TASKS.md` 的 T8 节 +> 分支：`feat/chapters-wiring`（本地，未 push） + +## 前情：上一个 session 为什么中断 + +第一个实施 session 主循环累积约 60 万 token（消息体 2.1MB），超过 provider 2MB +硬上限；compaction 请求本身也超限失败，session 死锁报废（导出留档： +仓库根目录 `kimi-export-session_-20260719-161035.md`，仅供参考，不要整份喂回上下文）。 +本交接文件 + git 提交历史是恢复的完整依据。 + +**教训（本 session 起执行）**： +- 大件探查/实现一律走 subagent，主循环只收结论，不整读几千行文件； +- 进度随时落盘到本目录；消息体接近上限前主动收工换新 session。 + +## 已完成（全部已提交，git log 为准） + +| 步 | commit | 内容 | +|----|--------|------| +| 方案修订 | `13b00cb` | T8 方案 v1 确定性段落化，增量 review 收敛 | +| S1 | `62aca84` | 确定性段落化工具 paragraphize（长度预算+授权断点），`transcriber/paragraphize.py` | +| S2 | `812ef97` | `structured_calibration_for_plain` 开关 + 段落化/分块配置字段 | +| S5 | `0a55f09` | 渲染器无 speaker 防御修复 + plain_structured 开关回退门控 | +| S3 | `822bb1b` | SpeakerAwareProcessor has_speaker 模式（无说话人逐段校对，零 SpeakerInferencer 调用，落盘前段落化）+ 19 个单测 | + +S3 提交时 `uv run pytest tests/unit` 全绿（exit 0，约 2400 项；warnings 摘要会 +盖住通过数行，以 exit code 为准）。 + +## 待办（按序） + +1. **S4 — llm_ops 接线**（`api/services/llm_ops.py`，当前未接线，grep 无 + `plain_structured`/`has_speaker` 痕迹）： + - `_prepare_llm_content`：**仅 `calibrate_requested=True`**（含 recalibrate）且 + plain 源有 segments → 返回 list（协调器 isinstance 路由自然生效）； + calibrate=false 补层任务维持纯文本（防指纹永久 mismatch → 永久 nolink）； + - structured 落盘 gate（现仅 `use_speaker_recognition`）扩展到本模式； + - plain 结构化产物写 provenance `"mode": "plain_structured"`； + - `llm_calibrated.txt` 继续生成（无前缀变体，保 export/raw 兼容）。 +2. **S6 — 集成测试**：开关开全链路（plain 源 → llm_processed.json 落盘 → + 渲染带 `dlg-{i}` 锚点 → 章节 fingerprint 匹配 → `jump_ok=1`）； + 开关关回归（plain 路径行为与现状一致）。验收全集见 TASKS.md T8「完成标准」。 +3. **review 循环**：独立 subagent review ≤20 轮；P1 必修，P2/P3 可判「接受不修」 + 记 `REVIEW-LOG.md` backlog；gate = 连续 2 轮无新增 P1。 + +## 纪律提醒 + +- 开关 `llm.structured_calibration_for_plain` 默认 **false**（暗启动，T9 验收后才翻 true）； +- `llm_processed.json` 序列化全文不含 "unknown"（硬断言）； +- 减法优先，新机制仅用于消除 P1； +- 勿把 worktree 本地 config 的 `min_chapters_threshold=1000` 当生产默认合入（示例默认 10000）。 diff --git a/docs/sessions/260719-1155-t8-implementation/REVIEW-LOG.md b/docs/sessions/260719-1155-t8-implementation/REVIEW-LOG.md new file mode 100644 index 00000000..f28a52ad --- /dev/null +++ b/docs/sessions/260719-1155-t8-implementation/REVIEW-LOG.md @@ -0,0 +1,17 @@ +# T8 实施 session —— review 日志与 backlog + +> 创建：2026-07-19 +> 任务：T8 阶段二逐段校对推广 v1（确定性段落化），任务卡 `docs/sessions/260719-0513-chapters/TASKS.md` +> 纪律：独立 subagent review ≤20 轮；P1（正确性/安全/数据丢失）必修；P2/P3 可判「接受不修」记本文件 backlog 并写理由；gate = 连续 2 轮无新增 P1；减法优先，新机制仅用于消除 P1。 + +## Review 轮次记录 + +| 轮次 | reviewer | 新增 P1 | P2/P3 新增 | 处置 | +|------|----------|---------|------------|------| +| （待第 1 轮） | | | | | + +## Backlog（接受不修的 P2/P3） + +| 日期 | 轮次 | 发现 | 级别 | 接受理由 | +|------|------|------|------|----------| +| （暂无） | | | | |