feat(kb-deep-analyzer): Smart subtitle chunking strategy

## 改进内容

### 动态分段
- 根据字符数动态计算分段数
- 每个chunk最多3000字符
- 限制最小3段,最多20段

### 自然断点分割
- 优先级:段落(双换行) > 句末(。\n) > 句号 > 换行
- 不截断,保持语义完整

### 时间估算
- 用字符比例估算每段时间
- 长视频时间轴更准确
This commit is contained in:
OpenClaw Agent
2026-02-10 16:41:15 +08:00
parent 3cfd36cf4e
commit 08066334c4
+105 -31
View File
@@ -269,27 +269,62 @@ class DeepAnalyzer:
return result
def _analyze_in_chunks(self, title: str, subtitle: str, duration: int) -> AnalysisResult:
"""分段分析长视频"""
"""分段分析长视频 - 动态分段策略"""
result = AnalysisResult()
# 确定分段数
if duration < 3600: # <60分钟
num_segments = 3
elif duration < 7200: # 60-120分钟
num_segments = 4
else: # >120分钟
num_segments = 6
# 常量配置
CHUNK_MAX_CHARS = 3000 # 每个chunk最大字符数
MIN_CHUNKS = 3 # 最少分段数
MAX_API_CHUNKS = 20 # 最多API调用次数
# 生成分段摘要
segment_duration = duration // num_segments
total_chars = len(subtitle)
# 1. 动态计算分段数
# 计算需要多少个chunk(每个不超过3000字)
min_required = total_chars // CHUNK_MAX_CHARS + 1
# 根据时长限制最大分段
if duration < 1800: # <30分钟: 最多6段
max_chunks = 6
elif duration < 7200: # <2小时: 最多12段
max_chunks = 12
else: # >2小时: 最多20段
max_chunks = 20
num_segments = min(max(min_required, MIN_CHUNKS), MAX_API_CHUNKS)
print(f"[DeepAnalyzer] 字幕{total_chars}字符,分{num_segments}段分析")
# 2. 按字符数分段(保持语义完整)
chunk_size = total_chars // num_segments
chunks = []
chunk_boundaries = [] # 记录每段的字符边界
for i in range(num_segments):
start = i * chunk_size
if i == num_segments - 1:
# 最后一段取到结尾
end = total_chars
chunk_text = subtitle[start:]
else:
# 在自然断点处分割
end = self._find_breakpoint(subtitle, start, start + chunk_size)
chunk_text = subtitle[start:end]
chunks.append(chunk_text)
chunk_boundaries.append((start, end))
print(f"[DeepAnalyzer] 各段字符: {[len(c) for c in chunks]}")
# 3. 用字符比例估算时间
char_per_second = total_chars / duration if duration > 0 else 1
# 合并结果
chunk_summaries = []
# 简单按字符数分割字幕
subtitle_chunks = self._chunk_subtitle(subtitle, num_segments)
for i, chunk_text in enumerate(subtitle_chunks):
start_sec = i * segment_duration
end_sec = min((i + 1) * segment_duration, duration)
for i, chunk_text in enumerate(chunks):
start_char, end_char = chunk_boundaries[i]
# 用字符位置估算时间
start_sec = max(0, int(start_char / char_per_second))
end_sec = min(duration, int(end_char / char_per_second))
start_time = self._format_time(start_sec)
end_time = self._format_time(end_sec)
@@ -299,6 +334,8 @@ class DeepAnalyzer:
'index': i,
'start_time': start_time,
'end_time': end_time,
'start_sec': start_sec,
'end_sec': end_sec,
'summary': chunk_summary
})
@@ -308,6 +345,58 @@ class DeepAnalyzer:
return result
def _find_breakpoint(self, text: str, start: int, preferred_end: int) -> int:
"""
在自然断点处分割文本
优先级:
1. 双换行(段落分隔)
2. 句号+换行(句末)
3. 句号(句末)
4. 换行(行末)
5. 回到preferred_end
Args:
text: 完整文本
start: 起始位置
preferred_end: 理想结束位置
Returns:
实际结束位置
"""
if not text or start >= len(text):
return start
# 搜索范围:理想位置前后各500字符
search_start = max(start, preferred_end - 500)
search_end = min(len(text), preferred_end + 500)
# 优先级1: 双换行
double_newline = text.rfind('\n\n', search_start, search_end)
if double_newline != -1:
return double_newline
# 优先级2: 句号+换行
period_newline = text.rfind('。\n', search_start, search_end)
if period_newline != -1:
return period_newline + 1 # 包含句号
# 优先级3: 句号(中文或英文)
last_period = max(
text.rfind('。', search_start, search_end),
text.rfind('.', search_start, search_end)
)
if last_period != -1:
return last_period + 1 # 包含句号
# 优先级4: 换行
last_newline = text.rfind('\n', search_start, search_end)
if last_newline != -1:
return last_newline
# 没找到自然断点,返回理想位置
return preferred_end
def _analyze_chunk(self, chunk_text: str, chunk_index: int) -> str:
"""分析单个分片"""
text_sample = self._get_text_sample(chunk_text, max_chars=3000)
@@ -435,21 +524,6 @@ class DeepAnalyzer:
text[-part_size:]
)
def _chunk_subtitle(self, subtitle: str, num_chunks: int) -> List[str]:
"""简单分割字幕"""
if not subtitle:
return [""] * num_chunks
chunk_size = len(subtitle) // num_chunks
chunks = []
for i in range(num_chunks):
start = i * chunk_size
end = start + chunk_size if i < num_chunks - 1 else len(subtitle)
chunks.append(subtitle[start:end])
return chunks
def _generate_simple_timeline_from_duration(self, title: str, duration: int) -> List[Dict]:
"""生成简单时间轴"""
segments = 2 if duration < 300 else 3