feat(kb-deep-analyzer): 添加流式写入器 StreamingWriter

- 支持多线程并发写入到文档不同位置 - 初始化时创建带占位符的文档结构 - 每批完成后写入对应位置 - 支持更新摘要、主题、见解等全局内容
2026-02-13 14:47:43 +08:00
parent 94370f225e
commit c8ddc1e83f
1 changed files with 176 additions and 0 deletions
@@ -47,6 +47,182 @@ class AnalysisResult:

 # ============== AI 客户端 ==============

+# ============== 流式写入器 ==============
+
+import threading
+
+
+class StreamingWriter:
+    """
+    流式写入器 - 支持多线程并发写入到文档不同位置
+    
+    使用方法:
+    1. 初始化: writer = StreamingWriter(output_path, num_chunks)
+    2. 预先创建带占位符的文档结构
+    3. 每批完成后调用 writer.write_batch(batch_results)
+    4. 完成后调用 writer.mark_complete() 清理占位符
+    """
+
+    def __init__(self, output_path: str, num_chunks: int = 0, batch_size: int = 3):
+        self.output_path = Path(output_path)
+        self.num_chunks = num_chunks
+        self.batch_size = batch_size
+        self.lock = threading.Lock()
+        self.written = {}  # 记录已写入的位置
+        self.batch_num = 0
+        
+    def initialize_document(self, metadata: Dict, num_chunks: int) -> str:
+        """
+        初始化文档结构，返回带占位符的内容
+        
+        文档结构:
+        ---
+        title: "xxx"
+        ...
+        ---
+        
+        # 标题
+        
+        ## 摘要
+        
+        ## 主要主题
+        
+        ## 关键见解
+        
+        ## 建议行动
+        
+        ## 时间轴
+        ### 第 N 部分 (00:00 - 00:00)
+        [ANALYZING...]
+        """
+        self.num_chunks = num_chunks
+        
+        # 生成时间轴占位符
+        timeline_placeholders = ""
+        for i in range(num_chunks):
+            timeline_placeholders += f"""### 第 {i + 1} 部分 (待分析)
+[ANALYZING_BATCH_{i // self.batch_size + 1}]
+
+"""
+
+        content = f"""---
+title: "{metadata.get('title', '未命名')}"
+date: {datetime.now().strftime('%Y-%m-%d')}
+duration: {format_duration(metadata.get('duration', 0))}
+author: {metadata.get('author', '未知')}
+analyze_time: {datetime.now().strftime('%Y-%m-%d %H:%M')}
+bvid: {metadata.get('bvid', '')}
+platform: {metadata.get('platform', 'bilibili')}
+---
+
+# {metadata.get('title', '未命名')}
+
+## 摘要
+[WAITING_FOR_ANALYSIS...]
+
+## 主要主题
+[WAITING_FOR_ANALYSIS...]
+
+## 关键见解
+[WAITING_FOR_ANALYSIS...]
+
+## 建议行动
+[WAITING_FOR_ANALYSIS...]
+
+## 时间轴
+{timeline_placeholders}
+
+---
+*分析时间: {datetime.now().strftime('%Y-%m-%d %H:%M')}*
+"""
+        
+        # 写入初始化内容
+        self.output_path.parent.mkdir(parents=True, exist_ok=True)
+        with open(self.output_path, 'w', encoding='utf-8') as f:
+            f.write(content)
+        
+        print(f"[StreamingWriter] 初始化文档: {self.output_path}")
+        return content
+    
+    def write_batch(self, batch_results: List[Dict], num_batches: int) -> None:
+        """
+        写入一批结果
+        
+        Args:
+            batch_results: 当前批的结果列表
+            num_batches: 总批次数
+        """
+        self.batch_num += 1
+        
+        with self.lock:
+            # 读取当前内容
+            if not self.output_path.exists():
+                print(f"[StreamingWriter] 警告: 文件不存在 {self.output_path}")
+                return
+                
+            content = self.output_path.read_text(encoding='utf-8')
+            
+            for result in batch_results:
+                idx = result['index']
+                start_time = result['start_time']
+                end_time = result['end_time']
+                summary = result['summary']
+                
+                # 查找并替换占位符
+                old_placeholder = f"### 第 {idx + 1} 部分 ({start_time} - {end_time})\n[ANALYZING_BATCH_{self.batch_num}]"
+                new_content = f"""### 第 {idx + 1} 部分 ({start_time} - {end_time})
+{summary}"""
+                
+                content = content.replace(old_placeholder, new_content, 1)
+                self.written[idx] = True
+                
+                print(f"[StreamingWriter] 写入段 {idx + 1}/{self.num_chunks}")
+            
+            # 如果这是最后一批，更新其他状态
+            if self.batch_num >= num_batches:
+                # 更新摘要等
+                pass
+            
+            # 写回文件
+            self.output_path.write_text(content, encoding='utf-8')
+            print(f"[StreamingWriter] 批次 {self.batch_num}/{num_batches} 完成")
+    
+    def update_summary(self, summary_text: str, topics: List[str], insights: List[str]) -> None:
+        """更新摘要等全局内容"""
+        with self.lock:
+            if not self.output_path.exists():
+                return
+            
+            content = self.output_path.read_text(encoding='utf-8')
+            
+            # 更新摘要
+            if "[WAITING_FOR_ANALYSIS...]" in content:
+                content = content.replace("## 摘要\n[WAITING_FOR_ANALYSIS...]", f"## 摘要\n{summary_text}")
+            
+            # 更新主题
+            topics_text = "\n".join(f"- {t}" for t in topics)
+            if "## 主要主题\n[WAITING_FOR_ANALYSIS...]" in content:
+                content = content.replace("## 主要主题\n[WAITING_FOR_ANALYSIS...]", f"## 主要主题\n{topics_text}")
+            
+            # 更新见解
+            insights_text = "\n".join(f"- {i}" for i in insights)
+            if "## 关键见解\n[WAITING_FOR_ANALYSIS...]" in content:
+                content = content.replace("## 关键见解\n[WAITING_FOR_ANALYSIS...]", f"## 关键见解\n{insights_text}")
+            
+            self.output_path.write_text(content, encoding='utf-8')
+            print(f"[StreamingWriter] 更新摘要和主题")
+    
+    def cleanup_placeholder(self, placeholder: str) -> None:
+        """清理单个占位符"""
+        with self.lock:
+            if not self.output_path.exists():
+                return
+            
+            content = self.output_path.read_text(encoding='utf-8')
+            content = content.replace(placeholder, "")
+            self.output_path.write_text(content, encoding='utf-8')
+
+
 class SimpleAIClient:
    """简化的 AI 客户端 - 支持多提供商 fallback"""