"""AnalyzeProcessor - AI-powered content analysis Unified processor for generating AI analysis reports from content. Wraps existing DeepAnalyzer for integration into the pipeline. """ from __future__ import annotations import json import logging import os from pathlib import Path from typing import Optional, Dict, Any from kb_workflow.pipeline.config import get_settings from kb_workflow.pipeline.schemas import AnalyzeInput, AnalyzeOutput from kb_workflow.pipeline.state import WorkflowStateManager from kb_workflow.pipeline.processors.base import BaseProcessor from kb_workflow.deep_analyzer import DeepAnalyzer, SimpleAIClient, generate_frontmatter logger = logging.getLogger(__name__) class AnalyzeProcessor(BaseProcessor): """Analyze 处理器 - AI 深度内容分析 处理流程： 1. 检查步骤是否已完成（幂等执行） 2. 加载元数据和内容文件 3. 初始化 AI 客户端 4. 调用 DeepAnalyzer 生成分析报告 5. 生成 frontmatter 并保存 Markdown 文件 6. 记录处理状态到 workflow_state.json Attributes: STEP_NAME: 步骤名称，用于状态记录 settings: KB 配置 state_manager: 状态管理器（可选） """ STEP_NAME = "analyze" async def process(self, input_data: AnalyzeInput) -> AnalyzeOutput: """执行分析处理 策略： 1. 检查步骤是否已完成（跳过已完成） 2. 加载元数据和内容 3. 调用 DeepAnalyzer 4. 保存分析报告 5. 记录状态 Args: input_data: Analyze 输入数据 Returns: Analyze 处理结果 Raises: Exception: 处理失败时抛出异常 """ try: # Step 1: 检查是否已完成（幂等执行） if self.state_manager: state = self.state_manager.load() if state and state.is_step_completed(self.STEP_NAME): logger.info(f"Step {self.STEP_NAME} already completed, skipping") # 返回已有结果 existing_file = Path(input_data.output_file) if existing_file.exists(): return AnalyzeOutput( success=True, analysis_file=str(existing_file), summary="Skipped - already completed", main_topics=[], key_insights=[], tags=[], stats={"skipped": True} ) # Step 2: 加载元数据和内容 metadata = self._load_json(input_data.metadata_file) content = self._load_json(input_data.content_file) # 处理 Whisper 字幕格式 - 提取完整文本 if content and 'transcription' in content: # Whisper JSON 格式，提取文本 text_parts = [] for item in content.get('transcription', []): text_parts.append(item.get('text', '')) content = {'text': ' '.join(text_parts)} logger.info(f"Extracted {len(content['text'])} chars from Whisper transcription") # 对于文章类型，内容可能是纯文本文件 if not content and input_data.content_type in ("article", "answer", "zhuanlan"): content_path = Path(input_data.content_file) if content_path.exists(): text = content_path.read_text(encoding="utf-8") # 去除 YAML frontmatter（如果存在） text = self._strip_frontmatter(text) content = {"text": text} # Step 3: 调用 DeepAnalyzer result = await self._run_analysis( metadata=metadata, content=content, content_type=input_data.content_type ) # Step 4: 生成并保存分析报告 output_path = Path(input_data.output_file) output_path.parent.mkdir(parents=True, exist_ok=True) self._write_analysis_report( output_path=output_path, metadata=metadata, result=result, content_type=input_data.content_type ) # Step 5: 记录成功状态 if self.state_manager: state = self.state_manager.load() if state: summary = f"topics={len(result.main_topics)}, insights={len(result.key_insights)}" state.mark_step_completed(self.STEP_NAME, summary) self.state_manager.save(state) return AnalyzeOutput( success=True, analysis_file=str(output_path), summary=result.summary, main_topics=result.main_topics, key_insights=result.key_insights, tags=result.recommended_tags, stats={ "topics_count": len(result.main_topics), "insights_count": len(result.key_insights), "tags_count": len(result.recommended_tags) } ) except Exception as e: logger.error(f"Analyze failed: {e}") # 记录失败状态 if self.state_manager: state = self.state_manager.load() if state: state.mark_step_failed(self.STEP_NAME, str(e)) self.state_manager.save(state) return AnalyzeOutput( success=False, error_message=str(e), analysis_file="", summary="", main_topics=[], key_insights=[], tags=[], stats={} ) def _load_json(self, file_path: str) -> Dict[str, Any]: """加载 JSON 文件 Args: file_path: JSON 文件路径 Returns: 解析后的字典，如果文件不存在或解析失败返回空字典 """ if not file_path: return {} path = Path(file_path) if not path.exists(): logger.warning(f"File not found: {file_path}") return {} try: with open(path, 'r', encoding='utf-8') as f: return json.load(f) except json.JSONDecodeError as e: logger.error(f"Failed to parse JSON: {file_path}, error: {e}") return {} async def _run_analysis(self, metadata: Dict, content: Dict, content_type: str): """调用 DeepAnalyzer 执行分析 Args: metadata: 元数据字典 content: 内容字典 content_type: 内容类型 Returns: AnalysisResult 对象 """ # 初始化 AI 客户端 ai_client = self._create_ai_client() # 创建分析器实例 analyzer = DeepAnalyzer(ai_client=ai_client) # 执行分析 return analyzer.analyze(metadata, content, content_type) def _create_ai_client(self): """创建 AI 客户端 Returns: SimpleAIClient 实例 """ # 从环境变量获取配置 api_key = os.environ.get("AI_API_KEY", "") provider = os.environ.get("AI_PROVIDER", "") model = os.environ.get("AI_MODEL", "glm-5") base_url = os.environ.get("AI_BASE_URL", "") return SimpleAIClient( api_key=api_key, provider=provider, model=model, base_url=base_url ) def _normalize_metadata(self, metadata: Dict) -> Dict: """标准化元数据结构（扁平化嵌套结构） Args: metadata: 原始元数据字典，可能是嵌套结构 Returns: 扁平化的元数据字典 """ if not metadata: return {} # 如果已经是扁平结构（顶层有 title） if 'title' in metadata: return metadata # 处理嵌套结构 {metadata: {...}, stats: {...}} if 'metadata' in metadata: return metadata['metadata'].copy() return metadata def _strip_frontmatter(self, text: str) -> str: """去除 Markdown 文件中的 YAML frontmatter Args: text: 可能包含 frontmatter 的文本 Returns: 去除 frontmatter 后的文本 """ if not text: return text # 检查是否以 --- 开头 if not text.startswith('---'): return text # 找到第二个 --- lines = text.split('\n') end_index = -1 for i in range(1, len(lines)): if lines[i].strip() == '---': end_index = i break if end_index == -1: return text # 返回 frontmatter 之后的内容（跳过空行） remaining = '\n'.join(lines[end_index + 1:]) return remaining.lstrip('\n') def _write_analysis_report( self, output_path: Path, metadata: Dict, result, content_type: str ) -> None: """写入分析报告 Markdown 文件 Args: output_path: 输出文件路径 metadata: 元数据字典 result: AnalysisResult 对象 content_type: 内容类型 """ # 标准化元数据（扁平化嵌套结构） normalized_metadata = self._normalize_metadata(metadata) # 生成 frontmatter frontmatter = generate_frontmatter(normalized_metadata, result, content_type) # 获取完整分析内容 full_analysis = result.full_analysis # 文章/回答类型：追加原文 if content_type in ("article", "answer", "zhuanlan") and result.original_text: full_analysis = full_analysis.rstrip() + "\n\n## 原文\n\n" + result.original_text # 写入文件 with open(output_path, 'w', encoding='utf-8') as f: f.write(frontmatter + "\n\n") f.write(full_analysis) logger.info(f"Analysis report saved to: {output_path}")