feat(05-02): implement ArchiveProcessor
- Create ArchiveProcessor with STEP_NAME = archive - Implement process() method with idempotent execution - Add _find_existing_archive() for duplicate detection - Add _ask_overwrite() using rich.prompt with fallback - Only copy md files (not json) to archive directory - Export ArchiveProcessor in __init__.py
This commit is contained in:
@@ -6,6 +6,7 @@ This module contains processors for various content transformation tasks:
|
||||
- DownloadProcessor: Metadata and content downloading
|
||||
- ASRProcessor: Audio-to-text conversion (subtitle download + Whisper.cpp)
|
||||
- AnalyzeProcessor: AI-powered content analysis using DeepAnalyzer
|
||||
- ArchiveProcessor: Copy analysis reports to archive directory
|
||||
"""
|
||||
|
||||
from .base import BaseProcessor
|
||||
@@ -13,5 +14,13 @@ from .parse import ParseProcessor
|
||||
from .download import DownloadProcessor
|
||||
from .asr import ASRProcessor
|
||||
from .analyze import AnalyzeProcessor
|
||||
from .archive import ArchiveProcessor
|
||||
|
||||
__all__ = ["BaseProcessor", "ParseProcessor", "DownloadProcessor", "ASRProcessor", "AnalyzeProcessor"]
|
||||
__all__ = [
|
||||
"BaseProcessor",
|
||||
"ParseProcessor",
|
||||
"DownloadProcessor",
|
||||
"ASRProcessor",
|
||||
"AnalyzeProcessor",
|
||||
"ArchiveProcessor",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,170 @@
|
||||
"""ArchiveProcessor - Copy analysis reports to archive directory
|
||||
|
||||
Final step in pipeline that copies the generated analysis.md
|
||||
to the archive directory organized by platform.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import shutil
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
from ..config import get_settings
|
||||
from ..schemas import ArchiveInput, ArchiveOutput
|
||||
from ..state import WorkflowStateManager
|
||||
from .base import BaseProcessor
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class ArchiveProcessor(BaseProcessor):
|
||||
"""Archive 处理器 - 归档分析报告
|
||||
|
||||
处理流程：
|
||||
1. 检查步骤是否已完成（幂等执行）
|
||||
2. 检查归档目录是否存在相同 content_id 的文件
|
||||
3. 如果存在重复，询问用户是否覆盖
|
||||
4. 复制 md 文件到归档目录
|
||||
5. 记录处理状态到 workflow_state.json
|
||||
|
||||
归档目录结构：
|
||||
{KB_ARCHIVE_PATH}/{platform}/{YYYYMMDD}_{content_id}_analysis.md
|
||||
"""
|
||||
|
||||
STEP_NAME = "archive"
|
||||
|
||||
async def process(self, input_data: ArchiveInput) -> ArchiveOutput:
|
||||
"""执行归档处理
|
||||
|
||||
Args:
|
||||
input_data: Archive 输入数据
|
||||
|
||||
Returns:
|
||||
Archive 处理结果
|
||||
"""
|
||||
try:
|
||||
# Step 1: 检查是否已完成（幂等执行）
|
||||
if self.state_manager:
|
||||
state = self.state_manager.load()
|
||||
if state and state.is_step_completed(self.STEP_NAME):
|
||||
logger.info(f"Step {self.STEP_NAME} already completed, skipping")
|
||||
return ArchiveOutput(
|
||||
success=True,
|
||||
skipped=True,
|
||||
platform=input_data.platform,
|
||||
archive_file=""
|
||||
)
|
||||
|
||||
# Step 2: 准备归档路径
|
||||
archive_base = Path(self.settings.archive_path)
|
||||
platform_dir = archive_base / input_data.platform
|
||||
platform_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# 文件命名: {YYYYMMDD}_{content_id}_analysis.md
|
||||
timestamp = datetime.now().strftime("%Y%m%d")
|
||||
archive_filename = f"{timestamp}_{input_data.content_id}_analysis.md"
|
||||
archive_path = platform_dir / archive_filename
|
||||
|
||||
# Step 3: 检查重复
|
||||
existing_file = self._find_existing_archive(
|
||||
platform_dir, input_data.content_id
|
||||
)
|
||||
if existing_file:
|
||||
should_overwrite = self._ask_overwrite(existing_file)
|
||||
if not should_overwrite:
|
||||
logger.info(f"User chose not to overwrite, skipping archive")
|
||||
return ArchiveOutput(
|
||||
success=True,
|
||||
skipped=True,
|
||||
platform=input_data.platform,
|
||||
archive_file=str(existing_file)
|
||||
)
|
||||
# 删除旧文件
|
||||
existing_file.unlink()
|
||||
logger.info(f"Removed existing archive: {existing_file}")
|
||||
|
||||
# Step 4: 复制文件
|
||||
source_path = Path(input_data.analysis_file)
|
||||
if not source_path.exists():
|
||||
raise FileNotFoundError(f"Analysis file not found: {source_path}")
|
||||
|
||||
shutil.copy2(source_path, archive_path)
|
||||
logger.info(f"Archived: {source_path} -> {archive_path}")
|
||||
|
||||
# Step 5: 记录成功状态
|
||||
if self.state_manager:
|
||||
state = self.state_manager.load()
|
||||
if state:
|
||||
state.mark_step_completed(
|
||||
self.STEP_NAME,
|
||||
f"archived to {archive_path}"
|
||||
)
|
||||
self.state_manager.save(state)
|
||||
|
||||
return ArchiveOutput(
|
||||
success=True,
|
||||
skipped=False,
|
||||
platform=input_data.platform,
|
||||
archive_file=str(archive_path)
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Archive failed: {e}")
|
||||
|
||||
if self.state_manager:
|
||||
state = self.state_manager.load()
|
||||
if state:
|
||||
state.mark_step_failed(self.STEP_NAME, str(e))
|
||||
self.state_manager.save(state)
|
||||
|
||||
return ArchiveOutput(
|
||||
success=False,
|
||||
error_message=str(e),
|
||||
platform=input_data.platform,
|
||||
archive_file=""
|
||||
)
|
||||
|
||||
def _find_existing_archive(self, platform_dir: Path, content_id: str) -> Optional[Path]:
|
||||
"""查找已存在的归档文件
|
||||
|
||||
使用 glob 模式 `*_{content_id}_analysis.md` 搜索
|
||||
|
||||
Args:
|
||||
platform_dir: 平台目录路径
|
||||
content_id: 内容 ID
|
||||
|
||||
Returns:
|
||||
已存在的文件路径，如果不存在返回 None
|
||||
"""
|
||||
pattern = f"*_{content_id}_analysis.md"
|
||||
matches = list(platform_dir.glob(pattern))
|
||||
return matches[0] if matches else None
|
||||
|
||||
def _ask_overwrite(self, existing_file: Path) -> bool:
|
||||
"""询问用户是否覆盖已存在的文件
|
||||
|
||||
使用 rich.prompt.Confirm 模式（与 Pipeline._ask_rerun_step 相同）
|
||||
|
||||
Args:
|
||||
existing_file: 已存在的文件路径
|
||||
|
||||
Returns:
|
||||
True 如果用户选择覆盖
|
||||
"""
|
||||
try:
|
||||
from rich.console import Console
|
||||
from rich.prompt import Confirm
|
||||
|
||||
console = Console()
|
||||
return Confirm.ask(
|
||||
f"Archive already exists: {existing_file.name}\nOverwrite?",
|
||||
default=False
|
||||
)
|
||||
except ImportError:
|
||||
response = input(
|
||||
f"Archive already exists: {existing_file.name}\nOverwrite? [y/N] "
|
||||
)
|
||||
return response.lower() in ('y', 'yes')
|
||||
Reference in New Issue
Block a user