From c8ddc1e83f7ee2265b114a6729df8cc1c983d2c5 Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Fri, 13 Feb 2026 14:47:43 +0800 Subject: [PATCH] =?UTF-8?q?feat(kb-deep-analyzer):=20=E6=B7=BB=E5=8A=A0?= =?UTF-8?q?=E6=B5=81=E5=BC=8F=E5=86=99=E5=85=A5=E5=99=A8=20StreamingWriter?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 支持多线程并发写入到文档不同位置 - 初始化时创建带占位符的文档结构 - 每批完成后写入对应位置 - 支持更新摘要、主题、见解等全局内容 --- skills/kb-deep-analyzer/deep_analyzer.py | 176 +++++++++++++++++++++++ 1 file changed, 176 insertions(+) diff --git a/skills/kb-deep-analyzer/deep_analyzer.py b/skills/kb-deep-analyzer/deep_analyzer.py index 8bca524..68677bc 100644 --- a/skills/kb-deep-analyzer/deep_analyzer.py +++ b/skills/kb-deep-analyzer/deep_analyzer.py @@ -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"""