From 4b4ed9cab62659287a4b8ade46273af952522c0d Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Fri, 13 Mar 2026 22:50:28 +0800 Subject: [PATCH] docs(04-analysis-orchestration): create phase plans --- .coverage | Bin 0 -> 53248 bytes .../04-analysis-orchestration/04-01-PLAN.md | 273 ++++++++ .../04-analysis-orchestration/04-02-PLAN.md | 256 +++++++ .../04-analysis-orchestration/04-03-PLAN.md | 255 +++++++ .../04-analysis-orchestration/04-04-PLAN.md | 658 ++++++++++++++++++ .../04-analysis-orchestration/04-05-PLAN.md | 288 ++++++++ 6 files changed, 1730 insertions(+) create mode 100644 .coverage create mode 100644 .planning/phases/04-analysis-orchestration/04-01-PLAN.md create mode 100644 .planning/phases/04-analysis-orchestration/04-02-PLAN.md create mode 100644 .planning/phases/04-analysis-orchestration/04-03-PLAN.md create mode 100644 .planning/phases/04-analysis-orchestration/04-04-PLAN.md create mode 100644 .planning/phases/04-analysis-orchestration/04-05-PLAN.md diff --git a/.coverage b/.coverage new file mode 100644 index 0000000000000000000000000000000000000000..a283e8f307e8085d78a1bb94319e40de0975a097 GIT binary patch literal 53248 zcmeI)%WoS+90%~(NA22i*p#TULKHHWDv9dYHRahJNZO`QQ3+aF)C=9N?Mb}EyX)?* z^FTu2QmG*R1^x<7aH4nk12`a2Ap{i?2qY?y%7NePBev7HR6(dp^xeevK4xZTKJ)T= z6Tf-kyd4U@?D>ur^5blbDGFQQoH3T8$0$A0B}*%Mx-}(JQ?XK{!DbmoLKs4l> zMf+&hk^#$6e7R=Y4T=?t6-uJ*+m7XL@>Q`}iUK5O3&lnl@1R74z2eFcd^U+b$9=IZ zeBstb5a(jnZp`K9K72o)Esl>Xw?tGX%dgY#$UtdmN3SU7mMNjS=Qm{TYrf^yn<6N2 zYpaBHkMg+@#-;T=rxJR+WxG)^-M|iQ&*fr6)H|VQ>{!5(L`xUK>n{zZbxjuNS{ySu zSD#KV@(wYR;7R0V%AD(^ti6cx)D$_&v)gfMME*|4Y0Ju?81aciZ&=J>aYHXD>?7CB z4t3cp??82?Dm@i$9y*vUPERYhu15_;@|ZO{47OHs1C3^D_mHL|j}LD*vX=}t64{Gd zE$%HxT}H;{aph2J?(Z{|6vOgmN8?@+kaGH1Tbs-*<$>2)UTv_gMJ*-})In%0NP{hR z+`8;4fhtL& zDd(PEe&Yms+o?T;nu~%{_kP0*0-L5)Ic)_e(!G49X$5i=3TAlC^IF1ki z*18q&x-aNpGM7dfBeCZ8T*b+}N=MQ$G%n5DjP|g$Bdge~%4d0%d{DlX>rN}#;?bi@ zx0CeXcyKZUdMDsmX<3s^b)aQ!JJ@SvuhS?^Z;|>Nmjv%Kl$-Gxh< zKjg(UZPyM>lhTp*|Me&6{{I(7$C#%TED(SI1Rwwb2tWV=5P$##AOL|qEO18ejwrjl z0SG_<0uX=z1Rwwb2tWV=5P-lw6wtKMT=f6{ z#wX19)A-)_j8?Ef00Izz00bZa0SG_<0uX=z1R(H#3LMdOC3jv|znx$|sMnazRJt`k zMrY*lSylGI)3j06DA-j^UHI;+sRCn-DSn0hb%Nb~d*LMeEr0B1#x7q}*j?qD=`ZiG zg|0TPQutX_{hCtI=;t=Q7P5+-%RQ;yQKSF=H-2Qsed7<~7h1sr0SG_<0uX=z1Rwwb z2tWV=5P-lQ70A#>0E(_hFPav;sH#pM0Yvxznf#Ce3j`nl0SG_<0uX=z1Rwwb2tWV= z`&K~S|Ht+JzO7yq3<3~<00bZa0SG_<0uX=z1RxLz$p8Px^* +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 + + + +After completion, create `.planning/phases/04-analysis-orchestration/04-01-SUMMARY.md` + 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 + + + +After completion, create `.planning/phases/04-analysis-orchestration/04-02-SUMMARY.md` + 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 + + + +After completion, create `.planning/phases/04-analysis-orchestration/04-03-SUMMARY.md` + 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 + + + +After completion, create `.planning/phases/04-analysis-orchestration/04-04-SUMMARY.md` + 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 + + + +After completion, create `.planning/phases/04-analysis-orchestration/04-05-SUMMARY.md` +