feat: move unsupported files by extension

Makefile

@@ -17,14 +17,15 @@ help::
 	@echo
 	@echo --[ $(shell poetry version) ]--
 	@echo
+	@echo "- sh.remove-unsupported		: remove unsupported media files from the music folder"
 	@echo "- beet.import    		: import music to beets library"
 	@echo "- beet.duplicatez  		: list duplicates with beets and export JSON"
 	@echo "- beet.reset       		: delete beets music library"
 	@echo "- nd.backup			: backup the Navidrome database to 'data/backup'"
 	@echo "- nd.merge-annotations		: read annotations of all duplicates, merge and store them"
 	@echo "- nd.eval-deletable		: evaluate deletable duplicates"
 	@echo
-	@echo "- dev.init			: init app"	
+	@echo "- dev.init			: init app"
 	@echo "- dev.shell			: start app shell"
 	@echo "- dev.spell        		: run spell check"
 	@echo "- dev.ruff         		: format code"
@@ -38,8 +39,10 @@ help::
 
 ### App targets ###
 
+sh.remove-unsupported::
+	poetry run python src/ndtoolbox/app.py action=remove-unsupported
 beet.import::
-	beet import -A $(MUSIC_DIR) -p 
+	beet import -A $(MUSIC_DIR) -p
 beet.duplicatez::
 	beet duplicatez
 beet.reset::
@@ -94,3 +97,4 @@ docker.run::
 		-v ./data:/data  \
 		-e TZ=${TIMEZONE} \
 		--entrypoint bash nd-toolbox
+

src/ndtoolbox/app.py

@@ -12,7 +12,7 @@
 
 from ndtoolbox.db import NavidromeDb
 from ndtoolbox.model import MediaFile
-from ndtoolbox.utils import CLI, DotDict, ToolboxConfig
+from ndtoolbox.utils import CLI, DotDict, FileTools, ToolboxConfig
 from ndtoolbox.utils import PrintUtils as PU
 
 
@@ -55,6 +55,7 @@ def __init__(self, nd_folder: str, data_folder: str, source_base: str, target_ba
         navidrome_db_path = nd_folder + "/navidrome.db"
         PU.bold("Initializing DuplicateProcessor")
         PU.ln()
+        PU.info(f"Dry-run: {config.dry_run}")
         PU.info(f"Navidrome database path: {navidrome_db_path}")
         PU.info(f"Output folder: {data_folder}")
         PU.info(f"Source base: {source_base}")
@@ -416,7 +417,6 @@ def _log_info(self, file_path: str, media: MediaFile):
                     PU.log(f"└───── {media.album.annotation}", 3)
             else:
                 # This is not seen as an error because not all media files have an album
-                aid = media.album_id if media.album_id else ""
                 PU.warning(f"└───── Album '{media.album_name}' not found in database!", 2)
         else:
             self.errors.append({"error": "media file not found", "path": file_path})
@@ -476,19 +476,18 @@ def _get_duration(self) -> float:
         PU.ln()
 
     # Read the action argument from the command line
-    action = sys.argv[1] if len(sys.argv) > 1 else None
+    action = sys.argv[1]
     action = action.split("action=")[1]
-
     processor = DuplicateProcessor(config.nd_dir, config.data_dir, config.source_base, config.target_base)
 
-    if action == "merge-annotations":
+    if action == "remove-unsupported":
+        FileTools.move_by_extension(config.music_dir, config.data_dir, config.remove_extensions)
+    elif action == "merge-annotations":
         processor.merge_and_store_annotations()
         processor.print_stats()
-        # processor.export_errors()
     elif action == "eval-deletable":
         processor.eval_deletable_duplicates()
         processor.print_stats()
-        # processor.export_errors()
     elif action == "delete-duplicates":
         processor.delete_duplicates()
     else:

src/ndtoolbox/utils.py

@@ -2,8 +2,10 @@
 Utility classes and functions for the ndtoolbox package.
 """
 
+import glob
 import logging
 import os
+import shutil
 import sys
 from datetime import datetime
 from enum import Enum
@@ -18,8 +20,10 @@ class ToolboxConfig:
     """
 
     timezone = None
+    dry_run: bool
     logger = None
-    pref_extensions = ["mp3", "flac"]
+    pref_extensions: list = None
+    remove_extensions: list = None
 
     nd_dir = None
     data_dir = None
@@ -28,11 +32,13 @@ class ToolboxConfig:
     source_base = None
     target_base = None
 
-    def __init__(self):
+    def __init__(self, dry_run: bool = True):
         """Init config from environment variables."""
         load_dotenv(find_dotenv())
-
         ToolboxConfig.timezone = os.getenv("TZ", "UTC")
+        ToolboxConfig.dry_run = False if os.getenv("DRY_RUN").lower() == "false" else True
+        ToolboxConfig.pref_extensions = os.getenv("PREFERRED_EXTENSIONS").split(" ")
+        ToolboxConfig.remove_extensions = os.getenv("UNSUPPORTED_EXTENSIONS").split(" ")
         ToolboxConfig.nd_dir = os.getenv("ND_DIR")
         ToolboxConfig.data_dir = os.getenv("DATA_DIR")
         ToolboxConfig.music_dir = os.getenv("MUSIC_DIR")
@@ -186,6 +192,46 @@ def log(msg, lvl=0):
         ToolboxConfig.logger.info(msg)
 
 
+class FileTools:
+    """
+    Utility class for file operations.
+    """
+
+    @staticmethod
+    def move_by_extension(source: str, target: str, extensions: list[str]):
+        """
+        Move files with specific extensions from source to target directory.
+
+        Args:
+            source (str): Source directory.
+            target (str): Target directory.
+            extensions (list): List of file extensions to move.
+        """
+        dry = ToolboxConfig.dry_run
+        if source.startswith("./"):
+            source = source[2:]
+        abs_target = os.path.join(os.path.abspath(target), "removed-media")
+        msg = f"[dry-run: {dry}] Moving files file in '{source}' having '{str(extensions)}' extensions, to '{target}'."
+        PrintUtils.info(msg)
+        for ext in extensions:
+            search = os.path.join(source, f"**/*.{ext}")
+            PrintUtils.info(f"[dry-run: {dry}] Searching .{ext} files ({search})", 1)
+            for file in glob.iglob(search, recursive=True):
+                PrintUtils.info(f"[dry-run: {dry}] Found '{file}'")
+
+                # Create folder hierarchy in target
+                abs_target_dir = os.path.join(abs_target, os.path.dirname(file))
+                PrintUtils.info(f"[dry-run: {dry}] Creating target directory: {abs_target_dir}", 2)
+                if not dry:
+                    os.makedirs(abs_target_dir, exist_ok=True)
+
+                # Move files
+                abs_file = os.path.abspath(file)
+                PrintUtils.info(f"[dry-run: {dry}] Move {abs_file} to {abs_target_dir}", 2)
+                if not dry:
+                    shutil.move(abs_file, abs_target_dir)
+
+
 class DotDict(dict):
     """
     Wrap a dictionary with `DotDict()` to allow property access using the dot.notation.