Skip broken directory entries during discovery
This commit is contained in:
@@ -33,10 +33,28 @@ def _discover_root(
|
||||
extensions: set[str],
|
||||
discovered: list[FileCandidate],
|
||||
) -> None:
|
||||
entries = sorted(root.iterdir(), key=lambda item: (not item.is_file(), item.name.lower()))
|
||||
entries: list[tuple[Path, bool]] = []
|
||||
try:
|
||||
for entry in root.iterdir():
|
||||
try:
|
||||
entries.append((entry, entry.is_file()))
|
||||
except OSError:
|
||||
continue
|
||||
except OSError:
|
||||
return
|
||||
|
||||
entries.sort(key=lambda item: (not item[1], item[0].name.lower()))
|
||||
|
||||
for entry, is_file in entries:
|
||||
if not is_file:
|
||||
try:
|
||||
is_dir = entry.is_dir()
|
||||
except OSError:
|
||||
continue
|
||||
|
||||
if not is_dir:
|
||||
continue
|
||||
|
||||
for entry in entries:
|
||||
if entry.is_dir():
|
||||
if recursive:
|
||||
_discover_root(
|
||||
root=entry,
|
||||
@@ -50,7 +68,11 @@ def _discover_root(
|
||||
if extension not in extensions:
|
||||
continue
|
||||
|
||||
stat = entry.stat()
|
||||
try:
|
||||
stat = entry.stat()
|
||||
except OSError:
|
||||
continue
|
||||
|
||||
discovered.append(
|
||||
FileCandidate(
|
||||
path=entry,
|
||||
|
||||
@@ -38,3 +38,44 @@ def test_discovery_can_disable_recursion(tmp_path: Path) -> None:
|
||||
discovered = discover_files((root,), extensions=("mp4",), recursive=False)
|
||||
|
||||
assert [item.path for item in discovered] == [direct]
|
||||
|
||||
|
||||
def test_discovery_skips_entries_that_raise_oserror_during_scan() -> None:
|
||||
class FakeStat:
|
||||
st_size = 123
|
||||
st_mtime_ns = 456
|
||||
|
||||
class FakeFileEntry:
|
||||
def __init__(self, name: str) -> None:
|
||||
self.name = name
|
||||
self.suffix = Path(name).suffix
|
||||
|
||||
def is_file(self) -> bool:
|
||||
return True
|
||||
|
||||
def is_dir(self) -> bool:
|
||||
return False
|
||||
|
||||
def stat(self) -> FakeStat:
|
||||
return FakeStat()
|
||||
|
||||
class BrokenEntry:
|
||||
name = "video_2545@25-06-2022_02-52-08.mp4"
|
||||
suffix = ".mp4"
|
||||
|
||||
def is_file(self) -> bool:
|
||||
raise OSError(
|
||||
123,
|
||||
"文件名、目录名或卷标语法不正确。",
|
||||
r"E:\同步东芝1备份用\V4\2022\22-06\6.24-6.25\补充\video_files\video_2545@25-06-2022_02-52-08.mp4",
|
||||
)
|
||||
|
||||
class FakeRoot:
|
||||
def iterdir(self) -> list[object]:
|
||||
return [BrokenEntry(), FakeFileEntry("good.mp4")]
|
||||
|
||||
discovered = discover_files((FakeRoot(),), extensions=("mp4",))
|
||||
|
||||
assert [item.path.name for item in discovered] == ["good.mp4"]
|
||||
assert [item.size for item in discovered] == [123]
|
||||
assert [item.discovery_order for item in discovered] == [0]
|
||||
|
||||
Reference in New Issue
Block a user