"""Audio Attribute queries.""" from sqlalchemy.sql import text from backend.connectors import mysql from backend.models.audio_attribute import AudioAttributes from backend.models.audio_attribute import TrackAudioAttributes from backend.models.audio_attribute import TrackAudioAttributesEdits from backend.utils import api as api_utils class AudioAttributeQuery(object): """Handles Audio Attribute queries.""" @classmethod @mysql.db_session_wrap def get_audio_attributes(cls, session): """Get all possible audio attributes from DB. Args: session (object): SQLAlchemy database session (optional) Returns: List of all possible audio attributes """ audio_attributes = session.query(AudioAttributes) \ .order_by(AudioAttributes.id).all() return api_utils.create_get_list_response( [aa.to_dict() for aa in audio_attributes]) @classmethod @mysql.db_session_wrap def get_track_audio_attributes(cls, tuid, session): """Get a track's currently set audio attributes from DB. Args: tuid (int): unique id of track session (object): SQLAlchemy database session (optional) Returns: List of the track's currently set audio attributes, or None if tuid not found in the database """ track_audio_attributes = session.query(TrackAudioAttributes) \ .filter(TrackAudioAttributes.unique_track_id == tuid) \ .order_by(TrackAudioAttributes.audio_attribute_id) if track_audio_attributes.count() != 0: return api_utils.create_get_list_response( [{'track_audio_attribute_id': taa.id, 'audio_attribute_id': taa.audio_attribute_id, 'description': taa.audio_attributes.description } for taa in track_audio_attributes]) vendor_audio_attributes_query = text( 'SELECT aa.description, aa.id ' 'FROM track t ' 'INNER JOIN releases r ' 'ON t.release_id = r.release_id ' 'INNER JOIN project p ' 'ON r.project_id = p.project_id ' 'INNER JOIN vendor_audio_attributes vaa ' 'ON p.vendor_id = vaa.vendor_id ' 'INNER JOIN audio_attributes aa ' 'ON vaa.audio_attribute_id = aa.id ' 'WHERE t.id = :tuid ' 'GROUP BY 1, 2 ' 'ORDER BY 2' ) vendor_audio_attributes = session.execute( vendor_audio_attributes_query, {'tuid': tuid}).fetchall() if len(vendor_audio_attributes) != 0: return api_utils.create_get_list_response( [{'track_audio_attribute_id': 0, 'audio_attribute_id': vaa[1], 'description': vaa[0]} for vaa in vendor_audio_attributes]) return api_utils.create_get_list_response([]) @classmethod @mysql.db_session_wrap def get_track_suggested_audio_attributes(cls, tuid, session): """Get suggested audio attributes from DB for a track. Args: tuid (int): unique id of track session (object): SQLAlchemy database session (optional) Returns: List of suggested audio attributes with reasons, or empty list if tuid not found in the database """ # SQLite (unit tests) does not support CONCAT; use || instead dialect = session.bind.dialect.name if dialect == 'sqlite': concat_expr = "'\\b' || aask.keyword || '\\b'" else: concat_expr = "CONCAT('\\\\b', aask.keyword, '\\\\b')" # Each branch returns: (aa_id, description, reason_type, keyword, matched_on, subgenre_id) suggestion_query = text( 'SELECT aa.id, aa.description, \'keyword_match\', aask.keyword, \'track_name\', NULL ' 'FROM track t ' 'INNER JOIN releases r ON t.release_id = r.release_id ' 'INNER JOIN audio_attributes_suggestion_keywords aask ' f'ON LOWER(t.track_name) REGEXP {concat_expr} ' 'INNER JOIN audio_attributes aa ON aa.id = aask.audio_attribute_id ' 'WHERE t.id = :tuid ' 'UNION ALL ' 'SELECT aa.id, aa.description, \'keyword_match\', aask.keyword, \'version\', NULL ' 'FROM track t ' 'INNER JOIN audio_attributes_suggestion_keywords aask ' f'ON LOWER(t.version) REGEXP {concat_expr} ' 'INNER JOIN audio_attributes aa ON aa.id = aask.audio_attribute_id ' 'WHERE t.id = :tuid ' 'UNION ALL ' 'SELECT aa.id, aa.description, \'keyword_match\', aask.keyword, \'release_name\', NULL ' 'FROM track t ' 'INNER JOIN releases r ON t.release_id = r.release_id ' 'INNER JOIN audio_attributes_suggestion_keywords aask ' f'ON LOWER(r.release_name) REGEXP {concat_expr} ' 'INNER JOIN audio_attributes aa ON aa.id = aask.audio_attribute_id ' 'WHERE t.id = :tuid ' 'UNION ALL ' 'SELECT aa.id, aa.description, \'keyword_match\', aask.keyword, \'artist_name\', NULL ' 'FROM track t ' 'LEFT JOIN track_artist ta ON t.id = ta.track_id ' 'INNER JOIN audio_attributes_suggestion_keywords aask ' f'ON LOWER(ta.name) REGEXP {concat_expr} ' 'INNER JOIN audio_attributes aa ON aa.id = aask.audio_attribute_id ' 'WHERE t.id = :tuid ' 'UNION ALL ' # specific case for genre 'Spoken Word' (genre_id=16) with any non-comedy subgenre 'SELECT 6, aa.description, \'spoken_word_check\', NULL, NULL, NULL ' 'FROM track t ' 'INNER JOIN releases r ON t.release_id = r.release_id ' 'INNER JOIN release_subgenre rs ON rs.release_id = r.release_id ' 'INNER JOIN subgenre s ON rs.subgenre_id = s.orchard_id ' 'INNER JOIN audio_attributes aa ON aa.id = 6 ' 'WHERE r.genre_id = 16 AND s.name NOT LIKE \'%comedy%\' AND t.id = :tuid ' 'UNION ALL ' # specific case for sub-genres Baroque(267), Pre-baroque(285), Gregorian(701), Medieval(730), Romantic(286) 'SELECT 4, aa.description, \'classical_traditional_check\', NULL, NULL, rs.subgenre_id ' 'FROM track t ' 'INNER JOIN releases r ON t.release_id = r.release_id ' 'INNER JOIN release_subgenre rs ON rs.release_id = r.release_id ' 'INNER JOIN audio_attributes aa ON aa.id = 4 ' 'WHERE rs.subgenre_id IN (267, 285, 701, 730, 286) AND t.id = :tuid ' ) rows = session.execute(suggestion_query, {'tuid': tuid}).fetchall() return api_utils.create_get_list_response( cls._group_suggestions_with_reasons(rows)) @staticmethod def _group_suggestions_with_reasons(rows): """Group raw suggestion rows into items with deduplicated reasons. Each row is (aa_id, description, reason_type, keyword, matched_on, subgenre_id). Returns items sorted by attribute_id with track_audio_attribute_id=0. """ attr_map = {} seen = {} for aa_id, description, reason_type, keyword, matched_on, subgenre_id in rows: if aa_id not in attr_map: attr_map[aa_id] = { 'track_audio_attribute_id': 0, 'audio_attribute_id': aa_id, 'description': description, 'reasons': [], } seen[aa_id] = set() reason = {'type': reason_type} if reason_type == 'keyword_match': reason['keyword'] = keyword reason['on'] = matched_on elif reason_type == 'classical_traditional_check': reason['subgenre_id'] = subgenre_id key = tuple(reason.items()) if key not in seen[aa_id]: seen[aa_id].add(key) attr_map[aa_id]['reasons'].append(reason) return [attr_map[aa_id] for aa_id in sorted(attr_map)] @classmethod @mysql.db_session_wrap def get_track_audio_attributes_edits(cls, tuid, session): """Get a track's currently set audio attributes edits from DB. Args: tuid (int): unique id of track session (object): SQLAlchemy database session (optional) Returns: List of the track's currently set audio attributes edits, or None if tuid not found in the database """ track_audio_attributes_edits = session.query(TrackAudioAttributesEdits) \ .filter(TrackAudioAttributesEdits.unique_track_id == tuid) \ .order_by(TrackAudioAttributesEdits.audio_attribute_id) return api_utils.create_get_list_response( [{'track_audio_attribute_id': taae.id, 'audio_attribute_id': taae.audio_attribute_id, 'description': taae.audio_attributes.description } for taae in track_audio_attributes_edits])