Upload app.py

app.py

@@ -1,160 +1,307 @@
 import gradio as gr
 import requests
 import os
-import time
+import json
+import uuid
+from datetime import datetime
 
 # Load Suno API key from environment variable
-SUNO_KEY = os.environ.get("SUNO_KEY")  # Changed to match typical naming
+SUNO_KEY = os.environ.get("SUNO_KEY", os.environ.get("SUNOKEY", ""))
 
-# Check if API key exists
-if not SUNO_KEY:
-    print("⚠️ Warning: SUNO_KEY environment variable not set!")
+# 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 in a dictionary for multiple users (better than global)
-user_tasks = {}
+# Store tasks with timestamps
+tasks_db = {}
 
-def generate_lyrics(prompt, username="default"):
-    """Submit lyrics generation task"""
+def generate_lyrics(prompt, callBackUrl=""):
+    """Submit lyrics generation task with callback URL"""
     if not SUNO_KEY:
-        return "Error: API key not configured. Please set SUNO_KEY environment variable."
+        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": ""
+        "callBackUrl": callback_url,
+        "customTaskId": task_id  # Optional: track your own ID
     }
-
+    
     try:
         resp = requests.post(url, headers=headers, json=payload, timeout=30)
-        resp.raise_for_status()
         data = resp.json()
         
-        if data.get("code") == 200 and "data" in data:
-            task_id = data["data"]["taskId"]
-            user_tasks[username] = task_id
-            return f"✅ Submitted!\nTask ID: {task_id}\nPrompt: {prompt[:50]}..."
+        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:
-            return f"❌ API Error: {data.get('message', 'Unknown error')}"
+            error_msg = data.get("msg", "Unknown error")
+            return f"❌ API Error: {error_msg} (Code: {data.get('code')})"
             
-    except requests.exceptions.RequestException as e:
-        return f"❌ Connection Error: {str(e)}"
     except Exception as e:
-        return f"❌ Unexpected Error: {str(e)}"
+        return f"❌ Error: {str(e)}"
 
-def poll_task(username="default"):
-    """Poll the current user's task"""
+def poll_task(task_id):
+    """Manual polling as fallback"""
     if not SUNO_KEY:
-        return "Error: API key not configured."
+        return "❌ Error: SUNO_KEY not configured"
+    
+    if task_id not in tasks_db:
+        return "❌ Task ID not found. Please submit a task first."
     
-    task_id = user_tasks.get(username)
-    if not task_id:
-        return "No task submitted yet! Generate lyrics 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={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)
-        resp.raise_for_status()
         data = resp.json()
         
-        if data.get("code") == 200 and "data" in data:
+        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"]
-                output_lines = []
+                tasks_db[task_id]["lyrics"] = lyrics_data
+                tasks_db[task_id]["callback_received"] = True
                 
-                for i, item in enumerate(lyrics_data, 1):
-                    output_lines.append(f"🎵 Variant {i}: {item.get('title', 'Untitled')}")
-                    output_lines.append(f"{item.get('text', 'No lyrics')}")
-                    output_lines.append("-" * 40)
-                
-                return "\n".join(output_lines)
+                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}\nTask ID: {task_id}\n\nCheck back in a few seconds..."
+                return f"⏳ Status: {status}\nLast checked: {datetime.now().strftime('%H:%M:%S')}"
         else:
-            return f"❌ API Error: {data.get('message', 'Unknown error')}"
+            return f"❌ Polling error: {data.get('msg', 'Unknown error')}"
             
-    except requests.exceptions.RequestException as e:
-        return f"❌ Connection Error: {str(e)}"
+    except Exception as e:
+        return f"❌ Error: {str(e)}"
 
-def auto_poll_task(username="default"):
-    """Auto-poll with loading animation"""
-    if username not in user_tasks:
-        return "No task submitted yet!"
+def format_lyrics(lyrics_data, task_id):
+    """Format lyrics for display"""
+    output = [f"🎵 **Task {task_id} - Generated Lyrics**", ""]
     
-    result = poll_task(username)
+    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."
     
-    # If still processing, show animation
-    if "⏳ Status: processing" in result:
-        return f"🔄 Processing...\n{result}"
+    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 result
+    return "\n".join(output)
 
-# Enhanced Gradio interface
+# 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 using Suno's AI API")
-    
-    with gr.Row():
-        with gr.Column(scale=2):
-            prompt_input = gr.Textbox(
-                label="Lyrics Prompt",
-                placeholder="Enter your lyrics idea (e.g., 'A love song about Paris in spring')",
-                lines=3
-            )
+    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():
-                submit_btn = gr.Button("✨ Generate Lyrics", variant="primary")
-                poll_btn = gr.Button("🔄 Check Status", variant="secondary")
-                auto_poll_btn = gr.Button("🔁 Auto Refresh", variant="secondary")
+                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)
                 
-            gr.Markdown("### How to use:")
-            gr.Markdown("""
-            1. Enter your lyrics idea
-            2. Click **Generate Lyrics** to submit
-            3. Wait 10-30 seconds
-            4. Click **Check Status** to see results
+                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
             """)
             
-        with gr.Column(scale=3):
-            output_area = gr.Textbox(
-                label="Output",
-                lines=20,
-                placeholder="Your generated lyrics will appear here..."
+            webhook_status = gr.Textbox(
+                label="Last Webhook Status",
+                value="No webhooks received yet",
+                lines=5
             )
     
-    # Event handlers
-    submit_btn.click(
-        generate_lyrics,
-        inputs=prompt_input,
-        outputs=output_area
-    )
-    
-    poll_btn.click(
-        poll_task,
-        inputs=None,
-        outputs=output_area
-    )
-    
-    auto_poll_btn.click(
-        auto_poll_task,
-        inputs=None,
-        outputs=output_area,
-        every=5  # Auto-refresh every 5 seconds while visible
-    )
+    # 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 with better configuration
+# 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(
-        share=False,  # Set to True for temporary public link
         server_name="0.0.0.0",
         server_port=7860,
+        share=False,
         show_error=True
     )