haraberget commited on
Commit
c3f6215
Β·
verified Β·
1 Parent(s): c063c00

Update app.py

Browse files
Files changed (1) hide show
  1. app.py +243 -527
app.py CHANGED
@@ -3,685 +3,401 @@ import requests
3
  import os
4
  import json
5
  import uuid
6
- import threading
7
  import time
8
- from datetime import datetime, timedelta
9
  from typing import Dict, Optional
10
- from fastapi import Request
11
 
12
- # Load Suno API key - using SunoKey as you specified
13
  SUNO_KEY = os.environ.get("SunoKey", "")
14
  if not SUNO_KEY:
15
  print("⚠️ Warning: SunoKey environment variable not set!")
16
 
17
- # Configuration
18
- POLLING_INTERVAL = 5 # seconds
19
- MAX_POLLING_TIME = 300 # 5 minutes max
20
- WEBHOOK_TIMEOUT = 30 # seconds to wait for webhook before starting polling
21
 
22
- # For Hugging Face Spaces - get the public URL
23
- SPACE_NAME = os.environ.get("SPACE_NAME", "")
24
- AUTHOR_NAME = os.environ.get("AUTHOR_NAME", "")
25
- if SPACE_NAME and AUTHOR_NAME:
26
- SPACE_URL = f"https://{AUTHOR_NAME}-{SPACE_NAME}.hf.space"
27
- elif os.environ.get("SYSTEM") == "spaces":
28
- # Try to construct from environment
29
- SPACE_URL = f"https://{os.environ.get('SPACE_ID', '').replace('_', '-')}.hf.space"
30
- else:
31
- SPACE_URL = "http://localhost:7860"
32
-
33
- # Task storage with automatic cleanup
34
- class TaskManager:
35
- def __init__(self):
36
- self.tasks: Dict[str, dict] = {}
37
- self._lock = threading.Lock()
38
-
39
- def create_task(self, prompt: str, callback_url: str = "") -> str:
40
- """Create a new task and start background monitoring"""
41
- task_id = str(uuid.uuid4())[:8]
42
-
43
- with self._lock:
44
- self.tasks[task_id] = {
45
- "id": task_id,
46
- "prompt": prompt,
47
- "callback_url": callback_url,
48
- "api_task_id": None,
49
- "status": "created",
50
- "result": None,
51
- "error": None,
52
- "created_at": datetime.now(),
53
- "last_checked": None,
54
- "method": None, # 'webhook' or 'polling'
55
- "webhook_received": False,
56
- "polling_started": False,
57
- "completed": False
58
- }
59
-
60
- return task_id
61
-
62
- def update_task(self, task_id: str, **kwargs):
63
- """Update task properties"""
64
- with self._lock:
65
- if task_id in self.tasks:
66
- self.tasks[task_id].update(kwargs)
67
- self.tasks[task_id]["last_checked"] = datetime.now()
68
-
69
- def get_task(self, task_id: str) -> Optional[dict]:
70
- """Get task by ID"""
71
- with self._lock:
72
- return self.tasks.get(task_id)
73
-
74
- def get_all_tasks(self):
75
- """Get all tasks"""
76
- with self._lock:
77
- return list(self.tasks.values())
78
-
79
- def cleanup_old_tasks(self, hours=24):
80
- """Remove tasks older than specified hours"""
81
- cutoff = datetime.now() - timedelta(hours=hours)
82
- with self._lock:
83
- to_delete = [
84
- task_id for task_id, task in self.tasks.items()
85
- if task["created_at"] < cutoff
86
- ]
87
- for task_id in to_delete:
88
- del self.tasks[task_id]
89
- return len(to_delete)
90
-
91
- task_manager = TaskManager()
92
-
93
- # Background polling thread
94
- class PollingThread(threading.Thread):
95
- def __init__(self, task_id: str):
96
- super().__init__(daemon=True)
97
- self.task_id = task_id
98
- self.stop_event = threading.Event()
99
-
100
- def run(self):
101
- """Poll Suno API for task completion"""
102
- task = task_manager.get_task(self.task_id)
103
- if not task or not task.get("api_task_id"):
104
- return
105
-
106
- api_task_id = task["api_task_id"]
107
- start_time = datetime.now()
108
-
109
- while not self.stop_event.is_set():
110
- # Check if we've been polling too long
111
- if (datetime.now() - start_time).seconds > MAX_POLLING_TIME:
112
- task_manager.update_task(
113
- self.task_id,
114
- status="timeout",
115
- error="Polling timeout - task took too long",
116
- completed=True
117
- )
118
- break
119
-
120
- try:
121
- # Poll the API
122
- result = _poll_suno_api(api_task_id)
123
-
124
- if result["status"] == "completed":
125
- task_manager.update_task(
126
- self.task_id,
127
- status="completed",
128
- result=result.get("data"),
129
- method="polling",
130
- completed=True
131
- )
132
- break
133
- elif result["status"] == "failed":
134
- task_manager.update_task(
135
- self.task_id,
136
- status="failed",
137
- error=result.get("error", "Unknown error"),
138
- completed=True
139
- )
140
- break
141
- else:
142
- # Still processing
143
- task_manager.update_task(
144
- self.task_id,
145
- status=result["status"]
146
- )
147
-
148
- # Wait before next poll
149
- self.stop_event.wait(POLLING_INTERVAL)
150
-
151
- except Exception as e:
152
- task_manager.update_task(
153
- self.task_id,
154
- status="polling_error",
155
- error=f"Polling error: {str(e)}",
156
- last_checked=datetime.now()
157
- )
158
- # Wait before retry
159
- self.stop_event.wait(POLLING_INTERVAL * 2)
160
-
161
- def stop(self):
162
- self.stop_event.set()
163
-
164
- def _poll_suno_api(api_task_id: str) -> dict:
165
- """Internal function to poll Suno API"""
166
  if not SUNO_KEY:
167
- raise ValueError("SunoKey not configured")
168
-
169
- url = f"https://api.sunoapi.org/api/v1/lyrics/details?taskId={api_task_id}"
170
- headers = {"Authorization": f"Bearer {SUNO_KEY}"}
171
 
172
- response = requests.get(url, headers=headers, timeout=30)
173
- response.raise_for_status()
174
- data = response.json()
175
 
176
- if data.get("code") == 200 and "data" in data:
177
- task_data = data["data"]
178
- return {
179
- "status": task_data.get("status", "unknown"),
180
- "data": task_data.get("data"),
181
- "error": task_data.get("error")
182
- }
183
- else:
184
- raise Exception(data.get("msg", "API polling failed"))
185
-
186
- def submit_to_suno(prompt: str, task_id: str, callback_url: str = "") -> dict:
187
- """Submit task to Suno API with error handling"""
188
- if not SUNO_KEY:
189
- raise ValueError("SunoKey environment variable not set")
190
 
 
191
  url = "https://api.sunoapi.org/api/v1/lyrics"
192
  headers = {
193
  "Authorization": f"Bearer {SUNO_KEY}",
194
  "Content-Type": "application/json"
195
  }
196
 
197
- # If no callback_url provided, use a dummy one
198
- # Suno requires a URL, but we'll rely on polling
199
- effective_callback = callback_url or "https://dummy.webhook.url/not-used"
200
 
201
  payload = {
202
  "prompt": prompt,
203
- "callBackUrl": effective_callback
204
  }
205
 
206
  try:
 
207
  response = requests.post(url, headers=headers, json=payload, timeout=30)
208
- response.raise_for_status()
209
  data = response.json()
210
 
211
- if data.get("code") == 200 and "data" in data:
212
  api_task_id = data["data"]["taskId"]
213
 
214
- # Start background polling with delay (give webhook a chance)
215
- polling_thread = PollingThread(task_id)
216
-
217
- # Start polling after WEBHOOK_TIMEOUT seconds
218
- def start_polling():
219
- time.sleep(WEBHOOK_TIMEOUT)
220
- task = task_manager.get_task(task_id)
221
- if task and not task.get("webhook_received") and not task.get("polling_started"):
222
- task_manager.update_task(task_id, polling_started=True)
223
- polling_thread.start()
224
-
225
- threading.Thread(target=start_polling, daemon=True).start()
226
-
227
- return {
228
- "success": True,
229
  "api_task_id": api_task_id,
230
- "message": "Task submitted successfully"
231
- }
232
- else:
233
- error_msg = data.get("msg", "Unknown API error")
234
- return {
235
- "success": False,
236
- "error": f"API Error: {error_msg}",
237
- "code": data.get("code")
238
  }
239
-
240
- except requests.exceptions.Timeout:
241
- return {
242
- "success": False,
243
- "error": "Request timeout - Suno API is not responding"
244
- }
245
- except requests.exceptions.ConnectionError:
246
- return {
247
- "success": False,
248
- "error": "Connection error - Check your internet connection"
249
- }
250
- except requests.exceptions.HTTPError as e:
251
- return {
252
- "success": False,
253
- "error": f"HTTP Error: {e.response.status_code if e.response else 'Unknown'}"
254
- }
255
- except Exception as e:
256
- return {
257
- "success": False,
258
- "error": f"Unexpected error: {str(e)}"
259
- }
260
-
261
- def generate_lyrics(prompt: str, use_callback: bool = True):
262
- """Main function to generate lyrics with hybrid approach"""
263
- if not prompt or not prompt.strip():
264
- return "❌ Please enter a prompt"
265
-
266
- if not SUNO_KEY:
267
- return "❌ Error: SunoKey environment variable not set. Please configure it in Settings."
268
-
269
- # Create task
270
- callback_url = f"{SPACE_URL}/callback" if use_callback else ""
271
- task_id = task_manager.create_task(prompt, callback_url)
272
-
273
- try:
274
- # Submit to Suno
275
- result = submit_to_suno(prompt, task_id, callback_url)
276
-
277
- if result["success"]:
278
- # Update task with API task ID
279
- task_manager.update_task(
280
- task_id,
281
- api_task_id=result["api_task_id"],
282
- status="submitted",
283
- method="webhook" if use_callback else "polling"
284
- )
285
 
286
  return f"""βœ… **Task Submitted Successfully!**
287
 
288
  **Your Task ID:** `{task_id}`
289
- **API Task ID:** `{result['api_task_id']}`
290
- **Method:** {'Webhook + Polling Fallback' if use_callback else 'Polling Only'}
291
- **Status:** Submitted - Processing...
292
 
293
  πŸ“ **Prompt:** {prompt[:100]}{'...' if len(prompt) > 100 else ''}
294
 
295
- ⏳ **What happens next:**
296
- 1. Task submitted to Suno AI
297
- 2. {'Webhook expected within 30 seconds' if use_callback else 'Polling started'}
298
- 3. Results will appear automatically
299
- 4. You can also check status manually
300
 
301
- πŸ†” **Save this Task ID:** {task_id}"""
302
 
303
  else:
304
- # Submission failed
305
- task_manager.update_task(
306
- task_id,
307
- status="submission_failed",
308
- error=result["error"],
309
- completed=True
310
- )
311
-
312
  return f"""❌ **Submission Failed**
313
 
314
- **Task ID:** `{task_id}`
315
- **Error:** {result['error']}
316
 
317
  πŸ’‘ **Possible solutions:**
318
- β€’ Check your SunoKey is valid
319
  β€’ Try a different prompt
320
- β€’ Wait a few minutes and try again"""
321
 
 
 
 
 
 
 
322
  except Exception as e:
323
- error_msg = str(e)
324
- task_manager.update_task(
325
- task_id,
326
- status="exception",
327
- error=error_msg,
328
- completed=True
329
- )
330
- return f"❌ Exception occurred: {error_msg}"
331
 
332
- def check_task_status(task_id: str):
333
- """Check status of a specific task"""
334
  if not task_id or not task_id.strip():
335
  return "❌ Please enter a Task ID"
336
 
337
- task = task_manager.get_task(task_id)
338
- if not task:
339
- return f"❌ Task ID `{task_id}` not found. Please check and try again."
340
 
341
- status = task["status"]
342
- method = task.get("method", "unknown")
343
- elapsed = (datetime.now() - task["created_at"]).seconds
 
 
 
344
 
345
- # Format status display
346
- if task["completed"]:
347
- if status == "completed" and task["result"]:
348
- return format_lyrics_result(task["result"], task_id, method)
349
- elif status == "failed" or status == "error":
350
- return f"""❌ **Task Failed**
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
351
 
352
  **Task ID:** `{task_id}`
353
- **Status:** {status}
354
- **Error:** {task.get('error', 'Unknown error')}
355
- **Method:** {method}
356
- **Elapsed:** {elapsed} seconds
357
 
358
- πŸ’‘ Please try generating again."""
359
- else:
360
- return f"πŸ“Š **Task Completed with status:** {status}"
361
-
362
- else:
363
- # Task still in progress
364
- status_icon = {
365
- "created": "πŸ†•",
366
- "submitted": "⏳",
367
- "processing": "πŸ”„",
368
- "pending": "⏱️"
369
- }.get(status, "❓")
370
-
371
- return f"""{status_icon} **Task In Progress**
372
 
373
  **Task ID:** `{task_id}`
374
  **Status:** {status}
375
- **Method:** {method}
376
- **Elapsed:** {elapsed} seconds
377
- **Last checked:** {task.get('last_checked', 'Never')}
378
 
379
- ⏳ **Processing...** Please check back in a few seconds.
 
380
 
381
- πŸ” **Auto-refreshing in 5 seconds...**"""
 
 
 
 
 
 
 
382
 
383
- def format_lyrics_result(lyrics_data, task_id, method):
384
- """Format lyrics for display"""
385
  if not lyrics_data:
386
  return "βœ… Task completed but no lyrics data received"
387
 
388
- output = [f"🎡 **Generated Lyrics (Task: {task_id})**", ""]
 
 
 
 
 
389
 
390
  for i, item in enumerate(lyrics_data, 1):
391
  title = item.get('title', f'Variant {i}')
392
- text = item.get('text', 'No lyrics generated')
393
 
394
- output.append(f"### Variant {i}: {title}")
395
- output.append("```")
396
- output.append(text)
397
- output.append("```")
398
- output.append("---")
 
399
 
400
- output.append(f"\nβœ… **Completed via:** {method}")
401
- output.append(f"πŸ•’ **Finished at:** {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
402
 
403
- return "\n".join(output)
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
404
 
405
  def list_all_tasks():
406
- """List all tasks with status"""
407
- tasks = task_manager.get_all_tasks()
408
- if not tasks:
409
  return "πŸ“­ No tasks found. Generate some lyrics first!"
410
 
411
- # Sort by creation time (newest first)
412
- tasks.sort(key=lambda x: x["created_at"], reverse=True)
413
 
414
- output = ["πŸ“‹ **All Tasks**", ""]
415
-
416
- for task in tasks:
 
 
417
  # Status icon
418
- if task["completed"]:
419
- icon = "βœ…" if task["status"] == "completed" else "❌"
 
 
420
  else:
421
  icon = "⏳"
422
 
423
- elapsed = (datetime.now() - task["created_at"]).seconds
424
- method = task.get("method", "unknown")
 
425
 
426
- output.append(f"{icon} **{task['id']}** - {task['status']}")
427
- output.append(f" Prompt: {task['prompt'][:50]}...")
428
- output.append(f" Method: {method} | Age: {elapsed}s")
429
 
430
- if task.get("error"):
431
- output.append(f" Error: {task['error'][:50]}...")
432
-
433
- output.append("")
434
 
435
- # Cleanup old tasks periodically
436
- cleaned = task_manager.cleanup_old_tasks(hours=1)
437
- if cleaned:
438
- output.append(f"🧹 Cleaned up {cleaned} old tasks")
439
 
440
- return "\n".join(output)
441
 
442
- # Webhook handler function for Gradio 6.0+
443
- async def webhook_handler(request: Request):
444
- """Handle incoming webhook from Suno"""
445
- try:
446
- # Get the raw request body
447
- body = await request.body()
448
- print(f"πŸ“₯ Webhook received: {request.method}")
449
-
450
- # Try to parse JSON
451
- try:
452
- data = json.loads(body.decode('utf-8'))
453
- except:
454
- # Try form data
455
- form_data = await request.form()
456
- data = dict(form_data) if form_data else {}
457
-
458
- print(f"πŸ“¦ Webhook data received")
459
-
460
- # Extract task info (adjust based on Suno's actual format)
461
- if isinstance(data, dict):
462
- print(f"Data structure: {json.dumps(data, indent=2)[:500]}...")
463
-
464
- # Try different possible response formats
465
- api_data = None
466
- if "data" in data:
467
- api_data = data["data"]
468
- elif "task" in data:
469
- api_data = data["task"]
470
- elif "result" in data:
471
- api_data = data["result"]
472
-
473
- if api_data and isinstance(api_data, dict):
474
- api_task_id = api_data.get("taskId") or api_data.get("id")
475
- status = api_data.get("status", "unknown")
476
-
477
- if api_task_id:
478
- # Find task by API task ID
479
- for task in task_manager.get_all_tasks():
480
- if task.get("api_task_id") == api_task_id:
481
- task_id = task["id"]
482
-
483
- if status == "completed":
484
- lyrics_data = api_data.get("data") or api_data.get("lyrics") or api_data.get("result")
485
- task_manager.update_task(
486
- task_id,
487
- status="completed",
488
- result=lyrics_data,
489
- webhook_received=True,
490
- method="webhook",
491
- completed=True
492
- )
493
- print(f"βœ… Webhook completed task {task_id}")
494
- return {"status": "success", "task_id": task_id}
495
-
496
- elif status == "failed":
497
- error_msg = api_data.get("error", "Unknown error")
498
- task_manager.update_task(
499
- task_id,
500
- status="failed",
501
- error=error_msg,
502
- webhook_received=True,
503
- completed=True
504
- )
505
- return {"status": "error", "task_id": task_id, "error": error_msg}
506
-
507
- else:
508
- task_manager.update_task(task_id, status=status)
509
- return {"status": "updated", "task_id": task_id}
510
-
511
- return {"status": "no_match", "message": "Task not found in database"}
512
 
513
- except Exception as e:
514
- print(f"❌ Webhook error: {str(e)}")
515
- return {"status": "error", "message": str(e)}
516
 
517
- # Create Gradio app
518
- with gr.Blocks() as app:
519
  gr.Markdown("# 🎡 Suno AI Lyrics Generator")
520
- gr.Markdown("Hybrid solution: Webhooks with automatic polling fallback")
521
 
522
  with gr.Tabs():
 
523
  with gr.TabItem("✨ Generate"):
524
  with gr.Row():
525
  with gr.Column(scale=2):
 
526
  prompt_input = gr.Textbox(
527
  label="Lyrics Prompt",
528
- placeholder="A jazz song about rainy nights in Tokyo...",
529
  lines=4
530
  )
531
 
532
- use_webhook = gr.Checkbox(
533
- label="Enable webhooks (recommended)",
534
- value=True,
535
- info="If webhook fails, automatic polling will start"
536
- )
537
 
538
- submit_btn = gr.Button("πŸš€ Generate Lyrics", variant="primary", scale=1)
539
-
540
- gr.Markdown("### πŸ“‹ Instructions:")
541
  gr.Markdown("""
542
- 1. Enter your lyrics idea
543
- 2. Submit (webhooks enabled by default)
544
- 3. Save your Task ID
545
- 4. Check status in the "Status" tab
546
- 5. Results arrive automatically
547
-
548
- **πŸ”§ Behind the scenes:**
549
- - Webhook attempted first
550
- - If no response in 30s, polling starts
551
- - Polling continues until completion
552
- - Results available in both tabs
553
  """)
554
 
555
  with gr.Column(scale=3):
556
- output_result = gr.Markdown(
557
- label="Submission Result",
558
- value="Your results will appear here..."
559
  )
560
 
561
  submit_btn.click(
562
- generate_lyrics,
563
- inputs=[prompt_input, use_webhook],
564
- outputs=output_result
565
  )
566
 
 
567
  with gr.TabItem("πŸ” Check Status"):
568
  with gr.Row():
569
  with gr.Column():
 
570
  task_id_input = gr.Textbox(
571
  label="Task ID",
572
- placeholder="Enter your Task ID (e.g., a1b2c3d4)"
 
573
  )
574
 
575
  with gr.Row():
576
- check_btn = gr.Button("πŸ”Ž Check Status", variant="primary")
577
- refresh_btn = gr.Button("πŸ”„ Auto-refresh", variant="secondary")
578
- list_all_btn = gr.Button("πŸ“‹ All Tasks")
579
 
580
- task_list_output = gr.Textbox(
581
- label="All Tasks",
582
- lines=10,
583
- interactive=False
584
- )
585
 
586
  with gr.Column():
587
  status_output = gr.Markdown(
588
- label="Task Status",
589
- value="Enter a Task ID and click Check Status"
590
  )
591
 
592
- # Event handlers
593
  check_btn.click(
594
- check_task_status,
595
  inputs=task_id_input,
596
  outputs=status_output
597
  )
598
 
599
- refresh_btn.click(
600
- check_task_status,
601
  inputs=task_id_input,
602
- outputs=status_output
 
 
 
 
 
 
 
 
603
  )
604
 
605
- list_all_btn.click(
606
- list_all_tasks,
607
  inputs=None,
608
- outputs=task_list_output
609
  )
610
 
611
- with gr.TabItem("βš™οΈ Settings & Info"):
612
- gr.Markdown("### βš™οΈ Configuration")
 
613
 
614
  with gr.Row():
615
  with gr.Column():
616
- gr.Markdown("**πŸ”‘ API Status:**")
617
- api_status = gr.Markdown(
618
- value=f"βœ… SunoKey: {'Configured' if SUNO_KEY else '❌ NOT SET'}"
619
- )
620
 
621
- gr.Markdown(f"**🌐 Space URL:** `{SPACE_URL}`")
622
-
623
- gr.Markdown("**πŸ”„ Webhook Endpoint:**")
624
- webhook_info = gr.Textbox(
625
- value=f"{SPACE_URL}/callback",
626
- interactive=False
627
- )
628
 
629
  with gr.Column():
630
- gr.Markdown("**πŸ“Š Statistics:**")
631
- stats_text = gr.Textbox(
632
- value=f"Active tasks: {len(task_manager.get_all_tasks())}",
633
- interactive=False
634
- )
635
 
636
- cleanup_btn = gr.Button("🧹 Cleanup Old Tasks", variant="secondary")
637
- cleanup_output = gr.Textbox(label="Cleanup Result", interactive=False)
638
-
639
- gr.Markdown("### πŸ“š How It Works:")
640
- gr.Markdown(f"""
641
- **Hybrid Approach:**
642
- 1. Submit with webhook URL (required by Suno API)
643
- 2. Wait 30 seconds for webhook response
644
- 3. If no webhook, start background polling
645
- 4. Poll every 5 seconds until completion
646
- 5. Return results via polling if webhook fails
647
 
648
- **Your Webhook URL:** `{SPACE_URL}/callback`
 
 
649
 
650
- **Error Recovery:**
651
- β€’ Network issues β†’ automatic retry
652
- β€’ API errors β†’ informative messages
653
- β€’ Timeouts β†’ fallback to polling
654
- β€’ All errors β†’ logged for debugging
655
  """)
656
-
657
- cleanup_btn.click(
658
- lambda: f"Cleaned {task_manager.cleanup_old_tasks(hours=1)} old tasks",
659
- inputs=None,
660
- outputs=cleanup_output
661
- )
662
-
663
- # Add the webhook route using the new Gradio 6.0 method
664
- app.add_api_route("/callback", webhook_handler, methods=["POST"])
665
-
666
- # For backward compatibility, also add a simple GET endpoint
667
- @app.get("/")
668
- def home():
669
- return {"status": "running", "app": "Suno Lyrics Generator"}
670
 
671
  # Launch the app
672
  if __name__ == "__main__":
 
 
673
  print("πŸš€ Starting Suno Lyrics Generator...")
674
- print(f"πŸ”‘ SunoKey: {'βœ… Set' if SUNO_KEY else '❌ NOT SET'}")
675
- print(f"🌐 Space URL: {SPACE_URL}")
676
- print(f"🌐 Webhook endpoint: {SPACE_URL}/callback")
677
- print(f"⏱️ Polling fallback: {WEBHOOK_TIMEOUT}s delay")
678
 
679
  app.launch(
680
  server_name="0.0.0.0",
681
  server_port=7860,
682
  share=False,
683
- show_error=True,
684
  debug=False,
685
- title="Suno Lyrics Generator",
686
- theme="soft"
687
  )
 
3
  import os
4
  import json
5
  import uuid
 
6
  import time
7
+ from datetime import datetime
8
  from typing import Dict, Optional
 
9
 
10
+ # Load Suno API key
11
  SUNO_KEY = os.environ.get("SunoKey", "")
12
  if not SUNO_KEY:
13
  print("⚠️ Warning: SunoKey environment variable not set!")
14
 
15
+ # Task storage
16
+ tasks_db = {}
 
 
17
 
18
+ def generate_lyrics(prompt: str) -> str:
19
+ """Submit lyrics generation task to Suno API"""
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
20
  if not SUNO_KEY:
21
+ return "❌ Error: SunoKey environment variable not set. Please add it in Space Settings."
 
 
 
22
 
23
+ if not prompt or not prompt.strip():
24
+ return "❌ Please enter a lyrics prompt"
 
25
 
26
+ # Generate task ID
27
+ task_id = str(uuid.uuid4())[:8]
 
 
 
 
 
 
 
 
 
 
 
 
28
 
29
+ # Prepare API request
30
  url = "https://api.sunoapi.org/api/v1/lyrics"
31
  headers = {
32
  "Authorization": f"Bearer {SUNO_KEY}",
33
  "Content-Type": "application/json"
34
  }
35
 
36
+ # Suno requires a callback URL, but we'll use a dummy one and rely on polling
37
+ dummy_callback = "https://dummy.callback.url/not-used"
 
38
 
39
  payload = {
40
  "prompt": prompt,
41
+ "callBackUrl": dummy_callback
42
  }
43
 
44
  try:
45
+ # Submit task
46
  response = requests.post(url, headers=headers, json=payload, timeout=30)
 
47
  data = response.json()
48
 
49
+ if response.status_code == 200 and data.get("code") == 200:
50
  api_task_id = data["data"]["taskId"]
51
 
52
+ # Store task information
53
+ tasks_db[task_id] = {
54
+ "id": task_id,
 
 
 
 
 
 
 
 
 
 
 
 
55
  "api_task_id": api_task_id,
56
+ "prompt": prompt,
57
+ "status": "submitted",
58
+ "result": None,
59
+ "error": None,
60
+ "created_at": datetime.now().isoformat(),
61
+ "last_checked": None
 
 
62
  }
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
63
 
64
  return f"""βœ… **Task Submitted Successfully!**
65
 
66
  **Your Task ID:** `{task_id}`
67
+ **API Task ID:** `{api_task_id}`
 
 
68
 
69
  πŸ“ **Prompt:** {prompt[:100]}{'...' if len(prompt) > 100 else ''}
70
 
71
+ ⏳ **Next Steps:**
72
+ 1. Task is now processing with Suno AI
73
+ 2. This usually takes 10-30 seconds
74
+ 3. Click **"Check Status"** tab to check progress
75
+ 4. Use your Task ID: `{task_id}`
76
 
77
+ πŸ’‘ **Tip:** Save your Task ID to check results later!"""
78
 
79
  else:
80
+ error_msg = data.get("msg", f"HTTP {response.status_code}")
 
 
 
 
 
 
 
81
  return f"""❌ **Submission Failed**
82
 
83
+ **Error:** {error_msg}
 
84
 
85
  πŸ’‘ **Possible solutions:**
86
+ β€’ Check if your SunoKey is valid
87
  β€’ Try a different prompt
88
+ β€’ Wait a few minutes and retry"""
89
 
90
+ except requests.exceptions.Timeout:
91
+ return "❌ Error: Request timeout - Suno API is not responding"
92
+ except requests.exceptions.ConnectionError:
93
+ return "❌ Error: Connection failed - Check your internet connection"
94
+ except requests.exceptions.HTTPError as e:
95
+ return f"❌ Error: HTTP {e.response.status_code if e.response else 'Unknown'}"
96
  except Exception as e:
97
+ return f"❌ Error: {str(e)}"
 
 
 
 
 
 
 
98
 
99
+ def check_task_status(task_id: str) -> str:
100
+ """Check the status of a task using polling"""
101
  if not task_id or not task_id.strip():
102
  return "❌ Please enter a Task ID"
103
 
104
+ if task_id not in tasks_db:
105
+ return f"❌ Task ID `{task_id}` not found. Please submit a task first."
 
106
 
107
+ task = tasks_db[task_id]
108
+ api_task_id = task["api_task_id"]
109
+
110
+ # Poll the Suno API
111
+ url = f"https://api.sunoapi.org/api/v1/lyrics/details?taskId={api_task_id}"
112
+ headers = {"Authorization": f"Bearer {SUNO_KEY}"}
113
 
114
+ try:
115
+ response = requests.get(url, headers=headers, timeout=30)
116
+ data = response.json()
117
+
118
+ if response.status_code == 200 and data.get("code") == 200:
119
+ task_data = data["data"]
120
+ status = task_data.get("status", "unknown")
121
+
122
+ # Update task status
123
+ tasks_db[task_id]["status"] = status
124
+ tasks_db[task_id]["last_checked"] = datetime.now().isoformat()
125
+
126
+ if status == "completed" and "data" in task_data:
127
+ # Task completed successfully
128
+ lyrics_data = task_data["data"]
129
+ tasks_db[task_id]["result"] = lyrics_data
130
+
131
+ # Format the output
132
+ return format_lyrics_output(lyrics_data, task_id)
133
+
134
+ elif status == "failed":
135
+ error_msg = task_data.get("error", "Unknown error")
136
+ tasks_db[task_id]["error"] = error_msg
137
+ return f"""❌ **Task Failed**
138
 
139
  **Task ID:** `{task_id}`
140
+ **Error:** {error_msg}
 
 
 
141
 
142
+ πŸ’‘ Please try generating again with a different prompt."""
143
+
144
+ else:
145
+ # Still processing
146
+ elapsed = time_since(task["created_at"])
147
+ return f"""⏳ **Task Processing...**
 
 
 
 
 
 
 
 
148
 
149
  **Task ID:** `{task_id}`
150
  **Status:** {status}
151
+ **Elapsed:** {elapsed}
 
 
152
 
153
+ ⏰ **Estimated time:** 10-30 seconds
154
+ πŸ”„ **Auto-refresh recommended**
155
 
156
+ πŸ’‘ Check back in a few seconds!"""
157
+
158
+ else:
159
+ error_msg = data.get("msg", f"HTTP {response.status_code}")
160
+ return f"❌ Error checking status: {error_msg}"
161
+
162
+ except Exception as e:
163
+ return f"❌ Error: {str(e)}"
164
 
165
+ def format_lyrics_output(lyrics_data, task_id):
166
+ """Format the lyrics for display"""
167
  if not lyrics_data:
168
  return "βœ… Task completed but no lyrics data received"
169
 
170
+ output_lines = [
171
+ f"# 🎡 Generated Lyrics (Task: {task_id})",
172
+ "",
173
+ f"**Generated at:** {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
174
+ ""
175
+ ]
176
 
177
  for i, item in enumerate(lyrics_data, 1):
178
  title = item.get('title', f'Variant {i}')
179
+ lyrics = item.get('text', 'No lyrics generated')
180
 
181
+ output_lines.append(f"## Variant {i}: {title}")
182
+ output_lines.append("```")
183
+ output_lines.append(lyrics)
184
+ output_lines.append("```")
185
+ output_lines.append("---")
186
+ output_lines.append("")
187
 
188
+ output_lines.append("### πŸŽ‰ All done!")
189
+ output_lines.append("You can generate more lyrics or try different prompts.")
190
 
191
+ return "\n".join(output_lines)
192
+
193
+ def time_since(iso_timestamp: str) -> str:
194
+ """Calculate time elapsed since timestamp"""
195
+ try:
196
+ created = datetime.fromisoformat(iso_timestamp)
197
+ elapsed = datetime.now() - created
198
+ seconds = int(elapsed.total_seconds())
199
+
200
+ if seconds < 60:
201
+ return f"{seconds} seconds"
202
+ elif seconds < 3600:
203
+ minutes = seconds // 60
204
+ return f"{minutes} minutes"
205
+ else:
206
+ hours = seconds // 3600
207
+ return f"{hours} hours"
208
+ except:
209
+ return "Unknown"
210
 
211
  def list_all_tasks():
212
+ """List all submitted tasks"""
213
+ if not tasks_db:
 
214
  return "πŸ“­ No tasks found. Generate some lyrics first!"
215
 
216
+ output_lines = ["# πŸ“‹ All Submitted Tasks", ""]
 
217
 
218
+ for task_id, task in tasks_db.items():
219
+ status = task.get("status", "unknown")
220
+ prompt_preview = task.get("prompt", "")[:50]
221
+ created = task.get("created_at", "")[:19]
222
+
223
  # Status icon
224
+ if status == "completed":
225
+ icon = "βœ…"
226
+ elif status in ["failed", "error"]:
227
+ icon = "❌"
228
  else:
229
  icon = "⏳"
230
 
231
+ output_lines.append(f"{icon} **{task_id}** - {status}")
232
+ output_lines.append(f" Prompt: {prompt_preview}...")
233
+ output_lines.append(f" Created: {created}")
234
 
235
+ if task.get("last_checked"):
236
+ last_checked = task["last_checked"][:19]
237
+ output_lines.append(f" Last checked: {last_checked}")
238
 
239
+ output_lines.append("")
 
 
 
240
 
241
+ # Add summary
242
+ completed = sum(1 for t in tasks_db.values() if t.get("status") == "completed")
243
+ total = len(tasks_db)
244
+ output_lines.append(f"**Summary:** {completed}/{total} tasks completed")
245
 
246
+ return "\n".join(output_lines)
247
 
248
+ def auto_refresh_status(task_id: str):
249
+ """Auto-refresh status with loading animation"""
250
+ if not task_id or task_id not in tasks_db:
251
+ return task_id, "❌ Invalid Task ID"
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
252
 
253
+ result = check_task_status(task_id)
254
+ return task_id, result
 
255
 
256
+ # Create the Gradio interface
257
+ with gr.Blocks(title="Suno Lyrics Generator", theme=gr.themes.Soft()) as app:
258
  gr.Markdown("# 🎡 Suno AI Lyrics Generator")
259
+ gr.Markdown("Generate song lyrics using Suno's AI API")
260
 
261
  with gr.Tabs():
262
+ # Tab 1: Generate Lyrics
263
  with gr.TabItem("✨ Generate"):
264
  with gr.Row():
265
  with gr.Column(scale=2):
266
+ gr.Markdown("### Enter your lyrics idea")
267
  prompt_input = gr.Textbox(
268
  label="Lyrics Prompt",
269
+ placeholder="Example: A romantic ballad about stargazing on a summer night...",
270
  lines=4
271
  )
272
 
273
+ submit_btn = gr.Button("πŸš€ Generate Lyrics", variant="primary", size="lg")
 
 
 
 
274
 
275
+ gr.Markdown("### πŸ“ Tips:")
 
 
276
  gr.Markdown("""
277
+ β€’ Be descriptive in your prompt
278
+ β€’ Include genre, mood, or theme
279
+ β€’ Specify if you want verses, chorus, bridge
280
+ β€’ Typical processing time: 10-30 seconds
 
 
 
 
 
 
 
281
  """)
282
 
283
  with gr.Column(scale=3):
284
+ output_area = gr.Markdown(
285
+ label="Result",
286
+ value="Your task submission result will appear here..."
287
  )
288
 
289
  submit_btn.click(
290
+ fn=generate_lyrics,
291
+ inputs=prompt_input,
292
+ outputs=output_area
293
  )
294
 
295
+ # Tab 2: Check Status
296
  with gr.TabItem("πŸ” Check Status"):
297
  with gr.Row():
298
  with gr.Column():
299
+ gr.Markdown("### Enter your Task ID")
300
  task_id_input = gr.Textbox(
301
  label="Task ID",
302
+ placeholder="Paste your Task ID here (e.g., a1b2c3d4)",
303
+ scale=1
304
  )
305
 
306
  with gr.Row():
307
+ check_btn = gr.Button("πŸ” Check Status", variant="primary")
308
+ auto_refresh_btn = gr.Button("πŸ”„ Auto-refresh", variant="secondary")
 
309
 
310
+ gr.Markdown("---")
311
+ refresh_all_btn = gr.Button("πŸ“‹ List All Tasks")
312
+ tasks_list = gr.Markdown(label="All Tasks")
 
 
313
 
314
  with gr.Column():
315
  status_output = gr.Markdown(
316
+ label="Status",
317
+ value="Enter a Task ID above and click Check Status"
318
  )
319
 
320
+ # Connect buttons to functions
321
  check_btn.click(
322
+ fn=check_task_status,
323
  inputs=task_id_input,
324
  outputs=status_output
325
  )
326
 
327
+ auto_refresh_btn.click(
328
+ fn=auto_refresh_status,
329
  inputs=task_id_input,
330
+ outputs=[task_id_input, status_output]
331
+ ).then(
332
+ fn=lambda: time.sleep(3),
333
+ inputs=None,
334
+ outputs=None
335
+ ).then(
336
+ fn=auto_refresh_status,
337
+ inputs=task_id_input,
338
+ outputs=[task_id_input, status_output]
339
  )
340
 
341
+ refresh_all_btn.click(
342
+ fn=list_all_tasks,
343
  inputs=None,
344
+ outputs=tasks_list
345
  )
346
 
347
+ # Tab 3: Help & Info
348
+ with gr.TabItem("ℹ️ Help"):
349
+ gr.Markdown("# Help & Information")
350
 
351
  with gr.Row():
352
  with gr.Column():
353
+ gr.Markdown("### πŸ”‘ API Status")
354
+ api_status = "βœ… Configured" if SUNO_KEY else "❌ Not configured"
355
+ gr.Markdown(f"**SunoKey:** {api_status}")
 
356
 
357
+ gr.Markdown("### πŸ“‹ How to Use:")
358
+ gr.Markdown("""
359
+ 1. **Generate Tab:** Enter a lyrics prompt and submit
360
+ 2. **Save your Task ID** from the result
361
+ 3. **Check Status Tab:** Paste your Task ID to check progress
362
+ 4. **Results:** Lyrics will appear when processing is complete
363
+ """)
364
 
365
  with gr.Column():
366
+ gr.Markdown("### 🚨 Troubleshooting")
367
+ gr.Markdown("""
368
+ **Common Issues:**
 
 
369
 
370
+ β€’ **"SunoKey not set":** Add your API key in Space Settings β†’ Repository secrets
371
+ β€’ **"Task ID not found":** Submit a new task and save the ID
372
+ β€’ **Long processing time:** Suno API can take 10-30 seconds
373
+ β€’ **API errors:** Check if your API key is valid and has credits
374
+
375
+ **For best results:**
376
+ - Use descriptive prompts
377
+ - Check status after 15-20 seconds
378
+ - Save your Task IDs
379
+ """)
 
380
 
381
+ gr.Markdown("### πŸ“ž Support")
382
+ gr.Markdown("""
383
+ If you continue to have issues:
384
 
385
+ 1. **Check your SunoKey** is correct and has available credits
386
+ 2. **Try a simpler prompt** to test the API
387
+ 3. **Wait a few minutes** if the API seems busy
388
+ 4. **Contact Suno API support** for API-specific issues
 
389
  """)
 
 
 
 
 
 
 
 
 
 
 
 
 
 
390
 
391
  # Launch the app
392
  if __name__ == "__main__":
393
+ print(f"πŸ”‘ SunoKey status: {'Configured' if SUNO_KEY else 'NOT SET'}")
394
+ print(f"πŸ“Š Tasks in memory: {len(tasks_db)}")
395
  print("πŸš€ Starting Suno Lyrics Generator...")
 
 
 
 
396
 
397
  app.launch(
398
  server_name="0.0.0.0",
399
  server_port=7860,
400
  share=False,
 
401
  debug=False,
402
+ show_error=True
 
403
  )