Dedupe individual tracks during import based on stable identifier

beets/config_default.yaml

@@ -53,6 +53,8 @@ import:
         item: artist title
     duplicate_action: ask
     duplicate_verbose_prompt: no
+    duplicate_track_resolution: no
+    duplicate_track_action: ''
     bell: no
     set_fields: {}
     ignored_alias_types: []

beets/importer/session.py

@@ -2,7 +2,7 @@
 
 import os
 import time
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 from beets import config, logging, plugins, util
 from beets.util import displayable_path, normpath, pipeline, syspath
@@ -46,6 +46,7 @@ class ImportSession:
     _is_resuming: dict[bytes, bool]
     _merged_items: set[PathBytes]
     _merged_dirs: set[PathBytes]
+    _seen_track_keys: set[tuple[Any, ...]]
 
     def __init__(
         self,
@@ -74,6 +75,7 @@ def __init__(
         self._is_resuming = {}
         self._merged_items = set()
         self._merged_dirs = set()
+        self._seen_track_keys = set()
 
         # Normalize the paths.
         self.paths = list(map(normpath, paths or []))
@@ -185,6 +187,22 @@ def get_duplicate_action(
     def choose_item(self, task: SingletonImportTask) -> TrackMatch | Action:
         raise NotImplementedError
 
+    def resolve_track_duplicates(
+        self,
+        task: ImportTask,
+        duplicates: dict[library.Item, list[library.Item]],
+    ) -> DuplicateAction:
+        """Decide what to do with album tracks that already exist in the
+        library: :attr:`~DuplicateAction.SKIP` (drop the duplicate tracks and
+        fold the remaining new tracks into the existing album),
+        :attr:`~DuplicateAction.KEEP` (import everything) or
+        :attr:`~DuplicateAction.REMOVE` (remove the old items).
+
+        ``duplicates`` maps each incoming :class:`~beets.library.Item` to the
+        existing library items it duplicates.
+        """
+        raise NotImplementedError
+
     def run(self) -> None:
         """Run the import task."""
         self.logger.info("import started {}", time.asctime())
@@ -205,6 +223,10 @@ def run(self) -> None:
                 # Split directory tasks into one task for each album.
                 stages += [stagefuncs.group_albums(self)]
 
+            # Optionally drop or replace album tracks that already exist in
+            # the library before the autotag lookup runs.
+            stages += [stagefuncs.resolve_track_duplicates(self)]
+
             # These stages either talk to the user to get a decision or,
             # in the case of a non-autotagged import, just choose to
             # import everything as-is. In *both* cases, these stages
@@ -285,6 +307,25 @@ def mark_merged(self, paths: Sequence[PathBytes]) -> None:
         }
         self._merged_dirs.update(dirs)
 
+    def track_key_seen(self, key: tuple[Any, ...]) -> bool:
+        """Return whether a per-track duplicate ``key`` was already claimed by
+        an earlier task in this import run (see :ref:`duplicate_track_resolution`).
+
+        This lets the track-level duplicate check catch tracks imported earlier
+        in the *same* run before they reach the database, which matters under
+        the default threaded import where a later task may be checked before an
+        earlier one has been added.
+        """
+        return key in self._seen_track_keys
+
+    def remember_track_key(self, key: tuple[Any, ...]) -> None:
+        """Record a per-track duplicate ``key`` claimed by the current task.
+
+        Only ever called from the single-threaded ``resolve_track_duplicates``
+        pipeline stage, so the shared set needs no locking.
+        """
+        self._seen_track_keys.add(key)
+
     def is_resuming(self, toppath: PathBytes) -> bool:
         """Return `True` if user wants to resume import of this path.
 

beets/importer/stages.py

@@ -3,10 +3,11 @@
 import contextvars
 import itertools
 import logging
-from typing import TYPE_CHECKING, TypeAlias
+from typing import TYPE_CHECKING, Any, TypeAlias
 
-from beets import config, plugins
+from beets import config, dbcore, plugins
 from beets.util import MoveOperation, displayable_path, pipeline
+from beets.util.color import colorize
 
 from .actions import Action, DuplicateAction
 from .tasks import (
@@ -119,6 +120,125 @@ def group(item: library.Item) -> tuple[str | None, str | None]:
         out = pipeline.multiple(tasks)
 
 
+@pipeline.mutator_stage
+def resolve_track_duplicates(session: ImportSession, task: ImportTask) -> None:
+    """Resolve tracks of an album that already exist in the library.
+
+    When ``import.duplicate_track_resolution`` is enabled, each item of an
+    album import is checked against the library using
+    ``import.duplicate_keys.item``. Matched tracks are resolved according to
+    ``import.duplicate_track_action`` (which falls back to
+    ``import.duplicate_action`` when unset):
+
+    * ``skip`` drops the duplicate tracks and adds the remaining new tracks to
+      the existing album they belong to (if every track is a duplicate, the
+      whole album is skipped);
+    * ``remove`` removes the matching old library items;
+    * ``keep`` (and ``merge``) import everything as-is;
+    * ``ask`` prompts the session for one of the above.
+
+    This runs before :func:`lookup_candidates` so that dropped tracks are
+    excluded from the autotag match. Singleton imports are handled by the
+    regular duplicate resolution and are ignored here.
+    """
+    if (
+        task.skip
+        or not task.is_album
+        or not task.items
+        or not config["import"]["duplicate_track_resolution"].get(bool)
+    ):
+        return
+
+    keys = config["import"]["duplicate_keys"]["item"].as_str_seq()
+    if not keys:
+        return
+
+    def item_key(item: library.Item) -> tuple[Any, ...]:
+        return tuple(item.get(k) for k in keys)
+
+    def has_key(item: library.Item) -> bool:
+        return any(item.get(k) for k in keys)
+
+    def remember(items: Iterable[library.Item]) -> None:
+        # Claim the keys of the tracks this task will import so that later
+        # tasks in the same run recognise them as duplicates before they reach
+        # the database (see ``ImportSession.track_key_seen``).
+        for item in items:
+            if has_key(item):
+                session.remember_track_key(item_key(item))
+
+    # Map each incoming item to the existing library items it duplicates. A
+    # track also counts as a duplicate when its key was already claimed by an
+    # earlier task in this same import run, even if that task has not been
+    # added to the library yet.
+    duplicates: dict[library.Item, list[library.Item]] = {}
+    for item in task.items:
+        if not has_key(item):
+            continue
+        matches = _find_track_duplicates(session.lib, item, keys)
+        if matches or session.track_key_seen(item_key(item)):
+            duplicates[item] = matches
+
+    if not duplicates:
+        remember(task.items)
+        return
+
+    action = _track_duplicate_action()
+    if action is DuplicateAction.ASK:
+        action = session.resolve_track_duplicates(task, duplicates)
+
+    if action is DuplicateAction.SKIP:
+        for item in duplicates:
+            log.info(
+                colorize("text_warning", "Skipping duplicate track: {}"),
+                displayable_path(item.path),
+            )
+            task.items.remove(item)
+        if not task.items:
+            # Every track was a duplicate: skip the whole album.
+            log.info(
+                colorize(
+                    "text_warning",
+                    "Skipping album, all tracks are duplicates: {}",
+                ),
+                next(iter(duplicates)).album,
+            )
+            task.set_choice(Action.SKIP)
+            return
+        # Only some tracks were duplicates; we have already dropped them, so
+        # don't let the album-level check skip the rest.
+        task.duplicate_tracks_resolved = True
+        # Fold the remaining new tracks into the existing album the matched
+        # duplicates belong to. Tracks matching a *singleton* are skipped
+        # individually but do not affect the fold target (their ``album_id``
+        # is ``None``), so a mix of album-member and singleton matches still
+        # completes the album. Only when the matched album members span more
+        # than one album -- or none of the matches belong to an album at all
+        # -- are the new tracks imported as their own album.
+        album_ids = {
+            match.album_id
+            for matches in duplicates.values()
+            for match in matches
+            if match.album_id is not None
+        }
+        if len(album_ids) == 1:
+            task.fold_into_album_id = album_ids.pop()
+        else:
+            log.warning(
+                "cannot fold tracks into a single existing album; "
+                "importing them as a new album"
+            )
+    elif action is DuplicateAction.REMOVE:
+        for matches in duplicates.values():
+            task.duplicate_track_items_to_remove.extend(matches)
+        task.duplicate_tracks_resolved = True
+    # KEEP and MERGE leave the incoming tracks untouched; whole-album
+    # duplicates are still handled by the regular resolution stage.
+
+    # Claim the keys of the tracks that remain to be imported.
+    remember(task.items)
+
+
 @pipeline.mutator_stage
 def lookup_candidates(session: ImportSession, task: ImportTask) -> None:
     """A coroutine for performing the initial MusicBrainz lookup for an
@@ -275,6 +395,9 @@ def manipulate_files(session: ImportSession, task: ImportTask) -> None:
         if task.duplicate_action is DuplicateAction.REMOVE:
             task.remove_duplicates(session.lib)
 
+        if task.duplicate_track_items_to_remove:
+            task.remove_duplicate_track_items(session.lib)
+
         if session.config["move"]:
             operation = MoveOperation.MOVE
         elif session.config["copy"]:
@@ -327,10 +450,48 @@ def _apply_choice(session: ImportSession, task: ImportTask) -> None:
         task.set_fields(session.lib)
 
 
+def _track_duplicate_action() -> DuplicateAction:
+    """Return the configured :class:`DuplicateAction` for per-track resolution.
+
+    Uses ``import.duplicate_track_action`` when set, otherwise falls back to
+    ``import.duplicate_action``.
+    """
+    cfg = config["import"]
+    view = (
+        cfg["duplicate_track_action"]
+        if cfg["duplicate_track_action"].get()
+        else cfg["duplicate_action"]
+    )
+    choice = view.as_choice(DuplicateAction.choices())
+    return DuplicateAction(choice)  # type: ignore[call-arg]
+
+
+def _find_track_duplicates(
+    lib: library.Library, item: library.Item, keys: list[str]
+) -> list[library.Item]:
+    """Return library items matching `item` on all `keys`, excluding the
+    item itself (so re-imports do not match their own files).
+
+    Unlike :meth:`Item.duplicates_query`, this matches *every* library item,
+    including tracks that belong to an album -- not just singletons -- so a
+    track is caught regardless of how it was originally imported.
+    """
+    query = dbcore.AndQuery(
+        [item.field_query(k, item.get(k), dbcore.MatchQuery) for k in keys]
+    )
+    return [other for other in lib.items(query) if other.path != item.path]
+
+
 def _resolve_duplicates(session: ImportSession, task: ImportTask) -> None:
     """Check if a task conflicts with items or albums already imported
     and ask the session to resolve this.
     """
+    if task.duplicate_tracks_resolved:
+        # Per-track duplicate resolution already pruned (or recorded for
+        # removal) the tracks of this album that exist in the library; the
+        # rest are new and should be imported without a whole-album skip.
+        return
+
     if task.choice_flag in (Action.ASIS, Action.APPLY, Action.RETAG):
         found_duplicates = task.find_duplicates(session.lib)
         if found_duplicates: