from dataclasses import dataclass, field from datetime import datetime from typing import 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", "song_id": "abc123...", "message": "노래 생성 요청이 접수되었습니다. song_id로 상태를 조회하세요.", "error_message": null } Example Response (Failure): { "success": false, "task_id": "019123ab-cdef-7890-abcd-ef1234567890", "song_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)") song_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/{song_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): """노래 생성 상태 조회 응답 스키마 (Suno API) Usage: GET /song/status/{song_id} Suno API 작업 상태를 조회합니다. Note: 상태 값 (Suno API 응답): - PENDING: Suno API 대기 중 - processing: Suno API에서 노래 생성 중 - SUCCESS: Suno API 노래 생성 완료 (백그라운드 Blob 업로드 시작) - TEXT_SUCCESS: Suno API 노래 생성 완료 - failed: Suno API 노래 생성 실패 - error: API 조회 오류 SUCCESS 상태 시: - 백그라운드에서 MP3 파일 다운로드 및 Azure Blob 업로드 시작 - Song 테이블의 status가 uploading으로 변경 - 업로드 완료 시 status가 completed로 변경, song_result_url에 Blob URL 저장 Example Response (Pending): { "success": true, "status": "PENDING", "message": "노래 생성 대기 중입니다.", "error_message": null } Example Response (Processing): { "success": true, "status": "processing", "message": "노래를 생성하고 있습니다.", "error_message": null } Example Response (Success): { "success": true, "status": "SUCCESS", "message": "노래 생성이 완료되었습니다.", "error_message": null } Example Response (Failure): { "success": false, "status": "error", "message": "상태 조회에 실패했습니다.", "error_message": "ConnectionError: ..." } """ success: bool = Field(..., description="조회 성공 여부") status: Optional[str] = Field( None, description="Suno API 작업 상태 (PENDING, processing, SUCCESS, TEXT_SUCCESS, failed, error)" ) message: str = Field(..., description="상태 메시지") 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): """노래 다운로드 응답 스키마 (DB Polling) Usage: GET /song/download/{task_id} DB의 Song 테이블 상태를 조회하고 완료 시 Project 정보와 노래 URL을 반환합니다. Note: 상태 값 (DB 상태): - processing: Suno API에서 노래 생성 중 (song_result_url은 null) - uploading: MP3 다운로드 및 Azure Blob 업로드 중 (song_result_url은 null) - completed: 모든 작업 완료 (song_result_url에 Azure Blob 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 (Uploading): { "success": true, "status": "uploading", "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": "https://blob.azure.com/.../song.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="DB 처리 상태 (processing, uploading, 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", ""), )