From c259e0ac2fda6dc623c0a46efe13f1e5134aa308 Mon Sep 17 00:00:00 2001
From: starmen1265 <starmen1265@163.com>
Date: Thu, 19 Mar 2026 23:32:57 +0800
Subject: [PATCH] feat(12-01): implement SRT parser module

- Add Segment dataclass for subtitle segments
- Add parse_srt() function to parse SRT content string
- Add parse_srt_file() function to parse from file path
- Support standard SRT format with timestamps
- Handle special characters and multi-line subtitles

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
---
 kb_workflow/srt/__init__.py |  25 ++++++
 kb_workflow/srt/parser.py   | 161 ++++++++++++++++++++++++++++++++++++
 2 files changed, 186 insertions(+)
 create mode 100644 kb_workflow/srt/__init__.py
 create mode 100644 kb_workflow/srt/parser.py

diff --git a/kb_workflow/srt/__init__.py b/kb_workflow/srt/__init__.py
new file mode 100644
index 0000000..91e0d10
--- /dev/null
+++ b/kb_workflow/srt/__init__.py
@@ -0,0 +1,25 @@
+"""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",
+]
+
+# 延迟导入，避免循环依赖
+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}")
diff --git a/kb_workflow/srt/parser.py b/kb_workflow/srt/parser.py
new file mode 100644
index 0000000..6565792
--- /dev/null
+++ b/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)