Centralize and remove duplicated functionality

Author: NeptuneHub

CONTRIBUTING.md

@@ -28,7 +28,7 @@ The following table details the most important paths in the repository, their pu
 | :---- | :---- |
 | app.py, app_*.py | The main entry point for the Flask web application. It handles the initialization of the Flask app, database connections, and the registration of API routes and blueprints. |
 | tasks/ | **The Core Logic Hub.** This is where the most intensive computations occur. Each API or async task then point to an specific implementation in this directory|
-| tasks/mediaserver/ | In this package the generic methods to interact with the mediaservers (`__init__.py`) dispatch to the specific backend (`jellyfin.py`, `navidrome.py`, `emby.py`, `lyrion.py`, `mpd.py`) |
+| tasks/mediaserver/ | In this package the generic methods to interact with the mediaservers (`__init__.py`) dispatch to the specific backend (`jellyfin.py`, `navidrome.py`, `emby.py`, `lyrion.py`) |
 | tasks/ai/ | All AI / MCP code. `tasks/ai/api.py` is the provider dispatcher (called via `tasks.ai.api.call_with_tools()`), with backends in `tasks/ai/providers/{openai,gemini,mistral,ollama}.py`. Prompts are in `tasks/ai/prompts.py`. MCP tool schemas + dispatcher in `tasks/ai/tools.py`; tool bodies in `tasks/ai/tool_impl.py`. Two-stage planner (intent classifier + plan validation + execution) in `tasks/ai/planner.py`. Vocabulary normalization helpers in `tasks/ai/vocab.py`. |
 | config.py | Contains the application's default, non-sensitive configuration parameters. These values serve as fallbacks and can be easily overridden by environment variables, providing a flexible and secure configuration system. |
 | Authentication | Configured in `config.py` by `AUTH_ENABLED`, `AUDIOMUSE_USER`, `AUDIOMUSE_PASSWORD`, `API_TOKEN`, and `JWT_SECRET`. Enforcement happens in `app.py` and `app_helepr.py` functionality |

app_clustering.py

@@ -247,7 +247,7 @@ def start_clustering_endpoint():
     """
     # Local imports to prevent circular dependency at startup
     from app_helper import rq_queue_high, get_active_main_task
-    from app_helper import clean_up_previous_main_tasks, save_task_status, TASK_STATUS_PENDING, TASK_STATUS_FAILURE
+    from app_helper import clean_up_previous_main_tasks, save_task_status, TASK_STATUS_PENDING
 
     # Check for any existing active main task to prevent parallel batch runs
     active_task = get_active_main_task()

app_helper.py

@@ -10,9 +10,8 @@
 import psycopg2
 from psycopg2.extras import DictCursor
 import numpy as np
-from flask import g
 
-from database import get_db, close_db
+from database import get_db
 from taskqueue import (
     redis_conn,
     rq_queue_high,
@@ -1775,18 +1774,7 @@ def cancel_job_and_children_recursive(job_id, task_type_from_db=None, reason="Ta
 # JWT handling, the Flask routes) lives in ``app_auth``. The re-exports
 # below keep the legacy ``from app_helper import ...`` paths working.
 from app_auth import (  # noqa: E402  (intentional late import to avoid cycles)
-    USER_ROLE_USER,
-    USER_ROLE_ADMIN,
     check_setup_needed,
-    check_auth_needed,
-    check_admin_needed,
-    is_admin_path,
-    list_additional_users,
     count_admin_users,
-    get_additional_user_by_id,
-    create_additional_user,
-    delete_additional_user_safe,
-    verify_additional_user,
     upsert_admin_user,
-    seed_admin_from_env,
 )

app_map.py

@@ -10,12 +10,11 @@
 
 # Try to reuse the shared projection helpers
 try:
-    from tasks.alchemy_projections import _project_with_umap, _project_to_2d, _project_aligned_add_sub, _project_with_discriminant
+    from tasks.alchemy_projections import _project_with_umap, _project_to_2d, _project_with_discriminant
 except Exception:
     # Fallbacks will be used if import fails
     _project_with_umap = None
     _project_to_2d = None
-    _project_aligned_add_sub = None
     _project_with_discriminant = None
 
 logger = logging.getLogger(__name__)

docs/ALGORITHM.md

@@ -266,13 +266,12 @@ The Song Analysis functionality is configured by the following environment varia
 
 #### **Media Server**
 
-* MEDIASERVER\_TYPE: **(Required)** Specifies the media server to connect to. (e.g., jellyfin, navidrome, emby, lyrion, mpd).  
+* MEDIASERVER\_TYPE: **(Required)** Specifies the media server to connect to. (e.g., jellyfin, navidrome, emby, lyrion).  
 * MUSIC\_LIBRARIES: (Optional) A comma-separated list of library names to scan. If empty, all music libraries are scanned.  
 * JELLYFIN\_URL, JELLYFIN\_USER\_ID, JELLYFIN\_TOKEN: Credentials for Jellyfin (if MEDIASERVER\_TYPE="jellyfin").  
 * EMBY\_URL, EMBY\_USER\_ID, EMBY\_TOKEN: Credentials for Emby (if MEDIASERVER\_TYPE="emby").  
 * NAVIDROME\_URL, NAVIDROME\_USER, NAVIDROME\_PASSWORD: Credentials for Navidrome (if MEDIASERVER\_TYPE="navidrome").  
-* LYRION\_URL: Credentials for Lyrion (if MEDIASERVER\_TYPE="lyrion").  
-* MPD\_HOST, MPD\_PORT, MPD\_PASSWORD, MPD\_MUSIC\_DIRECTORY: Credentials for MPD (if MEDIASERVER\_TYPE="mpd").
+* LYRION\_URL: Credentials for Lyrion (if MEDIASERVER\_TYPE="lyrion").
 
 #### **Task & Performance Tuning**
 

rq_janitor.py

@@ -7,7 +7,7 @@
 
 try:
     # We need the queue objects to get their registries
-    from app_helper import redis_conn, rq_queue_high, rq_queue_default
+    from app_helper import rq_queue_high, rq_queue_default
     from app_logging import configure_logging
 except ImportError as e:
     print(f"Error importing from app.py: {e}")

tasks/clap_text_search.py

@@ -3,12 +3,8 @@
 Provides in-memory caching and fast text-based music search using CLAP embeddings.
 """
 
-import gc
-import json
 import logging
-import re
 import sys
-import tempfile
 import threading
 import time
 
@@ -60,145 +56,35 @@ def get_clap_cache_size() -> int:
 
 
 def _fetch_clap_metadata(item_ids: list) -> Dict[str, Dict[str, str]]:
-    """Fetch metadata for CLAP result item_ids from the database."""
-    metadata_map: Dict[str, Dict[str, str]] = {}
-    if not item_ids:
-        return metadata_map
-
-    from app_helper import get_score_data_by_ids
-    try:
-        track_details_list = get_score_data_by_ids(item_ids)
-        for row in track_details_list:
-            item_id = row['item_id']
-            metadata_map[item_id] = {
-                'title': row.get('title', ''),
-                'author': row.get('author', ''),
-                'album': row.get('album', ''),
-            }
-    except Exception:
-        pass
-
-    return metadata_map
+    from .commons import fetch_track_metadata_map
+    return fetch_track_metadata_map(item_ids)
 
 
 def _load_clap_index_from_db() -> bool:
     """Load a persisted CLAP voyager index from the database."""
 
     from app_helper import get_db
     from config import CLAP_EMBEDDING_DIMENSION, VOYAGER_QUERY_EF
+    from .index_build_helpers import load_voyager_index_from_db
 
     try:
-        conn = get_db()
-        with conn.cursor() as cur:
-            cur.execute("SET LOCAL statement_timeout = 0")
-            cur.execute(
-                "SELECT index_data, id_map_json, embedding_dimension FROM clap_index_data WHERE index_name = %s",
-                ('clap_index',)
-            )
-            row = cur.fetchone()
-
-            index_stream = None
-            try:
-                if row:
-                    index_binary_data, id_map_json, db_embedding_dim = row
-                    index_stream = tempfile.TemporaryFile()
-                    index_stream.write(index_binary_data)
-                    index_stream.seek(0)
-                else:
-                    seg_pattern = re.compile(r'^clap_index_(\d+)_(\d+)$')
-                    parts = []
-                    total_expected = None
-                    with conn.cursor(name='clap_index_segments') as seg_cur:
-                        seg_cur.itersize = 50
-                        seg_cur.execute(
-                            "SELECT index_name, index_data, id_map_json, embedding_dimension FROM clap_index_data WHERE index_name LIKE %s ESCAPE '\\'",
-                            (r'clap_index\_%\_%',)
-                        )
-                        for name, part_data, part_id_map_json, part_dim in seg_cur:
-                            m = seg_pattern.match(name)
-                            if not m:
-                                continue
-                            part_no = int(m.group(1))
-                            total = int(m.group(2))
-                            if total_expected is None:
-                                total_expected = total
-                            elif total_expected != total:
-                                logger.error(f"Segment total mismatch for CLAP index parts ({total_expected} vs {total}).")
-                                return False
-                            parts.append((part_no, part_data, part_id_map_json, part_dim))
-
-                    if total_expected is None or len(parts) != total_expected:
-                        logger.error(f"Incomplete CLAP index segments: expected {total_expected}, found {len(parts)}.")
-                        return False
-
-                    parts.sort(key=lambda p: p[0])
-                    from .index_build_helpers import reassemble_segmented_id_map
-                    id_map_json_candidate = reassemble_segmented_id_map((p[0], p[2]) for p in parts)
-                    for _, _, _, part_dim in parts:
-                        if part_dim != CLAP_EMBEDDING_DIMENSION:
-                            logger.error(f"CLAP index embedding_dimension mismatch in segmented parts: expected {CLAP_EMBEDDING_DIMENSION}, got {part_dim}.")
-                            return False
-
-                    if not id_map_json_candidate:
-                        logger.error("No id_map_json found in segmented CLAP index rows.")
-                        return False
-
-                    db_embedding_dim = parts[0][3]
-                    index_stream = tempfile.TemporaryFile()
-                    for _, part_data, _, _ in parts:
-                        index_stream.write(part_data)
-                    index_stream.seek(0)
-                    id_map_json = id_map_json_candidate
-
-                if index_stream is None:
-                    logger.error("CLAP index binary data was empty.")
-                    return False
-
-                if db_embedding_dim != CLAP_EMBEDDING_DIMENSION:
-                    logger.error(f"CLAP index dimension mismatch: db={db_embedding_dim} expected={CLAP_EMBEDDING_DIMENSION}")
-                    index_stream.close()
-                    return False
-
-                try:
-                    try:
-                        import voyager  # type: ignore
-                    except ImportError:
-                        logger.warning("Voyager library is unavailable; cannot load persisted CLAP index.")
-                        return False
-
-                    loaded_index = voyager.Index.load(index_stream)
-                    loaded_index.ef = VOYAGER_QUERY_EF
-                finally:
-                    if index_stream is not None:
-                        try:
-                            index_stream.close()
-                        except Exception as close_error:
-                            logger.warning("Failed to close CLAP index stream: %s", close_error, exc_info=True)
-
-            except Exception:
-                if index_stream is not None:
-                    try:
-                        index_stream.close()
-                    except Exception:
-                        pass
-                raise
-
-            id_map = {int(k): v for k, v in json.loads(id_map_json).items()}
-            reverse_id_map = {v: k for k, v in id_map.items()}
-
-            if not id_map:
-                logger.error("CLAP index id_map is empty.")
-                return False
-
-            _CLAP_CACHE['loaded'] = True
-
-            _CLAP_INDEX_CACHE['index'] = loaded_index
-            _CLAP_INDEX_CACHE['id_map'] = id_map
-            _CLAP_INDEX_CACHE['reverse_id_map'] = reverse_id_map
-            _CLAP_INDEX_CACHE['loaded'] = True
-
-            logger.info(f"CLAP index loaded from database with {len(id_map)} items.")
-            return True
+        loaded = load_voyager_index_from_db(
+            get_db(), 'clap_index_data', 'clap_index',
+            CLAP_EMBEDDING_DIMENSION, VOYAGER_QUERY_EF, label='CLAP',
+        )
+        if loaded is None:
+            return False
+        loaded_index, id_map, reverse_id_map = loaded
+
+        _CLAP_CACHE['loaded'] = True
+
+        _CLAP_INDEX_CACHE['index'] = loaded_index
+        _CLAP_INDEX_CACHE['id_map'] = id_map
+        _CLAP_INDEX_CACHE['reverse_id_map'] = reverse_id_map
+        _CLAP_INDEX_CACHE['loaded'] = True
+
+        logger.info(f"CLAP index loaded from database with {len(id_map)} items.")
+        return True
     except Exception as e:
         logger.error(f"Failed to load CLAP index from DB: {e}", exc_info=True)
         return False
@@ -208,63 +94,21 @@ def build_and_store_clap_index(db_conn=None):
     """Build a CLAP text search voyager index from stored CLAP embeddings and save it to the DB."""
     from app_helper import get_db
     from config import CLAP_EMBEDDING_DIMENSION, VOYAGER_METRIC
-    from .index_build_helpers import (
-        iter_embedding_batches,
-        build_voyager_index_bytes_streaming,
-        store_voyager_index_segmented,
-        build_id_map,
-        EmptyIndexError,
-    )
-
-    try:
-        import voyager  # type: ignore  # noqa: F401
-    except ImportError:
-        logger.warning("Voyager library is unavailable; cannot build CLAP index.")
-        return False
+    from .index_build_helpers import build_and_store_index_streaming
 
     if db_conn is None:
         db_conn = get_db()
 
-    try:
-        logger.info("Building CLAP voyager index (streaming)...")
-        batches = iter_embedding_batches(
-            table="clap_embedding",
-            column="embedding",
-            dim=CLAP_EMBEDDING_DIMENSION,
-        )
-        try:
-            index_bytes, item_ids = build_voyager_index_bytes_streaming(
-                batches, CLAP_EMBEDDING_DIMENSION, metric=VOYAGER_METRIC,
-            )
-        except EmptyIndexError as ve:
-            logger.warning(f"No valid CLAP embedding vectors found for CLAP index build: {ve}")
-            return False
-        gc.collect()
-
-        if not index_bytes:
-            logger.error("Generated CLAP index binary is empty. Aborting storage.")
-            return False
-
-        id_map = build_id_map(item_ids)
-        store_voyager_index_segmented(
-            db_conn,
-            target_table="clap_index_data",
-            index_name="clap_index",
-            index_bytes=index_bytes,
-            id_map=id_map,
-            embedding_dimension=CLAP_EMBEDDING_DIMENSION,
-        )
-
-        db_conn.commit()
-        logger.info("CLAP text search index build successful.")
-        return True
-    except Exception as e:
-        logger.error(f"Failed to build and store CLAP index: {e}", exc_info=True)
-        try:
-            db_conn.rollback()
-        except Exception:
-            pass
-        return False
+    return build_and_store_index_streaming(
+        db_conn,
+        source_table="clap_embedding",
+        source_column="embedding",
+        dim=CLAP_EMBEDDING_DIMENSION,
+        target_table="clap_index_data",
+        index_name="clap_index",
+        metric=VOYAGER_METRIC,
+        label="CLAP",
+    )
 
 
 def _unload_timer_worker():
@@ -360,9 +204,8 @@ def load_clap_cache_from_db():
     Returns True if successful, False otherwise.
     """
 
-    from app_helper import get_db
     from config import CLAP_ENABLED
-    
+
     if not CLAP_ENABLED:
         logger.info("CLAP is disabled, skipping cache load.")
         return False

tasks/clustering.py

@@ -668,7 +668,7 @@ def _monitor_and_process_batches(state_dict, parent_task_id, initial_check=False
     CRITICAL: This prevents the main task from hanging at 4980/5000 runs
     by implementing timeouts and forced progress tracking.
     """
-    from app_helper import redis_conn, get_child_tasks_from_db, get_task_info_from_db, TASK_STATUS_SUCCESS, TASK_STATUS_FAILURE, TASK_STATUS_REVOKED, TASK_STATUS_STARTED, TASK_STATUS_PROGRESS
+    from app_helper import redis_conn, get_child_tasks_from_db, TASK_STATUS_SUCCESS, TASK_STATUS_FAILURE, TASK_STATUS_REVOKED, TASK_STATUS_STARTED
 
     current_time = time.time()
     timeout_seconds = CLUSTERING_BATCH_TIMEOUT_MINUTES * 60