"""断句 + 时间戳重算（纯算法，零模型开销）。 Whisper 原始 segment 的断句通常很混乱：每段不一定是完整句子，时间戳也不对齐句界。 本模块基于词级时间戳重新断句，得到规范的字幕条目。 两路策略： 1. 精确路（有 word_timestamps）：按句末标点（. ! ? ;）切句，超长句再按逗号拆， 每条字幕的时间戳直接取首词.start ~ 末词.end，精确无误。 2. 匀速估算路（无 word_timestamps）：段内按字符数比例分配时间—— 句start = 段start + (前缀字符数 / 段总字符数) × 段时长。 即「短时匀速」假设，无需大模型。 最后做 SRT 规范化：单条 1–7 秒、≤2 行、每行 ≤42 字符。 """ from __future__ import annotations import logging import re from ..config import get_settings from .types import Segment, Subtitle, Word logger = logging.getLogger("audio2text.segmenter") # 句末标点：句号、感叹号、问号、分号 _SENTENCE_END = re.compile(r"[.!?;]+") # 句内停顿：逗号、冒号、破折号 _CLAUSE_BREAK = re.compile(r"[,:\-—]+") def resegment(segments: list[Segment]) -> list[Subtitle]: """把 ASR segments 重组为规范字幕条目。 Args: segments: asr_service.transcribe 的输出（可能含词级时间戳） Returns: list[Subtitle]，已做时长/字数规范化。 """ cfg = get_settings().segmentation max_words = cfg.max_words_per_line max_dur = cfg.max_duration_seconds min_dur = cfg.min_duration_seconds max_chars = cfg.max_chars_per_line # 第一步：把所有词串成大列表（精确路）或退化为段级（估算路） all_words: list[Word] = [] has_word_ts = True for seg in segments: if not seg.words: has_word_ts = False break all_words.extend(seg.words) if has_word_ts and all_words: subs = _resegment_by_words(all_words, max_words, max_dur) else: logger.warning("无词级时间戳，退化为匀速估算路。") subs = _resegment_by_estimate(segments, max_words, max_dur) # 规范化：合并过短条目、拆分行宽 subs = _normalize(subs, min_dur, max_chars) logger.debug("断句完成：%d 条字幕。", len(subs)) return subs # ---------------- 精确路：按词级时间戳 ---------------- def _resegment_by_words( words: list[Word], max_words: int, max_dur: float, ) -> list[Subtitle]: """按句末标点切句，超长句按逗号拆，时间戳取首末词。""" subs: list[Subtitle] = [] # 当前句的词 current: list[Word] = [] def flush(wlist: list[Word]) -> None: if not wlist: return text = " ".join(w.text for w in wlist).strip() if not text: return subs.append(Subtitle( text=text, start=wlist[0].start, end=wlist[-1].end, )) for w in words: current.append(w) # 句末标点 → 收尾 if _SENTENCE_END.search(w.text): _maybe_split_and_flush(current, max_words, max_dur, flush) current = [] continue # 超长（词数或时长）→ 优先在最近的逗号处断 cur_dur = (current[-1].end - current[0].start) if len(current) > 1 else 0 if len(current) >= max_words or cur_dur >= max_dur: _maybe_split_and_flush(current, max_words, max_dur, flush) current = [] flush(current) return subs def _maybe_split_and_flush( wlist: list[Word], max_words: int, max_dur: float, flush, ) -> None: """若 wlist 过长，在逗号处再拆；否则整条 flush。""" if len(wlist) <= max_words and (len(wlist) <= 1 or wlist[-1].end - wlist[0].start < max_dur): flush(wlist) return # 找逗号断点 parts: list[list[Word]] = [] cur: list[Word] = [] for w in wlist: cur.append(w) if _CLAUSE_BREAK.search(w.text) and len(cur) >= max_words // 2: parts.append(cur) cur = [] if cur: parts.append(cur) # 若逗号拆不开（无逗号），强制按 max_words 等分 if len(parts) == 1 and len(parts[0]) > max_words: parts = [parts[0][i:i + max_words] for i in range(0, len(parts[0]), max_words)] for p in parts: flush(p) # ---------------- 匀速估算路：段内按字符比例 ---------------- def _resegment_by_estimate( segments: list[Segment], max_words: int, max_dur: float, ) -> list[Subtitle]: """无词级时间戳时：先按文本断句，再按字符数比例估算时间戳。""" subs: list[Subtitle] = [] for seg in segments: text = seg.text.strip() if not text: continue dur = seg.end - seg.start # 按句末标点切 sentences = _split_sentences(text) if not sentences: sentences = [text] # 段内按字符数比例分配时间 total_chars = sum(len(s) for s in sentences) or 1 cursor = seg.start for sent in sentences: frac = len(sent) / total_chars est_end = cursor + dur * frac # 超长句再按逗号拆（时间按字符比例再分） if len(sent.split()) > max_words or (est_end - cursor) > max_dur: for clause in _split_clauses(sent): cfrac = len(clause) / len(sent) if len(sent) else 1 c_end = cursor + (est_end - cursor) * cfrac subs.append(Subtitle(text=clause.strip(), start=cursor, end=c_end)) cursor = c_end else: subs.append(Subtitle(text=sent.strip(), start=cursor, end=est_end)) cursor = est_end return subs def _split_sentences(text: str) -> list[str]: """按句末标点切句，保留标点。""" parts = _SENTENCE_END.split(text) marks = _SENTENCE_END.findall(text) out = [] for i, p in enumerate(parts): p = p.strip() if not p: continue out.append(p + (marks[i] if i < len(marks) else "")) return out def _split_clauses(sentence: str) -> list[str]: """按逗号/冒号拆子句，保留标点。""" parts = _CLAUSE_BREAK.split(sentence) marks = _CLAUSE_BREAK.findall(sentence) out = [] for i, p in enumerate(parts): p = p.strip() if not p: continue out.append(p + (marks[i - 1] if 0 < i <= len(marks) else "")) return out # ---------------- 规范化 ---------------- def _normalize(subs: list[Subtitle], min_dur: float, max_chars: int) -> list[Subtitle]: """合并过短条目、拆分过宽行。""" # 1. 合并过短（< min_dur 且非末尾） merged: list[Subtitle] = [] for s in subs: if merged and (s.end - s.start) < min_dur: prev = merged[-1] prev.text = (prev.text + " " + s.text).strip() prev.end = s.end else: merged.append(Subtitle(text=s.text, start=s.start, end=s.end)) # 2. 拆分超过 max_chars 的行（按词折行，不改时间戳） out: list[Subtitle] = [] for s in merged: if len(s.text) <= max_chars: out.append(s) continue lines = _wrap_text(s.text, max_chars) out.append(Subtitle(text=lines, start=s.start, end=s.end)) return out def _wrap_text(text: str, max_chars: int) -> str: """按词折成 ≤2 行，每行 ≤max_chars 字符（SRT 规范）。""" words = text.split() lines: list[str] = [] cur = "" for idx, w in enumerate(words): if cur and len(cur) + 1 + len(w) > max_chars: lines.append(cur) # 已有一行 + 当前行，合并剩余为一行 cur = " ".join([w] + words[idx + 1:]) break else: cur = (cur + " " + w).strip() if cur else w if cur: lines.append(cur) return "\n".join(lines[:2])