castad
/
o2o-castad-backend
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
o2o-castad-backend/app/lyric/api/routers/v1/lyric.py

417 lines
12 KiB
Python
Raw Blame History

"""
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 app.database.session import get_session
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**: 생성 성공 여부
- **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,
) -> GenerateLyricResponse:
"""고객 정보를 기반으로 가사를 생성합니다."""
try:
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()
result = await service.generate(prompt=prompt)
# ERROR가 포함되어 있으면 실패 처리
if "ERROR:" in result:
return GenerateLyricResponse(
success=False,
lyric=None,
language=request_body.language,
prompt_used=prompt,
error_message=result,
)
return GenerateLyricResponse(
success=True,
lyric=result,
language=request_body.language,
prompt_used=prompt,
error_message=None,
)
except Exception as e:
return GenerateLyricResponse(
success=False,
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)
Reference in New Issue View Git Blame Copy Permalink