feat(05-02): integrate ArchiveProcessor into Pipeline

- Add ArchiveInput creation in Pipeline.run()
- Import ArchiveInput in pipeline.py
- Add archive_file and skipped to output field recording
- Add ArchiveProcessor to CLI processors list
- Update PipelineConfig steps to include archive
This commit is contained in:
2026-03-16 23:19:23 +08:00
parent 63dd88279c
commit d1497ca550
2 changed files with 21 additions and 4 deletions
+6 -2
View File
@@ -45,7 +45,10 @@ def process_command(
try: try:
from kb_pipeline.config import get_settings from kb_pipeline.config import get_settings
from kb_pipeline.pipeline import Pipeline from kb_pipeline.pipeline import Pipeline
from kb_pipeline.processors import ASRProcessor, AnalyzeProcessor, DownloadProcessor, ParseProcessor from kb_pipeline.processors import (
ASRProcessor, AnalyzeProcessor, ArchiveProcessor,
DownloadProcessor, ParseProcessor
)
from kb_pipeline.schemas import PipelineConfig from kb_pipeline.schemas import PipelineConfig
async def run_pipeline(): async def run_pipeline():
@@ -59,11 +62,12 @@ def process_command(
DownloadProcessor(), DownloadProcessor(),
ASRProcessor(), ASRProcessor(),
AnalyzeProcessor(), AnalyzeProcessor(),
ArchiveProcessor(),
] ]
# Create pipeline config # Create pipeline config
config = PipelineConfig( config = PipelineConfig(
steps=["parse", "download", "asr", "analyze"], steps=["parse", "download", "asr", "analyze", "archive"],
output_dir=out_dir, output_dir=out_dir,
skip_completed=skip_completed, skip_completed=skip_completed,
) )
+15 -2
View File
@@ -183,7 +183,7 @@ class Pipeline:
try: try:
# Create step input based on step type # Create step input based on step type
from kb_pipeline.schemas import ( from kb_pipeline.schemas import (
ParseInput, DownloadInput, ASRInput, AnalyzeInput ParseInput, DownloadInput, ASRInput, AnalyzeInput, ArchiveInput
) )
if step_name == "parse": if step_name == "parse":
@@ -228,6 +228,18 @@ class Pipeline:
content_type=parsed.content_type, content_type=parsed.content_type,
output_file=str(content_dir / "analysis.md"), output_file=str(content_dir / "analysis.md"),
) )
elif step_name == "archive":
# Archive needs analysis file from analyze step
analyze_result = results.get("analyze", {})
analysis_file = analyze_result.get("analysis_file", "")
step_input = ArchiveInput(
content_id=parsed.content_id,
platform=parsed.platform,
content_dir=str(content_dir),
analysis_file=analysis_file,
content_type=parsed.content_type,
)
else: else:
# Generic step input # Generic step input
from kb_pipeline.schemas import StepInput from kb_pipeline.schemas import StepInput
@@ -248,7 +260,8 @@ class Pipeline:
if step_output.success: if step_output.success:
# Add output fields to results # Add output fields to results
for field_name in ["metadata_file", "content_file", "title", "author", for field_name in ["metadata_file", "content_file", "title", "author",
"analysis_file", "summary", "output_files"]: "analysis_file", "summary", "output_files",
"archive_file", "skipped"]:
if hasattr(step_output, field_name): if hasattr(step_output, field_name):
results[step_name][field_name] = getattr(step_output, field_name) results[step_name][field_name] = getattr(step_output, field_name)