自动同步: 2026-03-20 02:10

skills/kb_workflow/srt/__init__.py

@@ -0,0 +1,27 @@
+"""SRT Processing Module
+
+B站字幕处理模块，包含：
+- SRT 解析器 (parser)
+- 语义分段器 (segmenter)
+- ASR 纠错器 (corrector)
+"""
+
+from kb_workflow.srt.parser import parse_srt, parse_srt_file, Segment
+
+__all__ = [
+    "parse_srt",
+    "parse_srt_file",
+    "Segment",
+    "SemanticSegmenter",
+    "ASRCorrector",
+]
+
+# 延迟导入，避免循环依赖
+def __getattr__(name: str):
+    if name == "SemanticSegmenter":
+        from kb_workflow.srt.segmenter import SemanticSegmenter
+        return SemanticSegmenter
+    if name == "ASRCorrector":
+        from kb_workflow.srt.corrector import ASRCorrector
+        return ASRCorrector
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

skills/kb_workflow/srt/corrector.py

@@ -0,0 +1,157 @@
+"""ASR Corrector Module
+
+使用 LLM 纠正语音识别错误。复用 SimpleAIClient。
+
+功能：
+- 纠正明显的语音识别错误（同音字、漏字、错字）
+- 分批处理长文本
+- AI 不可用时返回原文
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Optional, Protocol
+
+logger = logging.getLogger(__name__)
+
+
+class AIClientProtocol(Protocol):
+    """AI 客户端协议"""
+
+    def generate(self, prompt: str, max_tokens: int = 16000) -> str:
+        """生成文本"""
+        ...
+
+
+# 纠错 Prompt 模板
+CORRECTION_PROMPT = """你是一个文本校对专家。以下是语音识别生成的字幕，可能包含识别错误。
+请只纠正明显的错误（如同音字、漏字、错字），不要改变原意和风格。
+
+原文：
+{text}
+
+纠正后的文本（只输出纠正后的文本，不要解释）："""
+
+
+class ASRCorrector:
+    """ASR 纠错器
+
+    使用 LLM 纠正语音识别中的明显错误。
+
+    Attributes:
+        ai_client: AI 客户端（必须实现 generate 方法）
+        batch_size: 分批处理的字符数，默认 5000
+    """
+
+    def __init__(
+        self,
+        ai_client: Optional[AIClientProtocol] = None,
+        batch_size: int = 5000
+    ):
+        """初始化纠错器
+
+        Args:
+            ai_client: AI 客户端实例，如果为 None 则不进行纠错
+            batch_size: 分批处理的字符数
+        """
+        self.ai_client = ai_client
+        self.batch_size = batch_size
+
+    def correct(self, text: str) -> str:
+        """纠正文本中的语音识别错误
+
+        Args:
+            text: 原始文本
+
+        Returns:
+            纠正后的文本。如果 AI 不可用或文本为空，返回原文。
+
+        Example:
+            >>> corrector = ASRCorrector(ai_client=my_client)
+            >>> result = corrector.correct("什么是机器学习")
+            >>> print(result)
+            什么是机器学习
+        """
+        # 空文本直接返回
+        if not text or not text.strip():
+            return text or ""
+
+        # AI 不可用，返回原文
+        if self.ai_client is None:
+            logger.info("AI client not available, returning original text")
+            return text
+
+        # 短文本直接处理
+        if len(text) <= self.batch_size:
+            return self._correct_batch(text)
+
+        # 长文本分批处理
+        return self._correct_long_text(text)
+
+    def _correct_batch(self, text: str) -> str:
+        """处理单个批次的文本
+
+        Args:
+            text: 文本内容
+
+        Returns:
+            纠正后的文本
+        """
+        try:
+            prompt = CORRECTION_PROMPT.format(text=text)
+            result = self.ai_client.generate(prompt, max_tokens=len(text) + 500)
+            return result.strip()
+        except Exception as e:
+            logger.warning(f"ASR correction failed: {e}, returning original text")
+            return text
+
+    def _correct_long_text(self, text: str) -> str:
+        """分批处理长文本
+
+        按句子分割，每批不超过 batch_size 字符。
+
+        Args:
+            text: 长文本内容
+
+        Returns:
+            纠正后的文本
+        """
+        # 按句子分割（简单分割，基于句号、问号、感叹号）
+        sentences = []
+        current = ""
+        for char in text:
+            current += char
+            if char in '。！？\n':
+                sentences.append(current)
+                current = ""
+
+        # 添加最后一个不完整的句子
+        if current:
+            sentences.append(current)
+
+        # 分批处理
+        batches = []
+        current_batch = []
+        current_length = 0
+
+        for sentence in sentences:
+            if current_length + len(sentence) > self.batch_size:
+                if current_batch:
+                    batches.append(''.join(current_batch))
+                current_batch = [sentence]
+                current_length = len(sentence)
+            else:
+                current_batch.append(sentence)
+                current_length += len(sentence)
+
+        if current_batch:
+            batches.append(''.join(current_batch))
+
+        # 处理每个批次
+        corrected_batches = []
+        for batch in batches:
+            corrected = self._correct_batch(batch)
+            corrected_batches.append(corrected)
+
+        return ''.join(corrected_batches)

skills/kb_workflow/srt/parser.py

@@ -0,0 +1,161 @@
+"""SRT Parser Module
+
+解析 SRT 字幕文件格式，将字幕转换为结构化的 Segment 对象。
+
+SRT 格式示例:
+    1
+    00:00:00,000 --> 00:00:05,000
+    第一句字幕内容
+
+    2
+    00:00:05,000 --> 00:00:10,000
+    第二句字幕内容
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from pathlib import Path
+from typing import List
+
+
+@dataclass
+class Segment:
+    """字幕分段数据结构
+
+    Attributes:
+        id: 字幕序号
+        start_time: 开始时间（秒）
+        end_time: 结束时间（秒）
+        text: 字幕文本
+    """
+    id: int
+    start_time: float
+    end_time: float
+    text: str
+
+
+# SRT 块正则表达式
+# 格式: 序号\n时间码\n文本(可能多行)\n\n
+SRT_BLOCK_PATTERN = re.compile(
+    r'(\d+)\s*\n'
+    r'(\d{2}:\d{2}:\d{2},\d{3})\s*-->\s*(\d{2}:\d{2}:\d{2},\d{3})\s*\n'
+    r'(.*?)(?=\n\n|\Z)',
+    re.DOTALL
+)
+
+
+def parse_srt_time(time_str: str) -> float:
+    """解析 SRT 时间格式 (HH:MM:SS,mmm) 为秒数
+
+    Args:
+        time_str: SRT 时间字符串，如 "00:05:30,500"
+
+    Returns:
+        秒数，如 330.5
+
+    Example:
+        >>> parse_srt_time("00:05:30,500")
+        330.5
+        >>> parse_srt_time("01:02:03,004")
+        3723.004
+    """
+    # 分割时间和毫秒
+    main_part, millis = time_str.rsplit(',', 1)
+    hours, minutes, seconds = main_part.split(':')
+
+    total_seconds = (
+        int(hours) * 3600 +
+        int(minutes) * 60 +
+        int(seconds) +
+        int(millis) / 1000
+    )
+    return total_seconds
+
+
+def parse_srt(content: str) -> List[Segment]:
+    """解析 SRT 格式字符串
+
+    将 SRT 格式的字幕内容解析为 Segment 对象列表。
+
+    Args:
+        content: SRT 格式的字符串内容
+
+    Returns:
+        Segment 对象列表
+
+    Raises:
+        ValueError: 内容为空或格式无效
+
+    Example:
+        >>> content = "1\\n00:00:00,000 --> 00:00:05,000\\n第一句\\n\\n"
+        >>> segments = parse_srt(content)
+        >>> len(segments)
+        1
+        >>> segments[0].text
+        '第一句'
+    """
+    # 检查空内容
+    if not content or not content.strip():
+        raise ValueError("Empty SRT content")
+
+    segments = []
+
+    # 查找所有 SRT 块
+    matches = SRT_BLOCK_PATTERN.findall(content)
+
+    if not matches:
+        raise ValueError("Invalid SRT format: no valid blocks found")
+
+    for match in matches:
+        try:
+            segment_id = int(match[0])
+            start_time = parse_srt_time(match[1])
+            end_time = parse_srt_time(match[2])
+            # 清理文本：移除前后空白，合并多行为单行
+            text = ' '.join(match[3].strip().split())
+
+            segment = Segment(
+                id=segment_id,
+                start_time=start_time,
+                end_time=end_time,
+                text=text
+            )
+            segments.append(segment)
+        except (ValueError, IndexError) as e:
+            # 跳过解析失败的块
+            continue
+
+    if not segments:
+        raise ValueError("Invalid SRT format: no valid segments parsed")
+
+    return segments
+
+
+def parse_srt_file(file_path: Path) -> List[Segment]:
+    """从文件解析 SRT 格式
+
+    读取文件内容并解析为 Segment 对象列表。
+
+    Args:
+        file_path: SRT 文件路径
+
+    Returns:
+        Segment 对象列表
+
+    Raises:
+        FileNotFoundError: 文件不存在
+        ValueError: 内容格式无效
+
+    Example:
+        >>> from pathlib import Path
+        >>> segments = parse_srt_file(Path("subtitles.srt"))
+        >>> len(segments)
+        10
+    """
+    if not file_path.exists():
+        raise FileNotFoundError(f"SRT file not found: {file_path}")
+
+    content = file_path.read_text(encoding='utf-8')
+    return parse_srt(content)

skills/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