"""Generate Release nodes for v38.""" from typing import List from soundrecording_utils.constants.ddex.constants import DEFAULT_TERRITORY_CODE, EXPLICIT_STATUS_LOOKUP_TABLE from soundrecording_utils.ddex.utils.control_chars import remove_control_chars from soundrecording_utils.ddex.utils.rules import get_fingerprint_rules_by_track_id, get_territory_claims_by_rules from soundrecording_utils.metadata.types import OrchardSoundRecording, Track from .e import E from .sound_recording import generate_reference_title from .track import generate_display_artist_name, generate_p_line, generate_track_display_artists def _generate_release_details_by_territory( territories: List[str], track_metadata: Track, ): return E( 'ReleaseDetailsByTerritory', *(E('TerritoryCode', territory) for territory in territories), generate_display_artist_name(track_metadata.participations), E( 'LabelName', track_metadata.product.label.name, ), E( 'Title', { 'TitleType': 'DisplayTitle', }, E('TitleText', remove_control_chars(track_metadata.name)), ), *generate_track_display_artists(track_metadata), E('ParentalWarningType', EXPLICIT_STATUS_LOOKUP_TABLE[track_metadata.explicit]), E('Genre', E('GenreText', track_metadata.product.subgenre.genre.name)), ) def _generate_all_release_details_by_territories( sound_recording_metadata: OrchardSoundRecording, primary_track_metadata: Track, service: str, takedown: bool = False, ): # Extract and format rules on each level for each track if not takedown if not takedown: track_and_rules_by_tuid = get_fingerprint_rules_by_track_id( sound_recording_metadata.track_connection.tracks ) # Extract territory claims for each track tracks_with_claims = get_territory_claims_by_rules(track_and_rules_by_tuid, service) # Find if any territory claims / block otherwise handle takedown claimed_or_blocked_tracks = False for _, v in tracks_with_claims.items(): if len(v['claimed_rules']) or len(v['blocked_rules']): claimed_or_blocked_tracks = True break if claimed_or_blocked_tracks: sr_details = [] territories_to_use: List[str] = [] for _, track_id in enumerate(tracks_with_claims): if len(tracks_with_claims[track_id]['claimed_rules']): territories_to_use += [ r['territory'] for r in tracks_with_claims[track_id]['claimed_rules'] ] if len(tracks_with_claims[track_id]['blocked_rules']): territories_to_use += [ r['territory'] for r in tracks_with_claims[track_id]['blocked_rules'] ] # de-dup and sort territories territories_to_use = list(set(territories_to_use)) territories_to_use.sort() if territories_to_use: sr_details.append( _generate_release_details_by_territory( territories_to_use, primary_track_metadata, ) ) return sr_details # Handle 'takedown' return [ _generate_release_details_by_territory( [DEFAULT_TERRITORY_CODE], primary_track_metadata, ) ] def generate_release( release_reference: str, resource_reference: str, sound_recording_metadata: OrchardSoundRecording, primary_track_metadata: Track, service: str, takedown: bool = False, ): """Generate a Release node.""" details_by_territory = _generate_all_release_details_by_territories( sound_recording_metadata, primary_track_metadata, service, takedown=takedown ) return E( 'Release', E( 'ReleaseId', E('ISRC', sound_recording_metadata.isrc), ), E('ReleaseReference', release_reference), generate_reference_title(primary_track_metadata), E( 'ReleaseResourceReferenceList', E( 'ReleaseResourceReference', { 'ReleaseResourceType': 'PrimaryResource', }, resource_reference, ), ), E('ReleaseType', 'SingleResourceRelease'), *details_by_territory, generate_p_line(primary_track_metadata.p_info), )