diff --git a/skills/kb-deep-analyzer/deep_analyzer.py b/skills/kb-deep-analyzer/deep_analyzer.py index 7f51062..889cac0 100644 --- a/skills/kb-deep-analyzer/deep_analyzer.py +++ b/skills/kb-deep-analyzer/deep_analyzer.py @@ -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