rewrite segments analysis

config.py

@@ -3,7 +3,7 @@ import re
 import logging
 
 logging.basicConfig(
-    level=logging.INFO,
+    level=logging.DEBUG,
     format="%(asctime)s - %(levelname)s - %(message)s",
     datefmt="%H:%M:%S"
 )
@@ -14,6 +14,8 @@ logging.getLogger("pywhispercpp").setLevel(logging.WARNING)
 BASE_DIR = pathlib.Path(__file__).parent
 MODEL_DIR = BASE_DIR / "moyoyo_asr_models"
 ASSERT_DIR = BASE_DIR / "assets"
+
+SAMPLE_RATE = 16000
 # 标点
 SENTENCE_END_MARKERS =  ['.', '!', '?', '。', '！', '？', ';', '；', ':', '：']
 PAUSE_END_MARKERS = [',', '，', '、']

tests/test_transcript_buffer.py

@@ -0,0 +1,59 @@
+import unittest
+from transcribe.strategy import TranscriptBuffer  # 请替换为你的实际模块名
+
+class TestTranscriptBuffer(unittest.TestCase):
+
+    def setUp(self):
+        self.buffer = TranscriptBuffer()
+
+    def test_update_pending_text(self):
+        self.buffer.update_pending_text("Hello")
+        self.assertEqual(self.buffer.pending_text, "Hello")
+
+    def test_commit_line(self):
+        self.buffer.update_pending_text("This is a line.")
+        self.buffer.commit_line()
+        self.assertEqual(self.buffer.paragraph, "This is a line.")
+        self.assertEqual(self.buffer.pending_text, "")
+
+    def test_commit_paragraph(self):
+        self.buffer.update_pending_text("Sentence 1.")
+        self.buffer.commit_line()
+        self.buffer.update_pending_text("Sentence 2.")
+        self.buffer.commit_line()
+        self.buffer.commit_paragraph(end_of_sentence=True)
+
+        self.assertEqual(self.buffer.get_seg_id(), 1)
+        self.assertEqual(self.buffer.latest_paragraph, "Sentence 1.Sentence 2.")
+        self.assertEqual(self.buffer.paragraph, "")
+        self.assertEqual(self.buffer.pending_text, "")
+
+    def test_commit_paragraph_without_end(self):
+        self.buffer.update_pending_text("Incomplete sentence.")
+        self.buffer.commit_line()
+        self.buffer.commit_paragraph(end_of_sentence=False)
+
+        # 段落不应提交
+        self.assertEqual(self.buffer.get_seg_id(), 0)
+        self.assertEqual(self.buffer.paragraph, "Incomplete sentence.")
+
+    def test_update_and_commit_end_sentence(self):
+        self.buffer.update_and_commit("Stable.", "Remaining", is_end_sentence=True)
+
+        self.assertEqual(self.buffer.latest_paragraph, "Stable.")
+        self.assertEqual(self.buffer.pending_text, "Remaining")
+        self.assertEqual(self.buffer.paragraph, "")
+
+    def test_update_and_commit_partial_sentence(self):
+        self.buffer.update_and_commit("Partial", "New Buffer", is_end_sentence=False)
+
+        self.assertEqual(self.buffer.paragraph, "Partial")
+        self.assertEqual(self.buffer.pending_text, "New Buffer")
+        self.assertEqual(self.buffer.get_seg_id(), 0)
+
+    def test_current_not_commit_text(self):
+        self.buffer.update_and_commit("Part 1.", "Live text", is_end_sentence=False)
+        self.assertEqual(self.buffer.current_not_commit_text, "Part 1.Live text")
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_transcript_chunk.py

@@ -0,0 +1,57 @@
+import unittest
+from transcribe.strategy import TranscriptChunk, TranscriptToken, SplitMode
+
+class TestTranscriptChunk(unittest.TestCase):
+
+    def setUp(self):
+        self.tokens = [
+            TranscriptToken(text="Hello", t0=0, t1=100),
+            TranscriptToken(text=",", t0=100, t1=200),
+            TranscriptToken(text="world", t0=200, t1=300),
+            TranscriptToken(text=".", t0=300, t1=400),
+        ]
+        self.chunk = TranscriptChunk(items=self.tokens, separator=" ")
+
+    def test_split_by_punctuation(self):
+        chunks = self.chunk.split_by(SplitMode.PUNCTUATION)
+        self.assertEqual(len(chunks), 3)
+        self.assertEqual(chunks[0].join(), "Hello ,")
+        self.assertEqual(chunks[1].join(), "world .")
+        self.assertEqual(chunks[2].join(), "")
+
+    def test_get_split_first_rest(self):
+        first, rest = self.chunk.get_split_first_rest(SplitMode.PUNCTUATION)
+        self.assertEqual(first.join(), "Hello ,")
+        self.assertEqual(len(rest), 2)
+        self.assertEqual(rest[0].join(), "world .")
+        self.assertEqual(rest[1].join(), "")
+
+    def test_punctuation_numbers(self):
+        self.assertEqual(self.chunk.puncation_numbers(), 2)
+
+    def test_length(self):
+        self.assertEqual(self.chunk.length(), 4)
+
+    def test_join(self):
+        self.assertEqual(self.chunk.join(), "Hello , world .")
+
+    def test_compare(self):
+        other_chunk = TranscriptChunk(items=[
+            TranscriptToken(text="Hello", t0=0, t1=100),
+            TranscriptToken(text="!", t0=100, t1=200),
+        ], separator=" ")
+        similarity = self.chunk.compare(other_chunk)
+        self.assertTrue(0 < similarity < 1)
+
+    def test_has_punctuation(self):
+        self.assertTrue(self.chunk.has_punctuation())
+
+    def test_get_buffer_index(self):
+        # t1 = 400 -> index = 400 / 100 * 16000 = 64000
+        self.assertEqual(self.chunk.get_buffer_index(), 64000)
+
+    def test_is_end_sentence(self):
+        self.assertTrue(self.chunk.is_end_sentence())
+
+if __name__ == '__main__':
+    unittest.main()

transcribe/strategy.py

@@ -3,303 +3,291 @@ import collections
 import logging
 from difflib import SequenceMatcher
 from itertools import chain
-from dataclasses import dataclass
-from typing import List, Tuple, Optional, Deque, Any, Iterator
-from config import SENTENCE_END_MARKERS, ALL_MARKERS,SENTENCE_END_PATTERN,REGEX_MARKERS
+from dataclasses import dataclass, field
+from typing import List, Tuple, Optional, Deque, Any, Iterator,Literal
+from config import SENTENCE_END_MARKERS, ALL_MARKERS,SENTENCE_END_PATTERN,REGEX_MARKERS, PAUSEE_END_PATTERN,SAMPLE_RATE
 import numpy as np
-
+from enum import Enum
 logger = logging.getLogger("TranscriptionStrategy")
 
 
+class SplitMode(Enum):
+    PUNCTUATION = "punctuation"
+    PAUSE = "pause"
+    END = "end"
+
+
+
+@dataclass
+class TranscriptResult:
+    seg_id: int = 0
+    cut_index: int = 0
+    is_end_sentence: bool = False
+    context: str = ""
+
+    def partial(self):
+        return not self.is_end_sentence
+
 @dataclass
-class TranscriptSegment:
+class TranscriptToken:
     """表示一个转录片段，包含文本和时间信息"""
-    text: str
+    text: str  # 转录的文本内容
     t0: float  # 开始时间（百分之一秒）
     t1: float  # 结束时间（百分之一秒）
 
-def join_segment_text(segments: List[TranscriptSegment], separator: str = "") -> str:
-    """连接多个片段的文本"""
-    return separator.join(seg.text for seg in segments)
-
-class TextStabilityBuffer:
-    """
-    通过比较连续文本样本的相似度来确定转录文本的稳定性。
-    当连续样本的相似度超过阈值时，认为文本已稳定。
-    """
-    def __init__(self, max_history: int = 2):
-        self.history: Deque[Tuple[str, int]] = collections.deque(maxlen=max_history)
-
-    def add_entry(self, text: str, index: int) -> None:
-        """
-        添加新的文本和索引到历史记录中
-        
-        Args:
-            text: 文本内容
-            index: 当前buffer的相对下标
-        """
-        self.history.append((text, index))
-
-    def get_stable_index(self, similarity_threshold: float = 0.7) -> Optional[int]:
-        """
-        根据文本相似度，判断文本是否稳定，返回稳定文本的索引
-        
-        Args:
-            similarity_threshold: 相似度阈值，超过此值认为文本稳定
-            
-        Returns:
-            稳定文本的索引，如果没有找到稳定文本则返回None
-        """
-        if len(self.history) < 2:
-            return None
+    def is_punctuation(self):
+        """检查文本是否包含标点符号"""
+        return REGEX_MARKERS.search(self.text) is not  None
+    
+    def is_end(self):
+        """检查文本是否为句子结束标记"""
+        return SENTENCE_END_PATTERN.search(self.text)  is not  None
+    
+    def is_pause(self):
+        """检查文本是否为暂停标记"""
+        return PAUSEE_END_PATTERN.search(self.text) is not  None
 
-        text1, _ = self.history[0]
-        text2, idx2 = self.history[1]
+    def buffer_index(self) -> int:
+        return int(self.t1 / 100 * SAMPLE_RATE)
 
-        similarity = self._calculate_similarity(text1, text2)
-        
-        if similarity >= similarity_threshold:
-            self.history.clear()
-            return idx2
-        return None
+@dataclass
+class TranscriptChunk:
+    """表示一组转录片段，支持分割和比较操作"""
+    separator: str = ""  # 用于连接片段的分隔符
+    items: list[TranscriptToken] = field(default_factory=list)  # 转录片段列表
 
     @staticmethod
     def _calculate_similarity(text1: str, text2: str) -> float:
         """计算两段文本的相似度"""
         return SequenceMatcher(None, text1, text2).ratio()
 
+    def split_by(self, mode: SplitMode) -> list['TranscriptChunk']:
+        """根据文本中的标点符号分割片段列表"""
+        if mode == SplitMode.PUNCTUATION:
+            indexes = [i for i, seg in enumerate(self.items) if seg.is_punctuation()]
+        elif mode == SplitMode.PAUSE:
+            indexes = [i for i, seg in enumerate(self.items) if seg.is_pause()]
+        elif mode == SplitMode.END:
+            indexes = [i for i, seg in enumerate(self.items) if seg.is_end()]
+        else:
+            raise ValueError(f"Unsupported mode: {mode}")
 
+        # 每个切分点向后移一个索引，表示“分隔符归前段”
+        cut_points = [0] + sorted(i + 1 for i in indexes) + [len(self.items)]
+        return [
+            TranscriptChunk(items=self.items[start:end], separator=self.separator)
+            for start, end in zip(cut_points, cut_points[1:])
+        ]
 
-class TranscriptionManager:
-    """
-    管理转录文本的分级结构：临时字符串 -> 短句 -> 完整段落
-
-    |-- 已确认文本 --|-- 观察窗口 --|-- 新输入 --|
-    """
-    def __init__(self):
-        self._committed_segments: List[str] = []  # 确认的完整段落
-        self._committed_sentences: List[str] = []  # 确认的短句
-        self._temp_string: str = ""  # 临时字符串缓冲
-    
-    def check_line_break(self, min_length: int = 20) -> bool:
-        """检查当前短句长度是否达到换行标准"""
-        return self.sentence_length >= min_length
-
-    def force_line_break(self) -> None:
-        """强制换行，保留当前内容但创建新段落"""
-        if self.current_sentence:
-            self._committed_segments.append(self.current_sentence)
-            self._committed_sentences = []
-        
-    @property
-    def current_sentence(self) -> str:
-        """当前已确认的短句组合"""
-        return "".join(self._committed_sentences)
     
-    @property
-    def remaining_text(self) -> str:
-        return self._temp_string
+    def get_split_first_rest(self,  mode: SplitMode):
+        chunks = self.split_by(mode)
+        fisrt_chunk = chunks[0] if chunks else self
+        rest_chunks = chunks[1:] if chunks else []
+        return fisrt_chunk, rest_chunks
     
-    @property
-    def latest_segment(self) -> str:
-        """最新确认的完整段落"""
-        return self._committed_segments[-1] if self._committed_segments else ""
+    def puncation_numbers(self) -> int:
+        """计算片段中标点符号的数量"""
+        return sum(1 for seg in self.items if REGEX_MARKERS.search(seg.text))
+
+    def length(self) -> int:
+        """返回片段列表的长度"""
+        return len(self.items)
+
+    def join(self) -> str:
+        """将片段连接为一个字符串"""
+        return self.separator.join(seg.text for seg in self.items)
     
-    @property
-    def segment_count(self) -> int:
-        """已确认的段落数量"""
-        return len(self._committed_segments)
+    def compare(self, chunk: Optional['TranscriptChunk'] = None) -> float:
+        """比较当前片段与另一个片段的相似度"""
+        if not chunk:
+            return 0
+        return self._calculate_similarity(self.join(), chunk.join())
     
-    @property
-    def sentence_length(self) -> int:
-        """当前短句的总字符长度"""
-        return sum(len(s) for s in self._committed_sentences)
+    def has_punctuation(self) -> bool:
+        return any(seg.is_punctuation() for seg in self.items)
     
-    def update_temp(self, text: str) -> 'TranscriptionManager':
-        """更新临时字符串"""
-        self._temp_string = text
-        return self
+    def get_buffer_index(self) -> int:
+        return self.items[-1].buffer_index()
     
-    def commit_sentence(self) -> None:
-        """将临时字符串提交到短句列表"""
-        if self._temp_string:
-            self._committed_sentences.append(self._temp_string)
-            self._temp_string = ""
+    def is_end_sentence(self) ->bool:
+        return self.items[-1].is_end()
+ 
+
+class TranscriptHistory:
+    """管理转录片段的历史记录"""
+
+    def __init__(self) -> None:
+        self.history = collections.deque(maxlen=2)  # 存储最近的两个片段
     
-    def commit_segment(self, is_end_of_sentence: bool = False) -> None:
+    def add(self, chunk: TranscriptChunk):
+        """添加新的片段到历史记录"""
+        self.history.appendleft(chunk)
+
+    def previous_chunk(self) -> Optional[TranscriptChunk]:
+        """获取上一个片段（如果存在）"""
+        return self.history[1] if len(self.history) == 2 else None
+
+    def lastest_chunk(self):
+        """获取最后一个片段"""
+        return self.history[-1]
+
+
+class TranscriptBuffer:
+    """
+    管理转录文本的分级结构：临时字符串 -> 短句 -> 完整段落
+
+    |-- 已确认文本 --|-- 观察窗口 --|-- 新输入 --|
+
+    管理 pending -> line -> paragraph 的缓冲逻辑
+
+    """
+
+    def __init__(self):
+        self._segments: List[str] = []     # 确认的完整段落
+        self._sentences: List[str] = []    # 当前段落中的短句
+        self._buffer: str = ""             # 当前缓冲中的文本
+
+    def update_pending_text(self, text: str) -> None:
+        """更新临时缓冲字符串"""
+        self._buffer = text
+
+    def commit_line(self) -> None:
+        """将缓冲字符串提交为短句"""
+        if self._buffer:
+            self._sentences.append(self._buffer)
+            self._buffer = ""
+
+    def commit_paragraph(self, end_of_sentence: bool = False) -> None:
         """
-        提交当前内容到适当的层级
+        提交当前短句为完整段落（如句子结束）
         
         Args:
-            is_end_of_sentence: 是否为完整句子的结束
-        """
-        self.commit_sentence()
-        if is_end_of_sentence and self._committed_sentences:
-            self._committed_segments.append(self.current_sentence)
-            self._committed_sentences = []
-            
-    def get_all_text(self) -> str:
-        """获取所有已提交的文本"""
-        all_segments = self._committed_segments.copy()
-        if self.current_sentence:
-            all_segments.append(self.current_sentence)
-        if self._temp_string:
-            all_segments.append(self._temp_string)
-        return "\n".join(all_segments)
-
-
-class TranscriptionSplitter:
-    """负责根据语音和文本特征拆分转录片段"""
-    
-    @staticmethod
-    def group_by_sentences(segments: List[TranscriptSegment]) -> List[List[TranscriptSegment]]:
-        """将片段按照完整句子分组"""
-        sequences = []
-        temp_seq = []
-
-        for seg in segments:
-            temp_seq.append(seg)
-            if any(marker in seg.text for marker in SENTENCE_END_MARKERS):
-                sequences.append(temp_seq.copy())
-                temp_seq = []
-                
-        if temp_seq:
-            sequences.append(temp_seq)
-        return sequences
-    
-    @staticmethod
-    def split_by_punctuation(
-        segments: List[TranscriptSegment], 
-        sample_rate: int = 16000,
-        segment_skip_index=  0
-    ) -> Tuple[int, List[TranscriptSegment], List[TranscriptSegment], bool]:
-        """
-        根据标点符号将片段分为左侧（已确认）和右侧（待确认）
-        
-        Returns:
-            (分割索引, 左侧片段, 右侧片段, 是否为句子结束)
+            end_of_sentence: 是否为句子结尾（如检测到句号）
         """
-        left_segments = []
-        right_segments = []
-        split_index = 0
-        is_sentence_end = False
-        
-        # # 短音频使用所有标点符号作为分割依据
-        # buffer_duration = len(audio_buffer) / sample_rate
-        # markers = ALL_MARKERS if buffer_duration < 12 else SENTENCE_END_MARKERS
-        skip_segments = segments[:segment_skip_index+1]
-        skipped_segments = segments[segment_skip_index:]
-
-        markers = ALL_MARKERS
-        for idx, seg in enumerate(skipped_segments):
-            left_segments.append(seg)
-            if seg.text and seg.text[-1] in markers:
-                split_index = int(seg.t1 / 100 * sample_rate)
-                is_sentence_end = bool(SENTENCE_END_PATTERN.search(seg.text))
-                right_segments = skipped_segments[min(idx+1, len(skipped_segments)):]
-                break
-        left_segments = skip_segments+ left_segments
-        return split_index, left_segments, right_segments, is_sentence_end
-    
-    @staticmethod
-    def split_by_sequences(
-        segments: List[TranscriptSegment], 
-        sample_rate: int = 16000
-    ) -> Tuple[int, Iterator[TranscriptSegment], Iterator[TranscriptSegment], bool]:
-        """
-        对于长文本，按照句子组保留最新的两句
-        
-        Returns:
-            (分割索引, 左侧片段, 右侧片段, 是否为句子结束)
-        """
-        sequences = TranscriptionSplitter.group_by_sentences(segments)
-        
-        if len(sequences) > 2:
-            logger.info(f"Buffer clip via sequence, current length: {len(sequences)}")
-            left_segments = chain(*sequences[:-2])
-            right_segments = chain(*sequences[-2:])
-            
-            # 确定切分点
-            last_sequence = sequences[-3]
-            last_segment = last_sequence[-1]
-            split_index = int(last_segment.t1 / 100 * sample_rate)
-            
-            return split_index, left_segments, right_segments, True
-            
-        return 0, iter([]), iter(segments), False
-
-
-class TranscriptionStabilizer(TranscriptionSplitter):
-    """
-    转录结果稳定器，负责确认和管理转录片段
-    """
-    def __init__(self, sample_rate: int = 16000):
-        self.text_manager = TranscriptionManager()
-        self.sample_rate = sample_rate
+        self.commit_line()
+        if end_of_sentence and self._sentences:
+            self._segments.append("".join(self._sentences))
+            self._sentences.clear()
+
 
+    def update_and_commit(self, stable_string: str, remaining_string:str, is_end_sentence=False):
+        self.update_pending_text(stable_string)
+        if is_end_sentence:
+            self.commit_paragraph(end_of_sentence=True)
+        else:
+            self.commit_line()
+        self.update_pending_text(remaining_string)
+
+    def get_seg_id(self) -> int:
+        return len(self._segments)
+       
     @property
-    def latest_segment(self):
-        return self.text_manager.latest_segment
+    def paragraph(self) -> str:
+        """当前短句组合"""
+        return "".join(self._sentences)
 
+    @property
+    def pending_text(self) -> str:
+        """当前缓冲内容"""
+        return self._buffer
 
     @property
-    def segment_count(self):
-        return self.text_manager.segment_count
-    
+    def latest_paragraph(self) -> str:
+        """最新确认的段落"""
+        return self._segments[-1] if self._segments else ""
+
 
     @property
-    def remaining_text(self):
-        return self.text_manager.remaining_text
+    def current_not_commit_text(self) -> str:
+        return self.paragraph + self.pending_text
+
+
+
+class TranscriptStabilityAnalyzer:
+    def __init__(self) -> None:
+        self._transcript_buffer = TranscriptBuffer()
+        self._transcript_history = TranscriptHistory()
+
+    def merge_chunks(self, chunks: List[TranscriptChunk])->str:
+        return "".join(r.join() for r in chunks)
     
-    @property
-    def stable_string(self):
-        return self.text_manager.current_sentence
 
+    def analysis(self, separator, current: TranscriptChunk, buffer_duration: float) -> Iterator[TranscriptResult]:
+        current = TranscriptChunk(items=current, separator=separator)
+        self._transcript_history.add(current)
+
+        prev = self._transcript_history.previous_chunk()
+       
+        if not prev:
+            yield TranscriptResult(context=current.join())
+            return
         
-    def process_segments(self, segments: List[TranscriptSegment]) -> Tuple[Optional[int], bool]:
-        """
-        处理转录片段，确认稳定的文本
-        
-        Args:
-            segments: 转录片段列表
-            
-        Returns:
-            (音频分割点索引, 是否达到足够长度需要换行)
-        """
-        # 查找第一个包含标点的片段作为分割点
-        split_index = None
-        stable_segments = []
-        force_split = False
-        if len(segments) < 20:
-            remaining_text = join_segment_text(segments)
-            self.text_manager.update_temp(remaining_text)
-            return split_index, False, join_segment_text(segments), self.text_manager.remaining_text
-        
-        # 查找20个长度后的标点符号
-        split_index, left_segments, right_segments, is_sentence_end = self.split_by_punctuation(segments[20:],sample_rate=self.sample_rate)
-        
-        if split_index is not None:  # 找到标点，确认标点前的内容
-            stable_text = join_segment_text(left_segments)
-            self.text_manager.update_temp(stable_text).commit_sentence()
-            
-            # 更新剩余文本
-            remaining_text = join_segment_text(right_segments)
-            self.text_manager.update_temp(remaining_text)
+        self._transcript_buffer.update_pending_text(current.join())
+
+        if buffer_duration <= 12:
+            yield from self._handle_short_buffer(current, prev)
         else:
-            # 如果没有标点 但是累计超过22个字符 直接从20个字符的位置切掉
-            if len(segments) > 22 and not REGEX_MARKERS.search(join_segment_text(segments)):
-                split_index = int(segments[20].t1 / 100 * self.sample_rate)
-                stable_idx = 21  # 直接使用22个字符的索引
-                force_split = True
-                stable_text = join_segment_text(segments[:stable_idx])
-                self.text_manager.update_temp(stable_text).commit_sentence()
-                self.text_manager.update_temp(join_segment_text(segments[stable_idx:]))
-            else:
-                # 没有找到标点，全部作为临时文本
-                self.text_manager.update_temp(join_segment_text(segments))
-            
-        # 检查是否达到换行标准
-        should_linebreak = self.text_manager.sentence_length >= 20 or force_split
-        
-        return split_index, should_linebreak, join_segment_text(stable_segments), self.text_manager.remaining_text
- 
+            yield from self._handle_long_buffer(current)
+
+
+    def _handle_short_buffer(self, curr: TranscriptChunk, prev: TranscriptChunk) -> Iterator[TranscriptResult]:
+        curr_first, curr_rest = curr.get_split_first_rest(SplitMode.PUNCTUATION)
+        prev_first, _ = prev.get_split_first_rest(SplitMode.PUNCTUATION)
+        core = curr_first.compare(prev_first)
+        has_punctuation = curr_first.has_punctuation()
+        logger.debug(f"Compare with rev score:{core},is end :{curr_first.is_end_sentence()}, has_punctuation: {has_punctuation}, current_first: {curr_first.join()},")
+        if core >= 0.8:
+            yield from self._yield_commit_results(curr_first, curr_rest, curr_first.is_end_sentence())
+        else:
+            yield TranscriptResult(
+                seg_id=self._transcript_buffer.get_seg_id(),
+                context=self._transcript_buffer.current_not_commit_text
+            )
+
+
+    def _handle_long_buffer(self, curr: TranscriptChunk) -> Iterator[TranscriptResult]:
+        chunks = curr.split_by(SplitMode.PUNCTUATION)
+        if len(chunks) > 2:
+            stable, remaining = chunks[:-2], chunks[-2:]
+            stable_str = self.merge_chunks(stable)
+            remaining_str = self.merge_chunks(remaining)
+            yield from self._yield_commit_results(
+                stable[-1], remaining, is_end_sentence=True  # 暂时硬编码为True
+            )
+        else:
+            yield TranscriptResult(
+                seg_id=self._transcript_buffer.get_seg_id(),
+                context=self._transcript_buffer.current_not_commit_text
+            )
+
+
+    def _yield_commit_results(self, stable_chunk, remaining_chunks, is_end_sentence: bool) -> Iterator[TranscriptResult]:
+        stable_str = stable_chunk.join() if hasattr(stable_chunk, "join") else self.merge_chunks([stable_chunk])
+        remaining_str = self.merge_chunks(remaining_chunks)
+
+        frame_cut_index = stable_chunk.get_buffer_index()
+        logger.debug(f"Current cut index: {frame_cut_index}, Stable string: {stable_str}, Remaining_str:{remaining_str}")
+
+        prev_seg_id = self._transcript_buffer.get_seg_id()
+        self._transcript_buffer.update_and_commit(stable_str, remaining_str, is_end_sentence)
+        curr_seg_id = self._transcript_buffer.get_seg_id()
+
+        logger.debug(f"current buffer: {self._transcript_buffer.__dict__}")
+
+        if curr_seg_id > prev_seg_id:
+            # 表示生成了一个新段落
+            yield TranscriptResult(
+                seg_id=prev_seg_id,
+                cut_index=frame_cut_index,
+                context=self._transcript_buffer.latest_paragraph,
+                is_end_sentence=True
+            )
+
+        # 如果还有挂起的文本
+        if (pending_text := self._transcript_buffer.pending_text.strip()):
+            yield TranscriptResult(
+                seg_id=self._transcript_buffer.get_seg_id(),
+                cut_index=frame_cut_index,
+                context=pending_text
+            )

transcribe/whisper_llm_serve.py

@@ -13,12 +13,7 @@ from .server import ServeClientBase
 from .utils import log_block, save_to_wave
 from .translatepipes import TranslatePipes
 from .strategy import (
-    TextStabilityBuffer, 
-    TranscriptionManager, 
-    TranscriptionSplitter, 
-    TranscriptSegment,
-    TranscriptionStabilizer,
-    join_segment_text)
+    TranscriptStabilityAnalyzer, TranscriptToken)
 
 logger = getLogger("TranscriptionService")
 
@@ -34,8 +29,7 @@ class WhisperTranscriptionService(ServeClientBase):
         self.target_language = dst_lang  # 目标翻译语言
         
         # 转录结果稳定性管理
-        self._text_stability_buffer = TextStabilityBuffer()
-        self._transcription_manager = TranscriptionManager()
+        self._transcrible_analysis = TranscriptStabilityAnalyzer()
         self._translate_pipe = pipe
         
         # 音频处理相关
@@ -56,7 +50,6 @@ class WhisperTranscriptionService(ServeClientBase):
         self.translate_thread = self._start_thread(self._transcription_processing_loop)
         self.frame_processing_thread = self._start_thread(self._frame_processing_loop)
 
-        self.text_stablizer = TranscriptionStabilizer()
 
     def _start_thread(self, target_function) -> threading.Thread:
         """启动守护线程执行指定函数"""
@@ -138,7 +131,7 @@ class WhisperTranscriptionService(ServeClientBase):
         
         return frames.copy()
 
-    def _transcribe_audio(self, audio_buffer: np.ndarray) -> List[TranscriptSegment]:
+    def _transcribe_audio(self, audio_buffer: np.ndarray) -> List[TranscriptToken]:
         """转录音频并返回转录片段"""
         log_block("Audio buffer length", f"{audio_buffer.shape[0]/self.sample_rate:.2f}", "s")
         start_time = time.perf_counter()
@@ -149,7 +142,10 @@ class WhisperTranscriptionService(ServeClientBase):
         log_block("Whisper transcription output", f"{''.join(seg.text for seg in segments)}", "")
         log_block("Whisper transcription time", f"{(time.perf_counter() - start_time):.3f}", "s")
         
-        return segments
+        return [
+            TranscriptToken(text=s.text, t0=s.t0, t1=s.t1)
+            for s in segments
+        ]
 
     def _translate_text(self, text: str) -> str:
         """将文本翻译为目标语言"""
@@ -167,66 +163,7 @@ class WhisperTranscriptionService(ServeClientBase):
         
         return translated_text
 
-    def _find_best_split_position(self, segments: list, target_length: int = 20) -> int:
-        """找到最适合分割的位置，尽量靠近目标长度且在词/字的边界"""
-        if len(segments) <= target_length:
-            return 0
-            
-        # 从目标长度位置向前搜索适合的分割点
-        for i in range(target_length, min(target_length + 10, len(segments))):
-            # 对于中文，每个字符都可以作为分割点
-            # 对于英文，在空格处分割
-            if self.source_language == "zh" or segments[i] == " ":
-                return i
-                
-        # 如果找不到理想分割点，就在目标长度处分割
-        return target_length
-
-    def _analyze_segments(self, segments: List[TranscriptSegment], audio_buffer: np.ndarray) -> Tuple[Optional[int], str, str, bool]:
-        """
-        分析转录片段，确定稳定部分和需要继续观察的部分
-        
-        Returns:
-            (分割索引, 左侧稳定文本, 右侧观察文本, 是否为句子结束)
-        """
-        # 尝试基于标点符号进行分割
-        left_idx, left_segments, right_segments, is_end = TranscriptionSplitter.split_by_punctuation(
-            segments, audio_buffer, self.sample_rate
-        )
-        
-        left_text = join_segment_text(left_segments, self.text_separator)
-        right_text =  join_segment_text(right_segments, self.text_separator)
-        
-        # 如果找到分割点，检查左侧文本稳定性
-        if left_idx != 0:
-            self._text_stability_buffer.add_entry(left_text, left_idx)
-            stable_idx = self._text_stability_buffer.get_stable_index()
-            if stable_idx:
-                should_break = True if (self._transcription_manager.sentence_length>= 20) else False
-                return stable_idx, left_text, right_text, should_break
-
-        # 如果基于标点的方法没有找到稳定点，尝试检查句子的长度
-        if len(segments) >= 20:  # 设置更长的阈值，确保有足够内容进行分割
-            # 尝试在约20字符处找一个词的边界进行分割
-            split_pos = self._find_best_split_position(segments)
-            if split_pos > 0:
-                left_text = join_segment_text(segments[:split_pos], self.text_separator)
-                right_text = join_segment_text(segments[split_pos:], self.text_separator)
-                audio_pos = int(segments[split_pos].t1 / 100 * self.sample_rate)
-                return audio_pos, left_text, right_text, True
 
-        # 如果基于标点的方法未找到稳定点，尝试基于句子序列的方法
-        left_idx, left_segments, right_segments, is_end = TranscriptionSplitter.split_by_sequences(
-            segments, self.sample_rate
-        )
-        
-        if left_idx != 0:
-            left_text = join_segment_text(left_segments, self.text_separator)
-            right_text =  join_segment_text(right_segments, self.text_separator)
-            return left_idx, left_text, right_text, is_end
-        
-        # 如果都没有找到分割点
-        return None, left_text, right_text, is_end
 
     def _transcription_processing_loop(self) -> None:
         """主转录处理循环"""
@@ -248,79 +185,49 @@ class WhisperTranscriptionService(ServeClientBase):
                 time.sleep(0.2)
                 continue
 
-            c+= 1
-            save_to_wave(f"dev-{c}.wav", audio_buffer)
+            # c+= 1
+            # save_to_wave(f"dev-{c}.wav", audio_buffer)
             
             # try:
             segments = self._transcribe_audio(audio_buffer)
             
             # 处理转录结果并发送到客户端
             for result in self._process_transcription_results(segments, audio_buffer):
+                print(result)
                 self._send_result_to_client(result)
                     
             # except Exception as e:
             #     logger.error(f"Error processing audio: {e}")
 
-    def _process_transcription_results(self, segments: List[TranscriptSegment], audio_buffer: np.ndarray) -> Iterator[TransResult]:
+    def _process_transcription_results(self, segments: List[TranscriptToken], audio_buffer: np.ndarray) -> Iterator[TransResult]:
         """
         处理转录结果，生成翻译结果
         
         Returns:
             TransResult对象的迭代器
         """
-        # 合并所有片段的文本
-        full_text = self.text_separator.join(seg.text for seg in segments)
-        if not full_text:
+
+        if not segments:
             return
-            
-        cut_index, is_sentence_end, stable_text, remaining_text = self.text_stablizer.process_segments(segments)
-        # 如果找到稳定的分割点
-        if cut_index:
-            # 更新音频缓冲区，移除已处理部分
-            self._update_audio_buffer(cut_index)
-            
-            # 提交稳定的文本
-            log_block("Stable transcription", f"{stable_text}")
+        
+        for ana_result in self._transcrible_analysis.analysis(
+            self.text_separator,segments, len(audio_buffer)/self.sample_rate):
+            if (cut_index :=ana_result.cut_index)>0:
+                # 更新音频缓冲区，移除已处理部分
+                self._update_audio_buffer(cut_index)
 
-            # 如果是句子结束，发送完整句子的翻译结果
-            if is_sentence_end:
-                segment_text = self.text_stablizer.latest_segment
-                segment_id = self.text_stablizer.segment_count - 1
-                
-                # 生成已确认句子的翻译结果
-                yield TransResult(
-                    seg_id=segment_id,
-                    context=segment_text,
-                    from_=self.source_language,
-                    to=self.target_language,
-                    tran_content=self._translate_text(segment_text),
-                    partial=False
-                )
-                
-                # 如果还有剩余部分，生成临时翻译结果
-                if self.text_stablizer.remaining_text.strip():
-                    yield TransResult(
-                        seg_id=segment_id + 1,
-                        context=self.text_stablizer.remaining_text,
-                        from_=self.source_language,
-                        to=self.target_language,
-                        tran_content=self._translate_text(self.text_stablizer.remaining_text.strip()),
-                        partial=True
-                    )
-        else:
-            # 没有找到稳定点，发送当前所有内容的临时翻译结果
-            segment_id = self.text_stablizer.segment_count
-            current_text = self.text_stablizer.stable_string +  self.text_stablizer.remaining_text
-            
+            translated_context = self._translate_text(ana_result.context)
+            log_block("Translated context:", translated_context)
             yield TransResult(
-                seg_id=segment_id,
-                context=current_text,
+                seg_id=ana_result.seg_id,
+                context=ana_result.context,
                 from_=self.source_language,
                 to=self.target_language,
-                tran_content=self._translate_text(current_text),
-                partial=True
+                tran_content=translated_context,
+                partial=ana_result.partial()
             )
 
+
     def _send_result_to_client(self, result: TransResult) -> None:
         """发送翻译结果到客户端"""
         try: