diff --git a/kb_workflow/srt/segmenter.py b/kb_workflow/srt/segmenter.py new file mode 100644 index 0000000..7af8d12 --- /dev/null +++ b/kb_workflow/srt/segmenter.py @@ -0,0 +1,97 @@ +"""Semantic Segmenter Module + +将字幕分段按语义合并为段落。基于简单启发式规则: +1. 时间间隔超过阈值时分段 +2. 段落长度限制 + +不使用 LLM 进行分段判断,保持简单和快速。 +""" + +from __future__ import annotations + +from typing import List + +from kb_workflow.srt.parser import Segment + + +class SemanticSegmenter: + """语义分段器 + + 将字幕分段转换为语义段落。基于时间和长度规则: + - 长时间停顿(>gap_threshold 秒)作为分段点 + - 段落长度限制(max_paragraph_chars 字符) + + Attributes: + gap_threshold: 时间间隔阈值(秒),默认 3.0 + max_paragraph_chars: 单段落最大字符数,默认 500 + """ + + def __init__( + self, + gap_threshold: float = 3.0, + max_paragraph_chars: int = 500 + ): + """初始化分段器 + + Args: + gap_threshold: 时间间隔阈值(秒),超过此值时分段 + max_paragraph_chars: 单段落最大字符数 + """ + self.gap_threshold = gap_threshold + self.max_paragraph_chars = max_paragraph_chars + + def segment(self, segments: List[Segment]) -> List[str]: + """将字幕分段转换为语义段落 + + Args: + segments: 字幕分段列表 + + Returns: + 段落文本列表(已去除时间戳) + + Example: + >>> segmenter = SemanticSegmenter() + >>> segments = [Segment(1, 0, 5, "第一句"), Segment(2, 5, 10, "第二句")] + >>> paragraphs = segmenter.segment(segments) + >>> len(paragraphs) + 1 + """ + if not segments: + return [] + + paragraphs = [] + current_paragraph = [] + current_length = 0 + last_end_time = None + + for seg in segments: + # 检查是否需要分段 + should_start_new = False + + # 规则1: 时间间隔超过阈值 + if last_end_time is not None: + gap = seg.start_time - last_end_time + if gap > self.gap_threshold: + should_start_new = True + + # 规则2: 段落长度超过限制 + if current_length + len(seg.text) > self.max_paragraph_chars: + if current_paragraph: # 只有当前段落有内容时才分段 + should_start_new = True + + # 执行分段 + if should_start_new and current_paragraph: + paragraphs.append(' '.join(current_paragraph)) + current_paragraph = [] + current_length = 0 + + # 添加当前分段 + current_paragraph.append(seg.text) + current_length += len(seg.text) + last_end_time = seg.end_time + + # 添加最后一个段落 + if current_paragraph: + paragraphs.append(' '.join(current_paragraph)) + + return paragraphs