463 lines
14 KiB
Python
463 lines
14 KiB
Python
"""
|
|
Lyric API Router
|
|
|
|
이 모듈은 가사 관련 API 엔드포인트를 정의합니다.
|
|
모든 엔드포인트는 재사용 가능하도록 설계되었습니다.
|
|
|
|
엔드포인트 목록:
|
|
- GET /lyric/status/{task_id}: 가사 생성 상태 조회
|
|
- GET /lyric/{task_id}: 가사 상세 조회
|
|
- GET /lyrics: 가사 목록 조회 (페이지네이션)
|
|
|
|
사용 예시:
|
|
from app.lyric.api.routers.v1.lyric import router
|
|
app.include_router(router, prefix="/api/v1")
|
|
|
|
다른 서비스에서 재사용:
|
|
# 이 파일의 헬퍼 함수들을 import하여 사용 가능
|
|
from app.lyric.api.routers.v1.lyric import (
|
|
get_lyric_status_by_task_id,
|
|
get_lyric_by_task_id,
|
|
get_lyrics_paginated,
|
|
)
|
|
"""
|
|
|
|
import math
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from uuid_extensions import uuid7
|
|
|
|
from app.database.session import get_session
|
|
from app.home.models import Project
|
|
from app.lyric.models import Lyric
|
|
from app.lyric.schemas.lyric import (
|
|
GenerateLyricRequest,
|
|
GenerateLyricResponse,
|
|
LyricDetailResponse,
|
|
LyricListItem,
|
|
LyricStatusResponse,
|
|
PaginatedResponse,
|
|
)
|
|
from app.utils.chatgpt_prompt import ChatgptService
|
|
|
|
router = APIRouter(prefix="/lyric", tags=["lyric"])
|
|
|
|
|
|
# =============================================================================
|
|
# Reusable Service Functions (다른 모듈에서 import하여 사용 가능)
|
|
# =============================================================================
|
|
|
|
|
|
async def get_lyric_status_by_task_id(
|
|
session: AsyncSession, task_id: str
|
|
) -> LyricStatusResponse:
|
|
"""task_id로 가사 생성 작업의 상태를 조회합니다.
|
|
|
|
Args:
|
|
session: SQLAlchemy AsyncSession
|
|
task_id: 작업 고유 식별자
|
|
|
|
Returns:
|
|
LyricStatusResponse: 상태 정보
|
|
|
|
Raises:
|
|
HTTPException: 404 - task_id에 해당하는 가사가 없는 경우
|
|
|
|
Usage:
|
|
# 다른 서비스에서 사용
|
|
from app.lyric.api.routers.v1.lyric import get_lyric_status_by_task_id
|
|
|
|
status_info = await get_lyric_status_by_task_id(session, "some-task-id")
|
|
if status_info.status == "completed":
|
|
# 완료 처리
|
|
"""
|
|
result = await session.execute(select(Lyric).where(Lyric.task_id == task_id))
|
|
lyric = result.scalar_one_or_none()
|
|
|
|
if not lyric:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"task_id '{task_id}'에 해당하는 가사를 찾을 수 없습니다.",
|
|
)
|
|
|
|
status_messages = {
|
|
"processing": "가사 생성 중입니다.",
|
|
"completed": "가사 생성이 완료되었습니다.",
|
|
"failed": "가사 생성에 실패했습니다.",
|
|
}
|
|
|
|
return LyricStatusResponse(
|
|
task_id=lyric.task_id,
|
|
status=lyric.status,
|
|
message=status_messages.get(lyric.status, "알 수 없는 상태입니다."),
|
|
)
|
|
|
|
|
|
async def get_lyric_by_task_id(
|
|
session: AsyncSession, task_id: str
|
|
) -> LyricDetailResponse:
|
|
"""task_id로 생성된 가사 상세 정보를 조회합니다.
|
|
|
|
Args:
|
|
session: SQLAlchemy AsyncSession
|
|
task_id: 작업 고유 식별자
|
|
|
|
Returns:
|
|
LyricDetailResponse: 가사 상세 정보
|
|
|
|
Raises:
|
|
HTTPException: 404 - task_id에 해당하는 가사가 없는 경우
|
|
|
|
Usage:
|
|
# 다른 서비스에서 사용
|
|
from app.lyric.api.routers.v1.lyric import get_lyric_by_task_id
|
|
|
|
lyric = await get_lyric_by_task_id(session, task_id)
|
|
print(lyric.lyric_result)
|
|
"""
|
|
result = await session.execute(select(Lyric).where(Lyric.task_id == task_id))
|
|
lyric = result.scalar_one_or_none()
|
|
|
|
if not lyric:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"task_id '{task_id}'에 해당하는 가사를 찾을 수 없습니다.",
|
|
)
|
|
|
|
return LyricDetailResponse(
|
|
id=lyric.id,
|
|
task_id=lyric.task_id,
|
|
project_id=lyric.project_id,
|
|
status=lyric.status,
|
|
lyric_prompt=lyric.lyric_prompt,
|
|
lyric_result=lyric.lyric_result,
|
|
created_at=lyric.created_at,
|
|
)
|
|
|
|
|
|
async def get_lyrics_paginated(
|
|
session: AsyncSession,
|
|
page: int = 1,
|
|
page_size: int = 20,
|
|
status_filter: Optional[str] = None,
|
|
) -> PaginatedResponse[LyricListItem]:
|
|
"""페이지네이션으로 가사 목록을 조회합니다.
|
|
|
|
Args:
|
|
session: SQLAlchemy AsyncSession
|
|
page: 페이지 번호 (1부터 시작, 기본값: 1)
|
|
page_size: 페이지당 데이터 수 (기본값: 20, 최대: 100)
|
|
status_filter: 상태 필터 (optional) - "processing", "completed", "failed"
|
|
|
|
Returns:
|
|
PaginatedResponse[LyricListItem]: 페이지네이션된 가사 목록
|
|
|
|
Usage:
|
|
# 다른 서비스에서 사용
|
|
from app.lyric.api.routers.v1.lyric import get_lyrics_paginated
|
|
|
|
# 기본 페이지네이션
|
|
lyrics = await get_lyrics_paginated(session, page=1, page_size=20)
|
|
|
|
# 상태 필터링
|
|
completed_lyrics = await get_lyrics_paginated(
|
|
session, page=1, page_size=10, status_filter="completed"
|
|
)
|
|
"""
|
|
# 페이지 크기 제한
|
|
page_size = min(page_size, 100)
|
|
offset = (page - 1) * page_size
|
|
|
|
# 기본 쿼리
|
|
query = select(Lyric)
|
|
count_query = select(func.count(Lyric.id))
|
|
|
|
# 상태 필터 적용
|
|
if status_filter:
|
|
query = query.where(Lyric.status == status_filter)
|
|
count_query = count_query.where(Lyric.status == status_filter)
|
|
|
|
# 전체 개수 조회
|
|
total_result = await session.execute(count_query)
|
|
total = total_result.scalar() or 0
|
|
|
|
# 데이터 조회 (최신순 정렬)
|
|
query = query.order_by(Lyric.created_at.desc()).offset(offset).limit(page_size)
|
|
result = await session.execute(query)
|
|
lyrics = result.scalars().all()
|
|
|
|
# 페이지네이션 정보 계산
|
|
total_pages = math.ceil(total / page_size) if total > 0 else 1
|
|
|
|
items = [
|
|
LyricListItem(
|
|
id=lyric.id,
|
|
task_id=lyric.task_id,
|
|
status=lyric.status,
|
|
lyric_result=lyric.lyric_result,
|
|
created_at=lyric.created_at,
|
|
)
|
|
for lyric in lyrics
|
|
]
|
|
|
|
return PaginatedResponse[LyricListItem](
|
|
items=items,
|
|
total=total,
|
|
page=page,
|
|
page_size=page_size,
|
|
total_pages=total_pages,
|
|
has_next=page < total_pages,
|
|
has_prev=page > 1,
|
|
)
|
|
|
|
|
|
# =============================================================================
|
|
# API Endpoints
|
|
# =============================================================================
|
|
|
|
|
|
@router.post(
|
|
"/generate",
|
|
summary="가사 생성",
|
|
description="""
|
|
고객 정보를 기반으로 ChatGPT를 이용하여 가사를 생성합니다.
|
|
|
|
## 요청 필드
|
|
- **customer_name**: 고객명/가게명 (필수)
|
|
- **region**: 지역명 (필수)
|
|
- **detail_region_info**: 상세 지역 정보 (선택)
|
|
- **language**: 가사 출력 언어 (Korean, English, Chinese, Japanese, Thai, Vietnamese)
|
|
|
|
## 반환 정보
|
|
- **success**: 생성 성공 여부
|
|
- **task_id**: 작업 고유 식별자
|
|
- **lyric**: 생성된 가사
|
|
- **language**: 가사 언어
|
|
- **prompt_used**: 사용된 프롬프트
|
|
- **error_message**: 에러 메시지 (실패 시)
|
|
|
|
## 사용 예시
|
|
```
|
|
POST /lyric/generate
|
|
{
|
|
"customer_name": "스테이 머뭄",
|
|
"region": "군산",
|
|
"detail_region_info": "군산 신흥동 말랭이 마을",
|
|
"language": "English"
|
|
}
|
|
```
|
|
""",
|
|
response_model=GenerateLyricResponse,
|
|
responses={
|
|
200: {"description": "가사 생성 성공"},
|
|
500: {"description": "가사 생성 실패"},
|
|
},
|
|
)
|
|
async def generate_lyric(
|
|
request_body: GenerateLyricRequest,
|
|
session: AsyncSession = Depends(get_session),
|
|
) -> GenerateLyricResponse:
|
|
"""고객 정보를 기반으로 가사를 생성합니다."""
|
|
task_id = str(uuid7())
|
|
|
|
try:
|
|
# 1. ChatGPT 서비스 초기화 및 프롬프트 생성
|
|
service = ChatgptService(
|
|
customer_name=request_body.customer_name,
|
|
region=request_body.region,
|
|
detail_region_info=request_body.detail_region_info or "",
|
|
language=request_body.language,
|
|
)
|
|
prompt = service.build_lyrics_prompt()
|
|
|
|
# 2. Project 테이블에 데이터 저장
|
|
project = Project(
|
|
store_name=request_body.customer_name,
|
|
region=request_body.region,
|
|
task_id=task_id,
|
|
detail_region_info=request_body.detail_region_info,
|
|
language=request_body.language,
|
|
)
|
|
session.add(project)
|
|
await session.commit()
|
|
await session.refresh(project) # commit 후 project.id 동기화
|
|
|
|
# 3. Lyric 테이블에 데이터 저장 (status: processing)
|
|
lyric = Lyric(
|
|
project_id=project.id,
|
|
task_id=task_id,
|
|
status="processing",
|
|
lyric_prompt=prompt,
|
|
lyric_result=None,
|
|
language=request_body.language,
|
|
)
|
|
session.add(lyric)
|
|
await session.commit() # processing 상태를 확실히 저장 (다른 트랜잭션에서 조회 가능)
|
|
await session.refresh(lyric) # commit 후 객체 상태 동기화
|
|
|
|
# 4. ChatGPT를 통해 가사 생성
|
|
result = await service.generate(prompt=prompt)
|
|
|
|
# 5. ERROR가 포함되어 있으면 실패 처리
|
|
if "ERROR:" in result:
|
|
lyric.status = "failed"
|
|
lyric.lyric_result = result
|
|
await session.commit()
|
|
|
|
return GenerateLyricResponse(
|
|
success=False,
|
|
task_id=task_id,
|
|
lyric=None,
|
|
language=request_body.language,
|
|
prompt_used=prompt,
|
|
error_message=result,
|
|
)
|
|
|
|
# 6. 성공 시 Lyric 테이블 업데이트 (status: completed)
|
|
lyric.status = "completed"
|
|
lyric.lyric_result = result
|
|
await session.commit()
|
|
|
|
return GenerateLyricResponse(
|
|
success=True,
|
|
task_id=task_id,
|
|
lyric=result,
|
|
language=request_body.language,
|
|
prompt_used=prompt,
|
|
error_message=None,
|
|
)
|
|
except Exception as e:
|
|
await session.rollback()
|
|
return GenerateLyricResponse(
|
|
success=False,
|
|
task_id=task_id,
|
|
lyric=None,
|
|
language=request_body.language,
|
|
prompt_used=None,
|
|
error_message=str(e),
|
|
)
|
|
|
|
|
|
@router.get(
|
|
"/status/{task_id}",
|
|
summary="가사 생성 상태 조회",
|
|
description="""
|
|
task_id로 가사 생성 작업의 현재 상태를 조회합니다.
|
|
|
|
## 상태 값
|
|
- **processing**: 가사 생성 중
|
|
- **completed**: 가사 생성 완료
|
|
- **failed**: 가사 생성 실패
|
|
|
|
## 사용 예시
|
|
```
|
|
GET /lyric/status/019123ab-cdef-7890-abcd-ef1234567890
|
|
```
|
|
""",
|
|
response_model=LyricStatusResponse,
|
|
responses={
|
|
200: {"description": "상태 조회 성공"},
|
|
404: {"description": "해당 task_id를 찾을 수 없음"},
|
|
},
|
|
)
|
|
async def get_lyric_status(
|
|
task_id: str,
|
|
session: AsyncSession = Depends(get_session),
|
|
) -> LyricStatusResponse:
|
|
"""task_id로 가사 생성 작업 상태를 조회합니다."""
|
|
return await get_lyric_status_by_task_id(session, task_id)
|
|
|
|
|
|
@router.get(
|
|
"s",
|
|
summary="가사 목록 조회 (페이지네이션)",
|
|
description="""
|
|
생성된 모든 가사를 페이지네이션으로 조회합니다.
|
|
|
|
## 파라미터
|
|
- **page**: 페이지 번호 (1부터 시작, 기본값: 1)
|
|
- **page_size**: 페이지당 데이터 수 (기본값: 20, 최대: 100)
|
|
- **status**: 상태 필터 (선택사항) - "processing", "completed", "failed"
|
|
|
|
## 반환 정보
|
|
- **items**: 가사 목록
|
|
- **total**: 전체 데이터 수
|
|
- **page**: 현재 페이지
|
|
- **page_size**: 페이지당 데이터 수
|
|
- **total_pages**: 전체 페이지 수
|
|
- **has_next**: 다음 페이지 존재 여부
|
|
- **has_prev**: 이전 페이지 존재 여부
|
|
|
|
## 사용 예시
|
|
```
|
|
GET /lyrics # 기본 조회 (1페이지, 20개)
|
|
GET /lyrics?page=2 # 2페이지 조회
|
|
GET /lyrics?page=1&page_size=50 # 50개씩 조회
|
|
GET /lyrics?status=completed # 완료된 가사만 조회
|
|
```
|
|
|
|
## 다른 모델에서 PaginatedResponse 재사용
|
|
```python
|
|
from app.lyric.api.schemas.lyric import PaginatedResponse
|
|
|
|
# Song 목록에서 사용
|
|
@router.get("/songs", response_model=PaginatedResponse[SongListItem])
|
|
async def list_songs(...):
|
|
...
|
|
```
|
|
""",
|
|
response_model=PaginatedResponse[LyricListItem],
|
|
responses={
|
|
200: {"description": "가사 목록 조회 성공"},
|
|
},
|
|
)
|
|
async def list_lyrics(
|
|
page: int = Query(1, ge=1, description="페이지 번호 (1부터 시작)"),
|
|
page_size: int = Query(20, ge=1, le=100, description="페이지당 데이터 수"),
|
|
status: Optional[str] = Query(
|
|
None,
|
|
description="상태 필터 (processing, completed, failed)",
|
|
pattern="^(processing|completed|failed)$",
|
|
),
|
|
session: AsyncSession = Depends(get_session),
|
|
) -> PaginatedResponse[LyricListItem]:
|
|
"""페이지네이션으로 가사 목록을 조회합니다."""
|
|
return await get_lyrics_paginated(session, page, page_size, status)
|
|
|
|
|
|
@router.get(
|
|
"/{task_id}",
|
|
summary="가사 상세 조회",
|
|
description="""
|
|
task_id로 생성된 가사의 상세 정보를 조회합니다.
|
|
|
|
## 반환 정보
|
|
- **id**: 가사 ID
|
|
- **task_id**: 작업 고유 식별자
|
|
- **project_id**: 프로젝트 ID
|
|
- **status**: 처리 상태
|
|
- **lyric_prompt**: 가사 생성에 사용된 프롬프트
|
|
- **lyric_result**: 생성된 가사 (완료 시)
|
|
- **created_at**: 생성 일시
|
|
|
|
## 사용 예시
|
|
```
|
|
GET /lyric/019123ab-cdef-7890-abcd-ef1234567890
|
|
```
|
|
""",
|
|
response_model=LyricDetailResponse,
|
|
responses={
|
|
200: {"description": "가사 조회 성공"},
|
|
404: {"description": "해당 task_id를 찾을 수 없음"},
|
|
},
|
|
)
|
|
async def get_lyric_detail(
|
|
task_id: str,
|
|
session: AsyncSession = Depends(get_session),
|
|
) -> LyricDetailResponse:
|
|
"""task_id로 생성된 가사를 조회합니다."""
|
|
return await get_lyric_by_task_id(session, task_id)
|