From a11a2dde99c8fb07dcc0e852e51e02c32a2cc85c Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Mon, 16 Mar 2026 23:03:14 +0800 Subject: [PATCH] docs(05): create phase 5 plans for archive and directory optimization --- .planning/ROADMAP.md | 18 +- .../phases/05-archive-directory/05-01-PLAN.md | 171 +++++ .../phases/05-archive-directory/05-02-PLAN.md | 504 +++++++++++++++ .../phases/05-archive-directory/05-03-PLAN.md | 593 ++++++++++++++++++ 4 files changed, 1277 insertions(+), 9 deletions(-) create mode 100644 .planning/phases/05-archive-directory/05-01-PLAN.md create mode 100644 .planning/phases/05-archive-directory/05-02-PLAN.md create mode 100644 .planning/phases/05-archive-directory/05-03-PLAN.md diff --git a/.planning/ROADMAP.md b/.planning/ROADMAP.md index 1d60aba..f199581 100644 --- a/.planning/ROADMAP.md +++ b/.planning/ROADMAP.md @@ -53,19 +53,19 @@ Plans: #### Phase 5: 归档与目录优化 **Goal**: 统一中间目录结构，添加自动归档步骤，优化最终目录布局 **Depends on**: Phase 4 -**Requirements**: INT-01, INT-02, INT-03, ARC-01, ARC-02, ARC-03, ARC-04, OUT-01, OUT-02, OUT-03, OUT-04 +**Requirements**: INT-01, INT-02, ARC-01, ARC-02, ARC-03, ARC-04, OUT-01, OUT-02, OUT-03, OUT-04 **Success Criteria** (what must be TRUE): 1. 所有分析任务的中间文件统一存放在 `~/work/intermediate/analysis/{platform}_{content_id}/` 2. 分析完成后 md 文件自动复制到按平台分组的最终目录 - 3. 最终目录结构为 `~/kb/{platform}/{timestamp}_{content_id}_analysis.md` - 4. 用户可通过配置文件自定义中间目录路径 - 5. 归档后可选择性清理中间文件 -**Plans**: TBD + 3. 最终目录结构为 `{KB_ARCHIVE_PATH}/{platform}/{YYYYMMDD}_{content_id}_analysis.md` + 4. 归档时检测重复文件，询问用户是否覆盖 + 5. 不自动清理中间文件（用户手动维护） +**Plans**: 3 plans Plans: -- [ ] 05-01: 中间目录统一和配置 -- [ ] 05-02: ArchiveProcessor 归档处理器 -- [ ] 05-03: 最终目录结构优化 +- [ ] 05-01-PLAN.md — 中间目录统一 (INT-01, INT-02) +- [ ] 05-02-PLAN.md — ArchiveProcessor 归档处理器 (ARC-01, ARC-02, ARC-03, ARC-04) +- [ ] 05-03-PLAN.md — 最终目录结构验证和测试 (OUT-01, OUT-02, OUT-03, OUT-04) #### Phase 6: 代码整合 **Goal**: 将独立的元数据技能和 Pipeline 模块整合进 kb-workflow @@ -93,5 +93,5 @@ Phases execute in numeric order: 1 → 2 → 3 → 4 → 5 → 6 | 2. Adapters | v1.0 | 2/2 | Complete | 2026-03-15 | | 3. Processors | v1.0 | 3/3 | Complete | 2026-03-15 | | 4. Pipeline | v1.0 | 2/2 | Complete | 2026-03-15 | -| 5. 归档与目录优化 | v1.1 | 0/3 | Not started | - | +| 5. 归档与目录优化 | v1.1 | 0/3 | Ready to execute | - | | 6. 代码整合 | v1.1 | 0/2 | Not started | - | diff --git a/.planning/phases/05-archive-directory/05-01-PLAN.md b/.planning/phases/05-archive-directory/05-01-PLAN.md new file mode 100644 index 0000000..9f6ed4c --- /dev/null +++ b/.planning/phases/05-archive-directory/05-01-PLAN.md @@ -0,0 +1,171 @@ +--- +phase: 05-archive-directory +plan: 01 +type: execute +wave: 1 +depends_on: [] +files_modified: + - kb_pipeline/pipeline.py + - kb-workflow/kb_workflow.py +autonomous: true +requirements: + - INT-01 + - INT-02 + +must_haves: + truths: + - "Pipeline uses ~/work/intermediate/analysis/ as intermediate directory" + - "Content directory follows {platform}_{content_id}/ naming convention" + - "CLI defaults to intermediate directory when no output_dir specified" + artifacts: + - path: "kb_pipeline/pipeline.py" + provides: "Intermediate directory path logic" + contains: "intermediate" + - path: "kb-workflow/kb_workflow.py" + provides: "CLI default output directory" + contains: "intermediate" + key_links: + - from: "kb-workflow/kb_workflow.py" + to: "kb_pipeline/pipeline.py" + via: "output_dir parameter" + pattern: "output_dir.*intermediate" +--- + + +统一中间目录结构到 `~/work/intermediate/analysis/` + +Purpose: 为所有分析任务提供一致的中间文件存储位置，便于管理和清理 +Output: 修改 Pipeline 和 CLI 使用统一的中间目录路径 + + + +@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md +@/home/starmen1265/.claude/get-shit-done/templates/summary.md + + + +@.planning/PROJECT.md +@.planning/ROADMAP.md +@.planning/STATE.md +@.planning/phases/05-archive-directory/05-CONTEXT.md + + + + +From kb_pipeline/pipeline.py: +```python +class Pipeline: + def __init__( + self, + processors: List[BaseProcessor], + config: Optional[PipelineConfig] = None, + output_dir: Optional[Path] = None, + ): + self.output_dir = output_dir or Path.cwd() + + def _get_content_dir(self, parsed: ParsedLink) -> Path: + content_dir = self.output_dir / f"{parsed.platform}_{parsed.content_id}" + content_dir.mkdir(parents=True, exist_ok=True) + return content_dir +``` + +From kb-workflow/kb_workflow.py: +```python +def process_command( + url: str, + output_dir: Optional[str] = None, + skip_completed: bool = True +) -> Dict[str, Any]: + out_dir = Path(output_dir) if output_dir else Path.cwd() +``` + + + + + + + Task 1: Add intermediate directory constant and update Pipeline + kb_pipeline/pipeline.py + + 1. Add module-level constant for intermediate directory: + ```python + INTERMEDIATE_DIR = Path.home() / "work" / "intermediate" / "analysis" + ``` + + 2. Modify `Pipeline.__init__` to use intermediate directory as default: + - Change `self.output_dir = output_dir or Path.cwd()` to use `INTERMEDIATE_DIR` + - The output_dir parameter still allows override for testing + + 3. Update `_get_content_dir` to ensure it follows `{platform}_{content_id}/` naming: + - Current implementation already does this: `f"{parsed.platform}_{parsed.content_id}"` + - Verify no changes needed + + Key points: + - Hard-code path as `~/work/intermediate/analysis/` (per user decision) + - Do NOT add KB_INTERMEDIATE_PATH config option (user explicitly decided against this) + - Content directory naming stays as `{platform}_{content_id}/` + + + python3 -c "from kb_pipeline.pipeline import INTERMEDIATE_DIR; assert str(INTERMEDIATE_DIR) == str(Path.home() / 'work' / 'intermediate' / 'analysis')" + + + - INTERMEDIATE_DIR constant defined in pipeline.py + - Pipeline defaults to intermediate directory + - Content directory naming unchanged + + + + + Task 2: Update CLI to use intermediate directory default + kb-workflow/kb_workflow.py + + 1. Import INTERMEDIATE_DIR from kb_pipeline.pipeline: + ```python + from kb_pipeline.pipeline import Pipeline, INTERMEDIATE_DIR + ``` + + 2. Modify `process_command` to use intermediate directory as default: + - Change: `out_dir = Path(output_dir) if output_dir else Path.cwd()` + - To: `out_dir = Path(output_dir) if output_dir else INTERMEDIATE_DIR` + + 3. Update docstring to document the default output directory behavior + + Key points: + - When no -o/--output-dir specified, content goes to ~/work/intermediate/analysis/ + - User can still override with -o flag for testing or special cases + + + python3 -c "from kb_workflow import process_command; import inspect; src = inspect.getsource(process_command); assert 'INTERMEDIATE_DIR' in src" + + + - CLI imports INTERMEDIATE_DIR + - Default output directory is ~/work/intermediate/analysis/ + - Help text updated to reflect default + + + + + + +1. Run existing tests to ensure no regression: + ```bash + python3 -m pytest tests/ -v --tb=short + ``` + +2. Verify intermediate directory path is correct: + ```bash + python3 -c "from kb_pipeline.pipeline import INTERMEDIATE_DIR; print(INTERMEDIATE_DIR)" + ``` + Expected: /home/{user}/work/intermediate/analysis + + + +- INT-01: All analysis tasks use `~/work/intermediate/analysis/` as intermediate directory +- INT-02: Content directory naming is `{platform}_{content_id}/` +- No KB_INTERMEDIATE_PATH config option added (per user decision) +- Existing tests pass + + + +After completion, create `.planning/phases/05-archive-directory/05-01-SUMMARY.md` + diff --git a/.planning/phases/05-archive-directory/05-02-PLAN.md b/.planning/phases/05-archive-directory/05-02-PLAN.md new file mode 100644 index 0000000..64a1910 --- /dev/null +++ b/.planning/phases/05-archive-directory/05-02-PLAN.md @@ -0,0 +1,504 @@ +--- +phase: 05-archive-directory +plan: 02 +type: execute +wave: 2 +depends_on: + - 05-01 +files_modified: + - kb_pipeline/schemas.py + - kb_pipeline/processors/__init__.py + - kb_pipeline/processors/archive.py + - kb_pipeline/pipeline.py + - kb-workflow/kb_workflow.py +autonomous: true +requirements: + - ARC-01 + - ARC-02 + - ARC-03 + - ARC-04 + +must_haves: + truths: + - "ArchiveProcessor is the last step in pipeline" + - "After analysis, md file is copied to archive directory" + - "Only md files are archived, not json files" + - "Archive step status is recorded in workflow_state.json" + artifacts: + - path: "kb_pipeline/processors/archive.py" + provides: "ArchiveProcessor implementation" + contains: "class ArchiveProcessor" + - path: "kb_pipeline/schemas.py" + provides: "ArchiveInput and ArchiveOutput schemas" + contains: "class ArchiveInput" + - path: "kb_pipeline/pipeline.py" + provides: "Pipeline orchestration with archive step" + contains: "archive" + key_links: + - from: "kb_pipeline/pipeline.py" + to: "kb_pipeline/processors/archive.py" + via: "ArchiveProcessor instantiation" + pattern: "ArchiveProcessor" + - from: "ArchiveProcessor" + to: "KBSettings.archive_path" + via: "get_settings()" + pattern: "settings.archive_path" +--- + + +创建 ArchiveProcessor 作为管道最后一步，自动归档分析报告到最终目录 + +Purpose: 分析完成后自动将 md 文件复制到按平台分组的归档目录 +Output: ArchiveProcessor 类、ArchiveInput/ArchiveOutput schemas、Pipeline 集成 + + + +@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md +@/home/starmen1265/.claude/get-shit-done/templates/summary.md + + + +@.planning/PROJECT.md +@.planning/ROADMAP.md +@.planning/STATE.md +@.planning/phases/05-archive-directory/05-CONTEXT.md + + + + +From kb_pipeline/processors/base.py: +```python +class BaseProcessor(ABC): + STEP_NAME: str + + def __init__(self, state_manager: Optional[WorkflowStateManager] = None): + self.settings = get_settings() + self.state_manager = state_manager + + @abstractmethod + async def process(self, input_data: StepInput) -> StepOutput: + pass +``` + +From kb_pipeline/config.py: +```python +class KBSettings(BaseSettings): + archive_path: str = Field(..., description="归档目录") +``` + +From kb_pipeline/schemas.py: +```python +class AnalyzeOutput(StepOutput): + analysis_file: str = Field(..., description="Path to generated analysis report") + # ... other fields +``` + +From kb_pipeline/pipeline.py: +```python +class Pipeline: + def __init__(self, processors: List[BaseProcessor], config: Optional[PipelineConfig] = None, output_dir: Optional[Path] = None): + self.processors: Dict[str, BaseProcessor] = {p.STEP_NAME: p for p in processors} + + async def run(self, url: str) -> PipelineOutput: + for step_name in self.config.steps: + # ... step execution +``` + +From rich.prompt (existing pattern): +```python +try: + from rich.console import Console + from rich.prompt import Confirm + return Confirm.ask("Question?", default=False) +except ImportError: + response = input("Question? [y/N] ") + return response.lower() in ('y', 'yes') +``` + + + + + + + Task 1: Create ArchiveInput and ArchiveOutput schemas + kb_pipeline/schemas.py + + Add new schema classes at the end of the schemas file: + + ```python + # ============================================================================= + # Archive Schemas (Phase 5) + # ============================================================================= + + class ArchiveInput(StepInput): + """Archive 步骤输入 + + 继承 StepInput 的基础字段，添加归档特定字段。 + + Attributes: + content_dir: 内容目录路径（中间目录） + analysis_file: 分析报告 md 文件路径 + content_type: 内容类型 + """ + content_dir: str = Field(..., min_length=1, description="Content directory path (intermediate)") + analysis_file: str = Field(..., min_length=1, description="Path to analysis Markdown file") + content_type: str = Field(..., min_length=1, description="Content type (video, article, etc.)") + + + class ArchiveOutput(StepOutput): + """Archive 步骤输出 + + 继承 StepOutput 的基础字段，添加归档特定字段。 + + Attributes: + archive_file: 归档后的文件路径 + skipped: 是否跳过（已有相同文件） + platform: 平台名称 + """ + archive_file: str = Field(default="", description="Path to archived file") + skipped: bool = Field(default=False, description="Whether archiving was skipped") + platform: str = Field(default="", description="Platform name") + ``` + + Follow existing schema patterns: + - Inherit from StepInput/StepOutput + - Use Field() with descriptions + - Include min_length validation for required strings + + + python3 -c "from kb_pipeline.schemas import ArchiveInput, ArchiveOutput; print('OK')" + + + - ArchiveInput class defined with content_dir, analysis_file, content_type fields + - ArchiveOutput class defined with archive_file, skipped, platform fields + - Both inherit from base StepInput/StepOutput + + + + + Task 2: Implement ArchiveProcessor + kb_pipeline/processors/archive.py, kb_pipeline/processors/__init__.py + + Create new file `kb_pipeline/processors/archive.py`: + + ```python + """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') + ``` + + Update `kb_pipeline/processors/__init__.py` to export ArchiveProcessor: + ```python + from .archive import ArchiveProcessor + ``` + + Key points: + - Follow BaseProcessor pattern exactly + - Use settings.archive_path for archive base directory + - Only copy md files (not json) + - Glob pattern for duplicate detection + - Use rich.prompt.Confirm with input() fallback + + + python3 -c "from kb_pipeline.processors import ArchiveProcessor; p = ArchiveProcessor(); assert p.STEP_NAME == 'archive'" + + + - ArchiveProcessor class created with STEP_NAME = "archive" + - ArchiveInput/ArchiveOutput schemas imported + - _find_existing_archive uses glob pattern + - _ask_overwrite uses rich.prompt with fallback + - Export added to __init__.py + + + + + Task 3: Integrate ArchiveProcessor into Pipeline + kb_pipeline/pipeline.py, kb-workflow/kb_workflow.py + + 1. Update `kb_pipeline/pipeline.py`: + + a. Update default steps in PipelineConfig to include "archive": + - Find where PipelineConfig is referenced + - The default steps are defined in PipelineConfig.steps in schemas.py + - We need to update schemas.py default_factory + + b. Add archive step input creation in Pipeline.run(): + ```python + elif step_name == "archive": + # Archive needs analysis file from analyze step + analyze_result = results.get("analyze", {}) + analysis_file = analyze_result.get("analysis_file", "") + + step_input = ArchiveInput( + content_id=parsed.content_id, + platform=parsed.platform, + content_dir=str(content_dir), + analysis_file=analysis_file, + content_type=parsed.content_type, + ) + ``` + + c. Import ArchiveInput at top of file: + ```python + from kb_pipeline.schemas import ( + ParseInput, DownloadInput, ASRInput, AnalyzeInput, ArchiveInput + ) + ``` + + 2. Update `kb_pipeline/schemas.py` PipelineConfig default steps: + ```python + steps: List[str] = Field( + default_factory=lambda: ["parse", "download", "asr", "analyze", "archive"], + description="Steps to execute in order" + ) + ``` + + 3. Update `kb-workflow/kb_workflow.py`: + + a. Import ArchiveProcessor: + ```python + from kb_pipeline.processors import ( + ASRProcessor, AnalyzeProcessor, ArchiveProcessor, + DownloadProcessor, ParseProcessor + ) + ``` + + b. Add ArchiveProcessor to processors list: + ```python + processors = [ + ParseProcessor(), + DownloadProcessor(), + ASRProcessor(), + AnalyzeProcessor(), + ArchiveProcessor(), # Add this + ] + ``` + + c. Update config steps to include "archive": + ```python + config = PipelineConfig( + steps=["parse", "download", "asr", "analyze", "archive"], + # ... + ) + ``` + + Key points: + - Archive is always the last step + - Archive step needs analysis_file from analyze step + - Pipeline must create ArchiveInput properly + + + python3 -c "from kb_pipeline.schemas import PipelineConfig; c = PipelineConfig(); assert 'archive' in c.steps" + + + - ArchiveProcessor added to processors list in CLI + - PipelineConfig default steps include "archive" + - Pipeline.run() handles ArchiveInput creation + - ArchiveInput imported in pipeline.py + + + + + + +1. Verify ArchiveProcessor can be imported: + ```bash + python3 -c "from kb_pipeline.processors import ArchiveProcessor; print('OK')" + ``` + +2. Verify schemas: + ```bash + python3 -c "from kb_pipeline.schemas import ArchiveInput, ArchiveOutput; print('OK')" + ``` + +3. Verify PipelineConfig includes archive step: + ```bash + python3 -c "from kb_pipeline.schemas import PipelineConfig; print(PipelineConfig().steps)" + ``` + Expected: ['parse', 'download', 'asr', 'analyze', 'archive'] + +4. Run existing tests: + ```bash + python3 -m pytest tests/ -v --tb=short -k "not integration" + ``` + + + +- ARC-01: ArchiveProcessor is the last step in pipeline +- ARC-02: Analysis completion triggers automatic archiving +- ARC-03: Only md files are copied (not json) +- ARC-04: No automatic cleanup of intermediate files (user handles manually) +- Archive step status recorded in workflow_state.json +- Duplicate detection asks user for overwrite confirmation + + + +After completion, create `.planning/phases/05-archive-directory/05-02-SUMMARY.md` + diff --git a/.planning/phases/05-archive-directory/05-03-PLAN.md b/.planning/phases/05-archive-directory/05-03-PLAN.md new file mode 100644 index 0000000..596f713 --- /dev/null +++ b/.planning/phases/05-archive-directory/05-03-PLAN.md @@ -0,0 +1,593 @@ +--- +phase: 05-archive-directory +plan: 03 +type: execute +wave: 3 +depends_on: + - 05-02 +files_modified: + - tests/test_pipeline.py + - tests/test_processors.py +autonomous: true +requirements: + - OUT-01 + - OUT-02 + - OUT-03 + - OUT-04 + +must_haves: + truths: + - "Archive files are organized by platform (bilibili/, zhihu/)" + - "Archive files have timestamp prefix: {YYYYMMDD}_{content_id}_analysis.md" + - "No date-based subdirectories in archive" + - "Only md files in archive, no json files" + artifacts: + - path: "tests/test_pipeline.py" + provides: "Integration test for archive step" + contains: "test_archive" + - path: "tests/test_processors.py" + provides: "Unit tests for ArchiveProcessor" + contains: "ArchiveProcessor" + key_links: + - from: "ArchiveProcessor" + to: "{KB_ARCHIVE_PATH}/{platform}/" + via: "settings.archive_path / platform" + pattern: "platform_dir" +--- + + +验证最终目录结构和端到端归档流程 + +Purpose: 通过测试验证归档目录结构符合需求，确保只归档 md 文件 +Output: ArchiveProcessor 单元测试和 Pipeline 集成测试 + + + +@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md +@/home/starmen1265/.claude/get-shit-done/templates/summary.md + + + +@.planning/PROJECT.md +@.planning/ROADMAP.md +@.planning/STATE.md +@.planning/phases/05-archive-directory/05-CONTEXT.md + + + + +From tests/test_processors.py (existing pattern): +```python +import pytest +from unittest.mock import AsyncMock, patch, MagicMock + +@pytest.mark.asyncio +async def test_processor_success(): + # ... test implementation +``` + +From kb_pipeline/processors/archive.py (created in 05-02): +```python +class ArchiveProcessor(BaseProcessor): + STEP_NAME = "archive" + + async def process(self, input_data: ArchiveInput) -> ArchiveOutput: + # ... + + def _find_existing_archive(self, platform_dir: Path, content_id: str) -> Optional[Path]: + # ... + + def _ask_overwrite(self, existing_file: Path) -> bool: + # ... +``` + +Expected archive structure: +``` +{KB_ARCHIVE_PATH}/ +├── bilibili/ +│ ├── 20260316_BV1xxx_analysis.md +│ └── 20260316_BV1yyy_analysis.md +└── zhihu/ + └── 20260316_123456_analysis.md +``` + + + + + + + Task 1: Add ArchiveProcessor unit tests + tests/test_processors.py + + - Test 1: ArchiveProcessor copies md file to correct platform directory + - Test 2: ArchiveProcessor uses correct filename format {YYYYMMDD}_{content_id}_analysis.md + - Test 3: ArchiveProcessor skips when step already completed + - Test 4: ArchiveProcessor detects existing archive by content_id + - Test 5: ArchiveProcessor handles missing analysis file with error + + + Add the following tests to `tests/test_processors.py`: + + ```python + # ============================================================================= + # ArchiveProcessor Tests + # ============================================================================= + + import tempfile + import shutil + from datetime import datetime + from pathlib import Path + from unittest.mock import patch, MagicMock + + from kb_pipeline.processors.archive import ArchiveProcessor + from kb_pipeline.schemas import ArchiveInput, ArchiveOutput + from kb_pipeline.state import WorkflowStateManager + + + @pytest.fixture + def archive_test_env(tmp_path): + """Create test environment for archive tests""" + # Create intermediate content directory + content_dir = tmp_path / "intermediate" / "bilibili_BV1test" + content_dir.mkdir(parents=True) + + # Create analysis file + analysis_file = content_dir / "analysis.md" + analysis_file.write_text("# Test Analysis\n\nContent here.") + + # Create archive directory + archive_dir = tmp_path / "archive" + archive_dir.mkdir(parents=True) + + # Create state manager + state_manager = WorkflowStateManager(content_dir) + + return { + "content_dir": content_dir, + "analysis_file": analysis_file, + "archive_dir": archive_dir, + "state_manager": state_manager, + } + + + @pytest.mark.asyncio + async def test_archive_processor_copies_to_platform_dir(archive_test_env, monkeypatch): + """Test that archive file is copied to correct platform directory""" + # Mock settings + mock_settings = MagicMock() + mock_settings.archive_path = str(archive_test_env["archive_dir"]) + + with patch.object(ArchiveProcessor, '__init__', lambda self, sm: None): + processor = ArchiveProcessor() + processor.settings = mock_settings + processor.state_manager = None + + input_data = ArchiveInput( + content_id="BV1test", + platform="bilibili", + content_dir=str(archive_test_env["content_dir"]), + analysis_file=str(archive_test_env["analysis_file"]), + content_type="video" + ) + + result = await processor.process(input_data) + + assert result.success is True + assert result.skipped is False + assert "bilibili" in result.archive_file + assert result.archive_file.endswith("_analysis.md") + + + @pytest.mark.asyncio + async def test_archive_processor_filename_format(archive_test_env, monkeypatch): + """Test that archive filename follows {YYYYMMDD}_{content_id}_analysis.md format""" + mock_settings = MagicMock() + mock_settings.archive_path = str(archive_test_env["archive_dir"]) + + with patch.object(ArchiveProcessor, '__init__', lambda self, sm: None): + processor = ArchiveProcessor() + processor.settings = mock_settings + processor.state_manager = None + + input_data = ArchiveInput( + content_id="BV1test", + platform="bilibili", + content_dir=str(archive_test_env["content_dir"]), + analysis_file=str(archive_test_env["analysis_file"]), + content_type="video" + ) + + result = await processor.process(input_data) + + # Check filename format + archive_path = Path(result.archive_file) + filename = archive_path.name + today = datetime.now().strftime("%Y%m%d") + expected = f"{today}_BV1test_analysis.md" + assert filename == expected + + + @pytest.mark.asyncio + async def test_archive_processor_skips_completed(archive_test_env, monkeypatch): + """Test that archive step is skipped when already completed""" + mock_settings = MagicMock() + mock_settings.archive_path = str(archive_test_env["archive_dir"]) + + # Create state with archive already completed + state = archive_test_env["state_manager"].get_or_create( + content_id="BV1test", + platform="bilibili", + source_url="https://test.com" + ) + state.mark_step_completed("archive") + archive_test_env["state_manager"].save(state) + + with patch.object(ArchiveProcessor, '__init__', lambda self, sm: None): + processor = ArchiveProcessor() + processor.settings = mock_settings + processor.state_manager = archive_test_env["state_manager"] + + input_data = ArchiveInput( + content_id="BV1test", + platform="bilibili", + content_dir=str(archive_test_env["content_dir"]), + analysis_file=str(archive_test_env["analysis_file"]), + content_type="video" + ) + + result = await processor.process(input_data) + + assert result.success is True + assert result.skipped is True + + + @pytest.mark.asyncio + async def test_archive_processor_detects_existing(archive_test_env, monkeypatch): + """Test that archive processor detects existing archive by content_id""" + mock_settings = MagicMock() + mock_settings.archive_path = str(archive_test_env["archive_dir"]) + + # Create existing archive with different timestamp + platform_dir = archive_test_env["archive_dir"] / "bilibili" + platform_dir.mkdir(parents=True, exist_ok=True) + existing = platform_dir / "20260101_BV1test_analysis.md" + existing.write_text("Old content") + + with patch.object(ArchiveProcessor, '__init__', lambda self, sm: None): + processor = ArchiveProcessor() + processor.settings = mock_settings + processor.state_manager = None + processor._ask_overwrite = lambda f: False # User says no + + input_data = ArchiveInput( + content_id="BV1test", + platform="bilibili", + content_dir=str(archive_test_env["content_dir"]), + analysis_file=str(archive_test_env["analysis_file"]), + content_type="video" + ) + + result = await processor.process(input_data) + + assert result.success is True + assert result.skipped is True + + + @pytest.mark.asyncio + async def test_archive_processor_handles_missing_file(archive_test_env, monkeypatch): + """Test that archive processor handles missing analysis file""" + mock_settings = MagicMock() + mock_settings.archive_path = str(archive_test_env["archive_dir"]) + + with patch.object(ArchiveProcessor, '__init__', lambda self, sm: None): + processor = ArchiveProcessor() + processor.settings = mock_settings + processor.state_manager = None + + input_data = ArchiveInput( + content_id="BV1test", + platform="bilibili", + content_dir=str(archive_test_env["content_dir"]), + analysis_file="/nonexistent/analysis.md", # Missing file + content_type="video" + ) + + result = await processor.process(input_data) + + assert result.success is False + assert "not found" in result.error_message.lower() + ``` + + Run tests to verify: + ```bash + python3 -m pytest tests/test_processors.py -v -k "archive" + ``` + + + python3 -m pytest tests/test_processors.py -v -k "archive" --tb=short + + + - 5 unit tests for ArchiveProcessor pass + - Tests cover: copy, filename format, skip completed, detect existing, missing file + - All tests use tmp_path for isolation + + + + + Task 2: Add Pipeline integration test for archive step + tests/test_pipeline.py + + - Test 1: Full pipeline with archive step copies file to archive directory + - Test 2: Archive directory structure is {archive_path}/{platform}/ + - Test 3: Archive filename format is {YYYYMMDD}_{content_id}_analysis.md + + + Add the following tests to `tests/test_pipeline.py`: + + ```python + # ============================================================================= + # Archive Integration Tests + # ============================================================================= + + import tempfile + from datetime import datetime + from pathlib import Path + from unittest.mock import AsyncMock, patch, MagicMock + + from kb_pipeline.pipeline import Pipeline, INTERMEDIATE_DIR + from kb_pipeline.schemas import PipelineConfig + from kb_pipeline.processors import ( + ParseProcessor, DownloadProcessor, ASRProcessor, + AnalyzeProcessor, ArchiveProcessor + ) + + + @pytest.fixture + def mock_processors_with_archive(): + """Create mock processors including archive""" + processors = [] + + for step_name in ["parse", "download", "asr", "analyze", "archive"]: + processor = MagicMock() + processor.STEP_NAME = step_name + processor.process = AsyncMock() + + if step_name == "parse": + processor.process.return_value = MagicMock( + success=True, + platform="bilibili", + content_type="video", + content_id="BV1test", + resolved_url="https://www.bilibili.com/video/BV1test", + original_url="https://www.bilibili.com/video/BV1test" + ) + elif step_name == "download": + processor.process.return_value = MagicMock( + success=True, + metadata_file="/tmp/metadata.json", + content_file="/tmp/content.json", + title="Test Video", + author="Test Author" + ) + elif step_name == "asr": + processor.process.return_value = MagicMock( + success=True, + tool="whisper", + has_subtitle=False, + output_files={"json": "/tmp/subtitle.json"} + ) + elif step_name == "analyze": + processor.process.return_value = MagicMock( + success=True, + analysis_file="/tmp/analysis.md", + summary="Test summary", + main_topics=["topic1"], + key_insights=["insight1"], + tags=["tag1"], + stats={} + ) + elif step_name == "archive": + processor.process.return_value = MagicMock( + success=True, + skipped=False, + platform="bilibili", + archive_file="/archive/bilibili/20260316_BV1test_analysis.md" + ) + + processors.append(processor) + + return processors + + + @pytest.mark.asyncio + async def test_pipeline_includes_archive_step(mock_processors_with_archive, tmp_path): + """Test that pipeline includes archive as final step""" + config = PipelineConfig( + steps=["parse", "download", "asr", "analyze", "archive"], + output_dir=tmp_path, + skip_completed=False + ) + + pipeline = Pipeline( + processors=mock_processors_with_archive, + config=config, + output_dir=tmp_path + ) + + # Mock _parse_url and _get_content_dir + with patch.object(pipeline, '_parse_url') as mock_parse, \ + patch.object(pipeline, '_get_content_dir') as mock_dir: + + from kb_pipeline.link_parser import ParsedLink + mock_parse.return_value = ParsedLink( + platform="bilibili", + content_type="video", + content_id="BV1test", + original_url="https://www.bilibili.com/video/BV1test", + resolved_url="https://www.bilibili.com/video/BV1test" + ) + mock_dir.return_value = tmp_path / "bilibili_BV1test" + + result = await pipeline.run("https://www.bilibili.com/video/BV1test") + + assert result.success is True + # Verify archive processor was called + archive_processor = next(p for p in mock_processors_with_archive if p.STEP_NAME == "archive") + archive_processor.process.assert_called_once() + + + @pytest.mark.asyncio + async def test_archive_uses_platform_directory(tmp_path): + """Test that archive files go to {archive_path}/{platform}/ directory""" + from kb_pipeline.processors.archive import ArchiveProcessor + from kb_pipeline.schemas import ArchiveInput + + # Setup + archive_dir = tmp_path / "archive" + content_dir = tmp_path / "intermediate" / "bilibili_BV1test" + content_dir.mkdir(parents=True) + analysis_file = content_dir / "analysis.md" + analysis_file.write_text("# Test") + + # Mock settings + with patch('kb_pipeline.processors.archive.get_settings') as mock_get_settings: + mock_settings = MagicMock() + mock_settings.archive_path = str(archive_dir) + mock_get_settings.return_value = mock_settings + + processor = ArchiveProcessor() + input_data = ArchiveInput( + content_id="BV1test", + platform="bilibili", + content_dir=str(content_dir), + analysis_file=str(analysis_file), + content_type="video" + ) + + result = await processor.process(input_data) + + assert result.success is True + # Verify platform subdirectory + archive_path = Path(result.archive_file) + assert archive_path.parent.name == "bilibili" + + + @pytest.mark.asyncio + async def test_archive_no_date_subdirectories(tmp_path): + """Test that archive does not create date-based subdirectories""" + from kb_pipeline.processors.archive import ArchiveProcessor + from kb_pipeline.schemas import ArchiveInput + + archive_dir = tmp_path / "archive" + content_dir = tmp_path / "intermediate" / "bilibili_BV1test" + content_dir.mkdir(parents=True) + analysis_file = content_dir / "analysis.md" + analysis_file.write_text("# Test") + + with patch('kb_pipeline.processors.archive.get_settings') as mock_get_settings: + mock_settings = MagicMock() + mock_settings.archive_path = str(archive_dir) + mock_get_settings.return_value = mock_settings + + processor = ArchiveProcessor() + input_data = ArchiveInput( + content_id="BV1test", + platform="bilibili", + content_dir=str(content_dir), + analysis_file=str(analysis_file), + content_type="video" + ) + + result = await processor.process(input_data) + + # Verify no date-based subdirectories + archive_path = Path(result.archive_file) + # Parent should be platform dir, grandparent should be archive base + assert archive_path.parent.parent == archive_dir + # Check that parent is just "bilibili", not a date + assert archive_path.parent.name == "bilibili" + ``` + + Run tests: + ```bash + python3 -m pytest tests/test_pipeline.py -v -k "archive" --tb=short + ``` + + + python3 -m pytest tests/test_pipeline.py -v -k "archive" --tb=short + + + - 3 integration tests for archive step pass + - Tests verify: archive included in pipeline, platform directory, no date subdirs + - All tests use tmp_path for isolation + + + + + Task 3: Run full test suite and verify + tests/test_pipeline.py, tests/test_processors.py + + Run the complete test suite to verify all changes work together: + + ```bash + # Run all tests + python3 -m pytest tests/ -v --tb=short + + # Specifically verify archive-related tests + python3 -m pytest tests/ -v -k "archive" --tb=short + ``` + + Verify the following: + 1. All existing tests still pass (no regression) + 2. New archive tests pass + 3. ArchiveProcessor can be imported and instantiated + 4. PipelineConfig default steps include "archive" + + If any tests fail, fix the issues before marking complete. + + + python3 -m pytest tests/ -v --tb=short -x + + + - All tests pass (existing + new) + - No regressions introduced + - Archive functionality fully tested + + + + + + +1. Run full test suite: + ```bash + python3 -m pytest tests/ -v --tb=short + ``` + +2. Verify archive-specific tests: + ```bash + python3 -m pytest tests/ -v -k "archive" --tb=short + ``` + +3. Manual verification of archive structure: + ```bash + python3 -c " + from pathlib import Path + from kb_pipeline.processors.archive import ArchiveProcessor + # Archive structure should be: {archive_path}/{platform}/{timestamp}_{id}_analysis.md + print('ArchiveProcessor loaded successfully') + " + ``` + + + +- OUT-01: Archive files organized by platform (bilibili/, zhihu/) +- OUT-02: No date-based subdirectories in archive +- OUT-03: Filename format is {YYYYMMDD}_{content_id}_analysis.md +- OUT-04: Only md files archived (json files stay in intermediate) +- All tests pass +- No regressions in existing functionality + + + +After completion, create `.planning/phases/05-archive-directory/05-03-SUMMARY.md` +