Expand missing import paths as glob patterns

mapillary_tools/utils.py

@@ -6,12 +6,15 @@
 from __future__ import annotations
 
 import concurrent.futures
+import glob
 import hashlib
 import logging
 import os
 import typing as T
 from pathlib import Path
 
+from . import exceptions
+
 
 # Use "hashlib._Hash" instead of hashlib._Hash because:
 # AttributeError: module 'hashlib' has no attribute '_Hash'
@@ -130,6 +133,36 @@ def find_all_image_samples(
     return image_samples_by_video_path
 
 
+def expand_import_paths(
+    import_paths: T.Iterable[Path],
+    *,
+    predicate: T.Callable[[Path], bool] | None = None,
+    missing: str | None = None,
+) -> list[Path]:
+    """Keep existing files/dirs; otherwise expand each path as a glob pattern."""
+    out: list[Path] = []
+    for path in import_paths:
+        if path.is_file() or path.is_dir():
+            if predicate is None or predicate(path):
+                out.append(path)
+            continue
+        pattern = os.fspath(path)
+        matches = [
+            Path(p)
+            for p in glob.glob(pattern, recursive="**" in pattern)
+            if predicate is None or predicate(Path(p))
+        ]
+        if predicate is None:
+            matches = [p for p in matches if p.is_file() or p.is_dir()]
+        matches.sort(key=lambda p: p.name.lower())
+        if not matches:
+            raise exceptions.MapillaryFileNotFoundError(
+                missing or f"Import file or directory not found: {path}"
+            )
+        out.extend(matches)
+    return out
+
+
 def deduplicate_paths(paths: T.Iterable[Path]) -> T.Generator[Path, None, None]:
     resolved_paths: set[Path] = set()
     for p in paths:

tests/unit/test_utils.py

@@ -126,6 +126,17 @@ def test_filter_all(tmpdir: py.path.local):
         )
 
 
+def test_expand_import_paths_glob(tmp_path: Path):
+    (tmp_path / "GS100130.360").mkdir()
+    (tmp_path / "GS110130.360").mkdir()
+    (tmp_path / "GS090130.360").mkdir()
+    (tmp_path / "skip.txt").write_text("x")
+    matches = utils.expand_import_paths([tmp_path / "GS1?0130.360"])
+    assert [p.name for p in matches] == ["GS100130.360", "GS110130.360"]
+    one = utils.expand_import_paths([tmp_path / "GS090130.360"])
+    assert [p.name for p in one] == ["GS090130.360"]
+
+
 class TestSanitizeSerial:
     """Tests for sanitize_serial function"""