app.py

import gradio as gr
import requests
import os
import json
import uuid
from datetime import datetime

# Load Suno API key from environment variable
SUNO_KEY = os.environ.get("SUNO_KEY", os.environ.get("SUNOKEY", ""))

# In production, use your actual public URL
# For Spaces, get from environment or use ngrok for local testing
SPACE_URL = os.environ.get("SPACE_URL", "https://your-username.hf.space")

# Store tasks with timestamps
tasks_db = {}

def generate_lyrics(prompt, callBackUrl=""):
    """Submit lyrics generation task with callback URL"""
    if not SUNO_KEY:
        return "❌ Error: SUNO_KEY environment variable not set"
    
    # Generate unique task ID
    task_id = str(uuid.uuid4())[:8]
    
    # Use provided callback URL or default to your Space URL
    callback_url = callBackUrl or f"{SPACE_URL}/callback"
    
    url = "https://api.sunoapi.org/api/v1/lyrics"
    headers = {
        "Authorization": f"Bearer {SUNO_KEY}",
        "Content-Type": "application/json"
    }
    
    payload = {
        "prompt": prompt,
        "callBackUrl": callback_url,
        "customTaskId": task_id  # Optional: track your own ID
    }
    
    try:
        resp = requests.post(url, headers=headers, json=payload, timeout=30)
        data = resp.json()
        
        if resp.status_code == 200 and data.get("code") == 200:
            api_task_id = data["data"]["taskId"]
            
            # Store task info
            tasks_db[task_id] = {
                "api_task_id": api_task_id,
                "prompt": prompt,
                "status": "submitted",
                "submitted_at": datetime.now().isoformat(),
                "callback_received": False,
                "lyrics": None
            }
            
            return f"""✅ Task Submitted!
Your Task ID: {task_id}
API Task ID: {api_task_id}
Callback URL: {callback_url}

📝 Prompt: {prompt}

⏳ Processing... The results will be sent to the callback URL.
You can also poll manually using your Task ID above."""
        
        else:
            error_msg = data.get("msg", "Unknown error")
            return f"❌ API Error: {error_msg} (Code: {data.get('code')})"
            
    except Exception as e:
        return f"❌ Error: {str(e)}"

def poll_task(task_id):
    """Manual polling as fallback"""
    if not SUNO_KEY:
        return "❌ Error: SUNO_KEY not configured"
    
    if task_id not in tasks_db:
        return "❌ Task ID not found. Please submit a task first."
    
    task_info = tasks_db[task_id]
    api_task_id = task_info["api_task_id"]
    
    url = f"https://api.sunoapi.org/api/v1/lyrics/details?taskId={api_task_id}"
    headers = {"Authorization": f"Bearer {SUNO_KEY}"}
    
    try:
        resp = requests.get(url, headers=headers, timeout=30)
        data = resp.json()
        
        if resp.status_code == 200 and data.get("code") == 200:
            task_data = data["data"]
            status = task_data.get("status", "unknown")
            tasks_db[task_id]["status"] = status
            
            if status == "completed" and "data" in task_data:
                lyrics_data = task_data["data"]
                tasks_db[task_id]["lyrics"] = lyrics_data
                tasks_db[task_id]["callback_received"] = True
                
                return format_lyrics(lyrics_data, task_id)
            elif status == "failed":
                return f"❌ Task failed: {task_data.get('error', 'Unknown error')}"
            else:
                return f"⏳ Status: {status}\nLast checked: {datetime.now().strftime('%H:%M:%S')}"
        else:
            return f"❌ Polling error: {data.get('msg', 'Unknown error')}"
            
    except Exception as e:
        return f"❌ Error: {str(e)}"

def format_lyrics(lyrics_data, task_id):
    """Format lyrics for display"""
    output = [f"🎵 **Task {task_id} - Generated Lyrics**", ""]
    
    for i, item in enumerate(lyrics_data, 1):
        title = item.get('title', f'Variant {i}')
        text = item.get('text', 'No lyrics generated')
        
        output.append(f"**Variant {i}: {title}**")
        output.append("```")
        output.append(text)
        output.append("```")
        output.append("---")
    
    output.append(f"\n✅ Generated at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    return "\n".join(output)

def list_tasks():
    """Show all submitted tasks"""
    if not tasks_db:
        return "No tasks submitted yet."
    
    output = ["📋 **Submitted Tasks:**", ""]
    for task_id, info in tasks_db.items():
        status_icon = "✅" if info["callback_received"] else "⏳"
        output.append(f"{status_icon} **{task_id}** - {info['status']}")
        output.append(f"   Prompt: {info['prompt'][:50]}...")
        output.append(f"   Submitted: {info['submitted_at']}")
        output.append("")
    
    return "\n".join(output)

# WEBHOOK ENDPOINT (for receiving callbacks)
def webhook_callback(request: gr.Request):
    """Handle incoming webhook from Suno API"""
    try:
        # Try to get JSON data
        data = request.json()
        
        if not data:
            # Try form data
            data = dict(request.form)
        
        print(f"📥 Received webhook: {json.dumps(data, indent=2)}")
        
        # Extract task info from webhook
        # Suno API format might vary - adjust based on actual response
        if "data" in data and "taskId" in data["data"]:
            api_task_id = data["data"]["taskId"]
            status = data["data"].get("status", "unknown")
            
            # Find our task by API task ID
            for task_id, task_info in tasks_db.items():
                if task_info["api_task_id"] == api_task_id:
                    task_info["status"] = status
                    task_info["callback_received"] = True
                    
                    if status == "completed" and "data" in data["data"]:
                        task_info["lyrics"] = data["data"]["data"]
                        print(f"✅ Lyrics received for task {task_id}")
                    
                    return {"status": "success", "message": f"Updated task {task_id}"}
        
        return {"status": "error", "message": "Task not found"}
        
    except Exception as e:
        print(f"❌ Webhook error: {e}")
        return {"status": "error", "message": str(e)}

# Gradio Interface
with gr.Blocks(theme=gr.themes.Soft(), title="Suno Lyrics Generator") as app:
    gr.Markdown("# 🎵 Suno AI Lyrics Generator")
    gr.Markdown("Generate song lyrics with webhook support")
    
    with gr.Tabs():
        with gr.TabItem("🎤 Generate"):
            with gr.Row():
                with gr.Column():
                    prompt = gr.Textbox(
                        label="Lyrics Prompt",
                        placeholder="A romantic ballad about stargazing...",
                        lines=3
                    )
                    
                    # Optional: Custom callback URL
                    callback_url = gr.Textbox(
                        label="Callback URL (Optional)",
                        value=f"{SPACE_URL}/callback",
                        info="Where Suno should send results. Leave as default for Spaces."
                    )
                    
                    submit_btn = gr.Button("✨ Generate Lyrics", variant="primary")
                    
                    gr.Markdown("### ℹ️ Instructions:")
                    gr.Markdown("""
                    1. Enter your lyrics prompt
                    2. Click Generate
                    3. Save your Task ID
                    4. Check status in the "Poll Tasks" tab
                    5. Results will arrive via webhook automatically
                    """)
                
                with gr.Column():
                    output = gr.Textbox(
                        label="Submission Result",
                        lines=10,
                        interactive=False
                    )
            
            submit_btn.click(
                generate_lyrics,
                inputs=[prompt, callback_url],
                outputs=output
            )
        
        with gr.TabItem("🔄 Poll Tasks"):
            with gr.Row():
                with gr.Column():
                    task_id_input = gr.Textbox(
                        label="Your Task ID",
                        placeholder="Enter the Task ID from generation step"
                    )
                    poll_btn = gr.Button("🔍 Check Status", variant="primary")
                    
                    gr.Markdown("---")
                    refresh_btn = gr.Button("📋 List All Tasks")
                    tasks_list = gr.Textbox(label="All Tasks", lines=10)
                
                with gr.Column():
                    poll_result = gr.Textbox(
                        label="Task Status",
                        lines=15,
                        interactive=False
                    )
            
            poll_btn.click(
                poll_task,
                inputs=task_id_input,
                outputs=poll_result
            )
            
            refresh_btn.click(
                list_tasks,
                inputs=None,
                outputs=tasks_list
            )
        
        with gr.TabItem("⚙️ Webhook Info"):
            gr.Markdown("### 🌐 Webhook Configuration")
            gr.Markdown(f"""
            **Your Webhook URL:** `{SPACE_URL}/callback`
            
            **For Local Development:**
            1. Use [ngrok](https://ngrok.com/): `ngrok http 7860`
            2. Update callback URL: `https://your-ngrok-url.ngrok.io/callback`
            
            **For Hugging Face Spaces:**
            - The URL above should work automatically
            - Make sure your Space is public or has network access
            """)
            
            webhook_status = gr.Textbox(
                label="Last Webhook Status",
                value="No webhooks received yet",
                lines=5
            )
    
    # Register webhook endpoint
    # Note: In production, you'd set up proper route handling
    # For Gradio, we can simulate with a POST endpoint
    app.post("/callback")(webhook_callback)

# Launch configuration
if __name__ == "__main__":
    # For local testing with webhooks
    import socket
    
    # Get local IP for ngrok compatibility
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        s.connect(("8.8.8.8", 80))
        local_ip = s.getsockname()[0]
        s.close()
        print(f"🌐 Local IP: {local_ip}")
        print(f"🌐 Webhook URL: http://{local_ip}:7860/callback")
    except:
        pass
    
    app.launch(
        server_name="0.0.0.0",
        server_port=7860,
        share=False,
        show_error=True
    )