starmen1265
279ae684ea
- Move pipeline.py, schemas.py, config.py, state.py, errors.py - Move retry.py, idempotency.py, link_parser.py - Create pipeline/__init__.py with updated imports - Update all internal imports to use kb_workflow.pipeline prefix Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
81 lines
2.5 KiB
Python
81 lines
2.5 KiB
Python
"""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)
|