castad
/
o2o-castad-backend
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

노래 생성까지 완료

insta
bluebamus 2025-12-22 18:45:41 +09:00
parent dd855ff11a
commit caf7b1ab60
4 changed files with 287 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
.gitignore vendored
View File

@ -21,4 +21,9 @@ dist/
build/ build/
*.pyc *.pyc
*.pyo *.pyo
.eggs/ .eggs/
# Media files
*.mp3
*.mp4
media/

112
README.md
View File

@ -0,0 +1,112 @@
# CastAD Backend
AI 기반 광고 음악 생성 서비스의 백엔드 API 서버입니다.
## 기술 스택
- **Framework**: FastAPI
- **Database**: MySQL (asyncmy 비동기 드라이버)
- **ORM**: SQLAlchemy (async)
- **AI Services**:
- OpenAI ChatGPT (가사 생성, 마케팅 분석)
- Suno AI (음악 생성)
## 프로젝트 구조
```text
app/
├── core/ # 핵심 설정 및 공통 모듈
├── database/ # 데이터베이스 세션 및 설정
├── home/ # 홈 API (크롤링, 영상 생성 요청)
├── lyric/ # 가사 API (가사 생성)
├── song/ # 노래 API (Suno AI 음악 생성)
├── video/ # 비디오 관련 모듈
└── utils/ # 유틸리티 (ChatGPT, Suno, 크롤러)
```
## API 엔드포인트
### Home API
| Method | Endpoint | 설명 |
| ------ | ------------------ | ------------------------------ |
| POST | `/crawling` | 네이버 지도 장소 크롤링 |
| POST | `/generate` | 기본 영상 생성 요청 |
| POST | `/generate/urls` | URL 기반 영상 생성 요청 |
| POST | `/generate/upload` | 파일 업로드 기반 영상 생성 요청 |
### Lyric API
| Method | Endpoint | 설명 |
| ------ | ------------------------- | --------------------------- |
| POST | `/lyric/generate` | ChatGPT를 이용한 가사 생성 |
| GET | `/lyric/status/{task_id}` | 가사 생성 상태 조회 |
| GET | `/lyric/{task_id}` | 가사 상세 조회 |
| GET | `/lyrics` | 가사 목록 조회 (페이지네이션) |
### Song API
| Method | Endpoint | 설명 |
| ------ | -------------------------- | ----------------------------- |
| POST | `/song/generate` | Suno AI를 이용한 노래 생성 요청 |
| GET | `/song/status/{task_id}` | 노래 생성 상태 조회 (폴링) |
| GET | `/song/download/{task_id}` | 생성된 노래 MP3 다운로드 |
## 환경 설정
`.env` 파일에 다음 환경 변수를 설정합니다:
```env
# 프로젝트 설정
PROJECT_NAME=CastAD
PROJECT_DOMAIN=localhost:8000
DEBUG=True
# MySQL 설정
MYSQL_HOST=localhost
MYSQL_PORT=3306
MYSQL_USER=your_user
MYSQL_PASSWORD=your_password
MYSQL_DB=castad
# API Keys
CHATGPT_API_KEY=your_openai_api_key
SUNO_API_KEY=your_suno_api_key
SUNO_CALLBACK_URL=https://your-domain.com/api/suno/callback
```
## 실행 방법
### uv 설치
```bash
# Windows (PowerShell)
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"
# macOS / Linux
curl -LsSf https://astral.sh/uv/install.sh | sh
```
### 의존성 설치
```bash
# 기본 설치 (uv가 자동으로 가상환경 생성)
uv sync
# 이미 venv를 만든 경우 (기존 가상환경 활성화 필요)
uv sync --active
```
### 서버 실행
```bash
# 개발 서버 실행
fastapi dev main.py
# 프로덕션 서버 실행
fastapi run main.py
```
## API 문서
서버 실행 후 `/docs` 에서 Scalar API 문서를 확인할 수 있습니다.

152
app/song/api/routers/v1/song.py
View File

@ -6,21 +6,30 @@ Song API Router
엔드포인트 목록: 엔드포인트 목록:
- POST /song/generate: 노래 생성 요청 - POST /song/generate: 노래 생성 요청
- GET /song/status/{task_id}: 노래 생성 상태 조회 - GET /song/status/{task_id}: 노래 생성 상태 조회
- GET /song/download/{task_id}: 노래 다운로드
사용 예시: 사용 예시:
from app.song.api.routers.v1.song import router from app.song.api.routers.v1.song import router
app.include_router(router, prefix="/api/v1") app.include_router(router, prefix="/api/v1")
""" """
from datetime import date
from pathlib import Path
import aiofiles
import httpx
from fastapi import APIRouter from fastapi import APIRouter
from uuid_extensions import uuid7str
from app.song.schemas.song_schema import ( from app.song.schemas.song_schema import (
DownloadSongResponse,
GenerateSongRequest, GenerateSongRequest,
GenerateSongResponse, GenerateSongResponse,
PollingSongResponse, PollingSongResponse,
SongClipData, SongClipData,
) )
from app.utils.suno import SunoService from app.utils.suno import SunoService
from config import prj_settings
def _parse_suno_status_response(result: dict) -> PollingSongResponse: def _parse_suno_status_response(result: dict) -> PollingSongResponse:
@ -39,25 +48,30 @@ def _parse_suno_status_response(result: dict) -> PollingSongResponse:
) )
status = data.get("status", "unknown") status = data.get("status", "unknown")
clips_data = data.get("data", [])
# 상태별 메시지 # 클립 데이터는 data.response.sunoData에 있음 (camelCase)
response_data = data.get("response", {})
clips_data = response_data.get("sunoData", [])
# 상태별 메시지 (Suno API는 다양한 상태값 반환)
status_messages = { status_messages = {
"pending": "노래 생성 대기 중입니다.", "pending": "노래 생성 대기 중입니다.",
"processing": "노래를 생성하고 있습니다.", "processing": "노래를 생성하고 있습니다.",
"complete": "노래 생성이 완료되었습니다.", "complete": "노래 생성이 완료되었습니다.",
"SUCCESS": "노래 생성이 완료되었습니다.",
"TEXT_SUCCESS": "노래 생성이 완료되었습니다.",
"failed": "노래 생성에 실패했습니다.", "failed": "노래 생성에 실패했습니다.",
} }
# 클립 데이터 파싱 # 클립 데이터 파싱 (Suno API는 camelCase 사용)
clips = None clips = None
if clips_data: if clips_data:
clips = [ clips = [
SongClipData( SongClipData(
id=clip.get("id"), id=clip.get("id"),
audio_url=clip.get("audio_url"), audio_url=clip.get("audioUrl"),
stream_audio_url=clip.get("stream_audio_url"), stream_audio_url=clip.get("streamAudioUrl"),
image_url=clip.get("image_url"), image_url=clip.get("imageUrl"),
title=clip.get("title"), title=clip.get("title"),
status=clip.get("status"), status=clip.get("status"),
duration=clip.get("duration"), duration=clip.get("duration"),
@ -194,3 +208,129 @@ async def get_song_status(
raw_response=None, raw_response=None,
error_message=str(e), error_message=str(e),
) )
@router.get(
"/download/{task_id}",
summary="노래 다운로드",
description="""
완료된 노래를 서버에 다운로드하고 접근 가능한 URL을 반환합니다.
## 경로 파라미터
- **task_id**: 노래 생성 시 반환된 작업 ID (필수)
## 반환 정보
- **success**: 다운로드 성공 여부
- **message**: 응답 메시지
- **file_path**: 저장된 파일의 상대 경로
- **file_url**: 프론트엔드에서 접근 가능한 파일 URL
## 사용 예시
```
GET /song/download/abc123...
```
## 참고
- 노래 생성이 완료된 상태(complete)에서만 다운로드 가능합니다.
- 파일은 /media/{날짜}/{uuid7}/song.mp3 경로에 저장됩니다.
- 반환된 file_url을 사용하여 프론트엔드에서 MP3를 재생할 수 있습니다.
""",
response_model=DownloadSongResponse,
responses={
200: {"description": "다운로드 성공"},
400: {"description": "노래 생성이 완료되지 않음"},
500: {"description": "다운로드 실패"},
},
)
async def download_song(
task_id: str,
) -> DownloadSongResponse:
"""완료된 노래를 다운로드하여 서버에 저장하고 접근 URL을 반환합니다."""
try:
suno_service = SunoService()
result = await suno_service.get_task_status(task_id)
# API 응답 확인
if result.get("code") != 200:
return DownloadSongResponse(
success=False,
message="Suno API 응답 오류",
error_message=result.get("msg", "Unknown error"),
)
data = result.get("data", {})
status = data.get("status", "unknown")
# 완료 상태 확인 (Suno API는 다양한 완료 상태값 반환)
completed_statuses = {"complete", "SUCCESS", "TEXT_SUCCESS"}
if status not in completed_statuses:
return DownloadSongResponse(
success=False,
message=f"노래 생성이 완료되지 않았습니다. 현재 상태: {status}",
error_message="노래 생성 완료 후 다운로드해 주세요.",
)
# 클립 데이터는 data.response.sunoData에 있음 (camelCase)
response_data = data.get("response", {})
clips_data = response_data.get("sunoData", [])
if not clips_data:
return DownloadSongResponse(
success=False,
message="생성된 노래 클립이 없습니다.",
error_message="sunoData is empty",
)
# 첫 번째 클립의 streamAudioUrl 가져오기 (camelCase)
first_clip = clips_data[0]
stream_audio_url = first_clip.get("streamAudioUrl")
if not stream_audio_url:
return DownloadSongResponse(
success=False,
message="스트리밍 오디오 URL을 찾을 수 없습니다.",
error_message="stream_audio_url is missing",
)
# 저장 경로 생성: media/{날짜}/{uuid7}/song.mp3
today = date.today().isoformat()
unique_id = uuid7str()
relative_dir = f"{today}/{unique_id}"
file_name = "song.mp3"
# 절대 경로 생성
media_dir = Path("media") / today / unique_id
media_dir.mkdir(parents=True, exist_ok=True)
file_path = media_dir / file_name
# 오디오 파일 다운로드 (비동기 파일 쓰기)
async with httpx.AsyncClient() as client:
response = await client.get(stream_audio_url, timeout=60.0)
response.raise_for_status()
async with aiofiles.open(file_path, "wb") as f:
await f.write(response.content)
# 프론트엔드에서 접근 가능한 URL 생성
relative_path = f"/media/{relative_dir}/{file_name}"
base_url = f"http://{prj_settings.PROJECT_DOMAIN}"
file_url = f"{base_url}{relative_path}"
return DownloadSongResponse(
success=True,
message="노래 다운로드가 완료되었습니다.",
file_path=relative_path,
file_url=file_url,
)
except httpx.HTTPError as e:
return DownloadSongResponse(
success=False,
message="오디오 파일 다운로드에 실패했습니다.",
error_message=str(e),
)
except Exception as e:
return DownloadSongResponse(
success=False,
message="노래 다운로드에 실패했습니다.",
error_message=str(e),
)

23
app/song/schemas/song_schema.py
View File

@ -122,6 +122,29 @@ class PollingSongResponse(BaseModel):
error_message: Optional[str] = Field(None, description="에러 메시지 (실패 시)") error_message: Optional[str] = Field(None, description="에러 메시지 (실패 시)")
class DownloadSongResponse(BaseModel):
"""노래 다운로드 응답 스키마
Usage:
GET /song/download/{task_id}
Downloads the generated song and returns the local file path.
Example Response:
{
"success": true,
"message": "노래 다운로드가 완료되었습니다.",
"file_path": "/media/2025-01-15/01234567-89ab-7def-0123-456789abcdef/song.mp3",
"file_url": "http://localhost:8000/media/2025-01-15/01234567-89ab-7def-0123-456789abcdef/song.mp3"
}
"""
success: bool = Field(..., description="다운로드 성공 여부")
message: str = Field(..., description="응답 메시지")
file_path: Optional[str] = Field(None, description="저장된 파일 경로 (상대 경로)")
file_url: Optional[str] = Field(None, description="파일 접근 URL")
error_message: Optional[str] = Field(None, description="에러 메시지 (실패 시)")
# ============================================================================= # =============================================================================
# Dataclass Schemas (Legacy) # Dataclass Schemas (Legacy)
# ============================================================================= # =============================================================================