"""Module that contains track-level asset operations.""" from typing import Any from assets.constants import ( asset_upload_types, error, product as product_constants, ) from assets.exceptions import InvalidProductStatus from assets.logic import asset_copy from assets.models import ( ows_product, ows_track, ) def copy_track_asset(from_tuid: int, to_tuid: int, full_user_id: str) -> None: """Create a copy of the track assets. Copy assets from track with unique id to track with unique id Args: from_tuid (int): source track id. to_tuid (int): destination track id. full_user_id (str): Full user id. """ source_asset_data = _prepare_source_track(from_tuid) from_product = source_asset_data["product"] destination_asset_data = _prepare_destination_track(to_tuid) destination_product = destination_asset_data["product"] if destination_product["status"] == product_constants.PRODUCT_STATUS_IN_CONTENT: raise InvalidProductStatus(error.ERROR_MESSAGE_PRODUCT_IN_CONTENT) # Copy only the audio types that exist on source track, with stereo first. for asset_type in asset_upload_types.AUDIO_UPLOAD_TYPES: asset_copy.copy_v2_assets( from_product["product_id"], destination_product["product_id"], destination_product["upc"], full_user_id, from_tuid, to_tuid, asset_type, (asset_type != asset_upload_types.STEREO), ) def _prepare_source_track(unique_track_id: int) -> dict[str, Any]: """Get file name and upc for source tracks asset. Args: unique_track_id (int): source track id. Returns: dict: file name and upc. """ track_and_product_response = _get_track_and_product_by_track_id(unique_track_id) track = track_and_product_response["track"] product = track_and_product_response["product"] return { "source_upc": track["upc"], "track": track, "product": product, } def _prepare_destination_track(unique_track_id: int) -> dict[str, Any]: """Get file name and upc for destination tracks asset. Args: unique_track_id (int): destination track id. Returns: dict: dict with file name and upc. """ track_and_product_response = _get_track_and_product_by_track_id(unique_track_id) to_track = track_and_product_response["track"] to_product = track_and_product_response["product"] return { "destination_upc": to_track["upc"], "track": to_track, "product": to_product, } def _get_track_and_product_by_track_id(unique_track_id: int) -> dict[str, Any]: """Get the track and product by track id. Args: unique_track_id: Unique track id Returns: dict: dict with track and product object. """ track = ows_track.get_track_by_id(unique_track_id) product = ows_product.get_product_by_id(track["product_id"]) return {"track": track, "product": product}