⚡ Bolt: os.scandir를 이용한 미디어 파일 탐색 성능 향상

.gitignore

@@ -1,3 +1,4 @@
 __pycache__/
 *.py[cod]
 .DS_Store
+.coverage

.jules/bolt.md

@@ -57,6 +57,10 @@
 ## 2024-06-22 - Optimize FFprobe payload parsing with single pass iteration
 **Learning:** Using multiple generator expressions (`next` and `any`) to search through the same list (like FFprobe streams) requires iterating through the list multiple times. In `_parse_probe_payload`, parsing out both the audio stream and checking for a video stream with separate generator expressions introduces unnecessary loop overhead, which is measurable in batch processes.
 **Action:** Combine multiple searches over the same list into a single standard `for` loop, extracting all necessary information in one pass. This provides measurable CPU savings and avoids multiple iterator instantiations.
+## 2025-02-12 - File traversal optimization in `media_shrinker.py`
+**Learning:** Using `os.walk` paired with `os.lstat` for each file is significantly slower than using `os.scandir`. `os.scandir` caches directory entries and their stat structures inherently reducing the number of system calls needed to traverse a directory hierarchy.
+**Action:** When performing file discovery or traversing directories, particularly with a need for file metadata (like size or checking symlinks), prefer `os.scandir()` or `Path.rglob()` over `os.walk()` to avoid redundant `stat` system calls.
+
 ## 2026-06-25 - [Optimize Path.exists() when paired with stat()]
 **Learning:** Checking `Path.exists()` before `Path.stat()` introduces a redundant system call because `exists()` internally uses `stat()`.
 **Action:** Rely on catching the `OSError` from `Path.stat()` to simultaneously check for existence and retrieve file attributes, saving measurable I/O overhead on large filesystems.

media_shrinker.py

@@ -193,6 +193,7 @@ def find_candidates(
     """
 
     root = Path(root)
+    root_str = str(root)
     excluded = tuple(Path(item).resolve() for item in exclude_paths)
     excluded_prefixes = tuple(prefix.casefold() for prefix in exclude_dir_prefixes)
     candidates: list[tuple[Path, int]] = []
@@ -201,73 +202,74 @@ def find_candidates(
     excluded_prefix_strs = tuple(s + os.sep for s in excluded_exact_strs)
     excluded_exact_set = frozenset(excluded_exact_strs)
 
-    for dirpath_str, dirnames, filenames in os.walk(str(root)):
+    dirs_to_visit = [root_str]
+
+    while dirs_to_visit:
+        current_dir = dirs_to_visit.pop()
+
         try:
-            resolved_dir_str = os.path.realpath(dirpath_str)
+            resolved_dir_str = os.path.realpath(current_dir)
         except OSError:
             continue
 
         if excluded_exact_strs:
             if resolved_dir_str in excluded_exact_set or resolved_dir_str.startswith(
                 excluded_prefix_strs
             ):
-                dirnames[:] = []
                 continue
 
-        # Prune excluded directories
-        valid_dirs = []
-        for d in dirnames:
-            if d.casefold().startswith(excluded_prefixes):
-                continue
-
-            d_path_str = os.path.join(dirpath_str, d)
-
-            try:
-                d_stat = os.lstat(d_path_str)
-                is_symlink = stat.S_ISLNK(d_stat.st_mode)
-            except OSError:
-                continue
+        try:
+            with os.scandir(current_dir) as it:
+                for entry in it:
+                    name = entry.name
 
-            if excluded_exact_strs:
-                if not is_symlink:
-                    resolved_d_str = os.path.join(resolved_dir_str, d)
-                else:
                     try:
-                        resolved_d_str = os.path.realpath(d_path_str)
+                        is_symlink = entry.is_symlink()
                     except OSError:
                         continue
+                    if is_symlink:
+                        # Original implementation did not follow symlinks
+                        continue
 
-                if resolved_d_str in excluded_exact_set or resolved_d_str.startswith(
-                    excluded_prefix_strs
-                ):
-                    continue
-            valid_dirs.append(d)
-        dirnames[:] = valid_dirs
-
-        for f in filenames:
-            if not f.lower().endswith(SUPPORTED_EXTS_TUPLE):
-                continue
-
-            file_path_str = os.path.join(dirpath_str, f)
-
-            try:
-                st = os.lstat(file_path_str)
-                if stat.S_ISLNK(st.st_mode) or not stat.S_ISREG(st.st_mode):
-                    continue
-                size = st.st_size
-            except OSError:
-                continue
-
-            if excluded_exact_strs:
-                resolved_file_str = os.path.join(resolved_dir_str, f)
-                if (
-                    resolved_file_str in excluded_exact_set
-                    or resolved_file_str.startswith(excluded_prefix_strs)
-                ):
-                    continue
+                    try:
+                        is_dir = entry.is_dir(follow_symlinks=False)
+                        is_file = entry.is_file(follow_symlinks=False) if not is_dir else False
+                    except OSError:
+                        continue
 
-            if include_under_limit or size > size_limit_bytes:
-                candidates.append((Path(file_path_str), size))
+                    if is_dir:
+                        if name.casefold().startswith(excluded_prefixes):
+                            continue
+
+                        if excluded_exact_strs:
+                            resolved_d_str = os.path.join(resolved_dir_str, name)
+                            if resolved_d_str in excluded_exact_set or resolved_d_str.startswith(excluded_prefix_strs):
+                                continue
+
+                        dirs_to_visit.append(entry.path)
+
+                    elif is_file:
+                        if not name.lower().endswith(SUPPORTED_EXTS_TUPLE):
+                            continue
+
+                        try:
+                            st = entry.stat(follow_symlinks=False)
+                            size = st.st_size
+                        except OSError:
+                            continue
+
+                        if excluded_exact_strs:
+                            resolved_file_str = os.path.join(resolved_dir_str, name)
+                            if (
+                                resolved_file_str in excluded_exact_set
+                                or resolved_file_str.startswith(excluded_prefix_strs)
+                            ):
+                                continue
+
+                        if include_under_limit or size > size_limit_bytes:
+                            candidates.append((Path(entry.path), size))
+        except OSError:
+            pass
 
     # Fast path: Pre-compute the root string prefix to avoid slow Path.relative_to() instantiation in the sort loop
     # We add a trailing slash to handle cases where root represents a directory structure.

tests/test_media_shrinker.py

@@ -185,15 +185,31 @@ def test_find_candidates_skips_entries_when_symlink_check_fails(self) -> None:
 
             import os
 
-            original_lstat = os.lstat
-
-            def flaky_lstat(path):
-                name = os.path.basename(str(path))
-                if name in {"bad.mp3", "bad_dir"}:
-                    raise OSError("cannot inspect symlink state")
-                return original_lstat(path)
-
-            with patch("os.lstat", flaky_lstat):
+            class FlakyDirEntry:
+                def __init__(self, entry):
+                    self._entry = entry
+                def __getattr__(self, name):
+                    if name in ('is_symlink', 'is_dir', 'is_file', 'stat'):
+                        if self._entry.name in {"bad.mp3", "bad_dir"}:
+                            def raise_os_error(*args, **kwargs):
+                                raise OSError("cannot inspect state")
+                            return raise_os_error
+                    return getattr(self._entry, name)
+
+            original_scandir = os.scandir
+            class FlakyScandirIterator:
+                def __init__(self, path):
+                    self._it = original_scandir(path)
+                def __iter__(self):
+                    return self
+                def __next__(self):
+                    return FlakyDirEntry(next(self._it))
+                def __enter__(self):
+                    return self
+                def __exit__(self, *args):
+                    self._it.close()
+
+            with patch("os.scandir", FlakyScandirIterator):
                 candidates = [
                     p[0].relative_to(root)
                     for p in find_candidates(root, include_under_limit=True)