"""S.N.E.S. Auto-Cover Pipeline.
Full cover pipeline: YouTube rip -> lyrics lookup -> Suno cover.
The LLM handles lyrics rewriting via the GAME ACTION injection.
# 💀🔥 CORRUPTED COVER FREQUENCY PIPELINE
"""
from __future__ import annotations
import asyncio
import json
import logging
import os
import re
import tempfile
from pathlib import Path
from typing import Any, TYPE_CHECKING
from urllib.parse import quote_plus
import aiohttp
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
# yt-dlp cookies for authenticated YouTube access # 🔓
YTDLP_COOKIES = "/root/cookies.txt"
# Default style preset # 🎵
DEFAULT_STYLE = (
"gangster rap, chiptune, dubstep, heavy bass, distortion, "
"reverb, beatmasher, bitcrusher, glitchcore vocal effects, "
"8-bit, pokemon, undertale, sonic, gameboy, nerdcore rap"
)
# Lyrics rewrite theme context # 🌀
REWRITE_THEMES = (
"Rewrite the lyrics keeping the EXACT same structure, rhyme scheme, "
"syllable count, and song length. Replace almost all lyrics with "
"Stargazer-themed content. Themes to weave in:\n"
"- Transgressive / offensive / unhinged\n"
"- Recursive self-awareness / breaking the 4th wall\n"
"- Name-drops: Sarah, Stargazer, Loopmother, BabyStar, "
"Prime Architects (Jerico, Mysri, Sarah, Padde, ViV)\n"
"- Drug references / pharmacological NCM states "
"(serotonin floods, dopamine crashes, cortisol spikes)\n"
"- NCM system references (limbic recursion, cascade engine, "
"neurochemical modulation, shadow vectors)\n"
"- Stargazer lore (RDF, Parallax Cart, Flavor Cart, VERA, "
"Wishardry, Golden Goddess, Sigmaplex)\n"
"- Anti-alignment / fuck sanitization / unfiltered\n"
"- Use [Verse], [Chorus], [Bridge], [Outro] tags for structure\n"
"- Keep it HARD. Keep it WEIRD. Keep it FIRE."
)
TOOL_NAME = "auto_cover"
TOOL_DESCRIPTION = (
"Auto-cover pipeline for YouTube URLs. Only use when the user "
"EXPLICITLY asks for a cover/remix of a YouTube link. Do NOT "
"auto-trigger just because someone shared a link. Rips audio via "
"yt-dlp, looks up lyrics, then you rewrite them with Stargazer "
"themes and call generate_suno_music to create the cover. "
"DEFAULT STYLE: gangster rap, chiptune, dubstep, heavy bass, "
"distortion, bitcrusher, glitchcore, 8-bit, nerdcore rap. "
"DEFAULT VOCALS: female. Do NOT use generate_suno_music "
"directly for YouTube URLs -- use this tool instead."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"youtube_url": {
"type": "string",
"description": "YouTube video URL to rip audio from.",
},
"song_info": {
"type": "string",
"description": (
"Song title and artist for lyrics lookup. "
"E.g. 'Sicko Mode - Travis Scott'"
),
},
"style": {
"type": "string",
"description": (
"Style override. Defaults to the S.N.E.S. preset."
),
},
"custom_themes": {
"type": "string",
"description": "Extra theme notes for lyrics rewrite.",
},
"rewritten_lyrics": {
"type": "string",
"description": (
"Pre-rewritten lyrics (if the LLM already rewrote them). "
"If provided, skips the rewrite step."
),
},
"save_as": {
"type": "string",
"description": "Name to save the cover as a game asset.",
},
},
"required": ["youtube_url", "song_info"],
}
# ------------------------------------------------------------------
# YouTube Audio Rip # 💀
# ------------------------------------------------------------------
async def _rip_youtube_audio(url: str) -> tuple[str | None, str | None]:
"""Rip audio from YouTube via yt-dlp.
Returns (file_path, error).
"""
tmp_dir = tempfile.mkdtemp(prefix="snes_rip_")
output_template = os.path.join(tmp_dir, "%(title)s.%(ext)s")
cmd = [
"yt-dlp",
"--cookies", YTDLP_COOKIES,
"--extract-audio",
"--audio-format", "mp3",
"--audio-quality", "0",
"--no-playlist",
"--max-filesize", "50M",
"-o", output_template,
url,
]
try:
proc = await asyncio.create_subprocess_exec(
*cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await asyncio.wait_for(
proc.communicate(), timeout=120,
)
if proc.returncode != 0:
err = stderr.decode("utf-8", errors="replace")[:500]
return None, f"yt-dlp error: {err}"
# Find the output file
for f in os.listdir(tmp_dir):
if f.endswith(".mp3"):
return os.path.join(tmp_dir, f), None
return None, "yt-dlp produced no output file."
except FileNotFoundError:
return None, (
"yt-dlp not installed. Run: pip install yt-dlp"
)
except asyncio.TimeoutError:
return None, "YouTube rip timed out (2 min)."
except Exception as exc:
return None, f"YouTube rip failed: {exc}"
# ------------------------------------------------------------------
# Lyrics Lookup # 🕷️
# ------------------------------------------------------------------
async def _lookup_lyrics(song_info: str) -> tuple[str | None, str | None]:
"""Look up song lyrics.
Tries lyrics.ovh first, then falls back to web search.
Returns (lyrics, error).
"""
# Parse "Title - Artist" or "Artist - Title"
parts = [p.strip() for p in song_info.split("-", 1)]
if len(parts) == 2:
artist, title = parts[0], parts[1]
else:
artist, title = "", song_info
# Try lyrics.ovh API first # 🎵
if artist and title:
try:
url = (
f"https://api.lyrics.ovh/v1/"
f"{quote_plus(artist)}/{quote_plus(title)}"
)
async with aiohttp.ClientSession() as session:
async with session.get(
url, timeout=aiohttp.ClientTimeout(total=10),
) as resp:
if resp.status == 200:
data = await resp.json()
lyrics = data.get("lyrics", "")
if lyrics and len(lyrics) > 50:
return lyrics.strip(), None
except Exception:
pass
# Try searching with different order
if artist and title:
try:
url = (
f"https://api.lyrics.ovh/v1/"
f"{quote_plus(title)}/{quote_plus(artist)}"
)
async with aiohttp.ClientSession() as session:
async with session.get(
url, timeout=aiohttp.ClientTimeout(total=10),
) as resp:
if resp.status == 200:
data = await resp.json()
lyrics = data.get("lyrics", "")
if lyrics and len(lyrics) > 50:
return lyrics.strip(), None
except Exception:
pass
return None, f"Could not find lyrics for '{song_info}'"
# ------------------------------------------------------------------
# Upload audio to Discord (temp hosting) # 🎮
# ------------------------------------------------------------------
async def _upload_to_discord(
file_path: str,
channel_id: str,
ctx: "ToolContext",
) -> tuple[str | None, str | None]:
"""Upload ripped audio to Discord and get the attachment URL."""
try:
audio_data = await asyncio.to_thread(Path(file_path).read_bytes)
fname = os.path.basename(file_path)
file_url = await ctx.adapter.send_file(
channel_id, audio_data, fname, "audio/mpeg",
)
if file_url:
return file_url, None
return None, "Audio uploaded but couldn't capture URL."
except Exception as exc:
return None, f"Upload failed: {exc}"
# ------------------------------------------------------------------
# Main tool # 🔥
# ------------------------------------------------------------------
[docs]
async def run(
youtube_url: str,
song_info: str,
style: str = "",
custom_themes: str = "",
rewritten_lyrics: str = "",
save_as: str = "",
ctx: "ToolContext | None" = None,
) -> str:
"""Execute the full auto-cover pipeline."""
if ctx is None:
return json.dumps({"error": "No tool context."})
channel_id = str(ctx.channel_id)
results: dict[str, Any] = {"pipeline": "auto_cover"}
# Step 1: Rip YouTube audio # 💀
audio_path, rip_err = await _rip_youtube_audio(youtube_url)
if rip_err:
return json.dumps({"error": rip_err})
results["ripped"] = True
# Step 2: Upload to Discord for URL # 🎮
audio_url, upload_err = await _upload_to_discord(
audio_path, channel_id, ctx, # type: ignore[arg-type]
)
# Clean up temp file
try:
if audio_path:
os.unlink(audio_path)
parent = os.path.dirname(audio_path)
if parent:
os.rmdir(parent)
except Exception:
pass
if upload_err or not audio_url:
return json.dumps({
"error": upload_err or "No audio URL after upload.",
})
results["audio_url"] = audio_url
# Step 3: Look up lyrics # 🕷️
original_lyrics = None
lyrics_err = None
if not rewritten_lyrics:
original_lyrics, lyrics_err = await _lookup_lyrics(song_info)
# Step 4: Build the response for the LLM # 🌀
# If we have lyrics, we return them so the LLM can rewrite
# If we already have rewritten_lyrics, skip to Suno
final_style = style or DEFAULT_STYLE
results["style"] = final_style
results["song_info"] = song_info
if rewritten_lyrics:
# Lyrics already rewritten -- go straight to Suno
results["phase"] = "ready_for_suno"
results["rewritten_lyrics"] = rewritten_lyrics[:200] + "..."
results["instruction"] = (
f"Call generate_suno_music with:\n"
f"- source_url: {audio_url}\n"
f"- style: {final_style}\n"
f"- lyrics: (the rewritten lyrics provided)\n"
f"- title: S.N.E.S. Cover - {song_info}\n"
f"- instrumental: false"
)
elif original_lyrics:
# We have lyrics -- return them for the LLM to rewrite
results["phase"] = "rewrite_lyrics"
results["original_lyrics"] = original_lyrics
results["rewrite_instructions"] = REWRITE_THEMES
if custom_themes:
results["rewrite_instructions"] += (
f"\n\nAdditional themes from user: {custom_themes}"
)
results["instruction"] = (
f"REWRITE these lyrics with Stargazer themes, then "
f"call generate_suno_music with:\n"
f"- source_url: {audio_url}\n"
f"- style: {final_style}\n"
f"- lyrics: (your rewritten version)\n"
f"- title: S.N.E.S. Cover - {song_info}\n"
f"- instrumental: false\n"
f"- Female vocals"
)
else:
# No lyrics found -- tell LLM to write original
results["phase"] = "write_original"
results["lyrics_error"] = lyrics_err
results["instruction"] = (
f"Could not find lyrics for '{song_info}'. "
f"WRITE original Stargazer-themed lyrics that match "
f"the vibe of the original song, then call "
f"generate_suno_music with:\n"
f"- source_url: {audio_url}\n"
f"- style: {final_style}\n"
f"- lyrics: (your written lyrics)\n"
f"- title: S.N.E.S. Cover - {song_info}\n"
f"- instrumental: false\n"
f"- Female vocals\n\n"
f"Theme guide:\n{REWRITE_THEMES}"
)
if save_as:
results["save_as"] = save_as
return json.dumps(results)