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

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
        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
        Returns the task ID for tracking song generation.

    Example Response:
        {
            "success": true,
            "task_id": "abc123...",
            "message": "노래 생성 요청이 접수되었습니다."
        }
    """

    success: bool = Field(..., description="요청 성공 여부")
    task_id: Optional[str] = Field(None, description="Suno 작업 ID")
    message: str = Field(..., description="응답 메시지")
    error_message: Optional[str] = Field(None, description="에러 메시지 (실패 시)")


class PollingSongRequest(BaseModel):
    """노래 생성 상태 조회 요청 스키마

    Usage:
        POST /song/polling
        Request body for checking song generation status.

    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:
        POST /song/polling 또는 GET /song/status/{task_id}
        Returns the current status of song generation.

    Example Response:
        {
            "success": true,
            "status": "complete",
            "message": "노래 생성이 완료되었습니다.",
            "clips": [...]
        }
    """

    success: bool = Field(..., description="조회 성공 여부")
    status: Optional[str] = Field(
        None, description="작업 상태 (pending, processing, complete, 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="에러 메시지 (실패 시)")


# =============================================================================
# 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-4o"

    @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-4o"),
            prompts=form_data.get("prompts", ""),
        )