RomanNum3ral
/
Cybersecurity-Projects
mirror of https://github.com/CarterPerez-dev/Cybersecurity-Projects.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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.
This commit is contained in:
HERITAGE-XION 2026-01-06 00:22:04 +01:00
parent fbd303f6d9
commit ff367d1e37
14 changed files with 783 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
src/commands/read.py
View File

@ -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:

141
src/commands/scrub.py Normal file
View File

@ -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())

58
src/core/jpeg_metadata.py
View File

@ -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]

56
src/core/png_metadata.py
View File

@ -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:

17
src/main.py
View File

@ -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__":

257
src/services/batch_processor.py Normal file
View File

@ -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

58
src/services/image_handler.py
View File

@ -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()

33
src/services/metadata_factory.py
View File

@ -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"]:

54
src/services/metadata_handler.py
View File

@ -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

62
src/utils/display.py
View File

@ -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))

16
src/utils/exceptions.py
View File

@ -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

17
src/utils/formatter.py
View File

@ -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):

29
src/utils/get_target_files.py
View File

@ -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}")

22
src/utils/logger.py
View File

@ -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(