wehub-resource-sync
295 lines
10 KiB
Python
295 lines
10 KiB
Python
"""
|
|
文本处理工具
|
|
"""
|
|
import json
|
|
import logging
|
|
import re
|
|
from typing import List, Dict, Any, Optional
|
|
from pathlib import Path
|
|
|
|
# 修复导入问题
|
|
try:
|
|
from ..core.shared_config import CHUNK_SIZE
|
|
except ImportError:
|
|
# 如果相对导入失败，尝试绝对导入
|
|
import sys
|
|
from pathlib import Path
|
|
backend_path = Path(__file__).parent.parent
|
|
if str(backend_path) not in sys.path:
|
|
sys.path.insert(0, str(backend_path))
|
|
from core.shared_config import CHUNK_SIZE
|
|
|
|
import pysrt
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class TextProcessor:
|
|
"""文本处理工具类"""
|
|
|
|
@staticmethod
|
|
def chunk_text(text: str, chunk_size: int = CHUNK_SIZE) -> List[str]:
|
|
"""
|
|
将长文本按指定大小分块
|
|
|
|
Args:
|
|
text: 输入文本
|
|
chunk_size: 分块大小
|
|
|
|
Returns:
|
|
文本块列表
|
|
"""
|
|
if len(text) <= chunk_size:
|
|
return [text]
|
|
|
|
chunks = []
|
|
current_chunk = ""
|
|
|
|
# 按段落分割
|
|
paragraphs = text.split('\n')
|
|
|
|
for paragraph in paragraphs:
|
|
# 如果当前块加上新段落不超过限制，则添加
|
|
if len(current_chunk) + len(paragraph) + 1 <= chunk_size:
|
|
current_chunk += paragraph + '\n'
|
|
else:
|
|
# 如果当前块不为空，保存它
|
|
if current_chunk.strip():
|
|
chunks.append(current_chunk.strip())
|
|
|
|
# 如果单个段落就超过限制，需要进一步分割
|
|
if len(paragraph) > chunk_size:
|
|
# 按句子分割
|
|
sentences = re.split(r'[。！？]', paragraph)
|
|
temp_chunk = ""
|
|
for sentence in sentences:
|
|
if len(temp_chunk) + len(sentence) + 1 <= chunk_size:
|
|
temp_chunk += sentence + "。"
|
|
else:
|
|
if temp_chunk:
|
|
chunks.append(temp_chunk.strip())
|
|
temp_chunk = sentence + "。"
|
|
current_chunk = temp_chunk
|
|
else:
|
|
current_chunk = paragraph + '\n'
|
|
|
|
# 添加最后一个块
|
|
if current_chunk.strip():
|
|
chunks.append(current_chunk.strip())
|
|
|
|
return chunks
|
|
|
|
def chunk_srt_data(self, srt_data: List[Dict], interval_minutes: int = 30, pause_threshold_ms: int = 1000) -> List[Dict]:
|
|
"""
|
|
根据停顿时间，将SRT数据切分为大约相等时间长度的块。
|
|
这可以避免在对话中间断开。
|
|
|
|
Args:
|
|
srt_data: SRT数据列表
|
|
interval_minutes: 每个块的目标时间长度（分钟）
|
|
pause_threshold_ms: 识别为停顿的最小毫秒数
|
|
|
|
Returns:
|
|
结构化的块列表，其中的 srt_entries 不包含临时处理字段。
|
|
"""
|
|
if not srt_data:
|
|
return []
|
|
|
|
# 创建一个带有秒数的新列表，而不是修改原始数据
|
|
srt_data_with_seconds = []
|
|
for sub in srt_data:
|
|
entry = sub.copy()
|
|
entry['start_seconds'] = self.time_to_seconds(sub['start_time'])
|
|
entry['end_seconds'] = self.time_to_seconds(sub['end_time'])
|
|
srt_data_with_seconds.append(entry)
|
|
|
|
interval_seconds = interval_minutes * 60
|
|
chunks = []
|
|
current_chunk_start_index = 0
|
|
chunk_index = 0
|
|
|
|
last_cut_time = 0
|
|
|
|
while current_chunk_start_index < len(srt_data_with_seconds):
|
|
target_cut_time = last_cut_time + interval_seconds
|
|
|
|
# 寻找接近目标时间的最佳切分点
|
|
best_cut_index = -1
|
|
|
|
# 查找从当前块开始后的 90% 到 110% 目标时间内的一个停顿
|
|
search_start_index = current_chunk_start_index
|
|
while search_start_index < len(srt_data_with_seconds) and srt_data_with_seconds[search_start_index]['start_seconds'] < target_cut_time * 0.9:
|
|
search_start_index += 1
|
|
|
|
# 从搜索起点开始寻找超过阈值的停顿
|
|
for i in range(search_start_index, len(srt_data_with_seconds) - 1):
|
|
current_sub = srt_data_with_seconds[i]
|
|
next_sub = srt_data_with_seconds[i+1]
|
|
|
|
# 如果我们已经超出了目标时间的110%，就停止搜索
|
|
if current_sub['start_seconds'] > target_cut_time * 1.1:
|
|
break
|
|
|
|
# 计算两个字幕条目之间的停顿时间
|
|
pause = next_sub['start_seconds'] - current_sub['end_seconds']
|
|
if pause * 1000 >= pause_threshold_ms:
|
|
best_cut_index = i + 1 # 在停顿后切分
|
|
break
|
|
|
|
# 如果没有找到合适的停顿点，就在目标时间点强制切分
|
|
if best_cut_index == -1:
|
|
# 寻找最接近目标时间的字幕条目
|
|
i = current_chunk_start_index
|
|
while i < len(srt_data_with_seconds) and srt_data_with_seconds[i]['start_seconds'] < target_cut_time:
|
|
i += 1
|
|
best_cut_index = i if i < len(srt_data_with_seconds) else len(srt_data_with_seconds)
|
|
|
|
# 如果切分点无效或过小，则将所有剩余部分作为一个块
|
|
if best_cut_index <= current_chunk_start_index:
|
|
best_cut_index = len(srt_data_with_seconds)
|
|
|
|
# 创建块
|
|
chunk_entries_with_seconds = srt_data_with_seconds[current_chunk_start_index:best_cut_index]
|
|
if not chunk_entries_with_seconds:
|
|
break
|
|
|
|
# 移除临时字段，得到干净的srt_entries
|
|
chunk_entries = []
|
|
for entry in chunk_entries_with_seconds:
|
|
clean_entry = entry.copy()
|
|
del clean_entry['start_seconds']
|
|
del clean_entry['end_seconds']
|
|
chunk_entries.append(clean_entry)
|
|
|
|
start_time = chunk_entries[0]['start_time']
|
|
end_time = chunk_entries[-1]['end_time']
|
|
text = " ".join([entry['text'] for entry in chunk_entries])
|
|
|
|
chunks.append({
|
|
"chunk_index": chunk_index,
|
|
"text": text,
|
|
"start_time": start_time,
|
|
"end_time": end_time,
|
|
"srt_entries": chunk_entries
|
|
})
|
|
|
|
chunk_index += 1
|
|
last_cut_time = chunk_entries_with_seconds[-1]['end_seconds']
|
|
current_chunk_start_index = best_cut_index
|
|
|
|
return chunks
|
|
|
|
@staticmethod
|
|
def parse_srt(srt_path: Path) -> List[Dict]:
|
|
"""
|
|
解析SRT字幕文件
|
|
|
|
Args:
|
|
srt_path: SRT文件路径
|
|
|
|
Returns:
|
|
字幕数据列表，每个元素包含时间戳和文本
|
|
"""
|
|
if not srt_path.exists():
|
|
logger.error(f"SRT文件不存在: {srt_path}")
|
|
return []
|
|
|
|
if srt_path.stat().st_size == 0:
|
|
logger.warning(f"SRT文件为空: {srt_path}")
|
|
return []
|
|
|
|
try:
|
|
try:
|
|
subs = pysrt.open(str(srt_path), encoding='utf-8')
|
|
except UnicodeDecodeError:
|
|
logger.warning("UTF-8解码失败，尝试使用 utf-8-sig...")
|
|
subs = pysrt.open(str(srt_path), encoding='utf-8-sig')
|
|
|
|
subtitles = []
|
|
for sub in subs:
|
|
subtitles.append({
|
|
'start_time': str(sub.start),
|
|
'end_time': str(sub.end),
|
|
'text': sub.text.strip(),
|
|
'index': sub.index
|
|
})
|
|
|
|
if not subtitles:
|
|
logger.warning(f"成功打开SRT文件但未能解析出任何字幕内容: {srt_path}")
|
|
|
|
return subtitles
|
|
except Exception as e:
|
|
logger.error(f"使用pysrt解析SRT文件'{srt_path}'时发生未知错误: {e}", exc_info=True)
|
|
return []
|
|
|
|
@staticmethod
|
|
def extract_text_by_time_range(text: str, srt_data: List[Dict],
|
|
start_time: str, end_time: str) -> str:
|
|
"""
|
|
根据时间范围从文本中提取对应内容
|
|
|
|
Args:
|
|
text: 完整文本
|
|
srt_data: SRT字幕数据
|
|
start_time: 开始时间 (格式: "00:01:25")
|
|
end_time: 结束时间 (格式: "00:02:53")
|
|
|
|
Returns:
|
|
对应时间范围的文本内容
|
|
"""
|
|
# 找到时间范围内的字幕
|
|
target_subtitles = []
|
|
|
|
for sub in srt_data:
|
|
sub_start = sub['start_time']
|
|
sub_end = sub['end_time']
|
|
|
|
# 检查时间重叠
|
|
if (sub_start <= end_time and sub_end >= start_time):
|
|
target_subtitles.append(sub)
|
|
|
|
# 提取对应的文本
|
|
extracted_text = ""
|
|
for sub in target_subtitles:
|
|
extracted_text += sub['text'] + " "
|
|
|
|
return extracted_text.strip()
|
|
|
|
@staticmethod
|
|
def time_to_seconds(time_str: str) -> float:
|
|
"""
|
|
将SRT时间字符串（HH:MM:SS,mmm）转换为秒数
|
|
|
|
Args:
|
|
time_str: 时间字符串
|
|
|
|
Returns:
|
|
秒数
|
|
"""
|
|
time_str = time_str.replace(',', '.')
|
|
parts = time_str.split(':')
|
|
|
|
if len(parts) == 3:
|
|
h = int(parts[0])
|
|
m = int(parts[1])
|
|
s_parts = parts[2].split('.')
|
|
s = int(s_parts[0])
|
|
ms = int(s_parts[1]) if len(s_parts) > 1 else 0
|
|
return h * 3600 + m * 60 + s + ms / 1000.0
|
|
|
|
raise ValueError(f"无效的时间格式: {time_str}")
|
|
|
|
@staticmethod
|
|
def seconds_to_time(seconds: float) -> str:
|
|
"""
|
|
将秒数转换为时间字符串
|
|
|
|
Args:
|
|
seconds: 秒数
|
|
|
|
Returns:
|
|
时间字符串 (格式: "00:01:25")
|
|
"""
|
|
hours = int(seconds // 3600)
|
|
minutes = int((seconds % 3600) // 60)
|
|
secs = int(seconds % 60)
|
|
return f"{hours:02d}:{minutes:02d}:{secs:02d}" |