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>
This commit is contained in:
@@ -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}")
|
||||||
@@ -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)
|
||||||
Reference in New Issue
Block a user