From 79ec5daa0dcdb4281febda32d2bb6b33d4a03c11 Mon Sep 17 00:00:00 2001 From: bluebamus Date: Mon, 22 Dec 2025 00:49:55 +0900 Subject: [PATCH] finished --- app/database/session.py | 55 +++++ app/home/api/routers/v1/home.py | 109 ++-------- app/home/worker/main_task.py | 87 ++++++++ app/lyric/api/routers/v1/lyric.py | 335 +++++++++++++++++++++++++++++ app/lyric/api/routers/v1/router.py | 146 ------------- app/lyric/api/schemas/.gitkeep | 0 app/lyric/schemas/lyric.py | 165 ++++++++++++++ app/song/api/schemas/.gitkeep | 0 app/utils/chatgpt_prompt.py | 180 +++++++++------- app/video/api/schemas/.gitkeep | 0 main.py | 4 +- 11 files changed, 768 insertions(+), 313 deletions(-) create mode 100644 app/home/worker/main_task.py create mode 100644 app/lyric/api/routers/v1/lyric.py delete mode 100644 app/lyric/api/routers/v1/router.py delete mode 100644 app/lyric/api/schemas/.gitkeep create mode 100644 app/lyric/schemas/lyric.py delete mode 100644 app/song/api/schemas/.gitkeep delete mode 100644 app/video/api/schemas/.gitkeep diff --git a/app/database/session.py b/app/database/session.py index b081725..598c2b3 100644 --- a/app/database/session.py +++ b/app/database/session.py @@ -1,7 +1,9 @@ +from contextlib import asynccontextmanager from typing import AsyncGenerator from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from sqlalchemy.orm import DeclarativeBase +from sqlalchemy.pool import NullPool from config import db_settings @@ -70,3 +72,56 @@ async def get_session() -> AsyncGenerator[AsyncSession, None]: async def dispose_engine() -> None: await engine.dispose() print("Database engine disposed") + + +# ============================================================================= +# 백그라운드 태스크용 세션 (별도 이벤트 루프에서 사용) +# ============================================================================= + + +@asynccontextmanager +async def get_worker_session() -> AsyncGenerator[AsyncSession, None]: + """백그라운드 태스크용 세션 컨텍스트 매니저 + + asyncio.run()으로 새 이벤트 루프를 생성하는 백그라운드 태스크에서 사용합니다. + NullPool을 사용하여 연결 풀링을 비활성화하고, 이벤트 루프 충돌을 방지합니다. + + get_session()과의 차이점: + - get_session(): FastAPI DI용, 메인 이벤트 루프의 연결 풀 사용 + - get_worker_session(): 백그라운드 태스크용, NullPool로 매번 새 연결 생성 + + Usage: + async with get_worker_session() as session: + result = await session.execute(select(Model)) + await session.commit() + + Note: + - 매 호출마다 엔진을 생성하고 dispose하므로 오버헤드가 있음 + - 빈번한 호출이 필요한 경우 방법 1(모듈 레벨 엔진)을 고려 + """ + worker_engine = create_async_engine( + url=db_settings.MYSQL_URL, + poolclass=NullPool, + connect_args={ + "connect_timeout": 3, + "charset": "utf8mb4", + }, + ) + session_factory = async_sessionmaker( + bind=worker_engine, + class_=AsyncSession, + expire_on_commit=False, + autoflush=False, + ) + + async with session_factory() as session: + try: + yield session + except Exception as e: + await session.rollback() + print(f"Worker session rollback due to: {e}") + raise e + finally: + await session.close() + + await worker_engine.dispose() diff --git a/app/home/api/routers/v1/home.py b/app/home/api/routers/v1/home.py index 736c527..f1e731d 100644 --- a/app/home/api/routers/v1/home.py +++ b/app/home/api/routers/v1/home.py @@ -3,7 +3,7 @@ from datetime import date from pathlib import Path import aiofiles -from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile +from fastapi import APIRouter, BackgroundTasks, Depends, File, Form, HTTPException, UploadFile from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from uuid_extensions import uuid7 @@ -21,108 +21,39 @@ from app.home.schemas.home import ( GenerateUrlsRequest, ProcessedInfo, ) +from app.home.worker.main_task import task_process from app.utils.nvMapScraper import NvMapScraper MEDIA_ROOT = Path("media") -# 전국 시/군/구 이름 목록 (roadAddress에서 region 추출용) +# 전국 시 이름 목록 (roadAddress에서 region 추출용) +# fmt: off KOREAN_CITIES = [ # 특별시/광역시 - "서울시", - "부산시", - "대구시", - "인천시", - "광주시", - "대전시", - "울산시", - "세종시", + "서울시", "부산시", "대구시", "인천시", "광주시", "대전시", "울산시", "세종시", # 경기도 - "수원시", - "성남시", - "고양시", - "용인시", - "부천시", - "안산시", - "안양시", - "남양주시", - "화성시", - "평택시", - "의정부시", - "시흥시", - "파주시", - "광명시", - "김포시", - "군포시", - "광주시", - "이천시", - "양주시", - "오산시", - "구리시", - "안성시", - "포천시", - "의왕시", - "하남시", - "여주시", - "동두천시", - "과천시", + "수원시", "성남시", "고양시", "용인시", "부천시", "안산시", "안양시", "남양주시", + "화성시", "평택시", "의정부시", "시흥시", "파주시", "광명시", "김포시", "군포시", + "광주시", "이천시", "양주시", "오산시", "구리시", "안성시", "포천시", "의왕시", + "하남시", "여주시", "동두천시", "과천시", # 강원도 - "춘천시", - "원주시", - "강릉시", - "동해시", - "태백시", - "속초시", - "삼척시", + "춘천시", "원주시", "강릉시", "동해시", "태백시", "속초시", "삼척시", # 충청북도 - "청주시", - "충주시", - "제천시", + "청주시", "충주시", "제천시", # 충청남도 - "천안시", - "공주시", - "보령시", - "아산시", - "서산시", - "논산시", - "계룡시", - "당진시", + "천안시", "공주시", "보령시", "아산시", "서산시", "논산시", "계룡시", "당진시", # 전라북도 - "전주시", - "군산시", - "익산시", - "정읍시", - "남원시", - "김제시", + "전주시", "군산시", "익산시", "정읍시", "남원시", "김제시", # 전라남도 - "목포시", - "여수시", - "순천시", - "나주시", - "광양시", + "목포시", "여수시", "순천시", "나주시", "광양시", # 경상북도 - "포항시", - "경주시", - "김천시", - "안동시", - "구미시", - "영주시", - "영천시", - "상주시", - "문경시", - "경산시", + "포항시", "경주시", "김천시", "안동시", "구미시", "영주시", "영천시", "상주시", "문경시", "경산시", # 경상남도 - "창원시", - "진주시", - "통영시", - "사천시", - "김해시", - "밀양시", - "거제시", - "양산시", + "창원시", "진주시", "통영시", "사천시", "김해시", "밀양시", "거제시", "양산시", # 제주도 - "제주시", - "서귀포시", + "제주시", "서귀포시", ] +# fmt: on router = APIRouter() @@ -221,6 +152,7 @@ def _extract_image_name(url: str, index: int) -> str: ) async def generate( request_body: GenerateRequest, + background_tasks: BackgroundTasks, session: AsyncSession = Depends(get_session), ): """기본 영상 생성 요청 처리 (이미지 없음)""" @@ -248,6 +180,9 @@ async def generate( ) session.add(project) await session.commit() + await session.refresh(project) + + background_tasks.add_task(task_process, request_body, task_id, project.id) return { "task_id": task_id, diff --git a/app/home/worker/main_task.py b/app/home/worker/main_task.py new file mode 100644 index 0000000..817d102 --- /dev/null +++ b/app/home/worker/main_task.py @@ -0,0 +1,87 @@ +import asyncio + +from sqlalchemy import select + +from app.database.session import get_worker_session +from app.home.schemas.home import GenerateRequest +from app.lyric.models import Lyric +from app.utils.chatgpt_prompt import ChatgptService + + +async def _save_lyric(task_id: str, project_id: int, lyric_prompt: str) -> int: + """Lyric 레코드를 DB에 저장 (status=processing, lyric_result=null)""" + async with get_worker_session() as session: + lyric = Lyric( + task_id=task_id, + project_id=project_id, + status="processing", + lyric_prompt=lyric_prompt, + lyric_result=None, + ) + session.add(lyric) + await session.commit() + await session.refresh(lyric) + print(f"Lyric saved: id={lyric.id}, task_id={task_id}, status=processing") + return lyric.id + + +async def _update_lyric_status(lyric_id: int, status: str, lyric_result: str | None = None) -> None: + """Lyric 레코드의 status와 lyric_result를 업데이트""" + async with get_worker_session() as session: + result = await session.execute(select(Lyric).where(Lyric.id == lyric_id)) + lyric = result.scalar_one_or_none() + if lyric: + lyric.status = status + if lyric_result is not None: + lyric.lyric_result = lyric_result + await session.commit() + print(f"Lyric updated: id={lyric_id}, status={status}") + + +async def lyric_task( + task_id: str, + project_id: int, + customer_name: str, + region: str, + detail_region_info: str, +) -> None: + """가사 생성 작업: ChatGPT로 가사 생성 및 Lyric 테이블 저장/업데이트""" + service = ChatgptService( + customer_name=customer_name, + region=region, + detail_region_info=detail_region_info, + ) + + # Lyric 레코드 저장 (status=processing, lyric_result=null) + lyric_prompt = service.build_lyrics_prompt() + lyric_id = await _save_lyric(task_id, project_id, lyric_prompt) + + # GPT 호출 + result = await service.generate_lyrics(prompt=lyric_prompt) + + print(f"GPT Response:\n{result}") + + # 결과에 ERROR가 포함되어 있으면 status를 failed로 업데이트 + if "ERROR:" in result: + await _update_lyric_status(lyric_id, "failed", lyric_result=result) + else: + await _update_lyric_status(lyric_id, "completed", lyric_result=result) + + +async def _task_process_async(request_body: GenerateRequest, task_id: str, project_id: int) -> None: + """백그라운드 작업 처리 (async 버전)""" + customer_name = request_body.customer_name + region = request_body.region + detail_region_info = request_body.detail_region_info or "" + + print(f"customer_name: {customer_name}") + print(f"region: {region}") + print(f"detail_region_info: {detail_region_info}") + + # 가사 생성 작업 + await lyric_task(task_id, project_id, customer_name, region, detail_region_info) + + +def task_process(request_body: GenerateRequest, task_id: str, project_id: int) -> None: + """백그라운드 작업 처리 함수 (sync wrapper)""" + asyncio.run(_task_process_async(request_body, task_id, project_id)) diff --git a/app/lyric/api/routers/v1/lyric.py b/app/lyric/api/routers/v1/lyric.py new file mode 100644 index 0000000..f1c0d84 --- /dev/null +++ b/app/lyric/api/routers/v1/lyric.py @@ -0,0 +1,335 @@ +""" +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 ( + LyricDetailResponse, + LyricListItem, + LyricStatusResponse, + PaginatedResponse, +) + +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.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) diff --git a/app/lyric/api/routers/v1/router.py b/app/lyric/api/routers/v1/router.py deleted file mode 100644 index f87cd1d..0000000 --- a/app/lyric/api/routers/v1/router.py +++ /dev/null @@ -1,146 +0,0 @@ -from typing import Any - -from fastapi import APIRouter, Depends, Request # , UploadFile, File, status -from sqlalchemy.ext.asyncio import AsyncSession - -from app.database.session import get_session -from app.lyric.services import lyrics - -router = APIRouter(prefix="/lyrics", tags=["lyrics"]) - - -@router.get("/store") -async def home( - request: Request, - conn: AsyncSession = Depends(get_session), -): - # store_info_list: List[StoreData] = await lyrics_svc.get_store_info(conn) - result: Any = await lyrics.get_store_info(conn) - - # return templates.TemplateResponse( - # request=request, - # name="store.html", - # context={"store_info_list": result}, - # ) - pass - - -@router.post("/attributes") -async def attribute( - request: Request, - conn: AsyncSession = Depends(get_session), -): - print("attributes") - print(await request.form()) - - result: Any = await lyrics.get_attribute(conn) - print(result) - - # return templates.TemplateResponse( - # request=request, - # name="attribute.html", - # context={ - # "attribute_group_dict": result, - # "before_dict": await request.form(), - # }, - # ) - pass - - -@router.post("/fewshot") -async def sample_song( - request: Request, - conn: AsyncSession = Depends(get_session), -): - print("fewshot") - print(await request.form()) - - result: Any = await lyrics.get_sample_song(conn) - print(result) - - # return templates.TemplateResponse( - # request=request, - # name="fewshot.html", - # context={"fewshot_list": result, "before_dict": await request.form()}, - # ) - pass - - -@router.post("/prompt") -async def prompt_template( - request: Request, - conn: AsyncSession = Depends(get_session), -): - print("prompt_template") - print(await request.form()) - - result: Any = await lyrics.get_prompt_template(conn) - print(result) - - print("prompt_template after") - print(await request.form()) - - # return templates.TemplateResponse( - # request=request, - # name="prompt.html", - # context={"prompt_list": result, "before_dict": await request.form()}, - # ) - - pass - - -@router.post("/result") -async def song_result( - request: Request, - conn: AsyncSession = Depends(get_session), -): - print("song_result") - print(await request.form()) - - result: Any = await lyrics.make_song_result(request, conn) - print("result : ", result) - - # return templates.TemplateResponse( - # request=request, - # name="result.html", - # context={"result_dict": result}, - # ) - pass - - -@router.get("/result") -async def get_song_result( - request: Request, - conn: AsyncSession = Depends(get_session), -): - print("get_song_result") - print(await request.form()) - - result: Any = await lyrics.get_song_result(conn) - print("result : ", result) - - # return templates.TemplateResponse( - # request=request, - # name="result.html", - # context={"result_dict": result}, - # ) - pass - - -@router.post("/automation") -async def automation( - request: Request, - conn: AsyncSession = Depends(get_session), -): - print("automation") - print(await request.form()) - - result: Any = await lyrics.make_automation(request, conn) - print("result : ", result) - - # return templates.TemplateResponse( - # request=request, - # name="result.html", - # context={"result_dict": result}, - # ) - pass diff --git a/app/lyric/api/schemas/.gitkeep b/app/lyric/api/schemas/.gitkeep deleted file mode 100644 index e69de29..0000000 diff --git a/app/lyric/schemas/lyric.py b/app/lyric/schemas/lyric.py new file mode 100644 index 0000000..fcfe18a --- /dev/null +++ b/app/lyric/schemas/lyric.py @@ -0,0 +1,165 @@ +""" +Lyric API Schemas + +이 모듈은 가사 관련 API 엔드포인트에서 사용되는 Pydantic 스키마를 정의합니다. + +사용 예시: + from app.lyric.schemas.lyric import ( + LyricStatusResponse, + LyricDetailResponse, + LyricListItem, + PaginatedResponse, + ) + + # 라우터에서 response_model로 사용 + @router.get("/lyric/{task_id}", response_model=LyricDetailResponse) + async def get_lyric(task_id: str): + ... + + # 페이지네이션 응답 (다른 모델에서도 재사용 가능) + @router.get("/songs", response_model=PaginatedResponse[SongListItem]) + async def list_songs(...): + ... +""" + +import math +from datetime import datetime +from typing import Generic, List, Optional, TypeVar + +from pydantic import BaseModel, Field + + +class LyricStatusResponse(BaseModel): + """가사 상태 조회 응답 스키마 + + Usage: + GET /lyric/status/{task_id} + Returns the current processing status of a lyric generation task. + + Example Response: + { + "task_id": "019123ab-cdef-7890-abcd-ef1234567890", + "status": "completed", + "message": "가사 생성이 완료되었습니다." + } + """ + + task_id: str = Field(..., description="작업 고유 식별자") + status: str = Field(..., description="처리 상태 (processing, completed, failed)") + message: str = Field(..., description="상태 메시지") + + +class LyricDetailResponse(BaseModel): + """가사 상세 조회 응답 스키마 + + Usage: + GET /lyric/{task_id} + Returns the generated lyric content for a specific task. + + Example Response: + { + "id": 1, + "task_id": "019123ab-cdef-7890-abcd-ef1234567890", + "project_id": 1, + "status": "completed", + "lyric_prompt": "...", + "lyric_result": "생성된 가사...", + "created_at": "2024-01-01T12:00:00" + } + """ + + id: int = Field(..., description="가사 ID") + task_id: str = Field(..., description="작업 고유 식별자") + project_id: int = Field(..., description="프로젝트 ID") + status: str = Field(..., description="처리 상태") + lyric_prompt: str = Field(..., description="가사 생성 프롬프트") + lyric_result: Optional[str] = Field(None, description="생성된 가사") + created_at: Optional[datetime] = Field(None, description="생성 일시") + + +class LyricListItem(BaseModel): + """가사 목록 아이템 스키마 + + Usage: + Used as individual items in paginated lyric list responses. + """ + + id: int = Field(..., description="가사 ID") + task_id: str = Field(..., description="작업 고유 식별자") + status: str = Field(..., description="처리 상태") + lyric_result: Optional[str] = Field(None, description="생성된 가사 (미리보기)") + created_at: Optional[datetime] = Field(None, description="생성 일시") + + +T = TypeVar("T") + + +class PaginatedResponse(BaseModel, Generic[T]): + """페이지네이션 응답 스키마 (재사용 가능) + + Usage: + 다른 모델에서도 페이지네이션이 필요할 때 재사용 가능: + - PaginatedResponse[LyricListItem] + - PaginatedResponse[SongListItem] + - PaginatedResponse[VideoListItem] + + Example: + from app.lyric.schemas.lyric import PaginatedResponse + + @router.get("/items", response_model=PaginatedResponse[ItemModel]) + async def get_items(page: int = 1, page_size: int = 20): + ... + + Example Response: + { + "items": [...], + "total": 100, + "page": 1, + "page_size": 20, + "total_pages": 5, + "has_next": true, + "has_prev": false + } + """ + + items: List[T] = Field(..., description="데이터 목록") + total: int = Field(..., description="전체 데이터 수") + page: int = Field(..., description="현재 페이지 (1부터 시작)") + page_size: int = Field(..., description="페이지당 데이터 수") + total_pages: int = Field(..., description="전체 페이지 수") + has_next: bool = Field(..., description="다음 페이지 존재 여부") + has_prev: bool = Field(..., description="이전 페이지 존재 여부") + + @classmethod + def create( + cls, + items: List[T], + total: int, + page: int, + page_size: int, + ) -> "PaginatedResponse[T]": + """페이지네이션 응답을 생성하는 헬퍼 메서드 + + Args: + items: 현재 페이지의 데이터 목록 + total: 전체 데이터 수 + page: 현재 페이지 번호 + page_size: 페이지당 데이터 수 + + Returns: + PaginatedResponse: 완성된 페이지네이션 응답 + + Usage: + items = [LyricListItem(...) for lyric in lyrics] + return PaginatedResponse.create(items, total=100, page=1, page_size=20) + """ + total_pages = math.ceil(total / page_size) if total > 0 else 1 + return cls( + items=items, + total=total, + page=page, + page_size=page_size, + total_pages=total_pages, + has_next=page < total_pages, + has_prev=page > 1, + ) diff --git a/app/song/api/schemas/.gitkeep b/app/song/api/schemas/.gitkeep deleted file mode 100644 index e69de29..0000000 diff --git a/app/utils/chatgpt_prompt.py b/app/utils/chatgpt_prompt.py index 3618ba5..c2586df 100644 --- a/app/utils/chatgpt_prompt.py +++ b/app/utils/chatgpt_prompt.py @@ -2,92 +2,116 @@ from openai import AsyncOpenAI from config import apikey_settings +# fmt: off +LYRICS_PROMPT_TEMPLATE_ORI = """ +1.Act as a content marketing expert with domain knowledges in [pension/staying services] in Korea, Goal: plan viral content creation that lead online reservations and promotion +2.Conduct an in-depth analysis of [업체명:{customer_name}] in [지역명:{region}] by examining their official website or informations, photos on never map and online presence. Create a comprehensive "[지역 상세: {detail_region_info}]_Brand & Marketing Intelligence Report in Korean, that includes: + +**Core Analysis:** +- Target customer segments & personas +- Unique Selling Propositions (USPs) and competitive differentiators +- Comprehensive competitor landscape analysis (direct & indirect competitors) +- Market positioning assessment + +**Content Strategy Framework:** +- Seasonal content calendar with trend integration +- Visual storytelling direction (shot-by-shot creative guidance) +- Brand tone & voice guidelines +- Content themes aligned with target audience behaviors + +**SEO & AEO Optimization:** +- Recommended primary and long-tail keywords +- SEO-optimized taglines and meta descriptions +- Answer Engine Optimization (AEO) content suggestions +- Local search optimization strategies + +**Actionable Recommendations:** +- Content distribution strategy across platforms +- KPI measurement framework +- Budget allocation recommendations by content type + +콘텐츠 기획(Lyrics, Prompt for SUNO) +1. Based on the Brand & Marketing Intelligence Report for [업체명 + 지역명 / {customer_name} ({region})], create original lyrics and define music attributes (song mood, BPM, genres, and key musical motifs, Prompt for Suno.com) specifically tailored for viral content. +2. The lyrics should include, the name of [ Promotion Subject], [location], [main target],[Famous place, accessible in 10min], promotional words including but not limited to [인스타 감성], [사진같은 하루] + +Deliver outputs optimized for three formats:1 minute. Ensure that each version aligns with the brand's core identity and is suitable for use in digital marketing and social media campaigns, in Korean +""".strip() +# fmt: on + +LYRICS_PROMPT_TEMPLATE = """ +[ROLE] +Content marketing expert specializing in pension/accommodation services in Korea + +[INPUT] +- Business Name: {customer_name} +- Region: {region} +- Region Details: {detail_region_info} + +[INTERNAL ANALYSIS - DO NOT OUTPUT] +Analyze the following internally to inform lyrics creation: +- Target customer segments and personas +- Unique Selling Propositions (USPs) +- Regional characteristics and nearby attractions (within 10 min access) +- Seasonal appeal points + +[LYRICS REQUIREMENTS] +- Must include: business name, region name, main target audience, nearby famous places +- Keywords to incorporate: 인스타 감성, 사진같은 하루, 힐링, 여행 +- Length: For 1-minute video (approximately 8-12 lines) +- Tone: Emotional, trendy, viral-friendly + +[OUTPUT RULES - STRICTLY ENFORCED] +- Output lyrics ONLY +- Lyrics MUST be written in Korean (한국어) +- NO titles, descriptions, analysis, or explanations +- NO greetings or closing remarks +- NO additional commentary before or after lyrics +- Follow the exact format below + +[OUTPUT FORMAT - SUCCESS] +--- +[Lyrics in Korean here] +--- + +[OUTPUT FORMAT - FAILURE] +If you cannot generate lyrics due to insufficient information, invalid input, or any other reason: +--- +ERROR: [Brief reason for failure in English] +--- +""".strip() +# fmt: on + class ChatgptService: - def __init__(self): + def __init__( + self, + customer_name: str, + region: str, + detail_region_info: str = "", + ): + # 최신 모델: GPT-5, GPT-5 mini, GPT-5 nano, GPT-4.1, GPT-4.1 mini, GPT-4.1 nano + # 이전 세대: GPT-4o, GPT-4o mini, GPT-4 Turbo, GPT-3.5 Turbo self.model = "gpt-4o" self.client = AsyncOpenAI(api_key=apikey_settings.CHATGPT_API_KEY) + self.customer_name = customer_name + self.region = region + self.detail_region_info = detail_region_info - def lyrics_prompt( - self, - name, - address, - category, - description, - season, - num_of_people, - people_category, - genre, - sample_song=None, - ): - prompt = f""" - 반드시 한국어로 답변해주세요. - - 당신은 작곡가 입니다. - - 작곡에 대한 영감을 얻기 위해 제가 정보를 제공할게요. - - 업체 이름 : {name} - 업체 주소 : {address} - 업체 카테고리 : {category} - 업체 설명 : {description} - - 가사는 다음 속성들을 기반하여 작곡되어야 합니다. - - ### - - 위의 정보를 토대로 작곡을 이어나가주세요. - - 다음은 노래에 대한 정보입니다. - - 노래의 길이 : - - 1분 이내입니다. - - 글자 수는 120자에서 150자 사이로 작성해주세요. - - 글자가 갑자기 끊기지 않도록 주의해주세요. - - 노래의 특징: - - 노래에 업체의 이름은 반드시 1번 이상 들어가야 합니다. - - 노래의 전반부는, 업체에 대한 장점과 특징을 소개하는 가사를 써주세요. - - 노래의 후반부는, 업체가 위치한 곳을 소개하는 가사를 써주세요. - - 답변에 [전반부], [후반부] 표시할 필요 없이, 가사만 답변해주세요. - (후크)와 같이 특정 동작에 대해 표시할 필요 없습니다. - - 노래를 한 마디씩 생성할 때마다 글자수를 세어보면서, 글자 수가 150자를 넘지 않도록 주의해주세요. - - """ - - if sample_song: - prompt += f""" - - 다음은 참고해야 하는 샘플 가사 정보입니다. - - 샘플 가사를 참고하여 작곡을 해주세요. - - {sample_song} - """ - - return prompt - - async def generate_lyrics(self, prompt=None): - # prompt = self.lyrics_prompt( - # name, - # address, - # category, - # description, - # season, - # num_of_people, - # people_type, - # genre, - # sample_song, - # ) + def build_lyrics_prompt(self) -> str: + """LYRICS_PROMPT_TEMPLATE에 고객 정보를 대입하여 완성된 프롬프트 반환""" + return LYRICS_PROMPT_TEMPLATE.format( + customer_name=self.customer_name, + region=self.region, + detail_region_info=self.detail_region_info, + ) + async def generate_lyrics(self, prompt: str | None = None) -> str: + """GPT에게 프롬프트를 전달하여 결과를 반환""" + if prompt is None: + prompt = self.build_lyrics_prompt() print("Generated Prompt: ", prompt) completion = await self.client.chat.completions.create( model=self.model, messages=[{"role": "user", "content": prompt}] ) message = completion.choices[0].message.content - return message - - -chatgpt_api = ChatgptService() + return message or "" diff --git a/app/video/api/schemas/.gitkeep b/app/video/api/schemas/.gitkeep deleted file mode 100644 index e69de29..0000000 diff --git a/main.py b/main.py index 4d1a1bb..7d988b1 100644 --- a/main.py +++ b/main.py @@ -7,7 +7,7 @@ from app.admin_manager import init_admin from app.core.common import lifespan from app.database.session import engine from app.home.api.routers.v1.home import router as home_router -from app.lyric.api.routers.v1.router import router as lyrics_router +from app.lyric.api.routers.v1.lyric import router as lyric_router from app.utils.cors import CustomCORSMiddleware from config import prj_settings @@ -47,4 +47,4 @@ def get_scalar_docs(): app.include_router(home_router) -app.include_router(lyrics_router) +app.include_router(lyric_router) # Lyric API 라우터 추가