From d1497ca55042d61bb294c4ee5aa2927ebf6eb44f Mon Sep 17 00:00:00 2001
From: starmen1265 <starmen1265@163.com>
Date: Mon, 16 Mar 2026 23:19:23 +0800
Subject: [PATCH] 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
---
 kb-workflow/kb_workflow.py |  8 ++++++--
 kb_pipeline/pipeline.py    | 17 +++++++++++++++--
 2 files changed, 21 insertions(+), 4 deletions(-)

diff --git a/kb-workflow/kb_workflow.py b/kb-workflow/kb_workflow.py
index 73775cc..8915617 100644
--- a/kb-workflow/kb_workflow.py
+++ b/kb-workflow/kb_workflow.py
@@ -45,7 +45,10 @@ def process_command(
     try:
         from kb_pipeline.config import get_settings
         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
 
         async def run_pipeline():
@@ -59,11 +62,12 @@ def process_command(
                 DownloadProcessor(),
                 ASRProcessor(),
                 AnalyzeProcessor(),
+                ArchiveProcessor(),
             ]
 
             # Create pipeline config
             config = PipelineConfig(
-                steps=["parse", "download", "asr", "analyze"],
+                steps=["parse", "download", "asr", "analyze", "archive"],
                 output_dir=out_dir,
                 skip_completed=skip_completed,
             )
diff --git a/kb_pipeline/pipeline.py b/kb_pipeline/pipeline.py
index 439b330..bd979e4 100644
--- a/kb_pipeline/pipeline.py
+++ b/kb_pipeline/pipeline.py
@@ -183,7 +183,7 @@ class Pipeline:
             try:
                 # Create step input based on step type
                 from kb_pipeline.schemas import (
-                    ParseInput, DownloadInput, ASRInput, AnalyzeInput
+                    ParseInput, DownloadInput, ASRInput, AnalyzeInput, ArchiveInput
                 )
 
                 if step_name == "parse":
@@ -228,6 +228,18 @@ class Pipeline:
                         content_type=parsed.content_type,
                         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:
                     # Generic step input
                     from kb_pipeline.schemas import StepInput
@@ -248,7 +260,8 @@ class Pipeline:
                 if step_output.success:
                     # Add output fields to results
                     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):
                             results[step_name][field_name] = getattr(step_output, field_name)