Add cbz clean up script

Author: justinmstuart

README.md

@@ -11,6 +11,7 @@ Scripts are fully tested, linted, and compatible with both direct execution and
 - **scripts/trim_filenames.py**: Recursively removes a specified number of characters from the beginning of filenames in a directory and its subdirectories.
 - **scripts/mp3_metadata_stripper.py**: Recursively removes metadata from mp3 and m4a files in a directory and its subdirectories.
 - **scripts/cbz_processor.py**: Compresses and optimizes CBZ files in a directory (see script for details).
+- **scripts/cbz_clean_up.py**: Recursively deletes CBZ backup files ending in `_original.cbz`.
 
 ## Requirements
 
@@ -31,6 +32,7 @@ You can run any script directly or as a module:
 python -m scripts.trim_filenames
 python -m scripts.mp3_metadata_stripper
 python -m scripts.cbz_processor
+python -m scripts.cbz_clean_up
 ```
 
 ## Linting & Code Quality

scripts/cbz_clean_up.py

@@ -0,0 +1,82 @@
+"""
+CBZ Cleanup Utility
+
+Recursively scans a directory and removes backup CBZ files that end with
+"_original.cbz" (extension matching is case-insensitive).
+
+Features:
+    - Recursive processing of directories and subdirectories
+    - Deletes only files ending in "_original.cbz"
+    - Reads target directory from .env via CBZ_PROCESSOR_DIR
+    - Continues processing when a delete operation fails
+    - Prints a summary of deleted, skipped, and failed files
+
+Usage:
+    python -m scripts.cbz_clean_up
+
+Environment Variables:
+    CBZ_PROCESSOR_DIR (optional): Directory path to process files in
+"""
+
+import os
+
+from dotenv import load_dotenv
+
+from scripts.utils import get_directory_from_env_or_prompt, make_summary_dict, print_result
+
+TITLES = {
+    "success": "Successfully deleted",
+    "warning": "Skipped file",
+    "failed": "Failed to delete",
+}
+
+
+def _is_original_cbz(filename):
+    """Return True when filename ends with the exact '_original.cbz' pattern."""
+    base, extension = os.path.splitext(filename)
+    return extension.lower() == ".cbz" and base.endswith("_original")
+
+
+def clean_up_original_cbz_files(directory):
+    """
+    Recursively delete files matching the '_original.cbz' naming pattern.
+
+    Args:
+        directory (str): Root directory to scan.
+
+    Returns:
+        dict: Summary counts for successful, skipped, and failed operations.
+    """
+    success_count = 0
+    skipped_count = 0
+    failed_count = 0
+
+    for root, _, files in os.walk(directory):
+        for filename in files:
+            filepath = os.path.join(root, filename)
+            if not _is_original_cbz(filename):
+                skipped_count += 1
+                continue
+
+            try:
+                os.remove(filepath)
+                print(f"Deleted: {filepath}")
+                success_count += 1
+            except OSError as error:
+                print(f"Failed to delete {filepath}: {error}")
+                failed_count += 1
+
+    return make_summary_dict(success_count, skipped_count, failed_count)
+
+
+def main():
+    """Entry point for recursive cleanup of '_original.cbz' backup files."""
+    load_dotenv()
+    directory = get_directory_from_env_or_prompt("CBZ_PROCESSOR_DIR")
+    print(f"Processing directory: {directory}")
+    result = clean_up_original_cbz_files(directory)
+    print_result(result, TITLES)
+
+
+if __name__ == "__main__":
+    main()

scripts/cbz_processor.py

@@ -67,7 +67,7 @@ def _compress_and_write_image(img_path, temp_dir, zipf, quality, max_height):
         arcname = os.path.relpath(img_compressed_path, temp_dir)
         zipf.write(img_compressed_path, arcname)
 
-def compress_cbz(file_path, output_path=None, quality=80, max_height=1024):
+def compress_cbz(file_path, output_path=None, quality=100, max_height=1024):
     """
     Compress images in a CBZ file and optionally resize them.
     Returns the size saved in MB.
@@ -118,26 +118,27 @@ def process_cbz_files(directory, quality, max_height):
         "failed_count": 0
     }
 
-    def backup_and_compress_cbz(filepath, filename):
+    def backup_and_compress_cbz(filepath, filename, source_directory):
         base, ext = os.path.splitext(filename)
-        original_copy_path = os.path.join(directory, f"{base}_original{ext}")
+        original_copy_path = os.path.join(source_directory, f"{base}_original{ext}")
         print(f"Creating backup: {original_copy_path}")
         shutil.copy2(filepath, original_copy_path)
         size_saved = compress_cbz(filepath, quality=quality, max_height=max_height)
         print(f"✅ Optimized {os.path.basename(filepath)} | Size saved: {size_saved:.2f} MB")
         return size_saved
 
-    for filename in os.listdir(directory):
-        filepath = os.path.join(directory, filename)
-        if filename.endswith('.cbz'):
-            try:
-                backup_and_compress_cbz(filepath, filename)
-                counts["success_count"] += 1
-            except Exception as e:
-                print(f"🛑 Failed to process {filename}: {e}")
-                counts["failed_count"] += 1
-        else:
-            counts["skipped_count"] += 1
+    for source_directory, _, filenames in os.walk(directory):
+        for filename in filenames:
+            filepath = os.path.join(source_directory, filename)
+            if filename.lower().endswith('.cbz'):
+                try:
+                    backup_and_compress_cbz(filepath, filename, source_directory)
+                    counts["success_count"] += 1
+                except Exception as e:
+                    print(f"🛑 Failed to process {filename}: {e}")
+                    counts["failed_count"] += 1
+            else:
+                counts["skipped_count"] += 1
     return counts
 
 def main():

tests/test_cbz_clean_up.py

@@ -0,0 +1,135 @@
+"""Tests for scripts/cbz_clean_up.py."""
+
+import os
+
+import scripts.cbz_clean_up as cbz_clean_up
+from tests.test_helpers import run_cli_with_env
+
+
+def test_cleanup_deletes_matching_files(tmp_path):
+    """Delete only files ending in '_original.cbz'."""
+    keep_file = tmp_path / "book.cbz"
+    delete_file = tmp_path / "book_original.cbz"
+    non_cbz_file = tmp_path / "book_original.txt"
+
+    keep_file.write_text("keep")
+    delete_file.write_text("delete")
+    non_cbz_file.write_text("keep")
+
+    result = cbz_clean_up.clean_up_original_cbz_files(str(tmp_path))
+
+    assert result["success_count"] == 1
+    assert result["skipped_count"] == 2
+    assert result["failed_count"] == 0
+    assert keep_file.exists()
+    assert not delete_file.exists()
+    assert non_cbz_file.exists()
+
+
+def test_cleanup_is_recursive(tmp_path):
+    """Delete matching files from nested subdirectories."""
+    nested = tmp_path / "nested" / "deeper"
+    nested.mkdir(parents=True)
+    delete_file = nested / "comic_original.cbz"
+    keep_file = nested / "comic.cbz"
+
+    delete_file.write_text("delete")
+    keep_file.write_text("keep")
+
+    result = cbz_clean_up.clean_up_original_cbz_files(str(tmp_path))
+
+    assert result["success_count"] == 1
+    assert result["skipped_count"] == 1
+    assert result["failed_count"] == 0
+    assert not delete_file.exists()
+    assert keep_file.exists()
+
+
+def test_cleanup_does_not_delete_non_matching_original_names(tmp_path):
+    """Do not delete files that only contain '_original' as a substring."""
+    files = [
+        tmp_path / "book_original_backup.cbz",
+        tmp_path / "book.cbz",
+        tmp_path / "book_originals.cbz",
+        tmp_path / "other.txt",
+    ]
+    for file_path in files:
+        file_path.write_text("keep")
+
+    result = cbz_clean_up.clean_up_original_cbz_files(str(tmp_path))
+
+    assert result["success_count"] == 0
+    assert result["skipped_count"] == 4
+    assert result["failed_count"] == 0
+    for file_path in files:
+        assert file_path.exists()
+
+
+def test_cleanup_matches_case_insensitive_extension(tmp_path):
+    """Delete '_original.CBZ' with uppercase extension."""
+    delete_file = tmp_path / "book_original.CBZ"
+    delete_file.write_text("delete")
+
+    result = cbz_clean_up.clean_up_original_cbz_files(str(tmp_path))
+
+    assert result["success_count"] == 1
+    assert result["skipped_count"] == 0
+    assert result["failed_count"] == 0
+    assert not delete_file.exists()
+
+
+def test_cleanup_continue_on_delete_error(monkeypatch, tmp_path):
+    """Continue scanning and report failed_count when os.remove raises OSError."""
+    failing_file = tmp_path / "bad_original.cbz"
+    deleted_file = tmp_path / "good_original.cbz"
+    skipped_file = tmp_path / "keep.cbz"
+
+    failing_file.write_text("delete")
+    deleted_file.write_text("delete")
+    skipped_file.write_text("keep")
+
+    original_remove = os.remove
+
+    def fake_remove(path):
+        if path == str(failing_file):
+            raise OSError("permission denied")
+        return original_remove(path)
+
+    monkeypatch.setattr(cbz_clean_up.os, "remove", fake_remove)
+
+    result = cbz_clean_up.clean_up_original_cbz_files(str(tmp_path))
+
+    assert result["success_count"] == 1
+    assert result["skipped_count"] == 1
+    assert result["failed_count"] == 1
+    assert failing_file.exists()
+    assert not deleted_file.exists()
+    assert skipped_file.exists()
+
+
+def test_main_uses_cbz_processor_env_var(monkeypatch, tmp_path):
+    """main() should use CBZ_PROCESSOR_DIR without prompting."""
+    delete_file = tmp_path / "series_original.cbz"
+    delete_file.write_text("delete")
+
+    monkeypatch.setenv("CBZ_PROCESSOR_DIR", str(tmp_path))
+
+    captured = {}
+
+    def fake_print_result(stats, _titles):
+        captured["stats"] = stats
+
+    monkeypatch.setattr(cbz_clean_up, "print_result", fake_print_result)
+
+    cbz_clean_up.main()
+
+    assert captured["stats"]["success_count"] == 1
+    assert not delete_file.exists()
+
+
+def test_cli_entry(tmp_path):
+    """CLI module execution should run with CBZ_PROCESSOR_DIR set."""
+    env_vars = {"CBZ_PROCESSOR_DIR": str(tmp_path)}
+    result = run_cli_with_env("scripts.cbz_clean_up", env_vars)
+    assert result is not None
+    assert result.returncode == 0

tests/test_cbz_processor.py

@@ -490,3 +490,49 @@ def __exit__(self, *args):
 
     assert isinstance(size_saved, (int, float))
     assert output_path.exists()
+
+
+def test_process_cbz_files_recursive_discovers_nested_cbz(monkeypatch, tmp_path):
+    """Test process_cbz_files processes CBZ files in subdirectories."""
+    nested_dir = tmp_path / "nested"
+    nested_dir.mkdir()
+    cbz_path = nested_dir / "book.cbz"
+    cbz_path.write_bytes(b"dummy")
+
+    monkeypatch.setattr("scripts.cbz_processor.compress_cbz", lambda *_a, **_kw: 1.0)
+
+    result = process_cbz_files(str(tmp_path), 80, 1024)
+
+    assert result["success_count"] == 1
+    assert (nested_dir / "book_original.cbz").exists()
+
+
+def test_process_cbz_files_case_insensitive_extension(monkeypatch, tmp_path):
+    """Test process_cbz_files accepts uppercase/mixed-case .CBZ extension."""
+    cbz_path = tmp_path / "Book.CBZ"
+    cbz_path.write_bytes(b"dummy")
+
+    monkeypatch.setattr("scripts.cbz_processor.compress_cbz", lambda *_a, **_kw: 1.0)
+
+    result = process_cbz_files(str(tmp_path), 80, 1024)
+
+    assert result["success_count"] == 1
+    assert (tmp_path / "Book_original.CBZ").exists()
+
+
+def test_process_cbz_files_same_name_different_subdirectories(monkeypatch, tmp_path):
+    """Test backups for same basename in different folders do not collide."""
+    folder_a = tmp_path / "a"
+    folder_b = tmp_path / "b"
+    folder_a.mkdir()
+    folder_b.mkdir()
+    (folder_a / "comic.cbz").write_bytes(b"dummy")
+    (folder_b / "comic.cbz").write_bytes(b"dummy")
+
+    monkeypatch.setattr("scripts.cbz_processor.compress_cbz", lambda *_a, **_kw: 1.0)
+
+    result = process_cbz_files(str(tmp_path), 80, 1024)
+
+    assert result["success_count"] == 2
+    assert (folder_a / "comic_original.cbz").exists()
+    assert (folder_b / "comic_original.cbz").exists()