"""Idempotency Checker module Provides file existence checking to avoid reprocessing content. Checks for metadata JSON and content Markdown files. """ from pathlib import Path from typing import Tuple, List class IdempotencyChecker: """Checks if content has already been processed Determines if the output files for a given content already exist, enabling idempotent processing to avoid duplicate work. File naming convention: - Metadata: {platform}_{content_id}_metadata.json - Content: {platform}_{content_id}_content.md """ def check_exists( self, output_dir: Path, content_id: str, platform: str ) -> Tuple[bool, List[Path]]: """Check if output files already exist for this content Args: output_dir: Directory where output files are stored content_id: Unique content identifier (e.g., BV1GJ411x7Qw) platform: Platform name (e.g., bilibili, zhihu) Returns: Tuple of (all_exist, existing_files): - all_exist: True if both metadata and content files exist - existing_files: List of Path objects for existing files """ existing_files: List[Path] = [] if not content_id or not platform: return (False, existing_files) # Define expected file paths metadata_file = output_dir / f"{platform}_{content_id}_metadata.json" content_file = output_dir / f"{platform}_{content_id}_content.md" # Check if files exist if metadata_file.exists(): existing_files.append(metadata_file) if content_file.exists(): existing_files.append(content_file) # Return True only if both files exist all_exist = len(existing_files) == 2 return (all_exist, existing_files) def get_expected_files( self, output_dir: Path, content_id: str, platform: str ) -> Tuple[Path, Path]: """Get the expected file paths for a content Args: output_dir: Directory where output files are stored content_id: Unique content identifier platform: Platform name Returns: Tuple of (metadata_path, content_path) """ metadata_path = output_dir / f"{platform}_{content_id}_metadata.json" content_path = output_dir / f"{platform}_{content_id}_content.md" return (metadata_path, content_path)