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 <noreply@anthropic.com>
This commit is contained in:
@@ -1,9 +1,13 @@
|
|||||||
"""Processors module for KB Pipeline
|
"""Processors module for KB Pipeline
|
||||||
|
|
||||||
This module contains processors for various content transformation tasks:
|
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)
|
- ASRProcessor: Audio-to-text conversion (subtitle download + Whisper.cpp)
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
from .base import BaseProcessor
|
||||||
|
from .parse import ParseProcessor
|
||||||
from .asr import ASRProcessor
|
from .asr import ASRProcessor
|
||||||
|
|
||||||
__all__ = ["ASRProcessor"]
|
__all__ = ["BaseProcessor", "ParseProcessor", "ASRProcessor"]
|
||||||
|
|||||||
@@ -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
|
||||||
@@ -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
|
||||||
Reference in New Issue
Block a user