"""ArchiveProcessor - Copy analysis reports to archive directory Final step in pipeline that copies the generated analysis.md to the archive directory organized by platform. """ from __future__ import annotations import logging import re import shutil from datetime import datetime from pathlib import Path from typing import Optional from kb_workflow.pipeline.config import get_settings from kb_workflow.pipeline.schemas import ArchiveInput, ArchiveOutput from kb_workflow.pipeline.state import WorkflowStateManager from kb_workflow.pipeline.processors.base import BaseProcessor logger = logging.getLogger(__name__) class ArchiveProcessor(BaseProcessor): """Archive 处理器 - 归档分析报告 处理流程： 1. 检查步骤是否已完成（幂等执行） 2. 检查归档目录是否存在相同 content_id 的文件 3. 如果存在重复，询问用户是否覆盖 4. 复制 md 文件到归档目录 5. 记录处理状态到 workflow_state.json 归档目录结构： {KB_ARCHIVE_PATH}/{platform}/{YYYYMMDD}_{content_id}_analysis.md """ STEP_NAME = "archive" async def process(self, input_data: ArchiveInput) -> ArchiveOutput: """执行归档处理 Args: input_data: Archive 输入数据 Returns: Archive 处理结果 """ try: import sys # 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") return ArchiveOutput( success=True, skipped=True, platform=input_data.platform, archive_file="" ) # Step 2: 准备归档路径 archive_base = Path(self.settings.archive_path) platform_dir = archive_base / input_data.platform platform_dir.mkdir(parents=True, exist_ok=True) # 从源文件中提取标题（用于文件命名） title = self._extract_title_from_file(Path(input_data.analysis_file)) # 清理标题：移除特殊字符，限制长度 safe_title = self._sanitize_title(title) if title else input_data.content_id # 文件命名: {YYYYMMDD}_{content_id}_{title}_analysis.md timestamp = datetime.now().strftime("%Y%m%d") archive_filename = f"{timestamp}_{input_data.content_id}_{safe_title}_analysis.md" archive_path = platform_dir / archive_filename # Step 3: 检查重复 existing_file = self._find_existing_archive( platform_dir, input_data.content_id ) if existing_file: # 检测是否为非交互式环境 is_non_interactive = not sys.stdin.isatty() should_overwrite = self._ask_overwrite(existing_file, is_non_interactive) if not should_overwrite: logger.info(f"User chose not to overwrite, skipping archive") return ArchiveOutput( success=True, skipped=True, platform=input_data.platform, archive_file=str(existing_file) ) # 删除旧文件 existing_file.unlink() logger.info(f"Removed existing archive: {existing_file}") # Step 4: 复制文件 source_path = Path(input_data.analysis_file) if not source_path.exists(): raise FileNotFoundError(f"Analysis file not found: {source_path}") shutil.copy2(source_path, archive_path) logger.info(f"Archived: {source_path} -> {archive_path}") # Step 5: 记录成功状态 if self.state_manager: state = self.state_manager.load() if state: state.mark_step_completed( self.STEP_NAME, f"archived to {archive_path}" ) self.state_manager.save(state) return ArchiveOutput( success=True, skipped=False, platform=input_data.platform, archive_file=str(archive_path) ) except Exception as e: logger.error(f"Archive 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 ArchiveOutput( success=False, error_message=str(e), platform=input_data.platform, archive_file="" ) def _find_existing_archive(self, platform_dir: Path, content_id: str) -> Optional[Path]: """查找已存在的归档文件 使用 glob 模式 `*_{content_id}_analysis.md` 搜索 Args: platform_dir: 平台目录路径 content_id: 内容 ID Returns: 已存在的文件路径，如果不存在返回 None """ pattern = f"*_{content_id}_analysis.md" matches = list(platform_dir.glob(pattern)) return matches[0] if matches else None def _extract_title_from_file(self, file_path: Path) -> Optional[str]: """从 Markdown 文件的 frontmatter 中提取标题 Args: file_path: Markdown 文件路径 Returns: 标题字符串，如果未找到返回 None """ try: content = file_path.read_text(encoding='utf-8') # 查找 frontmatter 中的 title 字段 # 格式：title: "标题内容" 或 title: 标题内容 title_match = re.search(r'^title:\s*(?:"([^"]*)"|(.+))$', content, re.MULTILINE) if title_match: # 优先使用引号内的内容，否则使用第二个捕获组 title = title_match.group(1) or title_match.group(2) return title.strip() if title else None return None except Exception as e: logger.warning(f"Failed to extract title from {file_path}: {e}") return None def _sanitize_title(self, title: str) -> str: """清理标题用于文件命名 移除特殊字符，限制长度，转换为安全的文件名。 Args: title: 原始标题 Returns: 清理后的标题 """ # 移除或替换不适合文件名的字符 # 保留中文、字母、数字、下划线、连字符 # 替换所有特殊字符（包括标点符号和常见符号） safe = re.sub(r'[\\/:*?"<>|，。！？、；：""\'\'\s@#$%&*+\-=\[\]{}()<>]+', '_', title) # 移除连续的下划线 safe = re.sub(r'_+', '_', safe) # 去除首尾下划线和空格 safe = safe.strip('_') # 限制长度（避免文件名过长） if len(safe) > 50: safe = safe[:50] # 如果清理后为空，返回默认值 if not safe: safe = "untitled" return safe def _ask_overwrite(self, existing_file: Path, is_non_interactive: bool = False) -> bool: """询问用户是否覆盖已存在的文件 使用 rich.prompt.Confirm 模式（与 Pipeline._ask_rerun_step 相同） Args: existing_file: 已存在的文件路径 is_non_interactive: 是否为非交互式环境（stdin不是tty） Returns: True 如果用户选择覆盖 """ # 非交互式模式：使用配置决定 if is_non_interactive: if self.settings.overwrite_archives: logger.info(f"Non-interactive mode: Overwriting existing archive (KB_OVERWRITE_ARCHIVES=true): {existing_file.name}") return True else: logger.info(f"Non-interactive mode: Skipping existing archive (KB_OVERWRITE_ARCHIVES=false): {existing_file.name}") return False # 交互式模式：询问用户 try: from rich.console import Console from rich.prompt import Confirm console = Console() return Confirm.ask( f"Archive already exists: {existing_file.name}\nOverwrite?", default=False ) except ImportError: response = input( f"Archive already exists: {existing_file.name}\nOverwrite? [y/N] " ) return response.lower() in ('y', 'yes')