Add bitrate-based 'upgrade' duplicate_action

beets/importer/actions.py

@@ -34,7 +34,10 @@ def options(cls) -> list[str]:
 
     @classmethod
     def strict_options(cls) -> list[str]:
-        return [d.text for d in set(cls) - {DuplicateAction.ASK}]
+        return [
+            d.text
+            for d in set(cls) - {DuplicateAction.ASK, DuplicateAction.UPGRADE}
+        ]
 
     @classmethod
     def choices(cls) -> dict[str, str]:
@@ -45,3 +48,7 @@ def choices(cls) -> dict[str, str]:
     REMOVE = "r", "Remove old"
     KEEP = "k", "Keep all"
     ASK = "a", "Ask"
+    # Config-only: not offered in the interactive `ask` prompt (see
+    # `strict_options`). Resolved to SKIP or applied per-track by
+    # `beets.importer.stages._resolve_duplicates`.
+    UPGRADE = "u", "Upgrade"

beets/importer/stages.py

@@ -14,6 +14,9 @@
     ImportTaskFactory,
     SentinelImportTask,
     SingletonImportTask,
+    _dup_album_ids,
+    _dup_items,
+    resolve_upgrade,
 )
 
 if TYPE_CHECKING:
@@ -272,7 +275,10 @@ def manipulate_files(session: ImportSession, task: ImportTask) -> None:
     finalizes each task.
     """
     if not task.skip:
-        if task.duplicate_action is DuplicateAction.REMOVE:
+        if task.duplicate_action in (
+            DuplicateAction.REMOVE,
+            DuplicateAction.UPGRADE,
+        ):
             task.remove_duplicates(session.lib)
 
         if session.config["move"]:
@@ -340,6 +346,27 @@ def _resolve_duplicates(session: ImportSession, task: ImportTask) -> None:
                 task, found_duplicates
             )
 
+            if task.duplicate_action is DuplicateAction.UPGRADE:
+                if task.apply:
+                    # Apply metadata early so items carry their final
+                    # (post-tag) identity before we match them against
+                    # old library items, which store post-tag metadata
+                    # too. `_apply_choice` re-applies it later, which is
+                    # harmless (metadata application is idempotent).
+                    task.apply_metadata()
+                old_items = [i for d in found_duplicates for i in _dup_items(d)]
+                keys = config["import"]["duplicate_keys"]["item"].as_str_seq()
+                kept, superseded = resolve_upgrade(
+                    task.imported_items(), old_items, keys
+                )
+                if not kept:
+                    log.debug("upgrade: no track was an improvement, declining")
+                    task.duplicate_action = DuplicateAction.SKIP
+                else:
+                    task.apply_upgrade(
+                        kept, superseded, _dup_album_ids(found_duplicates)
+                    )
+
             session.log_choice(task, True)
 
 

beets/importer/tasks.py

@@ -67,6 +67,74 @@ class ImportAbortError(Exception):
     """Raised when the user aborts the tagging operation."""
 
 
+def _item_dup_key(item: library.Item, keys: list[str]) -> tuple[Any, ...]:
+    """Identity key for matching a track across old/new copies.
+
+    Uses the same `duplicate_keys.item` fields that duplicate
+    *detection* already relies on (`ImportTask.find_duplicates` /
+    `SingletonImportTask.find_duplicates`), so per-track matching can't
+    disagree with what got flagged as a duplicate in the first place.
+    `mb_trackid` is deliberately not preferred here: a re-ripped
+    duplicate commonly gets matched to a different MusicBrainz track
+    ID than the old copy, which would otherwise cause false negatives.
+    """
+    return tuple(item.get(k) for k in keys)
+
+
+def _dup_items(obj: library.Album | library.Item) -> list[library.Item]:
+    """Flatten a found-duplicate (`Album` or `Item`) into its items."""
+    if isinstance(obj, library.Album):
+        return list(obj.items())
+    return [obj]
+
+
+def _dup_album_ids(
+    found_duplicates: Iterable[library.Album | library.Item],
+) -> list[int]:
+    """IDs of the existing albums implicated by a set of found duplicates.
+
+    For album-level duplicates, that's the album itself; for
+    singleton-level duplicates, it's the album the matched item
+    belongs to (if any). Used so an upgrade keeps folding kept items
+    into the album that was actually detected as a duplicate, even
+    when none of its individual tracks end up superseded (e.g. the
+    new import only adds tracks that had no old counterpart).
+    """
+    ids = []
+    for d in found_duplicates:
+        album_id = d.id if isinstance(d, library.Album) else d.album_id
+        if album_id:
+            ids.append(album_id)
+    return list(dict.fromkeys(ids))
+
+
+def resolve_upgrade(
+    new_items: list[library.Item],
+    old_items: list[library.Item],
+    keys: list[str],
+) -> tuple[list[library.Item], list[library.Item]]:
+    """Decide, per track, which new items to keep and which old items
+    they supersede.
+
+    Returns `(kept_new_items, superseded_old_items)`. New items with no
+    matching old item are always kept (they aren't duplicates of
+    anything). New items matching an old item are kept only if their
+    bitrate is higher than the old item's; otherwise they're dropped
+    and the old item is left untouched.
+    """
+    by_key = {_item_dup_key(i, keys): i for i in old_items}
+    kept = []
+    superseded = []
+    for new in new_items:
+        old = by_key.get(_item_dup_key(new, keys))
+        if old is None:
+            kept.append(new)
+        elif new.bitrate > old.bitrate:
+            kept.append(new)
+            superseded.append(old)
+    return kept, superseded
+
+
 class BaseImportTask:
     """An abstract base class for importer tasks.
 
@@ -139,13 +207,22 @@ class ImportTask(BaseImportTask):
     choice_flag: Action | None = None
     match: AlbumMatch | TrackMatch | None = None
 
+    # Set by `add()`; only valid afterwards (see class docstring).
+    album: library.Album
+
     # Keep track of the current task item
     cur_album: str | None = None
     cur_artist: str | None = None
     candidates: Sequence[AlbumMatch | TrackMatch] | None = None
     rec: Recommendation | None = None
     duplicate_action: DuplicateAction | None = None
 
+    # Set by `apply_upgrade` when `duplicate_action` is UPGRADE; consumed
+    # by `remove_duplicates` to know which old items to remove and which
+    # old albums to graft the kept new items onto.
+    _upgrade_superseded: list[library.Item] | None = None
+    _upgrade_old_albums: list[int] | None = None
+
     def __init__(
         self,
         toppath: util.PathBytes | None,
@@ -232,6 +309,34 @@ def imported_items(self) -> list[library.Item]:
             return self.match.items
         return []
 
+    def apply_upgrade(
+        self,
+        kept: list[library.Item],
+        superseded: list[library.Item],
+        old_album_ids: list[int],
+    ) -> None:
+        """Narrow this task's items down to `kept`, so `add()` only
+        commits tracks that improve on (or have no) existing duplicate.
+
+        Called by `_resolve_duplicates`, before `add()` runs, when
+        `duplicate_action` is UPGRADE. Stashes `superseded` and
+        `old_album_ids` for `remove_duplicates` to consume later, once
+        `self.album` exists.
+        """
+        kept_ids = {id(i) for i in kept}
+        if self.choice_flag in (Action.ASIS, Action.RETAG):
+            self.items = kept
+        elif self.choice_flag == Action.APPLY and isinstance(
+            self.match, AlbumMatch
+        ):
+            self.match.mapping = {
+                item: track
+                for item, track in self.match.mapping.items()
+                if id(item) in kept_ids
+            }
+        self._upgrade_superseded = superseded
+        self._upgrade_old_albums = old_album_ids
+
     def apply_metadata(self) -> None:
         """Copy metadata from match info to the items."""
         if self.match:  # TODO: redesign to remove the conditional
@@ -244,6 +349,10 @@ def duplicate_items(self, lib: library.Library) -> list[library.Item]:
         return duplicate_items
 
     def remove_duplicates(self, lib: library.Library) -> None:
+        if self._upgrade_superseded is not None:
+            self._remove_upgrade_duplicates(lib)
+            return
+
         duplicate_albums = self.find_duplicates(lib)
         log.debug("removing {} old duplicate albums", len(duplicate_albums))
 
@@ -272,6 +381,61 @@ def remove_duplicates(self, lib: library.Library) -> None:
                     clutter=config["clutter"].as_str_seq(),
                 )
 
+    def _remove_upgrade_duplicates(self, lib: library.Library) -> None:
+        """Remove only the old items superseded by an upgrade.
+
+        Unlike plain `remove`, this never deletes tracks that weren't
+        actually replaced: if any of an old duplicate album's items
+        survive, the newly-kept items are folded into that album (so
+        the album stays as one row) instead of being left in the
+        fresh album `add()` created for them. Only when an old album
+        is left with no surviving items is it deleted outright.
+        """
+        superseded = self._upgrade_superseded or []
+        log.debug("upgrade: removing {} superseded item(s)", len(superseded))
+        for item in superseded:
+            item.remove(with_album=False)
+            if lib.directory in util.ancestry(item.path):
+                log.debug("deleting superseded {.filepath}", item)
+                util.remove(item.path)
+                util.prune_dirs(
+                    os.path.dirname(item.path),
+                    lib.directory,
+                    clutter=config["clutter"].as_str_seq(),
+                )
+
+        grafted = False
+        for album_id in self._upgrade_old_albums or []:
+            old_album = lib.get_album(album_id)
+            if old_album is None:
+                continue
+
+            surviving = list(old_album.items())
+            if not surviving:
+                artpath = old_album.artpath
+                old_album.remove(with_items=False)
+                if artpath and lib.directory in util.ancestry(artpath):
+                    log.debug("deleting duplicate album art {}", artpath)
+                    util.remove(artpath)
+                    util.prune_dirs(
+                        os.path.dirname(artpath),
+                        lib.directory,
+                        clutter=config["clutter"].as_str_seq(),
+                    )
+            elif not grafted:
+                # Partial overlap: keep the album together by moving the
+                # kept new items onto the surviving old album, rather than
+                # leaving them split across two Album rows.
+                for item in self.imported_items():
+                    item.album_id = old_album.id
+                    item.store()
+                self.album.remove(with_items=False)
+                self.album = old_album
+                grafted = True
+            # else: a second old album also has survivors; an item can
+            # only belong to one album, so leave it untouched rather than
+            # arbitrarily choosing between two candidates.
+
     def set_fields(self, lib: library.Library) -> None:
         """Sets the fields given at CLI or configuration to the specified
         values, for both the album and all its items.
@@ -685,6 +849,24 @@ def chosen_info(self) -> dict[str, Any]:
     def imported_items(self) -> list[library.Item]:
         return [self.item]
 
+    def apply_upgrade(
+        self,
+        kept: list[library.Item],
+        superseded: list[library.Item],
+        old_album_ids: list[int],
+    ) -> None:
+        """Stash the superseded old item for `remove_duplicates`.
+
+        `kept` is always `[self.item]` here: a singleton task has
+        exactly one item, and the caller only invokes this method when
+        that item was kept (otherwise the task is routed to SKIP).
+        `old_album_ids` is unused: `Item.duplicates_query` restricts
+        singleton duplicate detection to other singletons (items with
+        no album), so a matched duplicate never has an old album to
+        preserve.
+        """
+        self._upgrade_superseded = superseded
+
     def _emit_imported(self, lib: library.Library) -> None:
         for item in self.imported_items():
             plugins.send("item_imported", lib=lib, item=item)
@@ -716,6 +898,10 @@ def find_duplicates(self, lib: library.Library) -> list[library.Item]:  # type:
     duplicate_items = find_duplicates
 
     def remove_duplicates(self, lib: library.Library) -> None:
+        if self._upgrade_superseded is not None:
+            self._remove_upgrade_duplicates(lib)
+            return
+
         duplicate_items = self.find_duplicates(lib)
         log.debug("removing {} old duplicated items", len(duplicate_items))
         for item in duplicate_items:
@@ -729,6 +915,26 @@ def remove_duplicates(self, lib: library.Library) -> None:
                     clutter=config["clutter"].as_str_seq(),
                 )
 
+    def _remove_upgrade_duplicates(self, lib: library.Library) -> None:
+        """Remove the superseded old item(s).
+
+        Singleton duplicate detection only ever matches other
+        singletons (see `find_duplicates`), so there's never an old
+        album to preserve here, unlike the `ImportTask` version.
+        """
+        superseded = self._upgrade_superseded or []
+        log.debug("upgrade: removing {} superseded item(s)", len(superseded))
+        for item in superseded:
+            item.remove()
+            if lib.directory in util.ancestry(item.path):
+                log.debug("deleting superseded {.filepath}", item)
+                util.remove(item.path)
+                util.prune_dirs(
+                    os.path.dirname(item.path),
+                    lib.directory,
+                    clutter=config["clutter"].as_str_seq(),
+                )
+
     def add(self, lib: library.Library) -> None:
         with lib.transaction():
             self.record_replaced(lib)

docs/changelog.rst

@@ -54,6 +54,9 @@ New features
   various other sources.
 - :doc:`plugins/replaygain`: Add a ``metaflac`` backend that computes ReplayGain
   for FLAC files using the ``metaflac`` command-line tool. :bug:`1203`
+- :ref:`duplicate_action`: Add an ``upgrade`` option that replaces individual
+  duplicate tracks only if the new copy has a higher bitrate, adds any genuinely
+  new tracks, and keeps the album together rather than splitting it. :bug:`4471`
 
 Bug fixes
 ~~~~~~~~~

docs/reference/config.rst

@@ -822,11 +822,18 @@ Default:
 duplicate_action
 ~~~~~~~~~~~~~~~~
 
-Either ``skip``, ``keep``, ``remove``, ``merge`` or ``ask``. Controls how
-duplicates are treated in import task. "skip" means that new item(album or
-track) will be skipped; "keep" means keep both old and new items; "remove" means
-remove old item; "merge" means merge into one album; "ask" means the user should
-be prompted for the action each time. The default is ``ask``.
+Either ``skip``, ``keep``, ``remove``, ``merge``, ``upgrade`` or ``ask``.
+Controls how duplicates are treated in import task. "skip" means that new
+item(album or track) will be skipped; "keep" means keep both old and new items;
+"remove" means remove old item; "merge" means merge into one album; "ask" means
+the user should be prompted for the action each time. The default is ``ask``.
+
+"upgrade" compares each newly-imported track against any existing duplicate with
+the same :ref:`duplicate_keys`: a track only replaces its old counterpart if it
+has a higher bitrate, and tracks with no old counterpart are always added. The
+rest of the existing album is left untouched, and the kept tracks are added to
+that same album rather than a new one. Unlike the other options, "upgrade" is
+config-only and is not offered when ``duplicate_action`` is ``ask``.
 
 .. _duplicate_verbose_prompt:
 

test/plugins/test_art.py

@@ -845,7 +845,6 @@ def test_fanarttv_only_other_images(
 class TestArtImporter(UseThePlugin):
     @pytest.fixture(autouse=True)
     def _setup(self, setup_plugin):
-
         # Mock the album art fetcher to always return our test file.
         self.art_file = self.temp_path / "tmpcover.jpg"
         self.art_file.touch()