metadata-scrubber-tool:

refactored buggy implementation.
png processing supports reading EXIF and pngInfo and not-serialized-to-human-readable-format data but not XMP OR IPTC.

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)

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

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.

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),
+                )

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."

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