feat(deep_analyzer): 使用SRT简化时间格式 + 分块分析 + 纯文本AI

- 添加 load_srt() 支持 SRT 字幕格式
- 时间格式简化为 HH:MM:SS
- 超过3000字自动分块分析
- _merge_summaries 改用纯文本而非JSON
- 添加提取函数解析AI响应

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

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,6 +1242,12 @@ def main():
     metadata = normalize_metadata(raw_metadata)
 
     print(f"[Main] 加载内容: {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)
 
     # 初始化

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")