importsource: Add new plugin (+docs/tests/changlog)

beetsplug/importsource.py

@@ -0,0 +1,167 @@
+"""Adds a `source_path` attribute to imported albums indicating from what path
+the album was imported from. Also suggests removing that source path in case
+you've removed the album from the library.
+
+"""
+
+import os
+from pathlib import Path
+from shutil import rmtree
+
+from beets.dbcore.query import PathQuery
+from beets.plugins import BeetsPlugin
+from beets.ui import colorize as colorize_text
+from beets.ui import input_options
+
+
+class ImportSourcePlugin(BeetsPlugin):
+    """Main plugin class."""
+
+    def __init__(self):
+        """Initialize the plugin and read configuration."""
+        super(ImportSourcePlugin, self).__init__()
+        self.config.add(
+            {
+                "suggest_removal": False,
+            }
+        )
+        self.import_stages = [self.import_stage]
+        self.register_listener("item_removed", self.suggest_removal)
+        # In order to stop future removal suggestions for an album we keep
+        # track of `mb_albumid`s in this set.
+        self.stop_suggestions_for_albums = set()
+        # During reimports (import --library) both the import_task_choice and
+        # the item_removed event are triggered. The item_removed event is
+        # triggered first. For the import_task_choice event we prevent removal
+        # suggestions using the existing stop_suggestions_for_album mechanism.
+        self.register_listener(
+            "import_task_choice", self.prevent_suggest_removal
+        )
+
+    def prevent_suggest_removal(self, session, task):
+        for item in task.imported_items():
+            if "mb_albumid" in item:
+                self.stop_suggestions_for_albums.add(item.mb_albumid)
+
+    def import_stage(self, _, task):
+        """Event handler for albums import finished."""
+        for item in task.imported_items():
+            # During reimports (import --library), we prevent overwriting the
+            # source_path attribute with the path from the music library
+            if "source_path" in item:
+                self._log.info(
+                    "Preserving source_path of reimported item {}", item.id
+                )
+                continue
+            item["source_path"] = item.path
+            item.try_sync(write=False, move=False)
+
+    def suggest_removal(self, item):
+        """Prompts the user to delete the original path the item was imported from."""
+        if (
+            not self.config["suggest_removal"]
+            or item.mb_albumid in self.stop_suggestions_for_albums
+        ):
+            return
+
+        if "source_path" not in item:
+            self._log.warning(
+                "Item without source_path (probably imported before plugin "
+                "usage): {}",
+                item.filepath,
+            )
+            return
+
+        srcpath = Path(os.fsdecode(item.source_path))
+        if not srcpath.is_file():
+            self._log.warning(
+                "Original source file no longer exists or is not accessible: {}",
+                srcpath,
+            )
+            return
+
+        if not (
+            os.access(srcpath, os.W_OK)
+            and os.access(srcpath.parent, os.W_OK | os.X_OK)
+        ):
+            self._log.warning(
+                "Original source file cannot be deleted (insufficient permissions): {}",
+                srcpath,
+            )
+            return
+
+        # We ask the user whether they'd like to delete the item's source
+        # directory
+        item_path = colorize_text("text_warning", item.filepath)
+        source_path = colorize_text("text_warning", srcpath)
+
+        print(
+            f"The item:\n{item_path}\nis originated from:\n{source_path}\n"
+            "What would you like to do?"
+        )
+
+        resp = input_options(
+            [
+                "Delete the item's source",
+                "Recursively delete the source's directory",
+                "do Nothing",
+                "do nothing and Stop suggesting to delete items from this album",
+            ],
+            require=True,
+        )
+
+        # Handle user response
+        if resp == "d":
+            self._log.info(
+                "Deleting the item's source file: {}",
+                srcpath,
+            )
+            srcpath.unlink()
+
+        elif resp == "r":
+            self._log.info(
+                "Searching for other items with a source_path attr containing: {}",
+                srcpath.parent,
+            )
+
+            source_dir_query = PathQuery(
+                "source_path",
+                srcpath.parent,
+                # The "source_path" attribute may not be present in all
+                # items of the library, so we avoid errors with this:
+                fast=False,
+            )
+
+            print("Doing so will delete the following items' sources as well:")
+            for searched_item in item._db.items(source_dir_query):
+                print(colorize_text("text_warning", searched_item.filepath))
+
+            print("Would you like to continue?")
+            continue_resp = input_options(
+                ["Yes", "delete None", "delete just the File"],
+                require=False,  # Yes is the a default
+            )
+
+            if continue_resp == "y":
+                self._log.info(
+                    "Deleting the item's source directory: {}",
+                    srcpath.parent,
+                )
+                rmtree(srcpath.parent)
+
+            elif continue_resp == "n":
+                self._log.info("doing nothing - aborting hook function")
+                return
+
+            elif continue_resp == "f":
+                self._log.info(
+                    "removing just the item's original source: {}",
+                    srcpath,
+                )
+                srcpath.unlink()
+
+        elif resp == "s":
+            self.stop_suggestions_for_albums.add(item.mb_albumid)
+
+        else:
+            self._log.info("Doing nothing")

docs/changelog.rst

@@ -433,6 +433,7 @@ New features:
   ``beet list -a title:something`` or ``beet list artpath:cover``. Consequently
   album queries involving ``path`` field have been sped up, like ``beet list -a
   path:/path/``.
+- :doc:`plugins/importsource`: Added plugin
 - :doc:`plugins/ftintitle`: New ``keep_in_artist`` option for the plugin, which
   allows keeping the "feat." part in the artist metadata while still changing
   the title.

docs/plugins/importsource.rst

@@ -0,0 +1,80 @@
+ImportSource Plugin
+===================
+
+The ``importsource`` plugin adds a ``source_path`` field to every item imported
+to the library which stores the original media files' paths. Using this plugin
+makes most sense when the general importing workflow is using ``beet import
+--copy``. Additionally the plugin interactively suggests deletion of original
+source files whenever items are removed from the Beets library.
+
+To enable it, add ``importsource`` to the list of plugins in your configuration
+(see :ref:`using-plugins`).
+
+Tracking Source Paths
+---------------------
+
+The primary use case for the plugin is tracking the original location of
+imported files using the ``source_path`` field. Consider this scenario: you've
+imported all directories in your current working directory using:
+
+.. code-block:: bash
+
+    beet import --flat --copy */
+
+Later, for instance if the import didn't complete successfully, you'll need to
+rerun the import but don't want Beets to re-process the already successfully
+imported directories. You can view which files were successfully imported using:
+
+.. code-block:: bash
+
+    beet ls source_path:$PWD --format='$source_path'
+
+To extract just the directory names, pipe the output to standard UNIX utilities:
+
+.. code-block:: bash
+
+    beet ls source_path:$PWD --format='$source_path' | awk -F / '{print $(NF-1)}' | sort -u
+
+This might help to find out what's left to be imported.
+
+Removal Suggestion
+------------------
+
+Another feature of the plugin is suggesting removal of original source files
+when items are deleted from your library. Consider this scenario: you imported
+an album using:
+
+.. code-block:: bash
+
+    beet import --copy --flat ~/Desktop/interesting-album-to-check/
+
+After listening to that album and deciding it wasn't good, you want to delete it
+from your library as well as from your ``~/Desktop``, so you run:
+
+.. code-block:: bash
+
+    beet remove --delete source_path:$HOME/Desktop/interesting-album-to-check
+
+After approving the deletion, the plugin will prompt:
+
+.. code-block:: text
+
+    The item:
+    <music-library>/Interesting Album/01 Interesting Song.flac
+    is originated from:
+    <HOME>/Desktop/interesting-album-to-check/01-interesting-song.flac
+    What would you like to do?
+    Delete the item's source, Recursively delete the source's directory,
+    do Nothing,
+    do nothing and Stop suggesting to delete items from this album?
+
+Configuration
+-------------
+
+To configure the plugin, make an ``importsource:`` section in your configuration
+file. There is one option available:
+
+- **suggest_removal**: By default ``importsource`` suggests to remove the
+  original directories / files from which the items were imported whenever
+  library items (and files) are removed. To disable these prompts set this
+  option to ``no``. Default: ``yes``.

docs/plugins/index.rst

@@ -88,6 +88,7 @@ databases. They share the following configuration options:
     hook
     ihate
     importadded
+    importsource
     importfeeds
     info
     inline

test/plugins/test_importsource.py

@@ -0,0 +1,115 @@
+# This file is part of beets.
+# Copyright 2025, Stig Inge Lea Bjornsen.
+#
+# Permission is hereby granted, free of charge, to any person obtaining
+# a copy of this software and associated documentation files (the
+# "Software"), to deal in the Software without restriction, including
+# without limitation the rights to use, copy, modify, merge, publish,
+# distribute, sublicense, and/or sell copies of the Software, and to
+# permit persons to whom the Software is furnished to do so, subject to
+# the following conditions:
+#
+# The above copyright notice and this permission notice shall be
+# included in all copies or substantial portions of the Software.
+
+
+"""Tests for the `importsource` plugin."""
+
+import os
+import time
+
+from beets import importer
+from beets.test.helper import AutotagImportTestCase, PluginMixin, control_stdin
+from beets.util import syspath
+from beetsplug.importsource import ImportSourcePlugin
+
+_listeners = ImportSourcePlugin.listeners
+
+
+def preserve_plugin_listeners():
+    """Preserve the initial plugin listeners as they would otherwise be
+    deleted after the first setup / tear down cycle.
+    """
+    if not ImportSourcePlugin.listeners:
+        ImportSourcePlugin.listeners = _listeners
+
+
+class ImportSourceTest(PluginMixin, AutotagImportTestCase):
+    plugin = "importsource"
+    preload_plugin = False
+
+    def setUp(self):
+        preserve_plugin_listeners()
+        super().setUp()
+        self.config[self.plugin]["suggest_removal"] = True
+        self.load_plugins()
+        self.prepare_album_for_import(2)
+        self.importer = self.setup_importer()
+        self.importer.add_choice(importer.Action.APPLY)
+        self.importer.run()
+        self.all_items = self.lib.albums().get().items()
+        self.item_to_remove = self.all_items[0]
+
+    def interact(self, stdin_input: str):
+        with control_stdin(stdin_input):
+            self.run_command(
+                "remove",
+                f"path:{syspath(self.item_to_remove.path)}",
+            )
+
+    def test_do_nothing(self):
+        self.interact("N")
+
+        assert os.path.exists(self.item_to_remove.source_path)
+
+    def test_remove_single(self):
+        self.interact("y\nD")
+
+        assert not os.path.exists(self.item_to_remove.source_path)
+
+    def test_remove_all_from_single(self):
+        self.interact("y\nR\ny")
+
+        for item in self.all_items:
+            assert not os.path.exists(item.source_path)
+
+    def test_stop_suggesting(self):
+        self.interact("y\nS")
+
+        for item in self.all_items:
+            assert os.path.exists(item.source_path)
+
+    def test_source_path_attribute_written(self):
+        """Test that source_path attribute is correctly written to imported items.
+
+        The items should already have source_path from the setUp import
+        """
+        for item in self.all_items:
+            assert "source_path" in item
+            assert item.source_path  # Should not be empty
+
+    def test_source_files_not_modified_during_import(self):
+        """Test that source files timestamps are not changed during import."""
+        # Prepare fresh files and record timestamps
+        test_album_path = self.import_path / "test_album"
+        import_paths = self.prepare_album_for_import(
+            2, album_path=test_album_path
+        )
+        original_mtimes = {
+            path: os.stat(path).st_mtime for path in import_paths
+        }
+
+        # Small delay to detect timestamp changes
+        time.sleep(0.1)
+
+        # Run a fresh import
+        importer_session = self.setup_importer()
+        importer_session.add_choice(importer.Action.APPLY)
+        importer_session.run()
+
+        # Verify timestamps haven't changed
+        for path, original_mtime in original_mtimes.items():
+            current_mtime = os.stat(path).st_mtime
+            assert current_mtime == original_mtime, (
+                f"Source file timestamp changed: {path}"
+            )