From 06d36fc440806b9a2ec75096c77385a71e565c97 Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Fri, 13 Mar 2026 23:14:32 +0800 Subject: [PATCH] feat(04-01): implement ParseProcessor and update module exports - ParseProcessor wraps LinkParser for URL parsing - Returns ParseOutput with platform, content_type, content_id - Records state when state_manager is provided - Handles unknown URLs with success=False - Export BaseProcessor, ParseProcessor, ASRProcessor from __init__.py Co-Authored-By: Claude Opus 4.6 --- kb_pipeline/processors/__init__.py | 6 +- kb_pipeline/processors/parse.py | 83 +++++++++++++ tests/test_processors/test_parse.py | 184 ++++++++++++++++++++++++++++ 3 files changed, 272 insertions(+), 1 deletion(-) create mode 100644 kb_pipeline/processors/parse.py create mode 100644 tests/test_processors/test_parse.py diff --git a/kb_pipeline/processors/__init__.py b/kb_pipeline/processors/__init__.py index be2f254..8bb4bc0 100644 --- a/kb_pipeline/processors/__init__.py +++ b/kb_pipeline/processors/__init__.py @@ -1,9 +1,13 @@ """Processors module for KB Pipeline This module contains processors for various content transformation tasks: +- BaseProcessor: Abstract base class for all processors +- ParseProcessor: URL parsing for various platforms - ASRProcessor: Audio-to-text conversion (subtitle download + Whisper.cpp) """ +from .base import BaseProcessor +from .parse import ParseProcessor from .asr import ASRProcessor -__all__ = ["ASRProcessor"] +__all__ = ["BaseProcessor", "ParseProcessor", "ASRProcessor"] diff --git a/kb_pipeline/processors/parse.py b/kb_pipeline/processors/parse.py new file mode 100644 index 0000000..6164a25 --- /dev/null +++ b/kb_pipeline/processors/parse.py @@ -0,0 +1,83 @@ +"""ParseProcessor - URL parsing processor + +Parses URLs from various platforms (Bilibili, Zhihu) and extracts +platform, content type, and content ID information. +""" + +from __future__ import annotations + +from typing import Optional, TYPE_CHECKING + +from kb_pipeline.config import get_settings +from kb_pipeline.link_parser import LinkParser +from kb_pipeline.schemas import ParseInput, ParseOutput +from .base import BaseProcessor + +if TYPE_CHECKING: + from kb_pipeline.state import WorkflowStateManager + + +class ParseProcessor(BaseProcessor): + """URL parsing processor. + + Parses URLs from Bilibili and Zhihu platforms, extracting + platform, content type, and content ID. + + Attributes: + STEP_NAME: "parse" - identifies this processor step + """ + + STEP_NAME = "parse" + + async def process(self, input_data: ParseInput) -> ParseOutput: + """Parse the URL and extract platform information. + + Args: + input_data: ParseInput with URL to parse + + Returns: + ParseOutput with parsed information or error message + """ + parser = LinkParser() + parsed = parser.parse(input_data.url) + + if parsed is None: + # Unknown URL format + result = ParseOutput( + success=False, + error_message="Unsupported URL format", + platform="", + content_type="", + content_id="", + resolved_url="", + original_url=input_data.url + ) + + # Record failure state + if self.state_manager: + state = self.state_manager.load() + if state: + state.mark_step_failed(self.STEP_NAME, result.error_message) + self.state_manager.save(state) + + return result + + # Successful parse + result = ParseOutput( + success=True, + platform=parsed.platform, + content_type=parsed.content_type, + content_id=parsed.content_id, + resolved_url=parsed.resolved_url, + original_url=parsed.original_url + ) + + # Record success state + if self.state_manager: + state = self.state_manager.load() + if state: + summary = f"platform={parsed.platform}, type={parsed.content_type}, id={parsed.content_id}" + state.mark_step_completed(self.STEP_NAME, summary) + self.state_manager.save(state) + + return result diff --git a/tests/test_processors/test_parse.py b/tests/test_processors/test_parse.py new file mode 100644 index 0000000..3cbd41b --- /dev/null +++ b/tests/test_processors/test_parse.py @@ -0,0 +1,184 @@ +"""Tests for ParseProcessor + +Tests that ParseProcessor: +- Parses Bilibili video URLs correctly +- Parses Zhihu answer URLs correctly +- Returns ParseOutput with success=False for unknown URLs +- Records state when state_manager is provided +- Can be imported from kb_pipeline.processors +""" + +import pytest +from unittest.mock import MagicMock, patch, AsyncMock + + +class TestParseProcessorBilibili: + """ParseProcessor Bilibili URL 测试""" + + @pytest.mark.asyncio + async def test_parse_bilibili_video_url(self): + """ParseProcessor 正确解析 Bilibili 视频 URL""" + from kb_pipeline.processors.parse import ParseProcessor + from kb_pipeline.schemas import ParseInput + + with patch("kb_pipeline.processors.base.get_settings") as mock_settings: + mock_settings.return_value = MagicMock() + + processor = ParseProcessor() + input_data = ParseInput(url="https://www.bilibili.com/video/BV1xx411c7mD") + + result = await processor.process(input_data) + + assert result.success is True + assert result.platform == "bilibili" + assert result.content_type == "video" + assert result.content_id == "BV1xx411c7mD" + assert result.original_url == "https://www.bilibili.com/video/BV1xx411c7mD" + + @pytest.mark.asyncio + async def test_parse_bilibili_article_url(self): + """ParseProcessor 正确解析 Bilibili 专栏 URL""" + from kb_pipeline.processors.parse import ParseProcessor + from kb_pipeline.schemas import ParseInput + + with patch("kb_pipeline.processors.base.get_settings") as mock_settings: + mock_settings.return_value = MagicMock() + + processor = ParseProcessor() + input_data = ParseInput(url="https://www.bilibili.com/read/cv12345678") + + result = await processor.process(input_data) + + assert result.success is True + assert result.platform == "bilibili" + assert result.content_type == "article" + assert result.content_id == "cv12345678" + + +class TestParseProcessorZhihu: + """ParseProcessor Zhihu URL 测试""" + + @pytest.mark.asyncio + async def test_parse_zhihu_answer_url(self): + """ParseProcessor 正确解析 Zhihu 回答 URL""" + from kb_pipeline.processors.parse import ParseProcessor + from kb_pipeline.schemas import ParseInput + + with patch("kb_pipeline.processors.base.get_settings") as mock_settings: + mock_settings.return_value = MagicMock() + + processor = ParseProcessor() + input_data = ParseInput(url="https://www.zhihu.com/question/123456/answer/789012") + + result = await processor.process(input_data) + + assert result.success is True + assert result.platform == "zhihu" + assert result.content_type == "answer" + assert result.content_id == "789012" + + @pytest.mark.asyncio + async def test_parse_zhihu_zhuanlan_url(self): + """ParseProcessor 正确解析 Zhihu 专栏 URL""" + from kb_pipeline.processors.parse import ParseProcessor + from kb_pipeline.schemas import ParseInput + + with patch("kb_pipeline.processors.base.get_settings") as mock_settings: + mock_settings.return_value = MagicMock() + + processor = ParseProcessor() + input_data = ParseInput(url="https://zhuanlan.zhihu.com/p/12345678") + + result = await processor.process(input_data) + + assert result.success is True + assert result.platform == "zhihu" + assert result.content_type == "zhuanlan" + assert result.content_id == "12345678" + + +class TestParseProcessorUnknown: + """ParseProcessor 未知 URL 测试""" + + @pytest.mark.asyncio + async def test_parse_unknown_url_returns_failure(self): + """ParseProcessor 对未知 URL 返回 success=False""" + from kb_pipeline.processors.parse import ParseProcessor + from kb_pipeline.schemas import ParseInput + + with patch("kb_pipeline.processors.base.get_settings") as mock_settings: + mock_settings.return_value = MagicMock() + + processor = ParseProcessor() + input_data = ParseInput(url="https://www.youtube.com/watch?v=abc123") + + result = await processor.process(input_data) + + assert result.success is False + assert "Unsupported" in result.error_message or "unsupported" in result.error_message.lower() + + +class TestParseProcessorStateRecording: + """ParseProcessor 状态记录测试""" + + @pytest.mark.asyncio + async def test_state_recording_on_success(self): + """ParseProcessor 成功时记录状态""" + from kb_pipeline.processors.parse import ParseProcessor + from kb_pipeline.schemas import ParseInput, WorkflowState + + mock_state_manager = MagicMock() + mock_state = MagicMock(spec=WorkflowState) + mock_state_manager.load.return_value = mock_state + + with patch("kb_pipeline.processors.base.get_settings") as mock_settings: + mock_settings.return_value = MagicMock() + + processor = ParseProcessor(state_manager=mock_state_manager) + input_data = ParseInput(url="https://www.bilibili.com/video/BV1xx") + + result = await processor.process(input_data) + + assert result.success is True + mock_state.mark_step_completed.assert_called_once() + call_args = mock_state.mark_step_completed.call_args + assert call_args[0][0] == "parse" + + @pytest.mark.asyncio + async def test_no_state_recording_without_manager(self): + """无 state_manager 时不记录状态""" + from kb_pipeline.processors.parse import ParseProcessor + from kb_pipeline.schemas import ParseInput + + with patch("kb_pipeline.processors.base.get_settings") as mock_settings: + mock_settings.return_value = MagicMock() + + processor = ParseProcessor(state_manager=None) + input_data = ParseInput(url="https://www.bilibili.com/video/BV1xx") + + result = await processor.process(input_data) + + assert result.success is True + # 不应该抛异常 + + +class TestParseProcessorModuleExport: + """ParseProcessor 模块导出测试""" + + def test_parse_processor_importable_from_processors(self): + """ParseProcessor 可以从 kb_pipeline.processors 导入""" + from kb_pipeline.processors import ParseProcessor + + assert ParseProcessor is not None + + def test_base_processor_importable_from_processors(self): + """BaseProcessor 可以从 kb_pipeline.processors 导入""" + from kb_pipeline.processors import BaseProcessor + + assert BaseProcessor is not None + + def test_asr_processor_still_importable(self): + """ASRProcessor 仍然可以从 kb_pipeline.processors 导入""" + from kb_pipeline.processors import ASRProcessor + + assert ASRProcessor is not None