"""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 jsonutil as json
import logging
import os
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]:
"""Download and extract the audio track of a YouTube URL as an mp3.
First stage of the auto-cover pipeline: it shells out to the ``yt-dlp``
binary via ``asyncio.create_subprocess_exec`` (using the authenticated
cookies file at ``YTDLP_COOKIES``) to pull the best audio, transcode it to
mp3, and cap the download at 50MB so a pathological link cannot fill the
disk. The whole subprocess is bounded by a 120-second
``asyncio.wait_for`` timeout. It writes into a fresh ``tempfile.mkdtemp``
directory on the local filesystem and locates the resulting ``.mp3`` by
scanning that directory; the caller (``run``) is responsible for deleting
the file and its temp directory afterward. Missing-binary, timeout, and
generic failures are all caught and returned as error strings rather than
raised.
Called only by ``run`` in this module; no other callers in the repo.
Args:
url: The YouTube video URL to rip audio from.
Returns:
A ``(file_path, error)`` tuple: on success the absolute path to the
ripped mp3 with ``None`` error; on failure ``None`` path with a
human-readable error message.
"""
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]:
"""Fetch the original lyrics for a song from the lyrics.ovh API.
Second stage of the auto-cover pipeline, providing the source text the LLM
later rewrites with Stargazer themes. It splits ``song_info`` on the first
``-`` into artist and title, then makes up to two HTTP GET calls to
``api.lyrics.ovh`` over ``aiohttp`` (each with a 10-second timeout): one in
artist/title order and, if that yields nothing usable, a second with the
order swapped to tolerate ambiguous input. A result is accepted only when
it exceeds 50 characters, guarding against empty or stub responses. Network
errors are silently ignored so the pipeline can fall back to writing
original lyrics. Touches the network only; no local state.
Called only by ``run`` in this module; no other callers in the repo.
Args:
song_info: A ``Title - Artist`` (or ``Artist - Title``) string used to
build the lookup query.
Returns:
A ``(lyrics, error)`` tuple: the stripped lyrics text with ``None``
error on success, or ``None`` lyrics with a not-found message when no
usable lyrics were retrieved.
"""
# 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 the ripped mp3 to Discord and return its public attachment URL.
Third stage of the auto-cover pipeline: Suno needs a fetchable source URL,
so the local mp3 is published as a Discord attachment and its CDN URL is
captured for the downstream ``generate_suno_music`` call. It reads the file
bytes off the filesystem in a worker thread (``asyncio.to_thread`` around
``Path.read_bytes`` so the event loop is not blocked) and sends them
through ``ctx.adapter.send_file`` -- the proxy platform adapter that, in
this microservice split, relays the upload to the gateway over Redis
Streams. Any exception is caught and returned as an error string.
Called only by ``run`` in this module; no other callers in the repo.
Args:
file_path: Local path to the ripped mp3 to upload.
channel_id: The Discord channel the audio is sent to.
ctx: The ToolContext, whose ``adapter`` performs the cross-service
send.
Returns:
A ``(file_url, error)`` tuple: the attachment URL with ``None`` error
on success, or ``None`` URL with an error message when the upload fails
or no URL could be captured.
"""
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:
"""Run the auto-cover pipeline up to the point the LLM takes over.
Entry point for the ``auto_cover`` tool. It orchestrates the three helper
stages -- rip the YouTube audio (``_rip_youtube_audio``), upload it to
Discord for a fetchable URL (``_upload_to_discord``, deleting the temp mp3
and its directory immediately after), and look up the original lyrics
(``_lookup_lyrics``) -- then assembles a phase-tagged instruction payload
rather than calling Suno itself. This is the GAME ACTION injection hand-off
described in the module: the LLM reads the returned ``instruction`` and is
expected to rewrite (or write) lyrics with Stargazer themes and then call
``generate_suno_music`` with the captured ``audio_url`` and style. The
chosen ``phase`` is ``ready_for_suno`` when ``rewritten_lyrics`` is already
supplied, ``rewrite_lyrics`` when originals were found, or
``write_original`` when no lyrics could be located. Touches the filesystem
(temp files), the network (via the helpers), and the Discord channel
through ``ctx.adapter``.
Dispatched by the tool loader as the ``run`` handler for the ``auto_cover``
tool (the single-tool ``TOOL_NAME``/``run`` convention); it is not called
directly anywhere in the repo.
Args:
youtube_url: The YouTube link to rip and cover.
song_info: ``Title - Artist`` string used for the lyrics lookup and in
the generated Suno title.
style: Optional style override; defaults to the S.N.E.S. ``DEFAULT_STYLE``
preset when empty.
custom_themes: Optional extra theme notes appended to the rewrite
instructions.
rewritten_lyrics: Pre-rewritten lyrics; when present the rewrite step is
skipped and the phase jumps straight to Suno.
save_as: Optional name under which to save the resulting cover as a game
asset; echoed back in the payload.
ctx: The ToolContext supplying the channel ID and platform adapter.
Returns:
A JSON string describing the pipeline result -- the ``phase``, the
``audio_url``, the resolved ``style``, and an ``instruction`` for the
next LLM step -- or a JSON ``error`` object if the context is missing
or a stage failed.
"""
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)