typing: make albums_in_dir, sorted_walk and other utils generic over AnyStr

.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

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,44 +1156,61 @@ 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.
     """
     ancestors = util.ancestry(path)
     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

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,22 +242,23 @@ 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)
             else:
                 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,15 +634,16 @@ 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.
     """
     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
 
 

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

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.

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