You can now close this window and return to the terminal.
""" else: error = params.get("error", ["Unknown error"])[0] response = f"""Error: {error}
""" self.wfile.write(response.encode()) print("\nRedirect received. You can close the browser window.") return return super().do_GET() def log_message(self, format, *args): # Suppress server logs return def start_oauth_server(): """Start the OAuth callback server and return the server instance""" class CustomTCPServer(socketserver.TCPServer): allow_reuse_address = True auth_code = None server = CustomTCPServer(("", 3001), OAuthCallbackHandler) server.auth_code = None server_thread = threading.Thread(target=server.serve_forever) server_thread.daemon = True server_thread.start() print("Server started on port 3001") return server def get_authorization_url(): """Generate the authorization URL""" params = { "client_id": CLIENT_ID, "response_type": "code", "redirect_uri": REDIRECT_URI, "scope": "read_profile generate_music read_music", "state": f"test-{int(time.time())}", } return f"{AUTH_ENDPOINT}?{urllib.parse.urlencode(params)}" def exchange_code_for_token(code): """Exchange the authorization code for an access token""" data = { "grant_type": "authorization_code", "code": code, "redirect_uri": REDIRECT_URI, "client_id": CLIENT_ID, } headers = { "Content-Type": "application/x-www-form-urlencoded", "Authorization": f"Basic {base64.b64encode(f'{CLIENT_ID}:{CLIENT_SECRET}'.encode()).decode()}", "x-auth-type": "oauth", # Required by Suno API } response = requests.post(TOKEN_ENDPOINT, data=data, headers=headers) response.raise_for_status() return response.json() def get_oauth_token(): """Complete the OAuth flow and return an access token""" server = start_oauth_server() try: auth_url = get_authorization_url() print("\nAuthorization URL:") print(auth_url) webbrowser.open(auth_url) print("Waiting for authorization...") wait_time = 0 while server.auth_code is None and wait_time < 300: time.sleep(1) wait_time += 1 if server.auth_code: print("Authorization code received, exchanging for tokens...") token_data = exchange_code_for_token(server.auth_code) if token_data and "access_token" in token_data: print("Access token obtained successfully!") return token_data["access_token"] else: print("Failed to exchange authorization code for tokens.") return None else: print("Timed out waiting for authorization.") return None finally: # Improve server shutdown to prevent hanging try: # Set a timeout for the shutdown shutdown_thread = threading.Thread(target=server.shutdown) shutdown_thread.daemon = True shutdown_thread.start() # Wait for shutdown with timeout shutdown_thread.join(timeout=2.0) # Continue even if timeout occurs if shutdown_thread.is_alive(): print("Server shutdown timed out, but continuing...") else: server.server_close() print("Server stopped") except Exception as e: print(f"Error during server shutdown: {e}") print("Continuing despite server shutdown issue...") def create_get_next_item_request( access_token, clip_id, queue_id, is_user_initiated=True ): """Create a sample GetNextItem request with configurable parameters.""" request = { "header": { "messageId": f"test-{int(time.time())}", "namespace": "Alexa.Audio.PlayQueue", "name": "GetNextItem", "payloadVersion": "1.0", }, "payload": { "requestContext": { "user": { "id": "amzn1.ask.account.TEST_USER_GENERATE", "accessToken": access_token, }, "location": {"originatingLocale": "en-US"}, }, "currentItemReference": { "id": clip_id, "queueId": queue_id, "contentId": clip_id, }, "isUserInitiated": is_user_initiated, }, } return request def create_get_previous_item_request(access_token, clip_id, queue_id, is_user_initiated=True): """Create a sample GetPreviousItem request with clip_id""" return { "header": { "messageId": f"test-{int(time.time())}", "namespace": "Alexa.Audio.PlayQueue", "name": "GetPreviousItem", "payloadVersion": "1.0", }, "payload": { "requestContext": { "user": { "id": "amzn1.ask.account.TEST_USER_GENERATE", "accessToken": access_token, }, "location": { "originatingLocale": "en-US" } }, "currentItemReference": { "id": clip_id, "queueId": queue_id, "contentId": clip_id, }, "isUserInitiated": is_user_initiated, }, } def create_gpc_request(access_token, query="Create a song about a cat"): """Create a sample GetPlayableContent request with query AND generate action""" return { "header": { "messageId": f"test-{int(time.time())}", "namespace": "Alexa.Media.Search", "name": "GetPlayableContent", "payloadVersion": "3.0", }, "payload": { "requestContext": { "user": { # Add a default user ID for completeness, although not strictly used in generation yet "id": "amzn1.ask.account.TEST_USER_GENERATE", "accessToken": access_token, }, # Add location and advertising stubs as seen in example "location": {"originatingLocale": "en-US", "countryCode": "US"}, "advertising": { "limitAdTracking": True, "advertisingId": "00000000-0000-0000-0000-000000000000", }, }, "rankedSelectionCriteria": [ { "id": "NL_QUERY_CRITERIA", # Match example ID "type": "NL_QUERY", "query": query, }, # Add the dummy ATTRIBUTES criteria from the example for realism { "type": "ATTRIBUTES", "id": "MULTI_ATTR_CRITERIA", "attributes": [ { "type": "MEDIA_TYPE", "id": "U09ORw==", # Base64 for "SONG" "rawValue": "SONG", "value": "TRACK", } ], "completeness": {"score": 0.0, "bin": "LOW"}, "query": None, }, ], # Add the crucial action field "action": {"type": "GENERATE_CONTENT"}, # Add filters stub from example "filters": {"explicitLanguageAllowed": True}, }, } def create_simple_gpc_request(access_token): """Create a simplified GetPlayableContent request without query""" return { "header": { "messageId": f"test-{int(time.time())}", "namespace": "Alexa.Media.Search", "name": "GetPlayableContent", "payloadVersion": "3.0", }, "payload": { "requestContext": {"user": {"accessToken": access_token}}, "filters": {"explicitLanguageAllowed": True}, # Add mandatory rankedSelectionCriteria for GPC 3.0 "rankedSelectionCriteria": [ { "type": "ATTRIBUTES", "id": "criteria-simple-track-id-placeholder", "completeness": {"bin": "HIGH", "score": 1.0}, "attributes": [ { "type": "TRACK", "id": "attr-track-placeholder", "resolvedEntities": [ { # Replace with an actual track ID known to your system/Suno if needed for a more realistic test "entityId": "fb4afa08-e64b-41ef-8bac-738d23a8076b", } ], } ], "query": None, # No complex query needed for simple ID lookup } # Add other simple criteria if needed for different test cases ], }, } def create_initiate_request( access_token, content_id, user_id="amzn1.ask.account.DEFAULT_USER" ): """Create a sample Initiate request matching the format provided""" return { "header": { "namespace": "Alexa.Media.Playback", "name": "Initiate", "messageId": str(uuid.uuid4()), "payloadVersion": "1.0", # Updated to match lambda expectations }, "payload": { "requestContext": { "user": {"id": user_id, "accessToken": access_token}, "location": {"originatingLocale": "en-US"}, }, "filters": {"explicitLanguageAllowed": True}, "contentId": content_id, "playbackModes": { "shuffle": False, "loop": False, "repeat": {"status": "OFF"}, }, }, } def create_set_loop_request(access_token, clip_id, queue_id, enable=True): """Create a SetLoop request.""" return { "header": { "messageId": f"test-set-loop-{uuid.uuid4()}", "namespace": "Alexa.Media.PlayQueue", "name": "SetLoop", "payloadVersion": "1.0" }, "payload": { "requestContext": { "user": { "id": "amzn1.ask.account.TEST_USER_GENERATE", "accessToken": access_token } }, "enable": enable, "currentItemReference": { "namespace": "Alexa.Audio.PlayQueue", "name": "item", "value": { "contentId": clip_id, "queueId": queue_id, "id": clip_id, "content": { "id": clip_id, "metadataType": "TRACK" } } } } } def create_set_repeat_request(access_token, clip_id, queue_id, status="ON"): """Create a SetRepeat request.""" return { "header": { "messageId": f"test-set-repeat-{uuid.uuid4()}", "namespace": "Alexa.Media.PlayQueue", "name": "SetRepeat", "payloadVersion": "1.0" }, "payload": { "requestContext": { "user": { "id": "amzn1.ask.account.TEST_USER_GENERATE", "accessToken": access_token } }, "currentItemReference": { "namespace": "Alexa.Audio.PlayQueue", "name": "item", "value": { "contentId": clip_id, "queueId": queue_id, "id": clip_id, "content": { "id": clip_id, "metadataType": "TRACK" } } }, "mode": { "status": status } } } def load_request_from_file(filename, access_token=None): """Load a request from a JSON file and inject access token if provided""" full_path = os.path.join(REQUEST_EXAMPLES_DIR, filename) try: with open(full_path, "r") as f: request = json.load(f) # Inject access token if provided if ( access_token and "payload" in request and "requestContext" in request["payload"] and "user" in request["payload"]["requestContext"] ): request["payload"]["requestContext"]["user"]["accessToken"] = access_token return request except Exception as e: print(f"Error loading request from {filename}: {e}") return None def list_available_requests(): """List all available request examples in the request_examples directory""" if not os.path.exists(REQUEST_EXAMPLES_DIR): print(f"Request examples directory '{REQUEST_EXAMPLES_DIR}' does not exist.") return [] files = [f for f in os.listdir(REQUEST_EXAMPLES_DIR) if f.endswith(".json")] return files def save_request_to_file(request): """Save request JSON to a file for AWS Lambda console testing""" # Create directory if it doesn't exist os.makedirs("aws_console_requests", exist_ok=True) # Get request type and current timestamp for filename request_type = request.get("header", {}).get("name", "unknown") timestamp = int(time.time()) # Create filename with request type and timestamp filename = f"aws_console_requests/{request_type}_{timestamp}.json" # Save request to file with open(filename, "w") as f: json.dump(request, f, indent=2) print(f"\nRequest saved to {filename} for AWS console testing") return filename # Utility function for printing JSON consistently def print_json(data, title=None): """Print JSON data using Rich for better formatting""" if title: console.print(f"\n[bold cyan]{title}[/bold cyan]") if isinstance(data, str): # If data is already a JSON string console.print(JSON(data)) else: # Convert data to a JSON string console.print(JSON(json.dumps(data))) def test_lambda(request, mock_context=None, print_raw_request=False, function_url=None): """Test the lambda function with a request, either locally or via a URL.""" # Save request to file for AWS console testing save_request_to_file(request) # Print the raw request JSON for AWS console testing if print_raw_request: print("\nRaw request JSON (copy this for AWS Lambda console testing):") print_json(request) test_target_description = ( f"against URL: {function_url}" if function_url else "locally" ) print( f"\nTesting lambda {test_target_description} with request type: {request['header']['name']}" ) # Create a logging-safe copy of the payload with the token truncated/hidden only for display display_payload = json.loads(json.dumps(request["payload"])) if ( "requestContext" in display_payload and "user" in display_payload["requestContext"] and "accessToken" in display_payload["requestContext"]["user"] ): token = display_payload["requestContext"]["user"]["accessToken"] display_payload["requestContext"]["user"]["accessToken"] = ( f"{token[:10]}...{token[-5:]}" if len(token) > 15 else "[hidden]" ) print_json(display_payload, "Request payload") # Start timing the request/call start_time = time.time() response = None error_message = None if function_url: # Test against the provided URL try: headers = {"Content-Type": "application/json"} http_response = requests.post( function_url, headers=headers, json=request, timeout=30 ) # Increased timeout for network calls http_response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx) try: response = http_response.json() except json.JSONDecodeError: error_message = f"Error: Received non-JSON response from URL. Status: {http_response.status_code}, Body: {http_response.text[:200]}..." # Limit body size except requests.exceptions.Timeout: error_message = ( f"Error: Request timed out after 30 seconds when calling {function_url}" ) except requests.exceptions.RequestException as e: error_message = f"Error calling function URL {function_url}: {e}" if e.response is not None: error_message += f"\nStatus Code: {e.response.status_code}\nResponse: {e.response.text[:200]}..." # Limit body size else: # Test locally if mock_context is None: # Create a simple mock context object class MockContext: def __init__(self): self.function_name = "test_function" self.function_version = "$LATEST" self.memory_limit_in_mb = 128 self.aws_request_id = f"test-{int(time.time())}" def get_remaining_time_in_millis(self): # Provide a reasonable remaining time, e.g., 10 seconds return 10000 mock_context = MockContext() # For local testing, ensure we're not using real cold start timing # This makes tests more consistent lambda_function._cold_start_time = ( time.time() - lambda_function._initialization_time ) try: # Make a deep copy of the request to avoid any reference issues request_copy = json.loads(json.dumps(request)) # Call the lambda function directly with the copy response = lambda_function.lambda_handler(request_copy, mock_context) except Exception as e: error_message = f"Error calling local lambda_handler: {e}" logging.exception("Exception details:") # Log traceback for local errors # Calculate and print the elapsed time elapsed_time = time.time() - start_time if error_message: print(f"\nError during test: {error_message}") print(f"Time elapsed before error: {elapsed_time:.2f} sec") return None # Return None to indicate failure # Print summary timing information print_json(response, "Response") # Detailed timing summary # Initialization time is only relevant for local runs print( f"\n{request['header']['name']} Timing Summary ({'URL' if function_url else 'Local'}):" ) if not function_url: print(f" Initialization: {lambda_function._initialization_time:.2f} sec") print(f" Execution: {elapsed_time:.2f} sec") total_time = lambda_function._initialization_time + elapsed_time print( f" Total Time: {total_time:.2f} sec {'<' if total_time <= TEST_TARGET_TIME else '>'} {TEST_TARGET_TIME:.2f} sec target" ) # Show how much time margin remains margin = TEST_TARGET_TIME - total_time if margin >= 0: print(f" Time Margin: {margin:.2f} sec remaining") else: print(f" Time Exceeded: {abs(margin):.2f} sec over budget") else: # For URL tests, just show the request time print(f" Request Time: {elapsed_time:.2f} sec") # Comparison to target time might still be relevant depending on expected network latency print( f" Comparison: {elapsed_time:.2f} sec {'<' if elapsed_time <= TEST_TARGET_TIME else '>'} {TEST_TARGET_TIME:.2f} sec target (excluding network)" ) return response def clear_token_cache(): """Remove the cached token file""" token_file = os.path.expanduser("~/.suno_test_token.json") if os.path.exists(token_file): try: os.remove(token_file) print("Token cache cleared.") return True except Exception as e: print(f"Error removing token cache: {str(e)}") return False else: print("No cached token found.") return True def get_access_token(force_new=False): """Get the access token from a file or through OAuth""" token_file = os.path.expanduser("~/.suno_test_token.json") try: # If force_new is set, skip checking for existing token if not force_new and os.path.exists(token_file): with open(token_file, "r") as f: token_data = json.load(f) access_token = token_data.get("access_token") expires_at = token_data.get("expires_at") # Check if token is still valid if expires_at and datetime.fromisoformat(expires_at) > datetime.now(): print(f"Using existing access token (expires {expires_at})") return access_token else: print("Existing token expired.") else: if force_new: print("Forcing new authentication...") else: print("No valid token found. Starting OAuth flow...") # Get new token access_token = get_oauth_token() if access_token: # Save the token expires_at = (datetime.now() + timedelta(hours=1)).isoformat() with open(token_file, "w") as f: json.dump({"access_token": access_token, "expires_at": expires_at}, f) return access_token else: print("Failed to obtain access token.") return None except Exception as e: print(f"Error handling token: {e}") access_token = get_oauth_token() return access_token def test_resource_urls(initiate_response, download_time=5): """Test URLs for audio, cover art, and lyrics in parallel after a successful Initiate. Args: initiate_response: The response from Initiate download_time: Number of seconds to download from audio stream Returns: dict: Results of the resource tests with status and file paths """ if not initiate_response or "payload" not in initiate_response: console.print("No valid Initiate response to test resources", style="red") return None try: # Create a directory for output files if it doesn't exist output_dir = "resource_tests" os.makedirs(output_dir, exist_ok=True) # Extract the content ID and URLs from the response first_item = ( initiate_response.get("payload", {}) .get("playbackMethod", {}) .get("firstItem", {}) ) content_id = first_item.get("id") if not content_id: console.print("No content ID found in Initiate response", style="red") return None console.print( f"\n[bold cyan]Testing resource URLs for content ID: {content_id}[/bold cyan]" ) console.print( f"Audio stream will be downloaded for {download_time} seconds (use --download-time to change)" ) # Extract the URLs and headers audio_url = first_item.get("stream", {}).get("uri") # Get the first (highest quality) cover art URL art_sources = first_item.get("metadata", {}).get("art", {}).get("sources", []) cover_art_url = art_sources[0].get("url") if art_sources else None lyrics_info = first_item.get("transcript", {}) lyrics_url = lyrics_info.get("uri") if lyrics_info else None # Get any request headers needed for lyrics lyrics_headers = {} if lyrics_info and "headers" in lyrics_info: for header in lyrics_info["headers"]: lyrics_headers[header["name"]] = header["value"] # Define output file paths audio_file = os.path.join(output_dir, f"stream_{content_id}.mp3") cover_file = os.path.join(output_dir, f"image_{content_id}.jpeg") lyrics_file = os.path.join(output_dir, f"lyric_{content_id}.vtt") results = { "content_id": content_id, "audio": {"status": "not_tested", "file": audio_file}, "cover": {"status": "not_tested", "file": cover_file}, "lyrics": {"status": "not_tested", "file": lyrics_file}, } # Define functions for each resource test def test_audio(): if not audio_url: return {"status": "error", "message": "No audio URL found"} try: # Stream for the specified number of seconds and save to file console.print( f"[cyan]Downloading audio stream for {download_time} seconds...[/cyan]" ) start_time = time.time() with requests.get( audio_url, stream=True, timeout=max(30, download_time + 5) ) as r: r.raise_for_status() console.print( f"[cyan]Audio stream connected. Status: {r.status_code}[/cyan]" ) with open(audio_file, "wb") as f: # Download for specified time or until connection closes total_bytes = 0 last_update = 0 for chunk in r.iter_content(chunk_size=8192): elapsed = time.time() - start_time if elapsed > download_time: break if chunk: total_bytes += len(chunk) f.write(chunk) # Show progress every second current_second = int(elapsed) if current_second > last_update: console.print( f"[cyan] Downloaded {total_bytes / 1024:.1f} KB in {elapsed:.1f} sec[/cyan]", end="\r", ) last_update = current_second elapsed = time.time() - start_time file_size = os.path.getsize(audio_file) console.print() # Add a newline after progress updates result = { "status": "success" if file_size > 0 else "error", "file": audio_file, "size": file_size, "message": f"Downloaded {file_size} bytes ({total_bytes / 1024:.1f} KB) in {elapsed:.1f} seconds", } # Print immediate result status_color = "green" if result["status"] == "success" else "red" console.print( f"[bold]Audio[/bold]: [{status_color}]{result['status']}[/{status_color}] - {result['message']}" ) return result except Exception as e: console.print(f"[bold red]Audio stream error: {str(e)}[/bold red]") # Generate and print equivalent curl command for debugging curl_command = f"curl -v '{audio_url}'" console.print("\n[yellow]Debug with this curl command:[/yellow]") console.print(curl_command) return {"status": "error", "message": str(e)} def test_cover_art(): if not cover_art_url: return {"status": "error", "message": "No cover art URL found"} try: console.print("[cyan]Fetching cover art...[/cyan]") start_time = time.time() r = requests.get(cover_art_url, timeout=10) r.raise_for_status() elapsed = time.time() - start_time with open(cover_file, "wb") as f: f.write(r.content) file_size = os.path.getsize(cover_file) result = { "status": "success", "file": cover_file, "size": file_size, "message": f"Downloaded {file_size} bytes in {elapsed:.2f} seconds", } # Print immediate result console.print( f"[bold]Cover[/bold]: [green]success[/green] - {result['message']}" ) return result except Exception as e: console.print(f"[bold red]Cover art error: {str(e)}[/bold red]") return {"status": "error", "message": str(e)} def test_lyrics(): if not lyrics_url: return {"status": "error", "message": "No lyrics URL found"} try: console.print("[cyan]Fetching lyrics...[/cyan]") start_time = time.time() r = requests.get(lyrics_url, headers=lyrics_headers, timeout=10) r.raise_for_status() elapsed = time.time() - start_time content_type = r.headers.get("Content-Type", "unknown") with open(lyrics_file, "wb") as f: f.write(r.content) file_size = os.path.getsize(lyrics_file) result = { "status": "success", "file": lyrics_file, "size": file_size, "message": f"Downloaded {file_size} bytes ({content_type}) in {elapsed:.2f} seconds", } # Print immediate result console.print( f"[bold]Lyrics[/bold]: [green]success[/green] - {result['message']}" ) return result except Exception as e: console.print(f"[bold red]Lyrics error: {str(e)}[/bold red]") # Generate and print equivalent curl command for debugging curl_command = "curl -v" # Add headers for header_name, header_value in lyrics_headers.items(): curl_command += f" -H '{header_name}: {header_value}'" # Add URL (in quotes to handle special characters) curl_command += f" '{lyrics_url}'" console.print("\n[yellow]Debug with this curl command:[/yellow]") console.print(curl_command) return {"status": "error", "message": str(e)} # Execute the tests in parallel console.print("[cyan]Testing all resources in parallel...[/cyan]") with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor: future_audio = executor.submit(test_audio) if audio_url else None future_cover = executor.submit(test_cover_art) if cover_art_url else None future_lyrics = executor.submit(test_lyrics) if lyrics_url else None if future_audio: results["audio"] = future_audio.result() if future_cover: results["cover"] = future_cover.result() if future_lyrics: results["lyrics"] = future_lyrics.result() # Print final summary console.print("\n[bold green]Resource Test Results Summary:[/bold green]") for resource_type, result in results.items(): if resource_type == "content_id": continue status = result.get("status", "unknown") file_path = result.get("file", "N/A") if status == "success": console.print(f" Saved to: {file_path}") return results except Exception as e: console.print(f"\n[bold red]Error testing resource URLs: {str(e)}[/bold red]") return None def process_gpc_response( response, access_token, auto_initiate=0, print_raw_request=False, function_url=None, download_time=5, ): """Process a GetPlayableContent response and optionally auto-initiate playback Args: response: The response from GetPlayableContent access_token: The OAuth access token auto_initiate: Number of seconds to wait before auto-initiating (0 = disabled) print_raw_request: Whether to print the raw request JSON function_url: URL of the Lambda function to test against download_time: Number of seconds to download from audio stream """ if response.get("header", {}).get("name") == "GetPlayableContent.Response": # Updated to handle content as a single object, not a list content = response.get("payload", {}).get("content", {}) # Check if content has an ID content_id = content.get("id") if content_id: console.print(f"\nContent ID: {content_id}") # If auto-initiate is enabled (positive seconds value), automatically call Initiate after waiting if auto_initiate > 0: # Display countdown with Rich spinner and styling with Progress( SpinnerColumn(), TextColumn("[cyan]Waiting to initiate...[/cyan]"), TimeRemainingColumn(), transient=True, ) as progress: task = progress.add_task( "", total=auto_initiate, remaining=auto_initiate ) for remaining in range(auto_initiate, 0, -1): progress.update(task, advance=1, remaining=remaining) time.sleep(1) initiate_request = create_initiate_request(access_token, content_id) # Reset cold start timer for this test lambda_function._cold_start_time = ( time.time() - lambda_function._initialization_time ) # This is a separate Initiate execution initiate_response = test_lambda( initiate_request, print_raw_request=print_raw_request, function_url=function_url, ) # Test resource URLs if initiate was successful if ( initiate_response and initiate_response.get("header", {}).get("name") == "Initiate.Response" ): test_resource_urls(initiate_response, download_time=download_time) return initiate_response else: # Otherwise ask for user confirmation test_initiate = ( input("Test Initiate with this content? (y/n): ").lower() == "y" ) if test_initiate: # This is a separate execution - starts timing here initiate_request = create_initiate_request(access_token, content_id) # Reset cold start timer for this test lambda_function._cold_start_time = ( time.time() - lambda_function._initialization_time ) initiate_response = test_lambda( initiate_request, print_raw_request=print_raw_request, function_url=function_url, ) # Test resource URLs if initiate was successful if ( initiate_response and initiate_response.get("header", {}).get("name") == "Initiate.Response" ): test_resource_urls_option = ( input( "Test resource URLs (audio, cover art, lyrics)? (y/n): " ).lower() == "y" ) if test_resource_urls_option: test_resource_urls( initiate_response, download_time=download_time ) return initiate_response return None def create_gdc_request(access_token, genre="pop"): """Create a GetDisplayableContent request with optional genre filter""" return { "header": { "messageId": f"test-{int(time.time())}", "namespace": "Alexa.Media.Search", "name": "GetDisplayableContent", "payloadVersion": "3.0" }, "payload": { "requestContext": { "user": { "id": "7d2fa43ade80f308f0d761944f60f391", "accessToken": access_token }, "location": { "originatingLocale": "en-US", "countryCode": "US" }, "advertising": { "limitAdTracking": True, "advertisingId": "00000000-0000-0000-0000-000000000000" }, "apiAccessToken": None, "apiEndpoint": None }, "filters": { "explicitLanguageAllowed": False }, "policies": None, "endpoints": None, "rawText": None, "maxResultLimit": 8, "playQueuePreviewCriteria": None, "rankedSelectionCriteria": [{ "type": "ATTRIBUTES", "id": "MULTI_ATTR_CRITERIA", "attributes": [{ "type": "GENRE", "id": "YW1iaWVudA==", # Base64 for "ambient", but we'll use rawValue "rawValue": genre, "resolvedEntities": [] }], "completeness": { "score": 1.0, "bin": "HIGH" }, "query": None }], "paginationContext": { "paginationContextType": None, "limits": { "contentGroupsLimit": { "maxLimit": 0, "canPaginate": False }, "contentItemsPerContentListLimit": { "maxLimit": 8, "canPaginate": False } } } } } def create_gdc_search_request(access_token, search_term="pop"): """Create a GetDisplayableContent request with natural language search query""" return { "header": { "messageId": f"test-{int(time.time())}", "namespace": "Alexa.Media.Search", "name": "GetDisplayableContent", "payloadVersion": "3.0" }, "payload": { "requestContext": { "user": { "id": "amzn1.ask.account.AMAXEMEU6ORSTK6P5ZBQP3FKC7AQUNSG7QLYEXBYAZTO2RQIRZ5UZGR7FWB2D7SR3WR6Q4SGATPADCQ7KGP5AFRD2VVUOZVGL2GKRG7ZYEEULWLEF7RMJOJWKGI57OD266FD3SCBG6KN2AXVX25YXEYTUEZBANSCKVB5LRYA7VHBIRTM3ZDLOO3NKQBZ6PSIIJ5KNQ5LPPYOZXLVINCIDNKW7DMRGRWIJHDLWIJ55ERA", "accessToken": access_token }, "location": { "originatingLocale": "en-US", "countryCode": "US" }, "advertising": { "limitAdTracking": True, "advertisingId": "00000000-0000-0000-0000-000000000000" }, "apiAccessToken": None, "apiEndpoint": None }, "filters": { "explicitLanguageAllowed": False }, "policies": None, "endpoints": None, "rawText": None, "maxResultLimit": 8, "playQueuePreviewCriteria": None, "rankedSelectionCriteria": [ { "type": "NL_QUERY", "id": "NL_QUERY_CRITERIA", "query": f"find the {search_term} music i created on suno" } ], "paginationContext": { "paginationContextType": None, "limits": { "contentGroupsLimit": { "maxLimit": 0, "canPaginate": False }, "contentItemsPerContentListLimit": { "maxLimit": 8, "canPaginate": False } } } } } def create_gdc_my_music_request(access_token): """Create a GetDisplayableContent request for browsing 'my music' created on Suno""" return { "header": { "messageId": f"test-{int(time.time())}", "namespace": "Alexa.Media.Search", "name": "GetDisplayableContent", "payloadVersion": "3.0" }, "payload": { "requestContext": { "user": { "id": "amzn1.ask.account.AMAXEMEU6ORSTK6P5ZBQP3FKC7AQUNSG7QLYEXBYAZTO2RQIRZ5UZGR7FWB2D7SR3WR6Q4SGATPADCQ7KGP5AFRD2VVUOZVGL2GKRG7ZYEEULWLEF7RMJOJWKGI57OD266FD3SCBG6KN2AXVX25YXEYTUEZBANSCKVB5LRYA7VHBIRTM3ZDLOO3NKQBZ6PSIIJ5KNQ5LPPYOZXLVINCIDNKW7DMRGRWIJHDLWIJ55ERA", "accessToken": access_token }, "location": { "originatingLocale": "en-US", "countryCode": "US" }, "advertising": { "limitAdTracking": True, "advertisingId": "00000000-0000-0000-0000-000000000000" }, "apiAccessToken": None, "apiEndpoint": None }, "filters": { "explicitLanguageAllowed": False }, "policies": None, "endpoints": None, "rawText": None, "maxResultLimit": 8, "playQueuePreviewCriteria": None, "rankedSelectionCriteria": [ { "type": "NL_QUERY", "id": "NL_QUERY_CRITERIA", "query": "find the music i created on suno" }, { "type": "ATTRIBUTES", "id": "MULTI_ATTR_CRITERIA", "attributes": [ { "type": "MEDIA_TYPE", "id": "bXVzaWM=", "rawValue": "music", "value": "TRACK" } ], "completeness": { "score": 1, "bin": "HIGH" }, "query": None } ], "paginationContext": { "paginationContextType": None, "limits": { "contentGroupsLimit": { "maxLimit": 0, "canPaginate": False }, "contentItemsPerContentListLimit": { "maxLimit": 8, "canPaginate": False } } } } } def test_compound_song_flow(access_token, query="Create a song about a cat", print_raw_request=False, function_url=None, download_time=5): """Run a compound test that creates a song, initiates playback, and tests navigation""" console.print("\n[bold cyan]Starting compound song creation and navigation test[/bold cyan]") console.print(f"Creating song with prompt: {query}") # Step 1: Create song with GetPlayableContent gpc_request = create_gpc_request(access_token, query) lambda_function._cold_start_time = time.time() - lambda_function._initialization_time gpc_response = test_lambda( gpc_request, print_raw_request=print_raw_request, function_url=function_url ) if not gpc_response or gpc_response.get("header", {}).get("name") != "GetPlayableContent.Response": console.print("[bold red]Failed at step 1: GetPlayableContent failed[/bold red]") return False # Extract content ID from GPC response content = gpc_response.get("payload", {}).get("content", {}) content_id = content.get("id") if not content_id: console.print("[bold red]Failed at step 1: No content ID in GetPlayableContent response[/bold red]") return False console.print(f"[green]✓ Step 1: Song created with content ID: {content_id}[/green]") # Step 2: Initiate playback console.print("\n[cyan]Step 2: Initiating playback...[/cyan]") # Wait 5 seconds for generation to complete with Progress( SpinnerColumn(), TextColumn("[cyan]Waiting for generation to complete...[/cyan]"), TimeRemainingColumn(), transient=True, ) as progress: task = progress.add_task("", total=5, remaining=5) for remaining in range(5, 0, -1): progress.update(task, advance=1, remaining=remaining) time.sleep(1) initiate_request = create_initiate_request(access_token, content_id) lambda_function._cold_start_time = time.time() - lambda_function._initialization_time initiate_response = test_lambda( initiate_request, print_raw_request=print_raw_request, function_url=function_url ) if not initiate_response or initiate_response.get("header", {}).get("name") != "Initiate.Response": console.print("[bold red]Failed at step 2: Initiate failed[/bold red]") return False # Extract clip ID and queue ID from Initiate response playback_method = initiate_response.get("payload", {}).get("playbackMethod", {}) first_item = playback_method.get("firstItem", {}) clip_id = first_item.get("id") queue_id = playback_method.get("id") if not clip_id or not queue_id: console.print("[bold red]Failed at step 2: No clip ID or queue ID in Initiate response[/bold red]") return False console.print(f"[green]✓ Step 2: Playback initiated with clip ID: {clip_id}[/green]") # Optionally test resource URLs test_resource_urls(initiate_response, download_time=download_time) # Step 3: Test GetNextItem console.print("\n[cyan]Step 3: Testing GetNextItem to verify second version...[/cyan]") get_next_item_request = create_get_next_item_request(access_token, clip_id, queue_id) lambda_function._cold_start_time = time.time() - lambda_function._initialization_time next_item_response = test_lambda( get_next_item_request, print_raw_request=print_raw_request, function_url=function_url ) if not next_item_response or next_item_response.get("header", {}).get("name") != "GetNextItem.Response": console.print("[bold red]Failed at step 3: GetNextItem failed[/bold red]") return False next_item = next_item_response.get("payload", {}).get("item", {}) next_item_id = next_item.get("id") if not next_item_id: console.print("[bold red]Failed at step 3: No item ID in GetNextItem response[/bold red]") return False # Extract metadata to verify it's actually a second version of the same song # Get original song metadata original_metadata = first_item.get("metadata", {}) # Title is in metadata.name.display original_title = original_metadata.get("name", {}).get("display", "Unknown") # Artist might be in a different location depending on the response structure original_artist = original_metadata.get("artist", "Unknown") # Get next item metadata next_metadata = next_item.get("metadata", {}) next_title = next_metadata.get("name", {}).get("display", "Unknown") next_artist = next_metadata.get("artist", "Unknown") # Check if items are different (different versions) if next_item_id == clip_id: console.print("[bold yellow]Warning: GetNextItem returned the same clip ID[/bold yellow]") else: console.print(f"[green]✓ Step 3: GetNextItem returned different clip ID: {next_item_id}[/green]") # Compare metadata to check if it's a second version of the same song console.print("\n[cyan]Comparing metadata between original and next item:[/cyan]") console.print(f"Original: Title='{original_title}', Artist='{original_artist}'") console.print(f"Next Item: Title='{next_title}', Artist='{next_artist}'") if original_title == next_title: console.print("[green]✓ Titles match - confirmed to be second version of same song[/green]") else: console.print("[bold yellow]Warning: Titles don't match - next item may not be second version of the same song[/bold yellow]") console.print("[yellow]This may indicate an issue with song pairs in the backend[/yellow]") # Step 4: Test GetPreviousItem console.print("\n[cyan]Step 4: Testing GetPreviousItem to verify navigation to older song...[/cyan]") get_previous_item_request = create_get_previous_item_request(access_token, clip_id, queue_id) lambda_function._cold_start_time = time.time() - lambda_function._initialization_time previous_item_response = test_lambda( get_previous_item_request, print_raw_request=print_raw_request, function_url=function_url ) if not previous_item_response or previous_item_response.get("header", {}).get("name") != "GetPreviousItem.Response": console.print("[bold red]Failed at step 4: GetPreviousItem failed[/bold red]") return False # Check if there is a previous item previous_item = previous_item_response.get("payload", {}).get("item") # Handle the case where there's no previous item (which is a valid response) if previous_item is None: is_queue_finished = previous_item_response.get("payload", {}).get("isQueueFinished", False) if is_queue_finished: console.print("[yellow]No previous item available (isQueueFinished: True)[/yellow]") else: console.print("[yellow]No previous item available, but queue is not marked as finished[/yellow]") console.print("[green]✓ Step 4: GetPreviousItem returned a valid response with no previous item[/green]") else: # There is a previous item, check its ID previous_item_id = previous_item.get("id") if not previous_item_id: console.print("[bold red]Failed at step 4: No item ID in GetPreviousItem response[/bold red]") return False if previous_item_id == clip_id: console.print("[bold yellow]Warning: GetPreviousItem returned the same clip ID[/bold yellow]") else: console.print(f"[green]✓ Step 4: GetPreviousItem returned different clip ID: {previous_item_id}[/green]") console.print("\n[bold green]Compound test completed successfully![/bold green]") return True def test_repeat_flow(access_token, print_raw_request=False, function_url=None): """ Run a compound test for Initiate -> SetLoop -> GetNextItem (repeat ON) -> SetLoop -> GetNextItem (repeat OFF). """ console.print("\n[bold cyan]Starting compound test for Repeat functionality[/bold cyan]") # Step 1: Create a song to get a content_id console.print("\n[cyan]Step 1: Creating a song to get a content ID...[/cyan]") gpc_request = create_gpc_request(access_token, "A song to test repeat mode") gpc_response = test_lambda( gpc_request, print_raw_request=print_raw_request, function_url=function_url ) if not gpc_response or gpc_response.get("header", {}).get("name") != "GetPlayableContent.Response": console.print("[bold red]Failed at step 1: GetPlayableContent failed[/bold red]") return False content_id = gpc_response.get("payload", {}).get("content", {}).get("id") if not content_id: console.print("[bold red]Failed at step 1: No content ID in GPC response[/bold red]") return False console.print(f"[green]✓ Step 1: Song created with content ID: {content_id}[/green]") # Wait for generation console.print("[cyan]Waiting 5 seconds for song generation to settle...[/cyan]") time.sleep(5) # Step 2: Initiate playback, which sets initial repeat status to OFF console.print("\n[cyan]Step 2: Initiating playback (initial repeat=OFF)...[/cyan]") initiate_request = create_initiate_request(access_token, content_id) initiate_response = test_lambda( initiate_request, print_raw_request=print_raw_request, function_url=function_url ) if not initiate_response or initiate_response.get("header", {}).get("name") != "Initiate.Response": console.print("[bold red]Failed at step 2: Initiate failed[/bold red]") return False playback_method = initiate_response.get("payload", {}).get("playbackMethod", {}) first_item = playback_method.get("firstItem", {}) original_clip_id = first_item.get("id") queue_id = playback_method.get("id") if not original_clip_id or not queue_id: console.print("[bold red]Failed at step 2: No clip ID or queue ID in Initiate response[/bold red]") return False console.print(f"[green]✓ Step 2: Playback initiated for clip ID: {original_clip_id} in queue: {queue_id}[/green]") # Step 3: Send SetLoop to turn ON repeat mode console.print("\n[cyan]Step 3: Setting loop mode to ON (via SetLoop)...[/cyan]") set_loop_on_request = create_set_loop_request(access_token, original_clip_id, queue_id, enable=True) set_loop_on_response = test_lambda( set_loop_on_request, print_raw_request=print_raw_request, function_url=function_url ) # A simple success response is enough if not set_loop_on_response or "header" not in set_loop_on_response: console.print("[bold red]Failed at step 3: SetLoop(enable=True) request failed[/bold red]") return False console.print("[green]✓ Step 3: SetLoop(enable=True) sent successfully.[/green]") # Step 4: System-initiated GetNextItem with REPEAT=ON console.print("\n[cyan]Step 4: Testing system-initiated GetNextItem (expecting same song)...[/cyan]") gni_repeat_on_request = create_get_next_item_request( access_token, original_clip_id, queue_id, is_user_initiated=False ) gni_repeat_on_response = test_lambda( gni_repeat_on_request, print_raw_request=print_raw_request, function_url=function_url ) repeated_item = gni_repeat_on_response.get("payload", {}).get("item", {}) repeated_clip_id = repeated_item.get("id") if repeated_clip_id == original_clip_id: console.print(f"[green]✓ Step 4: Success! Received same clip ID: {repeated_clip_id}[/green]") else: console.print(f"[bold red]Failed at step 4: Expected clip ID {original_clip_id} but got {repeated_clip_id}[/bold red]") return False # Step 5: Send SetLoop to turn OFF repeat mode console.print("\n[cyan]Step 5: Setting loop mode to OFF (via SetLoop)...[/cyan]") set_loop_off_request = create_set_loop_request(access_token, original_clip_id, queue_id, enable=False) set_loop_off_response = test_lambda( set_loop_off_request, print_raw_request=print_raw_request, function_url=function_url ) if not set_loop_off_response or "header" not in set_loop_off_response: console.print("[bold red]Failed at step 5: SetLoop(enable=False) request failed[/bold red]") return False console.print("[green]✓ Step 5: SetLoop(enable=False) sent successfully.[/green]") # Step 6: System-initiated GetNextItem with REPEAT=OFF console.print("\n[cyan]Step 6: Testing system-initiated GetNextItem (expecting next song)...[/cyan]") gni_repeat_off_request = create_get_next_item_request( access_token, original_clip_id, queue_id, is_user_initiated=False ) gni_repeat_off_response = test_lambda( gni_repeat_off_request, print_raw_request=print_raw_request, function_url=function_url ) next_item = gni_repeat_off_response.get("payload", {}).get("item", {}) next_clip_id = next_item.get("id") if next_clip_id and next_clip_id != original_clip_id: console.print(f"[green]✓ Step 6: Success! Received a different clip ID: {next_clip_id}[/green]") elif not next_clip_id: # This is also a valid outcome if there is no next song in the queue is_finished = gni_repeat_off_response.get("payload", {}).get("isQueueFinished", False) if is_finished: console.print(f"[green]✓ Step 6: Success! No next item returned and queue is finished.[/green]") else: console.print("[bold red]Failed at step 6: GetNextItem did not return an item, but queue not finished.[/bold red]") return False else: console.print(f"[bold red]Failed at step 6: Expected a different clip ID but got the same one: {next_clip_id}[/bold red]") return False console.print("\n[bold green]Repeat functionality compound test completed successfully![/bold green]") return True def interactive_mode( access_token, auto_initiate=0, print_raw_request=False, function_url=None, download_time=5, ): """Run in interactive mode, allowing the user to select what to test""" # Ask the user what to test print("\nWhat would you like to test?") print("1. Generate a song with a custom prompt") print("2. Play my most recent song") print("3. Initiate playback with a specific song ID") print("4. Get next song in queue") print("5. Get previous song in queue") print("6. Browse my library (GetDisplayableContent)") print("7. Search my library (GetDisplayableContent)") print("8. Full flow test (Create -> Initiate -> Nav)") print("9. Repeat mode test (Initiate -> GetNextItem)") choice = input("Enter choice (1-9): ") if choice == "1": # Test GetPlayableContent with query query = ( input("Enter song prompt (or press Enter for default): ") or "Create a song about a cat" ) gpc_request = create_gpc_request(access_token, query) response = test_lambda( gpc_request, print_raw_request=print_raw_request, function_url=function_url ) process_gpc_response( response, access_token, auto_initiate, print_raw_request, function_url, download_time ) elif choice == "2": # Test "Play my music" request = create_gpc_play_my_music_request(access_token) response = test_lambda( request, print_raw_request=print_raw_request, function_url=function_url ) process_gpc_response( response, access_token, auto_initiate, print_raw_request, function_url, download_time ) elif choice == "3": # Test with initiate_specific_content.json content_id = input("Enter the Content ID to initiate: ") if content_id: request = create_initiate_request(access_token, content_id) initiate_response = test_lambda( request, print_raw_request=print_raw_request, function_url=function_url ) test_resource_urls(initiate_response, download_time=download_time) else: print("Content ID cannot be empty.") elif choice == "4": # Test GetNextItem clip_id = input("Enter the current clip ID: ") queue_id = input("Enter the current queue ID: ") if clip_id and queue_id: request = create_get_next_item_request(access_token, clip_id, queue_id) response = test_lambda( request, print_raw_request=print_raw_request, function_url=function_url ) process_gpc_response( response, access_token, auto_initiate, print_raw_request, function_url, download_time ) else: print("Clip ID and Queue ID cannot be empty.") elif choice == "5": # Test GetPreviousItem clip_id = input("Enter the current clip ID: ") queue_id = input("Enter the current queue ID: ") if clip_id and queue_id: request = create_get_previous_item_request(access_token, clip_id, queue_id) response = test_lambda( request, print_raw_request=print_raw_request, function_url=function_url ) process_gpc_response( response, access_token, auto_initiate, print_raw_request, function_url, download_time ) else: print("Clip ID and Queue ID cannot be empty.") elif choice == "6": # Test GetDisplayableContent with optional genre use_genre = input("Filter by genre? (y/n): ").lower() == "y" if use_genre: genre = input("Enter genre (or press Enter for default 'pop'): ") or "pop" print(f"Using genre filter: {genre}") gdc_request = create_gdc_request(access_token, genre) else: # Create an empty GDC request (no criteria) gdc_request = create_gdc_request(access_token) gdc_request["payload"]["rankedSelectionCriteria"] = [] print("Using empty browse (no genre filter)") response = test_lambda( gdc_request, print_raw_request=print_raw_request, function_url=function_url ) print_json(response, "GetDisplayableContent response") elif choice == "7": # Test GetDisplayableContent search search_term = input("Enter search term (or press Enter for default 'pop'): ") or "pop" print(f"Searching for '{search_term}'") gdc_request = create_gdc_search_request(access_token, search_term) response = test_lambda( gdc_request, print_raw_request=print_raw_request, function_url=function_url ) print_json(response, "GetDisplayableContent Search response") elif choice == "8": # Compound test: Create song -> Initiate -> GetNextItem -> GetPreviousItem query = ( input("Enter song prompt (or press Enter for default): ") or "Create a song about a cat" ) test_compound_song_flow( access_token, query=query, print_raw_request=print_raw_request, function_url=function_url, download_time=download_time ) elif choice == "9": # Repeat mode test test_repeat_flow( access_token, print_raw_request=print_raw_request, function_url=function_url ) else: print("Invalid choice.") def test_generate_endpoint(access_token): """Test the generate endpoint directly to diagnose issues""" print("\nTesting generate endpoint directly...") try: # Prepare the request data data = { "topic": "Test song - diagnosing API issue", # The API expects 'topic' as the main field "model": "chirp-v4-h-api", "extra": {"add_to_playlist": True}, } # Set up headers with the access token headers = { "Authorization": f"Bearer {access_token}", "Content-Type": "application/json", "x-auth-type": "oauth", # Required by Suno API } print_json(data, "Request data") print(f"Authorization: Bearer {access_token[:10]}...{access_token[-5:]}") # Make the request to Suno API response = requests.post( "https://studio-api.prod.suno.com/api/v2/external/oauth/generate", headers=headers, json=data, timeout=10, ) # Print complete response details print(f"Status code: {response.status_code}") print(f"Response headers: {dict(response.headers)}") try: json_response = response.json() print_json(json_response, "Response JSON") except json.JSONDecodeError: print("Response is not valid JSON:") print(response.text) return response.status_code == 200 except Exception as e: print(f"Error during generate test: {str(e)}") if hasattr(e, "response") and e.response: print(f"Error response: {e.response.text}") return False def main(): parser = argparse.ArgumentParser( description="Test the Lambda function with requests" ) parser.add_argument( "--auto-initiate", "-a", type=int, nargs="?", # Make it optional const=5, # Default value when flag is specified without a value default=5, # Default when flag is not specified at all (changed from 0 to 5) metavar="SECONDS", help="Automatically call Initiate after successful GetPlayableContent with SECONDS delay (default: 5 seconds)", ) parser.add_argument("--output", "-o", help="Save response to a file") parser.add_argument( "--raw-request", "-r", action="store_true", help="Print the raw request JSON for AWS Lambda console testing", ) parser.add_argument( "--clear-token", "-c", action="store_true", help="Clear the token cache file to force re-authentication", ) parser.add_argument( "--log-level", "-l", choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], default="INFO", help="Set the logging level (default: INFO)", ) parser.add_argument( "--timing-breakdown", "-b", action="store_true", help="Show detailed timing breakdown", ) parser.add_argument( "--url", help="URL of the deployed Lambda function to test against" ) parser.add_argument( "--clip-id", help="Test GetPlayableContent with a specific clip ID" ) parser.add_argument( "--test-resources", "-t", action="store_true", help="Automatically test resource URLs after successful Initiate", ) parser.add_argument( "--test-response-file", help="Test resource URLs from a saved Initiate response file", ) parser.add_argument( "--download-time", type=int, default=5, help="Number of seconds to download from audio stream (default: 5)", ) parser.add_argument( "--prompt", type=str, help="Generate a song with the specified prompt", ) args = parser.parse_args() # Set log level based on argument log_level = getattr(logging, args.log_level) logging.getLogger().setLevel(log_level) lambda_function.logger.setLevel(log_level) # Special case: If testing resources from a file, do that first and exit if args.test_response_file: try: with open(args.test_response_file, "r") as f: response_data = json.load(f) print(f"Testing resources from file: {args.test_response_file}") test_resource_urls(response_data, args.download_time) return 0 except Exception as e: print(f"Error testing resources from file: {e}") return 1 # Clear token cache if requested if args.clear_token: clear_token_cache() # Get access token - this happens before timing starts access_token = get_access_token(force_new=args.clear_token) # If no valid token, offer to clear and retry if not access_token: print("\nFailed to obtain a valid access token.") retry = ( input( "Would you like to clear the token cache and try again? (y/n): " ).lower() == "y" ) if retry: clear_token_cache() access_token = get_access_token(force_new=True) if not access_token: print("Still unable to obtain a valid token. Exiting.") return 1 else: print("Exiting.") return 1 # Test with specific clip ID if provided if args.clip_id: print(f"\nTesting GetPlayableContent with clip ID: {args.clip_id}") # Get the template request gpc_request = create_simple_gpc_request(access_token) # Replace the placeholder ID with the real clip ID criteria = gpc_request["payload"]["rankedSelectionCriteria"][0] entity = criteria["attributes"][0]["resolvedEntities"][0] # Update the entity ID original_id = entity["entityId"] entity["entityId"] = args.clip_id print(f"Updated entity ID from {original_id} to {args.clip_id}") # Reset initialization time to simulate a fresh Lambda environment lambda_function._cold_start_time = ( time.time() - lambda_function._initialization_time ) response = test_lambda( gpc_request, print_raw_request=args.raw_request, function_url=args.url ) # Process the response - this may include initiate request process_gpc_response( response, access_token, args.auto_initiate, args.raw_request, args.url, args.download_time, ) return 0 # Test with specific prompt if provided elif args.prompt: print(f"\nGenerating song with prompt: {args.prompt}") # Create GPC request with the provided prompt gpc_request = create_gpc_request(access_token, args.prompt) # Reset initialization time lambda_function._cold_start_time = ( time.time() - lambda_function._initialization_time ) # Execute the request response = test_lambda( gpc_request, print_raw_request=args.raw_request, function_url=args.url ) # Process the response process_gpc_response( response, access_token, args.auto_initiate, args.raw_request, args.url, args.download_time, ) return 0 else: # Run in interactive mode interactive_mode( access_token, args.auto_initiate, args.raw_request, args.url, args.download_time, ) return 0 if __name__ == "__main__": sys.exit(main())