applied formatting from yapf style file
This commit is contained in:
parent
d0ab014b9d
commit
79d8b538f8
|
|
@ -46,4 +46,7 @@ Thumbs.db
|
||||||
|
|
||||||
# ============== Project Specific ==============
|
# ============== Project Specific ==============
|
||||||
# Output directory for processed images
|
# Output directory for processed images
|
||||||
tests/assets/output/
|
tests/assets/output/
|
||||||
|
|
||||||
|
|
||||||
|
.yapfignore/
|
||||||
|
|
@ -0,0 +1,4 @@
|
||||||
|
.venv/
|
||||||
|
venv/
|
||||||
|
env/
|
||||||
|
typings/
|
||||||
|
|
@ -5,13 +5,14 @@ from pathlib import Path
|
||||||
|
|
||||||
from PIL import Image, PngImagePlugin
|
from PIL import Image, PngImagePlugin
|
||||||
|
|
||||||
|
|
||||||
dest_dir = Path("tests/assets/test_images")
|
dest_dir = Path("tests/assets/test_images")
|
||||||
|
|
||||||
# Generate 45 PNG images with various metadata
|
# Generate 45 PNG images with various metadata
|
||||||
for i in range(1, 46):
|
for i in range(1, 46):
|
||||||
# Create a simple colored image
|
# Create a simple colored image
|
||||||
r, g, b = random.randint(50, 200), random.randint(50, 200), random.randint(50, 200)
|
r, g, b = random.randint(50, 200), random.randint(50, 200), random.randint(50, 200)
|
||||||
img = Image.new("RGB", (100, 100), color=(r, g, b))
|
img = Image.new("RGB", (100, 100), color = (r, g, b))
|
||||||
|
|
||||||
# Add PNG text metadata
|
# Add PNG text metadata
|
||||||
pnginfo = PngImagePlugin.PngInfo()
|
pnginfo = PngImagePlugin.PngInfo()
|
||||||
|
|
@ -23,7 +24,7 @@ for i in range(1, 46):
|
||||||
|
|
||||||
# Save with metadata
|
# Save with metadata
|
||||||
filename = dest_dir / f"generated_test_{i:02d}.png"
|
filename = dest_dir / f"generated_test_{i:02d}.png"
|
||||||
img.save(filename, pnginfo=pnginfo)
|
img.save(filename, pnginfo = pnginfo)
|
||||||
|
|
||||||
print("Generated 45 PNG images with metadata")
|
print("Generated 45 PNG images with metadata")
|
||||||
print(f"Total PNG count: {len(list(dest_dir.glob('*.png')))}")
|
print(f"Total PNG count: {len(list(dest_dir.glob('*.png')))}")
|
||||||
|
|
|
||||||
|
|
@ -15,6 +15,7 @@ from src.services.metadata_factory import MetadataFactory
|
||||||
from src.utils.display import print_metadata_table
|
from src.utils.display import print_metadata_table
|
||||||
from src.utils.get_target_files import get_target_files
|
from src.utils.get_target_files import get_target_files
|
||||||
|
|
||||||
|
|
||||||
console = Console()
|
console = Console()
|
||||||
log = logging.getLogger("metadata-scrubber")
|
log = logging.getLogger("metadata-scrubber")
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -27,6 +27,7 @@ from src.services.batch_processor import BatchProcessor
|
||||||
from src.utils.display import print_batch_summary
|
from src.utils.display import print_batch_summary
|
||||||
from src.utils.get_target_files import get_target_files
|
from src.utils.get_target_files import get_target_files
|
||||||
|
|
||||||
|
|
||||||
console = Console()
|
console = Console()
|
||||||
log = logging.getLogger("metadata-scrubber")
|
log = logging.getLogger("metadata-scrubber")
|
||||||
|
|
||||||
|
|
@ -93,7 +94,9 @@ def scrub(
|
||||||
|
|
||||||
# Show dry-run banner
|
# Show dry-run banner
|
||||||
if dry_run:
|
if dry_run:
|
||||||
console.print("\n[bold yellow]🔍 DRY-RUN MODE[/bold yellow] - No files will be modified.\n")
|
console.print(
|
||||||
|
"\n[bold yellow]🔍 DRY-RUN MODE[/bold yellow] - No files will be modified.\n"
|
||||||
|
)
|
||||||
|
|
||||||
# Collect files to process
|
# Collect files to process
|
||||||
files = list(get_target_files(file_path, ext)) if recursive else [file_path]
|
files = list(get_target_files(file_path, ext)) if recursive else [file_path]
|
||||||
|
|
@ -104,25 +107,31 @@ def scrub(
|
||||||
|
|
||||||
# Show worker count for batch operations
|
# Show worker count for batch operations
|
||||||
if len(files) > 1:
|
if len(files) > 1:
|
||||||
console.print(f"[dim]Processing {len(files)} files with {workers} workers...[/dim]\n")
|
console.print(
|
||||||
|
f"[dim]Processing {len(files)} files with {workers} workers...[/dim]\n"
|
||||||
|
)
|
||||||
|
|
||||||
# Initialize processor with worker count
|
# Initialize processor with worker count
|
||||||
processor = BatchProcessor(output_dir=output_dir, dry_run=dry_run, max_workers=workers)
|
processor = BatchProcessor(
|
||||||
|
output_dir = output_dir,
|
||||||
|
dry_run = dry_run,
|
||||||
|
max_workers = workers
|
||||||
|
)
|
||||||
|
|
||||||
# Process with thread-safe progress bar
|
# Process with thread-safe progress bar
|
||||||
with Progress(
|
with Progress(
|
||||||
SpinnerColumn(),
|
SpinnerColumn(),
|
||||||
TextColumn("[progress.description]{task.description}"),
|
TextColumn("[progress.description]{task.description}"),
|
||||||
BarColumn(),
|
BarColumn(),
|
||||||
MofNCompleteColumn(),
|
MofNCompleteColumn(),
|
||||||
TaskProgressColumn(),
|
TaskProgressColumn(),
|
||||||
TimeElapsedColumn(),
|
TimeElapsedColumn(),
|
||||||
console=console,
|
console = console,
|
||||||
refresh_per_second=10, # Smooth updates for concurrent processing
|
refresh_per_second = 10, # Smooth updates for concurrent processing
|
||||||
) as progress:
|
) as progress:
|
||||||
task_id = progress.add_task(
|
task_id = progress.add_task(
|
||||||
"[cyan]Scrubbing metadata...",
|
"[cyan]Scrubbing metadata...",
|
||||||
total=len(files),
|
total = len(files),
|
||||||
)
|
)
|
||||||
|
|
||||||
def on_file_complete(result):
|
def on_file_complete(result):
|
||||||
|
|
@ -130,12 +139,12 @@ def scrub(
|
||||||
status = "✓" if result.success else "✗"
|
status = "✓" if result.success else "✗"
|
||||||
progress.update(
|
progress.update(
|
||||||
task_id,
|
task_id,
|
||||||
description=f"[cyan]{status} {result.filepath.name}",
|
description = f"[cyan]{status} {result.filepath.name}",
|
||||||
advance=1,
|
advance = 1,
|
||||||
)
|
)
|
||||||
|
|
||||||
# Use concurrent batch processing
|
# Use concurrent batch processing
|
||||||
processor.process_batch(files, progress_callback=on_file_complete)
|
processor.process_batch(files, progress_callback = on_file_complete)
|
||||||
|
|
||||||
# Display summary
|
# Display summary
|
||||||
print_batch_summary(processor.get_summary())
|
print_batch_summary(processor.get_summary())
|
||||||
|
|
|
||||||
|
|
@ -14,26 +14,27 @@ from rich.console import Console
|
||||||
from src.services.metadata_factory import MetadataFactory
|
from src.services.metadata_factory import MetadataFactory
|
||||||
from src.services.report_generator import ReportGenerator
|
from src.services.report_generator import ReportGenerator
|
||||||
|
|
||||||
|
|
||||||
console = Console()
|
console = Console()
|
||||||
log = logging.getLogger("metadata-scrubber")
|
log = logging.getLogger("metadata-scrubber")
|
||||||
|
|
||||||
|
|
||||||
def verify(
|
def verify(
|
||||||
original_path: Path = typer.Argument(
|
original_path: Path = typer.Argument(
|
||||||
exists=True,
|
exists = True,
|
||||||
file_okay=True,
|
file_okay = True,
|
||||||
dir_okay=False,
|
dir_okay = False,
|
||||||
readable=True,
|
readable = True,
|
||||||
resolve_path=True,
|
resolve_path = True,
|
||||||
help="Path to the original file (before scrubbing)",
|
help = "Path to the original file (before scrubbing)",
|
||||||
),
|
),
|
||||||
processed_path: Path = typer.Argument(
|
processed_path: Path = typer.Argument(
|
||||||
exists=True,
|
exists = True,
|
||||||
file_okay=True,
|
file_okay = True,
|
||||||
dir_okay=False,
|
dir_okay = False,
|
||||||
readable=True,
|
readable = True,
|
||||||
resolve_path=True,
|
resolve_path = True,
|
||||||
help="Path to the processed file (after scrubbing)",
|
help = "Path to the processed file (after scrubbing)",
|
||||||
),
|
),
|
||||||
) -> None:
|
) -> None:
|
||||||
"""
|
"""
|
||||||
|
|
@ -58,10 +59,10 @@ def verify(
|
||||||
# Generate comparison report
|
# Generate comparison report
|
||||||
generator = ReportGenerator()
|
generator = ReportGenerator()
|
||||||
report = generator.generate_report(
|
report = generator.generate_report(
|
||||||
original_file=str(original_path),
|
original_file = str(original_path),
|
||||||
processed_file=str(processed_path),
|
processed_file = str(processed_path),
|
||||||
before_metadata=before_metadata,
|
before_metadata = before_metadata,
|
||||||
after_metadata=after_metadata,
|
after_metadata = after_metadata,
|
||||||
)
|
)
|
||||||
|
|
||||||
# Render the report
|
# Render the report
|
||||||
|
|
|
||||||
|
|
@ -34,7 +34,6 @@ class JpegProcessor:
|
||||||
tags_to_delete: List of EXIF tag IDs to remove.
|
tags_to_delete: List of EXIF tag IDs to remove.
|
||||||
data: Dict of extracted metadata with human-readable keys.
|
data: Dict of extracted metadata with human-readable keys.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self):
|
def __init__(self):
|
||||||
"""Initialize the JPEG processor with empty data structures."""
|
"""Initialize the JPEG processor with empty data structures."""
|
||||||
self.tags_to_delete: list[int] = []
|
self.tags_to_delete: list[int] = []
|
||||||
|
|
@ -69,15 +68,13 @@ class JpegProcessor:
|
||||||
|
|
||||||
# Get the human-readable name for the tag
|
# Get the human-readable name for the tag
|
||||||
tag_name = str(
|
tag_name = str(
|
||||||
tag_info.get("name", "Unknown Tag") if tag_info else "Unknown Tag"
|
tag_info.get("name",
|
||||||
|
"Unknown Tag") if tag_info else "Unknown Tag"
|
||||||
)
|
)
|
||||||
|
|
||||||
# Exclude tags necessary for image display integrity
|
# Exclude tags necessary for image display integrity
|
||||||
if (
|
if (tag_name == "Orientation" or tag_name == "ColorSpace"
|
||||||
tag_name == "Orientation"
|
or tag_name == "ExifTag"):
|
||||||
or tag_name == "ColorSpace"
|
|
||||||
or tag_name == "ExifTag"
|
|
||||||
):
|
|
||||||
continue
|
continue
|
||||||
|
|
||||||
# Save to list and dict
|
# Save to list and dict
|
||||||
|
|
|
||||||
|
|
@ -95,7 +95,7 @@ class PngProcessor:
|
||||||
found_metadata = True
|
found_metadata = True
|
||||||
if isinstance(value, bytes):
|
if isinstance(value, bytes):
|
||||||
try:
|
try:
|
||||||
value = value.decode("utf-8", errors="replace")
|
value = value.decode("utf-8", errors = "replace")
|
||||||
except Exception:
|
except Exception:
|
||||||
value = str(value)
|
value = str(value)
|
||||||
|
|
||||||
|
|
@ -143,7 +143,9 @@ class PngProcessor:
|
||||||
raise MetadataProcessingError(f"Error processing PNG EXIF: {str(e)}")
|
raise MetadataProcessingError(f"Error processing PNG EXIF: {str(e)}")
|
||||||
|
|
||||||
def get_clean_pnginfo(
|
def get_clean_pnginfo(
|
||||||
self, img: Image, keys_to_remove: list[str] | None = None
|
self,
|
||||||
|
img: Image,
|
||||||
|
keys_to_remove: list[str] | None = None
|
||||||
) -> PngImagePlugin.PngInfo | None:
|
) -> PngImagePlugin.PngInfo | None:
|
||||||
"""
|
"""
|
||||||
Create a new PngInfo with sensitive keys removed.
|
Create a new PngInfo with sensitive keys removed.
|
||||||
|
|
|
||||||
|
|
@ -20,10 +20,9 @@ from src.commands.verify import verify
|
||||||
from src.utils.logger import setup_logging
|
from src.utils.logger import setup_logging
|
||||||
|
|
||||||
# Initialize the Typer app with helpful defaults
|
# Initialize the Typer app with helpful defaults
|
||||||
app = typer.Typer(no_args_is_help=True, pretty_exceptions_show_locals=False)
|
app = typer.Typer(no_args_is_help = True, pretty_exceptions_show_locals = False)
|
||||||
log = logging.getLogger("metadata-scrubber")
|
log = logging.getLogger("metadata-scrubber")
|
||||||
|
|
||||||
|
|
||||||
__version__ = "0.3.0"
|
__version__ = "0.3.0"
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -71,9 +70,9 @@ def main(
|
||||||
# fmt: on
|
# fmt: on
|
||||||
|
|
||||||
# register commands
|
# register commands
|
||||||
app.command(name="read")(read)
|
app.command(name = "read")(read)
|
||||||
app.command(name="scrub")(scrub)
|
app.command(name = "scrub")(scrub)
|
||||||
app.command(name="verify")(verify)
|
app.command(name = "verify")(verify)
|
||||||
|
|
||||||
# run app
|
# run app
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
|
|
|
||||||
|
|
@ -19,6 +19,7 @@ from rich.console import Console
|
||||||
|
|
||||||
from src.services.metadata_factory import MetadataFactory
|
from src.services.metadata_factory import MetadataFactory
|
||||||
|
|
||||||
|
|
||||||
log = logging.getLogger("metadata-scrubber")
|
log = logging.getLogger("metadata-scrubber")
|
||||||
console = Console()
|
console = Console()
|
||||||
|
|
||||||
|
|
@ -44,7 +45,7 @@ class BatchSummary:
|
||||||
failed: int = 0
|
failed: int = 0
|
||||||
dry_run: bool = False
|
dry_run: bool = False
|
||||||
output_dir: Path | None = None
|
output_dir: Path | None = None
|
||||||
results: list[FileResult] = field(default_factory=list)
|
results: list[FileResult] = field(default_factory = list)
|
||||||
|
|
||||||
|
|
||||||
class BatchProcessor:
|
class BatchProcessor:
|
||||||
|
|
@ -55,7 +56,6 @@ class BatchProcessor:
|
||||||
Supports dry-run mode, automatic duplicate suffix handling,
|
Supports dry-run mode, automatic duplicate suffix handling,
|
||||||
and concurrent processing via ThreadPoolExecutor.
|
and concurrent processing via ThreadPoolExecutor.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
output_dir: str | None = None,
|
output_dir: str | None = None,
|
||||||
|
|
@ -99,12 +99,12 @@ class BatchProcessor:
|
||||||
if self.dry_run:
|
if self.dry_run:
|
||||||
# Verify the file can be handled (will raise if not)
|
# Verify the file can be handled (will raise if not)
|
||||||
MetadataFactory.get_handler(str(file))
|
MetadataFactory.get_handler(str(file))
|
||||||
output_path = self._get_unique_output_path(file, reserve=False)
|
output_path = self._get_unique_output_path(file, reserve = False)
|
||||||
result = FileResult(
|
result = FileResult(
|
||||||
filepath=file,
|
filepath = file,
|
||||||
success=True,
|
success = True,
|
||||||
action="dry-run",
|
action = "dry-run",
|
||||||
output_path=output_path,
|
output_path = output_path,
|
||||||
)
|
)
|
||||||
self._append_result(result)
|
self._append_result(result)
|
||||||
log.debug(f"[DRY-RUN] Would process: {file}")
|
log.debug(f"[DRY-RUN] Would process: {file}")
|
||||||
|
|
@ -121,10 +121,10 @@ class BatchProcessor:
|
||||||
handler.save(str(output_path))
|
handler.save(str(output_path))
|
||||||
|
|
||||||
result = FileResult(
|
result = FileResult(
|
||||||
filepath=file,
|
filepath = file,
|
||||||
success=True,
|
success = True,
|
||||||
action="scrubbed",
|
action = "scrubbed",
|
||||||
output_path=output_path,
|
output_path = output_path,
|
||||||
)
|
)
|
||||||
self._append_result(result)
|
self._append_result(result)
|
||||||
if log.isEnabledFor(logging.DEBUG):
|
if log.isEnabledFor(logging.DEBUG):
|
||||||
|
|
@ -137,10 +137,10 @@ class BatchProcessor:
|
||||||
self._cleanup_reserved_path(output_path)
|
self._cleanup_reserved_path(output_path)
|
||||||
|
|
||||||
result = FileResult(
|
result = FileResult(
|
||||||
filepath=file,
|
filepath = file,
|
||||||
success=False,
|
success = False,
|
||||||
action="skipped",
|
action = "skipped",
|
||||||
error=str(e),
|
error = str(e),
|
||||||
)
|
)
|
||||||
self._append_result(result)
|
self._append_result(result)
|
||||||
if log.isEnabledFor(logging.DEBUG):
|
if log.isEnabledFor(logging.DEBUG):
|
||||||
|
|
@ -151,7 +151,8 @@ class BatchProcessor:
|
||||||
def process_batch(
|
def process_batch(
|
||||||
self,
|
self,
|
||||||
files: Iterable[Path],
|
files: Iterable[Path],
|
||||||
progress_callback: Callable[[FileResult], None] | None = None,
|
progress_callback: Callable[[FileResult],
|
||||||
|
None] | None = None,
|
||||||
) -> list[FileResult]:
|
) -> list[FileResult]:
|
||||||
"""
|
"""
|
||||||
Process multiple files concurrently using ThreadPoolExecutor.
|
Process multiple files concurrently using ThreadPoolExecutor.
|
||||||
|
|
@ -170,10 +171,12 @@ class BatchProcessor:
|
||||||
return self.results
|
return self.results
|
||||||
|
|
||||||
# Used ThreadPoolExecutor for I/O-bound concurrent processing
|
# Used ThreadPoolExecutor for I/O-bound concurrent processing
|
||||||
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
|
with ThreadPoolExecutor(max_workers = self.max_workers) as executor:
|
||||||
# Submit all files for processing
|
# Submit all files for processing
|
||||||
future_to_file = {
|
future_to_file = {
|
||||||
executor.submit(self.process_file, file): file for file in file_list
|
executor.submit(self.process_file,
|
||||||
|
file): file
|
||||||
|
for file in file_list
|
||||||
}
|
}
|
||||||
|
|
||||||
# Collect results as they complete
|
# Collect results as they complete
|
||||||
|
|
@ -209,14 +212,14 @@ class BatchProcessor:
|
||||||
results_copy = list(self.results)
|
results_copy = list(self.results)
|
||||||
|
|
||||||
summary = BatchSummary(
|
summary = BatchSummary(
|
||||||
total=len(results_copy),
|
total = len(results_copy),
|
||||||
success=sum(
|
success = sum(
|
||||||
1 for r in results_copy if r.success and r.action == "scrubbed"
|
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),
|
skipped = sum(1 for r in results_copy if not r.success),
|
||||||
dry_run=self.dry_run,
|
dry_run = self.dry_run,
|
||||||
output_dir=self.output_dir,
|
output_dir = self.output_dir,
|
||||||
results=results_copy,
|
results = results_copy,
|
||||||
)
|
)
|
||||||
# Count dry-run as separate from success for clarity
|
# Count dry-run as separate from success for clarity
|
||||||
if self.dry_run:
|
if self.dry_run:
|
||||||
|
|
@ -266,7 +269,7 @@ class BatchProcessor:
|
||||||
"""
|
"""
|
||||||
with self._path_lock:
|
with self._path_lock:
|
||||||
# creates the destination directory if it doesn't exist
|
# creates the destination directory if it doesn't exist
|
||||||
self.output_dir.mkdir(parents=True, exist_ok=True)
|
self.output_dir.mkdir(parents = True, exist_ok = True)
|
||||||
|
|
||||||
base_name = file.stem
|
base_name = file.stem
|
||||||
extension = file.suffix
|
extension = file.suffix
|
||||||
|
|
|
||||||
|
|
@ -44,7 +44,6 @@ class ExcelHandler(MetadataHandler):
|
||||||
Attributes:
|
Attributes:
|
||||||
keys_to_delete: List of property names to be wiped.
|
keys_to_delete: List of property names to be wiped.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, filepath: str):
|
def __init__(self, filepath: str):
|
||||||
"""
|
"""
|
||||||
Initialize the Excel handler.
|
Initialize the Excel handler.
|
||||||
|
|
@ -66,7 +65,7 @@ class ExcelHandler(MetadataHandler):
|
||||||
UnsupportedFormatError: If file extension is not a supported Excel format.
|
UnsupportedFormatError: If file extension is not a supported Excel format.
|
||||||
"""
|
"""
|
||||||
ext = Path(self.filepath).suffix.lower()
|
ext = Path(self.filepath).suffix.lower()
|
||||||
normalised = FORMAT_MAP.get(ext[1:]) # Remove leading dot
|
normalised = FORMAT_MAP.get(ext[1 :]) # Remove leading dot
|
||||||
if normalised is None:
|
if normalised is None:
|
||||||
raise UnsupportedFormatError(f"Unsupported format: {ext}")
|
raise UnsupportedFormatError(f"Unsupported format: {ext}")
|
||||||
|
|
||||||
|
|
@ -152,7 +151,7 @@ class ExcelHandler(MetadataHandler):
|
||||||
# Use keep_vba=True for macro-enabled workbooks
|
# Use keep_vba=True for macro-enabled workbooks
|
||||||
detected_format = self._detect_format()
|
detected_format = self._detect_format()
|
||||||
if detected_format == "xlsm":
|
if detected_format == "xlsm":
|
||||||
wb = load_workbook(destination_file_path, keep_vba=True)
|
wb = load_workbook(destination_file_path, keep_vba = True)
|
||||||
else:
|
else:
|
||||||
wb = load_workbook(destination_file_path)
|
wb = load_workbook(destination_file_path)
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -39,7 +39,6 @@ class ImageHandler(MetadataHandler):
|
||||||
tags_to_delete: List of EXIF tags to remove during wipe operation.
|
tags_to_delete: List of EXIF tags to remove during wipe operation.
|
||||||
detected_format: Actual image format detected by Pillow.
|
detected_format: Actual image format detected by Pillow.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, filepath: str):
|
def __init__(self, filepath: str):
|
||||||
"""
|
"""
|
||||||
Initialize the image handler.
|
Initialize the image handler.
|
||||||
|
|
@ -48,10 +47,11 @@ class ImageHandler(MetadataHandler):
|
||||||
filepath: Path to the image file to process.
|
filepath: Path to the image file to process.
|
||||||
"""
|
"""
|
||||||
super().__init__(filepath)
|
super().__init__(filepath)
|
||||||
self.processors: dict[str, JpegProcessor | PngProcessor] = {
|
self.processors: dict[str,
|
||||||
"jpeg": JpegProcessor(),
|
JpegProcessor | PngProcessor] = {
|
||||||
"png": PngProcessor(),
|
"jpeg": JpegProcessor(),
|
||||||
}
|
"png": PngProcessor(),
|
||||||
|
}
|
||||||
self.tags_to_delete: list[int] = []
|
self.tags_to_delete: list[int] = []
|
||||||
self.detected_format: str | None = None
|
self.detected_format: str | None = None
|
||||||
self.text_keys_to_delete: list[str] = []
|
self.text_keys_to_delete: list[str] = []
|
||||||
|
|
@ -128,12 +128,16 @@ class ImageHandler(MetadataHandler):
|
||||||
|
|
||||||
with Image.open(Path(self.filepath)) as img:
|
with Image.open(Path(self.filepath)) as img:
|
||||||
self.processed_metadata = cast(
|
self.processed_metadata = cast(
|
||||||
dict[str, Any], processor.delete_metadata(img, self.tags_to_delete)
|
dict[str,
|
||||||
|
Any],
|
||||||
|
processor.delete_metadata(img,
|
||||||
|
self.tags_to_delete)
|
||||||
)
|
)
|
||||||
# For PNG, also get clean PngInfo
|
# For PNG, also get clean PngInfo
|
||||||
if isinstance(processor, PngProcessor):
|
if isinstance(processor, PngProcessor):
|
||||||
self.clean_pnginfo = processor.get_clean_pnginfo(
|
self.clean_pnginfo = processor.get_clean_pnginfo(
|
||||||
img, self.text_keys_to_delete
|
img,
|
||||||
|
self.text_keys_to_delete
|
||||||
)
|
)
|
||||||
|
|
||||||
def save(self, output_path: str | Path | None = None) -> None:
|
def save(self, output_path: str | Path | None = None) -> None:
|
||||||
|
|
@ -160,7 +164,7 @@ class ImageHandler(MetadataHandler):
|
||||||
shutil.copy2(self.filepath, destination_file_path)
|
shutil.copy2(self.filepath, destination_file_path)
|
||||||
with Image.open(destination_file_path) as img:
|
with Image.open(destination_file_path) as img:
|
||||||
exif_bytes = piexif.dump(self.processed_metadata)
|
exif_bytes = piexif.dump(self.processed_metadata)
|
||||||
img.save(destination_file_path, exif=exif_bytes)
|
img.save(destination_file_path, exif = exif_bytes)
|
||||||
|
|
||||||
elif actual_format == "png":
|
elif actual_format == "png":
|
||||||
# PNG: Open original, save fresh copy without metadata
|
# PNG: Open original, save fresh copy without metadata
|
||||||
|
|
@ -169,7 +173,9 @@ class ImageHandler(MetadataHandler):
|
||||||
# Preserve image mode and data integrity
|
# Preserve image mode and data integrity
|
||||||
img.save(
|
img.save(
|
||||||
destination_file_path,
|
destination_file_path,
|
||||||
format="PNG",
|
format = "PNG",
|
||||||
exif=None,
|
exif = None,
|
||||||
pnginfo=getattr(self, "clean_pnginfo", None),
|
pnginfo = getattr(self,
|
||||||
|
"clean_pnginfo",
|
||||||
|
None),
|
||||||
)
|
)
|
||||||
|
|
|
||||||
|
|
@ -28,7 +28,6 @@ class MetadataFactory:
|
||||||
- Images: .jpg, .jpeg, .png
|
- Images: .jpg, .jpeg, .png
|
||||||
- Future: .pdf, .docx, .xlsx, .pptx
|
- Future: .pdf, .docx, .xlsx, .pptx
|
||||||
"""
|
"""
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def get_handler(filepath: str):
|
def get_handler(filepath: str):
|
||||||
"""
|
"""
|
||||||
|
|
|
||||||
|
|
@ -23,7 +23,6 @@ class MetadataHandler(ABC):
|
||||||
metadata: Dictionary containing the extracted metadata.
|
metadata: Dictionary containing the extracted metadata.
|
||||||
processed_metadata: Dictionary containing metadata after processing.
|
processed_metadata: Dictionary containing metadata after processing.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, filepath: str):
|
def __init__(self, filepath: str):
|
||||||
"""
|
"""
|
||||||
Initialize the metadata handler.
|
Initialize the metadata handler.
|
||||||
|
|
|
||||||
|
|
@ -33,7 +33,6 @@ class PDFHandler(MetadataHandler):
|
||||||
Attributes:
|
Attributes:
|
||||||
keys_to_delete: List of metadata keys to be wiped.
|
keys_to_delete: List of metadata keys to be wiped.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, filepath: str):
|
def __init__(self, filepath: str):
|
||||||
"""
|
"""
|
||||||
Initialize the PDF handler.
|
Initialize the PDF handler.
|
||||||
|
|
@ -58,7 +57,7 @@ class PDFHandler(MetadataHandler):
|
||||||
if ext != ".pdf":
|
if ext != ".pdf":
|
||||||
raise UnsupportedFormatError(f"Unsupported format: {ext}")
|
raise UnsupportedFormatError(f"Unsupported format: {ext}")
|
||||||
|
|
||||||
return ext[1:] # Return 'pdf' without the dot
|
return ext[1 :] # Return 'pdf' without the dot
|
||||||
|
|
||||||
def read(self) -> dict[str, Any]:
|
def read(self) -> dict[str, Any]:
|
||||||
"""
|
"""
|
||||||
|
|
|
||||||
|
|
@ -64,7 +64,6 @@ class PowerpointHandler(MetadataHandler):
|
||||||
Attributes:
|
Attributes:
|
||||||
keys_to_delete: List of property names to be wiped.
|
keys_to_delete: List of property names to be wiped.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, filepath: str):
|
def __init__(self, filepath: str):
|
||||||
"""
|
"""
|
||||||
Initialize the PowerPoint handler.
|
Initialize the PowerPoint handler.
|
||||||
|
|
@ -86,7 +85,7 @@ class PowerpointHandler(MetadataHandler):
|
||||||
UnsupportedFormatError: If file extension is not a supported PowerPoint format.
|
UnsupportedFormatError: If file extension is not a supported PowerPoint format.
|
||||||
"""
|
"""
|
||||||
ext = Path(self.filepath).suffix.lower()
|
ext = Path(self.filepath).suffix.lower()
|
||||||
normalised = FORMAT_MAP.get(ext[1:]) # Remove leading dot
|
normalised = FORMAT_MAP.get(ext[1 :]) # Remove leading dot
|
||||||
if normalised is None:
|
if normalised is None:
|
||||||
raise UnsupportedFormatError(f"Unsupported format: {ext}")
|
raise UnsupportedFormatError(f"Unsupported format: {ext}")
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -48,7 +48,7 @@ class ComparisonReport:
|
||||||
|
|
||||||
original_file: str
|
original_file: str
|
||||||
processed_file: str
|
processed_file: str
|
||||||
comparisons: list[PropertyComparison] = field(default_factory=list)
|
comparisons: list[PropertyComparison] = field(default_factory = list)
|
||||||
status: VerificationStatus = VerificationStatus.CLEAN
|
status: VerificationStatus = VerificationStatus.CLEAN
|
||||||
removed_count: int = 0
|
removed_count: int = 0
|
||||||
preserved_count: int = 0
|
preserved_count: int = 0
|
||||||
|
|
@ -80,8 +80,10 @@ class ReportGenerator:
|
||||||
|
|
||||||
def compare(
|
def compare(
|
||||||
self,
|
self,
|
||||||
before: dict[str, Any],
|
before: dict[str,
|
||||||
after: dict[str, Any],
|
Any],
|
||||||
|
after: dict[str,
|
||||||
|
Any],
|
||||||
preserved_keys: set[str] | None = None,
|
preserved_keys: set[str] | None = None,
|
||||||
) -> ComparisonReport:
|
) -> ComparisonReport:
|
||||||
"""
|
"""
|
||||||
|
|
@ -99,8 +101,8 @@ class ReportGenerator:
|
||||||
preserved_keys = self.PRESERVED_PROPERTIES
|
preserved_keys = self.PRESERVED_PROPERTIES
|
||||||
|
|
||||||
report = ComparisonReport(
|
report = ComparisonReport(
|
||||||
original_file="",
|
original_file = "",
|
||||||
processed_file="",
|
processed_file = "",
|
||||||
)
|
)
|
||||||
|
|
||||||
# Get all unique keys from both dictionaries
|
# Get all unique keys from both dictionaries
|
||||||
|
|
@ -112,17 +114,20 @@ class ReportGenerator:
|
||||||
|
|
||||||
# Determine property status
|
# Determine property status
|
||||||
status = self._determine_status(
|
status = self._determine_status(
|
||||||
key, before_value, after_value, preserved_keys
|
key,
|
||||||
|
before_value,
|
||||||
|
after_value,
|
||||||
|
preserved_keys
|
||||||
)
|
)
|
||||||
|
|
||||||
is_sensitive = key not in preserved_keys
|
is_sensitive = key not in preserved_keys
|
||||||
|
|
||||||
comparison = PropertyComparison(
|
comparison = PropertyComparison(
|
||||||
name=key,
|
name = key,
|
||||||
before_value=before_value,
|
before_value = before_value,
|
||||||
after_value=after_value,
|
after_value = after_value,
|
||||||
status=status,
|
status = status,
|
||||||
is_sensitive=is_sensitive,
|
is_sensitive = is_sensitive,
|
||||||
)
|
)
|
||||||
report.comparisons.append(comparison)
|
report.comparisons.append(comparison)
|
||||||
|
|
||||||
|
|
@ -214,14 +219,14 @@ class ReportGenerator:
|
||||||
}
|
}
|
||||||
|
|
||||||
table = Table(
|
table = Table(
|
||||||
title="[bold]Verification Report[/bold]",
|
title = "[bold]Verification Report[/bold]",
|
||||||
show_header=True,
|
show_header = True,
|
||||||
header_style="bold",
|
header_style = "bold",
|
||||||
)
|
)
|
||||||
|
|
||||||
table.add_column("Property", style="cyan")
|
table.add_column("Property", style = "cyan")
|
||||||
table.add_column("Before", style="dim")
|
table.add_column("Before", style = "dim")
|
||||||
table.add_column("After", justify="left")
|
table.add_column("After", justify = "left")
|
||||||
|
|
||||||
for comp in report.comparisons:
|
for comp in report.comparisons:
|
||||||
before_str = self._format_value(comp.before_value)
|
before_str = self._format_value(comp.before_value)
|
||||||
|
|
@ -248,7 +253,7 @@ class ReportGenerator:
|
||||||
return "[dim]None[/dim]"
|
return "[dim]None[/dim]"
|
||||||
if isinstance(value, str):
|
if isinstance(value, str):
|
||||||
if len(value) > 30:
|
if len(value) > 30:
|
||||||
return value[:27] + "..."
|
return value[: 27] + "..."
|
||||||
return value if value.strip() else "[dim]-[/dim]"
|
return value if value.strip() else "[dim]-[/dim]"
|
||||||
return str(value)
|
return str(value)
|
||||||
|
|
||||||
|
|
@ -283,8 +288,10 @@ class ReportGenerator:
|
||||||
self,
|
self,
|
||||||
original_file: str,
|
original_file: str,
|
||||||
processed_file: str,
|
processed_file: str,
|
||||||
before_metadata: dict[str, Any],
|
before_metadata: dict[str,
|
||||||
after_metadata: dict[str, Any],
|
Any],
|
||||||
|
after_metadata: dict[str,
|
||||||
|
Any],
|
||||||
) -> ComparisonReport:
|
) -> ComparisonReport:
|
||||||
"""
|
"""
|
||||||
Generate a full comparison report between two files.
|
Generate a full comparison report between two files.
|
||||||
|
|
|
||||||
|
|
@ -61,7 +61,6 @@ class WorddocHandler(MetadataHandler):
|
||||||
Attributes:
|
Attributes:
|
||||||
keys_to_delete: List of property names to be wiped.
|
keys_to_delete: List of property names to be wiped.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, filepath: str):
|
def __init__(self, filepath: str):
|
||||||
"""
|
"""
|
||||||
Initialize the Word document handler.
|
Initialize the Word document handler.
|
||||||
|
|
@ -83,7 +82,7 @@ class WorddocHandler(MetadataHandler):
|
||||||
UnsupportedFormatError: If file extension is not a supported Word format.
|
UnsupportedFormatError: If file extension is not a supported Word format.
|
||||||
"""
|
"""
|
||||||
ext = Path(self.filepath).suffix.lower()
|
ext = Path(self.filepath).suffix.lower()
|
||||||
normalised = FORMAT_MAP.get(ext[1:]) # Remove leading dot
|
normalised = FORMAT_MAP.get(ext[1 :]) # Remove leading dot
|
||||||
if normalised is None:
|
if normalised is None:
|
||||||
raise UnsupportedFormatError(f"Unsupported format: {ext}")
|
raise UnsupportedFormatError(f"Unsupported format: {ext}")
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -15,6 +15,7 @@ from rich.text import Text
|
||||||
|
|
||||||
from src.utils.formatter import clean_value
|
from src.utils.formatter import clean_value
|
||||||
|
|
||||||
|
|
||||||
console = Console()
|
console = Console()
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -37,7 +38,10 @@ def print_metadata_table(metadata: dict[str, Any]):
|
||||||
"/Author",
|
"/Author",
|
||||||
"/Creator",
|
"/Creator",
|
||||||
],
|
],
|
||||||
"📸 Device Info": ["Make", "Model", "Software", "ExifVersion"],
|
"📸 Device Info": ["Make",
|
||||||
|
"Model",
|
||||||
|
"Software",
|
||||||
|
"ExifVersion"],
|
||||||
"⚙️ Exposure Settings": [
|
"⚙️ Exposure Settings": [
|
||||||
"ExposureTime",
|
"ExposureTime",
|
||||||
"FNumber",
|
"FNumber",
|
||||||
|
|
@ -68,9 +72,9 @@ def print_metadata_table(metadata: dict[str, Any]):
|
||||||
}
|
}
|
||||||
|
|
||||||
# Create the main table
|
# Create the main table
|
||||||
table = Table(box=box.ROUNDED, show_header=True, header_style="bold magenta")
|
table = Table(box = box.ROUNDED, show_header = True, header_style = "bold magenta")
|
||||||
table.add_column("Property", style="cyan")
|
table.add_column("Property", style = "cyan")
|
||||||
table.add_column("Value", style="green")
|
table.add_column("Value", style = "green")
|
||||||
|
|
||||||
# Track which keys we have displayed to handle the "leftovers"
|
# Track which keys we have displayed to handle the "leftovers"
|
||||||
displayed_keys = set()
|
displayed_keys = set()
|
||||||
|
|
@ -82,7 +86,7 @@ def print_metadata_table(metadata: dict[str, Any]):
|
||||||
|
|
||||||
if section_data:
|
if section_data:
|
||||||
# Add a section row (acts as a sub-header)
|
# Add a section row (acts as a sub-header)
|
||||||
table.add_row(Text(section_name, style="bold yellow"), "")
|
table.add_row(Text(section_name, style = "bold yellow"), "")
|
||||||
|
|
||||||
for key, val in section_data.items():
|
for key, val in section_data.items():
|
||||||
table.add_row(f" {key}", clean_value(val))
|
table.add_row(f" {key}", clean_value(val))
|
||||||
|
|
@ -98,13 +102,16 @@ def print_metadata_table(metadata: dict[str, Any]):
|
||||||
if k not in displayed_keys and k != "JPEGInterchangeFormat"
|
if k not in displayed_keys and k != "JPEGInterchangeFormat"
|
||||||
} # skip binary blobs
|
} # skip binary blobs
|
||||||
if leftovers:
|
if leftovers:
|
||||||
table.add_row(Text("📝 Other", style="bold yellow"), "")
|
table.add_row(Text("📝 Other", style = "bold yellow"), "")
|
||||||
for key, val in leftovers.items():
|
for key, val in leftovers.items():
|
||||||
table.add_row(f" {key}", clean_value(val))
|
table.add_row(f" {key}", clean_value(val))
|
||||||
|
|
||||||
# Print nicely inside a panel
|
# Print nicely inside a panel
|
||||||
console.print(
|
console.print(
|
||||||
Panel(table, title="Metadata Report", border_style="blue", expand=False)
|
Panel(table,
|
||||||
|
title = "Metadata Report",
|
||||||
|
border_style = "blue",
|
||||||
|
expand = False)
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -116,9 +123,9 @@ def print_batch_summary(summary) -> None:
|
||||||
summary: BatchSummary object with processing statistics.
|
summary: BatchSummary object with processing statistics.
|
||||||
"""
|
"""
|
||||||
# Build the summary table
|
# Build the summary table
|
||||||
table = Table(box=box.ROUNDED, show_header=False, expand=False)
|
table = Table(box = box.ROUNDED, show_header = False, expand = False)
|
||||||
table.add_column("Metric", style="cyan")
|
table.add_column("Metric", style = "cyan")
|
||||||
table.add_column("Value", style="green")
|
table.add_column("Value", style = "green")
|
||||||
|
|
||||||
if summary.dry_run:
|
if summary.dry_run:
|
||||||
table.add_row("Mode", "[yellow]DRY-RUN (no changes made)[/yellow]")
|
table.add_row("Mode", "[yellow]DRY-RUN (no changes made)[/yellow]")
|
||||||
|
|
@ -135,8 +142,8 @@ def print_batch_summary(summary) -> None:
|
||||||
failed = [r for r in summary.results if not r.success]
|
failed = [r for r in summary.results if not r.success]
|
||||||
if failed:
|
if failed:
|
||||||
table.add_section()
|
table.add_section()
|
||||||
table.add_row(Text("Failed files:", style="bold red"), "")
|
table.add_row(Text("Failed files:", style = "bold red"), "")
|
||||||
for result in failed[:5]: # Show max 5 failures
|
for result in failed[: 5]: # Show max 5 failures
|
||||||
table.add_row(f" {result.filepath.name}", f"[dim]{result.error}[/dim]")
|
table.add_row(f" {result.filepath.name}", f"[dim]{result.error}[/dim]")
|
||||||
if len(failed) > 5:
|
if len(failed) > 5:
|
||||||
table.add_row("", f"[dim]... and {len(failed) - 5} more[/dim]")
|
table.add_row("", f"[dim]... and {len(failed) - 5} more[/dim]")
|
||||||
|
|
@ -152,4 +159,9 @@ def print_batch_summary(summary) -> None:
|
||||||
title = "⚠️ Scrub Complete (with warnings)"
|
title = "⚠️ Scrub Complete (with warnings)"
|
||||||
border_style = "yellow"
|
border_style = "yellow"
|
||||||
|
|
||||||
console.print(Panel(table, title=title, border_style=border_style, expand=False))
|
console.print(
|
||||||
|
Panel(table,
|
||||||
|
title = title,
|
||||||
|
border_style = border_style,
|
||||||
|
expand = False)
|
||||||
|
)
|
||||||
|
|
|
||||||
|
|
@ -32,14 +32,14 @@ def setup_logging(verbose: bool = False):
|
||||||
logging.getLogger().handlers.clear()
|
logging.getLogger().handlers.clear()
|
||||||
|
|
||||||
logging.basicConfig(
|
logging.basicConfig(
|
||||||
level=level,
|
level = level,
|
||||||
format="%(message)s",
|
format = "%(message)s",
|
||||||
datefmt="[%X]",
|
datefmt = "[%X]",
|
||||||
handlers=[
|
handlers = [
|
||||||
RichHandler(
|
RichHandler(
|
||||||
rich_tracebacks=True, # Beautiful colorful stack traces
|
rich_tracebacks = True, # Beautiful colorful stack traces
|
||||||
markup=True, # Allow [bold red] styles in logs
|
markup = True, # Allow [bold red] styles in logs
|
||||||
show_path=False, # Hide line number (cleaner for CLI tools)
|
show_path = False, # Hide line number (cleaner for CLI tools)
|
||||||
)
|
)
|
||||||
],
|
],
|
||||||
)
|
)
|
||||||
|
|
|
||||||
|
|
@ -22,6 +22,7 @@ from tests.conftest import (
|
||||||
get_xlsx_test_file,
|
get_xlsx_test_file,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
runner = CliRunner()
|
runner = CliRunner()
|
||||||
|
|
||||||
# Test file paths (cross-platform)
|
# Test file paths (cross-platform)
|
||||||
|
|
|
||||||
|
|
@ -22,6 +22,7 @@ from tests.conftest import (
|
||||||
get_xlsx_test_file,
|
get_xlsx_test_file,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
runner = CliRunner()
|
runner = CliRunner()
|
||||||
|
|
||||||
# Test file paths (cross-platform)
|
# Test file paths (cross-platform)
|
||||||
|
|
@ -38,7 +39,7 @@ XLSX_DIR = get_test_xlsx_dir()
|
||||||
def output_dir(tmp_path):
|
def output_dir(tmp_path):
|
||||||
"""Create isolated output directory for each test using tmp_path."""
|
"""Create isolated output directory for each test using tmp_path."""
|
||||||
output = tmp_path / "output"
|
output = tmp_path / "output"
|
||||||
output.mkdir(parents=True, exist_ok=True)
|
output.mkdir(parents = True, exist_ok = True)
|
||||||
return output
|
return output
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -58,7 +59,14 @@ def test_scrub_command_single_file_success(x, output_dir):
|
||||||
def test_scrub_command_recursive_jpg_success(output_dir):
|
def test_scrub_command_recursive_jpg_success(output_dir):
|
||||||
"""Test the 'scrub' command with recursive directory processing for JPG."""
|
"""Test the 'scrub' command with recursive directory processing for JPG."""
|
||||||
result = runner.invoke(
|
result = runner.invoke(
|
||||||
app, ["scrub", EXAMPLES_DIR, "-r", "-ext", "jpg", "--output", str(output_dir)]
|
app,
|
||||||
|
["scrub",
|
||||||
|
EXAMPLES_DIR,
|
||||||
|
"-r",
|
||||||
|
"-ext",
|
||||||
|
"jpg",
|
||||||
|
"--output",
|
||||||
|
str(output_dir)]
|
||||||
)
|
)
|
||||||
|
|
||||||
assert result.exit_code == 0, f"Failed with: {result.stdout}"
|
assert result.exit_code == 0, f"Failed with: {result.stdout}"
|
||||||
|
|
@ -69,7 +77,12 @@ def test_scrub_command_recursive_jpg_success(output_dir):
|
||||||
def test_scrub_command_dry_run(output_dir):
|
def test_scrub_command_dry_run(output_dir):
|
||||||
"""Test that --dry-run doesn't create files."""
|
"""Test that --dry-run doesn't create files."""
|
||||||
result = runner.invoke(
|
result = runner.invoke(
|
||||||
app, ["scrub", JPG_TEST_FILE, "--output", str(output_dir), "--dry-run"]
|
app,
|
||||||
|
["scrub",
|
||||||
|
JPG_TEST_FILE,
|
||||||
|
"--output",
|
||||||
|
str(output_dir),
|
||||||
|
"--dry-run"]
|
||||||
)
|
)
|
||||||
|
|
||||||
assert result.exit_code == 0, f"Failed with: {result.stdout}"
|
assert result.exit_code == 0, f"Failed with: {result.stdout}"
|
||||||
|
|
@ -136,7 +149,14 @@ def test_scrub_command_pdf_single_file_success(output_dir):
|
||||||
def test_scrub_command_recursive_pdf_success(output_dir):
|
def test_scrub_command_recursive_pdf_success(output_dir):
|
||||||
"""Test the 'scrub' command with recursive directory processing for PDF."""
|
"""Test the 'scrub' command with recursive directory processing for PDF."""
|
||||||
result = runner.invoke(
|
result = runner.invoke(
|
||||||
app, ["scrub", PDF_DIR, "-r", "-ext", "pdf", "--output", str(output_dir)]
|
app,
|
||||||
|
["scrub",
|
||||||
|
PDF_DIR,
|
||||||
|
"-r",
|
||||||
|
"-ext",
|
||||||
|
"pdf",
|
||||||
|
"--output",
|
||||||
|
str(output_dir)]
|
||||||
)
|
)
|
||||||
|
|
||||||
assert result.exit_code == 0, f"Failed with: {result.stdout}"
|
assert result.exit_code == 0, f"Failed with: {result.stdout}"
|
||||||
|
|
@ -147,7 +167,12 @@ def test_scrub_command_recursive_pdf_success(output_dir):
|
||||||
def test_scrub_command_pdf_dry_run(output_dir):
|
def test_scrub_command_pdf_dry_run(output_dir):
|
||||||
"""Test that --dry-run doesn't create PDF files."""
|
"""Test that --dry-run doesn't create PDF files."""
|
||||||
result = runner.invoke(
|
result = runner.invoke(
|
||||||
app, ["scrub", PDF_TEST_FILE, "--output", str(output_dir), "--dry-run"]
|
app,
|
||||||
|
["scrub",
|
||||||
|
PDF_TEST_FILE,
|
||||||
|
"--output",
|
||||||
|
str(output_dir),
|
||||||
|
"--dry-run"]
|
||||||
)
|
)
|
||||||
|
|
||||||
assert result.exit_code == 0, f"Failed with: {result.stdout}"
|
assert result.exit_code == 0, f"Failed with: {result.stdout}"
|
||||||
|
|
@ -171,7 +196,14 @@ def test_scrub_command_xlsx_single_file_success(output_dir):
|
||||||
def test_scrub_command_recursive_xlsx_success(output_dir):
|
def test_scrub_command_recursive_xlsx_success(output_dir):
|
||||||
"""Test the 'scrub' command with recursive directory processing for Excel."""
|
"""Test the 'scrub' command with recursive directory processing for Excel."""
|
||||||
result = runner.invoke(
|
result = runner.invoke(
|
||||||
app, ["scrub", XLSX_DIR, "-r", "-ext", "xlsx", "--output", str(output_dir)]
|
app,
|
||||||
|
["scrub",
|
||||||
|
XLSX_DIR,
|
||||||
|
"-r",
|
||||||
|
"-ext",
|
||||||
|
"xlsx",
|
||||||
|
"--output",
|
||||||
|
str(output_dir)]
|
||||||
)
|
)
|
||||||
|
|
||||||
assert result.exit_code == 0, f"Failed with: {result.stdout}"
|
assert result.exit_code == 0, f"Failed with: {result.stdout}"
|
||||||
|
|
@ -182,7 +214,12 @@ def test_scrub_command_recursive_xlsx_success(output_dir):
|
||||||
def test_scrub_command_xlsx_dry_run(output_dir):
|
def test_scrub_command_xlsx_dry_run(output_dir):
|
||||||
"""Test that --dry-run doesn't create Excel files."""
|
"""Test that --dry-run doesn't create Excel files."""
|
||||||
result = runner.invoke(
|
result = runner.invoke(
|
||||||
app, ["scrub", XLSX_TEST_FILE, "--output", str(output_dir), "--dry-run"]
|
app,
|
||||||
|
["scrub",
|
||||||
|
XLSX_TEST_FILE,
|
||||||
|
"--output",
|
||||||
|
str(output_dir),
|
||||||
|
"--dry-run"]
|
||||||
)
|
)
|
||||||
|
|
||||||
assert result.exit_code == 0, f"Failed with: {result.stdout}"
|
assert result.exit_code == 0, f"Failed with: {result.stdout}"
|
||||||
|
|
@ -232,7 +269,14 @@ def test_scrub_command_recursive_pptx_success(output_dir):
|
||||||
|
|
||||||
PPTX_DIR = get_test_pptx_dir()
|
PPTX_DIR = get_test_pptx_dir()
|
||||||
result = runner.invoke(
|
result = runner.invoke(
|
||||||
app, ["scrub", PPTX_DIR, "-r", "-ext", "pptx", "--output", str(output_dir)]
|
app,
|
||||||
|
["scrub",
|
||||||
|
PPTX_DIR,
|
||||||
|
"-r",
|
||||||
|
"-ext",
|
||||||
|
"pptx",
|
||||||
|
"--output",
|
||||||
|
str(output_dir)]
|
||||||
)
|
)
|
||||||
|
|
||||||
assert result.exit_code == 0, f"Failed with: {result.stdout}"
|
assert result.exit_code == 0, f"Failed with: {result.stdout}"
|
||||||
|
|
@ -246,7 +290,12 @@ def test_scrub_command_pptx_dry_run(output_dir):
|
||||||
|
|
||||||
PPTX_TEST_FILE = get_pptx_test_file()
|
PPTX_TEST_FILE = get_pptx_test_file()
|
||||||
result = runner.invoke(
|
result = runner.invoke(
|
||||||
app, ["scrub", PPTX_TEST_FILE, "--output", str(output_dir), "--dry-run"]
|
app,
|
||||||
|
["scrub",
|
||||||
|
PPTX_TEST_FILE,
|
||||||
|
"--output",
|
||||||
|
str(output_dir),
|
||||||
|
"--dry-run"]
|
||||||
)
|
)
|
||||||
|
|
||||||
assert result.exit_code == 0, f"Failed with: {result.stdout}"
|
assert result.exit_code == 0, f"Failed with: {result.stdout}"
|
||||||
|
|
@ -299,7 +348,14 @@ def test_scrub_command_recursive_docx_success(output_dir):
|
||||||
|
|
||||||
DOCX_DIR = get_test_docx_dir()
|
DOCX_DIR = get_test_docx_dir()
|
||||||
result = runner.invoke(
|
result = runner.invoke(
|
||||||
app, ["scrub", DOCX_DIR, "-r", "-ext", "docx", "--output", str(output_dir)]
|
app,
|
||||||
|
["scrub",
|
||||||
|
DOCX_DIR,
|
||||||
|
"-r",
|
||||||
|
"-ext",
|
||||||
|
"docx",
|
||||||
|
"--output",
|
||||||
|
str(output_dir)]
|
||||||
)
|
)
|
||||||
|
|
||||||
assert result.exit_code == 0, f"Failed with: {result.stdout}"
|
assert result.exit_code == 0, f"Failed with: {result.stdout}"
|
||||||
|
|
@ -313,7 +369,12 @@ def test_scrub_command_docx_dry_run(output_dir):
|
||||||
|
|
||||||
DOCX_TEST_FILE = get_docx_test_file()
|
DOCX_TEST_FILE = get_docx_test_file()
|
||||||
result = runner.invoke(
|
result = runner.invoke(
|
||||||
app, ["scrub", DOCX_TEST_FILE, "--output", str(output_dir), "--dry-run"]
|
app,
|
||||||
|
["scrub",
|
||||||
|
DOCX_TEST_FILE,
|
||||||
|
"--output",
|
||||||
|
str(output_dir),
|
||||||
|
"--dry-run"]
|
||||||
)
|
)
|
||||||
|
|
||||||
assert result.exit_code == 0, f"Failed with: {result.stdout}"
|
assert result.exit_code == 0, f"Failed with: {result.stdout}"
|
||||||
|
|
|
||||||
|
|
@ -13,6 +13,7 @@ from typer.testing import CliRunner
|
||||||
|
|
||||||
from src.main import app
|
from src.main import app
|
||||||
|
|
||||||
|
|
||||||
runner = CliRunner()
|
runner = CliRunner()
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -37,7 +38,11 @@ def test_verify_command_jpeg_clean_status(output_dir):
|
||||||
|
|
||||||
# First scrub the file
|
# First scrub the file
|
||||||
scrub_result = runner.invoke(
|
scrub_result = runner.invoke(
|
||||||
app, ["scrub", JPG_TEST_FILE, "--output", str(output_dir)]
|
app,
|
||||||
|
["scrub",
|
||||||
|
JPG_TEST_FILE,
|
||||||
|
"--output",
|
||||||
|
str(output_dir)]
|
||||||
)
|
)
|
||||||
assert scrub_result.exit_code == 0
|
assert scrub_result.exit_code == 0
|
||||||
|
|
||||||
|
|
@ -90,7 +95,11 @@ def test_verify_command_pdf_clean_status(output_dir):
|
||||||
|
|
||||||
# First scrub the file
|
# First scrub the file
|
||||||
scrub_result = runner.invoke(
|
scrub_result = runner.invoke(
|
||||||
app, ["scrub", PDF_TEST_FILE, "--output", str(output_dir)]
|
app,
|
||||||
|
["scrub",
|
||||||
|
PDF_TEST_FILE,
|
||||||
|
"--output",
|
||||||
|
str(output_dir)]
|
||||||
)
|
)
|
||||||
assert scrub_result.exit_code == 0
|
assert scrub_result.exit_code == 0
|
||||||
|
|
||||||
|
|
@ -116,7 +125,11 @@ def test_verify_command_xlsx_clean_status(output_dir):
|
||||||
|
|
||||||
# First scrub the file
|
# First scrub the file
|
||||||
scrub_result = runner.invoke(
|
scrub_result = runner.invoke(
|
||||||
app, ["scrub", XLSX_TEST_FILE, "--output", str(output_dir)]
|
app,
|
||||||
|
["scrub",
|
||||||
|
XLSX_TEST_FILE,
|
||||||
|
"--output",
|
||||||
|
str(output_dir)]
|
||||||
)
|
)
|
||||||
assert scrub_result.exit_code == 0
|
assert scrub_result.exit_code == 0
|
||||||
|
|
||||||
|
|
@ -142,7 +155,11 @@ def test_verify_command_pptx_clean_status(output_dir):
|
||||||
|
|
||||||
# First scrub the file
|
# First scrub the file
|
||||||
scrub_result = runner.invoke(
|
scrub_result = runner.invoke(
|
||||||
app, ["scrub", PPTX_TEST_FILE, "--output", str(output_dir)]
|
app,
|
||||||
|
["scrub",
|
||||||
|
PPTX_TEST_FILE,
|
||||||
|
"--output",
|
||||||
|
str(output_dir)]
|
||||||
)
|
)
|
||||||
assert scrub_result.exit_code == 0
|
assert scrub_result.exit_code == 0
|
||||||
|
|
||||||
|
|
@ -168,7 +185,11 @@ def test_verify_command_docx_clean_status(output_dir):
|
||||||
|
|
||||||
# First scrub the file
|
# First scrub the file
|
||||||
scrub_result = runner.invoke(
|
scrub_result = runner.invoke(
|
||||||
app, ["scrub", DOCX_TEST_FILE, "--output", str(output_dir)]
|
app,
|
||||||
|
["scrub",
|
||||||
|
DOCX_TEST_FILE,
|
||||||
|
"--output",
|
||||||
|
str(output_dir)]
|
||||||
)
|
)
|
||||||
assert scrub_result.exit_code == 0
|
assert scrub_result.exit_code == 0
|
||||||
|
|
||||||
|
|
@ -189,7 +210,10 @@ def test_verify_command_docx_clean_status(output_dir):
|
||||||
def test_verify_command_original_file_not_found():
|
def test_verify_command_original_file_not_found():
|
||||||
"""Test verify command handles missing original file."""
|
"""Test verify command handles missing original file."""
|
||||||
result = runner.invoke(
|
result = runner.invoke(
|
||||||
app, ["verify", "nonexistent_file.jpg", "also_nonexistent.jpg"]
|
app,
|
||||||
|
["verify",
|
||||||
|
"nonexistent_file.jpg",
|
||||||
|
"also_nonexistent.jpg"]
|
||||||
)
|
)
|
||||||
|
|
||||||
assert result.exit_code != 0
|
assert result.exit_code != 0
|
||||||
|
|
|
||||||
|
|
@ -46,8 +46,7 @@ def test_read_image_metadata(x):
|
||||||
|
|
||||||
if isinstance(handler, (JpegProcessor, PngProcessor)):
|
if isinstance(handler, (JpegProcessor, PngProcessor)):
|
||||||
assert (
|
assert (
|
||||||
handler.tags_to_delete is not None
|
handler.tags_to_delete is not None or handler.text_keys_to_delete is not None
|
||||||
or handler.text_keys_to_delete is not None
|
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -66,7 +65,7 @@ def test_wipe_image_metadata(x):
|
||||||
def test_save_processed_image_metadata(x):
|
def test_save_processed_image_metadata(x):
|
||||||
"""Test saving processed image metadata through factory."""
|
"""Test saving processed image metadata through factory."""
|
||||||
output_dir = Path("./tests/assets/output")
|
output_dir = Path("./tests/assets/output")
|
||||||
output_dir.mkdir(parents=True, exist_ok=True)
|
output_dir.mkdir(parents = True, exist_ok = True)
|
||||||
|
|
||||||
handler = MetadataFactory.get_handler(str(x))
|
handler = MetadataFactory.get_handler(str(x))
|
||||||
handler.read()
|
handler.read()
|
||||||
|
|
@ -116,7 +115,7 @@ def test_wipe_pdf_metadata_via_factory():
|
||||||
def test_save_processed_pdf_metadata_via_factory():
|
def test_save_processed_pdf_metadata_via_factory():
|
||||||
"""Test saving processed PDF metadata through MetadataFactory."""
|
"""Test saving processed PDF metadata through MetadataFactory."""
|
||||||
output_dir = Path("./tests/assets/output")
|
output_dir = Path("./tests/assets/output")
|
||||||
output_dir.mkdir(parents=True, exist_ok=True)
|
output_dir.mkdir(parents = True, exist_ok = True)
|
||||||
|
|
||||||
handler = MetadataFactory.get_handler(PDF_TEST_FILE)
|
handler = MetadataFactory.get_handler(PDF_TEST_FILE)
|
||||||
handler.read()
|
handler.read()
|
||||||
|
|
@ -163,7 +162,7 @@ def test_wipe_excel_metadata_via_factory():
|
||||||
def test_save_processed_excel_metadata_via_factory():
|
def test_save_processed_excel_metadata_via_factory():
|
||||||
"""Test saving processed Excel metadata through MetadataFactory."""
|
"""Test saving processed Excel metadata through MetadataFactory."""
|
||||||
output_dir = Path("./tests/assets/output")
|
output_dir = Path("./tests/assets/output")
|
||||||
output_dir.mkdir(parents=True, exist_ok=True)
|
output_dir.mkdir(parents = True, exist_ok = True)
|
||||||
|
|
||||||
handler = MetadataFactory.get_handler(XLSX_TEST_FILE)
|
handler = MetadataFactory.get_handler(XLSX_TEST_FILE)
|
||||||
handler.read()
|
handler.read()
|
||||||
|
|
@ -220,7 +219,7 @@ def test_save_processed_pptx_metadata_via_factory():
|
||||||
|
|
||||||
PPTX_TEST_FILE = get_pptx_test_file()
|
PPTX_TEST_FILE = get_pptx_test_file()
|
||||||
output_dir = Path("./tests/assets/output")
|
output_dir = Path("./tests/assets/output")
|
||||||
output_dir.mkdir(parents=True, exist_ok=True)
|
output_dir.mkdir(parents = True, exist_ok = True)
|
||||||
|
|
||||||
handler = MetadataFactory.get_handler(PPTX_TEST_FILE)
|
handler = MetadataFactory.get_handler(PPTX_TEST_FILE)
|
||||||
handler.read()
|
handler.read()
|
||||||
|
|
@ -279,7 +278,7 @@ def test_save_processed_docx_metadata_via_factory():
|
||||||
|
|
||||||
DOCX_TEST_FILE = get_docx_test_file()
|
DOCX_TEST_FILE = get_docx_test_file()
|
||||||
output_dir = Path("./tests/assets/output")
|
output_dir = Path("./tests/assets/output")
|
||||||
output_dir.mkdir(parents=True, exist_ok=True)
|
output_dir.mkdir(parents = True, exist_ok = True)
|
||||||
|
|
||||||
handler = MetadataFactory.get_handler(DOCX_TEST_FILE)
|
handler = MetadataFactory.get_handler(DOCX_TEST_FILE)
|
||||||
handler.read()
|
handler.read()
|
||||||
|
|
|
||||||
|
|
@ -68,7 +68,7 @@ def test_save_processed_excel_metadata(xlsx_file):
|
||||||
Test saving processed Excel to output path.
|
Test saving processed Excel to output path.
|
||||||
"""
|
"""
|
||||||
output_dir = Path("./tests/assets/output")
|
output_dir = Path("./tests/assets/output")
|
||||||
output_dir.mkdir(parents=True, exist_ok=True)
|
output_dir.mkdir(parents = True, exist_ok = True)
|
||||||
|
|
||||||
handler = ExcelHandler(xlsx_file)
|
handler = ExcelHandler(xlsx_file)
|
||||||
handler.read()
|
handler.read()
|
||||||
|
|
@ -99,7 +99,7 @@ def test_output_file_has_less_metadata(xlsx_file):
|
||||||
Test that the output file has metadata stripped.
|
Test that the output file has metadata stripped.
|
||||||
"""
|
"""
|
||||||
output_dir = Path("./tests/assets/output")
|
output_dir = Path("./tests/assets/output")
|
||||||
output_dir.mkdir(parents=True, exist_ok=True)
|
output_dir.mkdir(parents = True, exist_ok = True)
|
||||||
|
|
||||||
# Process original file
|
# Process original file
|
||||||
handler = ExcelHandler(xlsx_file)
|
handler = ExcelHandler(xlsx_file)
|
||||||
|
|
|
||||||
|
|
@ -32,8 +32,7 @@ def test_read_image_metadata(x):
|
||||||
|
|
||||||
# checks if tags_to_delete or text_keys_to_delete is not empty
|
# checks if tags_to_delete or text_keys_to_delete is not empty
|
||||||
assert (
|
assert (
|
||||||
processor.tags_to_delete is not None
|
processor.tags_to_delete is not None or processor.text_keys_to_delete is not None
|
||||||
or processor.text_keys_to_delete is not None
|
|
||||||
)
|
)
|
||||||
|
|
||||||
# checks if metadata is a dictionary
|
# checks if metadata is a dictionary
|
||||||
|
|
@ -62,7 +61,7 @@ def test_save_processed_image_metadata(x):
|
||||||
"""
|
"""
|
||||||
# creates output directory
|
# creates output directory
|
||||||
output_dir = Path("./tests/assets/output")
|
output_dir = Path("./tests/assets/output")
|
||||||
output_dir.mkdir(parents=True, exist_ok=True)
|
output_dir.mkdir(parents = True, exist_ok = True)
|
||||||
|
|
||||||
processor = ImageHandler(x)
|
processor = ImageHandler(x)
|
||||||
metadata = processor.read()
|
metadata = processor.read()
|
||||||
|
|
@ -116,7 +115,7 @@ def test_output_file_has_less_metadata(x):
|
||||||
Test that the output file has metadata stripped
|
Test that the output file has metadata stripped
|
||||||
"""
|
"""
|
||||||
output_dir = Path("./tests/assets/output")
|
output_dir = Path("./tests/assets/output")
|
||||||
output_dir.mkdir(parents=True, exist_ok=True)
|
output_dir.mkdir(parents = True, exist_ok = True)
|
||||||
|
|
||||||
# Process original file
|
# Process original file
|
||||||
processor = ImageHandler(x)
|
processor = ImageHandler(x)
|
||||||
|
|
|
||||||
|
|
@ -69,7 +69,7 @@ def test_save_processed_pdf_metadata(pdf_file):
|
||||||
Test saving processed PDF to output path.
|
Test saving processed PDF to output path.
|
||||||
"""
|
"""
|
||||||
output_dir = Path("./tests/assets/output")
|
output_dir = Path("./tests/assets/output")
|
||||||
output_dir.mkdir(parents=True, exist_ok=True)
|
output_dir.mkdir(parents = True, exist_ok = True)
|
||||||
|
|
||||||
handler = PDFHandler(pdf_file)
|
handler = PDFHandler(pdf_file)
|
||||||
handler.read()
|
handler.read()
|
||||||
|
|
@ -100,7 +100,7 @@ def test_output_file_has_less_metadata(pdf_file):
|
||||||
Test that the output file has metadata stripped.
|
Test that the output file has metadata stripped.
|
||||||
"""
|
"""
|
||||||
output_dir = Path("./tests/assets/output")
|
output_dir = Path("./tests/assets/output")
|
||||||
output_dir.mkdir(parents=True, exist_ok=True)
|
output_dir.mkdir(parents = True, exist_ok = True)
|
||||||
|
|
||||||
# Process original file
|
# Process original file
|
||||||
handler = PDFHandler(pdf_file)
|
handler = PDFHandler(pdf_file)
|
||||||
|
|
|
||||||
|
|
@ -66,7 +66,7 @@ def test_save_processed_pptx_metadata(pptx_file):
|
||||||
Test saving processed PowerPoint to output path.
|
Test saving processed PowerPoint to output path.
|
||||||
"""
|
"""
|
||||||
output_dir = Path("./tests/assets/output")
|
output_dir = Path("./tests/assets/output")
|
||||||
output_dir.mkdir(parents=True, exist_ok=True)
|
output_dir.mkdir(parents = True, exist_ok = True)
|
||||||
|
|
||||||
handler = PowerpointHandler(pptx_file)
|
handler = PowerpointHandler(pptx_file)
|
||||||
handler.read()
|
handler.read()
|
||||||
|
|
@ -97,7 +97,7 @@ def test_output_file_has_wiped_metadata(pptx_file):
|
||||||
Test that the output file has metadata wiped.
|
Test that the output file has metadata wiped.
|
||||||
"""
|
"""
|
||||||
output_dir = Path("./tests/assets/output")
|
output_dir = Path("./tests/assets/output")
|
||||||
output_dir.mkdir(parents=True, exist_ok=True)
|
output_dir.mkdir(parents = True, exist_ok = True)
|
||||||
|
|
||||||
# Process original file
|
# Process original file
|
||||||
handler = PowerpointHandler(pptx_file)
|
handler = PowerpointHandler(pptx_file)
|
||||||
|
|
|
||||||
|
|
@ -66,7 +66,7 @@ def test_save_processed_docx_metadata(docx_file):
|
||||||
Test saving processed Word document to output path.
|
Test saving processed Word document to output path.
|
||||||
"""
|
"""
|
||||||
output_dir = Path("./tests/assets/output")
|
output_dir = Path("./tests/assets/output")
|
||||||
output_dir.mkdir(parents=True, exist_ok=True)
|
output_dir.mkdir(parents = True, exist_ok = True)
|
||||||
|
|
||||||
handler = WorddocHandler(docx_file)
|
handler = WorddocHandler(docx_file)
|
||||||
handler.read()
|
handler.read()
|
||||||
|
|
@ -97,7 +97,7 @@ def test_output_file_has_wiped_metadata(docx_file):
|
||||||
Test that the output file has metadata wiped.
|
Test that the output file has metadata wiped.
|
||||||
"""
|
"""
|
||||||
output_dir = Path("./tests/assets/output")
|
output_dir = Path("./tests/assets/output")
|
||||||
output_dir.mkdir(parents=True, exist_ok=True)
|
output_dir.mkdir(parents = True, exist_ok = True)
|
||||||
|
|
||||||
# Process original file
|
# Process original file
|
||||||
handler = WorddocHandler(docx_file)
|
handler = WorddocHandler(docx_file)
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue