feat: add yaml configuration

Author: add-n2x

config/config.yaml

@@ -0,0 +1,19 @@
+
+# Heartbeets Configuration
+
+timezone: Europe/Vienna
+log-level: INFO
+dry-run: True
+pref-extensions: [mp3, flac]
+remove-extensions: [m4a, wma]
+data: ./data
+music: ./music
+
+beets:
+  config: ./config/beets/docker.config.yaml
+  database: ./data/beets/library.db
+  base-path: ./music
+
+navidrome:
+  database: ./navidrome/navidrome.db
+  base-path: ./music/library

src/ndtoolbox/app.py

@@ -8,11 +8,13 @@
 from datetime import datetime
 
 import jsonpickle
+import tomli
 from easydict import EasyDict
 
+from ndtoolbox.config import config
 from ndtoolbox.db import NavidromeDb, NavidromeDbConnection
 from ndtoolbox.model import Annotation, Folder, MediaFile
-from ndtoolbox.utils import CLI, FileTools, FileUtil, Stats, ToolboxConfig
+from ndtoolbox.utils import CLI, FileTools, FileUtil, PrintUtil, Stats
 from ndtoolbox.utils import PrintUtil as PU
 from ndtoolbox.utils import StringUtil as SU
 
@@ -48,13 +50,12 @@ class DuplicateProcessor:
         stop (float): The timestamp when processing stopped. This is set at the end of an action or an error occurs.
     """
 
-    config: ToolboxConfig
     db: NavidromeDb
     data: DuplicateData
     stats: Stats
     errors: list
 
-    def __init__(self, config: ToolboxConfig):
+    def __init__(self):
         """
         Initialize the DuplicateProcessor with a database and an input file containing duplicate media files.
 
@@ -67,10 +68,11 @@ def __init__(self, config: ToolboxConfig):
             base_path_navidrome (str): The actual location in the Navidrome music library.
             dry_run (bool): If True, no actual file operations will be performed.
         """
-        self.config = config
+        # self.config = config
         self.data = DuplicateData()
         self.errors = []
-        self.db = NavidromeDb(config.navidrome_db_path, config, self)
+        navidrome_db = config["navidrome"]["database"].get(str)
+        self.db = NavidromeDb(navidrome_db, self)
         self.stats = Stats(self)
 
     def merge_and_store_annotations(self):
@@ -120,7 +122,7 @@ def eval_deletable_duplicates(self):
             keepable = self._get_keepable_media(dups)
             PU.log(f"<- Found keepable: {keepable.path}", 0)
 
-        file_path = os.path.join(self.config.data_folder, "duplicates-with-keepers.json")
+        file_path = os.path.join(config["data"].get(str), "duplicates-with-keepers.json")
         with open(file_path, "w", encoding="utf-8") as file:
             file.write(jsonpickle.encode(self.data.media, indent=4))
 
@@ -165,18 +167,18 @@ def load_navidrome_database(self):
         PU.ln()
 
         # Read raw duplicates info generated by the Beets `duplicatez` plugin.
-        PU.info(f"Reading duplicates from Beets JSON file: {self.config.FILE_BEETS_INPUT_JSON}")
+        PU.info(f"Reading duplicates from Beets JSON file: {config["FILE_BEETS_INPUT_JSON"].get(str)}")
         # Read the input JSON file containing duplicate media files references from Beets.
-        with open(self.config.FILE_BEETS_INPUT_JSON, "r", encoding="utf-8") as file:
+        with open(config["FILE_BEETS_INPUT_JSON"].get(str), "r", encoding="utf-8") as file:
             dups_input = json.load(file)
         if not dups_input:
-            PU.error(f"No duplicates found in input file '{self.config.FILE_BEETS_INPUT_JSON}'")
+            PU.error(f"No duplicates found in input file '{config["FILE_BEETS_INPUT_JSON"].get(str)}'")
             PU.info("Please generate the duplicates info using Beets `duplicatez` plugin first.")
             sys.exit(1)
 
         # Check for existing data file.
-        if os.path.isfile(self.config.FILE_TOOLBOX_DATA_JSON):
-            PU.note(f"Data file '{self.config.FILE_TOOLBOX_DATA_JSON}' existing already.")
+        if os.path.isfile(config["FILE_TOOLBOX_DATA_JSON"].get(str)):
+            PU.note(f"Data file '{config["FILE_TOOLBOX_DATA_JSON"].get(str)}' existing already.")
             PU.note("Do you want to continue anyway and overwrite existing data?")
             CLI.ask_continue()
 
@@ -186,14 +188,14 @@ def load_navidrome_database(self):
 
         # Persist data.
         data = {"dups_media_files": self.data.media, "stats": self.stats, "cache": self.data}
-        with open(self.config.FILE_TOOLBOX_DATA_JSON, "w", encoding="utf-8") as file:
+        with open(config["FILE_TOOLBOX_DATA_JSON"].get(str), "w", encoding="utf-8") as file:
             file.write(jsonpickle.encode(data, indent=4, keys=True))
-        PU.success(f"Stored Navidrome data to '{self.config.FILE_TOOLBOX_DATA_JSON}'")
+        PU.success(f"Stored Navidrome data to '{config["FILE_TOOLBOX_DATA_JSON"].get(str)}'")
         self.stats.stop()
         self.stats.print_stats()
         if self._has_errors():
-            PU.error(f"Please review {len(self.errors)} errors in {self.config.ERROR_REPORT_JSON} ... ")
-            with open(self.config.ERROR_REPORT_JSON, "w") as f:
+            PU.error(f"Please review {len(self.errors)} errors in {config["ERROR_REPORT_JSON"].get(str)}...")
+            with open(config["ERROR_REPORT_JSON"].get(str), "w") as f:
                 json.dump(self.errors, f, indent=4)
         else:
             PU.success("No errors found.")
@@ -204,20 +206,20 @@ def _load_navidrome_data_file(self):
         Load data from a previously saved JSON file.
         """
         # Check for existing errors file.
-        if os.path.isfile(self.config.ERROR_REPORT_JSON):
-            PU.error(f"Errors file '{self.config.ERROR_REPORT_JSON}' found.")
+        if os.path.isfile(config["ERROR_REPORT_JSON"].get(str)):
+            PU.error(f"Errors file '{config["ERROR_REPORT_JSON"].get(str)}' found.")
             PU.note("Have you checked the errors and decided to continue anyway?")
             CLI.ask_continue()
 
         PU.bold("Loading duplicate records from JSON file")
         PU.ln()
         # Load data from JSON file.
-        with open(self.config.FILE_TOOLBOX_DATA_JSON, "r", encoding="utf-8") as file:
+        with open(config["FILE_TOOLBOX_DATA_JSON"].get(str), "r", encoding="utf-8") as file:
             data = jsonpickle.decode(file.read())
             self.data.media = data["dups_media_files"]
             self.stats = data["stats"]
             self.data = data["cache"]
-            PU.success(f"Loaded duplicate records from '{self.config.FILE_TOOLBOX_DATA_JSON}'")
+            PU.success(f"Loaded duplicate records from '{config["FILE_TOOLBOX_DATA_JSON"].get(str)}'")
 
     def _split_duplicates_by_album_folder(
         self, dups_media_files: dict[str, list[MediaFile]]
@@ -293,18 +295,22 @@ def _replace_base_path(self, dups_input: dict[str, list[MediaFile]]):
             dups_input (dict[str, list[str]]): A dictionary where the keys are duplicate identifiers and the values
                 are lists of file paths.
         """
-        if not self.config.base_path_beets or not self.config.base_path_navidrome:
+        if not config["beets"]["base-path"].get(str) or not config["navidrome"]["base-path"].get(str):
             PU.warning("Skipping base path update, since no paths are set")
             return
-        if self.config.base_path_beets == self.config.base_path_navidrome:
+        if config["beets"]["base-path"].get(str) == config["navidrome"]["base-path"].get(str):
             PU.warning("Skipping base path update as target equals source")
             return
 
         for paths in dups_input.values():
             item: str
             for i, item in enumerate(paths):
-                paths[i] = item.replace(self.config.base_path_beets, self.config.base_path_navidrome, 1)
-        PU.info(f"Updated all base paths from '{self.config.base_path_beets}' to '{self.config.base_path_navidrome}'.")
+                paths[i] = item.replace(
+                    config["beets"]["base-path"].get(str), config["navidrome"]["base-path"].get(str), 1
+                )
+        PU.info(
+            f"Updated all base paths from '{config["beets"]["base-path"].get(str)}' to '{config["navidrome"]["base-path"].get(str)}'."
+        )
 
     def _query_media_data(self, dups_input: dict[str, list[str]]):
         """
@@ -532,8 +538,8 @@ def is_keepable(self, this: MediaFile, that: MediaFile) -> MediaFile:
         # Skip, if none is a suffixed path
 
         # Having a preferred file extension is keepable
-        left = this.path.split(".")[-1].lower() in ToolboxConfig.pref_extensions
-        right = that.path.split(".")[-1].lower() in ToolboxConfig.pref_extensions
+        left = this.path.split(".")[-1].lower() in config["pref-extensions"].get(list)
+        right = that.path.split(".")[-1].lower() in config["pref-extensions"].get(list)
         PU.log(f"Compare if file extension is keepable: {left} || {right}", 1)
         if left != right:
             if left:
@@ -652,9 +658,32 @@ def is_keepable(self, this: MediaFile, that: MediaFile) -> MediaFile:
         this.delete_reason = f"No reason, since no condition matched | {SU.gray(that.path)}"
         return that
 
+    #    self.init_logger()
+    #     self.print_info()
+
+
+def print_info():
+    """Prints the current configuration details."""
+    # Print app version
+    with open("pyproject.toml", mode="rb") as file:
+        data = tomli.load(file)
+        version = data["tool"]["poetry"]["version"]
+        PrintUtil.ln()
+        PrintUtil.bold(f"  Heartbeets v{version}")
+        PrintUtil.ln()
+
+    # Print config details
+    PrintUtil.bold("\nInitializing configuration")
+    PrintUtil.ln()
+    PrintUtil.info(f"Dry-run: {config["dry-run"].get(bool)}")
+    PrintUtil.info(f"Navidrome database path: {config["navidrome"]["database"].get(str)}")
+    PrintUtil.info(f"Output folder: {config["data"].get(str)}")
+    PrintUtil.info(f"Beets library root: {config["beets"]["base-path"].get(str)}")
+    PrintUtil.info(f"Navidrome library root: {config["navidrome"]["base-path"].get(str)}")
+
 
 if __name__ == "__main__":
-    config = ToolboxConfig()
+    print_info()
 
     # Read the action argument from the command line
     action = ""
@@ -666,12 +695,14 @@ def is_keepable(self, this: MediaFile, that: MediaFile) -> MediaFile:
         PU.error("No action specified. Please use the format 'python app.py action=<action>'.")
         sys.exit(1)
 
-    processor = DuplicateProcessor(config)
-    # client = BeetsClient(config)
-    # print("BEETS CLIENT: " + client._query1())
+    processor = DuplicateProcessor()
 
     if action == "remove-unsupported":
-        FileTools.move_by_extension(config.music_dir, config.data_dir, config.remove_extensions, ToolboxConfig.dry_run)
+        music = config["music"].get(str)
+        data = config["data"].get(str)
+        remove_ext = config["remove-extensions"].get(list)
+        dry_run = config["music"].get(str)
+        FileTools.move_by_extension(music, data, remove_ext, dry_run)
     elif action == "load-duplicates":
         processor.load_navidrome_database()
     elif action == "merge-annotations":

src/ndtoolbox/config.py

@@ -0,0 +1,54 @@
+"""App configuration."""
+
+import logging
+import os
+
+import colorlog
+import confuse
+
+
+class Config(confuse.Configuration):
+    """
+    Configuration for application based on Confuse.
+    """
+
+    logger: logging.Logger = None
+
+    def init(self):
+        """Init configuration."""
+        dir_data = self["data"].get(str)
+        file_duplicates_json = os.path.join(dir_data, "beets/beets-duplicates.json")
+        file_data_json = os.path.join(dir_data, "nd-toolbox-data.json")
+        file_error_json = os.path.join(dir_data, "nd-toolbox-error.json")
+        file_log = os.path.join(dir_data, "nd-toolbox.log")
+        self.set_args(
+            {
+                "FILE_BEETS_INPUT_JSON": file_duplicates_json,
+                "FILE_TOOLBOX_DATA_JSON": file_data_json,
+                "ERROR_REPORT_JSON": file_error_json,
+                "file-log": file_log,
+            }
+        )
+
+    def init_logger(self):
+        """Setup logger."""
+        log_level = self["log-level"].get(str)
+        file_log = self["file-log"].get(str)
+        if log_level not in logging._nameToLevel:
+            raise ValueError(f"Invalid log-level: {log_level}")
+        self.logger = colorlog.getLogger("ndtoolbox")
+
+        colorlog.basicConfig(
+            filename=file_log,
+            filemode="w",
+            encoding="utf-8",
+            level=log_level,
+            format="%(log_color)s %(msecs)d %(name)s %(levelname)s %(message)s",
+        )
+        self.logger.info(f"Initialized logger with level: {log_level} and log file: {file_log}")
+
+
+config = Config("Heartbeets")
+config.set_file("config/config.yaml")
+config.init()
+config.init_logger()

src/ndtoolbox/db.py

@@ -6,9 +6,10 @@
 import unicodedata
 from typing import Generator
 
+from ndtoolbox.config import config
 from ndtoolbox.model import Album, Annotation, Artist, Folder, MediaFile
 from ndtoolbox.utils import DateUtil as DU
-from ndtoolbox.utils import FileUtil, ToolboxConfig
+from ndtoolbox.utils import FileUtil
 from ndtoolbox.utils import PrintUtil as PU
 
 
@@ -57,20 +58,18 @@ class NavidromeDb:
     """
 
     db_path: str
-    config: ToolboxConfig
     app: object
     user_id: str
     conn: NavidromeDbConnection
 
-    def __init__(self, db_path: str, config: ToolboxConfig, app):
+    def __init__(self, db_path: str, app):
         """
         Initialize the database connection and set the user ID.
 
         Args:
             db_path (str): Path to the database file.
         """
         NavidromeDbConnection.db_path = db_path
-        self.config = config
         self.app = app
         self.user_id = self.init_user()
 
@@ -166,7 +165,9 @@ def get_media_batch(self, file_paths: list[str], conn: NavidromeDbConnection) ->
             # and `á` (`\u00e1`) are not threaded as the same.
             nd_path = unicodedata.normalize("NFC", beets_path)
             # Restore the Beets prefix
-            beets_path = beets_path.replace(self.config.base_path_navidrome, self.config.base_path_beets, 1)
+            beets_path = beets_path.replace(
+                config["navidrome"]["base-path"].get(str), config["beets"]["base-path"].get(str), 1
+            )
             path_mapping[nd_path] = beets_path
 
         cursor = conn.cursor()

src/ndtoolbox/model.py

@@ -2,14 +2,14 @@
 Model classes representing the Navidrome database.
 """
 
-import os
 from datetime import datetime
 from enum import Enum
 from typing import Optional
 
 from ndtoolbox.beets import BeetsClient
+from ndtoolbox.config import config
 from ndtoolbox.utils import DateUtil as DU
-from ndtoolbox.utils import FileUtil, ToolboxConfig
+from ndtoolbox.utils import FileUtil
 from ndtoolbox.utils import PrintUtil as PU
 
 
@@ -249,11 +249,11 @@ def __init__(self, media: MediaFile):
 
         # Set folder type
         self.type = Folder.Type.ALBUM
-        if self.beets_path == ToolboxConfig.base_path_beets:
+        if self.beets_path == config["beets"]["base-path"].get(str):
             self.type = Folder.Type.ROOT
-        elif FileUtil.is_artist_folder(ToolboxConfig.base_path_beets, self.beets_path):
+        elif FileUtil.is_artist_folder(config["beets"]["base-path"].get(str), self.beets_path):
             self.type = Folder.Type.ARTIST
-        elif FileUtil.is_album_folder(ToolboxConfig.base_path_beets, self.beets_path):
+        elif FileUtil.is_album_folder(config["base-path"].get(str), self.beets_path):
             self.type = Folder.Type.ALBUM
         else:
             self.type = Folder.Type.UNKNOWN