|
|
""" |
|
|
Music Service |
|
|
|
|
|
High-level service for music generation operations. |
|
|
Abstracts all API complexity from the UI layer. |
|
|
""" |
|
|
|
|
|
from typing import Callable, Optional, List |
|
|
from dataclasses import dataclass, field |
|
|
|
|
|
from ..api.client import StackNetClient, MediaAction |
|
|
|
|
|
|
|
|
@dataclass |
|
|
class MusicClip: |
|
|
"""Generated music clip.""" |
|
|
title: str |
|
|
audio_url: str |
|
|
audio_path: Optional[str] = None |
|
|
duration: Optional[str] = None |
|
|
image_url: Optional[str] = None |
|
|
video_url: Optional[str] = None |
|
|
tags: List[str] = field(default_factory=list) |
|
|
|
|
|
|
|
|
@dataclass |
|
|
class StemResult: |
|
|
"""Extracted audio stems.""" |
|
|
vocals_path: Optional[str] = None |
|
|
drums_path: Optional[str] = None |
|
|
bass_path: Optional[str] = None |
|
|
other_path: Optional[str] = None |
|
|
|
|
|
|
|
|
class MusicService: |
|
|
""" |
|
|
Service for music generation and manipulation. |
|
|
|
|
|
Provides clean interfaces for: |
|
|
- Text-to-music generation |
|
|
- Cover song creation |
|
|
- Stem extraction |
|
|
""" |
|
|
|
|
|
def __init__(self, client: Optional[StackNetClient] = None): |
|
|
self.client = client or StackNetClient() |
|
|
|
|
|
async def generate_music( |
|
|
self, |
|
|
prompt: str, |
|
|
title: Optional[str] = None, |
|
|
tags: Optional[str] = None, |
|
|
lyrics: Optional[str] = None, |
|
|
instrumental: bool = False, |
|
|
on_progress: Optional[Callable[[float, str], None]] = None |
|
|
) -> List[MusicClip]: |
|
|
""" |
|
|
Generate original music from a text prompt. |
|
|
|
|
|
Args: |
|
|
prompt: Description of desired music |
|
|
title: Optional song title |
|
|
tags: Optional genre/style tags (comma-separated) |
|
|
lyrics: Optional lyrics (ignored if instrumental=True) |
|
|
instrumental: Generate instrumental only |
|
|
on_progress: Callback for progress updates |
|
|
|
|
|
Returns: |
|
|
List of generated MusicClip objects |
|
|
""" |
|
|
options = {} |
|
|
if tags: |
|
|
options["tags"] = tags |
|
|
if title: |
|
|
options["title"] = title |
|
|
if instrumental: |
|
|
options["make_instrumental"] = True |
|
|
if lyrics and not instrumental: |
|
|
options["lyrics"] = lyrics |
|
|
|
|
|
result = await self.client.submit_media_task( |
|
|
action=MediaAction.GENERATE_MUSIC, |
|
|
prompt=prompt, |
|
|
options=options if options else None, |
|
|
on_progress=on_progress |
|
|
) |
|
|
|
|
|
if not result.success: |
|
|
raise Exception(result.error or "Music generation failed") |
|
|
|
|
|
return self._parse_music_result(result.data) |
|
|
|
|
|
async def create_cover( |
|
|
self, |
|
|
audio_url: str, |
|
|
style_prompt: str, |
|
|
title: Optional[str] = None, |
|
|
tags: Optional[str] = None, |
|
|
on_progress: Optional[Callable[[float, str], None]] = None |
|
|
) -> List[MusicClip]: |
|
|
""" |
|
|
Create a cover version of audio. |
|
|
|
|
|
Args: |
|
|
audio_url: URL to source audio |
|
|
style_prompt: Style/voice direction for the cover |
|
|
title: Optional title for the cover |
|
|
tags: Optional genre/style tags |
|
|
on_progress: Progress callback |
|
|
|
|
|
Returns: |
|
|
List of generated cover clips |
|
|
""" |
|
|
options = {} |
|
|
if tags: |
|
|
options["tags"] = tags |
|
|
if title: |
|
|
options["title"] = title |
|
|
|
|
|
result = await self.client.submit_media_task( |
|
|
action=MediaAction.CREATE_COVER, |
|
|
audio_url=audio_url, |
|
|
prompt=style_prompt, |
|
|
options=options if options else None, |
|
|
on_progress=on_progress |
|
|
) |
|
|
|
|
|
if not result.success: |
|
|
raise Exception(result.error or "Cover creation failed") |
|
|
|
|
|
return self._parse_music_result(result.data) |
|
|
|
|
|
async def extract_stems( |
|
|
self, |
|
|
audio_url: str, |
|
|
on_progress: Optional[Callable[[float, str], None]] = None |
|
|
) -> StemResult: |
|
|
""" |
|
|
Extract stems (vocals, drums, bass, other) from audio. |
|
|
|
|
|
Args: |
|
|
audio_url: URL to source audio |
|
|
on_progress: Progress callback |
|
|
|
|
|
Returns: |
|
|
StemResult with paths to each stem |
|
|
""" |
|
|
result = await self.client.submit_media_task( |
|
|
action=MediaAction.EXTRACT_STEMS, |
|
|
audio_url=audio_url, |
|
|
on_progress=on_progress |
|
|
) |
|
|
|
|
|
if not result.success: |
|
|
raise Exception(result.error or "Stem extraction failed") |
|
|
|
|
|
stems_data = result.data.get("stems", result.data) |
|
|
|
|
|
stem_result = StemResult() |
|
|
|
|
|
|
|
|
if stems_data.get("vocals"): |
|
|
stem_result.vocals_path = await self.client.download_file( |
|
|
stems_data["vocals"], "vocals.mp3" |
|
|
) |
|
|
if stems_data.get("drums"): |
|
|
stem_result.drums_path = await self.client.download_file( |
|
|
stems_data["drums"], "drums.mp3" |
|
|
) |
|
|
if stems_data.get("bass"): |
|
|
stem_result.bass_path = await self.client.download_file( |
|
|
stems_data["bass"], "bass.mp3" |
|
|
) |
|
|
if stems_data.get("other"): |
|
|
stem_result.other_path = await self.client.download_file( |
|
|
stems_data["other"], "other.mp3" |
|
|
) |
|
|
|
|
|
return stem_result |
|
|
|
|
|
def _parse_music_result(self, data: dict) -> List[MusicClip]: |
|
|
"""Parse API response into MusicClip objects.""" |
|
|
clips = [] |
|
|
|
|
|
|
|
|
raw_clips = data.get("clips", []) |
|
|
|
|
|
|
|
|
if not raw_clips: |
|
|
if data.get("audio_url") or data.get("audioUrl"): |
|
|
raw_clips = [data] |
|
|
elif data.get("url"): |
|
|
raw_clips = [{"audio_url": data["url"], "title": data.get("title", "Generated")}] |
|
|
|
|
|
for clip_data in raw_clips: |
|
|
audio_url = clip_data.get("audio_url") or clip_data.get("audioUrl") or clip_data.get("url") |
|
|
if audio_url: |
|
|
clips.append(MusicClip( |
|
|
title=clip_data.get("title", "Generated Music"), |
|
|
audio_url=audio_url, |
|
|
duration=clip_data.get("duration"), |
|
|
image_url=clip_data.get("image_url") or clip_data.get("imageUrl"), |
|
|
video_url=clip_data.get("video_url") or clip_data.get("videoUrl"), |
|
|
tags=clip_data.get("tags", []) |
|
|
)) |
|
|
|
|
|
return clips |
|
|
|
|
|
async def download_clip(self, clip: MusicClip) -> str: |
|
|
"""Download a clip's audio to local file.""" |
|
|
if clip.audio_path: |
|
|
return clip.audio_path |
|
|
|
|
|
filename = f"{clip.title.replace(' ', '_')[:30]}.mp3" |
|
|
clip.audio_path = await self.client.download_file(clip.audio_url, filename) |
|
|
return clip.audio_path |
|
|
|
|
|
def cleanup(self): |
|
|
"""Clean up temporary files.""" |
|
|
self.client.cleanup() |
|
|
|
