"""Video Product Copy logic.""" from typing import Any from video.constants import job_io_fields from video.constants.error import ( ERROR_CODE_VALIDATION_ERROR, ERROR_MESSAGE_NO_RELEASE_FOUND, ) from video.exceptions import InvalidRequest, ReleaseNotFound, VideoAssetNotFound from video.logic import ( asset as asset_logic, track as logic_track, video_asset as video_asset_logic, ) from video.models.ows import track as ows_track from video.models.sql.classes import release, track as models_track, track_video def add_track_metadata(track_data: dict[str, Any]) -> dict[str, Any]: """Create track.""" # Here track_data['track_id'] is None track_response = models_track.upsert(track_data) add_track_metadata_message = { **track_data, "track_id": track_response["track_id"], } return add_track_metadata_message def copy_track_metadata( audio_product_id: Any, selected_thumbnail_id: Any, volume_data: list[Any], ) -> dict[int, dict[str, Any]]: """Copy Track Metadata.""" video_product_track_mapping = {} release_upc = release.get_upc_by_product_id(audio_product_id) for index, video_product_ids in enumerate(volume_data, start=1): volume_number = index for video_product_id in video_product_ids: tracks = ows_track.get_track(video_product_id) track_video_metadata_details = logic_track.get_tracks_metadata( video_product_id, None ) track = tracks["items"][0] track_video_details = track_video_metadata_details[0] # Extract Max Track Number product_tracks = models_track.get_tracks_by_product_id( audio_product_id, volume_number ) if product_tracks: track_number_counter = max( track["track_number"] for track in product_tracks ) else: track_number_counter = 0 # Extract Track data track_data = extract_track_data( track, track_video_details, audio_product_id, release_upc.get("upc"), track_number_counter + 1, volume_number, ) # Extract TrackVideo data track_video_data = extract_track_video_data(track_video_details) # Insert Track Details inserted_track_records = add_track_metadata(track_data) track_id = inserted_track_records["track_id"] video_product_track_mapping[int(video_product_id)] = { "track_id": track_id, "video_thumbnail_product": video_product_id == selected_thumbnail_id, } # Insert TrackVideo Details track_video_data["track_id"] = track_id track_video.upsert(track_video_data) track_number_counter += 1 if not video_product_track_mapping: raise InvalidRequest("Failed to add track metadata") return video_product_track_mapping def extract_track_data( track: dict[str, Any], track_video_details: dict[str, Any], audio_product_id: Any, audio_product_tracks_upc: str | None, track_number_counter: int, volume_number: int, ) -> dict[str, Any]: """Extract Track Metadata.""" return { "isrc": track["isrc"], "track_name": track["track_name"], "duration_minutes": track["duration_minutes"], "duration_seconds": track["duration_seconds"], "explicit": track["explicit"], "offer_type": track["offer_type"], "p_info": track["p_info"], "meta_language_code": track["meta_language_code"], "closed_caption_exists": track_video_details["closed_caption_exists"], "closed_caption_reason_id": track_video_details["closed_caption_reason_id"], "preview_start_time": track["preview_start_time"], "track_type": track["track_type"], "upc": audio_product_tracks_upc, "product_id": audio_product_id, "track_number": track_number_counter, "stereo_or_mono": "stereo", "volume_number": volume_number, } def extract_track_video_data(track_video_details: dict[str, Any]) -> dict[str, Any]: """Extract Track Video Metadata.""" return { "fps": track_video_details["fps"], "resolution": track_video_details["resolution"], "aspect_ratio": track_video_details["aspect_ratio"], "color": track_video_details["color"], "channel": track_video_details["channel"], } def format_video_assets( video_assets: list[dict[str, Any]], video_and_tracks_map: dict[int, dict[str, Any]], audio_product_id: Any, ) -> list[dict[str, Any]]: """Format Video Asset Records.""" formatted_video_assets = [] requires_track_id = [ job_io_fields.VIDEO_MASTER, job_io_fields.H264_SD, job_io_fields.H264_HD, ] video_image_asset_types = [ job_io_fields.VIDEO_IMAGE_S2, job_io_fields.VIDEO_IMAGE_S5, job_io_fields.VIDEO_IMAGE_S10, job_io_fields.VIDEO_IMAGE_S11, ] for video_asset in video_assets: asset_types_to_process = [ job_io_fields.VIDEO_MASTER, job_io_fields.H264_SD, job_io_fields.H264_HD, ] tuid = ( video_and_tracks_map[int(video_asset["product_id"])]["track_id"] if video_asset["asset_type"] in requires_track_id else 0 ) if video_and_tracks_map[int(video_asset["product_id"])][ "video_thumbnail_product" ]: asset_types_to_process.extend(video_image_asset_types) if video_asset["asset_type"] in asset_types_to_process: formatted_asset = { "product_id": int(audio_product_id), "asset_type": video_asset["asset_type"], "asset_path": video_asset["s3_key"], "tuid": tuid, "duration": video_asset["duration"], } formatted_video_assets.append(formatted_asset) return formatted_video_assets def copy_video_asset_details( video_product_track_mapping: dict[int, dict[str, Any]], audio_product_id: Any ) -> Any: """Copy Video Asset Details.""" product_ids = list(map(int, video_product_track_mapping.keys())) video_assets = asset_logic.get_product_assets(product_ids) video_asset_insert = format_video_assets( video_assets, video_product_track_mapping, audio_product_id ) return video_asset_logic.bulk_upsert_video_assets(video_asset_insert) def validate_bundle( audio_product_id: Any, bundle_type: str, volume_data: list[Any] ) -> None: """Validate Bundle Data before proceeding with insertion.""" if bundle_type == "VIDEO": product_tracks = models_track.get_tracks_by_product_id(audio_product_id) if product_tracks: error_message = ( "Delete existing tracks before proceeding with the video-only bundle." ) raise InvalidRequest(error_message, error_code=ERROR_CODE_VALIDATION_ERROR) try: release.get_upc_by_product_id(audio_product_id) except ReleaseNotFound as e: raise InvalidRequest( ERROR_MESSAGE_NO_RELEASE_FOUND, error_code=ERROR_CODE_VALIDATION_ERROR, ) from e for video_product_ids in volume_data: for video_product_id in video_product_ids: ows_track.get_track(video_product_id) logic_track.get_tracks_metadata(video_product_id, None) # Validate video assets for each video product product_ids = [ int(video_product_id) for volume in volume_data for video_product_id in volume ] try: video_assets = asset_logic.get_product_assets(product_ids) missing_product_ids = set(product_ids) - { asset["product_id"] for asset in video_assets } except VideoAssetNotFound: missing_product_ids = set(product_ids) if missing_product_ids: error_message = f"Video assets are missing for product IDs: {', '.join(map(str, missing_product_ids))}" raise VideoAssetNotFound(error_message) def create_bundle( audio_product_id: Any, bundle_type: str, selected_thumbnail_id: Any, volume_data: list[Any], ) -> int: """Create Bundle.""" validate_bundle(audio_product_id, bundle_type, volume_data) video_product_track_mapping = copy_track_metadata( audio_product_id, selected_thumbnail_id, volume_data ) copy_video_asset_details(video_product_track_mapping, audio_product_id) return int(audio_product_id)