import os import requests import gradio as gr from dotenv import load_dotenv import json import uuid import tempfile from datetime import datetime import logging import sys from typing import Dict, Any, Optional, Union, List, Tuple from dataclasses import dataclass import time import mimetypes import shutil from pathlib import Path # Load environment variables from .env file load_dotenv() # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.StreamHandler(sys.stdout), logging.FileHandler('suno_generator.log') ] ) logger = logging.getLogger(__name__) # Configuration class Config: """Configuration manager""" def __init__(self): self.suno_api_key = os.environ.get("SUNO_API_KEY") or os.environ.get("SunoKey") self.suno_api_url = "https://api.sunoapi.org/api/v1/generate/upload-cover" # Your PHP callback URL - kept private in app self.php_callback_url = "https://1hit.no/cover/cb.php" # Private, not exposed # Temporary server for uploaded files (simulated or real) self.temp_server_url = os.environ.get("TEMP_SERVER_URL", "https://temp.yourdomain.com") # Local storage for uploaded files (for simulation) self.upload_dir = Path("uploads") self.upload_dir.mkdir(exist_ok=True) # Public facing URL (what users see) self.public_callback_info = "Your callback URL (configured in app backend)" # Default values self.default_model = "V4_5ALL" self.default_vocal_gender = "m" self.max_retries = 3 self.request_timeout = 30 self.max_file_size_mb = 50 # Maximum file size for uploads # Default values for UI self.defaults = { "prompt": "A calm and relaxing piano track with soft melodies", "title": "Peaceful Piano Meditation", "style": "Classical", "style_weight": 0.65, "weirdness_constraint": 0.65, "audio_weight": 0.65, "instrumental": True, "custom_mode": True } config = Config() # Data models @dataclass class UploadedFile: """Information about uploaded file""" filename: str temp_url: str filepath: Path size: int mime_type: str upload_time: datetime @dataclass class ApiResponse: """Structured API response""" success: bool message: str task_id: Optional[str] = None upload_url: Optional[str] = None callback_url: Optional[str] = None raw_response: Optional[Dict] = None error_type: Optional[str] = None class SunoAPIError(Exception): """Custom exception for Suno API errors""" pass class FileUploadError(Exception): """Custom exception for file upload errors""" pass # Session management for requests session = requests.Session() session.headers.update({ "User-Agent": "SunoMusicGenerator/1.0", "Accept": "application/json" }) def validate_uploaded_file(file_path: str) -> Tuple[bool, str]: """Validate uploaded file""" try: # Check if file exists if not os.path.exists(file_path): return False, "File does not exist" # Check file size file_size = os.path.getsize(file_path) max_size = config.max_file_size_mb * 1024 * 1024 if file_size > max_size: return False, f"File too large (max {config.max_file_size_mb}MB)" # Check file type mime_type, _ = mimetypes.guess_type(file_path) if mime_type not in ['audio/mpeg', 'audio/mp3', 'audio/wav', 'audio/x-wav']: return False, "Only MP3 and WAV files are supported" return True, "File validated successfully" except Exception as e: return False, f"Validation error: {str(e)}" def upload_file_to_temp_server(file_path: str, original_filename: str) -> UploadedFile: """ Simulate uploading a file to a temporary server. In production, you would upload to S3, Cloud Storage, or your own server. """ try: # Create a unique filename timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") unique_id = str(uuid.uuid4())[:8] extension = Path(original_filename).suffix or '.mp3' new_filename = f"{timestamp}_{unique_id}{extension}" # Copy file to upload directory (simulating upload) dest_path = config.upload_dir / new_filename shutil.copy2(file_path, dest_path) # Generate a temporary URL (simulated) # In production, this would be a real URL to your storage service temp_url = f"{config.temp_server_url}/uploads/{new_filename}" # Get file info file_size = os.path.getsize(dest_path) mime_type, _ = mimetypes.guess_type(str(dest_path)) logger.info(f"File uploaded: {original_filename} -> {temp_url} ({file_size} bytes)") return UploadedFile( filename=new_filename, temp_url=temp_url, filepath=dest_path, size=file_size, mime_type=mime_type or 'audio/mpeg', upload_time=datetime.now() ) except Exception as e: logger.error(f"File upload failed: {e}") raise FileUploadError(f"Failed to upload file: {str(e)}") def cleanup_old_uploads(max_age_hours: int = 24): """Clean up old uploaded files""" try: cutoff_time = datetime.now().timestamp() - (max_age_hours * 3600) for file_path in config.upload_dir.glob("*"): if file_path.is_file(): if file_path.stat().st_mtime < cutoff_time: file_path.unlink() logger.info(f"Cleaned up old file: {file_path.name}") except Exception as e: logger.warning(f"Cleanup failed: {e}") def validate_inputs( prompt: str, title: str, style: str, use_reference_audio: bool, reference_audio_path: Optional[str] ) -> List[str]: """Validate user inputs and return list of errors""" errors = [] # Trim inputs prompt = prompt.strip() title = title.strip() style = style.strip() # Check required fields if len(prompt) < 5: errors.append("Prompt should be at least 5 characters") if not title: errors.append("Title is required") if not style: errors.append("Style is required") # Check reference audio if enabled if use_reference_audio and reference_audio_path: is_valid, error_msg = validate_uploaded_file(reference_audio_path) if not is_valid: errors.append(f"Reference audio: {error_msg}") return errors def make_api_request_with_retry( payload: Dict[str, Any], headers: Dict[str, str], max_retries: int = 3 ) -> requests.Response: """Make API request with retry logic""" last_exception = None for attempt in range(max_retries): try: logger.info(f"API Request Attempt {attempt + 1}/{max_retries}") response = session.post( config.suno_api_url, json=payload, headers=headers, timeout=config.request_timeout ) response.raise_for_status() return response except requests.exceptions.Timeout: logger.warning(f"Request timeout (attempt {attempt + 1})") last_exception = "Timeout" if attempt < max_retries - 1: time.sleep(2 ** attempt) # Exponential backoff except requests.exceptions.ConnectionError: logger.warning(f"Connection error (attempt {attempt + 1})") last_exception = "Connection error" if attempt < max_retries - 1: time.sleep(2 ** attempt) except requests.exceptions.HTTPError as e: logger.error(f"HTTP error: {e}") last_exception = f"HTTP {e.response.status_code}" break # Don't retry HTTP errors except Exception as e: logger.error(f"Request failed: {e}") last_exception = str(e) break # Don't retry other errors raise SunoAPIError(f"API request failed after {max_retries} attempts: {last_exception}") def generate_suno_music( prompt: str, title: str, style: str, use_reference_audio: bool, reference_audio_path: Optional[str], instrumental: bool = True, model: str = "V4_5ALL", persona_id: Optional[str] = None, negative_tags: Optional[str] = None, vocal_gender: str = "m", style_weight: float = 0.65, weirdness_constraint: float = 0.65, audio_weight: float = 0.65, custom_mode: bool = True ) -> ApiResponse: """ Generate music using Suno API with uploaded reference audio """ logger.info(f"Starting music generation: {title}") # Check if API key is available if not config.suno_api_key: error_msg = "Suno API key not found. Please set the 'SUNO_API_KEY' or 'SunoKey' environment variable." logger.error(error_msg) return ApiResponse( success=False, message=f"❌ {error_msg}", error_type="ConfigurationError" ) # Validate inputs validation_errors = validate_inputs(prompt, title, style, use_reference_audio, reference_audio_path) if validation_errors: error_msg = "Input validation failed:\n" + "\n".join([f"• {err}" for err in validation_errors]) logger.warning(f"Validation errors: {validation_errors}") return ApiResponse( success=False, message=f"❌ {error_msg}", error_type="ValidationError" ) # Handle reference audio upload upload_url = None if use_reference_audio and reference_audio_path: try: original_filename = os.path.basename(reference_audio_path) uploaded_file = upload_file_to_temp_server(reference_audio_path, original_filename) upload_url = uploaded_file.temp_url logger.info(f"Using uploaded reference audio: {upload_url}") except FileUploadError as e: return ApiResponse( success=False, message=f"❌ Failed to upload reference audio: {str(e)}", error_type="UploadError" ) # Prepare payload payload = { "customMode": custom_mode, "instrumental": instrumental, "model": model, "callBackUrl": config.php_callback_url, # Private callback URL "prompt": prompt, "style": style, "title": title, "personaId": persona_id or "", "negativeTags": negative_tags or "", "vocalGender": vocal_gender, "styleWeight": style_weight, "weirdnessConstraint": weirdness_constraint, "audioWeight": audio_weight } # Add upload URL if provided if upload_url: payload["uploadUrl"] = upload_url # Remove empty fields payload = {k: v for k, v in payload.items() if v not in ["", None]} # Prepare headers headers = { "Authorization": f"Bearer {config.suno_api_key}", "Content-Type": "application/json" } try: # Make API request response = make_api_request_with_retry(payload, headers, config.max_retries) result = response.json() # Check API response if response.status_code == 200 and result.get("code") == 200: task_id = result.get("data", {}).get("taskId", "Unknown") logger.info(f"Success! Task ID: {task_id}") return ApiResponse( success=True, message="✅ Music generation started successfully!", task_id=task_id, upload_url=upload_url, callback_url=config.php_callback_url, # Private URL, not shown to user raw_response=result ) else: error_msg = result.get('msg', 'Unknown API error') logger.error(f"API error: {error_msg}") return ApiResponse( success=False, message=f"❌ API Error: {error_msg}", raw_response=result, error_type="APIError" ) except SunoAPIError as e: logger.error(f"API request failed: {e}") return ApiResponse( success=False, message=f"❌ Request failed after retries: {str(e)}", error_type="RequestError" ) except requests.exceptions.HTTPError as e: logger.error(f"HTTP error: {e}") return ApiResponse( success=False, message=f"❌ HTTP Error {e.response.status_code}: {e.response.text}", error_type="HTTPError" ) except json.JSONDecodeError as e: logger.error(f"JSON decode error: {e}") return ApiResponse( success=False, message=f"❌ Failed to parse API response: {str(e)}", error_type="ParseError" ) except Exception as e: logger.error(f"Unexpected error: {e}") return ApiResponse( success=False, message=f"❌ Unexpected error: {str(e)}", error_type="UnexpectedError" ) def format_api_response(response: ApiResponse) -> str: """Format API response for display""" if response.success: # Don't expose the actual callback URL callback_display = "✓ Configured in app backend (private)" message = f"""{response.message} 📋 Task Details: • Task ID: {response.task_id} • Reference Audio: {response.upload_url or 'None (original generation)'} • Callback URL: {callback_display} ✅ Your music is being generated! The callback will be sent to our private endpoint. 🎧 Check your playlist for the result: https://1hit.no/cover/playlist.php 🔍 API Response Status: Success""" # Add raw response if needed for debugging if response.raw_response: message += f"\n\n📊 Raw Response (Task ID only):" # Only show task ID from raw response, not full URL safe_response = response.raw_response.copy() if 'data' in safe_response and 'callbackUrl' in safe_response['data']: safe_response['data']['callbackUrl'] = "[PRIVATE]" message += f"\n{json.dumps(safe_response, indent=2)}" return message else: error_details = f"\n\n🔍 Error Type: {response.error_type}" if response.error_type else "" return f"{response.message}{error_details}" def test_api_connection() -> str: """Test API connection and key validity""" if not config.suno_api_key: return "❌ API Key: Not found" try: # Simple test endpoint if available headers = { "Authorization": f"Bearer {config.suno_api_key}", } response = session.head( config.suno_api_url.replace("/upload-cover", ""), headers=headers, timeout=10 ) if response.status_code < 500: return f"✅ API Connection: OK" else: return f"⚠️ API Connection: May require valid payload (Status: {response.status_code})" except Exception as e: return f"❌ API Connection: Failed - {str(e)}" def load_example() -> Dict: """Load example values""" return { "prompt": config.defaults["prompt"], "title": config.defaults["title"], "style": config.defaults["style"], "use_reference_audio": False, "reference_audio": None, "instrumental": config.defaults["instrumental"], "model": config.default_model, "persona_id": "", "negative_tags": "Heavy Metal, Upbeat Drums", "vocal_gender": config.default_vocal_gender, "style_weight": config.defaults["style_weight"], "weirdness_constraint": config.defaults["weirdness_constraint"], "audio_weight": config.defaults["audio_weight"], "custom_mode": config.defaults["custom_mode"] } # Create Gradio interface with gr.Blocks( title="Suno Music Generator", theme=gr.themes.Soft(), css=""" .success-box { background: #d4edda; border: 1px solid #c3e6cb; border-radius: 5px; padding: 10px; margin: 10px 0; } .error-box { background: #f8d7da; border: 1px solid #f5c6cb; border-radius: 5px; padding: 10px; margin: 10px 0; } .info-box { background: #d1ecf1; border: 1px solid #bee5eb; border-radius: 5px; padding: 10px; margin: 10px 0; } .upload-info { background: #e7f3ff; border: 1px dashed #4a90e2; border-radius: 8px; padding: 15px; margin: 10px 0; } """ ) as app: # Custom CSS for better UI gr.HTML(""" """) gr.Markdown("# 🎵 Suno Music Generator") gr.Markdown("Generate music using Suno API with optional reference audio upload") # Status Bar with gr.Row(): with gr.Column(scale=1): api_status = gr.HTML( value=f"""
API Key: {'✅ Loaded' if config.suno_api_key else '❌ Not found'}
File Uploads: ✅ Enabled
Max File Size: {config.max_file_size_mb}MB
🎧 View PlaylistYour callback URL is configured privately in the app backend and is not exposed to users.
All generated music will appear in your playlist automatically.
Processing your request...
Uploading files and sending to Suno API...