diff --git a/src/commands/read.py b/src/commands/read.py index eb50d502..fb45e69b 100644 --- a/src/commands/read.py +++ b/src/commands/read.py @@ -43,19 +43,19 @@ def get_metadata( handler = MetadataFactory.get_handler(str(file)) # Read - console.print(f"🔎 Processing [bold cyan]{file.name}[/bold cyan]...") + console.print(f"🔎 Reading [bold cyan]{file.name}[/bold cyan]...") current_data = handler.read() if log.isEnabledFor(logging.DEBUG): # if verbose mode is enabled, log the Info - log.info(f"Successfully read metadata from {file_path.name}") + log.info(f"Successfully read metadata from {file.name}") print_metadata_table(current_data) except Exception as e: # display error in console - console.print(f"❌ [bold red]Skipped[/bold red] [cyan]{file_path.name}[/cyan]: [dim]{e}[/dim]") + console.print(f"❌ [bold red]Skipped[/bold red] [cyan]{file.name}[/cyan]: [dim]{e}[/dim]") # LOG: Full technical details (Stack trace) for you to debug if log.isEnabledFor(logging.DEBUG): # if verbose mode is enabled, log the traceback - log.error(f"Failed to process {file_path}", exc_info=True) + log.error(f"Failed to read metadata from {file}", exc_info=True) diff --git a/src/core/png_metadata.py b/src/core/png_metadata.py index 8d00fd08..f240a472 100644 --- a/src/core/png_metadata.py +++ b/src/core/png_metadata.py @@ -1,11 +1,15 @@ """ 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. +This module provides the PngProcessor class which handles metadata +extraction and manipulation for PNG images, including both EXIF data +and PNG textual metadata (PngInfo chunks). """ -from PIL import ExifTags +from typing import Any, Dict, List, Optional + +from PIL import ExifTags, PngImagePlugin +from PIL.Image import Exif, Image from src.utils.exceptions import MetadataNotFoundError, MetadataProcessingError @@ -14,71 +18,110 @@ 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. + Handles reading, extracting, and deleting metadata from PNG files. + Processes both EXIF data and PNG textual chunks (PngInfo). Attributes: tags_to_delete: List of EXIF tag IDs to remove. + text_keys_to_delete: List of PngInfo text keys to remove. data: Dict of extracted metadata with human-readable keys. """ + # Privacy-sensitive PNG text keys to remove + SENSITIVE_TEXT_KEYS = { + "Author", + "Comment", + "Copyright", + "Creation Time", + "Description", + "Disclaimer", + "Software", + "Source", + "Title", + "Warning", + "XML:com.adobe.xmp", # XMP metadata + } + def __init__(self): """Initialize the PNG processor with empty data structures.""" - self.tags_to_delete = [] - self.data = {} + self.tags_to_delete: List[int] = [] + self.text_keys_to_delete: List[str] = [] + self.data: Dict[str, Any] = {} - def get_metadata(self, img): + def get_metadata(self, img: Image) -> Dict[str, Any]: """ - Extract EXIF metadata from a PNG image. + Extract metadata from a PNG image. + + Extracts both EXIF data (if present) and PNG textual chunks (PngInfo). Args: img: PIL Image object. Returns: - Dict with 'data' (metadata dict) and 'tags_to_delete' (tag IDs list). + Dict with 'data' (metadata dict), 'tags_to_delete' (EXIF tag IDs), + and 'text_keys' (PngInfo keys to remove). Raises: - MetadataNotFoundError: If no EXIF data is found in the image. + MetadataNotFoundError: If no metadata is found in the image. """ img.load() + found_metadata = False + + # Extract EXIF data (if present) exif = img.getexif() + if exif: + found_metadata = True + # Main IFD + for tag, value in exif.items(): + tag_name = ExifTags.TAGS.get(tag, f"Tag_{tag}") + self.tags_to_delete.append(tag) + self.data[f"EXIF:{tag_name}"] = value - if not exif: - raise MetadataNotFoundError("No EXIF data found in the image.") + # GPS IFD + gps_ifd = exif.get_ifd(ExifTags.IFD.GPSInfo) + for tag, value in gps_ifd.items(): + tag_name = ExifTags.GPSTAGS.get(tag, f"GPSTag_{tag}") + self.tags_to_delete.append(tag) + self.data[f"GPS:{tag_name}"] = value - # 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) + # Extract PNG textual metadata (PngInfo chunks) + if hasattr(img, "info") and img.info: + for key, value in img.info.items(): + # Skip binary/internal data + if key in ("icc_profile", "exif", "transparency", "gamma"): + continue - # Save to list and dict - self.tags_to_delete.append(tag) - self.data[tag_name] = value - print(f"{tag_name}: {value}") + if isinstance(value, (str, bytes)): + found_metadata = True + if isinstance(value, bytes): + try: + value = value.decode("utf-8", errors="replace") + except Exception: + value = str(value) - # 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) + self.data[f"PNG:{key}"] = value + if isinstance(key, str): # Only add string keys + self.text_keys_to_delete.append(key) - # Save to list and dict - self.tags_to_delete.append(tag) - self.data[tag_name] = value - print(f"{tag_name}: {value}") + if not found_metadata: + raise MetadataNotFoundError("No metadata found in the PNG image.") - return {"data": self.data, "tags_to_delete": self.tags_to_delete} + return { + "data": self.data, + "tags_to_delete": self.tags_to_delete, + "text_keys": self.text_keys_to_delete, + } - def delete_metadata(self, img, tags_to_delete): + def delete_metadata(self, img: Image, tags_to_delete: List[int]) -> Exif: """ - Remove specified EXIF tags from a PNG image. + Remove EXIF tags from a PNG image. Args: img: PIL Image object. - tags_to_delete: List of tag IDs to remove. + tags_to_delete: List of EXIF tag IDs to remove. Returns: - Modified EXIF data with specified tags removed. + Mutated Exif object with specified tags removed. Raises: MetadataProcessingError: If an error occurs during processing. @@ -86,17 +129,53 @@ class PngProcessor: img.load() exif = img.getexif() try: - # Iterate through the (0th) IFD - for tag_id, value in exif.items(): + # Delete tags from main IFD (iterate over copy of keys) + for tag_id in list(exif.keys()): if tag_id in tags_to_delete: del exif[tag_id] - # Iterate through the (GPS) IFD + # Clear GPS IFD entirely (privacy-sensitive) gps_ifd = exif.get_ifd(ExifTags.IFD.GPSInfo) - for tag_id, value in gps_ifd.items(): - if tag_id in tags_to_delete: - del gps_ifd[tag_id] + gps_ifd.clear() - return exif + gps_ifd + return exif except Exception as e: - raise MetadataProcessingError(f"Error Processing image: {str(e)}") + raise MetadataProcessingError(f"Error processing PNG EXIF: {str(e)}") + + def get_clean_pnginfo( + self, img: Image, keys_to_remove: Optional[List[str]] = None + ) -> Optional[PngImagePlugin.PngInfo]: + """ + Create a new PngInfo with sensitive keys removed. + + Args: + img: PIL Image object. + keys_to_remove: Specific keys to remove. If None, removes all sensitive keys. + + Returns: + New PngInfo object with only safe metadata, or None if no safe metadata. + """ + if not hasattr(img, "info") or not img.info: + return None + + keys_to_remove = keys_to_remove or list(self.text_keys_to_delete) + + # Create new PngInfo with only safe keys + pnginfo = PngImagePlugin.PngInfo() + has_safe_data = False + + for key, value in img.info.items(): + # Skip keys to remove + if key in keys_to_remove: + continue + + # Skip binary/internal data + if key in ("icc_profile", "exif", "transparency", "gamma"): + continue + + # Only include text data with string keys + if isinstance(key, str) and isinstance(value, str): + pnginfo.add_text(key, value) + has_safe_data = True + + return pnginfo if has_safe_data else None diff --git a/src/services/batch_processor.py b/src/services/batch_processor.py index de90d503..cb5eed82 100644 --- a/src/services/batch_processor.py +++ b/src/services/batch_processor.py @@ -92,6 +92,8 @@ class BatchProcessor: Returns: FileResult with success status and details. """ + output_path: Optional[Path] = None # Track reserved path for cleanup + try: # Dry-run mode: just report what would happen if self.dry_run: @@ -131,6 +133,9 @@ class BatchProcessor: return result except Exception as e: + # Cleanup: remove empty placeholder file if reservation failed + self._cleanup_reserved_path(output_path) + result = FileResult( filepath=file, success=False, @@ -223,6 +228,28 @@ class BatchProcessor: with self._results_lock: self.results.append(result) + def _cleanup_reserved_path(self, output_path: Optional[Path]) -> None: + """ + Remove empty placeholder file created during path reservation. + + Called when processing fails after _get_unique_output_path() reserved + a path via touch(). Only removes files that are empty (0 bytes) to + avoid deleting partially written data. + + Args: + output_path: Path that was reserved, or None if not yet reserved. + """ + if output_path is None: + return + + try: + if output_path.exists() and output_path.stat().st_size == 0: + output_path.unlink() + log.debug(f"Cleaned up empty placeholder: {output_path}") + except OSError: + # Best effort cleanup - don't fail if we can't delete + pass + def _get_unique_output_path(self, file: Path) -> Path: """ Generate unique output path with suffix (_1, _2) if file exists. diff --git a/src/services/image_handler.py b/src/services/image_handler.py index 1134ffa2..0259d501 100644 --- a/src/services/image_handler.py +++ b/src/services/image_handler.py @@ -16,6 +16,14 @@ from PIL import Image from src.core.jpeg_metadata import JpegProcessor from src.core.png_metadata import PngProcessor from src.services.metadata_handler import MetadataHandler +from src.utils.exceptions import UnsupportedFormatError + +# Map Pillow format names to processor keys +FORMAT_MAP = { + "jpeg": "jpeg", + "jpg": "jpeg", + "png": "png", +} class ImageHandler(MetadataHandler): @@ -23,11 +31,13 @@ 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. + to read, wipe, and save image metadata. Uses Pillow's format detection + to handle files with incorrect extensions. Attributes: - processors: Dict mapping file extensions to processor instances. + processors: Dict mapping format names to processor instances. tags_to_delete: List of EXIF tags to remove during wipe operation. + detected_format: Actual image format detected by Pillow. """ def __init__(self, filepath: str): @@ -39,57 +49,121 @@ class ImageHandler(MetadataHandler): """ super().__init__(filepath) self.processors = { - ".jpeg": JpegProcessor(), - ".jpg": JpegProcessor(), - ".png": PngProcessor(), + "jpeg": JpegProcessor(), + "png": PngProcessor(), } self.tags_to_delete = [] + self.detected_format: Optional[str] = None + self.text_keys_to_delete = [] + + def _detect_format(self) -> str: + """ + Detect actual image format using Pillow, not file extension. + + This protects against misnamed files (e.g., a PNG saved as .jpg). + + Returns: + Normalized format string ('jpeg' or 'png'). + + Raises: + UnsupportedFormatError: If format is not supported or undetectable. + """ + with Image.open(Path(self.filepath)) as img: + if img.format is None: + raise UnsupportedFormatError( + f"Could not detect format for: {self.filepath}" + ) + + pillow_format = img.format.lower() + normalized = FORMAT_MAP.get(pillow_format) + + if normalized is None: + raise UnsupportedFormatError( + f"Unsupported format: {pillow_format} (file: {self.filepath})" + ) + + return normalized def read(self): - """Extract metadata from the file.""" + """ + Extract metadata from the file. + + Uses actual format detection to select the appropriate processor. + """ + self.detected_format = self._detect_format() + processor = self.processors.get(self.detected_format) + + if not processor: + raise UnsupportedFormatError(f"Unsupported format: {self.detected_format}") + with Image.open(Path(self.filepath)) as img: - extension = Path(self.filepath).suffix - processor = self.processors.get(extension) - - if not processor: - raise ValueError(f"Unsupported format: {extension}") - - self.metadata = processor.get_metadata(img)["data"] - self.tags_to_delete = processor.get_metadata(img)["tags_to_delete"] + result = processor.get_metadata(img) + self.metadata = result["data"] + self.tags_to_delete = result["tags_to_delete"] + # Store text keys for PNG processing + self.text_keys_to_delete = result.get("text_keys", []) return self.metadata def wipe(self) -> None: - """Wipes internal metadata state.""" + """ + Remove privacy-sensitive metadata from the file. + + Uses actual format detection to select the appropriate processor. + """ + # Use cached format if available, otherwise detect + if not self.detected_format: + self.detected_format = self._detect_format() + + processor = self.processors.get(self.detected_format) + + if not processor: + raise UnsupportedFormatError(f"Unsupported format: {self.detected_format}") + with Image.open(Path(self.filepath)) as img: - extension = Path(self.filepath).suffix - processor = self.processors.get(extension) - - if not processor: - raise ValueError(f"Unsupported format: {extension}") - self.processed_metadata = processor.delete_metadata( img, self.tags_to_delete ) + # For PNG, also get clean PngInfo + if self.detected_format == "png" and hasattr( + processor, "get_clean_pnginfo" + ): + self.clean_pnginfo = processor.get_clean_pnginfo( + img, self.text_keys_to_delete + ) def save(self, output_path: Optional[str] = None) -> None: """ Writes the changes to a copy of the original file. + Handles format-specific saving: + - JPEG: Uses piexif to write cleaned EXIF data + - PNG: Saves without EXIF and strips textual metadata + 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. + output_path: Full path to the destination file. """ - destination_file_path = "" - if output_path: - # setup the destination directory. - # which was created by the batch_processor - destination_file_path = Path(output_path) + destination_file_path = Path(output_path) if output_path else None - # copies the original file to the destination directory - shutil.copy2(self.filepath, destination_file_path) + if not destination_file_path: + raise ValueError("output_path is required") - # writes the processed metadata to the image in the destination directory - with Image.open(destination_file_path) as img: - exif_bytes = piexif.dump(self.processed_metadata) - img.save(destination_file_path, exif=exif_bytes) + # Use detected format (falls back to extension if not detected) + actual_format = self.detected_format or self._detect_format() + + if actual_format == "jpeg": + # JPEG: Copy then write cleaned EXIF data + shutil.copy2(self.filepath, destination_file_path) + with Image.open(destination_file_path) as img: + exif_bytes = piexif.dump(self.processed_metadata) + img.save(destination_file_path, exif=exif_bytes) + elif actual_format == "png": + # PNG: Open original, save fresh copy without metadata + with Image.open(Path(self.filepath)) as img: + # Save without exif and without pnginfo to strip all metadata + # Preserve image mode and data integrity + img.save( + destination_file_path, + format="PNG", + exif=None, + pnginfo=getattr(self, "clean_pnginfo", None), + ) diff --git a/src/services/metadata_factory.py b/src/services/metadata_factory.py index 3502aebf..27089b17 100644 --- a/src/services/metadata_factory.py +++ b/src/services/metadata_factory.py @@ -9,6 +9,7 @@ extensibility for supporting new file formats (PDF, Office docs, etc.). from pathlib import Path from src.services.image_handler import ImageHandler +from src.utils.exceptions import UnsupportedFormatError class MetadataFactory: @@ -36,8 +37,8 @@ class MetadataFactory: 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. + UnsupportedFormatError: If no handler is defined for the file type. + ValueError: If the path is not a valid file. """ ext = Path(filepath).suffix.lower() if Path(filepath).is_file(): @@ -52,7 +53,7 @@ class MetadataFactory: # elif ext == ".pptx": # return PowerPointHandler(filepath) else: - raise ValueError(f"No handler defined for {ext} files.") + raise UnsupportedFormatError(f"No handler defined for {ext} files.") else: raise ValueError( f"{filepath} is not a file. if you want to process a directory, use the --recursive flag." diff --git a/src/utils/get_target_files.py b/src/utils/get_target_files.py index 12b84c6c..89a3d9b8 100644 --- a/src/utils/get_target_files.py +++ b/src/utils/get_target_files.py @@ -13,16 +13,19 @@ def get_target_files(input_path_str: Path, ext: str) -> Generator[Path, None, No """ Yield files to process based on input path and extension filter. - Recursively searches the input directory for files matching the - specified extension. + Handles both directories (recursive search) and single files (defensive). Args: - input_path_str: Path object pointing to the target directory. + input_path_str: Path object pointing to a file or 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}") + # Directory: recursively yield all files with the specified extension + yield from input_path_str.rglob(f"*.{ext.lower()}") + elif input_path_str.is_file(): + # Defensive: if a file is passed, yield it if extension matches + if input_path_str.suffix.lstrip(".").lower() == ext.lower(): + yield input_path_str