From efddef98d312797d0d2d1b7d245d062da233c884 Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Fri, 13 Mar 2026 23:11:44 +0800 Subject: [PATCH] feat(04-01): add ParseInput and ParseOutput schemas - ParseInput extends StepInput with url field - ParseOutput extends StepOutput with platform, content_type, content_id, resolved_url, original_url - Tests for schema validation and inheritance Co-Authored-By: Claude Opus 4.6 --- kb_pipeline/schemas.py | 40 ++++++++++++ tests/test_pipeline/test_schemas.py | 97 +++++++++++++++++++++++++++++ 2 files changed, 137 insertions(+) diff --git a/kb_pipeline/schemas.py b/kb_pipeline/schemas.py index 1d3011f..7e5a5cc 100644 --- a/kb_pipeline/schemas.py +++ b/kb_pipeline/schemas.py @@ -233,3 +233,43 @@ class ASROutput(StepOutput): speakers: Optional[List[str]] = Field( None, description="说话者列表（仅说话者分离时有值）" ) + + +# ============================================================================= +# Parse Schemas (Phase 04) +# ============================================================================= + + +class ParseInput(StepInput): + """Parse 步骤输入 + + 继承 StepInput 的基础字段，添加 URL 字段。 + content_id 和 platform 在解析前是空的，解析后会被填充。 + + Attributes: + url: 要解析的 URL + content_id: 解析后填充的内容 ID（初始为空） + platform: 解析后填充的平台名称（初始为空） + """ + url: str = Field(..., min_length=1, description="URL to parse") + 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 步骤输出 + + 继承 StepOutput 的基础字段，添加解析特定字段。 + + Attributes: + platform: 检测到的平台 (bilibili, zhihu) + content_type: 内容类型 (video, article, answer, zhuanlan) + content_id: 提取的内容 ID + resolved_url: 短链解析后的完整 URL + original_url: 用户提供的原始 URL + """ + 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") diff --git a/tests/test_pipeline/test_schemas.py b/tests/test_pipeline/test_schemas.py index 0e200f5..8a37806 100644 --- a/tests/test_pipeline/test_schemas.py +++ b/tests/test_pipeline/test_schemas.py @@ -472,3 +472,100 @@ class TestJSONSerialization: restored = WorkflowState.model_validate_json(json_str) assert "fetch_metadata" in restored.steps assert restored.steps["fetch_metadata"].status == StepStatus.COMPLETED + + +# ============================================================================= +# Parse Schemas Tests (Phase 04-01) +# ============================================================================= + + +class TestParseInput: + """测试 ParseInput Schema""" + + def test_parse_input_validates_url_field(self): + """ParseInput 必须包含 url 字段，且 min_length=1""" + from kb_pipeline.schemas import ParseInput + + # 缺少 url + with pytest.raises(ValidationError) as exc_info: + ParseInput() + assert "url" in str(exc_info.value) + + # 空 url + with pytest.raises(ValidationError): + ParseInput(url="") + + # 有效 url + input_data = ParseInput(url="https://www.bilibili.com/video/BV1xx") + assert input_data.url == "https://www.bilibili.com/video/BV1xx" + + def test_parse_input_content_id_default_empty(self): + """ParseInput content_id 默认为空字符串""" + from kb_pipeline.schemas import ParseInput + + input_data = ParseInput(url="https://www.bilibili.com/video/BV1xx") + assert input_data.content_id == "" + + def test_parse_input_platform_default_empty(self): + """ParseInput platform 默认为空字符串""" + from kb_pipeline.schemas import ParseInput + + input_data = ParseInput(url="https://www.bilibili.com/video/BV1xx") + assert input_data.platform == "" + + +class TestParseOutput: + """测试 ParseOutput Schema""" + + def test_parse_output_inherits_step_output(self): + """ParseOutput 继承 StepOutput 的 success 和 error_message""" + from kb_pipeline.schemas import ParseOutput, StepOutput + + output = ParseOutput( + success=True, + platform="bilibili", + content_type="video", + content_id="BV1xx", + resolved_url="https://www.bilibili.com/video/BV1xx", + original_url="https://b23.tv/abc" + ) + + assert isinstance(output, StepOutput) + assert output.success is True + assert output.error_message is None + + def test_parse_output_has_required_fields(self): + """ParseOutput 包含所有必需字段""" + from kb_pipeline.schemas import ParseOutput + + output = ParseOutput( + success=True, + platform="bilibili", + content_type="video", + content_id="BV1xx", + resolved_url="https://www.bilibili.com/video/BV1xx", + original_url="https://b23.tv/abc" + ) + + assert output.platform == "bilibili" + assert output.content_type == "video" + assert output.content_id == "BV1xx" + assert output.resolved_url == "https://www.bilibili.com/video/BV1xx" + assert output.original_url == "https://b23.tv/abc" + + def test_parse_output_with_error(self): + """ParseOutput 可以表示失败状态""" + from kb_pipeline.schemas import ParseOutput + + output = ParseOutput( + success=False, + error_message="Unsupported URL format", + platform="", + content_type="", + content_id="", + resolved_url="", + original_url="https://unknown.com/123" + ) + + assert output.success is False + assert output.error_message == "Unsupported URL format"