diff --git a/.coverage b/.coverage
new file mode 100644
index 0000000..a283e8f
Binary files /dev/null and b/.coverage differ
diff --git a/.planning/phases/04-analysis-orchestration/04-01-PLAN.md b/.planning/phases/04-analysis-orchestration/04-01-PLAN.md
new file mode 100644
index 0000000..9f4cb96
--- /dev/null
+++ b/.planning/phases/04-analysis-orchestration/04-01-PLAN.md
@@ -0,0 +1,273 @@
+---
+phase: 04-analysis-orchestration
+plan: 01
+type: execute
+wave: 1
+depends_on: []
+files_modified:
+ - kb_pipeline/processors/base.py
+ - kb_pipeline/processors/parse.py
+ - kb_pipeline/processors/__init__.py
+ - kb_pipeline/schemas.py
+ - tests/test_processors/test_base.py
+ - tests/test_processors/test_parse.py
+autonomous: true
+requirements: [MNT-01]
+user_setup: []
+
+must_haves:
+ truths:
+ - "BaseProcessor defines unified interface for all processors"
+ - "ParseProcessor correctly parses Bilibili and Zhihu URLs"
+ - "All processors inherit from BaseProcessor with consistent process() method"
+ artifacts:
+ - path: "kb_pipeline/processors/base.py"
+ provides: "BaseProcessor abstract class"
+ contains: "class BaseProcessor"
+ min_lines: 80
+ - path: "kb_pipeline/processors/parse.py"
+ provides: "ParseProcessor implementation"
+ contains: "class ParseProcessor"
+ min_lines: 60
+ - path: "kb_pipeline/schemas.py"
+ provides: "ParseInput/ParseOutput schemas"
+ contains: "class ParseInput"
+ key_links:
+ - from: "kb_pipeline/processors/parse.py"
+ to: "kb_pipeline/link_parser.py"
+ via: "LinkParser.parse()"
+ pattern: "LinkParser\\(\\)\\.parse"
+ - from: "kb_pipeline/processors/base.py"
+ to: "kb_pipeline/state.py"
+ via: "WorkflowStateManager integration"
+ pattern: "state_manager\\."
+---
+
+
+Create unified BaseProcessor interface and ParseProcessor to establish the foundation for the pipeline architecture.
+
+Purpose: Provide consistent processor interface that all steps (parse, download, asr, analyze) will follow, enabling unified state management and error handling.
+Output: BaseProcessor abstract class, ParseProcessor implementation, and related schemas.
+
+
+
+@/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/04-analysis-orchestration/04-CONTEXT.md
+@.planning/phases/04-analysis-orchestration/04-RESEARCH.md
+
+
+
+
+From kb_pipeline/schemas.py:
+```python
+class StepStatus(str, Enum):
+ PENDING = "pending"
+ RUNNING = "running"
+ COMPLETED = "completed"
+ FAILED = "failed"
+ SKIPPED = "skipped"
+
+class StepInput(BaseModel):
+ content_id: str = Field(..., min_length=1)
+ platform: Literal["bilibili", "zhihu"]
+
+class StepOutput(BaseModel):
+ success: bool
+ error_message: Optional[str] = None
+ completed_at: datetime = Field(default_factory=datetime.now)
+
+class WorkflowState(BaseModel):
+ def is_step_completed(self, step_name: str) -> bool: ...
+ def mark_step_started(self, step_name: str) -> None: ...
+ def mark_step_completed(self, step_name: str, output_summary: Optional[str] = None) -> None: ...
+ def mark_step_failed(self, step_name: str, error_message: str) -> None: ...
+```
+
+From kb_pipeline/state.py:
+```python
+class WorkflowStateManager:
+ STATE_FILENAME = "workflow_state.json"
+ def __init__(self, content_dir: Path): ...
+ def load(self) -> Optional[WorkflowState]: ...
+ def save(self, state: WorkflowState) -> None: ...
+ def get_or_create(self, content_id: str, platform: str, source_url: str) -> WorkflowState: ...
+```
+
+From kb_pipeline/link_parser.py:
+```python
+@dataclass
+class ParsedLink:
+ platform: str
+ content_type: str
+ content_id: str
+ original_url: str
+ resolved_url: str = ""
+
+class LinkParser:
+ def parse(self, url: str) -> Optional[ParsedLink]: ...
+```
+
+From kb_pipeline/processors/asr.py (existing pattern reference):
+```python
+class ASRProcessor:
+ STEP_NAME = "asr"
+ def __init__(self, state_manager: Optional[WorkflowStateManager] = None):
+ self.settings = get_settings()
+ self.state_manager = state_manager
+ async def process(self, input_data: ASRInput) -> ASROutput: ...
+```
+
+
+
+
+
+
+ Task 1: Define BaseProcessor abstract class
+ kb_pipeline/processors/base.py, tests/test_processors/test_base.py
+
+ - Test 1: BaseProcessor cannot be instantiated directly (abstract)
+ - Test 2: Subclass must implement process() method
+ - Test 3: BaseProcessor provides state_manager attribute
+ - Test 4: BaseProcessor provides STEP_NAME class attribute
+
+
+Create `kb_pipeline/processors/base.py` with BaseProcessor abstract class:
+
+1. Define abstract base class with:
+ - STEP_NAME: str class attribute (must be overridden by subclass)
+ - __init__(self, state_manager: Optional[WorkflowStateManager] = None)
+ - settings: KBSettings from get_settings()
+ - @abstractmethod process(self, input_data: StepInput) -> StepOutput
+
+2. Use ABC (Abstract Base Class) from abc module
+3. Import types from typing: Optional, TypeVar, Generic
+4. Keep it minimal - no complex template methods, just the interface contract
+
+Reference existing ASRProcessor pattern but extract the common interface.
+Do NOT add retry logic - that belongs in concrete implementations.
+Do NOT add complex inheritance hierarchies - keep it flat and simple.
+
+
+ pytest tests/test_processors/test_base.py -v -x
+
+ BaseProcessor abstract class exists with process() method, cannot be instantiated directly
+
+
+
+ Task 2: Add ParseInput and ParseOutput schemas
+ kb_pipeline/schemas.py, tests/test_pipeline/test_schemas.py
+
+ - Test 1: ParseInput validates url field (required, min_length=1)
+ - Test 2: ParseOutput contains platform, content_type, content_id, resolved_url
+ - Test 3: ParseOutput inherits success and error_message from StepOutput
+
+
+Add ParseInput and ParseOutput schemas to `kb_pipeline/schemas.py`:
+
+```python
+class ParseInput(StepInput):
+ """Parse step input - URL to parse"""
+ url: str = Field(..., min_length=1, description="URL to parse")
+ # Override content_id and platform as optional for parse step
+ content_id: str = Field(default="", description="Will be populated after parsing")
+ platform: str = Field(default="", description="Will be populated after parsing")
+
+class ParseOutput(StepOutput):
+ """Parse step output"""
+ platform: str = Field(..., description="Detected platform")
+ content_type: str = Field(..., description="Content type (video, article, etc.)")
+ content_id: str = Field(..., description="Extracted content ID")
+ resolved_url: str = Field(..., description="Resolved URL after short link expansion")
+ original_url: str = Field(..., description="Original URL provided by user")
+```
+
+Place after ASROutput schema (around line 236).
+Add tests to existing test_schemas.py file.
+
+
+ pytest tests/test_pipeline/test_schemas.py -v -x -k "Parse"
+
+ ParseInput and ParseOutput schemas exist and validate correctly
+
+
+
+ Task 3: Implement ParseProcessor
+ kb_pipeline/processors/parse.py, tests/test_processors/test_parse.py
+
+ - Test 1: ParseProcessor parses Bilibili video URL correctly
+ - Test 2: ParseProcessor parses Zhihu answer URL correctly
+ - Test 3: ParseProcessor returns ParseOutput with success=False for unknown URLs
+ - Test 4: ParseProcessor records state when state_manager is provided
+
+
+Create `kb_pipeline/processors/parse.py`:
+
+1. Import BaseProcessor from .base, LinkParser from kb_pipeline.link_parser
+2. Import ParseInput, ParseOutput from kb_pipeline.schemas
+3. Implement ParseProcessor(BaseProcessor):
+ - STEP_NAME = "parse"
+ - async def process(self, input_data: ParseInput) -> ParseOutput:
+ a. Create LinkParser instance
+ b. Call parser.parse(input_data.url)
+ c. If parsed is None, return ParseOutput(success=False, error_message="Unsupported URL")
+ d. If parsed, return ParseOutput with parsed fields
+ e. Record state on success (if state_manager exists)
+
+4. Keep implementation simple - just wrap LinkParser
+5. Handle both success and failure cases with proper state recording
+
+
+ pytest tests/test_processors/test_parse.py -v -x
+
+ ParseProcessor correctly parses URLs and returns structured ParseOutput
+
+
+
+ Task 4: Update processors __init__.py
+ kb_pipeline/processors/__init__.py
+
+Update `kb_pipeline/processors/__init__.py` to export new classes:
+
+```python
+from .base import BaseProcessor
+from .parse import ParseProcessor
+from .asr import ASRProcessor
+
+__all__ = ["BaseProcessor", "ParseProcessor", "ASRProcessor"]
+```
+
+Keep existing ASRProcessor export and add new ones.
+
+
+ python -c "from kb_pipeline.processors import BaseProcessor, ParseProcessor, ASRProcessor; print('OK')"
+
+ All processors can be imported from kb_pipeline.processors
+
+
+
+
+
+- All tests pass: `pytest tests/test_processors/test_base.py tests/test_processors/test_parse.py -v`
+- BaseProcessor is abstract and cannot be instantiated
+- ParseProcessor correctly wraps LinkParser
+- State recording works when state_manager is provided
+
+
+
+- BaseProcessor abstract class defines unified interface
+- ParseProcessor inherits from BaseProcessor
+- ParseInput/ParseOutput schemas exist and validate
+- All tests pass
+- Code follows existing patterns from ASRProcessor
+
+
+
diff --git a/.planning/phases/04-analysis-orchestration/04-02-PLAN.md b/.planning/phases/04-analysis-orchestration/04-02-PLAN.md
new file mode 100644
index 0000000..f7f1685
--- /dev/null
+++ b/.planning/phases/04-analysis-orchestration/04-02-PLAN.md
@@ -0,0 +1,256 @@
+---
+phase: 04-analysis-orchestration
+plan: 02
+type: execute
+wave: 2
+depends_on: [04-01]
+files_modified:
+ - kb_pipeline/processors/download.py
+ - kb_pipeline/schemas.py
+ - tests/test_processors/test_download.py
+autonomous: true
+requirements: [MNT-01]
+user_setup: []
+
+must_haves:
+ truths:
+ - "DownloadProcessor fetches metadata from platforms"
+ - "DownloadProcessor downloads content (video/audio or text)"
+ - "State is recorded after successful download"
+ artifacts:
+ - path: "kb_pipeline/processors/download.py"
+ provides: "DownloadProcessor implementation"
+ contains: "class DownloadProcessor"
+ min_lines: 100
+ - path: "kb_pipeline/schemas.py"
+ provides: "DownloadInput/DownloadOutput schemas"
+ contains: "class DownloadInput"
+ key_links:
+ - from: "kb_pipeline/processors/download.py"
+ to: "kb_pipeline/platforms/bilibili.py"
+ via: "BilibiliAdapter"
+ pattern: "BilibiliAdapter"
+ - from: "kb_pipeline/processors/download.py"
+ to: "kb_pipeline/platforms/zhihu.py"
+ via: "ZhihuAdapter"
+ pattern: "ZhihuAdapter"
+ - from: "kb_pipeline/processors/download.py"
+ to: "kb_pipeline/state.py"
+ via: "WorkflowStateManager"
+ pattern: "state_manager\\."
+---
+
+
+Create DownloadProcessor to handle metadata extraction and content downloading for all platforms.
+
+Purpose: Unify metadata and content fetching into a single processor that works with PlatformAdapter implementations.
+Output: DownloadProcessor implementation with Bilibili and Zhihu adapter integration.
+
+
+
+@/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/04-analysis-orchestration/04-CONTEXT.md
+@.planning/phases/04-analysis-orchestration/04-RESEARCH.md
+@.planning/phases/04-analysis-orchestration/04-01-SUMMARY.md
+
+
+
+
+From kb_pipeline/schemas.py:
+```python
+class StepInput(BaseModel):
+ content_id: str
+ platform: Literal["bilibili", "zhihu"]
+
+class StepOutput(BaseModel):
+ success: bool
+ error_message: Optional[str]
+ completed_at: datetime
+```
+
+From kb_pipeline/processors/base.py (from 04-01):
+```python
+class BaseProcessor(ABC):
+ STEP_NAME: str
+ async def process(self, input_data: StepInput) -> StepOutput
+ async def _do_process(self, input_data: StepInput) -> StepOutput
+```
+
+From kb_pipeline/platforms/base.py:
+```python
+class PlatformAdapter(ABC):
+ @abstractmethod
+ async def fetch_metadata(self, content_id: str, content_type: str) -> Dict[str, Any]
+
+ @abstractmethod
+ async def fetch_content(self, content_id: str, content_type: str) -> str
+
+ @abstractmethod
+ def get_output_filename(self, content_id: str, content_type: str, suffix: str) -> str
+```
+
+From kb_pipeline/platforms/bilibili.py:
+```python
+class BilibiliAdapter(PlatformAdapter):
+ # Already implemented with fetch_metadata and fetch_content
+```
+
+From kb_pipeline/platforms/zhihu.py:
+```python
+class ZhihuAdapter(PlatformAdapter):
+ # Already implemented with fetch_metadata and fetch_content
+```
+
+
+
+
+
+
+ Task 1: Define Download Schemas
+ kb_pipeline/schemas.py, tests/test_processors/test_download.py
+
+ - Test 1: DownloadInput validates required fields
+ - Test 2: DownloadOutput validates required fields
+ - Test 3: DownloadOutput handles both video and article content types
+
+
+Add DownloadInput and DownloadOutput schemas to kb_pipeline/schemas.py.
+
+**DownloadInput fields (extends StepInput):**
+- content_dir: str - Directory to save downloaded content
+- content_type: str - Type of content (video, article, answer, zhuanlan)
+
+**DownloadOutput fields (extends StepOutput):**
+- metadata_file: str - Path to metadata JSON
+- content_file: str - Path to content file (video/audio or text)
+- title: str - Content title
+- author: str - Content author
+- duration: Optional[int] - Duration for video content
+
+Follow StepInput/StepOutput pattern from Phase 1.
+Create test file with schema validation tests.
+
+
+ pytest tests/test_processors/test_download.py -v -x
+
+ DownloadInput/DownloadOutput schemas defined and validated with tests
+
+
+
+ Task 2: Create DownloadProcessor with metadata fetching
+ kb_pipeline/processors/download.py, tests/test_processors/test_download.py
+
+ - Test 1: DownloadProcessor fetches Bilibili video metadata correctly
+ - Test 2: DownloadProcessor fetches Zhihu answer metadata correctly
+ - Test 3: DownloadProcessor records state on success
+ - Test 4: DownloadProcessor records state on failure
+
+
+Create `kb_pipeline/processors/download.py` with DownloadProcessor class.
+
+**Implementation:**
+1. Import BaseProcessor, PlatformAdapter classes
+2. Import BilibiliAdapter, ZhihuAdapter from platforms
+3. Create DownloadProcessor(BaseProcessor):
+ - STEP_NAME = "download"
+ - __init__ with state_manager, settings
+ - process() method that:
+ a. Checks if step already completed (skip if done)
+ b. Gets appropriate PlatformAdapter based on platform
+ c. Calls adapter.fetch_metadata()
+ d. Saves metadata to content_dir
+ e. Records state on success/failure
+
+**State Recording:**
+- On success: mark_step_completed("download", f"title={title}, type={content_type}")
+- On failure: mark_step_failed("download", error_message)
+
+Create comprehensive tests mocking adapters and state_manager.
+
+
+ pytest tests/test_processors/test_download.py -v -x
+
+ DownloadProcessor fetches metadata with state recording
+
+
+
+ Task 3: Add content downloading to DownloadProcessor
+ kb_pipeline/processors/download.py, tests/test_processors/test_download.py
+
+ - Test 1: DownloadProcessor downloads Bilibili article content
+ - Test 2: DownloadProcessor skips video download (handled by ASR)
+ - Test 3: DownloadProcessor creates correct output file paths
+
+
+Extend DownloadProcessor to download content.
+
+**For articles/text content:**
+- Call adapter.fetch_content() to get text
+- Save content to content_dir as Markdown file
+- Update DownloadOutput with content_file path
+
+**For video content:**
+- Skip actual video download (handled by ASR in Phase 3)
+- Set content_file to empty or indicate video needs ASR
+
+**File naming:**
+- Metadata: `{platform}_{content_id}_metadata.json`
+- Content: `{platform}_{content_id}_content.md` (for articles)
+
+Update tests to verify content downloading behavior.
+
+
+ pytest tests/test_processors/test_download.py -v -x
+
+ DownloadProcessor downloads content for text-based content types
+
+
+
+ Task 4: Update processors __init__.py
+ kb_pipeline/processors/__init__.py
+
+Update `kb_pipeline/processors/__init__.py` to export DownloadProcessor.
+
+```python
+from .base import BaseProcessor
+from .parse import ParseProcessor
+from .download import DownloadProcessor
+from .asr import ASRProcessor
+
+__all__ = ["BaseProcessor", "ParseProcessor", "DownloadProcessor", "ASRProcessor"]
+```
+
+
+ python -c "from kb_pipeline.processors import DownloadProcessor; print('OK')"
+
+ DownloadProcessor exported from processors module
+
+
+
+
+
+- All tests pass: `pytest tests/test_processors/test_download.py -v`
+- DownloadProcessor correctly uses PlatformAdapter implementations
+- State recording works for both success and failure cases
+- Content files are saved to correct locations
+
+
+
+- DownloadProcessor inherits from BaseProcessor
+- DownloadInput/DownloadOutput schemas exist and validate
+- Metadata is fetched and saved correctly
+- Text content is downloaded and saved
+- Video content is marked for ASR processing
+- All tests pass
+
+
+
diff --git a/.planning/phases/04-analysis-orchestration/04-03-PLAN.md b/.planning/phases/04-analysis-orchestration/04-03-PLAN.md
new file mode 100644
index 0000000..522feaf
--- /dev/null
+++ b/.planning/phases/04-analysis-orchestration/04-03-PLAN.md
@@ -0,0 +1,255 @@
+---
+phase: 04-analysis-orchestration
+plan: 03
+type: execute
+wave: 3
+depends_on: [04-01, 04-02]
+files_modified:
+ - kb_pipeline/processors/analyze.py
+ - kb_pipeline/schemas.py
+ - tests/test_processors/test_analyze.py
+autonomous: true
+requirements: [MNT-01]
+user_setup: []
+
+must_haves:
+ truths:
+ - "AnalyzeProcessor generates AI analysis reports"
+ - "Analysis reports are saved as Markdown files"
+ - "AnalyzeProcessor wraps existing DeepAnalyzer"
+ artifacts:
+ - path: "kb_pipeline/processors/analyze.py"
+ provides: "AnalyzeProcessor implementation"
+ contains: "class AnalyzeProcessor"
+ min_lines: 100
+ - path: "kb_pipeline/schemas.py"
+ provides: "AnalyzeInput/AnalyzeOutput schemas"
+ contains: "class AnalyzeInput"
+ - path: "tests/test_processors/test_analyze.py"
+ provides: "Comprehensive test suite"
+ contains: "test_analyze_processor"
+ key_links:
+ - from: "kb_pipeline/processors/analyze.py"
+ to: "kb-deep-analyzer/deep_analyzer.py"
+ via: "DeepAnalyzer class import"
+ pattern: "from.*deep_analyzer.*import"
+ - from: "kb_pipeline/processors/analyze.py"
+ to: "kb_pipeline/state.py"
+ via: "WorkflowStateManager"
+ pattern: "state_manager\\."
+---
+
+
+Create AnalyzeProcessor to wrap existing DeepAnalyzer and integrate it into the unified processor pipeline.
+
+Purpose: Provide AI-powered content analysis that generates structured reports as Markdown files.
+Output: AnalyzeProcessor implementation with DeepAnalyzer integration.
+
+
+
+@/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/04-analysis-orchestration/04-CONTEXT.md
+@.planning/phases/04-analysis-orchestration/04-RESEARCH.md
+@.planning/phases/04-analysis-orchestration/04-01-SUMMARY.md
+@.planning/phases/04-analysis-orchestration/04-02-SUMMARY.md
+
+
+
+
+From kb-deep-analyzer/deep_analyzer.py:
+```python
+class SimpleAIClient:
+ def __init__(self, api_key: str = None, provider: str = None, model: str = None, base_url: str = None):
+ ...
+ def generate(self, prompt: str, max_tokens: int = 16000) -> str:
+ ...
+
+class DeepAnalyzer:
+ def __init__(self, ai_client: SimpleAIClient = None):
+ ...
+ def analyze(self, metadata: Dict[str, Any], content: Dict[str, Any], content_type: str = "video") -> AnalysisResult:
+ ...
+
+@dataclass
+class AnalysisResult:
+ summary: str = ""
+ main_topics: List[str] = field(default_factory=list)
+ key_insights: List[str] = field(default_factory=list)
+ recommended_tags: List[str] = field(default_factory=list)
+ full_analysis: str = ""
+ original_text: str = ""
+
+def generate_frontmatter(metadata: Dict, result: AnalysisResult, content_type: str = "video") -> str:
+ ...
+```
+
+From kb_pipeline/schemas.py (after 04-01 and 04-02):
+```python
+class StepInput(BaseModel):
+ content_id: str
+ platform: Literal["bilibili", "zhihu"]
+
+class StepOutput(BaseModel):
+ success: bool
+ error_message: Optional[str]
+ completed_at: datetime
+```
+
+From kb_pipeline/processors/base.py (from 04-01):
+```python
+class BaseProcessor(ABC):
+ STEP_NAME: str
+ async def process(self, input_data: StepInput) -> StepOutput
+```
+
+
+
+
+
+
+ Task 1: Define Analyze Schemas
+ kb_pipeline/schemas.py, tests/test_processors/test_analyze.py
+
+ - Test 1: AnalyzeInput validates required fields
+ - Test 2: AnalyzeOutput validates required fields
+ - Test 3: AnalyzeOutput handles both video and article content types
+
+
+Add AnalyzeInput and AnalyzeOutput schemas to kb_pipeline/schemas.py.
+
+**AnalyzeInput fields (extends StepInput):**
+- metadata_file: str - Path to metadata JSON file
+- content_file: str - Path to content file (subtitle JSON or article content)
+- content_type: str - Type of content (video or article)
+- output_file: str - Path to output analysis Markdown file
+
+**AnalyzeOutput fields (extends StepOutput):**
+- analysis_file: str - Path to generated analysis report
+- summary: str - Content summary
+- main_topics: List[str] - Main topics extracted
+- key_insights: List[str] - Key insights
+- tags: List[str] - Recommended tags
+- stats: Dict[str, Any] - Statistics about analysis
+
+Create test file with schema validation tests.
+
+
+ pytest tests/test_processors/test_analyze.py -v -x -k "test_schema"
+
+ AnalyzeInput/AnalyzeOutput schemas defined and validated
+
+
+
+ Task 2: Create AnalyzeProcessor with DeepAnalyzer integration
+ kb_pipeline/processors/analyze.py, tests/test_processors/test_analyze.py
+
+ - Test 1: AnalyzeProcessor loads metadata and content from files
+ - Test 2: AnalyzeProcessor calls DeepAnalyzer.analyze() with correct parameters
+ - Test 3: AnalyzeProcessor saves analysis report as Markdown
+ - Test 4: AnalyzeProcessor records state on success
+ - Test 5: AnalyzeProcessor records state on failure
+
+
+Create `kb_pipeline/processors/analyze.py` with AnalyzeProcessor class.
+
+**Implementation:**
+1. Import BaseProcessor from .base
+2. Import DeepAnalyzer, SimpleAIClient, AnalysisResult, generate_frontmatter from kb_deep_analyzer
+3. Import AnalyzeInput, AnalyzeOutput from ..schemas
+4. Create AnalyzeProcessor(BaseProcessor):
+ - STEP_NAME = "analyze"
+ - __init__ with state_manager, settings (initialize AI client lazily)
+ - process() method that:
+ a. Checks if step already completed (skip if done)
+ b. Loads metadata JSON file
+ c. Loads content JSON/text file
+ d. Initializes SimpleAIClient with settings
+ e. Creates DeepAnalyzer instance
+ f. Calls analyzer.analyze(metadata, content, content_type)
+ g. Generates frontmatter using generate_frontmatter()
+ h. Writes analysis report as Markdown
+ i. Records state on success/failure
+
+**Output file format:**
+```markdown
+---
+title: "..."
+date: ...
+author: ...
+tags: [...]
+---
+
+## 标题
+...
+
+## 摘要
+...
+
+## 主要主题
+...
+
+## 关键见解
+...
+
+## 时间轴 (for video only)
+...
+```
+
+Create comprehensive tests with mocked DeepAnalyzer.
+
+
+ pytest tests/test_processors/test_analyze.py -v -x
+
+ AnalyzeProcessor generates analysis reports using DeepAnalyzer
+
+
+
+ Task 3: Update processors __init__.py
+ kb_pipeline/processors/__init__.py
+
+Update `kb_pipeline/processors/__init__.py` to export AnalyzeProcessor.
+
+```python
+from .base import BaseProcessor
+from .parse import ParseProcessor
+from .download import DownloadProcessor
+from .asr import ASRProcessor
+from .analyze import AnalyzeProcessor
+
+__all__ = ["BaseProcessor", "ParseProcessor", "DownloadProcessor", "ASRProcessor", "AnalyzeProcessor"]
+```
+
+
+ python -c "from kb_pipeline.processors import AnalyzeProcessor; print('OK')"
+
+ AnalyzeProcessor exported from processors module
+
+
+
+
+
+- All tests pass: `pytest tests/test_processors/test_analyze.py -v`
+- AnalyzeProcessor correctly wraps DeepAnalyzer
+- Analysis reports are saved as Markdown files
+- Frontmatter is generated correctly
+- State recording works for both success and failure cases
+
+
+
+- AnalyzeProcessor inherits from BaseProcessor
+- AnalyzeInput/AnalyzeOutput schemas exist and validate
+- DeepAnalyzer is integrated correctly (no modifications to existing code)
+- Analysis reports are saved with proper frontmatter
+- All tests pass
+
+
+
diff --git a/.planning/phases/04-analysis-orchestration/04-04-PLAN.md b/.planning/phases/04-analysis-orchestration/04-04-PLAN.md
new file mode 100644
index 0000000..b1f288b
--- /dev/null
+++ b/.planning/phases/04-analysis-orchestration/04-04-PLAN.md
@@ -0,0 +1,658 @@
+---
+phase: 04-analysis-orchestration
+plan: 04
+type: execute
+wave: 4
+depends_on: [04-01, 04-02, 04-03]
+files_modified:
+ - kb_pipeline/pipeline.py
+ - kb_pipeline/processors/__init__.py
+ - kb-workflow/kb_workflow.py
+ - tests/test_pipeline/test_pipeline.py
+ - tests/test_cli/test_process_command.py
+autonomous: true
+requirements: [MNT-01]
+user_setup: []
+
+must_haves:
+ truths:
+ - "Pipeline orchestrestr executes full workflow from URL to analysis"
+ - "Single command triggers entire workflow"
+ - "Progress is visible through simple logging"
+ - "New developers can understand code in 30 minutes"
+ artifacts:
+ - path: "kb_pipeline/pipeline.py"
+ provides: "Pipeline engine class"
+ min_lines: 100
+ - path: "kb-workflow/kb_workflow.py"
+ provides: "CLI entry point with process subcommand"
+ min_lines: 50
+ - path: "tests/test_pipeline/test_pipeline.py"
+ provides: "Pipeline tests"
+ min_lines: 100
+ - path: "tests/test_cli/test_process_command.py"
+ provides: "CLI tests"
+ key_links:
+ - from: "kb_pipeline/pipeline.py"
+ to: "kb_pipeline/processors"
+ via: "Pipeline.run()"
+ pattern: "Pipeline\\("
+ - from: "kb-workflow/kb_workflow.py"
+ to: "kb_pipeline.processors"
+ via: "import"
+ pattern: "from.* import"
+ - from: "kb_pipeline/processors" import ParseProcessor, DownloadProcessor, ASrProcessor, AnalyzeProcessor
+ pattern: "from.* import.*"
+---
+
+
+Create Pipeline engine to orchest all processors and execute the complete workflow from URL to analysis report.
+
+Purpose: Enable single-command workflow execution with automatic state management and error recovery, and and new developer to easily understand and code结构.
+Output: Pipeline engine class, process subcommand, and CLI entry point.
+
+
+
+@/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/04-analysis-orchestration/04-CONTEXT.md
+@.planning/phases/04-analysis-orchestration/04-RESEARCH.md
+
+@.planning/phases/04-analysis-orchestration/04-01-SUMMARY.md
+@.planning/phases/04-analysis-orchestration/04-02-SUMMARY.md
+@.planning/phases/04-analysis-orchestration/04-03-SUMMARY.md
+
+
+
+
+From kb_pipeline/processors/base.py:
+```python
+class BaseProcessor(ABC):
+ STEP_NAME: str
+ async def process(self, input_data: StepInput) -> StepOutput:
+```
+
+From kb_pipeline/processors/parse.py:
+```python
+class ParseProcessor(BaseProcessor):
+ STEP_NAME = "parse"
+ async def process(self, input_data: ParseInput) -> ParseOutput
+```
+
+from kb_pipeline/processors/download.py:
+```python
+class DownloadProcessor(BaseProcessor):
+ STEP_NAME = "download"
+ async def process(self, input_data: DownloadInput) -> DownloadOutput
+```
+
+from kb_pipeline/processors/asr.py:
+```python
+class ASRProcessor:
+ STEP_NAME = "asr"
+ async def process(self, input_data: ASRInput) -> ASROutput
+```
+
+from kb_pipeline/processors/analyze.py:
+```python
+class AnalyzeProcessor(BaseProcessor):
+ STEP_NAME = "analyze"
+ async def process(self, input_data: AnalyInput) -> AnalyOutput
+```
+
+from kb_pipeline/schemas.py:
+```python
+class StepInput(BaseModel):
+ content_id: str = platform: Literal["bilibili", "zhihu"]
+
+class StepOutput(BaseModel):
+ success: bool
+ error_message: Optional[str]
+ completed_at: datetime
+
+class ParseInput(StepInput):
+ url: str
+
+class ParseOutput(StepOutput):
+ parsed: ParsedLink
+ content_dir: Path
+ metadata_file: Optional[str] = None
+ content_file: Optional[str] = None
+ content_type: str = "video" # video needs AS
+
+ # article不需要
+
+ content_file: Optional[str] = None # 从 download步骤获取
+
+ content_type: str # "video" or "article"
+ download_input.content_dir
+ download_output(
+ success=True,
+ metadata_file=f"{content_dir}/{platform}_{content_id}_metadata.json",
+ content_file=content_file,
+ title=metadata.get('title', '未知'),
+ author=metadata.get('author', '未知'),
+ duration=metadata.get('duration') # seconds, )
+
+ return state.mark_step_completed(
+ self.STEP_NAME, f"title={title}, author={author}, type={content_type}"
+ )
+ self.state_manager.save(state)
+ return output
+
+```
+
+from kb_pipeline.state import WorkflowStateManager
+from kb_pipeline.config import get_settings
+from kb_pipeline.errors import PipelineError
+
+from kb_pipeline.processors.parse import ParseProcessor
+from kb_pipeline.processors.download import DownloadProcessor
+from kb_pipeline.processors.asr import ASRProcessor
+from kb_pipeline.processors.analyze import AnalyzeProcessor
+
+from typing import List, Optional, Dict, Any, TYPE_CHECKING
+from pathlib import Path
+from datetime import datetime
+
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class PipelineConfig:
+ """Pipeline configuration"""
+ steps: List[str] = Field(default_factory=lambda: ["parse", "download", "asr", "analyze"])
+ output_dir: Path
+ skip_completed: bool = Field(default=True, description="Skip completed steps")
+ max_retries: int = Field(default=3, description="Max retries for each step")
+
+
+class Pipeline:
+ """Pipeline engine - orchestrates multiple processors
+
+ Orchestrates the workflow:
+ 1. Parse URL to extract platform and content ID
+ 2. Run processors in order (parse, download, asr, analyze)
+ 3. Handle errors and retry on failures
+ 4. Return results
+
+ Key decisions:
+ - Sequential execution (not parallel - simpler to easier to debug
+ - State persistence after each step (via WorkflowStateManager)
+ - Progress logging to simple, clear
+ """
+
+ def __init__(
+ self,
+ processors: List[BaseProcessor],
+ config: Optional[PipelineConfig] = None,
+ output_dir: Optional[Path] = None,
+ skip_completed: bool = False
+ ):
+ self.processors = processors
+ self.config = config or PipelineConfig.default()
+ self.output_dir = output_dir
+
+ self.skip_completed = skip_completed
+
+ # Validate processors
+ for processor in self.processors:
+ if processor.STEP_NAME not in self.config.steps:
+ raise ValueError(f"Unknown processor: {processor.STEP_NAME}")
+
+ async def run(self, url: str) -> Dict[str, Any]:
+ """Execute full workflow
+
+ Args:
+ url: URL to process
+
+ Returns:
+ Dict containing all step results
+ """
+ results: Dict[str, Any] = {}
+ current_step = 0
+ total_steps = len(self.config.steps)
+
+ state: Optional[WorkflowState] = None
+
+ try:
+ for step_name in self.config.steps:
+ current_step += 1
+ logger.info(f"[Step {current_step}/{total_steps}] {step_name}...")
+
+
+ # Get processor
+ processor = self.processors.get(step_name)
+ if not processor:
+ raise ValueError(f"Processor not found: {step_name}")
+
+ # Get content directory from parsed URL
+ parsed = await self._parse_url(url)
+ if not parsed:
+ results["error"] = "Failed to parse URL"
+ return results
+
+ # Get or create content directory
+ content_dir = self.output_dir / f"{parsed.platform}_{parsed.content_id}"
+ content_dir.mkdir(parents=True, exist_ok=True)
+
+ # Initialize state manager for this content
+ state_manager = WorkflowStateManager(content_dir)
+
+ # Run each step
+ previous_result = None
+ for step_name in self.config.steps:
+ # Check if already completed
+ if state and state.is_step_completed(step_name):
+ current_step += 1
+ logger.info(f"[Step {current_step}/{total_steps}] {step_name} - skipping (already completed)")
+ results[step_name] = {"status": "skipped", "message": "Already completed"}
+ continue
+
+ # Create step input
+ input_data = self._create_step_input(
+ step_name=step_name,
+ parsed=parsed,
+ state=state,
+ previous_result=previous_result
+ )
+
+ # Run processor
+ try:
+ output = await processor.process(input_data)
+ results[step_name] = {
+ "status": "success" if output.success else "failed",
+ "output": output.model_dump() if output.success else None,
+ "error": output.error_message
+ }
+ previous_result = output
+ except Exception as e:
+ results[step_name] = {
+ "status": "failed",
+ "error": str(e)
+ }
+ logger.error(f"[Step {current_step}/{total_steps}] {step_name} - failed: {e}")
+ # Don't stop pipeline on failure
+ raise
+
+ return results
+
+ def _parse_url(self, url: str) -> Optional[ParsedLink]:
+ """Parse URL using LinkParser"""
+ if not url:
+ return None
+
+ parser = LinkParser()
+ parsed = parser.parse(url)
+ if not parsed:
+ logger.warning(f"Failed to parse URL: {url}")
+ return None
+
+ return ParsedLink
+
+ def _create_step_input(
+ self,
+ step_name: str,
+ parsed: ParsedLink,
+ state: Optional[WorkflowState],
+ previous_result: Optional[Dict[str, Any]],
+ ) -> Dict[str, Any]:
+ """Create step input based on step name and previous result"""
+ step_map = {
+ "parse": ParseProcessor,
+ "download": DownloadProcessor,
+ "asr": ASRProcessor,
+ "analyze": AnalyzeProcessor,
+ }
+
+ # Get processor instance
+ processor = self.processors[step_name]
+ if not processor:
+ raise ValueError(f"Unknown processor: {step_name}")
+
+ return step_map
+
+ async def _run_step(
+ self,
+ step_name: str,
+ input_data: Dict[str, Any],
+ state: Optional[WorkflowState],
+ previous_result: Optional[Dict[str, Any]]
+ ) -> StepOutput:
+ """Run a single step"""
+ processor = self.processors[step_name]
+ if not processor:
+ raise ValueError(f"No processor for {step_name}")
+
+ # Check if already completed
+ if state and state.is_step_completed(step_name):
+ logger.info(f"Step {step_name} already completed, skipping")
+ return self._create_skipped_output()
+
+
+ try:
+ # Run the
+ if previous_result and previous_result.get(step_name):
+ input_data = previous_result[step_name]
+ else:
+ raise ValueError(f"No previous result for {step_name}")
+
+ return await processor.process(input_data)
+ except Exception as e:
+ logger.error(f"Step {step_name} failed: {e}")
+ # Record failure
+ if state:
+ state.mark_step_failed(step_name, str(e))
+ self.state_manager.save(state)
+ raise
+
+ # Update previous result
+ previous_result[step_name] = output.model_dump()
+
+ return results
+
+ def get_output_dir(self) -> Path:
+ return Path(self.output_dir) / Path(self.config.output_dir)
+
+
+
+
+
+
+
+
+ Task 1: Define Pipeline Schemas
+ kb_pipeline/schemas.py, tests/test_pipeline/test_pipeline.py
+
+ - Test 1: PipelineInput validates required fields
+ - Test 2: PipelineOutput validates required fields
+ - Test 3: PipelineConfig default steps are correct
+
+
+Add PipelineInput, PipelineOutput, and PipelineConfig to kb_pipeline/schemas.py.
+
+**PipelineInput fields (extends nothing - all fields inherited from StepInput):**
+- content_dir: str - Directory to save downloaded content
+
+- content_type: str - Type of content (video, article, answer, zhuanlan)
+
+- skip_completed: bool - Whether to skip already completed steps (default True, content_type not to ["video", "article", "answer", "zhuanlan", "download", "asr", "analyze"]
+
+- steps: List[str] = Field(...)
+
+ output_dir: Path
+ skip_completed: bool = True
+
+
+ pytest tests/test_pipeline/test_pipeline.py -v -x -k"test_schema" in
+
+ PipelineInput/Output schemas defined and validated
+
+
+
+ Task 2: Create Pipeline class with processor initialization
+ kb_pipeline/pipeline.py, tests/test_pipeline/test_pipeline.py
+
+ - Test 1: Pipeline initializes with all processors
+ - Test 2: Pipeline.parse_url creates correct content directory
+ - Test 3: Pipeline.run() executes all steps in correct order
+ - Test 4: Pipeline.run() handles step skipping correctly
+ - Test 5: Pipeline.run() handles errors and continues to next step
+ - Test 6: Pipeline.run() records state for each step
+
+
+Create `kb_pipeline/pipeline.py` with Pipeline class.
+
+**Implementation:**
+1. Initialize with processors list (ParseProcessor, DownloadProcessor, ASRProcessor, AnalyzeProcessor)
+2. Set config with PipelineConfig.default()
+3. Set output_dir
+4. Initialize state_manager (WorkflowStateManager)
+5. Implement run() method that:
+ - Creates content directory
+ - Runs steps in order (parse, download, asr, analyze)
+ - Returns results dict
+
+6. Add helper methods:
+ - _parse_url(url) -> ParsedLink
+ - _get_content_dir(parsed)
+ - _run_step(step_name, parsed, state, previous_result)
+ - _record_state
+
+7. Add error handling with try/except
+ - logger.error for error message
+ if self.state_manager:
+ state.mark_step_failed(step_name, str(e))
+ self.state_manager.save(state)
+ raise
+
+ def _parse_url(self, url: str) -> Optional[ParsedLink]:
+ """Parse URL using LinkParser"""
+ if not url:
+ return None
+
+ parser = LinkParser()
+ parsed = parser.parse(url)
+ if not parsed:
+ return None
+
+ return Parsed
+
+ def _get_content_dir(self, parsed: ParsedLink) -> Path:
+ """Get content directory from parsed URL"""
+ content_dir = self.output_dir / f"{parsed.platform}_{parsed.content_id}"
+ content_dir.mkdir(parents=True, exist_ok=True)
+ return content_dir
+
+ def _create_step_input(
+ self,
+ step_name: str,
+ parsed: ParsedLink,
+ state: Optional[WorkflowState],
+ previous_result: Optional[Dict[str, Any]]
+ ) -> Dict[str, Any]:
+ """Create step input based on step name and previous result"""
+ step_map = {
+ "parse": ParseProcessor,
+ "download": DownloadProcessor,
+ "asr": ASRProcessor,
+ "analyze": AnalyzeProcessor
+ }
+
+ # Get processor instance
+ processor = self.processors[step_name]
+ if not processor:
+ raise ValueError(f"Unknown processor: {step_name}")
+
+ return step_map
+
+ async def _run_step(
+ self,
+ step_name: str,
+ input_data: Dict[str, Any],
+ state: Optional[WorkflowState],
+ previous_result: Optional[Dict[str, Any]],
+ ) -> StepOutput:
+ """Run a single step"""
+ processor = self.processors[step_name]
+ if not processor:
+ raise ValueError(f"No processor for {step_name}")
+
+ # Check if already completed
+ if state and state.is_step_completed(step_name):
+ logger.info(f"Step {step_name} already completed, skipping")
+ return self._create_skipped_output()
+
+ try:
+ # Run step
+ if previous_result and previous_result.get(step_name):
+ input_data = previous_result[step_name]
+ else:
+ raise ValueError(f"No previous result for {step_name}")
+
+ return await processor.process(input_data)
+ except Exception as e:
+ logger.error(f"Step {step_name} failed: {e}")
+ if state:
+ state.mark_step_failed(step_name, str(e))
+ self.state_manager.save(state)
+ raise
+
+ # Update previous result
+ previous_result[step_name] = output.model_dump()
+ return results
+
+ def _create_skipped_output(self) -> StepOutput:
+ return StepOutput(success=True, error_message="Skipped: already completed")
+
+
+
+
+ Task 3: Add step execution logic
+ kb_pipeline/pipeline.py
+
+Extend Pipeline.run() to implement step execution logic:
+
+1. Create content directory
+2. Run steps in order
+3. Handle step skipping (check state)
+4. Handle errors (try/except, log and continue)
+5. Record final state
+6. Return results
+
+**Step execution logic:**
+```python
+async def _execute_steps(
+ self,
+ parsed: ParsedLink,
+ state: WorkflowState,
+ previous_result: Dict[str, Any]
+) -> None:
+ for step_name in self.config.steps:
+ step_start = datetime.now()
+ state.mark_step_started(step_name)
+
+ logger.info(f"[Step {step_num}/{total_steps}] {step_name}...")
+
+ # Get or run processor
+ processor = self.processors[step_name]
+ if not processor:
+ raise ValueError(f"Unknown processor: {step_name}")
+
+ # Get step input
+ input_data = self._create_step_input(
+ step_name=step_name,
+ parsed=parsed,
+ state=state,
+ previous_result=previous_result.get(step_name)
+ )
+
+ # Run processor
+ try:
+ output = await processor.process(input_data)
+ previous_result[step_name] = output.model_dump()
+ results[step_name] = {
+ "status": "success" if output.success else "failed",
+ "output": output.model_dump() if output.success else None,
+ "error": output.error_message
+ }
+ previous_result = output
+ except Exception as e:
+ results[step_name] = {
+ "status": "failed",
+ "error": str(e)
+ }
+ logger.error(f"[Step {step_num}/{total_steps}] {step_name} - failed: {e}")
+ # Record failure
+ if state:
+ state.mark_step_failed(step_name, str(e))
+ self.state_manager.save(state)
+ # Stop on failure
+ raise
+
+ return results
+
+
+ pytest tests/test_pipeline/test_pipeline.py -v -x
+
+ Pipeline engine executes workflow with state persistence
+
+
+
+ Task 4: Add CLI process command
+ kb-workflow/kb_workflow.py
+
+Extend kb-workflow/kb_workflow.py to add process subcommand.
+
+**CLI additions:**
+- New argument: 'url' with help='URL to process'
+- New argument: '--output-dir', '-o' with help='Output directory (default: KB_BASE_PATH/content)
+- New argument: '--no-analyze', action='store_true', help='Skip analysis step')
+- Set default to process function that handles the command
+
+- Call Pipeline().run(url, output_dir, skip_completed=no)
+- Print success message
+- If error: print error message and exit(1)
+
+**Update existing help text and kb_workflow.py to document the new command.
+
+
+ python kb-workflow/kb_workflow.py --help | grep process
+
+ CLI process command added and functional
+
+
+
+ Task 5: Create Pipeline tests
+ tests/test_pipeline/test_pipeline.py,files>
+
+Create comprehensive tests for Pipeline class.
+
+**Test cases:**
+1. Test pipeline initialization
+2. Test URL parsing and content dir creation
+3. Test step execution order
+4. Test step skipping
+5. Test error handling and6. Test state persistence
+
+7. Test full workflow (mocked)
+
+Use fixtures for mocking processors, state_manager, settings.
+
+
+ pytest tests/test_pipeline/test_pipeline.py -v
+
+ Pipeline class fully tested
+
+
+
+
+
+- All tests pass: `pytest tests/test_pipeline/test_pipeline.py tests/test_cli/test_process_command.py -v`
+- Pipeline.run() executes workflow correctly
+- State persistence works
+- Error handling works
+- Step skipping works
+
+
+
+- Pipeline class orchestrates all processors
+- Pipeline.run() executes workflow from URL to analysis
+- Content directory created correctly
+- Steps executed in order
+- State persisted after each step
+- Errors handled with retry
+- Completed steps skipped
+- CLI process command works
+
+
+
diff --git a/.planning/phases/04-analysis-orchestration/04-05-PLAN.md b/.planning/phases/04-analysis-orchestration/04-05-PLAN.md
new file mode 100644
index 0000000..e12da3a
--- /dev/null
+++ b/.planning/phases/04-analysis-orchestration/04-05-PLAN.md
@@ -0,0 +1,288 @@
+---
+phase: 04-analysis-orchestration
+plan: 05
+type: execute
+wave: 5
+depends_on: [04-04]
+files_modified:
+ - README.md
+ - tests/test_integration/test_e2e_workflow.py
+autonomous: false
+requirements: [MNT-01]
+user_setup: []
+
+must_haves:
+ truths:
+ - "End-to-end workflow completes successfully"
+ - "Analysis report is saved to content directory"
+ - "workflow_state.json correctly records all steps"
+ artifacts:
+ - path: "README.md"
+ provides: "Project documentation"
+ contains: "## Installation"
+ - path: "tests/test_integration/test_e2e_workflow.py"
+ provides: "End-to-end integration test"
+ contains: "test_e2e"
+ key_links:
+ - from: "kb_pipeline/pipeline.py"
+ to: "kb_pipeline/processors"
+ via: "import processors"
+ pattern: "from.*processor import"
+ - from: "kb-workflow/kb_workflow.py"
+ to: "kb_pipeline"
+ via: "import pipeline"
+ pattern: "from.*pipeline import"
+---
+
+
+Create end-to-end integration test and documentation ( project overview, installation, and and usage.
+
+Purpose: Verify the complete workflow works and provide clear project documentation for maintainability.
+Output: README.md with project overview, tests/test_integration/test_e2e_workflow.py
+
+
+
+@/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/04-analysis-orchestration/04-CONTEXT.md
+@.planning/phases/04-analysis-orchestration/04-RESEARCH.md
+@.planning/phases/04-analysis-orchestration/04-04-SUMMARY.md
+
+
+
+
+From kb_pipeline/pipeline.py:
+```python
+class Pipeline:
+ """Pipeline orchestration engine"""
+ def __init__(self, ...): # From 04-01
+ ...
+```
+
+From kb_pipeline/processors/base.py:
+```python
+class BaseProcessor(ABC):
+ STEP_NAME: str
+ async def process(self, input_data: StepInput) -> StepOutput
+```
+
+from kb_pipeline/processors/parse.py:
+```python
+class ParseProcessor(BaseProcessor):
+ STEP_NAME = "parse"
+ async def process(self, input_data: ParseInput) -> ParseOutput
+```
+
+from kb_pipeline/processors/download.py:
+```python
+class DownloadProcessor(BaseProcessor):
+ STEP_NAME = "download"
+ async def process(self, input_data: DownloadInput) -> DownloadOutput
+```
+
+from kb_pipeline/processors/asr.py:
+```python
+class ASRProcessor:
+ STEP_NAME = "asr"
+ async def process(self, input_data: ASRInput) -> ASROutput
+```
+
+from kb_pipeline/processors/analyze.py
+```python
+class AnalyzeProcessor(BaseProcessor):
+ STEP_NAME = "analyze"
+ async def process(self, input_data: AnalyzeInput) -> AnalyOutput
+```
+
+
+
+
+
+
+ Task 1: Create README.md
+ README.md
+
+ - Test 1: README.md exists and contains project overview
+ - Test 2: README.md contains installation instructions
+
+
+Create `README.md` in project root directory with the following sections:
+
+**## kb-skills**
+
+知识库工作流工具，从链接到分析报告的端到端自动化处理系统。
+
+### 格 Features
+
+- 统一 Processor 模式： Parse -> Download -> ASR -> Analyze
+- 状态管理: workflow_state.json
+- 配置管理: 通过 ~/.kb_env
+- 输出: Markdown 格式分析报告
+
+- 幂等性保证: 重复执行不会产生重复文件
+
+- 断点续传: 中断后可继续
+
+- 模块化设计: 新人能在30分钟内理解代码结构
+- 清晰的项目结构
+
+- 统一的 CLI 入
+
+ entry
+
+### Installation
+
+```bash
+# 安装依赖
+pip install kb-pipeline
+```
+
+验证安装:
+```bash
+# 测试
+pytest tests/test_pipeline/test_schemas.py tests/test_pipeline/test_state.py tests/test_pipeline/test_config.py tests/test_pipeline/test_errors.py tests/test_pipeline/test_retry.py
+```
+
+
+
+ python -c "import os; content = open('README.md', 'r', encoding='utf-8'); print(content[:100]) > 0"
+
+ README.md created with project overview, installation, and, usage sections, docstrings
+
+
+
+ Task 2: Create end-to-end integration test
+ tests/test_integration/test_e2e_workflow.py
+
+ - Test 1: Full workflow executes from URL to analysis
+ - Test 2: Workflow handles errors gracefully
+ - Test 3: State is persisted correctly
+ - Test 4: Workflow is resumable
+
+
+Create `tests/test_integration/test_e2e_workflow.py` with comprehensive end-to-end tests.
+
+**Test structure:**
+```python
+import pytest
+import pytest_asyncio
+from unittest.mock import AsyncMock, patch, MagicMock
+from pathlib import Path
+from typing import Dict, Any
+
+from kb_pipeline.pipeline import Pipeline
+from kb_pipeline.config import get_settings
+from kb_pipeline.processors import ParseProcessor, DownloadProcessor, ASRProcessor, AnalyzeProcessor
+
+
+@pytest.fixture
+def temp_content_dir(tmp_path):
+ """Create temp content directory"""
+ content_dir.mkdir(parents=True, exist_ok=True)
+ yield content_dir
+
+ # Create test workflow state
+ state_path = content_dir / "workflow_state.json"
+ state = WorkflowState(
+ content_id="test123",
+ platform="bilibili",
+ source_url="https://test.url"
+ )
+ state_manager = WorkflowStateManager(content_dir)
+ yield state_manager
+
+ # Create real pipeline
+ settings = get_settings()
+ pipeline = Pipeline(
+ processors=[
+ ParseProcessor(state_manager),
+ DownloadProcessor(state_manager),
+ ASRProcessor(state_manager),
+ AnalyzeProcessor(state_manager),
+ ],
+ output_dir=content_dir,
+ config=PipelineConfig(
+ steps=["parse", "download", "asr", "analyze"],
+ )
+ )
+ yield pipeline
+ # Mock workflow run
+ result = await pipeline.run("https://test.url")
+
+ # Verify success
+ assert result["parse"]["success"]
+ assert result["download"]["success"]
+ assert result["asr"]["success"] or result["asr"]["skipped"] # ASR not applicable for text content
+ assert result["analyze"]["success"]
+
+ # Verify state was saved
+ state = state_manager.load()
+ assert state is not None
+ assert state.is_step_completed("parse")
+ assert state.is_step_completed("download")
+ if content_type in ["video", "zvideo"]:
+ assert state.is_step_completed("asr")
+ assert not state.is_step_completed("asr")
+ assert state.is_step_completed("analyze")
+
+ # Verify output files exist
+ assert (content_dir / "bilibili_test123_metadata.json").exists()
+ assert (content_dir / "bilibili_test123_content.json").exists() or result["content_type"] == "video":
+ assert (content_dir / "bilibili_test123_analysis.md").exists()
+
+
+ pytest tests/test_integration/test_e2e_workflow.py -v
+
+ End-to-end workflow tested with state persistence
+
+
+
+
+End-to-end workflow (Phase 1-2/3):
+- Pipeline Engine with 4 processors (Parse, Download, ASR, Analyze)
+- CLI entry point (kb-workflow process command)
+- DeepAnalyzer integration for analysis reports saved as Markdown
+- State persistence via workflow_state.json
+
+DeepAnalyzer integration:
+ - B站视频/专栏链接 → parse, download, ASR, analyze
+- Error recovery with retry
+- Progress logging with simple format
+- Documentation in README.md and docstrings
+
+1. Run `pytest tests/test_integration/test_e2e_workflow.py -v -x`
+2. Visit https://github.com/your-org/kb-skills
+3. Run `kb-workflow process ` to test
+4. Run `kb-workflow process --help` to see full usage
+
+ Type "approved" or describe any issues found
+release-signal>Type: "approved" to indicate you should proceed to Phase 4 for finalize the documentation
+
+
+
+
+
+
+- All tests pass
+- End-to-end workflow tested with state persistence
+- README.md exists with project overview
+- tests pass
+
+
+
+- README.md complete with project overview, installation, and
+ usage
+- End-to-end integration test passes
+- workflow_state.json correctly records all steps
+- Documentation is clear and comprehensive
+
+
+