app_external.py

# AudioMuse-AI - https://github.com/NeptuneHub/AudioMuse-AI
# Copyright (C) 2025 NeptuneHub
# SPDX-License-Identifier: AGPL-3.0-only
#
# This program is free software: you can redistribute it and/or modify it under
# the terms of the GNU Affero General Public License v3.0. See the LICENSE file
# in the project root or <https://github.com/NeptuneHub/AudioMuse-AI/blob/main/LICENSE>

"""Flask blueprint for the external track-lookup API (mounted at `/external`).

Read-only endpoints that expose stored analysis for third-party integrations,
returning per-track scores/embeddings and running unified similarity search via
`tasks.ivf_manager.search_tracks_unified`.

Main Features:
* Routes: `/get_score`, `/get_embedding` (by item id) and `/search` (similarity).
* Imports `get_db` lazily inside each handler to avoid a circular import.
"""

from flask import Blueprint, jsonify, request
from psycopg2.extras import DictCursor
import numpy as np
import logging

# Import ivf_manager functions for track lookups
from tasks.ivf_manager import search_tracks_unified
from error import error_manager
from error.error_dictionary import ERR_DB_QUERY
# NOTE: The import of 'get_db' has been moved inside each function to prevent circular imports.

logger = logging.getLogger(__name__)

# Create a Blueprint for external API routes
external_bp = Blueprint('external_bp', __name__)


@external_bp.route('/get_score', methods=['GET'])
def get_score_endpoint():
    """
    Get all content from the score database for a given id.
    ---
    tags:
      - External
    parameters:
      - name: id
        in: query
        required: true
        description: The Item ID of the track.
        schema:
          type: string
    responses:
      200:
        description: Score data for the track.
        content:
          application/json:
            schema:
              type: object
      400:
        description: Missing id parameter.
      404:
        description: Score not found for the given id.
      500:
        description: Internal server error.
    """
    # Local import to prevent circular dependency
    from app_helper import get_db

    item_id = request.args.get('id')
    if not item_id:
        return jsonify({"error": "Missing 'id' parameter"}), 400

    try:
        db = get_db()
        with db.cursor(cursor_factory=DictCursor) as cur:
            cur.execute("SELECT * FROM score WHERE item_id = %s", (item_id,))
            score_data = cur.fetchone()

        if score_data:
            # Convert DictRow to a standard dictionary for consistent JSON output
            return jsonify(dict(score_data))
        else:
            return jsonify({"error": f"Score not found for id: {item_id}"}), 404
    except Exception as e:
        logger.exception(f"Error fetching score for id {item_id}")
        err, status = error_manager.error_response(error_manager.classify(e, ERR_DB_QUERY))
        return jsonify(err), status


@external_bp.route('/get_embedding', methods=['GET'])
def get_embedding_endpoint():
    """
    Get the embedding vector from the database for a given id.
    ---
    tags:
      - External
    parameters:
      - name: id
        in: query
        required: true
        description: The Item ID of the track.
        schema:
          type: string
    responses:
      200:
        description: Embedding data for the track, with the vector as a list of floats.
      400:
        description: Missing id parameter.
      404:
        description: Embedding not found for the given id.
      500:
        description: Internal server error.
    """
    # Local import to prevent circular dependency
    from app_helper import get_db

    item_id = request.args.get('id')
    if not item_id:
        return jsonify({"error": "Missing 'id' parameter"}), 400

    try:
        db = get_db()
        with db.cursor(cursor_factory=DictCursor) as cur:
            cur.execute("SELECT * FROM embedding WHERE item_id = %s", (item_id,))
            embedding_data = cur.fetchone()

        if embedding_data:
            embedding_dict = dict(embedding_data)
            if embedding_dict.get('embedding'):
                # The embedding is stored as BYTEA, convert it back to a list of floats
                embedding_vector = np.frombuffer(embedding_dict['embedding'], dtype=np.float32)
                embedding_dict['embedding'] = embedding_vector.tolist()
            return jsonify(embedding_dict)
        else:
            return jsonify({"error": f"Embedding not found for id: {item_id}"}), 404
    except Exception as e:
        logger.exception(f"Error fetching embedding for id {item_id}")
        err, status = error_manager.error_response(error_manager.classify(e, ERR_DB_QUERY))
        return jsonify(err), status


@external_bp.route('/search', methods=['GET'])
def search_tracks_endpoint():
    """
    Provides autocomplete suggestions for tracks based on a unified search query
    or legacy title/artist parameters.
    A query must be at least 3 characters long.
    ---
    tags:
      - External
    parameters:
      - name: search_query
        in: query
        description: Partial or full elements of songs' titles, artist or album names.
        schema:
          type: string
      - name: title
        in: query
        description: (Legacy) Partial or full title of the track. Used as fallback when search_query is absent.
        schema:
          type: string
      - name: artist
        in: query
        description: (Legacy) Partial or full name of the artist. Used as fallback when search_query is absent.
        schema:
          type: string
    responses:
      200:
        description: A list of matching tracks.
      400:
        description: Query string too short.
      500:
        description: Internal server error.
    """
    search_query = request.args.get('search_query', '', type=str)

    # Backward compatibility: support legacy 'title' and 'artist' params
    # so external apps using the old API continue to work.
    if not search_query:
        legacy_title = request.args.get('title', '', type=str).strip()
        legacy_artist = request.args.get('artist', '', type=str).strip()
        search_query = f"{legacy_artist} {legacy_title}".strip()

    # Return empty list if query is empty
    if not search_query:
        return jsonify([])

    # Enforce minimum length constraint
    if len(search_query) < 1:
        return jsonify({"error": "Query must be at least 1 character long"}), 400

    try:
        results = search_tracks_unified(search_query)
        return jsonify(results)
    except Exception:
        logger.exception("Error during external track search")
        return jsonify({"error": "An error occurred during search."}), 500