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

Merge branch 'main' into prompt

feature-youtube-upload
jaehwang 2026-01-23 13:11:55 +09:00
parent 38323870ec 7038faaf74
commit 88a91aa6d7
50 changed files with 4166 additions and 2001 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

198
README.md
View File

@ -4,24 +4,29 @@ AI 기반 광고 음악 생성 서비스의 백엔드 API 서버입니다.
## 기술 스택
- **Language**: Python 3.13
- **Framework**: FastAPI
- **Database**: MySQL (asyncmy 비동기 드라이버)
- **Database**: MySQL (asyncmy 비동기 드라이버), Redis
- **ORM**: SQLAlchemy (async)
- **Package Manager**: uv
- **AI Services**:
- OpenAI ChatGPT (가사 생성, 마케팅 분석)
- Suno AI (음악 생성)
- Creatomate (비디오 생성)
## 프로젝트 구조
```text
app/
├── core/ # 핵심 설정 및 공통 모듈
├── database/ # 데이터베이스 세션 및 설정
├── core/ # 핵심 설정 및 공통 모듈 (logging, exceptions)
├── database/ # 데이터베이스 세션 및 Redis 설정
├── dependencies/ # FastAPI 의존성 주입
├── home/ # 홈 API (크롤링, 영상 생성 요청)
├── lyric/ # 가사 API (가사 생성)
├── song/ # 노래 API (Suno AI 음악 생성)
├── user/ # 사용자 모듈 (카카오 로그인, JWT 인증)
├── video/ # 비디오 관련 모듈
└── utils/ # 유틸리티 (ChatGPT, Suno, 크롤러)
└── utils/ # 유틸리티 (ChatGPT, Suno, 크롤러, 프롬프트)
```
## API 엔드포인트
@ -57,22 +62,84 @@ app/
`.env` 파일에 다음 환경 변수를 설정합니다:
```env
# 프로젝트 설정
PROJECT_NAME=CastAD
PROJECT_DOMAIN=localhost:8000
DEBUG=True
# ================================
# 프로젝트 기본 정보
# ================================
PROJECT_NAME=CastAD # 프로젝트 이름
PROJECT_DOMAIN=localhost:8000 # 프로젝트 도메인 (호스트:포트)
PROJECT_VERSION=0.1.0 # 프로젝트 버전
DESCRIPTION=FastAPI 기반 CastAD 프로젝트 # 프로젝트 설명
ADMIN_BASE_URL=/admin # 관리자 페이지 기본 URL
DEBUG=True # 디버그 모드 (True: 개발, False: 운영)
# ================================
# MySQL 설정
MYSQL_HOST=localhost
MYSQL_PORT=3306
MYSQL_USER=your_user
MYSQL_PASSWORD=your_password
MYSQL_DB=castad
# ================================
MYSQL_HOST=localhost # MySQL 호스트 주소
MYSQL_PORT=3306 # MySQL 포트 번호
MYSQL_USER=castad-admin # MySQL 사용자명
MYSQL_PASSWORD=o2o1324 # MySQL 비밀번호
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
# ================================
# Redis 설정
# ================================
REDIS_HOST=localhost # Redis 호스트 주소
REDIS_PORT=6379 # Redis 포트 번호
# ================================
# CORS 설정
# ================================
CORS_ALLOW_ORIGINS='["*"]' # 허용할 Origin 목록 (JSON 배열 형식)
CORS_ALLOW_CREDENTIALS=True # 자격 증명(쿠키 등) 허용 여부
CORS_ALLOW_METHODS='["*"]' # 허용할 HTTP 메서드 (JSON 배열 형식)
CORS_ALLOW_HEADERS='["*"]' # 허용할 HTTP 헤더 (JSON 배열 형식)
CORS_MAX_AGE=600 # Preflight 요청 캐시 시간 (초)
# ================================
# Azure Blob Storage 설정
# ================================
AZURE_BLOB_SAS_TOKEN=your_sas_token # Azure Blob Storage SAS 토큰
AZURE_BLOB_BASE_URL=https://... # Azure Blob Storage 기본 URL
# ================================
# Creatomate 템플릿 설정
# ================================
TEMPLATE_ID_VERTICAL=your_template_id # 세로형(9:16) 비디오 템플릿 ID
TEMPLATE_DURATION_VERTICAL=60.0 # 세로형 비디오 기본 길이 (초)
TEMPLATE_ID_HORIZONTAL=your_template_id # 가로형(16:9) 비디오 템플릿 ID
TEMPLATE_DURATION_HORIZONTAL=20.0 # 가로형 비디오 기본 길이 (초)
# ================================
# JWT 토큰 설정
# ================================
JWT_SECRET=your_secret_key # JWT 서명용 비밀키 (랜덤 문자열 권장)
JWT_ALGORITHM=HS256 # JWT 알고리즘 (기본: HS256)
JWT_ACCESS_TOKEN_EXPIRE_MINUTES=60 # Access Token 만료 시간 (분)
JWT_REFRESH_TOKEN_EXPIRE_DAYS=7 # Refresh Token 만료 시간 (일)
# ================================
# 프롬프트 설정
# ================================
PROMPT_FOLDER_ROOT=./app/utils/prompts # 프롬프트 파일 루트 디렉토리
MARKETING_PROMPT_NAME=marketing_prompt # 마케팅 분석용 프롬프트 파일명
SUMMARIZE_PROMPT_NAME=summarize_prompt # 요약용 프롬프트 파일명
LYLIC_PROMPT_NAME=lyric_prompt # 가사 생성용 프롬프트 파일명
# ================================
# 로그 설정
# ================================
LOG_CONSOLE_ENABLED=True # 콘솔 로그 출력 여부
LOG_FILE_ENABLED=True # 파일 로그 저장 여부
LOG_LEVEL=DEBUG # 전체 로그 레벨 (DEBUG, INFO, WARNING, ERROR, CRITICAL)
LOG_CONSOLE_LEVEL=DEBUG # 콘솔 출력 로그 레벨
LOG_FILE_LEVEL=DEBUG # 파일 저장 로그 레벨
LOG_MAX_SIZE_MB=15 # 로그 파일 최대 크기 (MB)
LOG_BACKUP_COUNT=30 # 로그 백업 파일 보관 개수
LOG_DIR=logs # 로그 저장 디렉토리 경로
# - 절대 경로: 해당 경로 사용
# - 상대 경로: 프로젝트 루트 기준
# - /www/log/uvicorn 존재 시: 자동으로 해당 경로 사용 (운영)
```
## 실행 방법
@ -110,3 +177,100 @@ fastapi run main.py
## API 문서
서버 실행 후 `/docs` 에서 Scalar API 문서를 확인할 수 있습니다.
## 서버 아키텍처
### 전체 시스템 흐름
```
┌─────────────────────────────────────────────────────────────────────────────┐
│ Client (Web/Mobile) │
└─────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ FastAPI Application │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────┐ │
│ │ Auth API │ │ Home API │ │ Lyric API │ │ Song/Video API │ │
│ │ (카카오) │ │ (크롤링) │ │ (가사생성) │ │ (음악/영상 생성) │ │
│ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ └──────────┬──────────┘ │
└─────────┼────────────────┼────────────────┼─────────────────────┼───────────┘
│ │ │ │
▼ ▼ ▼ ▼
┌─────────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────────┐
│ Kakao OAuth │ │ Naver Maps │ │ ChatGPT │ │ External AI Services │
│ (로그인) │ │ (크롤링) │ │ (OpenAI) │ │ ┌───────┐ ┌──────────┐ │
└─────────────────┘ └─────────────┘ └─────────────┘ │ │ Suno │ │Creatomate│ │
│ │ (음악) │ │ (영상) │ │
│ └───────┘ └──────────┘ │
└─────────────────────────┘
│ │ │ │
▼ ▼ ▼ ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ Data Layer │
│ ┌─────────────┐ ┌─────────────────────┐ │
│ │ MySQL │ │ Azure Blob Storage │ │
│ │ (메인 DB) │ │ (미디어 저장소) │ │
│ └─────────────┘ └─────────────────────┘ │
│ ┌─────────────┐ │
│ │ Redis │ │
│ │ (캐시/세션) │ │
│ └─────────────┘ │
└─────────────────────────────────────────────────────────────────────────────┘
```
### 광고 콘텐츠 생성 플로우
```
┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐
│ 1. 입력 │───▶│ 2. 크롤링 │───▶│ 3. 가사 │───▶│ 4. 음악 │───▶│ 5. 영상 │
│ │ │ │ │ 생성 │ │ 생성 │ │ 생성 │
└──────────┘ └──────────┘ └──────────┘ └──────────┘ └──────────┘
│ │ │ │ │
▼ ▼ ▼ ▼ ▼
┌────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐
│장소 URL │ │Naver Maps│ │ ChatGPT │ │ Suno AI │ │Creatomate│
│or 이미지 │ │ 크롤러 │ │ API │ │ API │ │ API │
└────────┘ └──────────┘ └──────────┘ └──────────┘ └──────────┘
│ │ │ │
▼ ▼ ▼ ▼
┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐
│장소 정보 │ │ 광고 가사 │ │ MP3 │ │ 광고 영상 │
│이미지 수집 │ │ 텍스트 │ │ 파일 │ │ 파일 │
└──────────┘ └──────────┘ └──────────┘ └──────────┘
```
### 인증 플로우 (카카오 OAuth)
```
┌────────┐ ┌────────────┐ ┌───────────┐ ┌────────────┐
│ Client │ │ CastAD │ │ Kakao │ │ MySQL │
│ │ │ Backend │ │ OAuth │ │ │
└───┬────┘ └─────┬──────┘ └─────┬─────┘ └─────┬──────┘
│ │ │ │
│ 1. 로그인 요청 │ │ │
│───────────────▶│ │ │
│ │ │ │
│ 2. 카카오 URL │ │ │
│◀───────────────│ │ │
│ │ │ │
│ 3. 카카오 로그인 │ │ │
│────────────────────────────────▶ │ │
│ │ │ │
│ 4. 인가 코드 │ │ │
│◀──────────────────────────────── │ │
│ │ │ │
│ 5. 콜백 (code) │ │ │
│───────────────▶│ 6. 토큰 요청 │ │
│ │─────────────────▶│ │
│ │ 7. Access Token │ │
│ │◀─────────────────│ │
│ │ │ │
│ │ 8. 사용자 저장/조회 │ │
│ │─────────────────────────────────▶ │
│ │◀───────────────────────────────── │
│ │ │ │
│ 9. JWT 토큰 발급 │ │ │
│◀───────────────│ │ │
│ │ │ │
```

0
app/auth/__init__.py → app/archive/__init__.py
View File

0
app/auth/api/__init__.py → app/archive/api/__init__.py
View File

0
app/auth/api/routers/__init__.py → app/archive/api/archive_admin.py
View File

0
app/auth/api/routers/v1/__init__.py → app/archive/api/routers/__init__.py
View File

0
app/auth/services/__init__.py → app/archive/api/routers/v1/__init__.py
View File

0
app/archive/api/routers/v1/archive.py Normal file
View File

0
app/archive/dependency.py Normal file
View File

0
app/archive/models.py Normal file
View File

0
app/archive/schemas/__init__.py Normal file
View File

0
app/archive/schemas/archive_schema.py Normal file
View File

0
app/archive/services/__init__.py Normal file
View File

0
app/archive/services/archive.py Normal file
View File

0
app/archive/tests/__init__.py Normal file
View File

0
app/archive/tests/archive/__init__.py Normal file
View File

0
app/archive/tests/archive/conftest.py Normal file
View File

0
app/archive/tests/archive/test_db.py Normal file
View File

0
app/archive/tests/conftest.py Normal file
View File

0
app/archive/tests/test_db.py Normal file
View File

0
app/archive/worker/__init__.py Normal file
View File

0
app/archive/worker/archive_task.py Normal file
View File

125
app/auth/api/routers/v1/auth.py
View File

@ -1,125 +0,0 @@
"""
카카오 로그인 API 라우터
"""
from datetime import datetime, timezone
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.responses import RedirectResponse
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.auth.dependencies import get_current_user_optional
from app.auth.models import User
from app.auth.schemas import AuthStatusResponse, TokenResponse, UserResponse
from app.auth.services.jwt import create_access_token
from app.auth.services.kakao import kakao_client
from app.database.session import get_session
router = APIRouter(tags=["Auth"])
@router.get("/auth/kakao/login")
async def kakao_login():
"""
카카오 로그인 페이지로 리다이렉트
프론트엔드에서 이 URL을 호출하면 카카오 로그인 페이지로 이동합니다.
"""
auth_url = kakao_client.get_authorization_url()
return RedirectResponse(url=auth_url)
@router.get("/kakao/callback", response_model=TokenResponse)
async def kakao_callback(
code: str,
session: AsyncSession = Depends(get_session),
):
"""
카카오 로그인 콜백
카카오 로그인 성공 후 인가 코드를 받아 JWT 토큰을 발급합니다.
"""
# 1. 인가 코드로 액세스 토큰 획득
token_data = await kakao_client.get_access_token(code)
if "error" in token_data:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"카카오 토큰 발급 실패: {token_data.get('error_description', token_data.get('error'))}",
)
access_token = token_data.get("access_token")
if not access_token:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="액세스 토큰을 받지 못했습니다",
)
# 2. 액세스 토큰으로 사용자 정보 조회
user_info = await kakao_client.get_user_info(access_token)
if "id" not in user_info:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="사용자 정보를 가져오지 못했습니다",
)
kakao_id = str(user_info["id"])
kakao_account = user_info.get("kakao_account", {})
profile = kakao_account.get("profile", {})
nickname = profile.get("nickname")
email = kakao_account.get("email")
profile_image = profile.get("profile_image_url")
# 3. 기존 회원 확인 또는 신규 가입
result = await session.execute(select(User).where(User.kakao_id == kakao_id))
user = result.scalar_one_or_none()
if user is None:
# 신규 가입
user = User(
kakao_id=kakao_id,
nickname=nickname,
email=email,
profile_image=profile_image,
)
session.add(user)
await session.commit()
await session.refresh(user)
else:
# 기존 회원 - 마지막 로그인 시간 및 정보 업데이트
user.nickname = nickname
user.email = email
user.profile_image = profile_image
user.last_login_at = datetime.now(timezone.utc)
await session.commit()
await session.refresh(user)
# 4. JWT 토큰 발급
jwt_token = create_access_token({"sub": str(user.id)})
return TokenResponse(
access_token=jwt_token,
user=UserResponse.model_validate(user),
)
@router.get("/auth/me", response_model=AuthStatusResponse)
async def get_auth_status(
current_user: User | None = Depends(get_current_user_optional),
):
"""
현재 인증 상태 확인
프론트엔드에서 로그인 상태를 확인할 때 사용합니다.
토큰이 유효하면 사용자 정보를, 아니면 is_authenticated=False를 반환합니다.
"""
if current_user is None:
return AuthStatusResponse(is_authenticated=False)
return AuthStatusResponse(
is_authenticated=True,
user=UserResponse.model_validate(current_user),
)

71
app/auth/dependencies.py
View File

@ -1,71 +0,0 @@
"""
Auth 모듈 의존성 주입
"""
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.auth.models import User
from app.auth.services.jwt import decode_access_token
from app.database.session import get_session
security = HTTPBearer(auto_error=False)
async def get_current_user(
credentials: HTTPAuthorizationCredentials | None = Depends(security),
session: AsyncSession = Depends(get_session),
) -> User:
"""현재 로그인한 사용자 반환 (필수 인증)"""
if credentials is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="인증이 필요합니다",
)
payload = decode_access_token(credentials.credentials)
if payload is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="유효하지 않은 토큰입니다",
)
user_id = payload.get("sub")
if user_id is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="유효하지 않은 토큰입니다",
)
result = await session.execute(select(User).where(User.id == int(user_id)))
user = result.scalar_one_or_none()
if user is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="사용자를 찾을 수 없습니다",
)
return user
async def get_current_user_optional(
credentials: HTTPAuthorizationCredentials | None = Depends(security),
session: AsyncSession = Depends(get_session),
) -> User | None:
"""현재 로그인한 사용자 반환 (선택적 인증)"""
if credentials is None:
return None
payload = decode_access_token(credentials.credentials)
if payload is None:
return None
user_id = payload.get("sub")
if user_id is None:
return None
result = await session.execute(select(User).where(User.id == int(user_id)))
return result.scalar_one_or_none()

88
app/auth/models.py
View File

@ -1,88 +0,0 @@
"""
Auth 모듈 SQLAlchemy 모델 정의
카카오 로그인 사용자 정보를 저장합니다.
"""
from datetime import datetime
from sqlalchemy import DateTime, Index, Integer, String, func
from sqlalchemy.orm import Mapped, mapped_column
from app.database.session import Base
class User(Base):
"""
사용자 테이블 (카카오 로그인)
Attributes:
id: 고유 식별자 (자동 증가)
kakao_id: 카카오 고유 ID
nickname: 카카오 닉네임
email: 카카오 이메일 (선택)
profile_image: 프로필 이미지 URL
created_at: 가입 일시
last_login_at: 마지막 로그인 일시
"""
__tablename__ = "user"
__table_args__ = (
Index("idx_user_kakao_id", "kakao_id"),
{
"mysql_engine": "InnoDB",
"mysql_charset": "utf8mb4",
"mysql_collate": "utf8mb4_unicode_ci",
},
)
id: Mapped[int] = mapped_column(
Integer,
primary_key=True,
nullable=False,
autoincrement=True,
comment="고유 식별자",
)
kakao_id: Mapped[str] = mapped_column(
String(50),
nullable=False,
unique=True,
comment="카카오 고유 ID",
)
nickname: Mapped[str | None] = mapped_column(
String(100),
nullable=True,
comment="카카오 닉네임",
)
email: Mapped[str | None] = mapped_column(
String(255),
nullable=True,
comment="카카오 이메일",
)
profile_image: Mapped[str | None] = mapped_column(
String(500),
nullable=True,
comment="프로필 이미지 URL",
)
created_at: Mapped[datetime] = mapped_column(
DateTime,
nullable=False,
server_default=func.now(),
comment="가입 일시",
)
last_login_at: Mapped[datetime] = mapped_column(
DateTime,
nullable=False,
server_default=func.now(),
onupdate=func.now(),
comment="마지막 로그인 일시",
)
def __repr__(self) -> str:
return f"<User(id={self.id}, kakao_id='{self.kakao_id}', nickname='{self.nickname}')>"

36
app/auth/schemas.py
View File

@ -1,36 +0,0 @@
"""
Auth 모듈 Pydantic 스키마
"""
from datetime import datetime
from pydantic import BaseModel
class UserResponse(BaseModel):
"""사용자 정보 응답"""
id: int
kakao_id: str
nickname: str | None
email: str | None
profile_image: str | None
created_at: datetime
last_login_at: datetime
model_config = {"from_attributes": True}
class TokenResponse(BaseModel):
"""토큰 응답"""
access_token: str
token_type: str = "bearer"
user: UserResponse
class AuthStatusResponse(BaseModel):
"""인증 상태 응답 (프론트엔드용)"""
is_authenticated: bool
user: UserResponse | None = None

34
app/auth/services/jwt.py
View File

@ -1,34 +0,0 @@
"""
JWT 토큰 유틸리티
"""
from datetime import datetime, timedelta, timezone
from jose import JWTError, jwt
from config import security_settings
def create_access_token(data: dict) -> str:
"""JWT 액세스 토큰 생성"""
to_encode = data.copy()
expire = datetime.now(timezone.utc) + timedelta(minutes=security_settings.JWT_EXPIRE_MINUTES)
to_encode.update({"exp": expire})
return jwt.encode(
to_encode,
security_settings.JWT_SECRET,
algorithm=security_settings.JWT_ALGORITHM,
)
def decode_access_token(token: str) -> dict | None:
"""JWT 액세스 토큰 디코딩"""
try:
payload = jwt.decode(
token,
security_settings.JWT_SECRET,
algorithms=[security_settings.JWT_ALGORITHM],
)
return payload
except JWTError:
return None

55
app/auth/services/kakao.py
View File

@ -1,55 +0,0 @@
"""
카카오 OAuth API 클라이언트
"""
import aiohttp
from config import kakao_settings
class KakaoOAuthClient:
"""카카오 OAuth API 클라이언트"""
AUTH_URL = "https://kauth.kakao.com/oauth/authorize"
TOKEN_URL = "https://kauth.kakao.com/oauth/token"
USER_INFO_URL = "https://kapi.kakao.com/v2/user/me"
def __init__(self):
self.client_id = kakao_settings.KAKAO_CLIENT_ID
self.client_secret = kakao_settings.KAKAO_CLIENT_SECRET
self.redirect_uri = kakao_settings.KAKAO_REDIRECT_URI
def get_authorization_url(self) -> str:
"""카카오 로그인 페이지 URL 반환"""
return (
f"{self.AUTH_URL}"
f"?client_id={self.client_id}"
f"&redirect_uri={self.redirect_uri}"
f"&response_type=code"
)
async def get_access_token(self, code: str) -> dict:
"""인가 코드로 액세스 토큰 획득"""
async with aiohttp.ClientSession() as session:
data = {
"grant_type": "authorization_code",
"client_id": self.client_id,
"client_secret": self.client_secret,
"redirect_uri": self.redirect_uri,
"code": code,
}
print(f"[kakao] Token request - client_id: {self.client_id}, redirect_uri: {self.redirect_uri}")
async with session.post(self.TOKEN_URL, data=data) as response:
result = await response.json()
print(f"[kakao] Token response: {result}")
return result
async def get_user_info(self, access_token: str) -> dict:
"""액세스 토큰으로 사용자 정보 조회"""
async with aiohttp.ClientSession() as session:
headers = {"Authorization": f"Bearer {access_token}"}
async with session.get(self.USER_INFO_URL, headers=headers) as response:
return await response.json()
kakao_client = KakaoOAuthClient()

6
app/home/api/routers/v1/home.py
View File

@ -102,7 +102,7 @@ def _extract_region_from_address(road_address: str | None) -> str:
"model": ErrorResponse,
},
},
tags=["crawling"],
tags=["Crawling"],
)
async def crawling(request_body: CrawlingRequest):
"""네이버 지도 장소 크롤링"""
@ -379,7 +379,7 @@ print(response.json())
200: {"description": "이미지 업로드 성공"},
400: {"description": "이미지가 제공되지 않음", "model": ErrorResponse},
},
tags=["image"],
tags=["Image-Server"],
)
async def upload_images(
images_json: Optional[str] = Form(
@ -597,7 +597,7 @@ curl -X POST "http://localhost:8000/image/upload/blob" \\
200: {"description": "이미지 업로드 성공"},
400: {"description": "이미지가 제공되지 않음", "model": ErrorResponse},
},
tags=["image"],
tags=["Image-Blob"],
)
async def upload_images_blob(
images_json: Optional[str] = Form(

10
app/lyric/api/routers/v1/lyric.py
View File

@ -49,7 +49,7 @@ import traceback as tb
# 로거 설정
logger = get_logger("lyric")
router = APIRouter(prefix="/lyric", tags=["lyric"])
router = APIRouter(prefix="/lyric", tags=["Lyric"])
# =============================================================================
@ -398,7 +398,7 @@ async def get_lyric_status(
@router.get(
"s",
"s/",
summary="가사 목록 조회 (페이지네이션)",
description="""
생성 완료된 가사를 페이지네이션으로 조회합니다.
@ -418,9 +418,9 @@ async def get_lyric_status(
## 사용 예시
```
GET /lyrics # 기본 조회 (1페이지, 20개)
GET /lyrics?page=2 # 2페이지 조회
GET /lyrics?page=1&page_size=50 # 50개씩 조회
GET /lyrics/ # 기본 조회 (1페이지, 20개)
GET /lyrics/?page=2 # 2페이지 조회
GET /lyrics/?page=1&page_size=50 # 50개씩 조회
```
## 참고

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

@ -5,7 +5,7 @@ Song API Router
엔드포인트 목록:
- POST /song/generate/{task_id}: 노래 생성 요청 (task_id로 Project/Lyric 연결)
- GET /song/status/{suno_task_id}: Suno API 노래 생성 상태 조회
- GET /song/status/{song_id}: Suno API 노래 생성 상태 조회
- GET /song/download/{task_id}: 노래 다운로드 상태 조회 (DB polling)
사용 예시:
@ -13,7 +13,7 @@ Song API Router
app.include_router(router, prefix="/api/v1")
"""
from fastapi import APIRouter, Depends, HTTPException
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
@ -24,7 +24,8 @@ from app.dependencies.pagination import (
)
from app.home.models import Project
from app.lyric.models import Lyric
from app.song.models import Song
from app.song.models import Song, SongTimestamp
from app.song.schemas.song_schema import (
DownloadSongResponse,
GenerateSongRequest,
@ -32,6 +33,7 @@ from app.song.schemas.song_schema import (
PollingSongResponse,
SongListItem,
)
from app.song.worker.song_task import download_and_upload_song_by_suno_task_id
from app.utils.logger import get_logger
from app.utils.pagination import PaginatedResponse
from app.utils.suno import SunoService
@ -58,7 +60,7 @@ Suno API를 통해 노래 생성을 요청합니다.
## 반환 정보
- **success**: 요청 성공 여부
- **task_id**: 내부 작업 ID (Project/Lyric task_id)
- **suno_task_id**: Suno API 작업 ID (상태 조회에 사용)
- **song_id**: Suno API 작업 ID (상태 조회에 사용)
- **message**: 응답 메시지
## 사용 예시
@ -73,7 +75,7 @@ POST /song/generate/019123ab-cdef-7890-abcd-ef1234567890
## 참고
- 생성되는 노래는 약 1분 이내 길이입니다.
- suno_task_id를 사용하여 /status/{suno_task_id} 엔드포인트에서 생성 상태를 확인할 수 있습니다.
- song_id를 사용하여 /status/{song_id} 엔드포인트에서 생성 상태를 확인할 수 있습니다.
- Song 테이블에 데이터가 저장되며, project_id와 lyric_id가 자동으로 연결됩니다.
""",
response_model=GenerateSongResponse,
@ -193,7 +195,7 @@ async def generate_song(
return GenerateSongResponse(
success=False,
task_id=task_id,
suno_task_id=None,
song_id=None,
message="노래 생성 요청에 실패했습니다.",
error_message=str(e),
)
@ -237,7 +239,7 @@ async def generate_song(
return GenerateSongResponse(
success=False,
task_id=task_id,
suno_task_id=None,
song_id=None,
message="노래 생성 요청에 실패했습니다.",
error_message=str(e),
)
@ -273,8 +275,8 @@ async def generate_song(
return GenerateSongResponse(
success=True,
task_id=task_id,
suno_task_id=suno_task_id,
message="노래 생성 요청이 접수되었습니다. suno_task_id로 상태를 조회하세요.",
song_id=suno_task_id,
message="노래 생성 요청이 접수되었습니다. song_id로 상태를 조회하세요.",
error_message=None,
)
@ -286,119 +288,170 @@ async def generate_song(
return GenerateSongResponse(
success=False,
task_id=task_id,
suno_task_id=suno_task_id,
song_id=suno_task_id,
message="노래 생성은 요청되었으나 DB 업데이트에 실패했습니다.",
error_message=str(e),
)
@router.get(
"/status/{suno_task_id}",
summary="노래 생성 상태 조회",
"/status/{song_id}",
summary="노래 생성 상태 조회 (Suno API)",
description="""
Suno API를 통해 노래 생성 작업의 상태를 조회합니다.
SUCCESS 상태인 경우 백그라운드에서 MP3 파일을 다운로드하고 Azure Blob Storage에 업로드한 뒤 Song 테이블을 업데이트합니다.
SUCCESS 상태인 경우 백그라운드에서 MP3 파일을 다운로드하고 Azure Blob Storage에 업로드를 시작합니다.
## 경로 파라미터
- **suno_task_id**: 노래 생성 시 반환된 Suno API 작업 ID (필수)
- **song_id**: 노래 생성 시 반환된 Suno API 작업 ID (필수)
## 반환 정보
- **success**: 조회 성공 여부
- **status**: 작업 상태 (PENDING, processing, SUCCESS, failed)
- **status**: Suno API 작업 상태
- **message**: 상태 메시지
- **clips**: 생성된 노래 클립 목록 (완료 시)
- **raw_response**: Suno API 원본 응답
## 사용 예시
```
GET /song/status/abc123...
```
## 상태 값
- **PENDING**: 대기 중
- **processing**: 생성 중
- **SUCCESS**: 생성 완료
- **failed**: 생성 실패
## 상태 값 (Suno API 응답)
- **PENDING**: Suno API 대기 중
- **processing**: Suno API에서 노래 생성 중
- **SUCCESS**: Suno API 노래 생성 완료 (백그라운드 Blob 업로드 시작)
- **TEXT_SUCCESS**: Suno API 노래 생성 완료
- **failed**: Suno API 노래 생성 실패
- **error**: API 조회 오류
## 참고
- 스트림 URL: 30-40초 내 생성
- 다운로드 URL: 2-3분 내 생성
- SUCCESS 시 백그라운드에서 MP3 다운로드 → Azure Blob Storage 업로드 → Song 테이블 업데이트 진행
- 저장 경로: Azure Blob Storage ({BASE_URL}/{task_id}/song/{store_name}.mp3)
- Song 테이블의 song_result_url에 Blob URL이 저장됩니다
- 이 엔드포인트는 Suno API의 상태를 반환합니다
- SUCCESS 응답 시 백그라운드에서 MP3 다운로드 → Azure Blob Storage 업로드가 시작됩니다
- 최종 완료 상태는 `/song/download/{task_id}` 엔드포인트에서 확인하세요
- Song 테이블 상태: processing → uploading → completed
""",
response_model=PollingSongResponse,
responses={
200: {"description": "상태 조회 성공"},
500: {"description": "상태 조회 실패"},
},
)
async def get_song_status(
suno_task_id: str,
song_id: str,
background_tasks: BackgroundTasks,
session: AsyncSession = Depends(get_session),
) -> PollingSongResponse:
"""suno_task_id로 노래 생성 작업의 상태를 조회합니다.
"""song_id로 노래 생성 작업의 상태를 조회합니다.
SUCCESS 상태인 경우 백그라운드에서 MP3 파일을 다운로드하고
Azure Blob Storage에 업로드한 뒤 Song 테이블의 status를 completed로,
song_result_url을 Blob URL로 업데이트합니다.
"""
logger.info(f"[get_song_status] START - suno_task_id: {suno_task_id}")
suno_task_id = song_id # 임시방편 / 외부 suno 노출 방지
logger.info(f"[get_song_status] START - song_id: {suno_task_id}")
try:
suno_service = SunoService()
result = await suno_service.get_task_status(suno_task_id)
logger.debug(f"[get_song_status] Suno API raw response - song_id: {suno_task_id}, result: {result}")
parsed_response = suno_service.parse_status_response(result)
logger.info(f"[get_song_status] Suno API response - suno_task_id: {suno_task_id}, status: {parsed_response.status}")
logger.info(f"[get_song_status] Suno API response - song_id: {suno_task_id}, status: {parsed_response.status}")
# SUCCESS 상태인 경우 첫 번째 클립 정보를 DB에 직접 저장
if parsed_response.status == "SUCCESS" and parsed_response.clips:
# 첫 번째 클립(clips[0])의 audioUrl과 duration 사용
first_clip = parsed_response.clips[0]
audio_url = first_clip.audio_url
clip_duration = first_clip.duration
logger.debug(f"[get_song_status] Using first clip - id: {first_clip.id}, audio_url: {audio_url}, duration: {clip_duration}")
# SUCCESS 상태인 경우 백그라운드에서 MP3 다운로드 및 Blob 업로드 진행
if parsed_response.status == "SUCCESS" and result:
# result에서 직접 clips 데이터 추출
data = result.get("data", {})
response_data = data.get("response") or {}
clips_data = response_data.get("sunoData") or []
if audio_url:
# suno_task_id로 Song 조회
song_result = await session.execute(
select(Song)
.where(Song.suno_task_id == suno_task_id)
.order_by(Song.created_at.desc())
.limit(1)
)
song = song_result.scalar_one_or_none()
if clips_data:
# 첫 번째 클립(clips[0])의 audioUrl과 duration 사용
first_clip = clips_data[0]
audio_url = first_clip.get("audioUrl")
clip_duration = first_clip.get("duration")
logger.debug(f"[get_song_status] Using first clip - id: {first_clip.get('id')}, audio_url: {audio_url}, duration: {clip_duration}")
if song and song.status != "completed":
# 첫 번째 클립의 audio_url과 duration을 직접 DB에 저장
song.status = "completed"
song.song_result_url = audio_url
if clip_duration is not None:
song.duration = clip_duration
await session.commit()
logger.info(f"[get_song_status] Song updated - suno_task_id: {suno_task_id}, status: completed, song_result_url: {audio_url}, duration: {clip_duration}")
elif song and song.status == "completed":
logger.info(f"[get_song_status] SKIPPED - Song already completed, suno_task_id: {suno_task_id}")
if audio_url:
# song_id로 Song 조회
song_result = await session.execute(
select(Song)
.where(Song.suno_task_id == suno_task_id)
.order_by(Song.created_at.desc())
.limit(1)
)
song = song_result.scalar_one_or_none()
logger.info(f"[get_song_status] SUCCESS - suno_task_id: {suno_task_id}")
# processing 상태인 경우에만 백그라운드 태스크 실행 (중복 방지)
if song and song.status == "processing":
# store_name 조회
project_result = await session.execute(
select(Project).where(Project.id == song.project_id)
)
project = project_result.scalar_one_or_none()
store_name = project.store_name if project else "song"
# 상태를 uploading으로 변경 (중복 호출 방지)
song.status = "uploading"
song.suno_audio_id = first_clip.get('id')
await session.commit()
logger.info(f"[get_song_status] Song status changed to uploading - song_id: {suno_task_id}")
# 백그라운드 태스크로 MP3 다운로드 및 Blob 업로드 실행
background_tasks.add_task(
download_and_upload_song_by_suno_task_id,
suno_task_id=suno_task_id,
audio_url=audio_url,
store_name=store_name,
duration=clip_duration,
)
logger.info(f"[get_song_status] Background task scheduled - song_id: {suno_task_id}, store_name: {store_name}")
suno_audio_id = first_clip.get('id')
word_data = await suno_service.get_lyric_timestamp(suno_task_id, suno_audio_id)
lyric_result = await session.execute(
select(Lyric)
.where(Lyric.task_id == song.task_id)
.order_by(Lyric.created_at.desc())
.limit(1)
)
lyric = lyric_result.scalar_one_or_none()
gt_lyric = lyric.lyric_result
lyric_line_list = gt_lyric.split("\n")
sentences = [lyric_line.strip(',. ') for lyric_line in lyric_line_list if lyric_line and lyric_line != "---"]
timestamped_lyrics = suno_service.align_lyrics(word_data, sentences)
# TODO : DB upload timestamped_lyrics
for order_idx, timestamped_lyric in enumerate(timestamped_lyrics):
song_timestamp = SongTimestamp(
suno_audio_id = suno_audio_id,
order_idx = order_idx,
lyric_line = timestamped_lyric["text"],
start_time = timestamped_lyric["start_sec"],
end_time = timestamped_lyric["end_sec"]
)
session.add(song_timestamp)
await session.commit()
elif song and song.status == "uploading":
logger.info(f"[get_song_status] SKIPPED - Song is already uploading, song_id: {suno_task_id}")
elif song and song.status == "completed":
logger.info(f"[get_song_status] SKIPPED - Song already completed, song_id: {suno_task_id}")
logger.info(f"[get_song_status] SUCCESS - song_id: {suno_task_id}")
return parsed_response
except Exception as e:
import traceback
logger.error(f"[get_song_status] EXCEPTION - suno_task_id: {suno_task_id}, error: {e}")
logger.error(f"[get_song_status] EXCEPTION - song_id: {song_id}, error: {e}")
return PollingSongResponse(
success=False,
status="error",
message="상태 조회에 실패했습니다.",
clips=None,
raw_response=None,
error_message=f"{type(e).__name__}: {e}\n{traceback.format_exc()}",
)
@router.get(
"/download/{task_id}",
summary="노래 생성 URL 조회",
summary="노래 다운로드 상태 조회 (DB Polling)",
description="""
task_id를 기반으로 Song 테이블의 상태를 조회하고,
completed인 경우 Project 정보와 노래 URL을 반환합니다.
@ -408,31 +461,38 @@ completed인 경우 Project 정보와 노래 URL을 반환합니다.
## 반환 정보
- **success**: 조회 성공 여부
- **status**: 처리 상태 (processing, completed, failed, not_found)
- **status**: DB 처리 상태 (processing, uploading, completed, failed, not_found, error)
- **message**: 응답 메시지
- **store_name**: 업체명
- **region**: 지역명
- **detail_region_info**: 상세 지역 정보
- **store_name**: 업체명 (completed 시)
- **region**: 지역명 (completed 시)
- **detail_region_info**: 상세 지역 정보 (completed 시)
- **task_id**: 작업 고유 식별자
- **language**: 언어
- **language**: 언어 (completed 시)
- **song_result_url**: 노래 결과 URL (completed 시, Azure Blob Storage URL)
- **created_at**: 생성 일시
- **created_at**: 생성 일시 (completed 시)
- **error_message**: 에러 메시지 (실패 시)
## 사용 예시
```
GET /song/download/019123ab-cdef-7890-abcd-ef1234567890
```
## 상태 값 (DB 상태)
- **processing**: Suno API에서 노래 생성 중
- **uploading**: MP3 다운로드 및 Azure Blob 업로드 중
- **completed**: 모든 작업 완료, Blob URL 사용 가능
- **failed**: 노래 생성 또는 업로드 실패
- **not_found**: task_id에 해당하는 Song 없음
- **error**: 조회 중 오류 발생
## 참고
- processing 상태인 경우 song_result_url은 null입니다.
- completed 상태인 경우 Project 정보와 함께 song_result_url (Azure Blob URL)을 반환합니다.
- 이 엔드포인트는 DB의 Song 테이블 상태를 반환합니다
- completed 상태인 경우 Project 정보와 함께 song_result_url (Azure Blob URL)을 반환합니다
- song_result_url 형식: {AZURE_BLOB_BASE_URL}/{task_id}/song/{store_name}.mp3
""",
response_model=DownloadSongResponse,
responses={
200: {"description": "조회 성공"},
404: {"description": "Song을 찾을 수 없음"},
500: {"description": "조회 실패"},
200: {"description": "조회 성공 (모든 상태에서 200 반환)"},
},
)
async def download_song(
@ -472,6 +532,16 @@ async def download_song(
task_id=task_id,
)
# uploading 상태인 경우
if song.status == "uploading":
logger.info(f"[download_song] UPLOADING - task_id: {task_id}")
return DownloadSongResponse(
success=True,
status="uploading",
message="노래 파일을 업로드 중입니다.",
task_id=task_id,
)
# failed 상태인 경우
if song.status == "failed":
logger.warning(f"[download_song] FAILED - task_id: {task_id}")
@ -544,7 +614,6 @@ GET /songs/?page=1&page_size=10
response_model=PaginatedResponse[SongListItem],
responses={
200: {"description": "노래 목록 조회 성공"},
500: {"description": "조회 실패"},
},
)
async def get_songs(

101
app/song/models.py
View File

@ -1,7 +1,7 @@
from datetime import datetime
from typing import TYPE_CHECKING, List, Optional
from sqlalchemy import DateTime, ForeignKey, Integer, String, Text, func
from sqlalchemy import DateTime, Float, ForeignKey, Integer, String, Text, func
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database.session import Base
@ -25,7 +25,7 @@ class Song(Base):
lyric_id: 연결된 Lyric의 id (외래키)
task_id: 노래 생성 작업의 고유 식별자 (UUID 형식)
suno_task_id: Suno API 작업 고유 식별자 (선택)
status: 처리 상태 (pending, processing, completed, failed 등)
status: 처리 상태 (processing, uploading, completed, failed)
song_prompt: 노래 생성에 사용된 프롬프트
song_result_url: 생성 결과 URL (선택)
language: 출력 언어
@ -82,10 +82,16 @@ class Song(Base):
comment="Suno API 작업 고유 식별자",
)
suno_audio_id: Mapped[Optional[str]] = mapped_column(
String(64),
nullable=True,
comment="Suno 첫번째 노래의 고유 식별자",
)
status: Mapped[str] = mapped_column(
String(50),
nullable=False,
comment="처리 상태 (processing, completed, failed)",
comment="처리 상태 (processing, uploading, completed, failed)",
)
song_prompt: Mapped[str] = mapped_column(
@ -150,3 +156,92 @@ class Song(Base):
f"status='{self.status}'"
f")>"
)
class SongTimestamp(Base):
"""
노래 타임스탬프 테이블
노래의 가사별 시작/종료 시간 정보를 저장합니다.
Suno API에서 반환된 타임스탬프 데이터를 기반으로 생성됩니다.
Attributes:
id: 고유 식별자 (자동 증가)
suno_audio_id: 가사의 원본 오디오 ID
order_idx: 오디오 내에서 가사의 순서
lyric_line: 가사 한 줄의 내용
start_time: 가사 시작 시점 (초)
end_time: 가사 종료 시점 (초)
created_at: 생성 일시 (자동 설정)
"""
__tablename__ = "song_timestamp"
__table_args__ = (
{
"mysql_engine": "InnoDB",
"mysql_charset": "utf8mb4",
"mysql_collate": "utf8mb4_unicode_ci",
},
)
id: Mapped[int] = mapped_column(
Integer,
primary_key=True,
nullable=False,
autoincrement=True,
comment="고유 식별자",
)
suno_audio_id: Mapped[str] = mapped_column(
String(64),
nullable=False,
index=True,
comment="가사의 원본 오디오 ID",
)
order_idx: Mapped[int] = mapped_column(
Integer,
nullable=False,
comment="오디오 내에서 가사의 순서",
)
lyric_line: Mapped[str] = mapped_column(
Text,
nullable=False,
comment="가사 한 줄의 내용",
)
start_time: Mapped[float] = mapped_column(
Float,
nullable=False,
comment="가사 시작 시점 (초)",
)
end_time: Mapped[float] = mapped_column(
Float,
nullable=False,
comment="가사 종료 시점 (초)",
)
created_at: Mapped[datetime] = mapped_column(
DateTime,
nullable=False,
server_default=func.now(),
comment="생성 일시",
)
def __repr__(self) -> str:
def truncate(value: str | None, max_len: int = 10) -> str:
if value is None:
return "None"
return (value[:max_len] + "...") if len(value) > max_len else value
return (
f"<SongTimestamp("
f"id={self.id}, "
f"suno_audio_id='{truncate(self.suno_audio_id)}', "
f"order_idx={self.order_idx}, "
f"start_time={self.start_time}, "
f"end_time={self.end_time}"
f")>"
)

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

@ -1,6 +1,6 @@
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Dict, List, Optional
from typing import Dict, List, Optional
from fastapi import Request
from pydantic import BaseModel, Field
@ -64,8 +64,8 @@ class GenerateSongResponse(BaseModel):
{
"success": true,
"task_id": "019123ab-cdef-7890-abcd-ef1234567890",
"suno_task_id": "abc123...",
"message": "노래 생성 요청이 접수되었습니다. suno_task_id로 상태를 조회하세요.",
"song_id": "abc123...",
"message": "노래 생성 요청이 접수되었습니다. song_id로 상태를 조회하세요.",
"error_message": null
}
@ -73,7 +73,7 @@ class GenerateSongResponse(BaseModel):
{
"success": false,
"task_id": "019123ab-cdef-7890-abcd-ef1234567890",
"suno_task_id": null,
"song_id": null,
"message": "노래 생성 요청에 실패했습니다.",
"error_message": "Suno API connection error"
}
@ -81,7 +81,7 @@ class GenerateSongResponse(BaseModel):
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")
song_id: Optional[str] = Field(None, description="Suno API 작업 ID")
message: str = Field(..., description="응답 메시지")
error_message: Optional[str] = Field(None, description="에러 메시지 (실패 시)")
@ -90,7 +90,7 @@ class PollingSongRequest(BaseModel):
"""노래 생성 상태 조회 요청 스키마 (Legacy)
Note:
현재 사용되지 않음. GET /song/status/{suno_task_id} 엔드포인트 사용.
현재 사용되지 않음. GET /song/status/{song_id} 엔드포인트 사용.
Example Request:
{
@ -114,32 +114,39 @@ class SongClipData(BaseModel):
class PollingSongResponse(BaseModel):
"""노래 생성 상태 조회 응답 스키마
"""노래 생성 상태 조회 응답 스키마 (Suno API)
Usage:
GET /song/status/{suno_task_id}
GET /song/status/{song_id}
Suno API 작업 상태를 조회합니다.
Note:
상태 값:
- PENDING: 대기 중
- processing: 생성 중
- SUCCESS / TEXT_SUCCESS / complete: 생성 완료
- failed: 생성 실패
상태 값 (Suno API 응답):
- PENDING: Suno API 대기 중
- processing: Suno API에서 노래 생성 중
- SUCCESS: Suno API 노래 생성 완료 (백그라운드 Blob 업로드 시작)
- TEXT_SUCCESS: Suno API 노래 생성 완료
- failed: Suno API 노래 생성 실패
- error: API 조회 오류
SUCCESS 상태 시:
- 백그라운드에서 MP3 파일 다운로드 시작
- Song 테이블의 status를 completed로 업데이트
- song_result_url에 로컬 파일 경로 저장
- 백그라운드에서 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": "노래를 생성하고 있습니다.",
"clips": null,
"raw_response": {...},
"error_message": null
}
@ -148,18 +155,6 @@ class PollingSongResponse(BaseModel):
"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
}
@ -168,19 +163,15 @@ class PollingSongResponse(BaseModel):
"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)"
None, description="Suno API 작업 상태 (PENDING, processing, SUCCESS, TEXT_SUCCESS, failed, error)"
)
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="에러 메시지 (실패 시)")
@ -210,17 +201,18 @@ class SongListItem(BaseModel):
class DownloadSongResponse(BaseModel):
"""노래 다운로드 응답 스키마
"""노래 다운로드 응답 스키마 (DB Polling)
Usage:
GET /song/download/{task_id}
Polls for song completion and returns project info with song URL.
DB의 Song 테이블 상태를 조회하고 완료 시 Project 정보와 노래 URL을 반환합니다.
Note:
상태 값:
- processing: 노래 생성 진행 중 (song_result_url은 null)
- completed: 노래 생성 완료 (song_result_url 포함)
- failed: 노래 생성 실패
상태 값 (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: 조회 중 오류 발생
@ -239,6 +231,21 @@ class DownloadSongResponse(BaseModel):
"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,
@ -249,7 +256,7 @@ class DownloadSongResponse(BaseModel):
"detail_region_info": "군산 신흥동 말랭이 마을",
"task_id": "019123ab-cdef-7890-abcd-ef1234567890",
"language": "Korean",
"song_result_url": "http://localhost:8000/media/2025-01-15/스테이머뭄.mp3",
"song_result_url": "https://blob.azure.com/.../song.mp3",
"created_at": "2025-01-15T12:00:00",
"error_message": null
}
@ -270,8 +277,8 @@ class DownloadSongResponse(BaseModel):
}
"""
success: bool = Field(..., description="다운로드 성공 여부")
status: str = Field(..., description="처리 상태 (processing, completed, failed, not_found, error)")
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="지역명")

86
app/song/worker/song_task.py
View File

@ -152,7 +152,7 @@ async def download_and_save_song(
# 프론트엔드에서 접근 가능한 URL 생성
relative_path = f"/media/song/{today}/{unique_id}/{file_name}"
base_url = f"http://{prj_settings.PROJECT_DOMAIN}"
base_url = f"{prj_settings.PROJECT_DOMAIN}"
file_url = f"{base_url}{relative_path}"
logger.info(f"[download_and_save_song] URL generated - task_id: {task_id}, url: {file_url}")
@ -173,90 +173,6 @@ async def download_and_save_song(
await _update_song_status(task_id, "failed")
async def download_and_upload_song_to_blob(
task_id: str,
audio_url: str,
store_name: str,
) -> None:
"""백그라운드에서 노래를 다운로드하고 Azure Blob Storage에 업로드한 뒤 Song 테이블을 업데이트합니다.
Args:
task_id: 프로젝트 task_id
audio_url: 다운로드할 오디오 URL
store_name: 저장할 파일명에 사용할 업체명
"""
logger.info(f"[download_and_upload_song_to_blob] START - task_id: {task_id}, store_name: {store_name}")
temp_file_path: Path | None = None
try:
# 파일명에 사용할 수 없는 문자 제거
safe_store_name = "".join(
c for c in store_name if c.isalnum() or c in (" ", "_", "-")
).strip()
safe_store_name = safe_store_name or "song"
file_name = f"{safe_store_name}.mp3"
# 임시 저장 경로 생성
temp_dir = Path("media") / "temp" / task_id
temp_dir.mkdir(parents=True, exist_ok=True)
temp_file_path = temp_dir / file_name
logger.info(f"[download_and_upload_song_to_blob] Temp directory created - path: {temp_file_path}")
# 오디오 파일 다운로드
logger.info(f"[download_and_upload_song_to_blob] Downloading audio - task_id: {task_id}, url: {audio_url}")
content = await _download_audio(audio_url, task_id)
async with aiofiles.open(str(temp_file_path), "wb") as f:
await f.write(content)
logger.info(f"[download_and_upload_song_to_blob] File downloaded - task_id: {task_id}, path: {temp_file_path}")
# Azure Blob Storage에 업로드
uploader = AzureBlobUploader(task_id=task_id)
upload_success = await uploader.upload_music(file_path=str(temp_file_path))
if not upload_success:
raise Exception("Azure Blob Storage 업로드 실패")
# SAS 토큰이 제외된 public_url 사용
blob_url = uploader.public_url
logger.info(f"[download_and_upload_song_to_blob] Uploaded to Blob - task_id: {task_id}, url: {blob_url}")
# Song 테이블 업데이트
await _update_song_status(task_id, "completed", blob_url)
logger.info(f"[download_and_upload_song_to_blob] SUCCESS - task_id: {task_id}")
except httpx.HTTPError as e:
logger.error(f"[download_and_upload_song_to_blob] DOWNLOAD ERROR - task_id: {task_id}, error: {e}", exc_info=True)
await _update_song_status(task_id, "failed")
except SQLAlchemyError as e:
logger.error(f"[download_and_upload_song_to_blob] DB ERROR - task_id: {task_id}, error: {e}", exc_info=True)
await _update_song_status(task_id, "failed")
except Exception as e:
logger.error(f"[download_and_upload_song_to_blob] EXCEPTION - task_id: {task_id}, error: {e}", exc_info=True)
await _update_song_status(task_id, "failed")
finally:
# 임시 파일 삭제
if temp_file_path and temp_file_path.exists():
try:
temp_file_path.unlink()
logger.info(f"[download_and_upload_song_to_blob] Temp file deleted - path: {temp_file_path}")
except Exception as e:
logger.warning(f"[download_and_upload_song_to_blob] Failed to delete temp file: {e}")
# 임시 디렉토리 삭제 시도
temp_dir = Path("media") / "temp" / task_id
if temp_dir.exists():
try:
temp_dir.rmdir()
except Exception:
pass # 디렉토리가 비어있지 않으면 무시
async def download_and_upload_song_by_suno_task_id(
suno_task_id: str,
audio_url: str,

91
app/user/api/routers/v1/auth.py
View File

@ -4,18 +4,22 @@
카카오 로그인, 토큰 갱신, 로그아웃, 내 정보 조회 엔드포인트를 제공합니다.
"""
import logging
from typing import Optional
from fastapi import APIRouter, Depends, Header, Request, status
from fastapi.responses import Response
from fastapi.responses import RedirectResponse, Response
from sqlalchemy.ext.asyncio import AsyncSession
from config import prj_settings
from app.database.session import get_session
logger = logging.getLogger(__name__)
from app.user.dependencies import get_current_user
from app.user.models import User
from app.user.schemas.user_schema import (
AccessTokenResponse,
KakaoCallbackRequest,
KakaoCodeRequest,
KakaoLoginResponse,
LoginResponse,
RefreshTokenRequest,
@ -39,30 +43,33 @@ async def kakao_login() -> KakaoLoginResponse:
프론트엔드에서 이 URL로 사용자를 리다이렉트하면
카카오 로그인 페이지가 표시됩니다.
"""
logger.info("[ROUTER] 카카오 로그인 URL 요청")
auth_url = kakao_client.get_authorization_url()
logger.debug(f"[ROUTER] 카카오 인증 URL 생성 완료 - auth_url: {auth_url}")
return KakaoLoginResponse(auth_url=auth_url)
@router.post(
@router.get(
"/kakao/callback",
response_model=LoginResponse,
summary="카카오 로그인 콜백",
description="카카오 인가 코드를 받아 로그인/가입을 처리하고 JWT 토큰을 발급합니다.",
description="카카오 인가 코드를 받아 로그인/가입을 처리하고 프론트엔드로 리다이렉트합니다.",
)
async def kakao_callback(
request: Request,
body: KakaoCallbackRequest,
code: str,
session: AsyncSession = Depends(get_session),
user_agent: Optional[str] = Header(None, alias="User-Agent"),
) -> LoginResponse:
) -> RedirectResponse:
"""
카카오 로그인 콜백
카카오 로그인 성공 후 발급받은 인가 코드로
JWT 토큰을 발급합니다.
JWT 토큰을 발급하고 프론트엔드로 리다이렉트합니다.
신규 사용자인 경우 자동으로 회원가입이 처리됩니다.
"""
logger.info(f"[ROUTER] 카카오 콜백 수신 - code: {code[:20]}...")
# 클라이언트 IP 추출
ip_address = request.client.host if request.client else None
@ -71,13 +78,79 @@ async def kakao_callback(
if forwarded_for:
ip_address = forwarded_for.split(",")[0].strip()
return await auth_service.kakao_login(
logger.debug(f"[ROUTER] 클라이언트 정보 - ip: {ip_address}, user_agent: {user_agent}")
result = await auth_service.kakao_login(
code=code,
session=session,
user_agent=user_agent,
ip_address=ip_address,
)
# 프론트엔드로 토큰과 함께 리다이렉트
redirect_url = (
f"https://{prj_settings.PROJECT_DOMAIN}"
f"?access_token={result.access_token}"
f"&refresh_token={result.refresh_token}"
)
logger.info(f"[ROUTER] 카카오 콜백 완료, 프론트엔드로 리다이렉트 - redirect_url: {redirect_url[:50]}...")
return RedirectResponse(url=redirect_url, status_code=302)
@router.post(
"/kakao/verify",
response_model=LoginResponse,
summary="카카오 인가 코드 검증 및 토큰 발급",
description="""
프론트엔드에서 카카오 로그인 후 받은 인가 코드를 검증하고 JWT 토큰을 발급합니다.
## 사용 시나리오
1. 프론트엔드가 카카오 로그인 완료 후 인가 코드(code)를 받음
2. 프론트엔드가 이 엔드포인트에 code를 POST로 전달
3. 서버가 카카오 서버에 code 검증 및 사용자 정보 조회
4. JWT 토큰 발급 및 사용자 정보 반환
## 응답
- 신규 사용자인 경우 `user.is_new_user`가 `true`로 반환됩니다.
- `redirect_url`은 로그인 후 이동할 프론트엔드 URL입니다.
""",
)
async def kakao_verify(
request: Request,
body: KakaoCodeRequest,
session: AsyncSession = Depends(get_session),
user_agent: Optional[str] = Header(None, alias="User-Agent"),
) -> LoginResponse:
"""
카카오 인가 코드 검증 및 토큰 발급
프론트엔드가 카카오 콜백에서 받은 인가 코드를 전달하면
카카오 서버에서 검증 후 JWT 토큰을 발급합니다.
신규 사용자인 경우 자동으로 회원가입이 처리됩니다.
"""
logger.info(f"[ROUTER] 카카오 인가 코드 검증 요청 - code: {body.code[:20]}...")
# 클라이언트 IP 추출
ip_address = request.client.host if request.client else None
# X-Forwarded-For 헤더 확인 (프록시/로드밸런서 뒤에 있는 경우)
forwarded_for = request.headers.get("X-Forwarded-For")
if forwarded_for:
ip_address = forwarded_for.split(",")[0].strip()
logger.debug(f"[ROUTER] 클라이언트 정보 - ip: {ip_address}, user_agent: {user_agent}")
result = await auth_service.kakao_login(
code=body.code,
session=session,
user_agent=user_agent,
ip_address=ip_address,
)
logger.info(f"[ROUTER] 카카오 인가 코드 검증 완료 - user_id: {result.user.id}, is_new_user: {result.user.is_new_user}")
return result
@router.post(
"/refresh",

191
app/user/models.py
View File

@ -7,7 +7,8 @@ User 모듈 SQLAlchemy 모델 정의
from datetime import date, datetime
from typing import TYPE_CHECKING, List, Optional
from sqlalchemy import BigInteger, Boolean, Date, DateTime, ForeignKey, Index, Integer, String, func
from sqlalchemy import BigInteger, Boolean, Date, DateTime, ForeignKey, Index, Integer, String, Text, func
from sqlalchemy.dialects.mysql import JSON
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database.session import Base
@ -246,6 +247,18 @@ class User(Base):
lazy="selectin",
)
# ==========================================================================
# SocialAccount 1:N 관계
# ==========================================================================
# 한 사용자는 여러 소셜 계정을 연동할 수 있음 (YouTube, Instagram, Facebook)
# ==========================================================================
social_accounts: Mapped[List["SocialAccount"]] = relationship(
"SocialAccount",
back_populates="user",
cascade="all, delete-orphan",
lazy="selectin",
)
def __repr__(self) -> str:
return (
f"<User("
@ -373,3 +386,179 @@ class RefreshToken(Base):
f"expires_at={self.expires_at}"
f")>"
)
class SocialAccount(Base):
"""
소셜 계정 연동 테이블
사용자가 연동한 외부 소셜 플랫폼 계정 정보를 저장합니다.
YouTube, Instagram, Facebook 계정 연동을 지원합니다.
Attributes:
id: 고유 식별자 (자동 증가)
user_id: 사용자 외래키 (User.id 참조)
platform: 플랫폼 구분 (youtube, instagram, facebook)
access_token: OAuth 액세스 토큰
refresh_token: OAuth 리프레시 토큰 (선택)
token_expires_at: 토큰 만료 일시
scope: 허용된 권한 범위
platform_user_id: 플랫폼 내 사용자 고유 ID
platform_username: 플랫폼 내 사용자명/핸들
platform_data: 플랫폼별 추가 정보 (JSON)
is_active: 연동 활성화 상태
connected_at: 연동 일시
updated_at: 정보 수정 일시
플랫폼별 platform_data 예시:
- YouTube: {"channel_id": "UC...", "channel_title": "채널명"}
- Instagram: {"business_account_id": "...", "facebook_page_id": "..."}
- Facebook: {"page_id": "...", "page_access_token": "..."}
Relationships:
user: 연동된 사용자 (User 테이블 참조)
"""
__tablename__ = "social_account"
__table_args__ = (
Index("idx_social_account_user_id", "user_id"),
Index("idx_social_account_platform", "platform"),
Index("idx_social_account_is_active", "is_active"),
Index(
"uq_user_platform_account",
"user_id",
"platform",
"platform_user_id",
unique=True,
),
{
"mysql_engine": "InnoDB",
"mysql_charset": "utf8mb4",
"mysql_collate": "utf8mb4_unicode_ci",
},
)
# ==========================================================================
# 기본 식별자
# ==========================================================================
id: Mapped[int] = mapped_column(
Integer,
primary_key=True,
nullable=False,
autoincrement=True,
comment="고유 식별자",
)
user_id: Mapped[int] = mapped_column(
BigInteger,
ForeignKey("user.id", ondelete="CASCADE"),
nullable=False,
comment="사용자 외래키 (User.id 참조)",
)
# ==========================================================================
# 플랫폼 구분
# ==========================================================================
platform: Mapped[str] = mapped_column(
String(20),
nullable=False,
comment="플랫폼 구분 (youtube, instagram, facebook)",
)
# ==========================================================================
# OAuth 토큰 정보
# ==========================================================================
access_token: Mapped[str] = mapped_column(
Text,
nullable=False,
comment="OAuth 액세스 토큰",
)
refresh_token: Mapped[Optional[str]] = mapped_column(
Text,
nullable=True,
comment="OAuth 리프레시 토큰 (플랫폼에 따라 선택적)",
)
token_expires_at: Mapped[Optional[datetime]] = mapped_column(
DateTime,
nullable=True,
comment="토큰 만료 일시",
)
scope: Mapped[Optional[str]] = mapped_column(
Text,
nullable=True,
comment="허용된 권한 범위 (OAuth scope)",
)
# ==========================================================================
# 플랫폼 계정 식별 정보
# ==========================================================================
platform_user_id: Mapped[str] = mapped_column(
String(100),
nullable=False,
comment="플랫폼 내 사용자 고유 ID",
)
platform_username: Mapped[Optional[str]] = mapped_column(
String(100),
nullable=True,
comment="플랫폼 내 사용자명/핸들 (@username)",
)
# ==========================================================================
# 플랫폼별 추가 정보 (JSON)
# ==========================================================================
platform_data: Mapped[Optional[dict]] = mapped_column(
JSON,
nullable=True,
comment="플랫폼별 추가 정보 (채널ID, 페이지ID 등)",
)
# ==========================================================================
# 연동 상태
# ==========================================================================
is_active: Mapped[bool] = mapped_column(
Boolean,
nullable=False,
default=True,
comment="연동 활성화 상태 (비활성화 시 사용 중지)",
)
# ==========================================================================
# 시간 정보
# ==========================================================================
connected_at: Mapped[datetime] = mapped_column(
DateTime,
nullable=False,
server_default=func.now(),
comment="연동 일시",
)
updated_at: Mapped[datetime] = mapped_column(
DateTime,
nullable=False,
server_default=func.now(),
onupdate=func.now(),
comment="정보 수정 일시",
)
# ==========================================================================
# User 관계
# ==========================================================================
user: Mapped["User"] = relationship(
"User",
back_populates="social_accounts",
)
def __repr__(self) -> str:
return (
f"<SocialAccount("
f"id={self.id}, "
f"user_id={self.user_id}, "
f"platform='{self.platform}', "
f"platform_username='{self.platform_username}', "
f"is_active={self.is_active}"
f")>"
)

4
app/user/schemas/__init__.py
View File

@ -1,6 +1,6 @@
from app.user.schemas.user_schema import (
AccessTokenResponse,
KakaoCallbackRequest,
KakaoCodeRequest,
KakaoLoginResponse,
KakaoTokenResponse,
KakaoUserInfo,
@ -13,7 +13,7 @@ from app.user.schemas.user_schema import (
__all__ = [
"AccessTokenResponse",
"KakaoCallbackRequest",
"KakaoCodeRequest",
"KakaoLoginResponse",
"KakaoTokenResponse",
"KakaoUserInfo",

8
app/user/schemas/user_schema.py
View File

@ -27,8 +27,8 @@ class KakaoLoginResponse(BaseModel):
}
class KakaoCallbackRequest(BaseModel):
"""카카오 콜백 요청 (인가 코드)"""
class KakaoCodeRequest(BaseModel):
"""카카오 인가 코드 검증 요청 (프론트엔드에서 전달)"""
code: str = Field(..., min_length=1, description="카카오 인가 코드")
@ -163,6 +163,7 @@ class LoginResponse(BaseModel):
token_type: str = Field(default="Bearer", description="토큰 타입")
expires_in: int = Field(..., description="액세스 토큰 만료 시간 (초)")
user: UserBriefResponse = Field(..., description="사용자 정보")
redirect_url: str = Field(..., description="로그인 후 리다이렉트할 프론트엔드 URL")
model_config = {
"json_schema_extra": {
@ -177,7 +178,8 @@ class LoginResponse(BaseModel):
"email": "user@kakao.com",
"profile_image_url": "https://k.kakaocdn.net/dn/.../profile.jpg",
"is_new_user": False
}
},
"redirect_url": "http://localhost:3000"
}
}
}

30
app/user/services/auth.py
View File

@ -4,12 +4,17 @@
카카오 로그인, JWT 토큰 관리, 사용자 인증 관련 비즈니스 로직을 처리합니다.
"""
import logging
from datetime import datetime, timezone
from typing import Optional
from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import AsyncSession
from config import prj_settings
logger = logging.getLogger(__name__)
from app.user.exceptions import (
InvalidTokenError,
TokenExpiredError,
@ -66,24 +71,36 @@ class AuthService:
Returns:
LoginResponse: 토큰 및 사용자 정보
"""
logger.info(f"[AUTH] 카카오 로그인 시작 - code: {code[:20]}..., ip: {ip_address}")
# 1. 카카오 토큰 획득
logger.info("[AUTH] 1단계: 카카오 토큰 획득 시작")
kakao_token = await kakao_client.get_access_token(code)
logger.debug(f"[AUTH] 카카오 토큰 획득 완료 - token_type: {kakao_token.token_type}")
# 2. 카카오 사용자 정보 조회
logger.info("[AUTH] 2단계: 카카오 사용자 정보 조회 시작")
kakao_user_info = await kakao_client.get_user_info(kakao_token.access_token)
logger.debug(f"[AUTH] 카카오 사용자 정보 조회 완료 - kakao_id: {kakao_user_info.id}")
# 3. 사용자 조회 또는 생성
logger.info("[AUTH] 3단계: 사용자 조회/생성 시작")
user, is_new_user = await self._get_or_create_user(kakao_user_info, session)
logger.info(f"[AUTH] 사용자 처리 완료 - user_id: {user.id}, is_new_user: {is_new_user}")
# 4. 비활성화 계정 체크
if not user.is_active:
logger.error(f"[AUTH] 비활성화 계정 접근 시도 - user_id: {user.id}")
raise UserInactiveError()
# 5. JWT 토큰 생성
logger.info("[AUTH] 5단계: JWT 토큰 생성 시작")
access_token = create_access_token(user.id)
refresh_token = create_refresh_token(user.id)
logger.debug(f"[AUTH] JWT 토큰 생성 완료 - user_id: {user.id}")
# 6. 리프레시 토큰 DB 저장
logger.info("[AUTH] 6단계: 리프레시 토큰 저장 시작")
await self._save_refresh_token(
user_id=user.id,
token=refresh_token,
@ -91,11 +108,16 @@ class AuthService:
user_agent=user_agent,
ip_address=ip_address,
)
logger.debug(f"[AUTH] 리프레시 토큰 저장 완료 - user_id: {user.id}")
# 7. 마지막 로그인 시간 업데이트
user.last_login_at = datetime.now(timezone.utc)
await session.commit()
redirect_url = f"https://{prj_settings.PROJECT_DOMAIN}"
logger.info(f"[AUTH] 카카오 로그인 완료 - user_id: {user.id}, redirect_url: {redirect_url}")
logger.debug(f"[AUTH] 응답 토큰 정보 - access_token: {access_token[:30]}..., refresh_token: {refresh_token[:30]}...")
return LoginResponse(
access_token=access_token,
refresh_token=refresh_token,
@ -108,6 +130,7 @@ class AuthService:
profile_image_url=user.profile_image_url,
is_new_user=is_new_user,
),
redirect_url=redirect_url,
)
async def refresh_tokens(
@ -221,14 +244,19 @@ class AuthService:
kakao_account = kakao_user_info.kakao_account
profile = kakao_account.profile if kakao_account else None
logger.info(f"[AUTH] 사용자 조회 시작 - kakao_id: {kakao_id}")
# 기존 사용자 조회
result = await session.execute(
select(User).where(User.kakao_id == kakao_id)
)
user = result.scalar_one_or_none()
logger.info(f"[AUTH] DB 조회 결과 - user_exists: {user is not None}, user_id: {user.id if user else None}")
if user is not None:
# 기존 사용자: 프로필 정보 업데이트
logger.info(f"[AUTH] 기존 사용자 발견 - user_id: {user.id}, is_new_user: False")
if profile:
user.nickname = profile.nickname
user.profile_image_url = profile.profile_image_url
@ -239,6 +267,7 @@ class AuthService:
return user, False
# 신규 사용자 생성
logger.info(f"[AUTH] 신규 사용자 생성 시작 - kakao_id: {kakao_id}")
new_user = User(
kakao_id=kakao_id,
email=kakao_account.email if kakao_account else None,
@ -249,6 +278,7 @@ class AuthService:
session.add(new_user)
await session.flush()
await session.refresh(new_user)
logger.info(f"[AUTH] 신규 사용자 생성 완료 - user_id: {new_user.id}, is_new_user: True")
return new_user, True
async def _save_refresh_token(

27
app/user/services/kakao.py
View File

@ -4,10 +4,14 @@
카카오 로그인 인증 흐름을 처리하는 클라이언트입니다.
"""
import logging
import aiohttp
from config import kakao_settings
logger = logging.getLogger(__name__)
from app.user.exceptions import KakaoAPIError, KakaoAuthFailedError
from app.user.schemas.user_schema import KakaoTokenResponse, KakaoUserInfo
@ -39,12 +43,15 @@ class KakaoOAuthClient:
Returns:
카카오 OAuth 인증 페이지 URL
"""
return (
auth_url = (
f"{self.AUTH_URL}"
f"?client_id={self.client_id}"
f"&redirect_uri={self.redirect_uri}"
f"&response_type=code"
)
logger.info(f"[KAKAO] 인증 URL 생성 - redirect_uri: {self.redirect_uri}")
logger.debug(f"[KAKAO] 인증 URL 상세 - auth_url: {auth_url}")
return auth_url
async def get_access_token(self, code: str) -> KakaoTokenResponse:
"""
@ -60,6 +67,7 @@ class KakaoOAuthClient:
KakaoAuthFailedError: 토큰 발급 실패 시
KakaoAPIError: API 호출 오류 시
"""
logger.info(f"[KAKAO] 액세스 토큰 요청 시작 - code: {code[:20]}...")
try:
async with aiohttp.ClientSession() as session:
data = {
@ -72,20 +80,27 @@ class KakaoOAuthClient:
if self.client_secret:
data["client_secret"] = self.client_secret
logger.debug(f"[KAKAO] 토큰 요청 데이터 - redirect_uri: {self.redirect_uri}, client_id: {self.client_id[:10]}...")
async with session.post(self.TOKEN_URL, data=data) as response:
result = await response.json()
logger.debug(f"[KAKAO] 토큰 응답 상태 - status: {response.status}")
if "error" in result:
error_desc = result.get(
"error_description", result.get("error", "알 수 없는 오류")
)
logger.error(f"[KAKAO] 토큰 발급 실패 - error: {result.get('error')}, description: {error_desc}")
raise KakaoAuthFailedError(f"카카오 토큰 발급 실패: {error_desc}")
logger.info("[KAKAO] 액세스 토큰 발급 성공")
logger.debug(f"[KAKAO] 토큰 정보 - token_type: {result.get('token_type')}, expires_in: {result.get('expires_in')}")
return KakaoTokenResponse(**result)
except KakaoAuthFailedError:
raise
except Exception as e:
logger.error(f"[KAKAO] API 호출 오류 - error: {str(e)}")
raise KakaoAPIError(f"카카오 API 호출 중 오류 발생: {str(e)}")
async def get_user_info(self, access_token: str) -> KakaoUserInfo:
@ -102,21 +117,31 @@ class KakaoOAuthClient:
KakaoAuthFailedError: 사용자 정보 조회 실패 시
KakaoAPIError: API 호출 오류 시
"""
logger.info("[KAKAO] 사용자 정보 조회 시작")
try:
async with aiohttp.ClientSession() as session:
headers = {"Authorization": f"Bearer {access_token}"}
async with session.get(self.USER_INFO_URL, headers=headers) as response:
result = await response.json()
logger.debug(f"[KAKAO] 사용자 정보 응답 상태 - status: {response.status}")
if "id" not in result:
logger.error(f"[KAKAO] 사용자 정보 조회 실패 - response: {result}")
raise KakaoAuthFailedError("카카오 사용자 정보를 가져올 수 없습니다.")
kakao_id = result.get("id")
kakao_account = result.get("kakao_account", {})
profile = kakao_account.get("profile", {})
logger.info(f"[KAKAO] 사용자 정보 조회 성공 - kakao_id: {kakao_id}")
logger.debug(f"[KAKAO] 사용자 상세 정보 - nickname: {profile.get('nickname')}, email: {kakao_account.get('email')}")
return KakaoUserInfo(**result)
except KakaoAuthFailedError:
raise
except Exception as e:
logger.error(f"[KAKAO] API 호출 오류 - error: {str(e)}")
raise KakaoAPIError(f"카카오 API 호출 중 오류 발생: {str(e)}")

2
app/utils/chatgpt_prompt.py
View File

@ -39,7 +39,7 @@ class ChatgptService:
) -> str:
prompt_text = prompt.build_prompt(input_data)
print(f"[ChatgptService] Generated Prompt (length: {len(prompt_text)})")
logger.debug(f"[ChatgptService] Generated Prompt (length: {len(prompt_text)})")
logger.info(f"[ChatgptService] Starting GPT request with structured output with model: {prompt.prompt_model}")
# GPT API 호출

90
app/utils/creatomate.py
View File

@ -58,6 +58,77 @@ CACHE_TTL_SECONDS = 300
# 모듈 레벨 공유 HTTP 클라이언트 (커넥션 풀 재사용)
_shared_client: httpx.AsyncClient | None = None
text_template_v_1 = {
"type": "composition",
"track": 3,
"elements": [
{
"type": "text",
"time": 0,
"y": "87.9086%",
"width": "100%",
"height": "40%",
"x_alignment": "50%",
"y_alignment": "50%",
"font_family": "Noto Sans",
"font_weight": "700",
"font_size": "8 vmin",
"background_color": "rgba(216,216,216,0)",
"background_x_padding": "33%",
"background_y_padding": "7%",
"background_border_radius": "28%",
"fill_color": "#ffffff",
"stroke_color": "rgba(51,51,51,1)",
"stroke_width": "0.6 vmin",
}
]
}
text_template_v_2 = {
"type": "composition",
"track": 3,
"elements": [
{
"type": "text",
"time": 0,
"x": "7.7233%",
"y": "82.9852%",
"width": "84.5534%",
"height": "5.7015%",
"x_anchor": "0%",
"y_anchor": "0%",
"x_alignment": "50%",
"y_alignment": "100%",
"font_family": "Noto Sans",
"font_weight": "700",
"font_size": "6.9999 vmin",
"fill_color": "#ffffff"
}
]
}
text_template_h_1 = {
"type": "composition",
"track": 3,
"elements": [
{
"type": "text",
"time": 0,
"x": "10%",
"y": "80%",
"width": "80%",
"height": "15%",
"x_anchor": "0%",
"y_anchor": "0%",
"x_alignment": "50%",
"font_family": "Noto Sans",
"font_weight": "700",
"font_size": "5.9998 vmin",
"fill_color": "#ffffff",
"stroke_color": "#333333",
"stroke_width": "0.2 vmin"
}
]
}
async def get_shared_client() -> httpx.AsyncClient:
"""공유 HTTP 클라이언트를 반환합니다. 없으면 생성합니다."""
@ -470,3 +541,22 @@ class CreatomateService:
)
return new_template
def lining_lyric(self, text_template: dict, lyric_index: int, lyric_text: str, start_sec: float, end_sec: float) -> dict:
duration = end_sec - start_sec
text_scene = copy.deepcopy(text_template)
text_scene["name"] = f"Caption-{lyric_index}"
text_scene["duration"] = duration
text_scene["time"] = start_sec
text_scene["elements"][0]["name"] = f"lyric-{lyric_index}"
text_scene["elements"][0]["text"] = lyric_text
return text_scene
def get_text_template(self):
match self.orientation:
case "vertical":
return text_template_v_2
case "horizontal":
return text_template_h_1

7
app/utils/prompts/prompts.py
View File

@ -1,6 +1,9 @@
import os, json
from abc import ABCMeta
from config import prompt_settings
from app.utils.logger import get_logger
logger = get_logger("prompt")
class Prompt():
prompt_name : str # ex) marketing_prompt
@ -37,8 +40,8 @@ class Prompt():
def build_prompt(self, input_data:dict) -> str:
self.check_input(input_data)
build_template = self.prompt_template
print("build_template", build_template)
print("input_data", input_data)
logger.debug(f"build_template: {build_template}")
logger.debug(f"input_data: {input_data}")
build_template = build_template.format(**input_data)
return build_template

117
app/utils/suno.py
View File

@ -180,6 +180,37 @@ class SunoService:
return data
async def get_lyric_timestamp(self, task_id: str, audio_id: str) -> dict[str, Any]:
"""
음악 타임스탬프 정보 추출
Args:
task_id: generate()에서 반환된 작업 ID
audio_id: 사용할 audio id
Returns:
data.alignedWords: 수노 가사 input - startS endS 시간 데이터 매핑
"""
payload = {
"task_id" : task_id,
"audio_id" : audio_id
}
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.BASE_URL}/generate/get-timestamped-lyrics",
headers=self.headers,
json=payload,
timeout=30.0,
)
response.raise_for_status()
data = response.json()
if not data or not data['data']:
raise ValueError("Suno API returned empty response for task status")
return data['data']['alignedWords']
def parse_status_response(self, result: dict | None) -> PollingSongResponse:
"""Suno API 상태 응답을 파싱하여 PollingSongResponse로 변환합니다.
@ -199,8 +230,6 @@ class SunoService:
success=False,
status="error",
message="Suno API 응답이 비어있습니다.",
clips=None,
raw_response=None,
error_message="Suno API returned None response",
)
@ -212,8 +241,6 @@ class SunoService:
success=False,
status="failed",
message="Suno API 응답 오류",
clips=None,
raw_response=result,
error_message=result.get("msg", "Unknown error"),
)
@ -255,7 +282,85 @@ class SunoService:
success=True,
status=status,
message=status_messages.get(status, f"상태: {status}"),
clips=clips,
raw_response=result,
error_message=None,
)
def align_lyrics(self, word_data: list[dict], sentences: list[str]) -> list[dict]:
"""
word의 시작/끝 포지션만 저장하고, 시간은 word에서 참조
"""
# Step 1: 전체 텍스트 + word별 포지션 범위 저장
full_text = ""
word_ranges = [] # [(start_pos, end_pos, entry), ...]
for entry in word_data:
word = entry['word']
start_pos = len(full_text)
full_text += word
end_pos = len(full_text) - 1
word_ranges.append((start_pos, end_pos, entry))
# Step 2: 메타데이터 제거 + 포지션 재매핑
meta_ranges = []
i = 0
while i < len(full_text):
if full_text[i] == '[':
start = i
while i < len(full_text) and full_text[i] != ']':
i += 1
meta_ranges.append((start, i + 1))
i += 1
clean_text = ""
new_to_old = {} # 클린 포지션 -> 원본 포지션
for old_pos, char in enumerate(full_text):
in_meta = any(s <= old_pos < e for s, e in meta_ranges)
if not in_meta:
new_to_old[len(clean_text)] = old_pos
clean_text += char
# Step 3: 포지션으로 word 찾기
def get_word_at(old_pos: int):
for start, end, entry in word_ranges:
if start <= old_pos <= end:
return entry
return None
# Step 4: 문장 매칭
def normalize(text):
return ''.join(c for c in text if c not in ' \n\t-')
norm_clean = normalize(clean_text)
norm_to_clean = [i for i, c in enumerate(clean_text) if c not in ' \n\t-']
results = []
search_pos = 0
for sentence in sentences:
norm_sentence = normalize(sentence)
found_pos = norm_clean.find(norm_sentence, search_pos)
if found_pos != -1:
clean_start = norm_to_clean[found_pos]
clean_end = norm_to_clean[found_pos + len(norm_sentence) - 1]
old_start = new_to_old[clean_start]
old_end = new_to_old[clean_end]
word_start = get_word_at(old_start)
word_end = get_word_at(old_end)
results.append({
'text': sentence,
'start_sec': word_start['startS'],
'end_sec': word_end['endS'],
})
search_pos = found_pos + len(norm_sentence)
else:
results.append({'text': sentence, 'start_sec': None, 'end_sec': None})
return results

15
app/video/api/routers/v1/video.py
View File

@ -27,7 +27,7 @@ from app.dependencies.pagination import (
)
from app.home.models import Image, Project
from app.lyric.models import Lyric
from app.song.models import Song
from app.song.models import Song, SongTimestamp
from app.video.models import Video
from app.video.schemas.video_schema import (
DownloadVideoResponse,
@ -304,6 +304,19 @@ async def generate_video(
)
logger.debug(f"[generate_video] Duration extended to {creatomate_service.target_duration}s - task_id: {task_id}")
# 이런거 추가해야하는데 AI가 자꾸 번호 달면 제가 번호를 다 밀어야 하나요?
song_timestamp_result = await session.execute(
select(SongTimestamp)
.where(SongTimestamp.suno_audio_id == song.suno_audio_id)
)
song_timestamp_list = song_timestamp_result.scalars().all()
text_template = creatomate_service.get_text_template()
for idx, aligned in enumerate(song_timestamp_list):
caption = creatomate_service.lining_lyric(text_template, idx, aligned.lyric_line, aligned.start_time, aligned.end_time )
final_template['source']['elements'].append(caption)
# 6-5. 커스텀 렌더링 요청 (비동기)
render_response = await creatomate_service.make_creatomate_custom_call_async(
final_template["source"],

2
docs/analysis/pool_problem.md
View File

@ -716,7 +716,7 @@ from app.services.image_processor import ImageProcessorService
from app.config import settings
router = APIRouter(prefix="/images", tags=["images"])
router = APIRouter(prefix="/images", tags=["Images"])
# 외부 서비스 인스턴스
processor_service = ImageProcessorService(settings.IMAGE_PROCESSOR_URL)

8
main.py
View File

@ -43,12 +43,12 @@ tags_metadata = [
"description": "홈 화면 및 프로젝트 관리 API",
},
{
"name": "crawling",
"name": "Crawling",
"description": "네이버 지도 크롤링 API - 장소 정보 및 이미지 수집",
},
{
"name": "image",
"description": "이미지 업로드 API - 로컬 서버 또는 Azure Blob Storage",
"name": "Image-Blob",
"description": "이미지 업로드 API - Azure Blob Storage",
},
{
"name": "Lyric",
@ -68,7 +68,7 @@ tags_metadata = [
## 노래 생성 흐름
1. `POST /api/v1/song/generate/{task_id}` - 노래 생성 요청
2. `GET /api/v1/song/status/{suno_task_id}` - Suno API 상태 확인
2. `GET /api/v1/song/status/{song_id}` - Suno API 상태 확인
3. `GET /api/v1/song/download/{task_id}` - 노래 다운로드 URL 조회
""",
},

1268
poc/timestamp_lyric/Untitled.ipynb
View File

File diff suppressed because it is too large Load Diff

42
poc/timestamp_lyric/lyric_timestamp_mapper.py
View File

@ -1,42 +0,0 @@
from openai import OpenAI
from difflib import SequenceMatcher
from dataclasses import dataclass
from typing import List, Tuple
import aiohttp, json
@dataclass
class TimestampedLyric:
text: str
start: float
end: float
SUNO_BASE_URL="https://api.sunoapi.org"
SUNO_TIMESTAMP_ROUTE = "/api/v1/generate/get-timestamped-lyrics"
SUNO_DETAIL_ROUTE = "/api/v1/generate/record-info"
class LyricTimestampMapper:
suno_api_key : str
def __init__(self, suno_api_key):
self.suno_api_key = suno_api_key
async def get_suno_audio_id_from_task_id(self, suno_task_id): # expire if db save audio id
url = f"{SUNO_BASE_URL}{SUNO_DETAIL_ROUTE}"
headers = {"Authorization": f"Bearer {self.suno_api_key}"}
async with aiohttp.ClientSession() as session:
async with session.get(url, headers=headers, params={"taskId" : suno_task_id}) as response:
detail = await response.json()
result = detail['data']['response']['sunoData'][0]['id']
return result
async def get_suno_timestamp(self, suno_task_id, suno_audio_id): # expire if db save audio id
url = f"{SUNO_BASE_URL}{SUNO_TIMESTAMP_ROUTE}"
headers = {"Authorization": f"Bearer {self.suno_api_key}"}
payload = {
"task_id" : suno_task_id,
"audio_id" : suno_audio_id
}
async with aiohttp.ClientSession() as session:
async with session.post(url, headers=headers, data=json.dumps(payload)) as response:
result = await response.json()
return result

19
poc/timestamp_lyric/test_mapper.py
View File

@ -1,19 +0,0 @@
from lyric_timestamp_mapper import LyricTimestampMapper
API_KEY = "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"
AUDIO_PATH = "test_audio.mp3"
GROUND_TRUTH_LYRICS = [
"첫 번째 가사 라인입니다",
"두 번째 가사 라인입니다",
"세 번째 가사 라인입니다",
]
mapper = LyricTimestampMapper(api_key=API_KEY)
result = mapper.map_ground_truth(AUDIO_PATH, GROUND_TRUTH_LYRICS)
for lyric in result:
if lyric.start >= 0:
print(f"[{lyric.start:.2f} - {lyric.end:.2f}] {lyric.text}")
else:
print(f"[매칭 실패] {lyric.text}")

3122
poc/timestamp_lyric/timestamp.ipynb Normal file
View File

File diff suppressed because it is too large Load Diff