feat(import): Add support for reading skipped paths from logfile

beets/ui/commands.py

@@ -79,6 +79,43 @@ def _do_query(lib, query, album, also_items=True):
     return items, albums
 
 
+def _paths_from_logfile(path):
+    """Parse the logfile and yield skipped paths to pass to the `import`
+    command.
+    """
+    with open(path, mode="r", encoding="utf-8") as fp:
+        for i, line in enumerate(fp, start=1):
+            verb, sep, paths = line.rstrip("\n").partition(" ")
+            if not sep:
+                raise ValueError(f"line {i} is invalid")
+
+            # Ignore informational lines that don't need to be re-imported.
+            if verb in {"import", "duplicate-keep", "duplicate-replace"}:
+                continue
+
+            if verb not in {"asis", "skip", "duplicate-skip"}:
+                raise ValueError(f"line {i} contains unknown verb {verb}")
+
+            yield os.path.commonpath(paths.split("; "))
+
+
+def _parse_logfiles(logfiles):
+    """Parse all `logfiles` and yield paths from it."""
+    for logfile in logfiles:
+        try:
+            yield from _paths_from_logfile(syspath(normpath(logfile)))
+        except ValueError as err:
+            raise ui.UserError('malformed logfile {}: {}'.format(
+                util.displayable_path(logfile),
+                str(err)
+            )) from err
+        except IOError as err:
+            raise ui.UserError('unreadable logfile {}: {}'.format(
+                util.displayable_path(logfile),
+                str(err)
+            )) from err
+
+
 # fields: Shows a list of available fields for queries and format strings.
 
 def _print_keys(query):
@@ -913,12 +950,6 @@ def import_files(lib, paths, query):
     """Import the files in the given list of paths or matching the
     query.
     """
-    # Check the user-specified directories.
-    for path in paths:
-        if not os.path.exists(syspath(normpath(path))):
-            raise ui.UserError('no such file or directory: {}'.format(
-                displayable_path(path)))
-
     # Check parameter consistency.
     if config['import']['quiet'] and config['import']['timid']:
         raise ui.UserError("can't be both quiet and timid")
@@ -960,7 +991,12 @@ def import_func(lib, opts, args):
     else:
         query = None
         paths = args
-        if not paths:
+
+        # The paths from the logfiles go into a separate list to allow handling
+        # errors differently from user-specified paths.
+        paths_from_logfiles = list(_parse_logfiles(opts.from_logfiles or []))
+
+        if not paths and not paths_from_logfiles:
             raise ui.UserError('no path specified')
 
         # On Python 2, we used to get filenames as raw bytes, which is
@@ -969,6 +1005,31 @@ def import_func(lib, opts, args):
         # filename.
         paths = [p.encode(util.arg_encoding(), 'surrogateescape')
                  for p in paths]
+        paths_from_logfiles = [p.encode(util.arg_encoding(), 'surrogateescape')
+                               for p in paths_from_logfiles]
+
+        # Check the user-specified directories.
+        for path in paths:
+            if not os.path.exists(syspath(normpath(path))):
+                raise ui.UserError('no such file or directory: {}'.format(
+                    displayable_path(path)))
+
+        # Check the directories from the logfiles, but don't throw an error in
+        # case those paths don't exist. Maybe some of those paths have already
+        # been imported and moved separately, so logging a warning should
+        # suffice.
+        for path in paths_from_logfiles:
+            if not os.path.exists(syspath(normpath(path))):
+                log.warning('No such file or directory: {}'.format(
+                    displayable_path(path)))
+                continue
+
+            paths.append(path)
+
+        # If all paths were read from a logfile, and none of them exist, throw
+        # an error
+        if not paths:
+            raise ui.UserError('none of the paths are importable')
 
     import_files(lib, paths, query)
 
@@ -1061,6 +1122,11 @@ def import_func(lib, opts, args):
     metavar='ID',
     help='restrict matching to a specific metadata backend ID'
 )
+import_cmd.parser.add_option(
+    '--from-logfile', dest='from_logfiles', action='append',
+    metavar='PATH',
+    help='read skipped paths from an existing logfile'
+)
 import_cmd.parser.add_option(
     '--set', dest='set_fields', action='callback',
     callback=_store_dict,

docs/changelog.rst

@@ -29,6 +29,9 @@ New features:
   :bug:`1840` :bug:`4302`
 * Added a ``-P`` (or ``--disable-plugins``) flag to specify one/multiple plugin(s) to be
   disabled at startup.
+* :ref:`import-options`: Add support for re-running the importer on paths in
+  log files that were created with the ``-l`` (or ``--logfile``) argument.
+  :bug:`4379` :bug:`4387`
 
 Bug fixes:
 

docs/guides/tagger.rst

@@ -80,6 +80,8 @@ all of these limitations.
 
 Now that that's out of the way, let's tag some music.
 
+.. _import-options:
+
 Options
 -------
 
@@ -101,7 +103,8 @@ command-line options you should know:
 * ``beet import -l LOGFILE``: write a message to ``LOGFILE`` every time you skip
   an album or choose to take its tags "as-is" (see below) or the album is
   skipped as a duplicate; this lets you come back later and reexamine albums
-  that weren't tagged successfully
+  that weren't tagged successfully. Run ``beet import --from-logfile=LOGFILE``
+  rerun the importer on such paths from the logfile.
 
 * ``beet import -q``: quiet mode. Never prompt for input and, instead,
   conservatively skip any albums that need your opinion. The ``-ql`` combination

docs/reference/cli.rst

@@ -86,7 +86,9 @@ Optional command flags:
   that weren't tagged successfully---either because they're not in the
   MusicBrainz database or because something's wrong with the files. Use the
   ``-l`` option to specify a filename to log every time you skip an album
-  or import it "as-is" or an album gets skipped as a duplicate.
+  or import it "as-is" or an album gets skipped as a duplicate. You can later
+  review the file manually or import skipped paths from the logfile
+  automatically by using the ``--from-logfile LOGFILE`` argument.
 
 * Relatedly, the ``-q`` (quiet) option can help with large imports by
   autotagging without ever bothering to ask for user input. Whenever the

test/test_ui.py

@@ -729,6 +729,40 @@ def test_quiet_timid_disallowed(self):
         self.assertRaises(ui.UserError, commands.import_files, None, [],
                           None)
 
+    def test_parse_paths_from_logfile(self):
+        if os.path.__name__ == 'ntpath':
+            logfile_content = (
+                "import started Wed Jun 15 23:08:26 2022\n"
+                "asis C:\\music\\Beatles, The\\The Beatles; C:\\music\\Beatles, The\\The Beatles\\CD 01; C:\\music\\Beatles, The\\The Beatles\\CD 02\n"  # noqa: E501
+                "duplicate-replace C:\\music\\Bill Evans\\Trio '65\n"
+                "skip C:\\music\\Michael Jackson\\Bad\n"
+                "skip C:\\music\\Soulwax\\Any Minute Now\n"
+            )
+            expected_paths = [
+                "C:\\music\\Beatles, The\\The Beatles",
+                "C:\\music\\Michael Jackson\\Bad",
+                "C:\\music\\Soulwax\\Any Minute Now",
+            ]
+        else:
+            logfile_content = (
+                "import started Wed Jun 15 23:08:26 2022\n"
+                "asis /music/Beatles, The/The Beatles; /music/Beatles, The/The Beatles/CD 01; /music/Beatles, The/The Beatles/CD 02\n"  # noqa: E501
+                "duplicate-replace /music/Bill Evans/Trio '65\n"
+                "skip /music/Michael Jackson/Bad\n"
+                "skip /music/Soulwax/Any Minute Now\n"
+            )
+            expected_paths = [
+                "/music/Beatles, The/The Beatles",
+                "/music/Michael Jackson/Bad",
+                "/music/Soulwax/Any Minute Now",
+            ]
+
+        logfile = os.path.join(self.temp_dir, b"logfile.log")
+        with open(logfile, mode="w") as fp:
+            fp.write(logfile_content)
+        actual_paths = list(commands._paths_from_logfile(logfile))
+        self.assertEqual(actual_paths, expected_paths)
+
 
 @_common.slow_test()
 class ConfigTest(unittest.TestCase, TestHelper, _common.Assertions):