diff --git a/.git-blame-ignore-revs b/.git-blame-ignore-revs index 26e317c590..94367db7ce 100644 --- a/.git-blame-ignore-revs +++ b/.git-blame-ignore-revs @@ -223,3 +223,9 @@ b1ae461723316b738f301406198382913207e100 900543baca4cf5fa4dbe52dbe748f98a0371d0ed # Move DummyIMBackend to fixtures aef9c64cb07c8f828d0dd6e69a5485abf2c43396 +# typing: add types to beets.util +94ef7d8d1f37d87268abed6a4bd5df054912092f +# typing: make path utils more flexible re path types +1418a80e0531fba1e01d0ad2a4decdefaed6b365 +# typing: add types to beets.test.helper +29c887db5bff1cfc4c6ea60505636742fa41cccf diff --git a/beets/importer/tasks.py b/beets/importer/tasks.py index 143d90e2b6..bf4de537dc 100644 --- a/beets/importer/tasks.py +++ b/beets/importer/tasks.py @@ -8,7 +8,7 @@ from collections import defaultdict from collections.abc import Callable from tempfile import mkdtemp -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, AnyStr import mediafile @@ -1156,22 +1156,24 @@ def read_item(self, path: util.PathBytes) -> library.Item | None: _MULTIDISC_MARKERS = ( - rb"dis[ck]", - rb"cd", - rb"cassette", - rb"digital\s+media", - rb"vinyl", + r"dis[ck]", + r"cd", + r"cassette", + r"digital\s+media", + r"vinyl", ) -MULTIDISC_PATTERNS = [ - re.compile(rb"^(.*" + marker + rb"[\W_]*)\d", re.I) +MULTIDISC_BYTES_PATTERNS = [ + re.compile(rf"^(.*{marker}[\W_]*)\d".encode(), re.I) for marker in _MULTIDISC_MARKERS ] +MULTIDISC_PATTERNS = [ + re.compile(rf"^(.*{marker}[\W_]*)\d", re.I) for marker in _MULTIDISC_MARKERS +] -def is_subdir_of_any_in_list( - path: util.PathBytes, dirs: list[util.PathBytes] -) -> bool: + +def is_subdir_of_any_in_list(path: AnyStr, dirs: list[AnyStr]) -> bool: """Returns True if path os a subdirectory of any directory in dirs (a list). In other case, returns False. """ @@ -1179,21 +1181,36 @@ def is_subdir_of_any_in_list( return any(d in ancestors for d in dirs) -def albums_in_dir( - path: util.PathBytes, -) -> Iterable[tuple[list[util.PathBytes], list[util.PathBytes]]]: +def albums_in_dir(path: AnyStr) -> Iterable[tuple[list[AnyStr], list[AnyStr]]]: """Recursively searches the given directory and returns an iterable of (paths, items) where paths is a list of directories and items is a list of Items that is probably an album. Specifically, any folder containing any media files is an album. """ - collapse_paths: list[util.PathBytes] = [] - collapse_items: list[util.PathBytes] = [] + collapse_paths: list[AnyStr] = [] + collapse_items: list[AnyStr] = [] collapse_pat = None - ignore: list[str] = config["ignore"].as_str_seq() + _ignore = config["ignore"].as_str_seq() + ignore: list[AnyStr] + if isinstance(path, str): + ignore = _ignore + else: + ignore = list(map(os.fsencode, _ignore)) ignore_hidden: bool = config["ignore_hidden"].get(bool) + patterns = ( + MULTIDISC_PATTERNS + if isinstance(path, str) + else MULTIDISC_BYTES_PATTERNS + ) + + def get_numbered_variant_pattern(string: AnyStr) -> re.Pattern[AnyStr]: + pat = rf"^{re.escape(os.fsdecode(string))}\d" + return re.compile( + pat if isinstance(string, str) else os.fsencode(pat), re.I + ) + for root, dirs, files in util.sorted_walk( path, ignore=ignore, ignore_hidden=ignore_hidden, logger=log ): @@ -1223,25 +1240,19 @@ def albums_in_dir( # 1") or it contains no items but only directories that are # named in this way. start_collapsing = False - for marker_pat in MULTIDISC_PATTERNS: - match = marker_pat.match(os.path.basename(root)) - + for marker_pat in patterns: # Is this directory the root of a nested multi-disc album? if dirs and not items: # Check whether all subdirectories have the same prefix. start_collapsing = True subdir_pat = None for subdir in dirs: - subdir = util.bytestring_path(subdir) # The first directory dictates the pattern for # the remaining directories. if not subdir_pat: match = marker_pat.match(subdir) if match: - match_group = re.escape(match.group(1)) - subdir_pat = re.compile( - b"".join([b"^", match_group, rb"\d"]), re.I - ) + subdir_pat = get_numbered_variant_pattern(match[1]) else: start_collapsing = False break @@ -1257,13 +1268,11 @@ def albums_in_dir( break # Is this directory the first in a flattened multi-disc album? - elif match: + elif match := marker_pat.match(os.path.basename(root)): start_collapsing = True # Set the current pattern to match directories with the same # prefix as this one, followed by a digit. - collapse_pat = re.compile( - b"".join([b"^", re.escape(match.group(1)), rb"\d"]), re.I - ) + collapse_pat = get_numbered_variant_pattern(match[1]) break # If either of the above heuristics indicated that this is the diff --git a/beets/util/__init__.py b/beets/util/__init__.py index c1aed64eec..fd998ddef2 100644 --- a/beets/util/__init__.py +++ b/beets/util/__init__.py @@ -50,6 +50,7 @@ MAX_FILENAME_LENGTH = 200 WINDOWS_MAGIC_PREFIX = "\\\\?\\" T = TypeVar("T") +AnyPath = TypeVar("AnyPath", str, bytes, Path) StrPath = str | Path PathLike = StrPath | bytes Replacements = Sequence[tuple[Pattern[str], str]] @@ -203,41 +204,33 @@ def ancestry(path: AnyStr) -> list[AnyStr]: def sorted_walk( - path: PathLike, - ignore: Sequence[PathLike] = (), + path: AnyStr, + ignore: Sequence[AnyStr] = (), ignore_hidden: bool = False, logger: Logger | None = None, -) -> Iterator[tuple[bytes, Sequence[bytes], Sequence[bytes]]]: +) -> Iterator[tuple[AnyStr, Sequence[AnyStr], Sequence[AnyStr]]]: """Like `os.walk`, but yields things in case-insensitive sorted, breadth-first order. Directory and file names matching any glob pattern in `ignore` are skipped. If `logger` is provided, then warning messages are logged there when a directory cannot be listed. """ - # Make sure the paths aren't Unicode strings. - bytes_path = bytestring_path(path) - ignore_bytes = [ # rename prevents mypy variable shadowing issue - bytestring_path(i) for i in ignore - ] - # Get all the directories and files at this level. try: - contents = os.listdir(syspath(bytes_path)) + contents = os.listdir(path) except OSError: if logger: logger.warning( "could not list directory {}", - displayable_path(bytes_path), + displayable_path(path), exc_info=True, ) return dirs = [] files = [] - for str_base in contents: - base = bytestring_path(str_base) - + for base in contents: # Skip ignored filenames. skip = False - for pat in ignore_bytes: + for pat in ignore: if fnmatch.fnmatch(base, pat): if logger: logger.debug( @@ -249,7 +242,7 @@ def sorted_walk( continue # Add to output as either a file or a directory. - cur = os.path.join(bytes_path, base) + cur = os.path.join(path, base) if (ignore_hidden and not hidden.is_hidden(cur)) or not ignore_hidden: if os.path.isdir(syspath(cur)): dirs.append(base) @@ -257,14 +250,15 @@ def sorted_walk( files.append(base) # Sort lists (case-insensitive) and yield the current level. - dirs.sort(key=bytes.lower) - files.sort(key=bytes.lower) - yield (bytes_path, dirs, files) + sort_key = path.__class__.lower + dirs.sort(key=sort_key) + files.sort(key=sort_key) + yield (path, dirs, files) # Recurse into directories. for base in dirs: - cur = os.path.join(bytes_path, base) - yield from sorted_walk(cur, ignore_bytes, ignore_hidden, logger) + cur = os.path.join(path, base) + yield from sorted_walk(cur, ignore, ignore_hidden, logger) def path_as_posix(path: bytes) -> bytes: @@ -640,7 +634,7 @@ def reflink( ) from exc -def unique_path(path: bytes) -> bytes: +def unique_path(path: AnyStr) -> AnyStr: """Returns a version of ``path`` that does not exist on the filesystem. Specifically, if ``path` itself already exists, then something unique is appended to the path. @@ -648,7 +642,8 @@ def unique_path(path: bytes) -> bytes: if not os.path.exists(syspath(path)): return path - base, ext = os.path.splitext(path) + byte_path = os.fsencode(path) + base, ext = os.path.splitext(byte_path) match = re.search(rb"\.(\d)+$", base) if match: num = int(match.group(1)) @@ -660,6 +655,8 @@ def unique_path(path: bytes) -> bytes: suffix = f".{num}".encode() + ext new_path = base + suffix if not os.path.exists(new_path): + if not isinstance(path, bytes): + return os.fsdecode(new_path) return new_path diff --git a/beets/util/extension.py b/beets/util/extension.py index 98d82ceaa4..7b87a98c93 100644 --- a/beets/util/extension.py +++ b/beets/util/extension.py @@ -1,15 +1,21 @@ """A tool that finds an extension for files without one""" +from __future__ import annotations + import os import subprocess from logging import Logger from pathlib import Path +from typing import TYPE_CHECKING import mutagen.wave import beets from beets import util +if TYPE_CHECKING: + from beets.util import AnyPath + logger = Logger.info PathBytes = bytes @@ -137,7 +143,7 @@ def fix_extension(path_bytes: PathBytes, logger: Logger | None = None): return new_path -def remux_mpeglayer3_wav(path: util.PathBytes) -> util.PathBytes | None: +def remux_mpeglayer3_wav(path: AnyPath) -> AnyPath | None: """If 'path' is a WAV file containing an MP3 stream (WAVE_FORMAT_MPEGLAYER3, wFormatTag = 0x0055), extract the MP3 stream to a new .mp3 file and return its path. Returns None if the file is not @@ -160,9 +166,14 @@ def remux_mpeglayer3_wav(path: util.PathBytes) -> util.PathBytes | None: # Skip 'data' marker (4 bytes) and chunk size (4 bytes). mp3_data = data[data_offset + 8 :] - mp3_path = os.path.splitext(path)[0] + b".mp3" - with open(util.syspath(mp3_path), "wb") as mp3_file: - mp3_file.write(mp3_data) + syspath = Path(util.syspath(path)) + mp3_path = syspath.with_suffix(".mp3") + mp3_path.write_bytes(mp3_data) util.remove(path) + + if isinstance(path, str): + return str(mp3_path) + if isinstance(path, bytes): + return os.fsencode(mp3_path) return mp3_path diff --git a/beets/util/hidden.py b/beets/util/hidden.py index a12b3fd969..fb3348d5bc 100644 --- a/beets/util/hidden.py +++ b/beets/util/hidden.py @@ -1,19 +1,25 @@ """Simple library to work out if a file is hidden on different platforms.""" +from __future__ import annotations + import ctypes import os import stat import sys from pathlib import Path +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from beets.util import PathLike -def is_hidden(path: bytes | Path) -> bool: +def is_hidden(path: PathLike) -> bool: """ Determine whether the given path is treated as a 'hidden file' by the OS. """ - - if isinstance(path, bytes): - path = Path(os.fsdecode(path)) + if not isinstance(path, Path): + path = os.fsdecode(path) + path = Path(path) # TODO: Avoid doing a platform check on every invocation of the function. # TODO: Stop supporting 'bytes' inputs once 'pathlib' is fully integrated. diff --git a/beetsplug/fetchart.py b/beetsplug/fetchart.py index 05f8a99f20..daa72711de 100644 --- a/beetsplug/fetchart.py +++ b/beetsplug/fetchart.py @@ -1058,7 +1058,7 @@ def get( # Find all files that look like images in the directory. images = [] - ignore = config["ignore"].as_str_seq() + ignore = list(map(os.fsencode, config["ignore"].as_str_seq())) ignore_hidden = config["ignore_hidden"].get(bool) for _, _, files in sorted_walk( path, ignore=ignore, ignore_hidden=ignore_hidden diff --git a/test/plugins/test_playlist.py b/test/plugins/test_playlist.py index 02c8cf6aa4..ecdeb49c8a 100644 --- a/test/plugins/test_playlist.py +++ b/test/plugins/test_playlist.py @@ -1,55 +1,61 @@ import os +from pathlib import Path from shlex import quote import beets -from beets.test import _common from beets.test.helper import PluginTestCase class PlaylistTestCase(PluginTestCase): plugin = "playlist" preload_plugin = False + c_track_path = Path("a") / "b" / "c.mp3" + f_track_path = Path("d") / "e" / "f.mp3" + i_track_path = Path("g") / "h" / "i.mp3" + w_track_path = Path("u") / "v" / "w.mp3" + z_track_path = Path("x") / "y" / "z.mp3" + nonexisting_track_path = Path("nonexisting.mp3") def setUp(self): super().setUp() - self.music_dir = os.path.expanduser(os.path.join("~", "Music")) + self.music_dir = (Path("~") / "Music").expanduser() - i1 = _common.item() - i1.path = beets.util.normpath( - os.path.join(self.music_dir, "a", "b", "c.mp3") - ) - i1.title = "some item" - i1.album = "some album" - self.lib.add(i1) - self.lib.add_album([i1]) - - i2 = _common.item() - i2.path = beets.util.normpath( - os.path.join(self.music_dir, "d", "e", "f.mp3") - ) - i2.title = "another item" - i2.album = "another album" - self.lib.add(i2) - self.lib.add_album([i2]) - - i3 = _common.item() - i3.path = beets.util.normpath( - os.path.join(self.music_dir, "x", "y", "z.mp3") - ) - i3.title = "yet another item" - i3.album = "yet another album" - self.lib.add(i3) - self.lib.add_album([i3]) + for p, title, album in [ + (self.c_track_path, "some item", "some album"), + (self.f_track_path, "another item", "another album"), + (self.z_track_path, "yet another item", "yet another album"), + ]: + self.add_album(path=self.music_dir / p, title=title, album=album) self.playlist_dir = self.temp_dir_path / "playlists" self.playlist_dir.mkdir(parents=True, exist_ok=True) - self.config["directory"] = self.music_dir + self.config["directory"] = str(self.music_dir) self.config["playlist"]["playlist_dir"] = str(self.playlist_dir) + self.absolute_playlist_path = self.playlist_dir / "absolute.m3u" + self.relative_playlist_path = self.playlist_dir / "relative.m3u" + self.write_absolute_playlist() + self.write_relative_playlist() self.setup_test() self.load_plugins() + def write_absolute_playlist(self): + lines = [ + self.music_dir / self.c_track_path, + self.music_dir / self.f_track_path, + self.music_dir / self.nonexisting_track_path, + ] + self.absolute_playlist_path.write_text("\n".join(map(str, lines))) + + def write_relative_playlist(self): + lines = [ + self.c_track_path, + self.f_track_path, + self.nonexisting_track_path, + ] + self.relative_playlist_path.write_text("\n".join(map(str, lines))) + def setup_test(self): raise NotImplementedError @@ -61,7 +67,7 @@ def test_name_query_with_absolute_paths_in_playlist(self): assert {i.title for i in results} == {"some item", "another item"} def test_path_query_with_absolute_paths_in_playlist(self): - q = f"playlist:{quote(os.path.join(self.playlist_dir, 'absolute.m3u'))}" + q = f"playlist:{quote(str(self.absolute_playlist_path))}" results = self.lib.items(q) assert {i.title for i in results} == {"some item", "another item"} @@ -71,7 +77,7 @@ def test_name_query_with_relative_paths_in_playlist(self): assert {i.title for i in results} == {"some item", "another item"} def test_path_query_with_relative_paths_in_playlist(self): - q = f"playlist:{quote(os.path.join(self.playlist_dir, 'relative.m3u'))}" + q = f"playlist:{quote(str(self.relative_playlist_path))}" results = self.lib.items(q) assert {i.title for i in results} == {"some item", "another item"} @@ -81,113 +87,43 @@ def test_name_query_with_nonexisting_playlist(self): assert set(results) == set() def test_path_query_with_nonexisting_playlist(self): - q = f"playlist:{os.path.join(self.playlist_dir, 'nonexisting.m3u')!r}" + q = f"playlist:{quote(str(self.nonexisting_track_path))}" results = self.lib.items(q) assert set(results) == set() class PlaylistTestRelativeToLib(PlaylistQueryTest, PlaylistTestCase): def setup_test(self): - with open(os.path.join(self.playlist_dir, "absolute.m3u"), "w") as f: - f.writelines( - [ - os.path.join(self.music_dir, "a", "b", "c.mp3") + "\n", - os.path.join(self.music_dir, "d", "e", "f.mp3") + "\n", - os.path.join(self.music_dir, "nonexisting.mp3") + "\n", - ] - ) - - with open(os.path.join(self.playlist_dir, "relative.m3u"), "w") as f: - f.writelines( - [ - os.path.join("a", "b", "c.mp3") + "\n", - os.path.join("d", "e", "f.mp3") + "\n", - "nonexisting.mp3\n", - ] - ) - self.config["playlist"]["relative_to"] = "library" class PlaylistTestRelativeToDir(PlaylistQueryTest, PlaylistTestCase): def setup_test(self): - with open(os.path.join(self.playlist_dir, "absolute.m3u"), "w") as f: - f.writelines( - [ - os.path.join(self.music_dir, "a", "b", "c.mp3") + "\n", - os.path.join(self.music_dir, "d", "e", "f.mp3") + "\n", - os.path.join(self.music_dir, "nonexisting.mp3") + "\n", - ] - ) - - with open(os.path.join(self.playlist_dir, "relative.m3u"), "w") as f: - f.writelines( - [ - os.path.join("a", "b", "c.mp3") + "\n", - os.path.join("d", "e", "f.mp3") + "\n", - "nonexisting.mp3\n", - ] - ) - - self.config["playlist"]["relative_to"] = self.music_dir + self.config["playlist"]["relative_to"] = str(self.music_dir) class PlaylistTestRelativeToPls(PlaylistQueryTest, PlaylistTestCase): - def setup_test(self): - with open(os.path.join(self.playlist_dir, "absolute.m3u"), "w") as f: - f.writelines( - [ - os.path.join(self.music_dir, "a", "b", "c.mp3") + "\n", - os.path.join(self.music_dir, "d", "e", "f.mp3") + "\n", - os.path.join(self.music_dir, "nonexisting.mp3") + "\n", - ] - ) - - with open(os.path.join(self.playlist_dir, "relative.m3u"), "w") as f: - f.writelines( - [ - os.path.relpath( - os.path.join(self.music_dir, "a", "b", "c.mp3"), - start=self.playlist_dir, - ) - + "\n", - os.path.relpath( - os.path.join(self.music_dir, "d", "e", "f.mp3"), - start=self.playlist_dir, - ) - + "\n", - os.path.relpath( - os.path.join(self.music_dir, "nonexisting.mp3"), - start=self.playlist_dir, - ) - + "\n", - ] - ) + def write_relative_playlist(self): + lines = [ + os.path.relpath( + self.music_dir / self.c_track_path, self.playlist_dir + ), + os.path.relpath( + self.music_dir / self.f_track_path, self.playlist_dir + ), + os.path.relpath( + self.music_dir / self.nonexisting_track_path, self.playlist_dir + ), + ] + self.relative_playlist_path.write_text("\n".join(lines) + "\n") + def setup_test(self): self.config["playlist"]["relative_to"] = "playlist" self.config["playlist"]["playlist_dir"] = str(self.playlist_dir) class PlaylistUpdateTest: def setup_test(self): - with open(os.path.join(self.playlist_dir, "absolute.m3u"), "w") as f: - f.writelines( - [ - os.path.join(self.music_dir, "a", "b", "c.mp3") + "\n", - os.path.join(self.music_dir, "d", "e", "f.mp3") + "\n", - os.path.join(self.music_dir, "nonexisting.mp3") + "\n", - ] - ) - - with open(os.path.join(self.playlist_dir, "relative.m3u"), "w") as f: - f.writelines( - [ - os.path.join("a", "b", "c.mp3") + "\n", - os.path.join("d", "e", "f.mp3") + "\n", - "nonexisting.mp3\n", - ] - ) - self.config["playlist"]["auto"] = True self.config["playlist"]["relative_to"] = "library" @@ -195,91 +131,75 @@ def setup_test(self): class PlaylistTestItemMoved(PlaylistUpdateTest, PlaylistTestCase): def test_item_moved(self): # Emit item_moved event for an item that is in a playlist - results = self.lib.items( - f"path:{quote(os.path.join(self.music_dir, 'd', 'e', 'f.mp3'))}" - ) + q = f"path:{quote(str(self.music_dir / self.f_track_path))}" + results = self.lib.items(q) item = results[0] beets.plugins.send( "item_moved", item=item, source=item.path, - destination=beets.util.bytestring_path( - os.path.join(self.music_dir, "g", "h", "i.mp3") - ), + destination=os.fsencode(self.music_dir / self.i_track_path), ) # Emit item_moved event for an item that is not in a playlist results = self.lib.items( - f"path:{quote(os.path.join(self.music_dir, 'x', 'y', 'z.mp3'))}" + f"path:{quote(str(self.music_dir / self.z_track_path))}" ) item = results[0] beets.plugins.send( "item_moved", item=item, source=item.path, - destination=beets.util.bytestring_path( - os.path.join(self.music_dir, "u", "v", "w.mp3") - ), + destination=os.fsencode(self.music_dir / self.w_track_path), ) # Emit cli_exit event beets.plugins.send("cli_exit", lib=self.lib) - # Check playlist with absolute paths - playlist_path = os.path.join(self.playlist_dir, "absolute.m3u") - with open(playlist_path) as f: - lines = [line.strip() for line in f.readlines()] - - assert lines == [ - os.path.join(self.music_dir, "a", "b", "c.mp3"), - os.path.join(self.music_dir, "g", "h", "i.mp3"), - os.path.join(self.music_dir, "nonexisting.mp3"), + expected_paths = [ + self.c_track_path, + self.i_track_path, + self.nonexisting_track_path, ] + # Check playlist with absolute paths + lines = list( + map(Path, self.absolute_playlist_path.read_text().splitlines()) + ) + assert lines == [self.music_dir / p for p in expected_paths] # Check playlist with relative paths - playlist_path = os.path.join(self.playlist_dir, "relative.m3u") - with open(playlist_path) as f: - lines = [line.strip() for line in f.readlines()] - - assert lines == [ - os.path.join("a", "b", "c.mp3"), - os.path.join("g", "h", "i.mp3"), - "nonexisting.mp3", - ] + lines = list( + map(Path, self.relative_playlist_path.read_text().splitlines()) + ) + assert lines == expected_paths class PlaylistTestItemRemoved(PlaylistUpdateTest, PlaylistTestCase): def test_item_removed(self): # Emit item_removed event for an item that is in a playlist - results = self.lib.items( - f"path:{quote(os.path.join(self.music_dir, 'd', 'e', 'f.mp3'))}" - ) + q = f"path:{quote(str(self.music_dir / self.f_track_path))}" + results = self.lib.items(q) item = results[0] beets.plugins.send("item_removed", item=item) # Emit item_removed event for an item that is not in a playlist - results = self.lib.items( - f"path:{quote(os.path.join(self.music_dir, 'x', 'y', 'z.mp3'))}" - ) + q = f"path:{quote(str(self.music_dir / self.z_track_path))}" + results = self.lib.items(q) item = results[0] beets.plugins.send("item_removed", item=item) # Emit cli_exit event beets.plugins.send("cli_exit", lib=self.lib) + expected_paths = [self.c_track_path, self.nonexisting_track_path] # Check playlist with absolute paths - playlist_path = os.path.join(self.playlist_dir, "absolute.m3u") - with open(playlist_path) as f: - lines = [line.strip() for line in f.readlines()] - - assert lines == [ - os.path.join(self.music_dir, "a", "b", "c.mp3"), - os.path.join(self.music_dir, "nonexisting.mp3"), - ] + lines = list( + map(Path, self.absolute_playlist_path.read_text().splitlines()) + ) + assert lines == [self.music_dir / p for p in expected_paths] # Check playlist with relative paths - playlist_path = os.path.join(self.playlist_dir, "relative.m3u") - with open(playlist_path) as f: - lines = [line.strip() for line in f.readlines()] - - assert lines == [os.path.join("a", "b", "c.mp3"), "nonexisting.mp3"] + lines = list( + map(Path, self.relative_playlist_path.read_text().splitlines()) + ) + assert lines == expected_paths