import json import logging import time from typing import Dict, Any, Optional, List from datetime import datetime, timedelta, timezone import boto3 from error_models import ( build_alexa_error_response, build_media_error_response, SkillError, AuthorizationError, ContentNotFoundError, ContentFilteredError, InvalidRequestError, RateLimitError, GeoRestrictionError, ) from suno import ( get_clip_details, get_item_in_queue, wait_for_clip_events, handle_generate_content_action, search_suno_library, update_catalog, notify_skill_account_linked, generate_art_sources, get_art_url_for_item, get_fallback_image_url, SUNO_IMAGE_CDN, SUNO_AUDIO_ENDPOINT, log_playback_event, get_title_and_tags_from_sse, ) # ============================================================================= # KV Store Helpers for Repeat Status # ============================================================================= KV_STORE_LAMBDA_ARN = "arn:aws:lambda:us-east-2:734185074900:function:redis-clip-accessor" lambda_client = boto3.client("lambda", region_name="us-east-2") def set_repeat_status_in_kv(queue_id: str, status: str): """Saves the repeat status for a given queue_id in the KV store.""" if not queue_id: logger.warning("Attempted to set repeat status with no queue_id.") return key = f"alexa-repeat-status:{queue_id}" payload = {"key": key, "value": status} try: lambda_client.invoke( FunctionName=KV_STORE_LAMBDA_ARN, InvocationType="Event", # Fire and forget Payload=json.dumps(payload), ) logger.info(f"Persisted repeat status for queue {queue_id}: {status}") except Exception as e: logger.error(f"Failed to invoke KV store to set repeat status for queue {queue_id}: {e}") def get_repeat_status_from_kv(queue_id: str) -> str: """Retrieves the repeat status for a given queue_id from the KV store.""" if not queue_id: logger.warning("Attempted to get repeat status with no queue_id.") return "OFF" key = f"alexa-repeat-status:{queue_id}" payload = {"key": key} try: response = lambda_client.invoke( FunctionName=KV_STORE_LAMBDA_ARN, InvocationType="RequestResponse", # We need the result Payload=json.dumps(payload), ) response_payload = json.loads(response["Payload"].read()) response_body = json.loads(response_payload.get("body", "{}")) status = response_body.get("value", "OFF") # Ensure status is either "ON" or "OFF" if status not in ["ON", "OFF"]: logger.warning(f"Invalid repeat status '{status}' retrieved from KV store for queue {queue_id}. Defaulting to OFF.") return "OFF" logger.info(f"Retrieved repeat status for queue {queue_id}: {status}") return status except Exception as e: logger.error(f"Failed to invoke KV store to get repeat status for queue {queue_id}: {e}") return "OFF" # Default to OFF on error MIN_GENERATE_CONTENT_EXECUTION_TIME_SECONDS = 3.5 # Global variable to track actual processing time before artificial delays _real_processing_end_time = None # ============================================================================= # Supported Attribute Allowlist Configuration # ============================================================================= # This allowlist defines which attribute types the Suno skill supports. # When new attributes are added to Alexa v3.0 (like "LIBRARY"), they will be # ignored until explicitly added to this list and implemented in the skill. # # To add support for a new attribute: # 1. Add the attribute type to SUPPORTED_ATTRIBUTE_TYPES # 2. Update handle_attribute_query() to process the new attribute # 3. (Optional) Update skill manifest to declare the new capability # # Current supported attributes: # - GENRE: Used for library filtering by music genre # - MEDIA_TYPE: Only supports "TRACK" value for song playback # - TRACK: Specific track entities with resolvedEntities # ============================================================================= # Supported attribute types - update this list as Suno adds more capabilities SUPPORTED_ATTRIBUTE_TYPES = ["GENRE", "MEDIA_TYPE", "TRACK"] class MessageIdFormatter(logging.Formatter): """Formatter that prefixes log messages with a request-specific message_id.""" def __init__(self, fmt=None, datefmt=None, style="%"): super().__init__(fmt, datefmt, style) self.message_id = None def format(self, record): if hasattr(self, "message_id") and self.message_id: prefix = f"[{self.message_id}] " if not hasattr(record, "message_id_added"): record.msg = prefix + str(record.msg) record.message_id_added = True return super().format(record) _default_formatter = logging.Formatter("%(levelname)s %(message)s") _message_id_formatter = MessageIdFormatter("%(levelname)s %(message)s") logger = logging.getLogger() logger.setLevel(logging.INFO) def setup_log_handlers(): if not logger.handlers: handler = logging.StreamHandler() handler.setFormatter(_default_formatter) logger.addHandler(handler) setup_log_handlers() def set_message_id_for_logs(message_id): """Set the message_id for all logs until it's reset""" if message_id: _message_id_formatter.message_id = message_id for handler in logger.handlers: handler.setFormatter(_message_id_formatter) def reset_message_id_for_logs(): """Reset the log formatter to default""" _message_id_formatter.message_id = None for handler in logger.handlers: handler.setFormatter(_default_formatter) def lambda_handler(event: Dict[Any, Any], context: Any) -> Dict[Any, Any]: global _real_processing_end_time handler_start_time = time.time() _real_processing_end_time = None # Reset for this invocation reset_message_id_for_logs() response = None logger.info(f"Request object: {json.dumps(event, default=str)}") directive_namespace = "UnknownNamespace" directive_name = "UnknownDirective" request_message_id = None try: if ( isinstance(event, dict) and "header" in event and isinstance(event["header"], dict) ): request_message_id = event["header"].get("messageId") if request_message_id: set_message_id_for_logs(request_message_id) match event: case { "header": {"namespace": "Alexa.Audio", "name": "SyncUserContent"}, "payload": payload, }: directive_namespace = "Alexa.Audio" directive_name = "SyncUserContent" user_id = payload.get("user", {}).get("id") access_token = payload.get("accessToken") message_id = event.get("header", {}).get("messageId") correlation_token = event.get("header", {}).get("correlationToken") previous_sync_token = payload.get("previousSyncToken") # Log the previous sync token if present if previous_sync_token: logger.info(f"Processing SyncUserContent with previousSyncToken: {previous_sync_token}") # Get sync status from backend sync_result = notify_skill_account_linked(access_token, user_id, previous_sync_token) # Log the detailed sync result for debugging if isinstance(sync_result, dict): logger.info(f"Sync response details: needs_full_update={sync_result.get('needs_full_update', False)}, has_no_content={sync_result.get('has_no_content', False)}") # If sync_result is False, it indicates an error if sync_result is False: logger.warning("Failed to process SyncUserContent event") response = build_alexa_error_response( message_id, "INTERNAL_ERROR", "Failed to process account sync request" ) else: # Determine sync status based on response type and content sync_status = "NO_UPDATE" # Default status # Check dictionary response for special conditions if isinstance(sync_result, dict): if sync_result.get("has_no_content", False): sync_status = "NO_CONTENT" elif sync_result.get("needs_full_update", False): sync_status = "ASYNC_UPDATE" # Construct proper Alexa.Audio.SyncUserContent response response = { "header": { "messageId": message_id, "name": "SyncUserContent", "namespace": "Alexa.Audio", "interfaceVersion": "1.0" }, "payload": { "syncStatus": sync_status } } # Add correlationToken if present in the request if correlation_token: response["header"]["correlationToken"] = correlation_token logger.info(f"SyncUserContent completed with status: {sync_status}") case ( { "request": {"type": "AlexaSkillEvent.SkillAccountLinked"} } as event_data ): directive_namespace = "Alexa" directive_name = "SkillAccountLinked" user_id = ( event_data.get("context", {}) .get("System", {}) .get("user", {}) .get("userId") ) access_token = ( event_data.get("request", {}).get("body", {}).get("accessToken") ) or ( event_data.get("context", {}) .get("System", {}) .get("user", {}) .get("accessToken") ) # Check for previousSyncToken in the event data previous_sync_token = event_data.get("request", {}).get("body", {}).get("previousSyncToken") success = notify_skill_account_linked(access_token, user_id, previous_sync_token) if success: response = { "statusCode": 200, "body": json.dumps( {"message": "Account linked successfully"} ), } else: logger.warning("Failed to process SkillAccountLinked event") response = { "statusCode": 500, "body": json.dumps( {"message": "Failed to process account linking"} ), } # Handle Alexa Audio PlayQueue playback events case { "request": {"type": "AlexaAudioPlayQueueEvent.ItemPlaybackStarted"} } as event_data: directive_namespace = "AlexaAudioPlayQueueEvent" directive_name = "ItemPlaybackStarted" response = handle_playback_event(event_data, "started") case { "request": {"type": "AlexaAudioPlayQueueEvent.ItemPlaybackFailed"} } as event_data: directive_namespace = "AlexaAudioPlayQueueEvent" directive_name = "ItemPlaybackFailed" response = handle_playback_event(event_data, "failed") case { "request": {"type": "AlexaAudioPlayQueueEvent.ItemPlaybackFinished"} } as event_data: directive_namespace = "AlexaAudioPlayQueueEvent" directive_name = "ItemPlaybackFinished" response = handle_playback_event(event_data, "finished") case { "request": {"type": "AlexaAudioPlayQueueEvent.ItemPlaybackStopped"} } as event_data: directive_namespace = "AlexaAudioPlayQueueEvent" directive_name = "ItemPlaybackStopped" response = handle_playback_event(event_data, "stopped") case {"body": body} if isinstance(body, str): try: alexa_request_data = json.loads(body) if ( isinstance(alexa_request_data, dict) and "header" in alexa_request_data ): header = alexa_request_data.get("header", {}) request_message_id = header.get("messageId") if request_message_id: set_message_id_for_logs(request_message_id) response = process_alexa_directive( alexa_request_data, handler_start_time ) if ( isinstance(alexa_request_data, dict) and "header" in alexa_request_data ): header = alexa_request_data.get("header", {}) directive_namespace = header.get( "namespace", "UnknownNamespace" ) directive_name = header.get("name", "UnknownDirective") payload_version = header.get("payloadVersion", "UnknownVersion") request_message_id = header.get("messageId") except json.JSONDecodeError as json_err: logger.error(f"Failed to parse event body JSON: {json_err}") response = build_alexa_error_response( None, "INVALID_DIRECTIVE", f"Request body is not valid JSON: {json_err}", ) case {"header": header, "payload": payload} if isinstance(header, dict): directive_namespace = header.get("namespace", "UnknownNamespace") directive_name = header.get("name", "UnknownDirective") payload_version = header.get("payloadVersion", "UnknownVersion") request_message_id = header.get("messageId") if request_message_id: set_message_id_for_logs(request_message_id) response = process_alexa_directive(event, handler_start_time) case _: logger.warning(f"Unrecognized event structure {json.dumps(event, default=str, indent=2)}") response = build_alexa_error_response( None, "INVALID_DIRECTIVE", "Unrecognized request format. Could not find header/payload.", ) except ContentNotFoundError as e: logger.warning(f"Content not found error: {e.message}") response = build_media_error_response( request_message_id, e.error_code or "CONTENT_NOT_FOUND", e.message, subtypes=e.subtypes, ) except ContentFilteredError as e: logger.warning(f"Content filtered error: {e.message}") response = build_media_error_response( request_message_id, "CONTENT_FILTERED", e.message, subtypes=e.subtypes, ) except GeoRestrictionError as e: logger.warning(f"Geo restriction error: {e.message}") response = build_media_error_response( request_message_id, "GEOGRAPHICAL_RESTRICTION_ERROR", e.message ) except AuthorizationError as e: logger.warning(f"Authorization error: {e.message}") response = build_alexa_error_response( request_message_id, e.error_code or "INVALID_AUTHORIZATION_CREDENTIAL", e.message, ) except RateLimitError as e: logger.warning(f"Rate limit error: {e.message}") response = build_alexa_error_response( request_message_id, "RATE_LIMIT_EXCEEDED", e.message ) except SkillError as e: logger.error(f"General skill error: {e.message}", exc_info=True) response = build_alexa_error_response( request_message_id, e.error_code or "INTERNAL_ERROR", e.message ) except Exception as e: logger.error(f"Unhandled top-level exception in {directive_namespace}.{directive_name}: {str(e)}", exc_info=True) response = build_alexa_error_response( request_message_id, "INTERNAL_ERROR", f"An unexpected internal error occurred: {str(e)}", ) finally: if response is None: logger.error("Reached end of handler without response/exception") response = build_alexa_error_response( request_message_id, "INTERNAL_ERROR", "Handler failed to generate a response or raise an exception.", ) # Calculate real processing time, excluding artificial delays if _real_processing_end_time is None: # If never set, use current time _real_processing_end_time = time.time() real_duration_ms = (_real_processing_end_time - handler_start_time) * 1000 full_duration_ms = (time.time() - handler_start_time) * 1000 is_error_resp = False response_name = "UnknownResponse" if isinstance(response, dict) and "header" in response: final_response_header = response.get("header", {}) response_name = final_response_header.get("name", "UnknownResponse") is_error_resp = "ErrorResponse" in response_name logger.info( json.dumps( { "directive": f"{directive_namespace}.{directive_name}", "handler_duration_ms": int(real_duration_ms), "full_duration_ms": int(full_duration_ms), "status": "error" if is_error_resp else "success", } ) ) logger.info(f"Response object: {json.dumps(response, default=str)}") reset_message_id_for_logs() return response def handle_playback_event(event_data: Dict[Any, Any], event_type: str) -> Dict[Any, Any]: """ Handle Alexa Audio PlayQueue playback events (Started, Failed, Finished, Stopped). Logs the event to the Suno backend and returns a simple 200 OK response. """ # Extract information from the event request = event_data.get("request", {}) body = request.get("body", {}) item = body.get("item", {}) # Extract required fields content_id = item.get("contentId") queue_id = item.get("queueId") offset_ms = body.get("offsetInMilliseconds", 0) request_id = request.get("requestId") # Get access token from context user = event_data.get("context", {}).get("System", {}).get("user", {}) access_token = user.get("accessToken") user_id = user.get("userId") # Extract error info if applicable error_type = None error_message = None if event_type == "failed" and body.get("error"): error = body.get("error", {}) error_type = error.get("type") error_message = error.get("message") # Extract cause info for stopped events cause_type = None if event_type == "stopped" and body.get("cause"): cause = body.get("cause") cause_type = cause.get("type") # Extract playback attributes if present playback_attributes = body.get("playbackAttributes") # Log the event to Suno backend log_playback_event( access_token=access_token, user_id=user_id, content_id=content_id, queue_id=queue_id, event_type=event_type, offset_ms=offset_ms, error_type=error_type, error_message=error_message, request_id=request_id, cause_type=cause_type, playback_attributes=playback_attributes ) # Return a simple empty 200 response as Alexa ignores the payload return {"statusCode": 200} def process_alexa_directive( event: Dict[Any, Any], handler_start_time: float ) -> Dict[Any, Any]: """ Process an Alexa directive and route to the appropriate interface handler. """ header = event.get("header", {}) payload = event.get("payload", {}) namespace = header.get("namespace", "UnknownNamespace") directive_name = header.get("name", "UnknownDirective") payload_version = header.get("payloadVersion", "UnknownVersion") directive_message_id = header.get("messageId") match (namespace, directive_name, payload_version): case ("Alexa.Audio.PlayQueue", directive, "1.0") if directive in [ "GetNextItem", "GetPreviousItem", ]: return handle_alexa_audio_playqueue( directive, directive_message_id, payload, handler_start_time ) case ("Alexa.Media.Search", directive, "3.0") if directive in [ "GetPlayableContent", "GetDisplayableContent", ]: return handle_alexa_media_search( directive, directive_message_id, header, payload, handler_start_time ) case ("Alexa.Media.Playback", "Initiate", _): return handle_alexa_media_playback( directive_message_id, header, payload, payload_version ) case ("Alexa.Media.PlayQueue", directive, "1.0") if directive in [ "SetLoop", "SetRepeat", "SetPlaybackContinuation", ]: return handle_alexa_media_playqueue( directive, directive_message_id, payload ) case _: msg = f"Unsupported directive or version: {namespace}.{directive_name} v{payload_version}" logger.warning(msg) return build_alexa_error_response( directive_message_id, "INVALID_DIRECTIVE", msg ) def handle_alexa_audio_playqueue( directive_name: str, directive_message_id: str, payload: Dict[Any, Any], handler_start_time: float, ) -> Dict[Any, Any]: """ Handle Alexa.Audio.PlayQueue directives (GetNextItem, GetPreviousItem). These directives are used to navigate through a queue of audio content. """ if directive_message_id: set_message_id_for_logs(directive_message_id) current_item_reference = payload.get("currentItemReference") current_item_id = current_item_reference.get("id") queue_id = current_item_reference.get("queueId") try: # Get the persisted repeat status from KV store repeat_status = get_repeat_status_from_kv(queue_id) is_next_item = directive_name == "GetNextItem" access_token = None request_context = payload.get("requestContext") if request_context and isinstance(request_context, dict): user = request_context.get("user") if user and isinstance(user, dict): access_token = user.get("accessToken") # Check for repeat song logic if GetNextItem is system-initiated is_user_initiated = payload.get("isUserInitiated", True) if directive_name == "GetNextItem" and not is_user_initiated: if repeat_status == "ON": logger.info(f"Repeating track {current_item_id} due to system-initiated GetNextItem with repeat ON.") # Get details of the current clip to rebuild the item clip_details = get_clip_details(access_token, current_item_id) # When repeating, next/prev should still be enabled play_queue_item = build_play_queue_item( clip_id=current_item_id, directive_message_id=directive_message_id, clip_details=clip_details, access_token=access_token, enable_next=True, enable_previous=True, repeat_status=repeat_status, ) return build_play_queue_item_response( directive_message_id, directive_name, play_queue_item, is_queue_finished=False, ) # Request 2 items in the queue to determine if there are more items in that direction api_response = get_item_in_queue( access_token, current_item_id, handler_start_time, is_next_item, limit=2, # Request 2 items to determine if more are available return_list=True # Get the full list to check if there are more items ) if api_response is None: return build_play_queue_item_response( directive_message_id, directive_name, None, True, # isQueueFinished = True ) if not api_response: return build_play_queue_item_response( directive_message_id, directive_name, None, True, # isQueueFinished = True ) # Check if we got a valid list response if isinstance(api_response, list): # Store whether there are more items in this direction has_more_items = len(api_response) > 1 # Extract the first item for playback if len(api_response) > 0: api_response = api_response[0] # Extract the first item from the list else: # Return empty queue finished response when we get an empty list return build_play_queue_item_response( directive_message_id, directive_name, None, True, # isQueueFinished = True ) # Make sure we have an api_response before trying to access its fields if not api_response: return build_play_queue_item_response( directive_message_id, directive_name, None, True, # isQueueFinished = True ) content_id = api_response["id"] # Set navigation logic for prev/next buttons if is_next_item: # If we're going forward in the queue (GetNextItem) # - Enable Next if there are more items ahead # - Always enable Previous (since we've moved forward, we can go back) enable_next = has_more_items enable_previous = True else: # If we're going backward in the queue (GetPreviousItem) # - Always enable Next (since we've moved backward, we can go forward) # - Enable Previous if there are more items behind enable_next = True enable_previous = has_more_items play_queue_item = build_play_queue_item( content_id, directive_message_id, api_response, access_token=access_token, enable_next=enable_next, enable_previous=enable_previous, repeat_status=repeat_status, ) # Now just return the play queue item response directly return build_play_queue_item_response( directive_message_id, directive_name, play_queue_item, False, # isQueueFinished = False ) except Exception as e: # Don't wrap specific error types - let them propagate to the top-level handler # which knows how to properly map them to appropriate Alexa error types logger.warning(f"Error in {directive_name} request, propagating: {type(e).__name__}: {str(e)}") raise # Re-raise to be caught by the top-level handler def handle_alexa_media_search( directive_name: str, directive_message_id: str, header: Dict[Any, Any], payload: Dict[Any, Any], handler_start_time: float, ) -> Dict[Any, Any]: """ Handle Alexa.Media.Search directives (GetPlayableContent, GetDisplayableContent). These directives are used to search and retrieve media content based on user queries. """ if directive_message_id: set_message_id_for_logs(directive_message_id) # Create a complete request object with header and payload media_search_request = {"header": header, "payload": payload} # Process with our handler function that handles both GPC and GDC return process_media_search(media_search_request, handler_start_time) def handle_alexa_media_playback( directive_message_id: str, header: Dict[Any, Any], payload: Dict[Any, Any], payload_version: str, ) -> Dict[Any, Any]: """ Handle Alexa.Media.Playback directives (Initiate). These directives are used to start playback of media content. """ if directive_message_id: set_message_id_for_logs(directive_message_id) try: if payload_version != "1.0": logger.warning( f"Initiate request with unexpected version '{payload_version}'" ) # Create a dictionary representing the request instead of a Pydantic model initiate_request = {"header": header, "payload": payload} return process_initiate_playback(initiate_request) except Exception as e: # Don't wrap specific error types - let them propagate to the top-level handler # which knows how to properly map them to appropriate Alexa error types logger.warning(f"Error in Initiate request, propagating: {type(e).__name__}: {str(e)}") raise # Re-raise to be caught by the top-level handler def handle_alexa_media_playqueue( directive_name: str, directive_message_id: str, payload: Dict[Any, Any], ) -> Dict[Any, Any]: """ Handle Alexa.Media.PlayQueue directives (SetLoop, SetRepeat, SetPlaybackContinuation). These directives are used to control playback modes for the media queue. """ if directive_message_id: set_message_id_for_logs(directive_message_id) try: # Handle SetLoop directive if directive_name == "SetLoop": # Extract enable flag from payload enable_loop = payload.get("enable", False) # --- Temporary fix for Alexa sending SetLoop instead of SetRepeat --- # Convert boolean to "ON"/"OFF" status repeat_mode = "ON" if enable_loop else "OFF" logger.info(f"SetLoop request received with enable: {enable_loop}. Treating as SetRepeat with mode: {repeat_mode}.") # Extract queueId from currentItemReference queue_id = None current_item_ref = payload.get("currentItemReference", {}) if current_item_ref and "value" in current_item_ref: queue_id = current_item_ref.get("value", {}).get("queueId") if queue_id: set_repeat_status_in_kv(queue_id, repeat_mode) else: logger.warning("SetLoop request received without a queueId. Cannot persist repeat status.") # --- End temporary fix --- # Build and return SetLoop response return { "header": { "messageId": directive_message_id, "namespace": "Alexa", "name": "Response", "payloadVersion": "3.0", }, "payload": {} } # Handle SetRepeat directive elif directive_name == "SetRepeat": # Extract repeat mode from payload repeat_mode = payload.get("mode", {}).get("status", "OFF") # Extract queueId from currentItemReference queue_id = None current_item_ref = payload.get("currentItemReference", {}) if current_item_ref and "value" in current_item_ref: queue_id = current_item_ref.get("value", {}).get("queueId") if queue_id: set_repeat_status_in_kv(queue_id, repeat_mode) else: logger.warning("SetRepeat request received without a queueId.") logger.info(f"SetRepeat request received with repeatMode: {repeat_mode}") # Build and return SetRepeat response return { "header": { "messageId": directive_message_id, "namespace": "Alexa", "name": "Response", "payloadVersion": "3.0", }, "payload": {} } # Handle SetPlaybackContinuation directive elif directive_name == "SetPlaybackContinuation": # Extract continuation mode from payload continuation_enabled = payload.get("enabled", False) logger.info(f"SetPlaybackContinuation request received with enabled: {continuation_enabled}") # Build and return SetPlaybackContinuation response return { "header": { "messageId": directive_message_id, "namespace": "Alexa", "name": "Response", "payloadVersion": "3.0", }, "payload": {} } else: # Unsupported directive within the namespace msg = f"Unsupported Alexa.Media.PlayQueue directive: {directive_name}" logger.warning(msg) return build_alexa_error_response( directive_message_id, "INVALID_DIRECTIVE", msg ) except Exception as e: # Don't wrap specific error types - let them propagate to the top-level handler logger.warning(f"Error in {directive_name} request, propagating: {type(e).__name__}: {str(e)}") raise # Re-raise to be caught by the top-level handler def build_media_search_playable_content_response( directive_message_id: str, item: Dict[str, Any], selected_criteria_id: str, ) -> Dict[str, Any]: """ Builds response for GetPlayableContent directive with a single content item. """ art_url = get_art_url_for_item(item) item_id = item.get("id", "default") # Generate art sources art_sources = generate_art_sources(image_url=art_url) # Basic metadata structure metadata = { "type": "TRACK", "name": { "speech": { "type": "PLAIN_TEXT", "text": item.get("title", "Unknown Title"), }, "display": item.get("title", "Unknown Title"), }, "art": {"sources": art_sources}, } return { "header": { "messageId": directive_message_id, "namespace": "Alexa.Media.Search", "name": "GetPlayableContent.Response", "payloadVersion": "3.0", }, "payload": { "matchedCriteria": { "criteriaId": selected_criteria_id }, "content": { "id": item_id, "metadata": metadata, } } } def build_media_search_displayable_content_response( directive_message_id: str, results: List[Dict[str, Any]], selected_criteria_id: str, ) -> Dict[str, Any]: """ Builds response for GetDisplayableContent directive with content groups. """ # Format the content items content_list = [] # Create default group title group_title = "Your Library" # If we have results, process them if results and len(results) > 0: for item in results: # Format each track item item_id = item.get("id", "unknown") title = item.get("title", "Unknown Title") status = item.get("status", "complete") art_url = get_art_url_for_item(item) art_sources = generate_art_sources(image_url=art_url) # Check if the song is explicit is_explicit = item.get("explicit", False) # Get duration in milliseconds if available duration_ms = None metadata = item.get("metadata", {}) if metadata and "duration" in metadata: try: duration_float = float(metadata["duration"]) duration_ms = int(duration_float * 1000) except (ValueError, TypeError): pass # Get tags if available tags = [] if metadata and "tags" in metadata and metadata["tags"]: tags_str = metadata["tags"] tags = [tag.strip() for tag in tags_str.split(",") if tag.strip()] # Build metadata object metadata_obj = { "type": "TRACK", "name": { "speech": {"type": "PLAIN_TEXT", "text": title}, "display": title, }, "art": {"sources": art_sources}, } # Add optionally available metadata if duration_ms: metadata_obj["durationInMilliseconds"] = duration_ms if is_explicit: metadata_obj["contentAdvisory"] = {"isExplicit": True} # Format content item content_item = { "id": item_id, "contentActions": [], "metadata": metadata_obj, } # Add generation status if it's not complete if status != "complete": content_item["generationStatus"] = { "state": "IN_PROGRESS" if status == "in_progress" else "PENDING", "estimatedTimeToCompletionInMilliseconds": 60000, } content_list.append(content_item) # If we have results from a search, use a more descriptive title if any(item.get("search_term") for item in results): search_term = next( (item.get("search_term") for item in results if item.get("search_term")), "", ) if search_term: group_title = f"Results for '{search_term}'" # Create content group content_groups = [ { "metadata": { "name": { "speech": {"type": "PLAIN_TEXT", "text": group_title}, "display": group_title, } }, "contentList": content_list, } ] return { "header": { "messageId": directive_message_id, "namespace": "Alexa.Media.Search", "name": "GetDisplayableContent.Response", "payloadVersion": "3.0", }, "payload": { "matchedCriteria": { "criteriaId": selected_criteria_id }, "contentGroups": content_groups } } def build_play_queue_item_response( directive_message_id: str, directive_name: str, item: Optional[Dict[str, Any]], is_queue_finished: bool, ) -> Dict[str, Any]: """ Builds response for GetNextItem or GetPreviousItem directives. """ payload = {"isQueueFinished": is_queue_finished} if item and not is_queue_finished: payload["item"] = item return { "header": { "messageId": directive_message_id, "namespace": "Alexa.Audio.PlayQueue", "name": f"{directive_name}.Response", "payloadVersion": "1.0", }, "payload": payload } def build_alexa_media_search_response( directive_message_id: str, directive_name: str, results: Any, selected_criteria_id: str, is_queue_finished: bool = False, ) -> Dict[str, Any]: """ Builds response for various content retrieval directives by calling the appropriate builder. For GetPlayableContent: Uses build_media_search_playable_content_response For GetDisplayableContent: Uses build_media_search_displayable_content_response For GetNextItem/GetPreviousItem: Uses build_play_queue_item_response """ if directive_name == "GetPlayableContent": # Ensure we have a single item (not a list) result_item = results[0] if isinstance(results, list) and results else results return build_media_search_playable_content_response( directive_message_id, result_item, selected_criteria_id ) elif directive_name == "GetDisplayableContent": # Ensure results is a list results_list = results if isinstance(results, list) else [results] if results else [] return build_media_search_displayable_content_response( directive_message_id, results_list, selected_criteria_id ) elif directive_name in ["GetNextItem", "GetPreviousItem"]: return build_play_queue_item_response( directive_message_id, directive_name, results, is_queue_finished ) else: # Return error for unknown directive return build_alexa_error_response( directive_message_id, "INTERNAL_ERROR", f"Cannot build success response for unknown request type: {directive_name}", ) # ============================================================================= # Alexa.Audio.PlayQueue Item Builder # ============================================================================= def build_play_queue_item( clip_id: str, directive_message_id: str, clip_details: Optional[Dict[str, Any]] = None, access_token: Optional[str] = None, enable_next: bool = True, enable_previous: bool = True, repeat_status: str = "OFF", ) -> Dict[str, Any]: """ Builds PlayQueueItem from clip details, with fallback to API and SSE if needed. Retrieves or generates all required fields including metadata, stream, controls, transcript, art sources and duration for playback in Alexa. """ # Start measuring time to enforce 4-second limit handler_start_time = time.time() # Initialize with fallback values title = "" image_url = None audio_url = f"{SUNO_AUDIO_ENDPOINT}/stream_with_filler/?item_id={clip_id}" duration_ms = None status = "unknown" suno_request_id = None # If clip details are already provided, extract information if clip_details: title = clip_details.get("title", title) status = clip_details.get("status", status) # For completed clips, we can return immediately with all data if status == "complete": image_url = get_art_url_for_item(clip_details) audio_url = clip_details.get("audio_url", audio_url) clip_metadata = clip_details.get("metadata", {}) if clip_metadata.get("duration"): try: duration_val = float(clip_metadata["duration"]) if duration_val > 0: duration_ms = int(duration_val * 1000) except (ValueError, TypeError): pass # Handle "submitted" status specifically - fetch hydrated data from clip details API elif status == "submitted" and access_token: suno_request_id = clip_details.get("request_id") try: # Fetch fully hydrated clip details hydrated_details = get_clip_details(access_token, clip_id) # Get the title directly if hydrated_details and hydrated_details.get("title"): title = hydrated_details.get("title") # Get other metadata image_url = get_art_url_for_item(hydrated_details) audio_url = hydrated_details.get("audio_url", audio_url) status = hydrated_details.get("status", status) # Get duration if available hydrated_metadata = hydrated_details.get("metadata", {}) if hydrated_metadata.get("duration"): try: duration_val = float(hydrated_metadata["duration"]) if duration_val > 0: duration_ms = int(duration_val * 1000) except (ValueError, TypeError): pass except Exception as e: logger.warning(f"Failed to fetch hydrated details for submitted clip {clip_id}: {str(e)}") # If still no title but we have a request_id, try to get title from SSE if not title and suno_request_id: try: sse_data = get_title_and_tags_from_sse(suno_request_id, timeout=1.0) if sse_data.get("title"): title = sse_data.get("title") except Exception as e: logger.warning(f"Failed to get title from SSE: {str(e)}") else: # For non-complete clips, get what we can from the details first image_url = get_art_url_for_item(clip_details) audio_url = clip_details.get("audio_url", audio_url) suno_request_id = clip_details.get("request_id") clip_metadata = clip_details.get("metadata", {}) if clip_metadata.get("duration"): try: duration_val = float(clip_metadata["duration"]) if duration_val > 0: duration_ms = int(duration_val * 1000) except (ValueError, TypeError): pass # Fetch Content Details if Not Provided but Token Exists elif access_token: try: fetched_clip_details = get_clip_details(access_token, clip_id) return build_play_queue_item( clip_id, directive_message_id, fetched_clip_details, access_token, enable_next, enable_previous, repeat_status ) except Exception as e: logger.error(f"Error fetching clip details: {e}") # Continue with default values in case of error # For non-complete clips, we need to ensure both audio and image are ready # Only do this if we didn't already determine it's complete or submitted if status != "complete" and status != "submitted": # Call wait_for_clip_events which now returns success even if only audio is ready clip_events = wait_for_clip_events( clip_id=clip_id, handler_start_time=handler_start_time ) # Check if we should use fallback image due to image generation timeout if clip_events.get("use_fallback_image", False): # Audio is ready but image generation timed out - use fallback image_url = get_fallback_image_url() logger.info(f"Using fallback image for clip {clip_id} due to image generation timeout") else: # Both audio and image are ready - use the generated image image_id = clip_events.get("image_id") image_url = f"{SUNO_IMAGE_CDN}/{image_id}.jpeg" # Get title from events if available event_title = clip_events.get("title") if event_title: title = event_title # Use a fallback title if we still don't have one if not title: title = "Untitled" logger.warning(f"Using fallback title: {title}") # Set up stream stream_id = f"stream-{clip_id}" valid_until_dt = datetime.now(timezone.utc) + timedelta(minutes=10) valid_until_iso = ( valid_until_dt.replace(tzinfo=None).isoformat(timespec="seconds") + "Z" ) transcript_uri = ( f"https://studio-api.prod.suno.com/api/v2/external/lyrics?clip_id={clip_id}" ) transcript = { "id": f"LYRIC_{clip_id}", "uri": transcript_uri, "format": "WEBVTT", "headers": [ { "name": "x-auth-token", "value": "fbfd3378-a72d-45ef-9782-4edf140f5dd7", } ], } # Build art sources based on image URL art_sources = generate_art_sources( image_url=image_url ) # Create Metadata item_metadata = { "type": "TRACK", "name": { "speech": {"type": "PLAIN_TEXT", "text": title}, "display": title, }, "art": {"sources": art_sources}, } # Create Stream item_stream = { "id": stream_id, "uri": audio_url, "offsetInMilliseconds": 0, "validUntil": valid_until_iso, } # Create PlaybackInfo item_playback_info = {"type": "DEFAULT"} # Create controls controls = [ {"type": "COMMAND", "name": "NEXT", "enabled": enable_next}, {"type": "COMMAND", "name": "PREVIOUS", "enabled": enable_previous}, { "type": "ADJUST", "name": "SEEK_POSITION", "enabled": (duration_ms is not None), }, { "name": "REPEAT", "type": "CYCLE", "enabled": True, "value": {"status": repeat_status}, }, ] # Construct PlayQueueItem with conditional duration item = { "id": clip_id, "playbackInfo": item_playback_info, "metadata": item_metadata, "stream": item_stream, "controls": controls, "rules": {"feedbackEnabled": False}, "transcript": transcript, } # Only include durationInMilliseconds if we have a valid value if duration_ms is not None: item["durationInMilliseconds"] = duration_ms return item # ============================================================================= # GPC 3.0 Request Processing Logic # ============================================================================= def filter_supported_attributes(criteria: Dict[str, Any]) -> List[Dict[str, Any]]: """ Filter out unsupported attributes from criteria, returning only supported ones. Args: criteria: The criteria dictionary containing attributes Returns: List of supported attributes only """ supported_attributes = [] for attr in criteria.get("attributes", []): attr_type = attr.get("type") if attr_type not in SUPPORTED_ATTRIBUTE_TYPES: continue # Special handling for MEDIA_TYPE - we only support TRACK if attr_type == "MEDIA_TYPE": value = attr.get("value", "").upper() if value != "TRACK" and value != "SONG": continue supported_attributes.append(attr) return supported_attributes def check_attribute_support(criteria: Dict[str, Any]) -> bool: """ Verifies if the skill supports all attributes and predicates in criteria. Currently we support attributes in SUPPORTED_ATTRIBUTE_TYPES. Returns True if criteria can be processed (even with some attributes ignored). """ has_supported_attributes = False ignored_attributes = [] for attr in criteria.get("attributes", []): attr_type = attr.get("type") attr_id = attr.get("id", "unknown") if attr_type not in SUPPORTED_ATTRIBUTE_TYPES: ignored_attributes.append(f"{attr_type} (id: {attr_id})") logger.info(f"Ignoring unsupported attribute type: {attr_type}") continue # Special handling for MEDIA_TYPE - we only support TRACK if attr_type == "MEDIA_TYPE": value = attr.get("value", "").upper() if value != "TRACK" and value != "SONG": ignored_attributes.append(f"MEDIA_TYPE={value} (id: {attr_id})") logger.info(f"Ignoring unsupported MEDIA_TYPE value: {value}") continue # At least one supported attribute found has_supported_attributes = True # Log summary of ignored attributes if ignored_attributes: logger.info(f"Ignored {len(ignored_attributes)} unsupported attributes: {', '.join(ignored_attributes)}") # Return True if we have at least one supported attribute, or if no attributes at all # (some criteria might only have queries without attributes) return has_supported_attributes or len(criteria.get("attributes", [])) == 0 def handle_attribute_query( criteria: Dict[str, Any], filters: Optional[Dict[str, Any]], access_token: Optional[str] = None, request_name: str = "GetPlayableContent", max_results: int = 8, ) -> Optional[List[Dict[str, Any]]]: """ Resolves content search criteria to matching media items. For GetDisplayableContent: - Filters by GENRE if specified - Falls back to recent library items For GetPlayableContent: - Prioritizes explicit TRACK identifier - Uses MEDIA_TYPE for general track playback Only processes attributes that are in SUPPORTED_ATTRIBUTE_TYPES. """ # Filter to only supported attributes supported_attributes = filter_supported_attributes(criteria) # Handle GetDisplayableContent differently if request_name == "GetDisplayableContent": # Extract genre from attributes if present genre = None has_media_type_track = False # Check attributes for GENRE and MEDIA_TYPE for attr in supported_attributes: if attr.get("type") == "GENRE": # Access raw_value directly from the dictionary genre = attr.get("rawValue") elif attr.get("type") == "MEDIA_TYPE": # For MEDIA_TYPE, ignore rawValue and only check the value field # since rawValue can vary ("song", "music", etc.) but value is consistent value = attr.get("value", "").upper() if value == "TRACK": has_media_type_track = True # If we don't have MEDIA_TYPE=TRACK, log a warning but continue anyway if not has_media_type_track: # Check if we have any MEDIA_TYPE attribute at all has_media_type_attr = any(attr.get("type") == "MEDIA_TYPE" for attr in supported_attributes) if has_media_type_attr: logger.warning("GetDisplayableContent with unsupported MEDIA_TYPE (not TRACK)") # Get library content based on parameters: # 1. If genre is specified, search by genre # 2. If no genre, return recent library songs if genre: search_results = search_suno_library( access_token=access_token, search_term=genre, page_size=max_results, ) else: # Get library content with empty search term for most recent songs search_results = search_suno_library( access_token=access_token, search_term="", # Empty search returns recent songs page_size=max_results, ) return search_results # GetPlayableContent logic # 1. Look for a specific TRACK attribute found_track_id = None for attr in supported_attributes: if attr.get("type") == "TRACK": resolved_entities = attr.get("resolvedEntities", []) if resolved_entities and len(resolved_entities) > 0: entity = resolved_entities[0] found_track_id = entity.get("entityId") if found_track_id: break # Check for "play my song" request via MEDIA_TYPE=TRACK criteria has_media_type_track = False for attr in supported_attributes: if attr.get("type") == "MEDIA_TYPE": # For MEDIA_TYPE, ignore rawValue and only check the value field # since rawValue can vary ("song", "music", etc.) but value is consistent value = attr.get("value", "").upper() if value == "TRACK": has_media_type_track = True break # 2. Handle predicate tree if present if criteria.get("query"): # Get the query object which represents the predicate tree query = criteria.get("query", {}) predicate_type = query.get("type") # Extract search terms based on predicate type if predicate_type == "ATTRIBUTE": # Direct attribute reference - use the referenced attribute for search attribute_id = query.get("attributeId") # Find the attribute with this ID for attr in supported_attributes: if attr.get("id") == attribute_id: # For ARTIST or similar entities with resolvedEntities if "resolvedEntities" in attr: # Just use the first entityId for simplicity entities = attr.get("resolvedEntities", []) if entities and len(entities) > 0: return search_suno_library( access_token=access_token, search_term=entities[0].get("entityId", ""), page_size=max_results, ) # For text attributes like queries elif "rawValue" in attr: return search_suno_library( access_token=access_token, search_term=attr.get("rawValue", ""), page_size=max_results, ) # For a simple implementation, we'll just extract any ATTRIBUTE predicates # from the tree and use them for search elif predicate_type in ["NARY", "UNARY"]: # Log the specific condition to acknowledge its intended semantics if predicate_type == "NARY": condition = query.get("condition") logger.debug(f"NARY predicate with condition {condition} - simplifying implementation") # UNION: Any predicate can be satisfied # INTERSECTION: All predicates must be satisfied elif predicate_type == "UNARY": condition = query.get("condition") logger.debug(f"UNARY predicate with condition {condition} - simplifying implementation") # SIMILAR: Find content similar to predicate # NOT: Exclude content matching predicate # For simplicity in this initial implementation, just extract any attribute IDs # we can find and search for each of them separately attribute_ids = extract_attribute_ids_from_predicate(query) # Use the first attribute ID for search if attribute_ids and len(attribute_ids) > 0: for attr in supported_attributes: if attr.get("id") == attribute_ids[0]: # For ARTIST or similar entities with resolvedEntities if "resolvedEntities" in attr: entities = attr.get("resolvedEntities", []) if entities and len(entities) > 0: return search_suno_library( access_token=access_token, search_term=entities[0].get("entityId", ""), page_size=max_results, ) # For text attributes like queries elif "rawValue" in attr: return search_suno_library( access_token=access_token, search_term=attr.get("rawValue", ""), page_size=max_results, ) # Log that we couldn't fully process this predicate structure logger.warning(f"Could not fully process predicate structure: {predicate_type}") # 3. Process track ID to build response if found_track_id: # Create initial empty result with id result = { "id": found_track_id, "title": "Unknown Title", } clip_details = get_clip_details(access_token, found_track_id) title = clip_details.get("title", result["title"]) image_url = get_art_url_for_item(clip_details) result = { "id": found_track_id, "title": title, "image_url": image_url, "source_criteria_id": criteria.get("id"), "metadata": { "duration_ms": int( float(clip_details.get("metadata", {}).get("duration", 180)) * 1000 ) if clip_details.get("metadata", {}).get("duration") else 180_000, }, } return [result] # Return as a list containing the one item # Return most recent song when MEDIA_TYPE=TRACK with no specific song ID elif has_media_type_track and request_name == "GetPlayableContent": search_results = search_suno_library( access_token=access_token, search_term="", # Empty search returns recent songs page_size=1, ) # Make sure we have results and log what we're returning if search_results and len(search_results) > 0: return search_results else: logger.warning("No recent songs found in library for MEDIA_TYPE=TRACK request") return None else: # No track ID found or other attribute search not implemented logger.info( "No specific TRACK ID found, or predicate logic not fully implemented." ) # Signal that this criteria didn't yield usable results return None def extract_attribute_ids_from_predicate(predicate): """ Simple helper function to extract attribute IDs from a predicate tree. This is a minimal implementation that doesn't handle the full predicate semantics. """ result = [] predicate_type = predicate.get("type") if predicate_type == "ATTRIBUTE": # Direct attribute reference attribute_id = predicate.get("attributeId") if attribute_id: result.append(attribute_id) elif predicate_type == "NARY": # Process sub-predicates in a NARY predicate for sub_predicate in predicate.get("predicates", []): result.extend(extract_attribute_ids_from_predicate(sub_predicate)) elif predicate_type == "UNARY": # Process the sub-predicate in a UNARY predicate sub_predicate = predicate.get("predicate") if sub_predicate: result.extend(extract_attribute_ids_from_predicate(sub_predicate)) return result def handle_generate_content_request( access_token: str, prompt: str, message_id: str, request_name: str, user_id: Optional[str], nl_criteria_id: Optional[str], handler_start_time: float, ) -> Dict[str, Any]: """ Handle GENERATE_CONTENT action for content generation. Calls the generation handler, processes the result, updates catalog, and returns the appropriate response. """ global _real_processing_end_time # Call the generation handler with time budget tracking generation_result = handle_generate_content_action( access_token, prompt, handler_start_time ) if "error_code" in generation_result: error_code = generation_result.get("error_code", "CONTENT_FILTERED") return build_media_error_response( message_id, error_code, error_code, ) # Extract generated content details clip_id = generation_result["clip_id"] title = generation_result["title"] image_url = generation_result["image_url"] # Prepare result for response builder result_item = { "id": clip_id, "title": title, "image_url": image_url, } selected_criteria_id = nl_criteria_id or "ACTION_GENERATED_CONTENT" # Update actual processing time before adding artificial delay _real_processing_end_time = time.time() # Calculate elapsed time and add delay if needed elapsed_time = time.time() - handler_start_time if elapsed_time < MIN_GENERATE_CONTENT_EXECUTION_TIME_SECONDS: delay_seconds = MIN_GENERATE_CONTENT_EXECUTION_TIME_SECONDS - elapsed_time time.sleep(delay_seconds) if request_name == "GetPlayableContent": return build_media_search_playable_content_response( message_id, result_item, selected_criteria_id, ) else: # GetDisplayableContent return build_media_search_displayable_content_response( message_id, [result_item], selected_criteria_id, ) def handle_empty_browse_request( access_token: str, message_id: str, request_name: str, max_results: int, ) -> Dict[str, Any]: """ Handle empty browse/play requests with no criteria. Returns recent library items for browsing or the most recent song for playback. """ if request_name == "GetDisplayableContent": # Get recent songs when no search criteria provided search_results = search_suno_library( access_token=access_token, search_term="", # Empty search returns recent songs page_size=max_results, ) selected_criteria_id = "EMPTY_BROWSE_ALL_CONTENT" return build_media_search_displayable_content_response( message_id, search_results, selected_criteria_id ) else: # GetPlayableContent search_results = search_suno_library( access_token=access_token, search_term="", page_size=1, ) if search_results and len(search_results) > 0: return build_media_search_playable_content_response( message_id, search_results[0], "EMPTY_PLAYABLE_CONTENT" ) else: return build_media_error_response( message_id, error_type="CONTENT_NOT_FOUND", message="Could not find content in your library.", ) def handle_ranked_criteria_evaluation( ranked_criteria: List[Dict[str, Any]], filters: Dict[str, Any], access_token: str, message_id: str, request_name: str, max_results: int, ) -> Dict[str, Any]: """ Process ranked criteria evaluating each in priority order. Tries each criteria in order until finding valid results or determining no results are available. """ selected_criteria_id = None results = None last_error = None # Track significant errors for potential re-raising for criteria in ranked_criteria: criteria_id = criteria.get("id", "UNKNOWN_CRITERIA") criteria_type = criteria.get("type", "UNKNOWN_TYPE") try: potential_results = None if criteria_type == "NL_QUERY": logger.debug(f"Skipping NL_QUERY criteria: {criteria_id}") continue # Skip NL criteria as we don't support it yet elif criteria_type == "ATTRIBUTES": if check_attribute_support(criteria): logger.debug(f"Processing ATTRIBUTES criteria: {criteria_id}") potential_results = handle_attribute_query( criteria, filters, access_token, request_name, max_results, ) logger.debug(f"Attribute query returned: {potential_results is not None}") else: logger.debug(f"Unsupported attributes in criteria: {criteria_id}") continue else: logger.warning(f"Unknown criteria type encountered: {criteria_type}") continue # If we got valid results, use them if potential_results is not None: if isinstance(potential_results, list) and len(potential_results) > 0: results = potential_results selected_criteria_id = criteria_id break # Exit the loop, we found usable results else: logger.warning(f"Criteria {criteria_id} returned empty results list") else: logger.debug(f"No results for criteria: {criteria_id}") except SkillError as e: last_error = e # Only continue for non-critical errors if isinstance(e, (ContentNotFoundError, ContentFilteredError)): continue else: # Critical errors halt processing immediately raise e # Build final response based on processing results if results is not None and selected_criteria_id is not None: if request_name == "GetPlayableContent": result_item = results[0] if len(results) > 0 else None if result_item: return build_media_search_playable_content_response( message_id, result_item, selected_criteria_id ) else: # GetDisplayableContent return build_media_search_displayable_content_response( message_id, results, selected_criteria_id ) # Report last encountered error if we have one if last_error: logger.warning( f"No criteria succeeded, returning last encountered error: {last_error.__class__.__name__}" ) raise last_error else: logger.warning("No suitable criteria yielded results or results were empty.") return build_media_error_response( message_id, error_type="CONTENT_NOT_FOUND", message="Could not find content matching the request.", subtypes=None, ) def process_media_search( media_request: Dict[str, Any], handler_start_time: float ) -> Dict[str, Any]: """ Core handler for media search and content generation requests. Handles both GetPlayableContent and GetDisplayableContent directives including: - Content generation actions - Ranked criteria evaluation - Library browsing - Search functionality Returns appropriate responses based on matching content or errors encountered. """ payload = media_request.get("payload", {}) header = media_request.get("header", {}) ranked_criteria = payload.get("rankedSelectionCriteria", []) filters = payload.get("filters", {}) message_id = header.get("messageId") request_name = header.get("name") action = payload.get("action") max_results = 8 if request_name == "GetDisplayableContent": # Direct dictionary access if payload.get("maxResultLimit") is not None: max_results = payload.get("maxResultLimit") # Access pagination context through dictionary pagination_context = payload.get("paginationContext", {}) if pagination_context: limits = pagination_context.get("limits", {}) if limits: content_items_limit = limits.get("contentItemsPerContentListLimit", {}) if content_items_limit and content_items_limit.get("maxLimit", 0) > 0: max_results = content_items_limit.get("maxLimit") # Extract access token from the dictionary access_token = None user_id = None # Access request_context directly request_context = payload.get("requestContext", {}) if request_context and request_context.get("user"): user = request_context.get("user", {}) access_token = user.get("accessToken") user_id = user.get("id") # Check for explicit action first if action and action.get("type") == "GENERATE_CONTENT": # Extract prompt from natural language criteria prompt = None nl_criteria_id = None for criteria in ranked_criteria: if criteria.get("type") == "NL_QUERY" and criteria.get("query"): prompt = criteria.get("query") nl_criteria_id = criteria.get("id") break return handle_generate_content_request( access_token, prompt, message_id, request_name, user_id, nl_criteria_id, handler_start_time ) # Handle empty browse case for library content if len(ranked_criteria) == 0 and not action: return handle_empty_browse_request( access_token, message_id, request_name, max_results ) # Process ranked criteria if we have them return handle_ranked_criteria_evaluation( ranked_criteria, filters, access_token, message_id, request_name, max_results ) # ============================================================================= # Initiate V1.0 Request Processing # ============================================================================= def process_initiate_playback(request: Dict[str, Any]) -> Dict[str, Any]: """ Processes media playback initiation by creating a playback queue. Extracts authentication details, builds the first queue item, and constructs a complete playback response with controls. """ global _real_processing_end_time # Track execution time for potential delay handler_start_time = time.time() # Extract access token directly from the dictionary access_token = None payload = request.get("payload", {}) request_context = payload.get("requestContext") if request_context and request_context.get("user"): user_context = request_context.get("user", {}) access_token = user_context.get("accessToken") content_id = payload.get("contentId") message_id = request.get("header", {}).get("messageId") # Get playback modes from the request payload playback_modes = payload.get("playbackModes", {}) repeat_status = playback_modes.get("repeat", {}).get("status", "OFF") # Construct queue_id and persist initial repeat status queue_id = f"queue-{message_id}" set_repeat_status_in_kv(queue_id, repeat_status) # Build the first item in the play queue # For Initiate response: disable previous (it's the start of the queue) and enable next first_item = build_play_queue_item( content_id, message_id, None, access_token, enable_next=True, # Always enable next for Initiate enable_previous=False, # Always disable previous for Initiate (start of queue) repeat_status=repeat_status, ) # Add additional check to make sure controls have the right settings if isinstance(first_item, dict) and "controls" in first_item: for control in first_item["controls"]: if control.get("name") == "PREVIOUS": # Force PREVIOUS to be disabled at item level control["enabled"] = False response = { "header": { "namespace": "Alexa.Media.Playback", "name": "Initiate.Response", "messageId": message_id, "payloadVersion": "1.0", }, "payload": { "playbackMethod": { "type": "ALEXA_AUDIO_PLAYER_QUEUE", "id": queue_id, "firstItem": first_item, "controls": [ {"type": "TOGGLE", "name": "SHUFFLE", "enabled": False, "selected": False}, {"type": "TOGGLE", "name": "LOOP", "enabled": True, "selected": False}, { "name": "REPEAT", "type": "CYCLE", "enabled": True, "value": { "status": repeat_status } }, { "type": "ADJUST", "name": "SEEK_POSITION", "enabled": ("durationInMilliseconds" in first_item), "selected": False, }, # ALWAYS enable NEXT and PREVIOUS at queue level {"type": "COMMAND", "name": "NEXT", "enabled": True}, {"type": "COMMAND", "name": "PREVIOUS", "enabled": True}, ], "rules": {"feedback": {"type": "PREFERENCE", "enabled": False}}, } } } # Only add delay if we're NOT dealing with a complete song # Determine if this clip is a complete one is_complete_song = False if first_item and "durationInMilliseconds" in first_item: # If we have duration, the song is complete is_complete_song = True # Update actual processing time before adding artificial delay _real_processing_end_time = time.time() # Calculate elapsed time and add delay if needed - only for incomplete/generating songs elapsed_time = time.time() - handler_start_time if not is_complete_song and elapsed_time < MIN_GENERATE_CONTENT_EXECUTION_TIME_SECONDS: delay_seconds = MIN_GENERATE_CONTENT_EXECUTION_TIME_SECONDS - elapsed_time time.sleep(delay_seconds) return response