o2o-castad-backend/app/song/schemas/song_schema.py

375 lines
13 KiB
Python

from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Dict, List, Optional
from fastapi import Request
from pydantic import BaseModel, Field
# =============================================================================
# Pydantic Schemas for Song Generation API
# =============================================================================
class GenerateSongRequest(BaseModel):
"""노래 생성 요청 스키마
Usage:
POST /song/generate/{task_id}
Request body for generating a song via Suno API.
Example Request:
{
"lyrics": "인스타 감성의 스테이 머뭄...",
"genre": "k-pop",
"language": "Korean"
}
"""
model_config = {
"json_schema_extra": {
"example": {
"lyrics": "인스타 감성의 스테이 머뭄, 머물러봐요 \n군산 신흥동 말랭이 마을의 마음 힐링 \n사진같은 하루, 여행의 시작 \n보석 같은 이곳은 감성 숙소의 느낌 \n\n인근 명소와 아름다움이 가득한 거리 \n힐링의 바람과 여행의 추억 \n글로벌 감성의 스테이 머뭄, 인스타 감성 \n사진으로 남기고 싶은 그 순간들이 되어줘요",
"genre": "k-pop",
"language": "Korean",
}
}
}
lyrics: str = Field(..., description="노래에 사용할 가사")
genre: str = Field(
...,
description="음악 장르 (K-Pop, Pop, R&B, Hip-Hop, Ballad, EDM, Rock, Jazz 등)",
)
language: str = Field(
default="Korean",
description="노래 언어 (Korean, English, Chinese, Japanese, Thai, Vietnamese)",
)
class GenerateSongResponse(BaseModel):
"""노래 생성 응답 스키마
Usage:
POST /song/generate/{task_id}
Returns the task IDs for tracking song generation.
Note:
실패 조건:
- task_id에 해당하는 Project가 없는 경우 (404 HTTPException)
- task_id에 해당하는 Lyric이 없는 경우 (404 HTTPException)
- Suno API 호출 실패
Example Response (Success):
{
"success": true,
"task_id": "019123ab-cdef-7890-abcd-ef1234567890",
"suno_task_id": "abc123...",
"message": "노래 생성 요청이 접수되었습니다. suno_task_id로 상태를 조회하세요.",
"error_message": null
}
Example Response (Failure):
{
"success": false,
"task_id": "019123ab-cdef-7890-abcd-ef1234567890",
"suno_task_id": null,
"message": "노래 생성 요청에 실패했습니다.",
"error_message": "Suno API connection error"
}
"""
success: bool = Field(..., description="요청 성공 여부")
task_id: Optional[str] = Field(None, description="내부 작업 ID (Project/Lyric task_id)")
suno_task_id: Optional[str] = Field(None, description="Suno API 작업 ID")
message: str = Field(..., description="응답 메시지")
error_message: Optional[str] = Field(None, description="에러 메시지 (실패 시)")
class PollingSongRequest(BaseModel):
"""노래 생성 상태 조회 요청 스키마 (Legacy)
Note:
현재 사용되지 않음. GET /song/status/{suno_task_id} 엔드포인트 사용.
Example Request:
{
"task_id": "abc123..."
}
"""
task_id: str = Field(..., description="Suno 작업 ID")
class SongClipData(BaseModel):
"""생성된 노래 클립 정보"""
id: Optional[str] = Field(None, description="클립 ID")
audio_url: Optional[str] = Field(None, description="오디오 URL")
stream_audio_url: Optional[str] = Field(None, description="스트리밍 오디오 URL")
image_url: Optional[str] = Field(None, description="이미지 URL")
title: Optional[str] = Field(None, description="곡 제목")
status: Optional[str] = Field(None, description="클립 상태")
duration: Optional[float] = Field(None, description="노래 길이 (초)")
class PollingSongResponse(BaseModel):
"""노래 생성 상태 조회 응답 스키마
Usage:
GET /song/status/{suno_task_id}
Suno API 작업 상태를 조회합니다.
Note:
상태 값:
- PENDING: 대기 중
- processing: 생성 중
- SUCCESS / TEXT_SUCCESS / complete: 생성 완료
- failed: 생성 실패
- error: API 조회 오류
SUCCESS 상태 시:
- 백그라운드에서 MP3 파일 다운로드 시작
- Song 테이블의 status를 completed로 업데이트
- song_result_url에 로컬 파일 경로 저장
Example Response (Processing):
{
"success": true,
"status": "processing",
"message": "노래를 생성하고 있습니다.",
"clips": null,
"raw_response": {...},
"error_message": null
}
Example Response (Success):
{
"success": true,
"status": "SUCCESS",
"message": "노래 생성이 완료되었습니다.",
"clips": [
{
"id": "clip-id",
"audio_url": "https://...",
"stream_audio_url": "https://...",
"image_url": "https://...",
"title": "Song Title",
"status": "complete",
"duration": 60.0
}
],
"raw_response": {...},
"error_message": null
}
Example Response (Failure):
{
"success": false,
"status": "error",
"message": "상태 조회에 실패했습니다.",
"clips": null,
"raw_response": null,
"error_message": "ConnectionError: ..."
}
"""
success: bool = Field(..., description="조회 성공 여부")
status: Optional[str] = Field(
None, description="작업 상태 (PENDING, processing, SUCCESS, failed)"
)
message: str = Field(..., description="상태 메시지")
clips: Optional[List[SongClipData]] = Field(None, description="생성된 노래 클립 목록")
raw_response: Optional[Dict[str, Any]] = Field(None, description="Suno API 원본 응답")
error_message: Optional[str] = Field(None, description="에러 메시지 (실패 시)")
class SongListItem(BaseModel):
"""노래 목록 아이템 스키마
Usage:
GET /songs 응답의 개별 노래 정보
Example:
{
"store_name": "스테이 머뭄",
"region": "군산",
"task_id": "019123ab-cdef-7890-abcd-ef1234567890",
"language": "Korean",
"song_result_url": "http://localhost:8000/media/2025-01-15/스테이머뭄.mp3",
"created_at": "2025-01-15T12:00:00"
}
"""
store_name: Optional[str] = Field(None, description="업체명")
region: Optional[str] = Field(None, description="지역명")
task_id: str = Field(..., description="작업 고유 식별자")
language: Optional[str] = Field(None, description="언어")
song_result_url: Optional[str] = Field(None, description="노래 결과 URL")
created_at: Optional[datetime] = Field(None, description="생성 일시")
class DownloadSongResponse(BaseModel):
"""노래 다운로드 응답 스키마
Usage:
GET /song/download/{task_id}
Polls for song completion and returns project info with song URL.
Note:
상태 값:
- processing: 노래 생성 진행 중 (song_result_url은 null)
- completed: 노래 생성 완료 (song_result_url 포함)
- failed: 노래 생성 실패
- not_found: task_id에 해당하는 Song 없음
- error: 조회 중 오류 발생
Example Response (Processing):
{
"success": true,
"status": "processing",
"message": "노래 생성이 진행 중입니다.",
"store_name": null,
"region": null,
"detail_region_info": null,
"task_id": "019123ab-cdef-7890-abcd-ef1234567890",
"language": null,
"song_result_url": null,
"created_at": null,
"error_message": null
}
Example Response (Completed):
{
"success": true,
"status": "completed",
"message": "노래 다운로드가 완료되었습니다.",
"store_name": "스테이 머뭄",
"region": "군산",
"detail_region_info": "군산 신흥동 말랭이 마을",
"task_id": "019123ab-cdef-7890-abcd-ef1234567890",
"language": "Korean",
"song_result_url": "http://localhost:8000/media/2025-01-15/스테이머뭄.mp3",
"created_at": "2025-01-15T12:00:00",
"error_message": null
}
Example Response (Not Found):
{
"success": false,
"status": "not_found",
"message": "task_id 'xxx'에 해당하는 Song을 찾을 수 없습니다.",
"store_name": null,
"region": null,
"detail_region_info": null,
"task_id": null,
"language": null,
"song_result_url": null,
"created_at": null,
"error_message": "Song not found"
}
"""
success: bool = Field(..., description="다운로드 성공 여부")
status: str = Field(..., description="처리 상태 (processing, completed, failed, not_found, error)")
message: str = Field(..., description="응답 메시지")
store_name: Optional[str] = Field(None, description="업체명")
region: Optional[str] = Field(None, description="지역명")
detail_region_info: Optional[str] = Field(None, description="상세 지역 정보")
task_id: Optional[str] = Field(None, description="작업 고유 식별자")
language: Optional[str] = Field(None, description="언어")
song_result_url: Optional[str] = Field(None, description="노래 결과 URL")
created_at: Optional[datetime] = Field(None, description="생성 일시")
error_message: Optional[str] = Field(None, description="에러 메시지 (실패 시)")
# =============================================================================
# Dataclass Schemas (Legacy)
# =============================================================================
@dataclass
class StoreData:
id: int
created_at: datetime
store_name: str
store_category: str | None = None
store_region: str | None = None
store_address: str | None = None
store_phone_number: str | None = None
store_info: str | None = None
@dataclass
class AttributeData:
id: int
attr_category: str
attr_value: str
created_at: datetime
@dataclass
class SongSampleData:
id: int
ai: str
ai_model: str
sample_song: str
season: str | None = None
num_of_people: int | None = None
people_category: str | None = None
genre: str | None = None
@dataclass
class PromptTemplateData:
id: int
prompt: str
description: str | None = None
@dataclass
class SongFormData:
store_name: str
store_id: str
prompts: str
attributes: Dict[str, str] = field(default_factory=dict)
attributes_str: str = ""
lyrics_ids: List[int] = field(default_factory=list)
llm_model: str = "gpt-5-mini"
@classmethod
async def from_form(cls, request: Request):
"""Request의 form 데이터로부터 dataclass 인스턴스 생성"""
form_data = await request.form()
# 고정 필드명들
fixed_keys = {"store_info_name", "store_id", "llm_model", "prompts"}
# lyrics-{id} 형태의 모든 키를 찾아서 ID 추출
lyrics_ids = []
attributes = {}
for key, value in form_data.items():
if key.startswith("lyrics-"):
lyrics_id = key.split("-")[1]
lyrics_ids.append(int(lyrics_id))
elif key not in fixed_keys:
attributes[key] = value
# attributes를 문자열로 변환
attributes_str = (
"\r\n\r\n".join([f"{key} : {value}" for key, value in attributes.items()])
if attributes
else ""
)
return cls(
store_name=form_data.get("store_info_name", ""),
store_id=form_data.get("store_id", ""),
attributes=attributes,
attributes_str=attributes_str,
lyrics_ids=lyrics_ids,
llm_model=form_data.get("llm_model", "gpt-5-mini"),
prompts=form_data.get("prompts", ""),
)