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
+
+
+
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
+
+
+
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
+
+
+