This document walks through the actual code. We'll build key features step by step and explain the decisions along the way.
src/
├── commands/
│ ├── read.py # Display metadata without modification
│ ├── scrub.py # Remove metadata from files
│ └── verify.py # Compare before/after states
├── core/
│ ├── jpeg_metadata.py # EXIF parsing for JPEG
│ └── png_metadata.py # Metadata handling for PNG
├── services/
│ ├── metadata_handler.py # Abstract base class
│ ├── image_handler.py # Images (JPEG/PNG)
│ ├── pdf_handler.py # PDF documents
│ ├── excel_handler.py # Excel workbooks
│ ├── metadata_factory.py # Route files to handlers
│ └── batch_processor.py # Concurrent processing
└── utils/
├── display.py # Rich terminal formatting
└── exceptions.py # Custom error types
When a user runs mst scrub photo.jpg, we need to determine which handler to use. Different file types need different handlers. We could scatter if ext == ".jpg" checks everywhere, but that's unmaintainable.
Centralize routing in MetadataFactory (src/services/metadata_factory.py):
class MetadataFactory:
@staticmethod
def get_handler(filepath: str):
ext = Path(filepath).suffix.lower()
if Path(filepath).is_file():
if ext in [".jpg", ".jpeg", ".png"]:
return ImageHandler(filepath)
elif ext == ".pdf":
return PDFHandler(filepath)
elif ext in [".xlsx", ".xlsm"]:
return ExcelHandler(filepath)
else:
raise UnsupportedFormatError(
f"No handler for {ext} files"
)
else:
raise ValueError(f"{filepath} is not a file")Why this code works:
- Line 4: Extract extension and normalize to lowercase (.JPG → .jpg)
- Line 5: Verify it's actually a file (prevents directory processing)
- Lines 6-12: Route based on extension to appropriate handler class
- Line 14: Explicit error for unsupported formats (better than silent failure)
Common mistake here:
# Wrong - trusts extension blindly
if filepath.endswith(".jpg"):
return ImageHandler(filepath)
# Better - verifies file exists first
if Path(filepath).is_file() and ext == ".jpg":
return ImageHandler(filepath)The wrong approach fails when users pass non-existent paths or when file extensions lie about content.
From JpegProcessor.get_metadata() (src/core/jpeg_metadata.py:32-64):
def get_metadata(self, img: Image.Image) -> JpegMetadataResult:
if "exif" not in img.info:
raise MetadataNotFoundError("No EXIF data found")
exif_dict = piexif.load(img.info["exif"])
for ifd, value in exif_dict.items():
if not isinstance(exif_dict[ifd], dict):
continue # Skip thumbnail blob
for tag, tag_value in exif_dict[ifd].items():
tag_name = str(piexif.TAGS[ifd][tag]["name"])
# Preserve structural tags
if tag_name in ("Orientation", "ColorSpace", "ExifTag"):
continue
self.tags_to_delete.append(tag)
self.data[tag_name] = tag_value
return {"data": self.data, "tags_to_delete": self.tags_to_delete}What's happening:
- Line 2-3: Check if EXIF exists before trying to parse it
- Line 5: piexif.load() parses binary EXIF into nested dicts
- Line 6-8: Iterate IFDs (Image File Directories) - containers for tags
- Line 10-11: Get human-readable tag name from numeric ID
- Line 13-14: Skip tags needed for proper image display
- Line 16-17: Mark tag for deletion and store its value
Why we do it this way: Removing Orientation breaks image rotation. Removing ColorSpace breaks color rendering. We preserve what's needed for display, delete everything else.
Alternative approach:
# Simpler but wrong - removes everything
for tag in exif_dict["0th"]:
del exif_dict["0th"][tag]This corrupts images. The photo displays upside-down or with wrong colors.
From JpegProcessor.delete_metadata() (src/core/jpeg_metadata.py:66-96):
def delete_metadata(self, img: Image.Image, tags_to_delete: list[int]):
try:
exif_dict = piexif.load(img.info["exif"])
for ifd, value in exif_dict.items():
if not isinstance(exif_dict[ifd], dict):
continue
for tag in list(exif_dict[ifd]):
if tag in tags_to_delete:
del exif_dict[ifd][tag]
return exif_dict
except Exception as e:
raise MetadataProcessingError(f"Error: {str(e)}")Key parts explained:
- Line 8:
list(exif_dict[ifd])creates a copy of keys before iteration (prevents "dict changed size during iteration" error) - Line 9-10: Only delete tags we marked during read phase
- Line 12: Return modified dict for piexif.dump() to serialize
From ImageHandler.save() (src/services/image_handler.py:117-147):
def save(self, output_path: str | None = None) -> None:
if not output_path:
raise ValueError("output_path is required")
actual_format = self.detected_format or self._detect_format()
if actual_format == "jpeg":
shutil.copy2(self.filepath, output_path)
with Image.open(output_path) as img:
exif_bytes = piexif.dump(self.processed_metadata)
img.save(output_path, exif=exif_bytes)
elif actual_format == "png":
with Image.open(self.filepath) as img:
img.save(output_path, format="PNG", exif=None, pnginfo=None)What's happening:
- Lines 2-3: Validate output path exists
- Line 5: Use cached format or detect it
- Lines 7-11: JPEG - copy file then rewrite with cleaned EXIF
- Lines 12-14: PNG - save fresh copy without any metadata
Why JPEG copies then modifies: We preserve JPEG compression. If we re-encode, quality degrades. Copy preserves original compression, we just swap EXIF.
Processing 1000 files sequentially takes minutes. We need concurrency.
From BatchProcessor.process_batch() (src/services/batch_processor.py:134-168):
def process_batch(
self,
files: Iterable[Path],
progress_callback: Callable[[FileResult], None] | None = None,
) -> list[FileResult]:
file_list = list(files)
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
future_to_file = {
executor.submit(self.process_file, file): file
for file in file_list
}
for future in as_completed(future_to_file):
result = future.result()
if progress_callback:
progress_callback(result)
return self.resultsWhat's happening:
- Line 8: Create thread pool with configurable worker count
- Lines 9-12: Submit all files to executor, get Future objects
- Lines 14-17: Process results as workers complete (not in submission order)
- Line 16: Callback updates progress bar in real-time
Thread safety:
The _get_unique_output_path() method uses locks to prevent race conditions:
def _get_unique_output_path(self, file: Path, reserve: bool = True) -> Path:
with self._path_lock: # Thread-safe
output_path = self.output_dir / f"processed_{file.name}"
counter = 1
while output_path.exists():
output_path = self.output_dir / f"processed_{file.name}_{counter}{file.suffix}"
counter += 1
if reserve:
output_path.touch() # Reserve path
return output_pathWithout the lock, two threads processing "photo.jpg" simultaneously could both see the path as available and overwrite each other.
When one file fails, don't stop the batch:
# From BatchProcessor.process_file() (batch_processor.py:98-132)
try:
handler = MetadataFactory.get_handler(str(file))
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,
)
except Exception as e:
result = FileResult(
filepath=file,
success=False,
action="skipped",
error=str(e),
)
self._append_result(result)
return resultWhy this specific handling: Each file gets its own try/except. One corrupted file doesn't stop processing 999 others. The result object tracks success/failure for later reporting.
From src/utils/exceptions.py:
class MetadataException(Exception):
"""Base class for all metadata-related exceptions."""
class UnsupportedFormatError(MetadataException):
"""Raised when attempting to process unsupported file format."""
class MetadataNotFoundError(MetadataException):
"""Raised when no metadata is found in a file."""Why custom exceptions:
Callers can catch specific errors: except MetadataNotFoundError vs generic except Exception. Better than checking error message strings.
From tests/unit/test_image_handler.py:
def test_read_image_metadata(jpg_test_file):
processor = ImageHandler(jpg_test_file)
metadata = processor.read()
assert processor.metadata == metadata
assert processor.tags_to_delete is not None
assert isinstance(metadata, dict)What this tests:
- Read returns a dictionary
- Internal state (tags_to_delete) is populated
- Metadata field matches return value
From tests/integration/test_metadata_factory.py:
def test_save_processed_image_metadata(jpg_test_file):
output_dir = Path("./tests/assets/output")
output_dir.mkdir(parents=True, exist_ok=True)
handler = MetadataFactory.get_handler(str(jpg_test_file))
handler.read()
handler.wipe()
output_file = output_dir / Path(jpg_test_file).name
handler.save(str(output_file))
assert output_file.exists()Why these specific assertions: We test the full pipeline through the factory. If the output file exists and is valid, the entire read→wipe→save flow worked.
Symptom:
TypeError: expected str, bytes or os.PathLike object, not NoneType
Cause:
# Bad - no validation
def save(self, output_path):
shutil.copy2(self.filepath, output_path) # Crashes if NoneFix:
# Good - explicit validation
def save(self, output_path: str | None = None) -> None:
if not output_path:
raise ValueError("output_path is required")
# Now safe to useWhy this matters: Clear error messages help debugging. "output_path is required" is better than a cryptic TypeError.
Symptom:
RuntimeError: dictionary changed size during iteration
Cause:
# Bad
for tag in exif_dict[ifd]:
del exif_dict[ifd][tag] # Modifies dict while iteratingFix:
# Good
for tag in list(exif_dict[ifd]): # Iterate over copy
del exif_dict[ifd][tag]Symptom: Renamed PNG as .jpg processes incorrectly
Fix:
# From ImageHandler._detect_format()
with Image.open(Path(self.filepath)) as img:
if img.format is None:
raise UnsupportedFormatError("Could not detect format")
pillow_format = img.format.lower()
normalized = FORMAT_MAP.get(pillow_format)
if normalized is None:
raise UnsupportedFormatError(f"Unsupported: {pillow_format}")Use Pillow's actual format detection, not file extension.
# commands/scrub.py - UI concerns
console.print("🔎 Processing...")
progress = Progress(...)
# services/batch_processor.py - Business logic
def process_file(self, file: Path) -> FileResult:
# No UI code hereBenefit: You can use BatchProcessor in a web API without Rich/Typer dependencies. Commands stay thin, services stay reusable.
*Handlerclasses inherit from MetadataHandler*Processorclasses handle low-level format parsing*Resultdataclasses represent operation outcomesget_*functions retrieve data without side effectsprocess_*functions modify state or files
- typer (0.21.0): CLI framework with automatic help generation and type validation
- rich (14.0.0): Terminal formatting for progress bars and tables
- pillow (12.0.0): Image loading and EXIF access
- piexif (1.1.3): EXIF manipulation (Pillow is read-only for EXIF)
- pypdf (6.5.0): PDF metadata reading/writing
- openpyxl (3.1.5): Excel file handling
- python-pptx (1.0.2): PowerPoint metadata
- python-docx (1.2.0): Word document metadata
Check for vulnerabilities:
pip install safety
safety check --file pyproject.tomlIf you see vulnerabilities in dependencies, update to patched versions or find alternatives.
# Install in development mode
pip install -e .
# Run tests
pytest
# Type checking
mypy src/
# Linting
ruff check src/# Start development with auto-reload
# (Not applicable for CLI - just run directly)
mst scrub test.jpg
# Verbose logging for debugging
mst scrub test.jpg --verboseYou've seen how the code works. Now:
- Try the challenges - 04-CHALLENGES.md has extension ideas
- Add a feature - Try implementing video metadata support
- Read related projects - Study ExifTool source to see production patterns