From b7b40b1ece564fb72e714433c3d471c753e322a3 Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Fri, 13 Feb 2026 12:28:21 +0800 Subject: [PATCH] =?UTF-8?q?feat(deep=5Fanalyzer):=20=E4=BD=BF=E7=94=A8SRT?= =?UTF-8?q?=E7=AE=80=E5=8C=96=E6=97=B6=E9=97=B4=E6=A0=BC=E5=BC=8F=20+=20?= =?UTF-8?q?=E5=88=86=E5=9D=97=E5=88=86=E6=9E=90=20+=20=E7=BA=AF=E6=96=87?= =?UTF-8?q?=E6=9C=ACAI?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 添加 load_srt() 支持 SRT 字幕格式 - 时间格式简化为 HH:MM:SS - 超过3000字自动分块分析 - _merge_summaries 改用纯文本而非JSON - 添加提取函数解析AI响应 --- memory/2026-02-13.md | 31 ++ skills/kb-deep-analyzer/deep_analyzer.py | 286 +++++++++++++++++- .../whisper_diarization.py | 10 +- 3 files changed, 307 insertions(+), 20 deletions(-) diff --git a/memory/2026-02-13.md b/memory/2026-02-13.md index 0af2bdc..6a1d8be 100644 --- a/memory/2026-02-13.md +++ b/memory/2026-02-13.md @@ -3,6 +3,37 @@ ## 目标 整合说话者分离（speaker diarization）到 kb-whisper 工作流，替换 OCR 为 Whisper 视频转文字 +## glm-4.7 AI 分析返回问题（2026-02-13 新增） + +### 发现 +- deep_analyzer 使用 `.json` 文件传给 AI +- 实际传给 AI 的是 `text` 字段的前 3000 字 +- json 文件结构： + ``` + ├── text: "為什麼我要說美國就變了..." (~4500字) + └── segments: 168段 (带时间戳) + ``` + +### 三种字幕格式对比 +| 格式 | 大小 | 特点 | +|------|------|------| +| .json | 55KB | 完整元数据 + segments | +| .srt | 23KB | 带时间戳的字幕格式 | +| .txt | 9KB | 纯文本，无时间戳 | + +### 问题分析 +1. **截断问题**：只取前 3000 字传给 AI，可能不够全面 +2. **未利用时间轴**：segments 有 168 段时间信息未使用 +3. **格式问题**：传给 AI 是纯文本，没有分段结构 + +### 待用户确认改进方案 +- A. 用 `.txt` (9KB) 更简洁的格式？ +- B. 用 `.srt` (23KB) 带时间戳的格式？ +- C. 改进 json 解析，把 segments 整理成更易读的格式？ +- D. 其他方案？ + +--- + ## Passwordless Sudo 配置成功 ```bash sudo visudo -f /etc/sudoers.d/nopasswd diff --git a/skills/kb-deep-analyzer/deep_analyzer.py b/skills/kb-deep-analyzer/deep_analyzer.py index 583be0b..c3dd188 100644 --- a/skills/kb-deep-analyzer/deep_analyzer.py +++ b/skills/kb-deep-analyzer/deep_analyzer.py @@ -270,9 +270,17 @@ class DeepAnalyzer: print(f"[DeepAnalyzer] 开始分析: {title}") print(f"[DeepAnalyzer] 类型: {content_type}, 时长: {duration}秒") print(f"[DeepAnalyzer] 文本存在: {has_content}, 分段数: {len(segments)}") + print(f"[DeepAnalyzer] 文本长度: {len(text) if text else 0} 字符") + + # 强制分块阈值：无论视频长短，超过3000字都用分块分析 + CHUNK_THRESHOLD = 3000 + use_chunking = has_content and len(text) > CHUNK_THRESHOLD if has_content or has_segments: - if duration < self.SHORT_VIDEO_THRESHOLD: + if use_chunking: + print(f"[DeepAnalyzer] 文本超过 {CHUNK_THRESHOLD} 字，使用分块分析") + result = self._analyze_in_chunks_with_segments(title, text, duration, segments) + elif duration < self.SHORT_VIDEO_THRESHOLD: print(f"[DeepAnalyzer] 使用轻量分析策略 (时长 < 10分钟)") result = self._analyze_short_with_subtitle(title, text, duration, segments) else: @@ -396,6 +404,98 @@ class DeepAnalyzer: return result + def _analyze_in_chunks_with_segments(self, title: str, text: str, duration: int, segments: list = None) -> AnalysisResult: + """ + 使用实际字幕时间戳进行分块分析 + + 当有 SRT 格式的 segments 时，使用实际时间戳来分块 + 否则退化为字符位置估算 + """ + result = AnalysisResult() + + CHUNK_MAX_CHARS = 3000 + total_chars = len(text) + + # 检查是否有有效的 SRT segments + has_valid_segments = segments and len(segments) > 0 and all('start' in seg and 'text' in seg for seg in segments) + + if has_valid_segments: + print(f"[DeepAnalyzer] 使用 SRT 时间戳进行分块") + # 按时间戳分块 + chunks = [] + chunk_times = [] + current_chunk = "" + current_start = segments[0]['start'] + current_end = "" + + for seg in segments: + seg_text = f"[{seg['start']}] {seg['text']}\n" + + if len(current_chunk) + len(seg_text) > CHUNK_MAX_CHARS and current_chunk: + # 保存当前块 + chunks.append(current_chunk.strip()) + chunk_times.append((current_start, current_end or seg['start'])) + # 开始新块 + current_chunk = seg_text + current_start = seg['start'] + else: + current_chunk += seg_text + current_end = seg['end'] + + # 保存最后一块 + if current_chunk: + chunks.append(current_chunk.strip()) + chunk_times.append((current_start, current_end)) + + print(f"[DeepAnalyzer] SRT分块: {len(chunks)} 段") + else: + # 退化为字符位置估算 + print(f"[DeepAnalyzer] 无有效时间戳，使用字符位置分块") + min_required = total_chars // CHUNK_MAX_CHARS + 1 + num_segments = min(max(min_required, 3), 20) + + chunk_size = total_chars // num_segments + chunks = [] + chunk_times = [] + + for i in range(num_segments): + start = i * chunk_size + if i == num_segments - 1: + end = total_chars + else: + end = self._find_breakpoint(text, start, start + chunk_size) + + chunks.append(text[start:end]) + # 估算时间 + char_per_second = total_chars / duration if duration > 0 else 1 + start_sec = int(start / char_per_second) + end_sec = int(end / char_per_second) + chunk_times.append((self._format_time(start_sec), self._format_time(end_sec))) + + print(f"[DeepAnalyzer] 字符分块: {len(chunks)} 段") + + print(f"[DeepAnalyzer] 各段字符: {[len(c) for c in chunks]}") + + # 分析每个块 + chunk_summaries = [] + for i, chunk_text in enumerate(chunks): + start_time, end_time = chunk_times[i] + + # AI 分析每段 + chunk_summary = self._analyze_chunk(chunk_text, i + 1) + chunk_summaries.append({ + 'index': i, + 'start_time': start_time, + 'end_time': end_time, + 'summary': chunk_summary + }) + + # 合并结果 + result = self._merge_summaries(chunk_summaries, title) + result.timeline = self._build_timeline_from_chunks(chunk_summaries) + + return result + def _find_breakpoint(self, text: str, start: int, preferred_end: int) -> int: """ 在自然断点处分割文本 @@ -472,6 +572,7 @@ class DeepAnalyzer: for s in chunk_summaries ]) + # 使用纯文本提示，不再要求 JSON 格式 prompt = f"""基于以下视频各段分析，生成完整详细的分析报告： 标题：{title} @@ -479,23 +580,24 @@ class DeepAnalyzer: 各段内容： {summaries_text} -请返回JSON格式： -{{ - "summary": "整体摘要（300字以内）", - "topics": ["主题1", "主题2", "主题3", "主题4", "主题5"], - "insights": ["见解1", "见解2", "见解3", "见解4"], - "actions": ["建议1", "建议2", "建议3"], - "full_analysis": "详细分析（1500字以内）" -}}""" +请用中文详细回复，包含： +1. 整体摘要（300字以内） +2. 5个核心主题 +3. 4个关键见解 +4. 3条建议 +5. 详细分析（1500字以内）""" try: - analysis = self.ai_client.generate_json(prompt) + # 使用纯文本生成而非 JSON + full_response = self.ai_client.generate(prompt, max_tokens=4000) + + # 解析响应 result = AnalysisResult() - result.summary = analysis.get('summary', title) - result.main_topics = analysis.get('topics', []) - result.key_insights = analysis.get('insights', []) - result.action_items = analysis.get('actions', []) - result.full_analysis = analysis.get('full_analysis', summaries_text) + result.summary = self._extract_summary(full_response) + result.main_topics = self._extract_topics(full_response) + result.key_insights = self._extract_insights(full_response) + result.action_items = self._extract_actions(full_response) + result.full_analysis = full_response return result except Exception as e: print(f"[DeepAnalyzer] 合并摘要失败: {e}") @@ -506,6 +608,61 @@ class DeepAnalyzer: full_analysis=summaries_text[:500] ) + def _extract_summary(self, text: str) -> str: + """从 AI 响应中提取摘要""" + lines = text.split('\n') + summary_lines = [] + capture = False + for line in lines: + if '摘要' in line or '总结' in line: + capture = True + if capture and line.strip(): + summary_lines.append(line.strip()) + if len(summary_lines) >= 3: + break + return ' '.join(summary_lines[:3])[:300] if summary_lines else text[:200] + + def _extract_topics(self, text: str) -> List[str]: + """从 AI 响应中提取主题""" + topics = [] + lines = text.split('\n') + for line in lines: + # 匹配常见主题格式 + if any(m in line for m in ['主题', '核心', '要点', '1.', '2.', '3.', '4.', '5.']): + # 清理并提取 + cleaned = line.strip().lstrip('0123456789.、 ').strip() + if cleaned and len(cleaned) > 2: + topics.append(cleaned) + if len(topics) >= 5: + break + return topics[:5] if topics else [f"主题{i+1}" for i in range(5)] + + def _extract_insights(self, text: str) -> List[str]: + """从 AI 响应中提取见解""" + insights = [] + lines = text.split('\n') + for line in lines: + if any(m in line for m in ['见解', '观点', '发现', '洞察']): + cleaned = line.strip().lstrip('0123456789.、 ').strip() + if cleaned and len(cleaned) > 2: + insights.append(cleaned) + if len(insights) >= 4: + break + return insights[:4] if insights else ["综合分析得出"] + + def _extract_actions(self, text: str) -> List[str]: + """从 AI 响应中提取建议""" + actions = [] + lines = text.split('\n') + for line in lines: + if any(m in line for m in ['建议', '行动', '做法', '可以', '应该']): + cleaned = line.strip().lstrip('0123456789.、 ').strip() + if cleaned and len(cleaned) > 2: + actions.append(cleaned) + if len(actions) >= 3: + break + return actions[:3] if actions else ["暂无建议"] + def _analyze_fallback(self, title: str, description: str, metadata: Dict) -> AnalysisResult: """无字幕降级处理""" result = AnalysisResult() @@ -838,6 +995,97 @@ def load_json(file_path: str) -> Dict: return json.load(f) +def load_srt(file_path: str) -> Dict: + """ + 加载并解析 SRT 字幕文件 + + 返回格式: + { + "text": "纯文本内容", + "segments": [ + {"start": "00:00:00", "end": "00:00:10", "text": "字幕文本"} + ] + } + """ + if not file_path or not os.path.exists(file_path): + return {} + + with open(file_path, 'r', encoding='utf-8') as f: + content = f.read() + + # 解析 SRT 格式 + # 格式: 序号 -> 时间 -> 文本 -> 空行 + blocks = re.split(r'\n\s*\n', content.strip()) + segments = [] + + for block in blocks: + lines = block.strip().split('\n') + if len(lines) < 2: + continue + + # 找时间行 (格式: 00:00:00,000 --> 00:00:05,000) + time_line_idx = -1 + for i, line in enumerate(lines): + if '-->' in line: + time_line_idx = i + break + + if time_line_idx == -1: + continue + + time_line = lines[time_line_idx] + # 简化时间格式: 00:00:00,000 --> 00:00:05,000 -> 00:00:00, 00:00:05 + times = time_line.split('-->') + if len(times) != 2: + continue + + # 转换为 HH:MM:SS 格式 + start_raw = times[0].strip().replace(',', '.') + end_raw = times[1].strip().replace(',', '.') + + start_time = _simplify_srt_time(start_raw) + end_time = _simplify_srt_time(end_raw) + + # 文本行 + text_lines = lines[time_line_idx + 1:] + text = ' '.join(line.strip() for line in text_lines) + + if text: + segments.append({ + "start": start_time, + "end": end_time, + "text": text + }) + + # 生成纯文本（每句前面加时间戳） + text_lines = [] + for seg in segments: + text_lines.append(f"[{seg['start']}] {seg['text']}") + + return { + "text": '\n'.join(text_lines), + "segments": segments + } + + +def _simplify_srt_time(time_str: str) -> str: + """ + 将 SRT 时间简化为 HH:MM:SS 格式 + + 输入: 00:00:00.000 或 00:00:00,000 + 输出: 00:00:00 + """ + # 移除毫秒和逗号 + time_str = re.sub(r'[,\.]\d{3}$', '', time_str) + # 确保是 HH:MM:SS 格式 + parts = time_str.split(':') + if len(parts) == 3: + return f"{parts[0]}:{parts[1]}:{parts[2]}" + elif len(parts) == 2: + return f"00:{parts[0]}:{parts[1]}" + return time_str + + def normalize_metadata(metadata: Dict) -> Dict: """ 标准化元数据格式，支持多种嵌套结构 @@ -994,7 +1242,13 @@ def main(): metadata = normalize_metadata(raw_metadata) print(f"[Main] 加载内容: {args.content}") - content = load_json(args.content) + # 支持 SRT 和 JSON 格式 + content_path = Path(args.content) + if content_path.suffix.lower() == '.srt': + print(f"[Main] 检测到 SRT 格式") + content = load_srt(args.content) + else: + content = load_json(args.content) # 初始化 ai_client = SimpleAIClient( diff --git a/skills/kb-whisper-diarization/whisper_diarization.py b/skills/kb-whisper-diarization/whisper_diarization.py index a73ea0c..805fb4c 100644 --- a/skills/kb-whisper-diarization/whisper_diarization.py +++ b/skills/kb-whisper-diarization/whisper_diarization.py @@ -127,11 +127,13 @@ def run_diarization(audio_path, num_speakers=None): raise RuntimeError("需要设置 HF_TOKEN 环境变量\n" "获取方式: https://huggingface.co/settings/tokens") + # 新版 pyannote 需要先登录 + from huggingface_hub import login + print("登录 HuggingFace...") + login(token=hf_token) + print("加载 pyannote 流水线...") - pipeline = Pipeline.from_pretrained( - "pyannote/speaker-diarization", - use_auth_token=hf_token - ) + pipeline = Pipeline.from_pretrained("pyannote/speaker-diarization") # 设置设备 device = torch.device("cuda" if torch.cuda.is_available() else "cpu")