"""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