starmen1265
fdca690bd6
- Move all processor implementations (parse, download, asr, analyze, archive) - Move all platform adapters (base, bilibili, zhihu) - Update all internal imports to use kb_workflow.pipeline prefix Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
84 lines
2.5 KiB
Python
84 lines
2.5 KiB
Python
"""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_workflow.pipeline.config import get_settings
|
|
from kb_workflow.pipeline.link_parser import LinkParser
|
|
from kb_workflow.pipeline.schemas import ParseInput, ParseOutput
|
|
from kb_workflow.pipeline.processors.base import BaseProcessor
|
|
|
|
if TYPE_CHECKING:
|
|
from kb_workflow.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
|