"""Model schemas for delivery metadata.""" from typing import Any from uuid import UUID from pydantic import Field, computed_field, field_validator, model_validator from delivery_metadata.clients.direct_delivery.dms_encoding_profiles import ( AudioDmsEncodingProfile, DmsEncodingProfile, ImageDmsEncodingProfile, MetadataDmsEncodingProfile, VideoDmsEncodingProfile, ) from delivery_metadata.clients.ows_track.instant_grats import InstantGrats from delivery_metadata.constants import ( AssetType, CommercialModelType, DeliveryType, DistributionContext, DistributionFeatureId, DistributionFormatId, DownloadStreamRights, Genres, InstrumentRole, LyricsExplicitness, ParticipantRole, ProductType, ProductTypeId, TerritoryStandard, TrackOfferType, TrackTypes, UseType, ) from delivery_metadata.exceptions import NoDistributionFeatures, ProductIneligible from delivery_metadata.models import Model, SortedSet from delivery_metadata.utils import vector_templated_strings as templater from delivery_metadata.utils.standardized_name import get_standardized_name class TerritoryCodeA2(Model): country_code: str = Field(..., min_length=2, max_length=2) class TerritoryCodeA3(Model): country_code: str = Field(..., min_length=3, max_length=3) class ParticipantFullnameLocalization(Model): fullname: str language_code: str localized_fullname: str | None def __hash__(self) -> int: return hash((get_standardized_name(self.fullname), self.language_code)) def __eq__(self, other: object) -> bool: if not isinstance(other, ParticipantFullnameLocalization): return NotImplemented return ( get_standardized_name(self.fullname) == get_standardized_name(other.fullname) and self.language_code == other.language_code ) class Participant(Model): fullname: str roles: set[ParticipantRole] instruments: set[InstrumentRole] label_participant_id: UUID | None = None apple_music_id: str | None = None spotify_id: str | None = None spotify_artist_key: str | None = None global_participant_id: UUID | None = None fullname_localizations: frozenset[ParticipantFullnameLocalization] | None = None def __hash__(self) -> int: return hash(get_standardized_name(self.fullname)) def __eq__(self, other: object) -> bool: if not isinstance(other, Participant): return NotImplemented return get_standardized_name(self.fullname) == get_standardized_name( other.fullname ) class TerritoryDates(Model): country_code: str country_release_date: str | None country_sale_start_date: str | None country_preorder_date: str | None class CustomPricing(Model): custom_currency_code: str custom_price: str def __eq__(self, other: Any) -> bool: if not isinstance(other, CustomPricing): return NotImplemented return ( self.custom_currency_code == other.custom_currency_code and self.custom_price == other.custom_price ) def __hash__(self) -> int: return hash((self.custom_currency_code, self.custom_price)) class ReleasePricing(Model): price_code: str country_code: TerritoryCodeA2 custom_pricing: CustomPricing | None start_date: str | None end_date: str | None class TrackReleasePricing(Model): price_code: str country_code: TerritoryCodeA2 track_ids: list[int] custom_pricing: CustomPricing | None start_date: str | None end_date: str | None def __hash__(self) -> int: return hash( ( self.price_code, self.country_code, tuple(self.track_ids), self.custom_pricing, self.start_date, self.end_date, ) ) class CopyrightLine(Model): year: str | None text: str class LocalizedMetadata(Model): language_code: str localized_name: str | None localized_version: str | None class Asset(Model): filename: str bucket: str asset_type: AssetType class ArtworkAsset(Asset): @field_validator("asset_type", mode="before") @classmethod def is_valid_asset_type(cls, value: AssetType) -> AssetType: if value not in [AssetType.TIF, AssetType.JPG]: raise ValueError(f"Invalid asset type: {value.value} for Artwork asset.") return value class VideoArtworkAsset(Asset): @field_validator("asset_type", mode="before") @classmethod def is_valid_asset_type(cls, value: AssetType) -> AssetType: if value not in [ AssetType.VIDEO_IMAGE_S2, AssetType.VIDEO_IMAGE_S5, AssetType.VIDEO_IMAGE_S10, AssetType.VIDEO_IMAGE_S11, ]: raise ValueError( f"Invalid asset type: {value.value} for Video Artwork asset." ) return value class DigitalBookletAsset(Asset): @field_validator("asset_type", mode="before") @classmethod def is_valid_asset_type(cls, value: AssetType) -> AssetType: if value != AssetType.DIGITAL_BOOKLET: raise ValueError( f"Invalid asset type: {value.value} for Digital Booklet asset." ) return value class AudioAsset(Asset): duration: float @field_validator("asset_type", mode="before") @classmethod def is_valid_asset_type(cls, value: AssetType) -> AssetType: if value not in [ AssetType.WAV, AssetType.ATMOS, AssetType.FLAC, AssetType.MP3_192, ]: raise ValueError(f"Invalid asset type: {value.value} for Audio asset.") return value class VideoAsset(Asset): duration: float @field_validator("asset_type", mode="before") @classmethod def is_valid_asset_type(cls, value: AssetType) -> AssetType: if value not in [AssetType.VIDEO_MASTER, AssetType.H264_HD, AssetType.H264_SD]: raise ValueError(f"Invalid asset type: {value.value} for Video asset.") return value class Track(Model): track_id: int volume_number: int track_number: int duration_seconds: int = Field(..., ge=0) preview_start_time_seconds: int track_type: TrackTypes lyrics_explicitness: LyricsExplicitness upc: int isrc: str p_line: CopyrightLine track_name: str track_version: str | None language_of_performance_ietf_rfc_5646_code: str | None participants: list[Participant] distribution_rights: set[DownloadStreamRights] offer_type: TrackOfferType track_territory_pricing: set[ReleasePricing] instant_grats: InstantGrats | None localized_metadata: list[LocalizedMetadata] | None assets: list[AudioAsset | VideoAsset] spatial_isrc: str | None # the name of this property is confusing. # I believe it means if the track has no rights # or should assume release level rights. @computed_field # type: ignore[prop-decorator] @property def is_track_release(self) -> bool: return bool(self.distribution_rights) @computed_field # type: ignore[prop-decorator] @property def territory_pricing_dict(self) -> dict[str, str | CustomPricing]: pricing_by_price_code = {} for price in self.track_territory_pricing: pricing_by_price_code[price.country_code.country_code] = ( price.custom_pricing if price.custom_pricing else price.price_code ) return pricing_by_price_code @computed_field # type: ignore[prop-decorator] @property def has_spatial_assets(self) -> bool: return any( asset for asset in self.assets if asset.asset_type == AssetType.ATMOS ) @computed_field # type: ignore[prop-decorator] @property def asset_duration_seconds(self) -> int: wav_asset = next( ( asset for asset in self.assets if isinstance(asset, AudioAsset) and asset.asset_type == AssetType.WAV ), None, ) return ( int(wav_asset.duration) if wav_asset and wav_asset.duration is not None else self.duration_seconds ) class GenreMapping(Model): genre: str genre_code: str | None subgenre: str | None subgenre_code: str | None subgenre_id: int class Product(Model): upc: int display_upc: str tracks: list[Track] product_id: int original_release_date: str | None release_date: str sale_start_date: str preorder_date: str | None product_name: str product_type_id: ProductTypeId delivered_version: str | None release_grid: str | None metadata_language_ietf_rfc_5646_code: str | None distribution_format_id: DistributionFormatId product_code: str | None release_sony_product_no: str | None vendor_catalog_number: str | None participants: list[Participant] genre_id: int genre_name: str subgenre_id: int subgenre_name: str | None genre_mappings: list[GenreMapping] | None timed_release_datetime: str | None staggered_release_date: str | None territory_dates: list[TerritoryDates] release_territory_pricing: set[ReleasePricing] imprint: str p_line: CopyrightLine | None c_line: CopyrightLine distribution_context: DistributionContext distribution_rights: set[DownloadStreamRights] content_id: int project_id: int vendor_id: int vendor_owner: str localized_metadata: list[LocalizedMetadata] | None assets: list[ArtworkAsset | VideoArtworkAsset | DigitalBookletAsset] @computed_field # type: ignore[prop-decorator] @property def is_product_timed_release(self) -> bool: return bool(self.timed_release_datetime or self.staggered_release_date) @computed_field # type: ignore[prop-decorator] @property def deal_term_start_date(self) -> str: start_date = self.sale_start_date if self.timed_release_datetime: start_date = self.timed_release_datetime elif self.staggered_release_date: start_date = self.staggered_release_date return start_date @computed_field # type: ignore[prop-decorator] @property def lyrics_explicitness(self) -> LyricsExplicitness: lyrics_explicitness_to_ranking = { LyricsExplicitness.EXPLICIT: 3, LyricsExplicitness.UNKNOWN: 2, LyricsExplicitness.CLEAN: 1, LyricsExplicitness.NOT_EXPLICIT: 0, } return sorted( self.tracks, key=lambda track: lyrics_explicitness_to_ranking[track.lyrics_explicitness], reverse=True, )[0].lyrics_explicitness @computed_field # type: ignore[prop-decorator] @property def album_duration_seconds(self) -> int: return sum(track.duration_seconds for track in self.tracks) @computed_field # type: ignore[prop-decorator] @property def track_count(self) -> int: return len(self.music_tracks) + len(self.video_tracks) @computed_field # type: ignore[prop-decorator] @property def product_type(self) -> ProductType: if self.track_count == 1: return ( ProductType.VIDEO_SINGLE if self.has_video_track else ProductType.MUSIC_SINGLE ) if self.has_video_track and not self.has_music_track: return ProductType.VIDEO_ALBUM if not self.has_video_track and 3 <= self.track_count <= 4: return ProductType.EP return ( ProductType.MUSIC_BUNDLE if self.has_video_track else ProductType.MUSIC_ALBUM ) @computed_field # type: ignore[prop-decorator] @property def is_classical(self) -> bool: return self.genre_id == Genres.CLASSICAL @computed_field # type: ignore[prop-decorator] @property def is_video_single(self) -> bool: return self.has_video_track and len(self.tracks) == 1 @computed_field # type: ignore[prop-decorator] @property def has_video_track(self) -> bool: return len(self.video_tracks) > 0 @computed_field # type: ignore[prop-decorator] @property def has_music_track(self) -> bool: return len(self.music_tracks) > 0 @computed_field # type: ignore[prop-decorator] @property def video_tracks(self) -> list[Track]: return [track for track in self.tracks if track.track_type == TrackTypes.VIDEO] @computed_field # type: ignore[prop-decorator] @property def music_tracks(self) -> list[Track]: return [track for track in self.tracks if track.track_type == TrackTypes.MUSIC] @computed_field # type: ignore[prop-decorator] @property def music_tracks_grouped_by_volume(self) -> dict[int, list[Track]]: return self.group_tracks_by_volume(TrackTypes.MUSIC) @computed_field # type: ignore[prop-decorator] @property def video_tracks_grouped_by_volume(self) -> dict[int, list[Track]]: return self.group_tracks_by_volume(TrackTypes.VIDEO) def group_tracks_by_volume(self, track_type: TrackTypes) -> dict[int, list[Track]]: grouped_tracks: dict[int, list[Track]] = {} for track in self.tracks: if track.track_type == track_type: grouped_tracks.setdefault(track.volume_number, []).append(track) return grouped_tracks @computed_field # type: ignore[prop-decorator] @property def first_track(self) -> Track: return (self.music_tracks + self.video_tracks)[0] @computed_field # type: ignore[prop-decorator] @property def territory_pricing_dict(self) -> dict[str, str | CustomPricing]: pricing_by_price_code = {} for price in self.release_territory_pricing: pricing_by_price_code[price.country_code.country_code] = ( price.custom_pricing if price.custom_pricing else price.price_code ) return pricing_by_price_code class ProductAudio(Product): ... class ProductVideo(Product): channel_selection: str | None description: str | None type_of_video: str keywords: str | None associated_track_isrc: str | None class ProductBundle(ProductVideo): ... class Store(Model): store_id: int sender_ddex_party_id: str sender_ddex_party_name: str recipient_ddex_party_id: str | None recipient_ddex_party_name: str audio_encoding_profiles: list[AudioDmsEncodingProfile] | None image_encoding_profiles: list[ImageDmsEncodingProfile] | None video_image_encoding_profiles: list[ImageDmsEncodingProfile] | None metadata_encoding_profiles: list[MetadataDmsEncodingProfile] video_encoding_profiles: list[VideoDmsEncodingProfile] | None product_provided_store_artists: str | None distribution_feature_ids: set[DistributionFeatureId] is_sony_gras_store: bool = False is_preorder_supported: bool = False is_feature_to_primary_artist_supported: bool = False is_video_supported: bool = False is_bundle_supported: bool = False is_pricing_supported: bool = False territory_standard: TerritoryStandard = TerritoryStandard.ORCHARD_2016 various_artists_limit: int = 5 ddex_commercial_model_to_use_type_map_extras: dict[ DownloadStreamRights, dict[CommercialModelType, set[UseType]] ] = {} is_instant_gratification_supported: bool = False is_localization_supported: bool is_user_defined_contributors_supported: bool @model_validator(mode="after") def validate_distribution_feature_ids(self): # type: ignore[no-untyped-def] if not self.distribution_feature_ids: raise NoDistributionFeatures("Distribution features not found.") return self @computed_field # type: ignore[prop-decorator] @property def ddex_commercial_model_to_use_type_map( self, ) -> dict[DownloadStreamRights, dict[CommercialModelType, set[UseType]]]: commercial_model_type_to_use_types_map: dict[ DownloadStreamRights, dict[CommercialModelType, set[UseType]] ] = { DownloadStreamRights.DOWNLOAD: {}, DownloadStreamRights.STREAM: {}, } def add_commercial_model_type_and_use_types( *, distribution_rights: DownloadStreamRights, commercial_model_type: CommercialModelType, use_types: set[UseType], ) -> None: use_type_extras = set() if ( distribution_rights in self.ddex_commercial_model_to_use_type_map_extras and commercial_model_type in self.ddex_commercial_model_to_use_type_map_extras[ distribution_rights ] ): use_type_extras = self.ddex_commercial_model_to_use_type_map_extras[ distribution_rights ][commercial_model_type] commercial_model_type_to_use_types_map[distribution_rights].setdefault( commercial_model_type, SortedSet(), ).update(use_types.union(use_type_extras)) for distribution_feature_id in sorted( self.distribution_feature_ids, key=lambda x: x.value, ): match distribution_feature_id: case ( DistributionFeatureId.A_LA_CARTE_DOWNLOAD | DistributionFeatureId.OTA_INCLUDING_DUAL_DELIVERY ): add_commercial_model_type_and_use_types( distribution_rights=DownloadStreamRights.DOWNLOAD, commercial_model_type=CommercialModelType.PAY, use_types={UseType.PERMANENT_DOWNLOAD}, ) case ( DistributionFeatureId.SUBSCRIPTION_DOWNLOAD | DistributionFeatureId.SUBSCRIPTION_PORTABLE_TETHERED ): add_commercial_model_type_and_use_types( distribution_rights=DownloadStreamRights.DOWNLOAD, commercial_model_type=CommercialModelType.SUBSCRIPTION, use_types={UseType.CONDITIONAL_DOWNLOAD}, ) case DistributionFeatureId.SUBSCRIPTION_STREAMING: add_commercial_model_type_and_use_types( distribution_rights=DownloadStreamRights.STREAM, commercial_model_type=CommercialModelType.SUBSCRIPTION, use_types={UseType.STREAM}, ) case DistributionFeatureId.OTA_WITH_RINGTONE_USE: add_commercial_model_type_and_use_types( distribution_rights=DownloadStreamRights.DOWNLOAD, commercial_model_type=CommercialModelType.PAY, use_types={UseType.PERMANENT_DOWNLOAD, UseType.RINGTONE}, ) case DistributionFeatureId.STREAMING_SUBSCRIPTION_TO_MOBILE: add_commercial_model_type_and_use_types( distribution_rights=DownloadStreamRights.STREAM, commercial_model_type=CommercialModelType.SUBSCRIPTION, use_types={UseType.NON_INTERACTIVE}, ) case DistributionFeatureId.STREAMING_ON_DEMAND_TO_MOBILE: add_commercial_model_type_and_use_types( distribution_rights=DownloadStreamRights.STREAM, commercial_model_type=CommercialModelType.SUBSCRIPTION, use_types={UseType.ON_DEMAND}, ) case DistributionFeatureId.RINGTONES: add_commercial_model_type_and_use_types( distribution_rights=DownloadStreamRights.DOWNLOAD, commercial_model_type=CommercialModelType.PAY, use_types={UseType.RINGTONE}, ) case DistributionFeatureId.RINGBACK_TONES: add_commercial_model_type_and_use_types( distribution_rights=DownloadStreamRights.DOWNLOAD, commercial_model_type=CommercialModelType.PAY, use_types={UseType.RINGBACK}, ) case DistributionFeatureId.AD_SUPPORTED_STREAMING: add_commercial_model_type_and_use_types( distribution_rights=DownloadStreamRights.STREAM, commercial_model_type=CommercialModelType.AD_SUPPORTED, use_types={UseType.STREAM}, ) return commercial_model_type_to_use_types_map @computed_field # type: ignore[prop-decorator] @property def is_download_supported(self) -> bool: return bool( self.distribution_feature_ids.intersection( [ DistributionFeatureId.A_LA_CARTE_DOWNLOAD, DistributionFeatureId.OTA_INCLUDING_DUAL_DELIVERY, DistributionFeatureId.SUBSCRIPTION_DOWNLOAD, DistributionFeatureId.SUBSCRIPTION_PORTABLE_TETHERED, DistributionFeatureId.OTA_WITH_RINGTONE_USE, DistributionFeatureId.RINGTONES, DistributionFeatureId.RINGBACK_TONES, ] ) ) @computed_field # type: ignore[prop-decorator] @property def is_streaming_supported(self) -> bool: return bool( self.distribution_feature_ids.intersection( [ DistributionFeatureId.SUBSCRIPTION_STREAMING, DistributionFeatureId.STREAMING_SUBSCRIPTION_TO_MOBILE, DistributionFeatureId.STREAMING_ON_DEMAND_TO_MOBILE, DistributionFeatureId.AD_SUPPORTED_STREAMING, ] ) ) @computed_field # type: ignore[prop-decorator] @property def has_atmos_profile(self) -> bool: return self.audio_encoding_profiles is not None and any( encoding_profile for encoding_profile in self.audio_encoding_profiles if encoding_profile.profile_subtype == "atmos" ) class Delivery(Model): delivery_type: DeliveryType store: Store allowed_territories: list[TerritoryCodeA2] @model_validator(mode="after") def validate_model(self): # type: ignore[no-untyped-def] if ( self.delivery_type != DeliveryType.METADATA_UPDATE and not self.allowed_territories ): raise ProductIneligible( f"allowed_territories cannot be empty for delivery_type: {self.delivery_type.value}" ) return self @computed_field # type: ignore[prop-decorator] @property def is_metadata_update(self) -> bool: return self.delivery_type == DeliveryType.METADATA_UPDATE class TechnicalFileDetail(Model): profile_subtype: str format: str filename: str delivery_location: str hash_algorithm: str placeholder_hash: str | None @computed_field # type: ignore[prop-decorator] @property def full_file_path_and_name(self) -> tuple[str, str]: parts = "/".join( [self.delivery_location.strip("/"), self.filename.strip("/")] ).split("/") return "/".join(parts[:-1]), parts[-1] class ImageTechnicalFileDetail(TechnicalFileDetail): image_width: int | None image_height: int | None class AudioTechnicalFileDetail(TechnicalFileDetail): audio_number_of_channels: int | None audio_clip_length: int | None audio_bitrate: int | None audio_sampling_rate: int | None # frequency audio_bits_per_sample: int | None # bit depth class VideoTechnicalFileDetail(TechnicalFileDetail): ... class DeliveryMetadata(Model): product: Product delivery: Delivery @computed_field # type: ignore[prop-decorator] @property def metadata_file_details(self) -> list[TechnicalFileDetail]: return [ self._transform_encoding_profile( profile, self._get_placeholder_replacements(), ) for profile in self.delivery.store.metadata_encoding_profiles ] @computed_field # type: ignore[prop-decorator] @property def image_file_details(self) -> list[ImageTechnicalFileDetail]: if not self.delivery.store.image_encoding_profiles: return [] return [ self._transform_image_encoding_profile( profile, self._get_placeholder_replacements(), ) for profile in self.delivery.store.image_encoding_profiles ] @computed_field # type: ignore[prop-decorator] @property def video_image_file_details( self, ) -> dict[int, list[ImageTechnicalFileDetail]] | None: if ( not self.product.video_tracks or not self.delivery.store.video_image_encoding_profiles ): return None profiles: dict[int, list[ImageTechnicalFileDetail]] = {} for track in self.product.video_tracks: profiles[track.track_id] = [ self._transform_image_encoding_profile( profile, self._get_placeholder_replacements(track), ) for profile in self.delivery.store.video_image_encoding_profiles ] return profiles @computed_field # type: ignore[prop-decorator] @property def audio_file_details(self) -> dict[int, list[AudioTechnicalFileDetail]] | None: if ( not self.product.music_tracks or not self.delivery.store.audio_encoding_profiles ): return None profiles: dict[int, list[AudioTechnicalFileDetail]] = {} for track in self.product.music_tracks: replacements = self._get_placeholder_replacements(track) profiles[track.track_id] = [ self._transform_audio_encoding_profile(profile, replacements) for profile in self.delivery.store.audio_encoding_profiles ] return profiles @computed_field # type: ignore[prop-decorator] @property def video_file_details(self) -> dict[int, list[VideoTechnicalFileDetail]] | None: if ( not self.product.video_tracks or not self.delivery.store.video_encoding_profiles ): return None profiles: dict[int, list[VideoTechnicalFileDetail]] = {} for track in self.product.video_tracks: replacements = self._get_placeholder_replacements(track) profiles[track.track_id] = [ self._transform_video_encoding_profile(profile, replacements) for profile in self.delivery.store.video_encoding_profiles ] return profiles def _get_placeholder_replacements( self, track: Track | None = None ) -> dict[str, str | int]: replacement_variables: dict[str, str | int] = { "upc": self.product.upc, "display_upc": self.product.display_upc, "content_id": self.product.content_id, "vendor_id": self.product.vendor_id, } if track: replacement_variables.update( { "cd": track.volume_number, "track_id": track.track_number, "isrc": track.isrc, } ) return replacement_variables def _transform_encoding_profile( self, profile: DmsEncodingProfile, replacements: dict[str, str | int] ) -> TechnicalFileDetail: return TechnicalFileDetail( profile_subtype=profile.profile_subtype, format=profile.format, hash_algorithm=profile.hashcode_algorithm, filename=self._parse_filename(profile.filename_template, replacements), delivery_location=self._parse_delivery_location( profile.delivery_location_template, replacements ), placeholder_hash=( templater.process_string( original_string=profile.placeholder_hashcode_template, variables=replacements, ) if profile.placeholder_hashcode_template else None ), ) def _transform_image_encoding_profile( self, profile: ImageDmsEncodingProfile, replacements: dict[str, str | int] ) -> ImageTechnicalFileDetail: technical_detail = self._transform_encoding_profile(profile, replacements) return ImageTechnicalFileDetail( **technical_detail.__dict__, image_width=profile.width, image_height=profile.height, ) def _transform_audio_encoding_profile( self, profile: AudioDmsEncodingProfile, replacements: dict[str, str | int] ) -> AudioTechnicalFileDetail: technical_detail = self._transform_encoding_profile(profile, replacements) return AudioTechnicalFileDetail( **technical_detail.__dict__, audio_number_of_channels=profile.number_of_audio_channels, audio_clip_length=profile.clip_length, audio_bitrate=profile.bitrate, audio_sampling_rate=profile.frequency, audio_bits_per_sample=profile.bits_per_sample, ) def _transform_video_encoding_profile( self, profile: VideoDmsEncodingProfile, replacements: dict[str, str | int] ) -> VideoTechnicalFileDetail: technical_detail = self._transform_encoding_profile(profile, replacements) return VideoTechnicalFileDetail(**technical_detail.__dict__) @computed_field # type: ignore[prop-decorator] @property def territory_codes_with_start_date(self) -> dict[str, str]: start_date = self.product.deal_term_start_date timed_release_product = self.product.is_product_timed_release territories_with_start_date = { allowed_country.country_code: start_date for allowed_country in self.delivery.allowed_territories } if timed_release_product: return territories_with_start_date for territory_date in self.product.territory_dates: if ( territory_date.country_code in territories_with_start_date ) and territory_date.country_sale_start_date: territories_with_start_date.update( { territory_date.country_code: territory_date.country_sale_start_date } ) return territories_with_start_date def _parse_delivery_location( self, delivery_location_template: str, replacements: dict[str, str | int] ) -> str: return templater.process_string( original_string=delivery_location_template, variables=replacements, skip_timestamp=True, ) def _parse_filename( self, filename_template: str, replacements: dict[str, str | int] ) -> str: return templater.process_string( original_string=filename_template, variables=replacements, ) class LabelParticipantData(Model): label_participant_id: UUID | None apple_music_id: str | None spotify_id: str | None spotify_artist_key: str | None global_participant_id: UUID | None class LabelParticipantMap(Model): map: dict[str, LabelParticipantData]