docs(05): create phase 5 plans for archive and directory optimization
This commit is contained in:
@@ -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 | - |
|
||||
|
||||
@@ -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"
|
||||
---
|
||||
|
||||
<objective>
|
||||
统一中间目录结构到 `~/work/intermediate/analysis/`
|
||||
|
||||
Purpose: 为所有分析任务提供一致的中间文件存储位置，便于管理和清理
|
||||
Output: 修改 Pipeline 和 CLI 使用统一的中间目录路径
|
||||
</objective>
|
||||
|
||||
<execution_context>
|
||||
@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md
|
||||
@/home/starmen1265/.claude/get-shit-done/templates/summary.md
|
||||
</execution_context>
|
||||
|
||||
<context>
|
||||
@.planning/PROJECT.md
|
||||
@.planning/ROADMAP.md
|
||||
@.planning/STATE.md
|
||||
@.planning/phases/05-archive-directory/05-CONTEXT.md
|
||||
|
||||
<interfaces>
|
||||
<!-- Key interfaces from existing codebase -->
|
||||
|
||||
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()
|
||||
```
|
||||
</interfaces>
|
||||
</context>
|
||||
|
||||
<tasks>
|
||||
|
||||
<task type="auto">
|
||||
<name>Task 1: Add intermediate directory constant and update Pipeline</name>
|
||||
<files>kb_pipeline/pipeline.py</files>
|
||||
<action>
|
||||
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}/`
|
||||
</action>
|
||||
<verify>
|
||||
<automated>python3 -c "from kb_pipeline.pipeline import INTERMEDIATE_DIR; assert str(INTERMEDIATE_DIR) == str(Path.home() / 'work' / 'intermediate' / 'analysis')"</automated>
|
||||
</verify>
|
||||
<done>
|
||||
- INTERMEDIATE_DIR constant defined in pipeline.py
|
||||
- Pipeline defaults to intermediate directory
|
||||
- Content directory naming unchanged
|
||||
</done>
|
||||
</task>
|
||||
|
||||
<task type="auto">
|
||||
<name>Task 2: Update CLI to use intermediate directory default</name>
|
||||
<files>kb-workflow/kb_workflow.py</files>
|
||||
<action>
|
||||
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
|
||||
</action>
|
||||
<verify>
|
||||
<automated>python3 -c "from kb_workflow import process_command; import inspect; src = inspect.getsource(process_command); assert 'INTERMEDIATE_DIR' in src"</automated>
|
||||
</verify>
|
||||
<done>
|
||||
- CLI imports INTERMEDIATE_DIR
|
||||
- Default output directory is ~/work/intermediate/analysis/
|
||||
- Help text updated to reflect default
|
||||
</done>
|
||||
</task>
|
||||
|
||||
</tasks>
|
||||
|
||||
<verification>
|
||||
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
|
||||
</verification>
|
||||
|
||||
<success_criteria>
|
||||
- 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
|
||||
</success_criteria>
|
||||
|
||||
<output>
|
||||
After completion, create `.planning/phases/05-archive-directory/05-01-SUMMARY.md`
|
||||
</output>
|
||||
@@ -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"
|
||||
---
|
||||
|
||||
<objective>
|
||||
创建 ArchiveProcessor 作为管道最后一步，自动归档分析报告到最终目录
|
||||
|
||||
Purpose: 分析完成后自动将 md 文件复制到按平台分组的归档目录
|
||||
Output: ArchiveProcessor 类、ArchiveInput/ArchiveOutput schemas、Pipeline 集成
|
||||
</objective>
|
||||
|
||||
<execution_context>
|
||||
@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md
|
||||
@/home/starmen1265/.claude/get-shit-done/templates/summary.md
|
||||
</execution_context>
|
||||
|
||||
<context>
|
||||
@.planning/PROJECT.md
|
||||
@.planning/ROADMAP.md
|
||||
@.planning/STATE.md
|
||||
@.planning/phases/05-archive-directory/05-CONTEXT.md
|
||||
|
||||
<interfaces>
|
||||
<!-- Key interfaces from existing codebase -->
|
||||
|
||||
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')
|
||||
```
|
||||
</interfaces>
|
||||
</context>
|
||||
|
||||
<tasks>
|
||||
|
||||
<task type="auto">
|
||||
<name>Task 1: Create ArchiveInput and ArchiveOutput schemas</name>
|
||||
<files>kb_pipeline/schemas.py</files>
|
||||
<action>
|
||||
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
|
||||
</action>
|
||||
<verify>
|
||||
<automated>python3 -c "from kb_pipeline.schemas import ArchiveInput, ArchiveOutput; print('OK')"</automated>
|
||||
</verify>
|
||||
<done>
|
||||
- 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
|
||||
</done>
|
||||
</task>
|
||||
|
||||
<task type="auto">
|
||||
<name>Task 2: Implement ArchiveProcessor</name>
|
||||
<files>kb_pipeline/processors/archive.py, kb_pipeline/processors/__init__.py</files>
|
||||
<action>
|
||||
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
|
||||
</action>
|
||||
<verify>
|
||||
<automated>python3 -c "from kb_pipeline.processors import ArchiveProcessor; p = ArchiveProcessor(); assert p.STEP_NAME == 'archive'"</automated>
|
||||
</verify>
|
||||
<done>
|
||||
- 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
|
||||
</done>
|
||||
</task>
|
||||
|
||||
<task type="auto">
|
||||
<name>Task 3: Integrate ArchiveProcessor into Pipeline</name>
|
||||
<files>kb_pipeline/pipeline.py, kb-workflow/kb_workflow.py</files>
|
||||
<action>
|
||||
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
|
||||
</action>
|
||||
<verify>
|
||||
<automated>python3 -c "from kb_pipeline.schemas import PipelineConfig; c = PipelineConfig(); assert 'archive' in c.steps"</automated>
|
||||
</verify>
|
||||
<done>
|
||||
- ArchiveProcessor added to processors list in CLI
|
||||
- PipelineConfig default steps include "archive"
|
||||
- Pipeline.run() handles ArchiveInput creation
|
||||
- ArchiveInput imported in pipeline.py
|
||||
</done>
|
||||
</task>
|
||||
|
||||
</tasks>
|
||||
|
||||
<verification>
|
||||
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"
|
||||
```
|
||||
</verification>
|
||||
|
||||
<success_criteria>
|
||||
- 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
|
||||
</success_criteria>
|
||||
|
||||
<output>
|
||||
After completion, create `.planning/phases/05-archive-directory/05-02-SUMMARY.md`
|
||||
</output>
|
||||
@@ -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"
|
||||
---
|
||||
|
||||
<objective>
|
||||
验证最终目录结构和端到端归档流程
|
||||
|
||||
Purpose: 通过测试验证归档目录结构符合需求，确保只归档 md 文件
|
||||
Output: ArchiveProcessor 单元测试和 Pipeline 集成测试
|
||||
</objective>
|
||||
|
||||
<execution_context>
|
||||
@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md
|
||||
@/home/starmen1265/.claude/get-shit-done/templates/summary.md
|
||||
</execution_context>
|
||||
|
||||
<context>
|
||||
@.planning/PROJECT.md
|
||||
@.planning/ROADMAP.md
|
||||
@.planning/STATE.md
|
||||
@.planning/phases/05-archive-directory/05-CONTEXT.md
|
||||
|
||||
<interfaces>
|
||||
<!-- Key interfaces from existing codebase -->
|
||||
|
||||
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
|
||||
```
|
||||
</interfaces>
|
||||
</context>
|
||||
|
||||
<tasks>
|
||||
|
||||
<task type="auto" tdd="true">
|
||||
<name>Task 1: Add ArchiveProcessor unit tests</name>
|
||||
<files>tests/test_processors.py</files>
|
||||
<behavior>
|
||||
- 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
|
||||
</behavior>
|
||||
<action>
|
||||
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"
|
||||
```
|
||||
</action>
|
||||
<verify>
|
||||
<automated>python3 -m pytest tests/test_processors.py -v -k "archive" --tb=short</automated>
|
||||
</verify>
|
||||
<done>
|
||||
- 5 unit tests for ArchiveProcessor pass
|
||||
- Tests cover: copy, filename format, skip completed, detect existing, missing file
|
||||
- All tests use tmp_path for isolation
|
||||
</done>
|
||||
</task>
|
||||
|
||||
<task type="auto" tdd="true">
|
||||
<name>Task 2: Add Pipeline integration test for archive step</name>
|
||||
<files>tests/test_pipeline.py</files>
|
||||
<behavior>
|
||||
- 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
|
||||
</behavior>
|
||||
<action>
|
||||
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
|
||||
```
|
||||
</action>
|
||||
<verify>
|
||||
<automated>python3 -m pytest tests/test_pipeline.py -v -k "archive" --tb=short</automated>
|
||||
</verify>
|
||||
<done>
|
||||
- 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
|
||||
</done>
|
||||
</task>
|
||||
|
||||
<task type="auto">
|
||||
<name>Task 3: Run full test suite and verify</name>
|
||||
<files>tests/test_pipeline.py, tests/test_processors.py</files>
|
||||
<action>
|
||||
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.
|
||||
</action>
|
||||
<verify>
|
||||
<automated>python3 -m pytest tests/ -v --tb=short -x</automated>
|
||||
</verify>
|
||||
<done>
|
||||
- All tests pass (existing + new)
|
||||
- No regressions introduced
|
||||
- Archive functionality fully tested
|
||||
</done>
|
||||
</task>
|
||||
|
||||
</tasks>
|
||||
|
||||
<verification>
|
||||
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')
|
||||
"
|
||||
```
|
||||
</verification>
|
||||
|
||||
<success_criteria>
|
||||
- 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
|
||||
</success_criteria>
|
||||
|
||||
<output>
|
||||
After completion, create `.planning/phases/05-archive-directory/05-03-SUMMARY.md`
|
||||
</output>
|
||||
Reference in New Issue
Block a user