metadata-scrubber-tool:

added pdf handler,
modifed the display to accepts pdf specific data,
and pdf format data,
added full test coverage
This commit is contained in:
HERITAGE-XION 2026-01-09 17:27:22 +01:00
parent 45ca2089b2
commit 955bb8f6a1
13 changed files with 590 additions and 10 deletions

View File

@ -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.services.pdf_handler import PDFHandler
from src.utils.exceptions import UnsupportedFormatError
@ -45,10 +46,10 @@ class MetadataFactory:
if Path(filepath).is_file():
if ext in [".jpg", ".jpeg", ".png"]:
return ImageHandler(filepath)
elif ext == ".pdf":
return PDFHandler(filepath)
# TODO: implement other handlers
# elif ext == ".pdf":
# return PDFHandler(filepath)
# elif ext == ".xlsx":
# return ExcelHandler(filepath)
# elif ext == ".pptx":

View File

@ -61,7 +61,7 @@ class MetadataHandler(ABC):
pass
@abstractmethod
def save(self, output_path: str | None = None) -> None:
def save(self, output_path: str) -> None:
"""
Save the processed file with cleaned metadata.

102
src/services/pdf_handler.py Normal file
View File

@ -0,0 +1,102 @@
"""
PDF metadata handler for PDF files.
Does not support encrypted files.
This module provides the PdfHandler class which implements the MetadataHandler
interface for PDF files. It delegates the actual metadata operations to
format-specific processors (PdfProcessor).
"""
import shutil
from pathlib import Path
from pypdf import PdfReader, PdfWriter
from src.services.metadata_handler import MetadataHandler
from src.utils.exceptions import (
MetadataNotFoundError,
MetadataReadingError,
UnsupportedFormatError,
)
class PDFHandler(MetadataHandler):
"""
PDF metadata handler for PDF files.
"""
def __init__(self, filepath: str):
"""
Initialize the pdf handler.
Args:
filepath: Path to the pdf file to process.
"""
super().__init__(filepath)
self.keys_to_delete: list[str] = []
def _detect_format(self) -> str:
"""
Detect actual pdf format using, not file extension.
Returns:
Normalized format string ('pdf').
Raises:
UnsupportedFormatError: If format is not supported or undetectable.
"""
normalise = Path(self.filepath).suffix.lower()
if normalise != ".pdf":
raise UnsupportedFormatError(f"Unsupported format: {normalise}")
return normalise[1:]
def read(self):
"""
Extract metadata from the file.
Uses actual format detection to select the appropriate processor.
"""
with PdfReader(Path(self.filepath)) as reader:
if reader.is_encrypted:
raise MetadataReadingError("File is encrypted.")
if reader.metadata is None:
raise MetadataNotFoundError("No metadata found in the file.")
for key, value in reader.metadata.items():
self.metadata[key] = value
self.keys_to_delete.append(key)
return self.metadata
def wipe(self) -> None:
"""
Remove metadata from PDF file.
"""
with PdfReader(Path(self.filepath)) as reader:
metadata = reader.metadata
if metadata is None:
raise MetadataNotFoundError("No metadata found in the file.")
for key in list(metadata):
if key in self.keys_to_delete:
del metadata[key]
self.processed_metadata = metadata
def save(self, output_path: str | Path | None = None) -> None:
"""
Save the changes to a copy of the original file.
"""
destination_file_path = Path(output_path) if output_path else None
if not destination_file_path:
raise ValueError("output_path is required")
shutil.copy2(self.filepath, destination_file_path)
with PdfReader(destination_file_path) as reader, PdfWriter() as writer:
# Copy all pages
for page in reader.pages:
writer.add_page(page)
writer.add_metadata(self.processed_metadata)
writer.write(destination_file_path)

View File

@ -31,6 +31,11 @@ def print_metadata_table(metadata: dict[str, Any]):
# Define the groups using simple lists of keys
groups = {
"📄 Document Info": [
"Author",
"/Author",
"/Creator",
],
"📸 Device Info": ["Make", "Model", "Software", "ExifVersion"],
"⚙️ Exposure Settings": [
"ExposureTime",
@ -49,7 +54,14 @@ def print_metadata_table(metadata: dict[str, Any]):
"Orientation",
"ResolutionUnit",
],
"📅 Dates": ["DateTime", "DateTimeOriginal", "DateTimeDigitized", "OffsetTime"],
"📅 Dates": [
"DateTime",
"DateTimeOriginal",
"DateTimeDigitized",
"OffsetTime",
"/CreationDate",
"/ModDate",
],
}
# Create the main table

View File

@ -28,3 +28,9 @@ class MetadataProcessingError(MetadataException):
"""Raised when an error occurs during metadata processing."""
pass
class MetadataReadingError(MetadataException):
"""Raised when an error occurs during metadata reading."""
pass

View File

@ -31,6 +31,10 @@ def clean_value(value: Any) -> str:
if isinstance(value, tuple) or isinstance(value, list):
return "/".join(map(str, value))
if isinstance(value, str) and value.startswith("D:"):
clean_iso = f"{value[2:6]}-{value[6:8]}-{value[8:10]}T{value[10:12]}:{value[12:14]}:{value[14:16]}"
return clean_iso
# Handle empty values
if value == "":
return "-"

Binary file not shown.

Binary file not shown.

View File

@ -12,6 +12,7 @@ import pytest
TESTS_DIR = Path(__file__).parent
ASSETS_DIR = TESTS_DIR / "assets"
TEST_IMAGES_DIR = ASSETS_DIR / "test_images"
TEST_PDFS_DIR = ASSETS_DIR / "test_pdfs"
@pytest.fixture
@ -46,3 +47,31 @@ def get_png_test_file() -> str:
def get_test_images_dir() -> str:
"""Get test images directory path as string."""
return str(TEST_IMAGES_DIR)
@pytest.fixture
def pdf_test_file() -> Path:
"""Return path to a PDF test file with metadata."""
return TEST_PDFS_DIR / "sample.pdf"
@pytest.fixture
def test_pdfs_dir() -> Path:
"""Return path to test PDFs directory."""
return TEST_PDFS_DIR
# String versions for parametrize (PDF)
def get_pdf_test_file() -> str:
"""Get PDF test file path as string."""
return str(TEST_PDFS_DIR / "sample.pdf")
def get_large_pdf_test_file() -> str:
"""Get large PDF test file path as string."""
return str(TEST_PDFS_DIR / "file-example_PDF_1MB.pdf")
def get_test_pdfs_dir() -> str:
"""Get test PDFs directory path as string."""
return str(TEST_PDFS_DIR)

View File

@ -6,7 +6,13 @@ from typer.testing import CliRunner
from src.main import app
# Import path helpers from conftest
from tests.conftest import get_jpg_test_file, get_png_test_file, get_test_images_dir
from tests.conftest import (
get_jpg_test_file,
get_pdf_test_file,
get_png_test_file,
get_test_images_dir,
get_test_pdfs_dir,
)
runner = CliRunner()
@ -14,6 +20,8 @@ runner = CliRunner()
JPG_TEST_FILE = get_jpg_test_file()
PNG_TEST_FILE = get_png_test_file()
TEST_DIR = get_test_images_dir()
PDF_TEST_FILE = get_pdf_test_file()
PDF_DIR = get_test_pdfs_dir()
# ============== Success Case Tests ==============
@ -74,3 +82,27 @@ def test_read_command_file_not_found():
assert result.exit_code == 2
assert "Invalid value for 'FILE_PATH'" in result.stderr
assert "does not exist" in result.stderr
# ============== PDF E2E Read Tests ==============
def test_read_command_pdf_single_file_success():
"""
Test the 'read' command with a single PDF file.
"""
result = runner.invoke(app, ["read", PDF_TEST_FILE])
assert result.exit_code == 0, f"Failed with: {result.stdout}"
assert "Reading" in result.stdout
assert Path(PDF_TEST_FILE).name in result.stdout
def test_read_command_recursive_pdf_success():
"""
Test the 'read' command with recursive directory processing for PDF.
"""
result = runner.invoke(app, ["read", PDF_DIR, "-r", "-ext", "pdf"])
assert result.exit_code == 0, f"Failed with: {result.stdout}"
assert "Reading" in result.stdout

View File

@ -12,7 +12,13 @@ from typer.testing import CliRunner
from src.main import app
# Import path helpers from conftest
from tests.conftest import get_jpg_test_file, get_png_test_file, get_test_images_dir
from tests.conftest import (
get_jpg_test_file,
get_pdf_test_file,
get_png_test_file,
get_test_images_dir,
get_test_pdfs_dir,
)
runner = CliRunner()
@ -20,6 +26,8 @@ runner = CliRunner()
JPG_TEST_FILE = get_jpg_test_file()
PNG_TEST_FILE = get_png_test_file()
EXAMPLES_DIR = get_test_images_dir()
PDF_TEST_FILE = get_pdf_test_file()
PDF_DIR = get_test_pdfs_dir()
@pytest.fixture
@ -127,3 +135,69 @@ def test_scrub_command_requires_recursive_with_ext():
result = runner.invoke(app, ["scrub", JPG_TEST_FILE, "-ext", "jpg"])
assert result.exit_code != 0
# ============== PDF E2E Tests ==============
def test_scrub_command_pdf_single_file_success(output_dir):
"""
Test the 'scrub' command with a single PDF file.
"""
result = runner.invoke(app, ["scrub", PDF_TEST_FILE, "--output", str(output_dir)])
assert result.exit_code == 0, f"Failed with: {result.stdout}"
# Check output file was created (has processed_ prefix)
output_file = output_dir / f"processed_{Path(PDF_TEST_FILE).name}"
assert output_file.exists()
def test_scrub_command_recursive_pdf_success(output_dir):
"""
Test the 'scrub' command with recursive directory processing for PDF.
"""
result = runner.invoke(
app, ["scrub", PDF_DIR, "-r", "-ext", "pdf", "--output", str(output_dir)]
)
assert result.exit_code == 0, f"Failed with: {result.stdout}"
# Check at least one output file was created
output_files = list(output_dir.glob("processed_*.pdf"))
assert len(output_files) > 0
def test_scrub_command_pdf_dry_run(output_dir):
"""
Test that --dry-run doesn't create PDF files.
"""
result = runner.invoke(
app, ["scrub", PDF_TEST_FILE, "--output", str(output_dir), "--dry-run"]
)
assert result.exit_code == 0, f"Failed with: {result.stdout}"
assert "DRY-RUN" in result.stdout
# No files should be created in dry-run mode
output_file = output_dir / f"processed_{Path(PDF_TEST_FILE).name}"
assert not output_file.exists()
def test_scrub_command_pdf_with_workers(output_dir):
"""
Test the --workers option for concurrent PDF processing.
"""
result = runner.invoke(
app,
[
"scrub",
PDF_DIR,
"-r",
"-ext",
"pdf",
"--output",
str(output_dir),
"--workers",
"2",
],
)
assert result.exit_code == 0, f"Failed with: {result.stdout}"

View File

@ -3,22 +3,26 @@ from pathlib import Path
import pytest
from src.core.jpeg_metadata import JpegProcessor
from src.core.png_metadata import PngProcessor
from src.services.metadata_factory import MetadataFactory
from src.services.pdf_handler import PDFHandler
from src.utils.exceptions import MetadataNotFoundError, UnsupportedFormatError
# Import path helpers from conftest
from tests.conftest import get_jpg_test_file, get_png_test_file
from tests.conftest import get_jpg_test_file, get_pdf_test_file, get_png_test_file
# Test file paths (cross-platform)
JPG_TEST_FILE = get_jpg_test_file()
PNG_TEST_FILE = get_png_test_file()
PDF_TEST_FILE = get_pdf_test_file()
# ============== Success Case Tests ==============
@pytest.mark.parametrize("x", [JPG_TEST_FILE, PNG_TEST_FILE])
def test_read_image_metadata(x):
def test_read_metadata(x):
"""
test for reading image metadata
"""
@ -30,8 +34,12 @@ def test_read_image_metadata(x):
# checks if metadata is not empty
assert handler.metadata == metadata
# checks if tags_to_delete or text_keys_to_delete is not empty
assert handler.tags_to_delete is not None or handler.text_keys_to_delete is not None
if isinstance(handler, JpegProcessor | PngProcessor):
# checks if tags_to_delete or text_keys_to_delete is not empty
assert (
handler.tags_to_delete is not None
or handler.text_keys_to_delete is not None
)
# checks if metadata is a dictionary
assert isinstance(metadata, dict)
@ -146,3 +154,92 @@ def test_save_without_output_path_raises_error():
handler.wipe()
with pytest.raises(ValueError):
handler.save(None)
# ============== PDF Integration Tests ==============
def test_read_pdf_metadata_via_factory():
"""
Test reading PDF metadata through MetadataFactory.
"""
assert Path(PDF_TEST_FILE).exists(), f"Test file not found: {PDF_TEST_FILE}"
handler = MetadataFactory.get_handler(PDF_TEST_FILE)
# Verify correct handler type is returned
assert isinstance(handler, PDFHandler)
metadata = handler.read()
assert handler.metadata == metadata
assert isinstance(metadata, dict)
assert handler.keys_to_delete is not None
def test_wipe_pdf_metadata_via_factory():
"""
Test wiping PDF metadata through MetadataFactory.
"""
assert Path(PDF_TEST_FILE).exists(), f"Test file not found: {PDF_TEST_FILE}"
handler = MetadataFactory.get_handler(PDF_TEST_FILE)
metadata = handler.read()
handler.wipe()
assert handler.processed_metadata != metadata
def test_save_processed_pdf_metadata_via_factory():
"""
Test saving processed PDF metadata through MetadataFactory.
"""
output_dir = Path("./tests/assets/output")
output_dir.mkdir(parents=True, exist_ok=True)
handler = MetadataFactory.get_handler(PDF_TEST_FILE)
metadata = handler.read()
handler.wipe()
assert handler.processed_metadata != metadata
output_file = output_dir / Path(PDF_TEST_FILE).name
handler.save(str(output_file))
assert output_file.exists()
shutil.rmtree(output_dir)
def test_pdf_format_detection_via_factory():
"""
Test that format detection returns 'pdf' for PDF files.
"""
handler = MetadataFactory.get_handler(PDF_TEST_FILE)
detected = handler._detect_format()
assert detected == "pdf"
def test_output_pdf_file_has_less_metadata():
"""
Test that the output PDF file has metadata stripped.
"""
output_dir = Path("./tests/assets/output")
output_dir.mkdir(parents=True, exist_ok=True)
# Process original file
handler = MetadataFactory.get_handler(PDF_TEST_FILE)
original_metadata = handler.read()
original_count = len(original_metadata)
handler.wipe()
# Save processed file
output_file = output_dir / Path(PDF_TEST_FILE).name
handler.save(str(output_file))
# Read output file and verify metadata is reduced or gone
try:
output_handler = MetadataFactory.get_handler(str(output_file))
output_metadata = output_handler.read()
assert len(output_metadata) < original_count
except MetadataNotFoundError:
# If no metadata found, that's expected for fully stripped files
pass
shutil.rmtree(output_dir)

View File

@ -0,0 +1,223 @@
"""
Unit tests for PDFHandler.
Tests the PDFHandler class in isolation, focusing on individual methods
and edge cases including encrypted PDFs, missing metadata, and corrupted files.
"""
import shutil
from pathlib import Path
import pytest
from pypdf.errors import PdfReadError
from src.services.pdf_handler import PDFHandler
from src.utils.exceptions import (
MetadataNotFoundError,
MetadataReadingError,
UnsupportedFormatError,
)
# Import path helpers from conftest
from tests.conftest import get_large_pdf_test_file, get_pdf_test_file
# Test file paths (cross-platform)
PDF_TEST_FILE = get_pdf_test_file()
LARGE_PDF_TEST_FILE = get_large_pdf_test_file()
# ============== Success Case Tests ==============
@pytest.mark.parametrize("pdf_file", [PDF_TEST_FILE, LARGE_PDF_TEST_FILE])
def test_read_pdf_metadata(pdf_file):
"""
Test reading metadata from PDF files.
Verifies that read() extracts metadata and populates keys_to_delete.
"""
assert Path(pdf_file).exists(), f"Test file not found: {pdf_file}"
handler = PDFHandler(pdf_file)
metadata = handler.read()
# Check metadata was extracted
assert handler.metadata == metadata
assert isinstance(metadata, dict)
# Check keys_to_delete is populated
assert handler.keys_to_delete is not None
assert len(handler.keys_to_delete) > 0
@pytest.mark.parametrize("pdf_file", [PDF_TEST_FILE, LARGE_PDF_TEST_FILE])
def test_wipe_pdf_metadata(pdf_file):
"""
Test wiping metadata from PDF files.
Verifies that wipe() removes metadata entries.
"""
assert Path(pdf_file).exists(), f"Test file not found: {pdf_file}"
handler = PDFHandler(pdf_file)
metadata = handler.read()
handler.wipe()
# processed_metadata should differ from original
assert handler.processed_metadata != metadata
@pytest.mark.parametrize("pdf_file", [PDF_TEST_FILE, LARGE_PDF_TEST_FILE])
def test_save_processed_pdf_metadata(pdf_file):
"""
Test saving processed PDF to output path.
"""
output_dir = Path("./tests/assets/output")
output_dir.mkdir(parents=True, exist_ok=True)
handler = PDFHandler(pdf_file)
handler.read()
handler.wipe()
output_file = output_dir / Path(pdf_file).name
handler.save(str(output_file))
# Verify output file exists
assert output_file.exists()
# Cleanup
shutil.rmtree(output_dir)
def test_format_detection_works():
"""
Test that _detect_format() correctly identifies PDF files.
"""
handler = PDFHandler(PDF_TEST_FILE)
detected = handler._detect_format()
assert detected == "pdf"
@pytest.mark.parametrize("pdf_file", [PDF_TEST_FILE, LARGE_PDF_TEST_FILE])
def test_output_file_has_less_metadata(pdf_file):
"""
Test that the output file has metadata stripped.
"""
output_dir = Path("./tests/assets/output")
output_dir.mkdir(parents=True, exist_ok=True)
# Process original file
handler = PDFHandler(pdf_file)
original_metadata = handler.read()
original_count = len(original_metadata)
handler.wipe()
# Save processed file
output_file = output_dir / Path(pdf_file).name
handler.save(str(output_file))
# Read output file and verify metadata is reduced or gone
try:
output_handler = PDFHandler(str(output_file))
output_metadata = output_handler.read()
assert len(output_metadata) < original_count
except MetadataNotFoundError:
# If no metadata found, that's expected for fully stripped files
pass
# Cleanup
shutil.rmtree(output_dir)
# ============== Error Case Tests ==============
def test_unsupported_format_raises_error(tmp_path):
"""
Test that non-PDF files raise UnsupportedFormatError.
"""
# Create a fake text file with .txt extension
fake_file = tmp_path / "test.txt"
fake_file.write_text("not a pdf")
handler = PDFHandler(str(fake_file))
with pytest.raises(UnsupportedFormatError):
handler._detect_format()
def test_save_without_output_path_raises_error():
"""
Test that save() raises ValueError when output_path is None.
"""
handler = PDFHandler(PDF_TEST_FILE)
handler.read()
handler.wipe()
with pytest.raises(ValueError):
handler.save(None)
# ============== Edge Case Tests ==============
def test_encrypted_pdf_raises_error(tmp_path):
"""
Test that encrypted PDFs raise MetadataReadingError.
Note: This test creates a minimal encrypted PDF for testing.
"""
# Create a minimal encrypted PDF structure
encrypted_pdf = tmp_path / "encrypted.pdf"
# Write minimal PDF with encryption marker
# This is a simplified test - in production you'd use a real encrypted PDF
encrypted_pdf.write_bytes(
b"%PDF-1.4\n"
b"1 0 obj\n<</Type/Catalog/Pages 2 0 R>>\nendobj\n"
b"2 0 obj\n<</Type/Pages/Kids[3 0 R]/Count 1>>\nendobj\n"
b"3 0 obj\n<</Type/Page/MediaBox[0 0 612 792]/Parent 2 0 R>>\nendobj\n"
b"xref\n0 4\n"
b"0000000000 65535 f \n"
b"0000000009 00000 n \n"
b"0000000058 00000 n \n"
b"0000000115 00000 n \n"
b"trailer\n<</Size 4/Root 1 0 R/Encrypt<</V 1>>>>\n"
b"startxref\n191\n%%EOF"
)
handler = PDFHandler(str(encrypted_pdf))
# Encrypted PDF should raise MetadataReadingError or PdfReadError
with pytest.raises((MetadataReadingError, PdfReadError, Exception)):
handler.read()
def test_pdf_without_metadata_raises_error(tmp_path):
"""
Test that PDFs with no metadata raise MetadataNotFoundError.
"""
# Create a minimal PDF without metadata
no_metadata_pdf = tmp_path / "no_metadata.pdf"
no_metadata_pdf.write_bytes(
b"%PDF-1.4\n"
b"1 0 obj\n<</Type/Catalog/Pages 2 0 R>>\nendobj\n"
b"2 0 obj\n<</Type/Pages/Kids[3 0 R]/Count 1>>\nendobj\n"
b"3 0 obj\n<</Type/Page/MediaBox[0 0 612 792]/Parent 2 0 R>>\nendobj\n"
b"xref\n0 4\n"
b"0000000000 65535 f \n"
b"0000000009 00000 n \n"
b"0000000058 00000 n \n"
b"0000000115 00000 n \n"
b"trailer\n<</Size 4/Root 1 0 R>>\n"
b"startxref\n191\n%%EOF"
)
handler = PDFHandler(str(no_metadata_pdf))
with pytest.raises((MetadataNotFoundError, Exception)):
handler.read()
def test_corrupted_pdf_graceful_error(tmp_path):
"""
Test that corrupted PDFs are handled gracefully.
"""
# Create a corrupted PDF (invalid structure)
corrupted_pdf = tmp_path / "corrupted.pdf"
corrupted_pdf.write_bytes(b"not a valid pdf content at all")
handler = PDFHandler(str(corrupted_pdf))
# Should raise PdfReadError or similar from pypdf
with pytest.raises((PdfReadError, Exception)):
handler.read()