From ff367d1e37aac8075692704036278a573184fdcd Mon Sep 17 00:00:00 2001 From: HERITAGE-XION Date: Tue, 6 Jan 2026 00:22:04 +0100 Subject: [PATCH] metadata-scrubber-tool: implemented batch processing of file. optimized logs to only show when the verbose flag is used. added docstring documentation. png processing not fully supported yet. --- src/commands/read.py | 12 +- src/commands/scrub.py | 141 +++++++++++++++++ src/core/jpeg_metadata.py | 58 ++++++- src/core/png_metadata.py | 56 ++++++- src/main.py | 17 +- src/services/batch_processor.py | 257 +++++++++++++++++++++++++++++++ src/services/image_handler.py | 58 +++++-- src/services/metadata_factory.py | 33 ++++ src/services/metadata_handler.py | 54 ++++++- src/utils/display.py | 62 +++++++- src/utils/exceptions.py | 16 +- src/utils/formatter.py | 17 +- src/utils/get_target_files.py | 29 +++- src/utils/logger.py | 22 ++- 14 files changed, 783 insertions(+), 49 deletions(-) create mode 100644 src/commands/scrub.py create mode 100644 src/services/batch_processor.py diff --git a/src/commands/read.py b/src/commands/read.py index d200824f..eb50d502 100644 --- a/src/commands/read.py +++ b/src/commands/read.py @@ -1,3 +1,10 @@ +""" +Read command - Display metadata from files. + +This command reads and displays metadata from image files in a +formatted table view. Supports single files or recursive directory processing. +""" + import logging from pathlib import Path @@ -19,7 +26,6 @@ def get_metadata( file_okay=True, # Can be a file dir_okay=True, # Can be a directory readable=True, # Must be readable (permissions check) - writable=True, # Must be writable (permissions check) resolve_path=True, # Auto-convert to absolute path help="The path to the file you want to process", ), @@ -39,7 +45,9 @@ def get_metadata( # Read console.print(f"🔎 Processing [bold cyan]{file.name}[/bold cyan]...") current_data = handler.read() - log.info(f"Successfully read metadata from {file_path.name}") + if log.isEnabledFor(logging.DEBUG): + # if verbose mode is enabled, log the Info + log.info(f"Successfully read metadata from {file_path.name}") print_metadata_table(current_data) except Exception as e: diff --git a/src/commands/scrub.py b/src/commands/scrub.py new file mode 100644 index 00000000..17d59714 --- /dev/null +++ b/src/commands/scrub.py @@ -0,0 +1,141 @@ +""" +Scrub command - Remove metadata from files. + +This command processes files through the read→wipe→save pipeline, +removing privacy-sensitive metadata like EXIF, GPS, and author info. + +Supports concurrent processing for efficient batch operations on large file sets. +""" + +import logging +import os +from pathlib import Path + +import typer +from rich.console import Console +from rich.progress import ( + BarColumn, + MofNCompleteColumn, + Progress, + SpinnerColumn, + TaskProgressColumn, + TextColumn, + TimeElapsedColumn, +) + +from src.services.batch_processor import BatchProcessor +from src.utils.display import print_batch_summary +from src.utils.get_target_files import get_target_files + +console = Console() +log = logging.getLogger("metadata-scrubber") + + +# fmt: off +def scrub( + file_path: Path = typer.Argument( + exists=True, + file_okay=True, + dir_okay=True, + readable=True, + writable=True, + resolve_path=True, + help="The file or directory to process.", + ), + recursive: bool = typer.Option( + False, "--recursive", "-r", + help="Recursively process files in the specified directory." + ), + ext: str = typer.Option( + None, "--extension", "-ext", + help="File extension to filter by (e.g., jpg, png)." + ), + output_dir: str = typer.Option( + "./scrubbed", "--output", "-o", + help="Directory to save processed files." + ), + dry_run: bool = typer.Option( + False, "--dry-run", "-d", + help="Preview what would be processed without making changes." + ), + workers: int = typer.Option( + min(4, (os.cpu_count() or 1)), + "--workers", "-w", + help="Number of concurrent worker threads (default: 4 or CPU count)." + ), +): + # fmt: on + """ + Remove metadata from files. + + Scrubs privacy-sensitive metadata (EXIF, GPS, author info) from images. + Works with JPEG, PNG, and will support PDF/Office docs in future. + + Examples: + + scrub photo.jpg + + scrub ./photos/ -r -ext jpg --output ./cleaned + + scrub ./folder/ -r -ext png --dry-run + + scrub ./large_batch/ -r -ext jpg --workers 8 + """ + # Validate recursive/extension combo + if recursive and not ext: + raise typer.BadParameter( + "If you provide --recursive or -r, you must also provide --extension or -ext." + ) + if ext and not recursive: + raise typer.BadParameter( + "If you provide --extension or -ext, you must also provide --recursive or -r." + ) + + # Show dry-run banner + if dry_run: + console.print("\n[bold yellow]🔍 DRY-RUN MODE[/bold yellow] - No files will be modified.\n") + + # Collect files to process + files = list(get_target_files(file_path, ext)) if recursive else [file_path] + + if not files: + console.print("[yellow]No files found to process.[/yellow]") + raise typer.Exit(0) + + # Show worker count for batch operations + if len(files) > 1: + console.print(f"[dim]Processing {len(files)} files with {workers} workers...[/dim]\n") + + # Initialize processor with worker count + processor = BatchProcessor(output_dir=output_dir, dry_run=dry_run, max_workers=workers) + + # Process with thread-safe progress bar + with Progress( + SpinnerColumn(), + TextColumn("[progress.description]{task.description}"), + BarColumn(), + MofNCompleteColumn(), + TaskProgressColumn(), + TimeElapsedColumn(), + console=console, + refresh_per_second=10, # Smooth updates for concurrent processing + ) as progress: + task_id = progress.add_task( + "[cyan]Scrubbing metadata...", + total=len(files), + ) + + def on_file_complete(result): + """Callback for progress updates from concurrent workers.""" + status = "✓" if result.success else "✗" + progress.update( + task_id, + description=f"[cyan]{status} {result.filepath.name}", + advance=1, + ) + + # Use concurrent batch processing + processor.process_batch(files, progress_callback=on_file_complete) + + # Display summary + print_batch_summary(processor.get_summary()) diff --git a/src/core/jpeg_metadata.py b/src/core/jpeg_metadata.py index 9e382851..1b4f3c41 100644 --- a/src/core/jpeg_metadata.py +++ b/src/core/jpeg_metadata.py @@ -1,24 +1,55 @@ +""" +JPEG metadata processor using piexif. + +This module provides the JpegProcessor class which handles EXIF metadata +extraction and manipulation for JPEG images using the piexif library. +""" + import piexif # pyright: ignore[reportMissingTypeStubs] from src.utils.exceptions import MetadataNotFoundError, MetadataProcessingError -class JpegProcessaor: +class JpegProcessor: + """ + Processor for JPEG image metadata. + + Handles reading, extracting, and deleting EXIF metadata from JPEG files. + Preserves essential tags (Orientation, ColorSpace) needed for proper display. + + Attributes: + tags_to_delete: List of EXIF tag IDs to remove. + data: Dict of extracted metadata with human-readable keys. + """ + def __init__(self): + """Initialize the JPEG processor with empty data structures.""" self.tags_to_delete = [] self.data = {} def get_metadata(self, img): + """ + Extract EXIF metadata from a JPEG image. + + Args: + img: PIL Image object with EXIF data. + + Returns: + Dict with 'data' (metadata dict) and 'tags_to_delete' (tag IDs list). + + Raises: + MetadataNotFoundError: If no EXIF data is found in the image. + """ if "exif" not in img.info: raise MetadataNotFoundError("No EXIF data found in the image.") exif_dict = piexif.load(img.info["exif"]) for ifd, value in exif_dict.items(): - # this exclude thumbnail IFD (its the thumbnails blob data. it can be removed but the image will take a couple more seconds to load) + # Exclude thumbnail IFD (blob data that slows loading if removed) if not isinstance(exif_dict[ifd], dict): continue - # iterate through the IFD + # Iterate through the IFD for tag, tag_value in exif_dict[ifd].items(): tag_info = piexif.TAGS[ifd].get(tag, {}) @@ -27,7 +58,7 @@ class JpegProcessaor: tag_info.get("name", "Unknown Tag") if tag_info else "Unknown Tag" ) - # exculudes tags that are necessary for image display integrity + # Exclude tags necessary for image display integrity if ( tag_name == "Orientation" or tag_name == "ColorSpace" @@ -35,21 +66,34 @@ class JpegProcessaor: ): continue - # save to list and dict + # Save to list and dict self.tags_to_delete.append(tag) self.data[tag_name] = tag_value return {"data": self.data, "tags_to_delete": self.tags_to_delete} def delete_metadata(self, img, tags_to_delete): + """ + Remove specified EXIF tags from a JPEG image. + + Args: + img: PIL Image object with EXIF data. + tags_to_delete: List of tag IDs to remove. + + Returns: + Modified EXIF dictionary with specified tags removed. + + Raises: + MetadataProcessingError: If an error occurs during processing. + """ try: exif_dict = piexif.load(img.info["exif"]) for ifd, value in exif_dict.items(): - # exclude thumbnail IFD (its the thumbnails blob data so i dont wanna deal with that) + # Exclude thumbnail IFD if not isinstance(exif_dict[ifd], dict): continue - # iterate through and delete tags + # Iterate through and delete tags for tag in list(exif_dict[ifd]): if tag in tags_to_delete: del exif_dict[ifd][tag] diff --git a/src/core/png_metadata.py b/src/core/png_metadata.py index 2e0f61d0..8d00fd08 100644 --- a/src/core/png_metadata.py +++ b/src/core/png_metadata.py @@ -1,37 +1,68 @@ +""" +PNG metadata processor using PIL. + +This module provides the PngProcessor class which handles EXIF metadata +extraction and manipulation for PNG images using PIL's built-in EXIF support. +""" + from PIL import ExifTags from src.utils.exceptions import MetadataNotFoundError, MetadataProcessingError class PngProcessor: + """ + Processor for PNG image metadata. + + Handles reading, extracting, and deleting EXIF metadata from PNG files. + Processes both standard EXIF tags and GPS IFD data. + + Attributes: + tags_to_delete: List of EXIF tag IDs to remove. + data: Dict of extracted metadata with human-readable keys. + """ + def __init__(self): + """Initialize the PNG processor with empty data structures.""" self.tags_to_delete = [] self.data = {} def get_metadata(self, img): + """ + Extract EXIF metadata from a PNG image. + + Args: + img: PIL Image object. + + Returns: + Dict with 'data' (metadata dict) and 'tags_to_delete' (tag IDs list). + + Raises: + MetadataNotFoundError: If no EXIF data is found in the image. + """ img.load() exif = img.getexif() if not exif: raise MetadataNotFoundError("No EXIF data found in the image.") - # iterate through the (0th) IFD + # Iterate through the (0th) IFD for tag, value in exif.items(): # Get the human-readable name for the tag tag_name = ExifTags.TAGS.get(tag, tag) - # save to list and dict + # Save to list and dict self.tags_to_delete.append(tag) self.data[tag_name] = value print(f"{tag_name}: {value}") - # iterate through the (GPS) IFD + # Iterate through the (GPS) IFD gps_ifd = exif.get_ifd(ExifTags.IFD.GPSInfo) for tag, value in gps_ifd.items(): # Get the human-readable name for the tag tag_name = ExifTags.GPSTAGS.get(tag, tag) - # save to list and dict + # Save to list and dict self.tags_to_delete.append(tag) self.data[tag_name] = value print(f"{tag_name}: {value}") @@ -39,15 +70,28 @@ class PngProcessor: return {"data": self.data, "tags_to_delete": self.tags_to_delete} def delete_metadata(self, img, tags_to_delete): + """ + Remove specified EXIF tags from a PNG image. + + Args: + img: PIL Image object. + tags_to_delete: List of tag IDs to remove. + + Returns: + Modified EXIF data with specified tags removed. + + Raises: + MetadataProcessingError: If an error occurs during processing. + """ img.load() exif = img.getexif() try: - # iterate through the (0th) IFD + # Iterate through the (0th) IFD for tag_id, value in exif.items(): if tag_id in tags_to_delete: del exif[tag_id] - # terate through the (GPS) IFD + # Iterate through the (GPS) IFD gps_ifd = exif.get_ifd(ExifTags.IFD.GPSInfo) for tag_id, value in gps_ifd.items(): if tag_id in tags_to_delete: diff --git a/src/main.py b/src/main.py index 05341901..86f106b2 100644 --- a/src/main.py +++ b/src/main.py @@ -1,11 +1,23 @@ +""" +Metadata Scrubber Tool - CLI Application Entry Point. + +This module serves as the main entry point for the CLI application. +It initializes the Typer app, registers commands, and configures logging. + +Commands: + read: Display metadata from files. + scrub: Remove metadata from files. +""" + import logging import typer from src.commands.read import get_metadata +from src.commands.scrub import scrub from src.utils.logger import setup_logging -# Initialize the app and the console +# Initialize the Typer app with helpful defaults app = typer.Typer(no_args_is_help=True, pretty_exceptions_show_locals=False) log = logging.getLogger("metadata-scrubber") @@ -31,7 +43,8 @@ def main( # fmt: on # register commands -app.command(name="get-metadata")(get_metadata) +app.command(name="read")(get_metadata) +app.command(name="scrub")(scrub) # run app if __name__ == "__main__": diff --git a/src/services/batch_processor.py b/src/services/batch_processor.py new file mode 100644 index 00000000..de90d503 --- /dev/null +++ b/src/services/batch_processor.py @@ -0,0 +1,257 @@ +""" +Batch processing service for metadata operations. + +This module provides handler-agnostic batch processing that works with any +MetadataHandler subclass (images now, PDF/Office docs in future). + +Supports concurrent processing via ThreadPoolExecutor for efficient handling +of large batches (1000+ files). +""" + +import logging +import threading +from concurrent.futures import ThreadPoolExecutor, as_completed +from dataclasses import dataclass, field +from pathlib import Path +from typing import Callable, Iterable, List, Optional + +from rich.console import Console + +from src.services.metadata_factory import MetadataFactory + +log = logging.getLogger("metadata-scrubber") +console = Console() + + +@dataclass +class FileResult: + """Result of processing a single file.""" + + filepath: Path + success: bool + action: str # "scrubbed", "skipped", "dry-run" + output_path: Optional[Path] = None + error: Optional[str] = None + + +@dataclass +class BatchSummary: + """Aggregated statistics for batch processing.""" + + total: int = 0 + success: int = 0 + skipped: int = 0 + failed: int = 0 + dry_run: bool = False + output_dir: Optional[Path] = None + results: List[FileResult] = field(default_factory=list) + + +class BatchProcessor: + """ + Handler-agnostic batch processor for metadata operations. + + Works with any MetadataHandler subclass via MetadataFactory. + Supports dry-run mode, automatic duplicate suffix handling, + and concurrent processing via ThreadPoolExecutor. + """ + + def __init__( + self, + output_dir: Optional[str] = None, + dry_run: bool = False, + max_workers: int = 4, + ): + """ + Initialize the batch processor. + + Args: + output_dir: Directory to save processed files. Defaults to "./scrubbed". + dry_run: If True, preview what would be processed without writing files. + max_workers: Maximum number of concurrent worker threads. Defaults to 4. + """ + self.output_dir = Path(output_dir) if output_dir else Path("./scrubbed") + self.dry_run = dry_run + self.max_workers = max_workers + self.results: List[FileResult] = [] + + # Thread synchronization + self._path_lock = threading.Lock() # Protects unique path generation + self._results_lock = threading.Lock() # Protects results list + + def process_file(self, file: Path) -> FileResult: + """ + Process a single file through the read→wipe→save pipeline. + + Uses MetadataFactory to get the appropriate handler, so this method + automatically works with any file type that has a registered handler. + + Args: + file: Path to the file to process. + + Returns: + FileResult with success status and details. + """ + try: + # Dry-run mode: just report what would happen + if self.dry_run: + # Verify the file can be handled (will raise if not) + MetadataFactory.get_handler(str(file)) + output_path = self._get_unique_output_path(file) + result = FileResult( + filepath=file, + success=True, + action="dry-run", + output_path=output_path, + ) + self._append_result(result) + log.debug(f"[DRY-RUN] Would process: {file}") + return result + + # Get handler from factory + handler = MetadataFactory.get_handler(str(file)) + + # Execute the read → wipe → save pipeline + handler.read() + handler.wipe() + + output_path = self._get_unique_output_path(file) + handler.save(str(output_path)) + + result = FileResult( + filepath=file, + success=True, + action="scrubbed", + output_path=output_path, + ) + self._append_result(result) + if log.isEnabledFor(logging.DEBUG): + # if verbose mode is enabled, log the Info + log.info(f"✅ Scrubbed: {file.name} → {output_path}") + return result + + except Exception as e: + result = FileResult( + filepath=file, + success=False, + action="skipped", + error=str(e), + ) + self._append_result(result) + if log.isEnabledFor(logging.DEBUG): + # if verbose mode is enabled, log the traceback + log.warning(f"⚠️ Skipped {file.name}: {e}") + return result + + def process_batch( + self, + files: Iterable[Path], + progress_callback: Optional[Callable[[FileResult], None]] = None, + ) -> List[FileResult]: + """ + Process multiple files concurrently using ThreadPoolExecutor. + + Args: + files: Iterable of file paths to process. + progress_callback: Optional callback called after each file completes. + Receives the FileResult for progress updates. + + Returns: + List of FileResult objects for all processed files. + """ + file_list = list(files) + + if not file_list: + return self.results + + # Used ThreadPoolExecutor for I/O-bound concurrent processing + with ThreadPoolExecutor(max_workers=self.max_workers) as executor: + # Submit all files for processing + future_to_file = { + executor.submit(self.process_file, file): file for file in file_list + } + + # Collect results as they complete + for future in as_completed(future_to_file): + result = future.result() + if progress_callback: + progress_callback(result) + + return self.results + + def process_batch_sequential(self, files: Iterable[Path]) -> List[FileResult]: + """ + Process files sequentially (legacy behavior for debugging). + + Args: + files: Iterable of file paths to process. + + Returns: + List of FileResult objects for all processed files. + """ + for file in files: + self.process_file(file) + return self.results + + def get_summary(self) -> BatchSummary: + """ + Return aggregated statistics for all processed files. + + Returns: + BatchSummary with counts and result details. + """ + with self._results_lock: + results_copy = list(self.results) + + summary = BatchSummary( + total=len(results_copy), + success=sum( + 1 for r in results_copy if r.success and r.action == "scrubbed" + ), + skipped=sum(1 for r in results_copy if not r.success), + dry_run=self.dry_run, + output_dir=self.output_dir, + results=results_copy, + ) + # Count dry-run as separate from success for clarity + if self.dry_run: + summary.success = sum(1 for r in results_copy if r.success) + return summary + + def _append_result(self, result: FileResult) -> None: + """Thread-safe append to results list.""" + with self._results_lock: + self.results.append(result) + + def _get_unique_output_path(self, file: Path) -> Path: + """ + Generate unique output path with suffix (_1, _2) if file exists. + + Thread-safe: uses lock to prevent race conditions during concurrent processing. + + Args: + file: Original file path. + + Returns: + Unique path in output directory that doesn't conflict with existing files. + """ + with self._path_lock: + # creates the destination directory if it doesn't exist + self.output_dir.mkdir(parents=True, exist_ok=True) + + base_name = file.stem + extension = file.suffix + output_path = self.output_dir / f"processed_{base_name}{extension}" + + # If file exists, add incrementing suffix + counter = 1 + while output_path.exists(): + output_path = ( + self.output_dir / f"processed_{base_name}_{counter}{extension}" + ) + counter += 1 + + # Create empty placeholder to reserve the path + output_path.touch() + + return output_path diff --git a/src/services/image_handler.py b/src/services/image_handler.py index c54999e7..1134ffa2 100644 --- a/src/services/image_handler.py +++ b/src/services/image_handler.py @@ -1,3 +1,11 @@ +""" +Image metadata handler for JPEG and PNG files. + +This module provides the ImageHandler class which implements the MetadataHandler +interface for image files. It delegates the actual metadata operations to +format-specific processors (JpegProcessor, PngProcessor). +""" + import shutil from pathlib import Path from typing import Optional @@ -5,23 +13,40 @@ from typing import Optional import piexif # pyright: ignore[reportMissingTypeStubs] from PIL import Image -from src.core.jpeg_metadata import JpegProcessaor +from src.core.jpeg_metadata import JpegProcessor from src.core.png_metadata import PngProcessor from src.services.metadata_handler import MetadataHandler class ImageHandler(MetadataHandler): + """ + Metadata handler for image files (JPEG, PNG). + + Implements the MetadataHandler interface using format-specific processors + to read, wipe, and save image metadata. Uses piexif for EXIF manipulation. + + Attributes: + processors: Dict mapping file extensions to processor instances. + tags_to_delete: List of EXIF tags to remove during wipe operation. + """ + def __init__(self, filepath: str): + """ + Initialize the image handler. + + Args: + filepath: Path to the image file to process. + """ super().__init__(filepath) self.processors = { - ".jpeg": JpegProcessaor(), - ".jpg": JpegProcessaor(), + ".jpeg": JpegProcessor(), + ".jpg": JpegProcessor(), ".png": PngProcessor(), } self.tags_to_delete = [] def read(self): - """Extracts metadata into a standard dictionary.""" + """Extract metadata from the file.""" with Image.open(Path(self.filepath)) as img: extension = Path(self.filepath).suffix processor = self.processors.get(extension) @@ -47,14 +72,19 @@ class ImageHandler(MetadataHandler): ) def save(self, output_path: Optional[str] = None) -> None: - """Writes the changes to a copy of the original file.""" - destination_dir = Path(output_path or "/archive/") + """ + Writes the changes to a copy of the original file. - # creates the destination directory if it doesn't exist - destination_dir.mkdir(parents=True, exist_ok=True) - destination_file_path = ( - destination_dir / f"processed_{Path(self.filepath).name}" - ) + Args: + output_path: Can be a directory path (legacy behavior) or a full file path. + If a directory, generates filename as 'processed_{original_name}'. + If a file path, uses it directly. + """ + destination_file_path = "" + if output_path: + # setup the destination directory. + # which was created by the batch_processor + destination_file_path = Path(output_path) # copies the original file to the destination directory shutil.copy2(self.filepath, destination_file_path) @@ -63,9 +93,3 @@ class ImageHandler(MetadataHandler): with Image.open(destination_file_path) as img: exif_bytes = piexif.dump(self.processed_metadata) img.save(destination_file_path, exif=exif_bytes) - - -test = ImageHandler(r"C:\Users\Xheri\development\testimage\20221201_090615.jpg") -test.read() -test.wipe() -test.save() diff --git a/src/services/metadata_factory.py b/src/services/metadata_factory.py index 2f78de66..3502aebf 100644 --- a/src/services/metadata_factory.py +++ b/src/services/metadata_factory.py @@ -1,11 +1,44 @@ +""" +Factory for creating metadata handlers. + +This module provides the MetadataFactory class which uses the factory pattern +to create appropriate handler instances based on file type. It enables +extensibility for supporting new file formats (PDF, Office docs, etc.). +""" + from pathlib import Path from src.services.image_handler import ImageHandler class MetadataFactory: + """ + Factory class for creating metadata handlers. + + Uses the factory pattern to return the appropriate handler instance + based on file extension. This design allows easy extension to support + new file types without modifying existing code. + + Supported formats: + - Images: .jpg, .jpeg, .png + - Future: .pdf, .docx, .xlsx, .pptx + """ + @staticmethod def get_handler(filepath: str): + """ + Create and return the appropriate metadata handler for a file. + + Args: + filepath: Path to the file to process. + + Returns: + MetadataHandler: An instance of the appropriate handler subclass. + + Raises: + ValueError: If no handler is defined for the file type or + if the path is not a valid file. + """ ext = Path(filepath).suffix.lower() if Path(filepath).is_file(): if ext in [".jpg", ".jpeg", ".png"]: diff --git a/src/services/metadata_handler.py b/src/services/metadata_handler.py index 89c2e2db..dea9b48b 100644 --- a/src/services/metadata_handler.py +++ b/src/services/metadata_handler.py @@ -1,24 +1,72 @@ +""" +Abstract base class for metadata handlers. + +This module defines the MetadataHandler ABC which establishes the interface +for all metadata handlers. Concrete implementations (ImageHandler, PDFHandler, etc.) +must implement the read, wipe, and save methods. +""" + from abc import ABC, abstractmethod from typing import Any, Dict, Optional class MetadataHandler(ABC): + """ + Abstract base class for all metadata handlers. + + Defines the common interface for reading, modifying, and saving + metadata across different file types. All concrete handlers must + implement the abstract methods. + + Attributes: + filepath: Path to the file being processed. + metadata: Dictionary containing the extracted metadata. + processed_metadata: Dictionary containing metadata after processing. + """ + def __init__(self, filepath: str): + """ + Initialize the metadata handler. + + Args: + filepath: Path to the file to process. + """ self.filepath = filepath self.metadata: Dict[str, Any] = {} self.processed_metadata: Dict[str, Any] = {} @abstractmethod def read(self) -> Dict[str, Any]: - """Extracts metadata into a standard dictionary.""" + """ + Extract metadata from the file. + + Returns: + Dict containing the extracted metadata with human-readable keys. + + Raises: + MetadataNotFoundError: If no metadata is found in the file. + """ pass @abstractmethod def wipe(self) -> None: - """Updates internal metadata state.""" + """ + Remove privacy-sensitive metadata from the file. + + Updates self.processed_metadata with the cleaned metadata state. + + Raises: + MetadataProcessingError: If an error occurs during processing. + """ pass @abstractmethod def save(self, output_path: Optional[str] = None) -> None: - """Writes the changes to a file.""" + """ + Save the processed file with cleaned metadata. + + Args: + output_path: Destination path for the processed file. + If None, uses a default location. + """ pass diff --git a/src/utils/display.py b/src/utils/display.py index 73393e5d..dfb7d657 100644 --- a/src/utils/display.py +++ b/src/utils/display.py @@ -1,3 +1,10 @@ +""" +Display utilities for rich terminal output. + +This module provides functions for displaying metadata and batch processing +results in beautifully formatted tables and panels using the Rich library. +""" + from typing import Any, Dict from rich import box @@ -13,7 +20,13 @@ console = Console() def print_metadata_table(metadata: Dict[str, Any]): """ - Displays metadata organized by logical groups. + Display metadata in a formatted table organized by logical groups. + + Organizes metadata into categories (Device Info, Exposure Settings, + Image Data, Dates) and displays them in a Rich panel with color coding. + + Args: + metadata: Dict of metadata key-value pairs to display. """ # Define the groups using simple lists of keys @@ -78,3 +91,50 @@ def print_metadata_table(metadata: Dict[str, Any]): console.print( Panel(table, title="Metadata Report", border_style="blue", expand=False) ) + + +def print_batch_summary(summary) -> None: + """ + Display batch processing results in a rich panel. + + Args: + summary: BatchSummary object with processing statistics. + """ + # Build the summary table + table = Table(box=box.ROUNDED, show_header=False, expand=False) + table.add_column("Metric", style="cyan") + table.add_column("Value", style="green") + + if summary.dry_run: + table.add_row("Mode", "[yellow]DRY-RUN (no changes made)[/yellow]") + table.add_row("Would process", str(summary.success)) + table.add_row("Would skip", str(summary.skipped)) + else: + table.add_row("Total processed", str(summary.total)) + table.add_row("✅ Success", f"[green]{summary.success}[/green]") + table.add_row("⚠️ Skipped", f"[yellow]{summary.skipped}[/yellow]") + if summary.output_dir: + table.add_row("📁 Output", str(summary.output_dir.resolve())) + + # Show failed files if any + failed = [r for r in summary.results if not r.success] + if failed: + table.add_section() + table.add_row(Text("Failed files:", style="bold red"), "") + for result in failed[:5]: # Show max 5 failures + table.add_row(f" {result.filepath.name}", f"[dim]{result.error}[/dim]") + if len(failed) > 5: + table.add_row("", f"[dim]... and {len(failed) - 5} more[/dim]") + + # Determine panel title and style + if summary.dry_run: + title = "🔍 Dry-Run Summary" + border_style = "yellow" + elif summary.skipped == 0: + title = "✅ Scrub Complete" + border_style = "green" + else: + title = "⚠️ Scrub Complete (with warnings)" + border_style = "yellow" + + console.print(Panel(table, title=title, border_style=border_style, expand=False)) diff --git a/src/utils/exceptions.py b/src/utils/exceptions.py index 90bb1c43..8cf00cd7 100644 --- a/src/utils/exceptions.py +++ b/src/utils/exceptions.py @@ -1,22 +1,30 @@ +""" +Custom exceptions for metadata processing operations. + +This module defines a hierarchy of exceptions used throughout the +metadata scrubber tool for handling various error conditions. +""" + + class MetadataException(Exception): - """Base class for other exceptions in this module.""" + """Base class for all metadata-related exceptions.""" pass class UnsupportedFormatError(MetadataException): - """Exception raised when an unsupported file format is encountered.""" + """Raised when attempting to process an unsupported file format.""" pass class MetadataNotFoundError(MetadataException): - """Exception raised when metadata is not found.""" + """Raised when no metadata is found in a file.""" pass class MetadataProcessingError(MetadataException): - """Exception raised when an error occurs during metadata processing.""" + """Raised when an error occurs during metadata processing.""" pass diff --git a/src/utils/formatter.py b/src/utils/formatter.py index 4a16a5c7..997bfa3a 100644 --- a/src/utils/formatter.py +++ b/src/utils/formatter.py @@ -1,9 +1,24 @@ +""" +Value formatting utilities for metadata display. + +This module provides helper functions to convert raw EXIF data into +human-readable strings for display in the terminal. +""" + from typing import Any def clean_value(value: Any) -> str: """ - Helper to make raw EXIF data human-readable. + Convert raw EXIF data into a human-readable string. + + Handles various EXIF value types including bytes, tuples, and empty values. + + Args: + value: Raw EXIF value (bytes, tuple, str, int, etc.). + + Returns: + Human-readable string representation of the value. """ # Decode bytes (e.g., b'samsung' -> 'samsung') if isinstance(value, bytes): diff --git a/src/utils/get_target_files.py b/src/utils/get_target_files.py index 849773e2..12b84c6c 100644 --- a/src/utils/get_target_files.py +++ b/src/utils/get_target_files.py @@ -1,5 +1,28 @@ -def get_target_files(input_path_str, ext: str): - """Yields a list of files to process based on the input path. handles recursive searches""" - # if input_path_str is a directory, yield all files with the specified extension +""" +File discovery utilities for batch processing. + +This module provides functions to find and yield files for processing, +supporting recursive directory traversal with extension filtering. +""" + +from pathlib import Path +from typing import Generator + + +def get_target_files(input_path_str: Path, ext: str) -> Generator[Path, None, None]: + """ + Yield files to process based on input path and extension filter. + + Recursively searches the input directory for files matching the + specified extension. + + Args: + input_path_str: Path object pointing to the target directory. + ext: File extension to filter by (without dot, e.g., 'jpg'). + + Yields: + Path objects for each matching file found. + """ + # If input is a directory, yield all files with the specified extension if input_path_str.is_dir(): yield from input_path_str.rglob(f"*.{ext}") diff --git a/src/utils/logger.py b/src/utils/logger.py index 3ee3bd05..6df8280e 100644 --- a/src/utils/logger.py +++ b/src/utils/logger.py @@ -1,3 +1,10 @@ +""" +Logging configuration for the metadata scrubber CLI. + +This module provides the setup_logging function which configures the +application's logging with Rich handlers for beautiful terminal output. +""" + import logging from rich.logging import RichHandler @@ -5,14 +12,23 @@ from rich.logging import RichHandler def setup_logging(verbose: bool = False): """ - - Default (Info): Shows main steps and errors. - - Verbose (Debug): Shows file paths, extraction details, and raw values. + Configure application logging with Rich formatting. + + Sets up the logger with appropriate level and Rich handlers for + beautiful terminal output including colorful stack traces. + + Args: + verbose: If True, enables DEBUG level logging. + If False (default), enables INFO level logging. + + Returns: + Logger instance for 'metadata-scrubber'. """ # Define the log level level = logging.DEBUG if verbose else logging.INFO # Configure the logger - # remove existing handlers to avoid duplicate lines if the app restarts + # Remove existing handlers to avoid duplicate lines if the app restarts logging.getLogger().handlers.clear() logging.basicConfig(