from typing import Dict, Any, Optional, List, Union import requests import json import time import logging import threading import random import functools import re from constants import GENRE_LIST from instrumental_determinator import song_is_vocal from anodyne import generate_song_title from error_models import ( SkillError, AuthorizationError, RateLimitError, ContentFilteredError, ) logger = logging.getLogger() # Suno API endpoint Base SUNO_BASE_API_URL = "https://studio-api.prod.suno.com" # Specific API Paths CLIPS_PATH = "/api/v2/external/oauth/clips" PLAY_QUEUE_PATH = "/api/v2/external/oauth/search-user-clips" # Suno CDN URLs SUNO_AUDIO_ENDPOINT = "https://audiopipe.suno.ai" SUNO_AUDIO_CDN = "https://cdn1.suno.ai" SUNO_IMAGE_CDN = "https://cdn2.suno.ai" SUNO_REQUEST_EVENTS_ENDPOINT = "https://audiopipe.suno.ai/request_events/" SUNO_CLIP_EVENTS_ENDPOINT = "https://audiopipe.suno.ai/clip_events/" # Generation specific API SUNO_GENERATE_PATH = "/api/v2/external/oauth/generate" # Added for Generate action # Timeouts and Settings SUNO_API_TIMEOUT = 4.5 MAX_HANDLER_TIME_BUDGET_SECONDS = 4.5 TARGET_GENERATION_TIME_SECONDS = 3.5 # Target time for generation requests SSE_SAFETY_BUFFER_SECONDS = 0.1 # Small safety buffer for return SSE_TIMEOUT = 4.5 # SSE Max Timeout Settings SSE_DEFAULT_MAX_TIMEOUT = 3.5 SSE_SONG_GENERATION_MAX_TIMEOUT = 4.0 SSE_TITLE_EXTRACTION_MAX_TIMEOUT = 3.5 SSE_INITIAL_WAIT_TIME = 0.5 # Initial wait time before checking for early return # ============================================================================= # Shared API utility functions # ============================================================================= def handle_moderation_error(error_type: str, error_message: str): """ Handles moderation errors consistently, raising the appropriate exception. Args: error_type: The type of error from the SSE stream error_message: The error message from the SSE stream Raises: ContentFilteredError: For content moderation issues SkillError: For other types of errors """ if error_type != "moderation_failure": # For non-moderation errors, raise standard SkillError raise SkillError( error_message, error_code="INTERNAL_ERROR", ) # Handle different types of moderation failures logger.warning(f"Content moderation failure: {error_message}") if error_message == "Song Description flagged for moderation": # Explicit language filter case raise ContentFilteredError(error_message, subtypes=["EXPLICIT_LANGUAGE_FILTER"]) elif "Please try rephrasing with more specific details" in error_message: # This is a context mismatch, not a content filter raise SkillError(error_message, error_code="USER_CONTEXT_MISMATCH") elif "contained artist name:" in error_message: # Artist name mention - content filtered without subtypes raise ContentFilteredError(error_message, subtypes=None) else: # Other moderation failures - content filtered without subtypes raise SkillError(error_message, error_code="CONTENT_FILTERED") def create_auth_headers(access_token: str) -> Dict[str, str]: """ Create standardized authorization headers used by Suno API requests. Args: access_token: OAuth access token Returns: Dictionary of HTTP headers """ return { "Authorization": f"Bearer {access_token}", "Content-Type": "application/json", "x-auth-type": "oauth", } def handle_api_errors(operation_name: str = "API request"): """ Decorator that standardizes error handling for Suno API requests. Args: operation_name: Description of the operation for logging Returns: Decorated function with standardized error handling """ def decorator(func): def wrapper(*args, **kwargs): try: return func(*args, **kwargs) except requests.exceptions.Timeout as e: logger.warning(f"Timeout during {operation_name}: {str(e)}") raise # Re-raise Timeout to be handled by caller except requests.exceptions.RequestException as e: # Log more details if it's an HTTPError with a response if hasattr(e, "response") and e.response is not None: logger.error( f"HTTP Error during {operation_name}: " f"Status={e.response.status_code}, Reason={e.response.reason}, " f"URL={e.response.url}, Response Body: {e.response.text[:500]}" ) else: logger.error(f"Request Error during {operation_name}: {str(e)}") raise # Re-raise RequestException to be handled by caller except (json.JSONDecodeError, ValueError) as e: # Catch JSON parsing errors or the specific ValueError logger.error( f"Error processing response for {operation_name}: {str(e)}" ) raise # Re-raise error to be handled by caller except Exception as e: logger.error( f"Unexpected error during {operation_name}: {str(e)}", exc_info=True ) raise return wrapper return decorator # ============================================================================= # Clip Details Fetcher (Used by Initiate) # ============================================================================= @handle_api_errors("clip details fetch") def get_clip_details(access_token: str, clip_id: str) -> Dict[str, Any]: """ Fetch details for a specific clip from the Suno API. Can raise ValueError, requests.exceptions.Timeout, requests.exceptions.RequestException. """ headers = create_auth_headers(access_token) url = f"{SUNO_BASE_API_URL}{CLIPS_PATH}?ids={clip_id}" logger.debug(f"Fetching clip details from: {url}") response = requests.get( url, headers=headers, timeout=3.0 ) # Increase timeout from 1.0 to 3.0 response.raise_for_status() # Raises HTTPError for bad responses (4xx or 5xx) clips = response.json() # Raises JSONDecodeError for invalid JSON if not clips: # Raise specific error if clip not found by API, allows caller to distinguish raise ValueError(f"No clip found with ID {clip_id} via Suno API") logger.debug(f"Clip details retrieved successfully for {clip_id}") return clips[0] @handle_api_errors("playlist item fetch") def get_item_in_queue( access_token: str, clip_id: str, handler_start_time: float, is_next_item: bool, limit: int = 1, return_list: bool = False, ) -> Dict[str, Any]: """ Fetch details for a clip in the user's playlist from the Suno API. Can raise ValueError, requests.exceptions.Timeout, requests.exceptions.RequestException. Args: access_token: User authentication token clip_id: ID of the current clip handler_start_time: Start time of the handler for latency tracking is_next_item: If True, fetch the next (older) item; if False, fetch the previous (newer) item limit: Number of items to fetch (default 1) return_list: If True, return the entire list of clips; if False, return only the first clip (default False) """ headers = create_auth_headers(access_token) query_criteria = "before_id" if is_next_item else "after_id" print(f"Access token: {access_token}") url = ( f"{SUNO_BASE_API_URL}{PLAY_QUEUE_PATH}?{query_criteria}={clip_id}&limit={limit}" ) if not clip_id: url = f"{SUNO_BASE_API_URL}{PLAY_QUEUE_PATH}?limit={limit}" logger.debug(f"Fetching clip details from: {url}") response = requests.get( url, headers=headers, timeout=3.0 ) # Increase timeout from 1.0 to 3.0 response.raise_for_status() # Raises HTTPError for bad responses (4xx or 5xx) clips = response.json() # Raises JSONDecodeError for invalid JSON print(f"response: {clips}") if not clips: # Raise specific error if clip not found by API, allows caller to distinguish return None logger.debug(f"Clip details retrieved successfully for {clip_id}") if len(clips) > 0: # Return either the full list or just the first clip based on return_list parameter return clips if return_list else clips[0] else: return None # ============================================================================= # SSE Helper Functions # ============================================================================= def connect_to_sse_stream( url: str, process_event_callback, timeout: float, initial_result: Dict[str, Any] = None, max_timeout: float = SSE_DEFAULT_MAX_TIMEOUT, ) -> Dict[str, Any]: """ Generic function to connect to an SSE stream and extract data using a callback. Args: url: The SSE endpoint URL process_event_callback: Function that processes each event and updates the result timeout: Maximum time allowed for SSE processing initial_result: Initial result dictionary with any required keys max_timeout: Maximum allowed timeout (defaults to SSE_DEFAULT_MAX_TIMEOUT) Returns: Dictionary with extracted data and status information """ # Ensure timeout is not negative and enforce a specified maximum timeout = max(0, min(timeout, max_timeout)) if timeout <= 0.01: # Use a small threshold logger.warning("Skipping SSE connection due to zero remaining time budget.") return {"finished": False} # Use a thread-safe container for the result, starting with initial values if provided result = { "finished": False, "events_received": 0, "event_types_seen": set(), "connection_established": False, } if initial_result: result.update(initial_result) # Keep track of error count to avoid excessive logging error_count = {"total": 0, "logged": 0} # Use a shared variable to hold the response object for proper cleanup response_container = {"response": None, "session": None} # Event for signaling when connection is established connection_established = threading.Event() # Event for signaling when processing should stop stop_event = threading.Event() # Create a requests session with appropriate timeout values session = requests.Session() session.request = functools.partial(session.request, timeout=(1.0, timeout)) response_container["session"] = session def sse_worker(): """Worker thread to handle the SSE connection and extract data.""" response = None try: response = session.get(url, stream=True, timeout=(1.0, timeout)) response_container["response"] = response # Store for cleanup # Signal that connection has been established (successful or not) connection_established.set() # Log connection status only if there's an issue if response.status_code != 200: logger.warning(f"SSE connection established with unexpected status code: {response.status_code}") response.raise_for_status() buffer = "" for line in response.iter_lines(decode_unicode=True): # Check if stop was requested before processing more data if stop_event.is_set(): break if not line: # Empty line signifies end of an event if buffer: try: event_type = None event_data_raw = None for part in buffer.split("\n"): if part.startswith("event:"): event_type = part[6:].strip() elif part.startswith("data:"): event_data_raw = part[5:].strip() if event_type and event_data_raw: result["events_received"] += 1 # Count events result["event_types_seen"].add(event_type) # Track which types we've seen try: # Call the callback with proper error handling should_stop = process_event_callback(result, event_type, event_data_raw) if should_stop: return # Stop processing if callback returns True except Exception as e: # Count errors but limit logging to avoid flooding error_count["total"] += 1 if error_count["total"] <= 3 or error_count["total"] % 10 == 0: logger.warning(f"Error in callback processing {event_type} event: {str(e)}") error_count["logged"] += 1 except Exception as e: logger.warning(f"Error parsing SSE event: {e}") finally: buffer = "" else: buffer += line + "\n" except requests.exceptions.ConnectionError as e: logger.warning(f"Failed to establish SSE connection: {str(e)}") # Signal that connection attempt is complete even if it failed connection_established.set() except Exception as e: logger.warning(f"Exception in SSE thread: {str(e)}") # Signal that connection attempt is complete even if it failed connection_established.set() finally: # Signal that we're done processing stop_event.set() # Ensure the connection is closed even if an exception occurs try: if response_container["response"] is not None: response_container["response"].close() except Exception as cleanup_error: logger.warning(f"Error closing SSE connection: {str(cleanup_error)}") # Create and start the worker thread worker_thread = threading.Thread(target=sse_worker) worker_thread.daemon = True worker_thread.start() # Track when we started start_time = time.time() try: # Wait for connection to be established with timeout connected = connection_established.wait(min(1.0, timeout / 2)) if not connected: logger.warning("Timed out waiting for SSE connection to establish") # Wait for thread to complete or until timeout is reached, checking for ready flag exit_time = start_time + timeout while worker_thread.is_alive() and time.time() < exit_time: # Check if the callback has indicated we're ready to return if result.get("ready", False) is True: break # Wait a short time (10ms) before checking again time.sleep(0.01) # If we're within 50ms of timeout, break early to ensure we have time for cleanup if time.time() > exit_time - 0.05: break # Signal worker thread to stop stop_event.set() # Force cleanup immediately without delay cleanup_connections(response_container) # Calculate execution time actual_duration = time.time() - start_time # Log completion status only for timeouts if worker_thread.is_alive(): logger.warning(f"SSE thread timed out after {actual_duration:.2f}s") # Log error summary if there were many errors if error_count["total"] > error_count["logged"]: logger.warning(f"SSE processing had {error_count['total']} errors, {error_count['logged']} were logged") # Convert set to list for JSON serialization if "event_types_seen" in result: result["event_types_seen"] = list(result["event_types_seen"]) # Add actual duration for debugging result["actual_duration"] = actual_duration return result finally: # Always signal thread to stop stop_event.set() # Always force cleanup cleanup_connections(response_container) # Force session close if response_container["session"] is not None: try: response_container["session"].close() except Exception: pass def cleanup_connections(response_container): """Aggressively clean up HTTP connections to prevent hangs""" # First try graceful close if response_container["response"] is not None: try: response_container["response"].close() except Exception: pass # Then try to close the underlying socket more forcefully try: if ( hasattr(response_container["response"], "raw") and response_container["response"].raw ): response_container["response"].raw.close() if hasattr(response_container["response"].raw, "_connection"): conn = response_container["response"].raw._connection if conn: if hasattr(conn, "sock") and conn.sock: conn.sock.close() conn.close() except Exception: pass # Clear the reference response_container["response"] = None def get_title_and_tags_from_sse( suno_request_id: str, timeout: Optional[float] = None ) -> Dict[str, Any]: """ Connects to Suno SSE endpoint and fetches the title and tags from line events. The title is extracted from the first line event with content surrounded by curly braces {like this}. The tags are extracted from the next line event after the title is found. Args: suno_request_id: The request ID to get events for timeout: Optional custom timeout (uses SSE_TIMEOUT if not provided) Returns: A dictionary with title string and tags list. Raises: ContentFilteredError: When a moderation failure event is received SkillError: When a non-moderation error event is received from the SSE stream """ # Use the timeout constant if no custom timeout provided total_timeout = SSE_TIMEOUT if timeout is None else timeout # Initialize result with required keys base_result = { "title": "", "tags": ["default"], "found_title": False, # Track if we've found a title "next_is_tags": False, # Track if the next line should be treated as tags } def process_line_event(result_data, event_type, event_data_raw): # Check for error events if event_type == "error": try: error_data = json.loads(event_data_raw) error_type = error_data.get("error_type", "unknown_error") error_message = error_data.get( "error_message", "Unknown error occurred during content generation." ) logger.error( f"Error event received from SSE: {error_type} - {error_message}" ) # Store the error in the result data to raise later result_data["error"] = {"type": error_type, "message": error_message} # Signal to stop processing as we've found an error return True except json.JSONDecodeError: logger.error(f"Failed to parse error event JSON: {event_data_raw}") result_data["error"] = { "type": "parse_error", "message": "Failed to parse error event from content generation service.", } return True # Skip if not a line event if event_type != "line": return False try: parsed_content = json.loads(event_data_raw) if not isinstance(parsed_content, str): return False content = parsed_content.strip() # Check if this is a line after we found the title (for tags) if result_data.get("next_is_tags", False): # Parse tags from format "tag1, tag2, tag3" tags = [tag.strip() for tag in content.split(",") if tag.strip()] if tags: result_data["tags"] = tags # As an optimization for latency, we don't wait further once tags are processed result_data["ready"] = True return True # Look for content surrounded by curly braces for title if ( not result_data.get("found_title", False) and content.startswith("{") and content.endswith("}") ): # Extract the content between braces title = content[1:-1].strip() if title: result_data["title"] = title result_data["found_title"] = True # Optimize for latency: return as soon as title is found result_data["ready"] = True return True # Continue processing return False except json.JSONDecodeError: # Silently ignore JSON decode errors - likely not our event format pass except Exception as e: # Log other errors without stack trace to reduce noise logger.warning(f"Error processing line event: {str(e)}") return False # Continue processing # Create the URL and connect to SSE stream - single attempt only url = f"{SUNO_REQUEST_EVENTS_ENDPOINT}?request_id={suno_request_id}" # Single attempt with all available time result = connect_to_sse_stream( url, process_line_event, total_timeout, initial_result=dict(base_result), max_timeout=SSE_TITLE_EXTRACTION_MAX_TIMEOUT, ) # Check if an error was encountered during processing if "error" in result: error_type = result["error"].get("type", "unknown_error") error_message = result["error"].get( "message", "Unknown error occurred during content generation." ) handle_moderation_error(error_type, error_message) # Return the title and tags we got (or empty/defaults if SSE failed) return { "title": result.get("title", ""), "tags": result.get("tags", ["default"]) } def wait_for_clip_events( clip_id: str, timeout: Optional[float] = None, handler_start_time: Optional[float] = None, ) -> Dict[str, Any]: """ Connects to Suno SSE endpoint and waits for both gen_streaming AND image_generated events. Now returns success when audio is ready but image is not ready by timeout, with a flag to indicate using fallback images instead of failing completely. Args: clip_id: The ID of the clip to wait for events from timeout: Optional custom timeout (uses SSE_TIMEOUT if not provided) handler_start_time: If provided, enforces absolute time constraint Returns: Dictionary with image_id, events status, and other metadata. Includes 'use_fallback_image' flag when audio is ready but image is not. Raises: SkillError: Only if audio events (gen_streaming or lyrics) are not received """ # Use the timeout constant if no custom timeout provided if timeout is None: timeout = SSE_TIMEOUT # If handler start time provided, calculate remaining time if handler_start_time is not None: time_elapsed = time.time() - handler_start_time remaining_time = ( MAX_HANDLER_TIME_BUDGET_SECONDS - time_elapsed - 0.5 ) # 0.5s safety buffer timeout = min(timeout, max(0.1, remaining_time)) # Initialize result with required keys initial_result = { "image_id": "", "title": "", "events": {"image_generated": False, "gen_streaming": False, "lyrics": False}, "ready": False, "use_fallback_image": False, } def process_clip_event(result_data, event_type, event_data_raw): # Process error events immediately - always return early on errors if event_type == "error": error_data = json.loads(event_data_raw) error_type = error_data.get("error_type", "unknown_error") error_message = error_data.get("error_message", "Unknown error occurred") logger.error(f"Error event from SSE: {error_type} - {error_message}") # Store error info in result to raise later result_data["error"] = {"type": error_type, "message": error_message} return True # Signal to stop processing # Process image_generated event elif event_type == "image_generated": parsed_content = json.loads(event_data_raw) if "image_id" in parsed_content: result_data["image_id"] = parsed_content["image_id"] result_data["events"]["image_generated"] = True # Process lyrics event elif event_type == "lyrics": result_data["events"]["lyrics"] = True parsed_content = json.loads(event_data_raw) if "title" in parsed_content: result_data["title"] = parsed_content["title"] # Process gen_streaming event elif event_type == "gen_streaming": result_data["events"]["gen_streaming"] = True # Check if both required conditions are met has_audio = result_data["events"]["gen_streaming"] or result_data["events"]["lyrics"] has_image = result_data["events"]["image_generated"] if has_audio and has_image: result_data["ready"] = True return True # Signal to stop processing return False # Continue processing # Connect to SSE stream with higher timeout for song generation url = f"{SUNO_CLIP_EVENTS_ENDPOINT}?clip_id={clip_id}" result = connect_to_sse_stream(url, process_clip_event, timeout, initial_result, max_timeout=SSE_SONG_GENERATION_MAX_TIMEOUT) # Check if an error was encountered during processing if "error" in result: error_type = result["error"].get("type", "unknown_error") error_message = result["error"].get("message", "Unknown error occurred") handle_moderation_error(error_type, error_message) # If we got here and ready is True, return the successful result if result.get("ready", False): return result # Otherwise, we didn't get both required events within the timeout has_audio = result["events"]["gen_streaming"] or result["events"]["lyrics"] has_image = result["events"]["image_generated"] # New logic: If we have audio but not image, return success with fallback image flag if has_audio and not has_image: logger.warning( f"Audio ready but image not generated for clip {clip_id} within timeout. Using fallback image." ) result["ready"] = True result["use_fallback_image"] = True return result # Only fail if we don't have audio (audio is critical for playback) logger.error( f"Timeout waiting for required events. Got audio: {has_audio}, image: {has_image}" ) # Provide more specific error message about what's missing if not has_audio and not has_image: error_message = "MISSING_AUDIO_AND_IMAGE" elif not has_audio: error_message = "MISSING_AUDIO" else: # This shouldn't happen since we handle the audio-but-no-image case above error_message = "GENERATION_NOT_READY" raise SkillError(error_message, error_code="INTERNAL_ERROR") # ============================================================================= # Content Generation Handler (for GENERATE_CONTENT action) # ============================================================================= def call_suno_generate_api( access_token: str, prompt: str, make_instrumental: bool ) -> Dict[str, Any]: """ Calls the Suno Generate API to create a song from a prompt. Args: access_token: OAuth access token prompt: The text prompt to generate from make_instrumental: Whether to make an instrumental track Returns: API response dictionary with clip information Raises: AuthorizationError: For authentication issues RateLimitError: For rate limit issues SkillError: For other API errors requests.exceptions.RequestException: For network issues """ headers = create_auth_headers(access_token) # Generate the fallback title once so we can reuse it if SSE fails title_fallback = generate_song_title() data = { "topic": prompt, "model": "chirp-v4-h-api", "make_instrumental": make_instrumental, "extra": {"early_callback": True, "title_fallback": title_fallback}, } url = f"{SUNO_BASE_API_URL}{SUNO_GENERATE_PATH}" logger.debug(f"Calling Suno Generate API: {url}") response = requests.post(url, headers=headers, json=data, timeout=SUNO_API_TIMEOUT) # Check for specific HTTP errors if response.status_code == 401: logger.error( "Suno Generate API returned 401 Unauthorized. Token likely invalid/expired." ) raise AuthorizationError( "Invalid credentials for content generation.", error_code="INVALID_AUTHORIZATION_CREDENTIAL", ) if response.status_code == 402: logger.error( "Suno Generate API returned 402 Payment Required. Account likely does not have sufficient credits." ) return {"error_code": "INSUFFICIENT_USER_SUBSCRIPTION"} if response.status_code == 429: logger.warning("Suno Generate API returned 429 Rate Limited.") raise RateLimitError( "Generation request rate limit exceeded.", error_code="RATE_LIMIT_EXCEEDED", ) # Raise for other 4xx/5xx errors response.raise_for_status() # Parse the response clips_list = response.json() if not isinstance(clips_list, list) or not clips_list: logger.error( f"Suno Generate API returned unexpected response (not a non-empty list): {clips_list}" ) raise SkillError( "Failed to generate content: Unexpected response from provider.", error_code="CONTENT_GENERATION_ERROR", ) # Return the clip along with the title_fallback we used result = clips_list[1] # second clip is created later, so it should be at the top of the play queue result["_title_fallback"] = title_fallback # Store the fallback for later use return result def calculate_sse_timeout(handler_start_time: float, generate_api_duration: float) -> float: """ Calculates dynamic timeout for SSE based on remaining budget. Uses all available time up to the 3.5s target, minus safety buffer. Args: handler_start_time: Time when handler started processing generate_api_duration: How long the generate API call took Returns: Safe timeout value in seconds for SSE operation """ # Total time elapsed since handler started total_elapsed = time.time() - handler_start_time # Calculate remaining budget: target (3.5s) - elapsed - safety buffer (0.1s) remaining_budget = TARGET_GENERATION_TIME_SECONDS - total_elapsed - SSE_SAFETY_BUFFER_SECONDS # If we're already over budget, give minimal time if remaining_budget <= 0: logger.warning(f"Already over time budget. Elapsed: {total_elapsed:.2f}s, Generate API: {generate_api_duration:.2f}s") return 0.1 # Minimal timeout # Use all remaining budget for SSE (but cap at a reasonable max) sse_timeout = min(remaining_budget, 2.0) # Cap at 2s max for SSE logger.info(f"SSE timeout calculation: elapsed={total_elapsed:.2f}s, generate_api={generate_api_duration:.2f}s, sse_timeout={sse_timeout:.2f}s") return sse_timeout def handle_generate_content_action( access_token: str, prompt: str, handler_start_time: float, ) -> Dict[str, Any]: """ Handles the GENERATE_CONTENT action with dynamic SSE timeout. 1. Calls Suno Generate API. 2. Calculates remaining time budget. 3. Extracts first clip ID and Suno request ID. 4. Calls SSE helper with dynamic timeout to get the title and tags. 5. Returns a dictionary with clip_id, title, and image_url. Can raise SkillError, AuthorizationError, RateLimitError, ContentFilteredError. """ generate_api_start_time = time.time() try: # --- 1. Call Suno Generate API --- make_instrumental = not song_is_vocal(prompt) # Remove "on suno" and similar phrases from prompt cleaned_prompt = re.sub(r'\s+on\s+su[nm]o\b', '', prompt, flags=re.IGNORECASE) first_clip = call_suno_generate_api(access_token, cleaned_prompt, make_instrumental) if "error_code" in first_clip: return first_clip # Return error directly generate_api_duration = time.time() - generate_api_start_time logger.info(f"Suno Generate API call duration: {generate_api_duration:.2f}s") # --- 2. Extract Clip ID and Suno Request ID --- clip_id = first_clip.get("id") suno_request_id = first_clip.get("request_id") initial_image_url = first_clip.get("image_url") or first_clip.get( "image_large_url" ) # Extract the title_fallback we stored title_fallback = first_clip.get("_title_fallback", "") if not clip_id: logger.error( f"Suno Generate API response missing 'id' in first clip: {first_clip}" ) raise SkillError( "Failed to generate content: Missing content identifier.", error_code="CONTENT_GENERATION_ERROR", ) # --- 3. Calculate remaining time budget for SSE --- dynamic_sse_timeout = calculate_sse_timeout(handler_start_time, generate_api_duration) # --- 4. Call SSE helper with dynamic timeout to get title and tags --- title = "" image_url = "" if suno_request_id: logger.info(f"Starting SSE request with timeout: {dynamic_sse_timeout:.2f}s") sse_start = time.time() sse_result = get_title_and_tags_from_sse( suno_request_id, dynamic_sse_timeout ) sse_duration = time.time() - sse_start logger.info(f"SSE completed in {sse_duration:.2f}s, got title: {bool(sse_result.get('title'))}") title = sse_result["title"] # If we have image URL already from API, use it if initial_image_url: image_url = initial_image_url else: # For generating clips, use clip-specific CDN URL instead of fallback # This ensures we'll get the proper image when it becomes available image_url = f"{SUNO_IMAGE_CDN}/image_{clip_id}.jpeg" else: logger.warning( f"No 'request_id' found in Suno response for clip {clip_id}. Cannot fetch title/tags via SSE." ) # Use clip-specific CDN URL for all generating clips image_url = f"{SUNO_IMAGE_CDN}/image_{clip_id}.jpeg" # Create a suitable fallback title if SSE failed or wasn't possible if not title: # Use the same title_fallback that was sent to the API, or generate a new one title = title_fallback if title_fallback else "Untitled" logger.warning( f"No title from SSE for clip {clip_id}, using fallback: '{title}'" ) # --- 5. Return results --- return { "clip_id": clip_id, "title": title, "image_url": image_url, } except ContentFilteredError as e: # Pass through ContentFilteredError directly to be handled by the caller logger.warning(f"Content filtering error during generation: {e.message}") raise except requests.exceptions.Timeout as e: logger.error( f"Timeout calling Suno Generate API ({SUNO_API_TIMEOUT}s).", exc_info=True ) raise SkillError( "Content generation timed out.", error_code="INTERNAL_ERROR", cause=e ) except requests.exceptions.RequestException as e: logger.error(f"RequestException calling Suno Generate API: {e}", exc_info=True) raise SkillError( "Could not connect to content generation service.", error_code="INTERNAL_ERROR", cause=e, ) except (json.JSONDecodeError, KeyError, IndexError, TypeError) as e: logger.error(f"Error processing Suno Generate API response: {e}", exc_info=True) raise SkillError( "Failed to process generation response.", error_code="INTERNAL_ERROR", cause=e, ) except SkillError as e: logger.error(f"SkillError during generation handling: {e.message}") raise e except Exception as e: logger.error(f"Unexpected error during generation handling: {e}", exc_info=True) raise SkillError( f"An unexpected error occurred during content generation: {str(e)}", error_code="INTERNAL_ERROR", cause=e, ) # ============================================================================= # Library Search Function # ============================================================================= @handle_api_errors("library search") def search_suno_library( access_token: str, search_term: str, from_index: int = 0, page_size: int = 20, rank_by: str = "most_relevant", ) -> List[Dict[str, Any]]: """ Search the user's Suno library using the external search API. Args: access_token: The OAuth access token search_term: The search term to query for from_index: Starting index for pagination page_size: Number of results to return rank_by: Ranking method for results Returns: List of song items from the search results """ headers = create_auth_headers(access_token) search_url = f"{SUNO_BASE_API_URL}/api/v2/external/oauth/search" # Prepare the search query payload payload = { "search_queries": [ { "search_type": "library_song", "name": "library_song", "term": search_term, "from_index": from_index, "rank_by": rank_by, } ] } response = requests.post(search_url, headers=headers, json=payload, timeout=3.0) # Check for specific error codes if response.status_code == 401: raise AuthorizationError( "Invalid credentials for content search.", error_code="INVALID_AUTHORIZATION_CREDENTIAL", ) elif response.status_code == 429: raise RateLimitError( "Too many requests. Please try again later.", error_code="RATE_LIMIT_EXCEEDED", ) response.raise_for_status() search_results = response.json() # Extract the song results from the response if ( search_results and "result" in search_results and "library_song" in search_results["result"] and "result" in search_results["result"]["library_song"] ): songs = search_results["result"]["library_song"]["result"] # Format the songs for display formatted_songs = [] for song in songs: formatted_song = { "id": song.get("id"), "title": song.get("title", f"Song {song.get('id')[:8]}"), "image_url": song.get("image_url") or song.get("image_large_url", ""), # Include additional metadata that might be useful "status": song.get("status"), "created_at": song.get("created_at"), "metadata": song.get("metadata", {}), "explicit": song.get("explicit", False), # Add the search term so we can use it in the group title "search_term": search_term, } formatted_songs.append(formatted_song) return formatted_songs else: return [] # ============================================================================= # Image URL Handling # ============================================================================= def get_image_url( clip_id: str = None, image_id: str = None, fallback: bool = False, status: str = "unknown", ) -> str: """ Get an appropriate image URL based on available identifiers. Args: clip_id: Optional clip ID to use for image image_id: Optional image ID to use (takes precedence if provided) fallback: Whether to generate a fallback image if no IDs are provided status: Status of the clip ("complete", "streaming", "submitted", etc.) Returns: A suitable image URL """ # Use image_id first if available if image_id: return f"{SUNO_IMAGE_CDN}/{image_id}.jpeg" # Then use clip_id if available if clip_id: return f"{SUNO_IMAGE_CDN}/image_{clip_id}.jpeg" # Only use fallback for completed clips with missing image or when no clip_id is available if fallback and (not clip_id or status == "complete"): return get_fallback_image_url() # Return empty string if nothing available and no fallback requested return "" def get_fallback_image_url() -> str: """ Generate a fallback image URL using a random genre from GENRE_LIST. Returns: A Suno CDN URL for a genre cover image """ random_genre = random.choice(GENRE_LIST) return f"{SUNO_IMAGE_CDN}/GENRE_COVER_IMG_{random_genre}.jpeg" def update_catalog(user_id, clip_id, title, access_token): """ Updates the Alexa music catalog with a new song by notifying the Suno backend. Args: user_id (str): The Alexa user ID clip_id (str): The Suno clip ID to add to the catalog title (str): The title of the song (optional) access_token (str): The OAuth access token """ try: # Prepare request to Suno backend API headers = create_auth_headers(access_token) # Prepare the query parameters according to the correct API format request_params = {"clip_id": clip_id, "alexa_user_id": user_id} # Add title if provided (optional) if title: request_params["title"] = title # Call the Suno backend API with query parameters song_created_url = ( f"{SUNO_BASE_API_URL}/api/v2/external/oauth/alexa/song-created" ) response = requests.post( song_created_url, params=request_params, headers=headers, timeout=5.0 ) if response.status_code == 200: response_data = response.json() if response_data and "new_sync_token" in response_data: logger.info("Received new sync token for Alexa catalog updates") else: logger.warning( f"Failed to notify Suno backend of song creation: {response.status_code}: {response.text}" ) except Exception as e: logger.error( f"Unexpected error updating catalog via Suno backend: {str(e)}", exc_info=True, ) def notify_skill_account_linked(access_token, alexa_user_id, previous_sync_token=None, alexa_api_endpoint=None): """ Notifies the Suno backend that a skill account has been linked. Args: access_token (str): The OAuth access token alexa_user_id (str): The Alexa user ID previous_sync_token (str, optional): The previous sync token from Alexa alexa_api_endpoint (str, optional): The Alexa API endpoint to use Returns: Union[bool, Dict[str, Any]]: - False: If an error occurred during the request - Dict with sync information, including 'needs_full_update' and 'has_no_content' flags """ try: # Prepare headers headers = create_auth_headers(access_token) # Prepare query parameters instead of JSON body request_params = {"alexa_user_id": alexa_user_id} # Add previous sync token if available if previous_sync_token: request_params["previous_sync_token"] = previous_sync_token # Add Alexa API endpoint if available, otherwise use default if alexa_api_endpoint: request_params["alexa_api_endpoint"] = alexa_api_endpoint else: request_params["alexa_api_endpoint"] = "https://api.amazonalexa.com" # Call the Suno backend API account_linked_url = ( f"{SUNO_BASE_API_URL}/api/v2/external/oauth/alexa/skill-account-linked" ) response = requests.post( account_linked_url, params=request_params, # Use params instead of json headers=headers, timeout=5.0, ) if response.status_code == 200: response_data = response.json() # If the response includes sync_token information, return it if isinstance(response_data, dict) and "sync_info" in response_data: return { "needs_full_update": response_data.get("sync_info", {}).get( "needs_full_update", False ), "has_no_content": response_data.get("sync_info", {}).get( "has_no_content", False ), "sync_token": response_data.get("sync_info", {}).get( "sync_token", "" ), } # For backward compatibility, return True for success with no sync_info return True else: logger.warning( f"Failed to notify Suno backend of skill account linking: {response.status_code}: {response.text}" ) return False except Exception as e: logger.error( f"Unexpected error notifying account linking via Suno backend: {str(e)}", exc_info=True, ) return False def log_playback_event( access_token: str, user_id: str, content_id: str, queue_id: str, event_type: str, offset_ms: int = 0, error_type: str = None, error_message: str = None, request_id: str = None, timestamp: str = None, cause_type: str = None, playback_attributes: Dict[str, Any] = None ) -> None: """ Logs a playback event to the Suno backend for analytics and user history. """ try: # Prepare headers and request data headers = create_auth_headers(access_token) request_data = { "alexa_user_id": user_id, "content_id": content_id, "queue_id": queue_id, "event_type": event_type, "offset_ms": offset_ms } # Add optional fields if they exist if error_type: request_data["error_type"] = error_type if error_message: request_data["error_message"] = error_message if request_id: request_data["alexa_request_id"] = request_id if timestamp: request_data["timestamp"] = timestamp if cause_type: request_data["cause_type"] = cause_type if playback_attributes: request_data["playback_attributes"] = playback_attributes # Log event info logger.debug(f"Logging {event_type} event for content_id {content_id}") # Send to Suno backend playback_event_url = f"{SUNO_BASE_API_URL}/api/v2/external/oauth/alexa/playback-event" requests.post( playback_event_url, json=request_data, headers=headers, timeout=1.0 ) except Exception as e: # Just log the error but don't disrupt the Alexa flow logger.warning(f"Error logging playback event: {str(e)}") # ============================================================================= # Shared Helper Functions # ============================================================================= def generate_art_sources( image_url: str, suno_image_cdn: str = None, custom_dimensions: List[Dict[str, Any]] = None, ) -> List[Union[Dict[str, Any], Any]]: """ Generate standardized art sources array based on image URL. Args: image_url: The URL for the image suno_image_cdn: Optional CDN base URL to use for checking CDN URLs custom_dimensions: Optional custom size dimensions to override defaults Returns: A list of art sources (either as dictionaries or ArtSource objects) """ art_sources = [] # Check if we're using a CDN image cdn_base = suno_image_cdn or SUNO_IMAGE_CDN is_cdn_image = image_url and ( "cdn1.suno.ai" in image_url or "cdn2.suno.ai" in image_url or image_url.startswith(cdn_base) ) if is_cdn_image: # Standardize CDN domain to cdn2.suno.ai if it's using cdn1 if "cdn1.suno.ai" in image_url: image_url = image_url.replace("cdn1.suno.ai", "cdn2.suno.ai") # For CDN images, provide multiple sizes # Use custom dimensions if provided, otherwise use defaults size_dimensions = custom_dimensions or [ {"size": "LARGE", "dimension": 360}, {"size": "MEDIUM", "dimension": 256}, {"size": "SMALL", "dimension": 100}, ] for size_info in size_dimensions: art_sources.append( { "url": f"{image_url}?width={size_info['dimension']}", "size": size_info["size"], "widthPixels": size_info["dimension"], "heightPixels": size_info["dimension"], } ) elif image_url: art_sources.append( { "url": image_url, "size": "X_LARGE", "widthPixels": 600, "heightPixels": 600, } ) else: # For empty URL (generated content or missing), still need to provide a sources array art_sources.append( { "url": "", "size": "X_LARGE", "widthPixels": 600, "heightPixels": 600, } ) return art_sources def get_art_url_for_item(item: Dict[str, Any]) -> str: """ Determine the best art URL for an item based on its properties. Args: item: Dictionary containing clip/item data with potential image URLs Returns: The most appropriate image URL for the item """ # Extract basic properties art_url = item.get("image_url", "") item_id = item.get("id", "unknown") status = item.get("status", "complete") # If we already have a valid image URL, use it if art_url: return art_url # Also check for large image URL if the standard one isn't available large_url = item.get("image_large_url", "") if large_url: return large_url # For complete clips with missing images, use the fallback image if status == "complete": return get_fallback_image_url() # For in-progress clips or any other case, use the standard CDN URL format # This ensures we'll get the proper image when it becomes available return f"{SUNO_IMAGE_CDN}/image_{item_id}.jpeg"