AP-768: make file copy exclusions configurable as a Param (#81)

Author: anarchivist

mokelumne/dags/copy_files.py

@@ -1,5 +1,7 @@
 """File copy DAG to transfer files from one location to another."""
 
+# pyright: reportTypedDictNotRequiredAccess=false
+
 from __future__ import annotations
 
 import logging
@@ -41,6 +43,14 @@
             description="Directory where source files will be copied to",
             type="string",
         ),
+        "exclude_regex": Param(
+            default="",
+            title="Filename exclusion pattern",
+            description_md="""Regular expression to exclude files from the copy process. Defaults to excluding Thumbs.db and file or path name that starts with a period: `^(\.(.*)|(?i:Thumbs\.db))$`""", # pyright: ignore[reportInvalidStringEscapeSequence]
+            type=["string", "null"],
+            format="regex",
+            section="Exclude Files"
+        ),
     },
     tags=["file-transfer"],
 )
@@ -88,7 +98,9 @@ def build_manifest(source: str) -> str:
         """Build a manifest of all files under the source directory."""
 
         source_path = Path(source)
-        manifest = build_file_manifest(source_path)
+        ctx = get_current_context()
+        exclude_regex = ctx["params"].get("exclude_regex")
+        manifest = build_file_manifest(source_path, exclude_regex=exclude_regex)
 
         if not manifest:
             raise AirflowFailException(f"No files found in source: {source_path}")

mokelumne/util/file_transfer.py

@@ -2,6 +2,7 @@
 
 import hashlib
 import json
+import re
 import shutil
 
 from pathlib import Path
@@ -20,34 +21,20 @@ def sha256_for_file(path: Path) -> str:
     return sha256.hexdigest()
 
 
-def _should_include_manifest_entry(relative_path: Path, item: Path) -> bool:
-    """Return True when a path should be added to the manifest."""
-    if any(part.startswith(".") for part in relative_path.parts):
-        return False
+def build_file_manifest(
+        source_path: Path,
+        exclude_regex: str | None = None
+    ) -> Manifest:
+    exregex = re.compile(exclude_regex or r"^(\.(.*)|(?i:Thumbs\.db))$")
 
-    if item.name.casefold() in {"thumbs.db"}:
-        return False
-
-    return True
-
-def build_file_manifest(source_path: Path) -> Manifest:
-    files = []
-
-    for item in source_path.rglob("*"):
-        if not item.is_file():
-            continue
-
-        relative_path = item.relative_to(source_path)
-        if not _should_include_manifest_entry(relative_path, item):
-            continue
-
-        files.append(
-            {
-                "path": str(relative_path),
-                "size": item.stat().st_size,
-                "sha256": sha256_for_file(item),
-            }
+    files = [
+        { "path": str(f.relative_to(source_path)), "size": f.stat().st_size, "sha256": sha256_for_file(f) }
+        for f in source_path.rglob("*")
+        if (
+            f.is_file()
+            and not any(re.search(exregex, p) for p in f.relative_to(source_path).parts)
         )
+    ]
 
     return {
         "source_root": str(source_path),
@@ -74,11 +61,11 @@ def clean_destination_path(destination_path: Path) -> Path:
 def verify_file_manifest(destination_path: Path, manifest_path: Path) -> list[ManifestEntry]:
     """Verify all copied files exist at the destination."""
 
-    verification_report: ManifestEntry = []
+    verification_report: list[ManifestEntry] = []
     manifest = load_json(manifest_path)
 
     for entry in manifest["files"]:
-        relative_path = Path(str(entry["path"]))
+        relative_path = Path(str(entry.get("path")))
         expected_size = entry["size"]
         expected_sha256 = entry["sha256"]
 

test/unit/test_file_transfer.py

@@ -6,6 +6,27 @@
 
 from mokelumne.util import file_transfer
 
+@pytest.fixture(scope="session")
+def manifest_source_path(tmpdir_factory):
+    tmp_path = Path(tmpdir_factory.mktemp("manifest_source"))
+    visible_file = tmp_path / "visible.tif"
+    visible_file.write_text("hello", encoding="utf-8")
+
+    hidden_file = tmp_path / ".hidden.txt"
+    hidden_file.write_text("secret", encoding="utf-8")
+
+    hidden_dir = tmp_path / ".hidden"
+    hidden_dir.mkdir()
+    hidden_dir_file = hidden_dir / "nested.txt"
+    hidden_dir_file.write_text("nested", encoding="utf-8")
+
+    ds_store = tmp_path / ".DS_Store"
+    ds_store.write_text("folder metadata", encoding="utf-8")
+
+    thumbs_db = tmp_path / "Thumbs.db"
+    thumbs_db.write_text("thumbnail images db", encoding="utf-8")
+
+    return tmp_path
 
 class TestFileTransfer:
     """Tests for the Mokelumne file transfer module."""
@@ -69,31 +90,25 @@ def test_build_manifest_includes_nested_files(self, tmp_path: Path):
         }
         assert len(result) == 2
 
-    def test_build_manifest_excludes_hidden_and_excluded_files(self, tmp_path: Path):
+    @pytest.mark.parametrize(
+        "pattern,expected",
+        [
+            pytest.param(
+                None, {"visible.tif"}, id="with_default_regex"
+            ),
+            pytest.param(
+                r"^(visible\.tif|\.(.*))$", {"Thumbs.db"}, id="with_custom_regex"
+            )
+        ]
+    )
+    def test_build_manifest_excludes_hidden_and_excluded_files(self, manifest_source_path, pattern, expected):
         """Ensure that build_manifest skips hidden and system junk files."""
-        visible_file = tmp_path / "visible.tif"
-        visible_file.write_text("hello", encoding="utf-8")
-
-        hidden_file = tmp_path / ".hidden.txt"
-        hidden_file.write_text("secret", encoding="utf-8")
-
-        hidden_dir = tmp_path / ".hidden"
-        hidden_dir.mkdir()
-        hidden_dir_file = hidden_dir / "nested.txt"
-        hidden_dir_file.write_text("nested", encoding="utf-8")
-
-        ds_store = tmp_path / ".DS_Store"
-        ds_store.write_text("folder metadata", encoding="utf-8")
-
-        thumbs_db = tmp_path / "Thumbs.db"
-        thumbs_db.write_text("thumbnail images db", encoding="utf-8")
-
-        result = file_transfer.build_file_manifest(tmp_path)
+        result = file_transfer.build_file_manifest(manifest_source_path, exclude_regex=pattern)
 
         paths = {entry["path"] for entry in result["files"]}
 
-        assert paths == {"visible.tif"}
-        assert len(result["files"]) == 1
+        assert paths == expected
+        assert len(result["files"]) == len(expected)
 
     @pytest.mark.parametrize(
         ("input_path", "expected_path"),