diff --git a/kb_pipeline/processors/__init__.py b/kb_pipeline/processors/__init__.py index 8cf19c2..e74f168 100644 --- a/kb_pipeline/processors/__init__.py +++ b/kb_pipeline/processors/__init__.py @@ -6,6 +6,7 @@ This module contains processors for various content transformation tasks: - DownloadProcessor: Metadata and content downloading - ASRProcessor: Audio-to-text conversion (subtitle download + Whisper.cpp) - AnalyzeProcessor: AI-powered content analysis using DeepAnalyzer +- ArchiveProcessor: Copy analysis reports to archive directory """ from .base import BaseProcessor @@ -13,5 +14,13 @@ from .parse import ParseProcessor from .download import DownloadProcessor from .asr import ASRProcessor from .analyze import AnalyzeProcessor +from .archive import ArchiveProcessor -__all__ = ["BaseProcessor", "ParseProcessor", "DownloadProcessor", "ASRProcessor", "AnalyzeProcessor"] +__all__ = [ + "BaseProcessor", + "ParseProcessor", + "DownloadProcessor", + "ASRProcessor", + "AnalyzeProcessor", + "ArchiveProcessor", +] diff --git a/kb_pipeline/processors/archive.py b/kb_pipeline/processors/archive.py new file mode 100644 index 0000000..eb38716 --- /dev/null +++ b/kb_pipeline/processors/archive.py @@ -0,0 +1,170 @@ +"""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 shutil +from datetime import datetime +from pathlib import Path +from typing import Optional + +from ..config import get_settings +from ..schemas import ArchiveInput, ArchiveOutput +from ..state import WorkflowStateManager +from .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: + # 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) + + # 文件命名: {YYYYMMDD}_{content_id}_analysis.md + timestamp = datetime.now().strftime("%Y%m%d") + archive_filename = f"{timestamp}_{input_data.content_id}_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: + should_overwrite = self._ask_overwrite(existing_file) + 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 _ask_overwrite(self, existing_file: Path) -> bool: + """询问用户是否覆盖已存在的文件 + + 使用 rich.prompt.Confirm 模式（与 Pipeline._ask_rerun_step 相同） + + Args: + existing_file: 已存在的文件路径 + + Returns: + True 如果用户选择覆盖 + """ + 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')