repo_description: AutoClip: AI-powered video clipping and highlight generation · 一款智能高光提取与剪辑的二创工具
repo_topics: ai-video-editing auto-highlight llm video video-editing
code: """
  文本处理工具 - 文本分块、SRT解析等功能
  """
  import re
  import json
  from typing import List, Dict, Tuple
  from pathlib import Path
  import pysrt
  import logging
  
  from ..config import CHUNK_SIZE
  
  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}" 