diff --git a/beetsplug/importsource.py b/beetsplug/importsource.py new file mode 100644 index 000000000..1c686d334 --- /dev/null +++ b/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") diff --git a/docs/changelog.rst b/docs/changelog.rst index 749ddf005..a78787273 100644 --- a/docs/changelog.rst +++ b/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. diff --git a/docs/plugins/importsource.rst b/docs/plugins/importsource.rst new file mode 100644 index 000000000..dda2d5e08 --- /dev/null +++ b/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: + /Interesting Album/01 Interesting Song.flac + is originated from: + /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``. diff --git a/docs/plugins/index.rst b/docs/plugins/index.rst index 2c9d94dfd..d1590504d 100644 --- a/docs/plugins/index.rst +++ b/docs/plugins/index.rst @@ -88,6 +88,7 @@ databases. They share the following configuration options: hook ihate importadded + importsource importfeeds info inline diff --git a/test/plugins/test_importsource.py b/test/plugins/test_importsource.py new file mode 100644 index 000000000..e05a8f177 --- /dev/null +++ b/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}" + )