import logging import os import re import urllib.parse import requests import slack_sdk from bs4 import BeautifulSoup from dotenv import load_dotenv from slack_bolt import App load_dotenv() logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", handlers=[logging.StreamHandler()], ) logger = logging.getLogger(__name__) app = App( token=os.environ.get("SLACK_BOT_TOKEN"), signing_secret=os.environ.get("SLACK_SIGNING_SECRET"), ) CLIPS_CHANNEL_ID = "C051J3QFXQA" SUNO_API_TOKEN = os.environ.get("SUNO_API_TOKEN", "") API_URL = "https://studio-api.prod.suno.com/api/playlist/update_clips/" PLAYLIST_ID = "ee2e603d-68f1-4af3-8d88-d1291eef0fe8" # Clip City playlist for the Clipzilla user SLACK_WORKSPACE_DOMAIN = os.environ.get("SLACK_WORKSPACE_DOMAIN", "suno-main") # Default fallback # Test monitoring configuration SUNO_TEST_RUNS_CHANNEL = "suno-test-runs" # Channel to monitor for Ranger messages TECH_ALERTS_CHANNEL = "tech-alerts" # Channel to send alerts to TEST_FAILURE_PATTERN = re.compile(r"(\d+)\s+tests?\s+failed", re.IGNORECASE) # Domain to API base URL mapping DOMAINS = { "suno.com": "https://studio-api.prod.suno.com", "b.suno.fm": "https://studio-api.staging.suno.com", "a.suno.fm": "https://studio-api.staging.suno.com", } @app.event("link_shared") def on_link_shared(client: slack_sdk.WebClient, event, logger): url = event["links"][0]["url"] logger.info(f"Link shared: {url}") # Parse URL to extract parts parsed = urllib.parse.urlparse(url) domain = parsed.netloc.replace("www.", "") path = parsed.path # Get API base URL based on domain api_base = DOMAINS.get(domain, DOMAINS["suno.com"]) # Extract song ID from URL song_id = None if "/s/" in path: # Handle short link share_id = path.split("/s/")[1].strip("/") try: resp = requests.get(f"{api_base}/api/share/code/{share_id}") if resp.status_code == 200: data = resp.json() if data.get("success") and data.get("content_type") == "song": song_id = data.get("content_id") logger.info(f"Resolved share {share_id} to song {song_id}") except Exception as e: logger.error(f"Error resolving share: {e}") elif "/song/" in path: # Extract song ID directly song_id = path.split("/song/")[1].strip("/") # If no valid song ID found, return if not song_id or "?" in song_id: song_id = song_id.split("?")[0] if song_id else None if not song_id: return # Get metadata and unfurl try: # Get metadata embed_url = f"https://suno.com/embed/{song_id}" resp = requests.get( embed_url, headers={"x-vercel-protection-bypass": os.environ.get("VERCEL_BYPASS_TOKEN", "")} ) soup = BeautifulSoup(resp.content, "html.parser") img_meta = soup.find("meta", property="og:image") title_meta = soup.find("meta", property="og:title") video_meta = soup.find("meta", property="og:video") # Extract content from meta tags title = "" if title_meta and title_meta.has_attr("content"): title = title_meta["content"] video_url = "" if video_meta and video_meta.has_attr("content"): video_url = video_meta["content"] thumbnail_url = "" if img_meta and img_meta.has_attr("content"): thumbnail_url = img_meta["content"] # Unfurl in Slack client.chat_unfurl( channel=event["channel"], ts=event["message_ts"], unfurls={ url: { "blocks": [ { "type": "video", "title": {"type": "plain_text", "text": title, "emoji": True}, "title_url": url, "description": { "type": "plain_text", "text": "Listen on Suno.", "emoji": True, }, "video_url": video_url, "alt_text": title, "thumbnail_url": thumbnail_url, "author_name": "Suno", "provider_name": "Suno", } ] } }, ) # Add to playlist if in clips channel if event["channel"] == CLIPS_CHANNEL_ID: try: requests.post( API_URL, json={ "playlist_id": PLAYLIST_ID, "update_type": "add", "metadata": {"clip_ids": [song_id]}, }, headers={ "Authorization": f"Bearer {SUNO_API_TOKEN}", "Content-Type": "application/json", }, ) except Exception as e: logger.error(f"Error adding to playlist: {e}") except Exception as e: logger.error(f"Error processing song {song_id}: {e}") @app.message() def handle_ranger_messages(message, client: slack_sdk.WebClient, logger): """Monitor messages from Ranger for test failures""" # Get channel info try: channel_info = client.conversations_info(channel=message["channel"]) channel_name = channel_info["channel"]["name"] except Exception: return # Only process messages from #suno-test-runs channel if channel_name != SUNO_TEST_RUNS_CHANNEL: return message_text = message.get("text", "") logger.info(f"Processing message from Ranger in #{channel_name}: {message_text}") # Search for test failure pattern match = TEST_FAILURE_PATTERN.search(message_text) if match: failed_count = int(match.group(1)) logger.info(f"Found {failed_count} test failures") # Alert if more than 4 tests failed if ( failed_count > 11 ): # this is really high but staging is very slow right now, making tests super flaky # Create link to original message message_link = f"https://{SLACK_WORKSPACE_DOMAIN}.slack.com/archives/{message['channel']}/{message['ts'].replace('.', '')}" alert_message = ( f"🚨 *High Test Failure Alert* 🚨\n\n" f"Ranger reported *{failed_count} tests failed* in #{SUNO_TEST_RUNS_CHANNEL}" ) try: # Send alert to tech-alerts channel client.chat_postMessage( channel=f"#{TECH_ALERTS_CHANNEL}", text=alert_message, blocks=[ { "type": "header", "text": {"type": "plain_text", "text": "🚨 High Test Failure Alert"}, }, { "type": "section", "fields": [ {"type": "mrkdwn", "text": f"*Failed Tests:* {failed_count}"}, {"type": "mrkdwn", "text": f"*Source:* #{SUNO_TEST_RUNS_CHANNEL}"}, ], }, { "type": "section", "text": { "type": "mrkdwn", "text": f"<{message_link}|View original message>", }, }, ], ) logger.info(f"Alert sent to #{TECH_ALERTS_CHANNEL} for {failed_count} test failures") except Exception as e: logger.error(f"Failed to send alert to #{TECH_ALERTS_CHANNEL}: {e}") if __name__ == "__main__": app.start(port=int(os.environ.get("PORT", 3008)))