노래 백그라운드 추가

app/lyric/api/lyrics_admin.py

@@ -15,6 +15,7 @@ class LyricAdmin(ModelView, model=Lyric):
         "project_id",
         "task_id",
         "status",
+        "language",
         "created_at",
     ]
 
@@ -23,6 +24,7 @@ class LyricAdmin(ModelView, model=Lyric):
         "project_id",
         "task_id",
         "status",
+        "language",
         "lyric_prompt",
         "lyric_result",
         "created_at",
@@ -34,6 +36,7 @@ class LyricAdmin(ModelView, model=Lyric):
     column_searchable_list = [
         Lyric.task_id,
         Lyric.status,
+        Lyric.language,
     ]
 
     column_default_sort = (Lyric.created_at, True)  # True: DESC (최신순)
@@ -42,6 +45,7 @@ class LyricAdmin(ModelView, model=Lyric):
         Lyric.id,
         Lyric.project_id,
         Lyric.status,
+        Lyric.language,
         Lyric.created_at,
     ]
 
@@ -50,6 +54,7 @@ class LyricAdmin(ModelView, model=Lyric):
         "project_id": "프로젝트 ID",
         "task_id": "작업 ID",
         "status": "상태",
+        "language": "언어",
         "lyric_prompt": "프롬프트",
         "lyric_result": "생성 결과",
         "created_at": "생성일시",

app/lyric/api/routers/v1/lyric.py

@@ -5,6 +5,7 @@ Lyric API Router
 모든 엔드포인트는 재사용 가능하도록 설계되었습니다.
 
 엔드포인트 목록:
+    - POST /lyric/generate: 가사 생성
     - GET /lyric/status/{task_id}: 가사 생성 상태 조회
     - GET /lyric/{task_id}: 가사 상세 조회
     - GET /lyrics: 가사 목록 조회 (페이지네이션)
@@ -18,17 +19,17 @@ Lyric API Router
     from app.lyric.api.routers.v1.lyric import (
         get_lyric_status_by_task_id,
         get_lyric_by_task_id,
-        get_lyrics_paginated,
     )
+
+    # 페이지네이션은 pagination 모듈 사용
+    from app.utils.pagination import PaginatedResponse, get_paginated
 """
 
-import math
 from typing import Optional
 
 from fastapi import APIRouter, Depends, HTTPException, Query, status
-from sqlalchemy import func, select
+from sqlalchemy import 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
@@ -39,9 +40,10 @@ from app.lyric.schemas.lyric import (
     LyricDetailResponse,
     LyricListItem,
     LyricStatusResponse,
-    PaginatedResponse,
 )
 from app.utils.chatgpt_prompt import ChatgptService
+from app.utils.common import generate_task_id
+from app.utils.pagination import PaginatedResponse, get_paginated
 
 router = APIRouter(prefix="/lyric", tags=["lyric"])
 
@@ -74,10 +76,12 @@ async def get_lyric_status_by_task_id(
         if status_info.status == "completed":
             # 완료 처리
     """
+    print(f"[get_lyric_status_by_task_id] START - task_id: {task_id}")
     result = await session.execute(select(Lyric).where(Lyric.task_id == task_id))
     lyric = result.scalar_one_or_none()
 
     if not lyric:
+        print(f"[get_lyric_status_by_task_id] NOT FOUND - task_id: {task_id}")
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
             detail=f"task_id '{task_id}'에 해당하는 가사를 찾을 수 없습니다.",
@@ -89,6 +93,9 @@ async def get_lyric_status_by_task_id(
         "failed": "가사 생성에 실패했습니다.",
     }
 
+    print(
+        f"[get_lyric_status_by_task_id] SUCCESS - task_id: {task_id}, status: {lyric.status}"
+    )
     return LyricStatusResponse(
         task_id=lyric.task_id,
         status=lyric.status,
@@ -116,17 +123,19 @@ async def get_lyric_by_task_id(
         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)
     """
+    print(f"[get_lyric_by_task_id] START - task_id: {task_id}")
     result = await session.execute(select(Lyric).where(Lyric.task_id == task_id))
     lyric = result.scalar_one_or_none()
 
     if not lyric:
+        print(f"[get_lyric_by_task_id] NOT FOUND - task_id: {task_id}")
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
             detail=f"task_id '{task_id}'에 해당하는 가사를 찾을 수 없습니다.",
         )
 
+    print(f"[get_lyric_by_task_id] SUCCESS - task_id: {task_id}, lyric_id: {lyric.id}")
     return LyricDetailResponse(
         id=lyric.id,
         task_id=lyric.task_id,
@@ -138,82 +147,6 @@ async def get_lyric_by_task_id(
     )
 
 
-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
 # =============================================================================
@@ -234,11 +167,15 @@ async def get_lyrics_paginated(
 ## 반환 정보
 - **success**: 생성 성공 여부
 - **task_id**: 작업 고유 식별자
-- **lyric**: 생성된 가사
+- **lyric**: 생성된 가사 (성공 시)
 - **language**: 가사 언어
-- **prompt_used**: 사용된 프롬프트
 - **error_message**: 에러 메시지 (실패 시)
 
+## 실패 조건
+- ChatGPT API 오류
+- ChatGPT 거부 응답 (I'm sorry, I cannot 등)
+- 응답에 ERROR: 포함
+
 ## 사용 예시
 ```
 POST /lyric/generate
@@ -246,14 +183,36 @@ POST /lyric/generate
     "customer_name": "스테이 머뭄",
     "region": "군산",
     "detail_region_info": "군산 신흥동 말랭이 마을",
-    "language": "English"
+    "language": "Korean"
+}
+```
+
+## 응답 예시 (성공)
+```json
+{
+    "success": true,
+    "task_id": "0694b716-dbff-7219-8000-d08cb5fce431",
+    "lyric": "인스타 감성의 스테이 머뭄...",
+    "language": "Korean",
+    "error_message": null
+}
+```
+
+## 응답 예시 (실패)
+```json
+{
+    "success": false,
+    "task_id": "0694b716-dbff-7219-8000-d08cb5fce431",
+    "lyric": null,
+    "language": "Korean",
+    "error_message": "I'm sorry, I can't comply with that request."
 }
 ```
     """,
     response_model=GenerateLyricResponse,
     responses={
-        200: {"description": "가사 생성 성공"},
+        200: {"description": "가사 생성 성공 또는 실패 (success 필드로 구분)"},
-        500: {"description": "가사 생성 실패"},
+        500: {"description": "서버 내부 오류"},
     },
 )
 async def generate_lyric(
@@ -261,7 +220,10 @@ async def generate_lyric(
     session: AsyncSession = Depends(get_session),
 ) -> GenerateLyricResponse:
     """고객 정보를 기반으로 가사를 생성합니다."""
-    task_id = str(uuid7())
+    task_id = await generate_task_id(session=session, table_name=Project)
+    print(
+        f"[generate_lyric] START - task_id: {task_id}, customer_name: {request_body.customer_name}, region: {request_body.region}"
+    )
 
     try:
         # 1. ChatGPT 서비스 초기화 및 프롬프트 생성
@@ -284,6 +246,9 @@ async def generate_lyric(
         session.add(project)
         await session.commit()
         await session.refresh(project)  # commit 후 project.id 동기화
+        print(
+            f"[generate_lyric] Project saved - project_id: {project.id}, task_id: {task_id}"
+        )
 
         # 3. Lyric 테이블에 데이터 저장 (status: processing)
         lyric = Lyric(
@@ -295,14 +260,37 @@ async def generate_lyric(
             language=request_body.language,
         )
         session.add(lyric)
-        await session.commit()  # processing 상태를 확실히 저장 (다른 트랜잭션에서 조회 가능)
+        await (
+            session.commit()
+        )  # processing 상태를 확실히 저장 (다른 트랜잭션에서 조회 가능)
         await session.refresh(lyric)  # commit 후 객체 상태 동기화
+        print(
+            f"[generate_lyric] Lyric saved (processing) - lyric_id: {lyric.id}, task_id: {task_id}"
+        )
 
         # 4. ChatGPT를 통해 가사 생성
+        print(f"[generate_lyric] ChatGPT generation started - task_id: {task_id}")
         result = await service.generate(prompt=prompt)
+        print(f"[generate_lyric] ChatGPT generation completed - task_id: {task_id}")
 
-        # 5. ERROR가 포함되어 있으면 실패 처리
+        # 5. 실패 응답 검사 (ERROR 또는 ChatGPT 거부 응답)
-        if "ERROR:" in result:
+        failure_patterns = [
+            "ERROR:",
+            "I'm sorry",
+            "I cannot",
+            "I can't",
+            "I apologize",
+            "I'm unable",
+            "I am unable",
+            "I'm not able",
+            "I am not able",
+        ]
+        is_failure = any(
+            pattern.lower() in result.lower() for pattern in failure_patterns
+        )
+
+        if is_failure:
+            print(f"[generate_lyric] FAILED - task_id: {task_id}, error: {result}")
             lyric.status = "failed"
             lyric.lyric_result = result
             await session.commit()
@@ -312,7 +300,6 @@ async def generate_lyric(
                 task_id=task_id,
                 lyric=None,
                 language=request_body.language,
-                prompt_used=prompt,
                 error_message=result,
             )
 
@@ -321,22 +308,22 @@ async def generate_lyric(
         lyric.lyric_result = result
         await session.commit()
 
+        print(f"[generate_lyric] SUCCESS - task_id: {task_id}")
         return GenerateLyricResponse(
             success=True,
             task_id=task_id,
             lyric=result,
             language=request_body.language,
-            prompt_used=prompt,
             error_message=None,
         )
     except Exception as e:
+        print(f"[generate_lyric] EXCEPTION - task_id: {task_id}, error: {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),
         )
 
@@ -375,15 +362,14 @@ async def get_lyric_status(
     "s",
     summary="가사 목록 조회 (페이지네이션)",
     description="""
-생성된 모든 가사를 페이지네이션으로 조회합니다.
+생성 완료된 가사를 페이지네이션으로 조회합니다.
 
 ## 파라미터
 - **page**: 페이지 번호 (1부터 시작, 기본값: 1)
 - **page_size**: 페이지당 데이터 수 (기본값: 20, 최대: 100)
-- **status**: 상태 필터 (선택사항) - "processing", "completed", "failed"
 
 ## 반환 정보
-- **items**: 가사 목록
+- **items**: 가사 목록 (completed 상태만)
 - **total**: 전체 데이터 수
 - **page**: 현재 페이지
 - **page_size**: 페이지당 데이터 수
@@ -396,18 +382,11 @@ async def get_lyric_status(
 GET /lyrics                           # 기본 조회 (1페이지, 20개)
 GET /lyrics?page=2                    # 2페이지 조회
 GET /lyrics?page=1&page_size=50       # 50개씩 조회
-GET /lyrics?status=completed          # 완료된 가사만 조회
 ```
 
-## 다른 모델에서 PaginatedResponse 재사용
+## 참고
-```python
+- 생성 완료(completed)된 가사만 조회됩니다.
-from app.lyric.api.schemas.lyric import PaginatedResponse
+- processing, failed 상태의 가사는 조회되지 않습니다.
-
-# Song 목록에서 사용
-@router.get("/songs", response_model=PaginatedResponse[SongListItem])
-async def list_songs(...):
-    ...
-```
     """,
     response_model=PaginatedResponse[LyricListItem],
     responses={
@@ -417,15 +396,19 @@ async def list_songs(...):
 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)
+    return await get_paginated(
+        session=session,
+        model=Lyric,
+        item_schema=LyricListItem,
+        page=page,
+        page_size=page_size,
+        filters={"status": "completed"},
+        order_by="created_at",
+        order_desc=True,
+    )
 
 
 @router.get(

app/lyric/schemas/lyric.py

@@ -8,23 +8,22 @@ Lyric API Schemas
         LyricStatusResponse,
         LyricDetailResponse,
         LyricListItem,
-        PaginatedResponse,
     )
+    from app.utils.pagination import 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 typing import Optional
 
 from pydantic import BaseModel, Field
 
@@ -72,22 +71,36 @@ class GenerateLyricResponse(BaseModel):
         POST /lyric/generate
         Returns the generated lyrics.
 
-    Example Response:
+    Note:
+        실패 조건:
+        - ChatGPT API 오류
+        - ChatGPT 거부 응답 (I'm sorry, I cannot, I can't, I apologize 등)
+        - 응답에 ERROR: 포함
+
+    Example Response (Success):
         {
             "success": true,
             "task_id": "019123ab-cdef-7890-abcd-ef1234567890",
-            "lyric": "생성된 가사...",
+            "lyric": "인스타 감성의 스테이 머뭄...",
             "language": "Korean",
-            "prompt_used": "..."
+            "error_message": null
+        }
+
+    Example Response (Failure):
+        {
+            "success": false,
+            "task_id": "019123ab-cdef-7890-abcd-ef1234567890",
+            "lyric": null,
+            "language": "Korean",
+            "error_message": "I'm sorry, I can't comply with that request."
         }
     """
 
     success: bool = Field(..., description="생성 성공 여부")
-    task_id: Optional[str] = Field(None, description="작업 고유 식별자")
+    task_id: Optional[str] = Field(None, description="작업 고유 식별자 (uuid7)")
-    lyric: Optional[str] = Field(None, description="생성된 가사")
+    lyric: Optional[str] = Field(None, description="생성된 가사 (성공 시)")
     language: str = Field(..., description="가사 언어")
-    prompt_used: Optional[str] = Field(None, description="사용된 프롬프트")
+    error_message: Optional[str] = Field(None, description="에러 메시지 (실패 시, ChatGPT 거부 응답 포함)")
-    error_message: Optional[str] = Field(None, description="에러 메시지 (실패 시)")
 
 
 class LyricStatusResponse(BaseModel):
@@ -150,77 +163,3 @@ class LyricListItem(BaseModel):
     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,
-        )

app/song/api/routers/v1/router.py

@@ -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.lyrics.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

app/song/api/routers/v1/song.py

@@ -4,110 +4,45 @@ Song API Router
 이 모듈은 Suno API를 통한 노래 생성 관련 API 엔드포인트를 정의합니다.
 
 엔드포인트 목록:
-    - POST /song/generate: 노래 생성 요청
+    - POST /song/generate/{task_id}: 노래 생성 요청 (task_id로 Project/Lyric 연결)
-    - GET /song/status/{task_id}: 노래 생성 상태 조회
+    - GET /song/status/{suno_task_id}: Suno API 노래 생성 상태 조회
-    - GET /song/download/{task_id}: 노래 다운로드
+    - GET /song/download/{task_id}: 노래 다운로드 상태 조회 (DB polling)
 
 사용 예시:
     from app.song.api.routers.v1.song import router
     app.include_router(router, prefix="/api/v1")
 """
 
-from datetime import date
+from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
-from pathlib import Path
+from sqlalchemy import select
-
+from sqlalchemy.ext.asyncio import AsyncSession
-import aiofiles
-import httpx
-from fastapi import APIRouter
-from uuid_extensions import uuid7str
 
+from app.database.session import get_session
+from app.home.models import Project
+from app.lyric.models import Lyric
+from app.song.models import Song
 from app.song.schemas.song_schema import (
     DownloadSongResponse,
     GenerateSongRequest,
     GenerateSongResponse,
     PollingSongResponse,
-    SongClipData,
 )
+from app.song.worker.song_task import download_and_save_song
 from app.utils.suno import SunoService
-from config import prj_settings
 
 
-def _parse_suno_status_response(result: dict | None) -> PollingSongResponse:
-    """Suno API 상태 응답을 파싱하여 PollingSongResponse로 변환합니다."""
-    if result is None:
-        return PollingSongResponse(
-            success=False,
-            status="error",
-            message="Suno API 응답이 비어있습니다.",
-            clips=None,
-            raw_response=None,
-            error_message="Suno API returned None response",
-        )
-
-    code = result.get("code", 0)
-    data = result.get("data", {})
-
-    if code != 200:
-        return PollingSongResponse(
-            success=False,
-            status="failed",
-            message="Suno API 응답 오류",
-            clips=None,
-            raw_response=result,
-            error_message=result.get("msg", "Unknown error"),
-        )
-
-    status = data.get("status", "unknown")
-
-    # 클립 데이터는 data.response.sunoData에 있음 (camelCase)
-    # data.get()이 None을 반환할 수 있으므로 or {}로 처리
-    response_data = data.get("response") or {}
-    clips_data = response_data.get("sunoData") or []
-
-    # 상태별 메시지 (Suno API는 다양한 상태값 반환)
-    status_messages = {
-        "pending": "노래 생성 대기 중입니다.",
-        "processing": "노래를 생성하고 있습니다.",
-        "complete": "노래 생성이 완료되었습니다.",
-        "SUCCESS": "노래 생성이 완료되었습니다.",
-        "TEXT_SUCCESS": "노래 생성이 완료되었습니다.",
-        "failed": "노래 생성에 실패했습니다.",
-    }
-
-    # 클립 데이터 파싱 (Suno API는 camelCase 사용)
-    clips = None
-    if clips_data:
-        clips = [
-            SongClipData(
-                id=clip.get("id"),
-                audio_url=clip.get("audioUrl"),
-                stream_audio_url=clip.get("streamAudioUrl"),
-                image_url=clip.get("imageUrl"),
-                title=clip.get("title"),
-                status=clip.get("status"),
-                duration=clip.get("duration"),
-            )
-            for clip in clips_data
-        ]
-
-    return PollingSongResponse(
-        success=True,
-        status=status,
-        message=status_messages.get(status, f"상태: {status}"),
-        clips=clips,
-        raw_response=result,
-        error_message=None,
-    )
-
 router = APIRouter(prefix="/song", tags=["song"])
 
 
 @router.post(
-    "/generate",
+    "/generate/{task_id}",
     summary="노래 생성 요청",
     description="""
 Suno API를 통해 노래 생성을 요청합니다.
 
+## 경로 파라미터
+- **task_id**: Project/Lyric의 task_id (필수) - 연관된 프로젝트와 가사를 조회하는 데 사용
+
 ## 요청 필드
 - **lyrics**: 노래에 사용할 가사 (필수)
 - **genre**: 음악 장르 (필수) - K-Pop, Pop, R&B, Hip-Hop, Ballad, EDM, Rock, Jazz 등
@@ -115,12 +50,13 @@ Suno API를 통해 노래 생성을 요청합니다.
 
 ## 반환 정보
 - **success**: 요청 성공 여부
-- **task_id**: Suno 작업 ID (폴링에 사용)
+- **task_id**: 내부 작업 ID (Project/Lyric task_id)
+- **suno_task_id**: Suno API 작업 ID (상태 조회에 사용)
 - **message**: 응답 메시지
 
 ## 사용 예시
 ```
-POST /song/generate
+POST /song/generate/019123ab-cdef-7890-abcd-ef1234567890
 {
     "lyrics": "여기 군산에서 만나요\\n아름다운 하루를 함께",
     "genre": "K-Pop",
@@ -130,53 +66,124 @@ POST /song/generate
 
 ## 참고
 - 생성되는 노래는 약 1분 이내 길이입니다.
-- task_id를 사용하여 /status/{task_id} 엔드포인트에서 생성 상태를 확인할 수 있습니다.
+- suno_task_id를 사용하여 /status/{suno_task_id} 엔드포인트에서 생성 상태를 확인할 수 있습니다.
+- Song 테이블에 데이터가 저장되며, project_id와 lyric_id가 자동으로 연결됩니다.
     """,
     response_model=GenerateSongResponse,
     responses={
         200: {"description": "노래 생성 요청 성공"},
+        404: {"description": "Project 또는 Lyric을 찾을 수 없음"},
         500: {"description": "노래 생성 요청 실패"},
     },
 )
 async def generate_song(
+    task_id: str,
     request_body: GenerateSongRequest,
+    session: AsyncSession = Depends(get_session),
 ) -> GenerateSongResponse:
-    """가사와 장르를 기반으로 Suno API를 통해 노래를 생성합니다."""
+    """가사와 장르를 기반으로 Suno API를 통해 노래를 생성합니다.
-    try:
-        suno_service = SunoService()
 
-        task_id = await suno_service.generate(
+    1. task_id로 Project와 Lyric 조회
+    2. Song 테이블에 초기 데이터 저장 (status: processing)
+    3. Suno API 호출
+    4. suno_task_id 업데이트 후 응답 반환
+    """
+    print(f"[generate_song] START - task_id: {task_id}, genre: {request_body.genre}, language: {request_body.language}")
+    try:
+        # 1. task_id로 Project 조회
+        project_result = await session.execute(
+            select(Project).where(Project.task_id == task_id)
+        )
+        project = project_result.scalar_one_or_none()
+
+        if not project:
+            print(f"[generate_song] Project NOT FOUND - task_id: {task_id}")
+            raise HTTPException(
+                status_code=404,
+                detail=f"task_id '{task_id}'에 해당하는 Project를 찾을 수 없습니다.",
+            )
+        print(f"[generate_song] Project found - project_id: {project.id}, task_id: {task_id}")
+
+        # 2. task_id로 Lyric 조회
+        lyric_result = await session.execute(
+            select(Lyric).where(Lyric.task_id == task_id)
+        )
+        lyric = lyric_result.scalar_one_or_none()
+
+        if not lyric:
+            print(f"[generate_song] Lyric NOT FOUND - task_id: {task_id}")
+            raise HTTPException(
+                status_code=404,
+                detail=f"task_id '{task_id}'에 해당하는 Lyric을 찾을 수 없습니다.",
+            )
+        print(f"[generate_song] Lyric found - lyric_id: {lyric.id}, task_id: {task_id}")
+
+        # 3. Song 테이블에 초기 데이터 저장
+        song_prompt = (
+            f"[Lyrics]\n{request_body.lyrics}\n\n[Genre]\n{request_body.genre}"
+        )
+
+        song = Song(
+            project_id=project.id,
+            lyric_id=lyric.id,
+            task_id=task_id,
+            suno_task_id=None,
+            status="processing",
+            song_prompt=song_prompt,
+            language=request_body.language,
+        )
+        session.add(song)
+        await session.flush()  # ID 생성을 위해 flush
+        print(f"[generate_song] Song saved (processing) - task_id: {task_id}")
+
+        # 4. Suno API 호출
+        print(f"[generate_song] Suno API generation started - task_id: {task_id}")
+        suno_service = SunoService()
+        suno_task_id = await suno_service.generate(
             prompt=request_body.lyrics,
             genre=request_body.genre,
         )
 
+        # 5. suno_task_id 업데이트
+        song.suno_task_id = suno_task_id
+        await session.commit()
+        print(f"[generate_song] SUCCESS - task_id: {task_id}, suno_task_id: {suno_task_id}")
+
         return GenerateSongResponse(
             success=True,
             task_id=task_id,
-            message="노래 생성 요청이 접수되었습니다. task_id로 상태를 조회하세요.",
+            suno_task_id=suno_task_id,
+            message="노래 생성 요청이 접수되었습니다. suno_task_id로 상태를 조회하세요.",
             error_message=None,
         )
+
+    except HTTPException:
+        raise
     except Exception as e:
+        print(f"[generate_song] EXCEPTION - task_id: {task_id}, error: {e}")
+        await session.rollback()
         return GenerateSongResponse(
             success=False,
-            task_id=None,
+            task_id=task_id,
+            suno_task_id=None,
             message="노래 생성 요청에 실패했습니다.",
             error_message=str(e),
         )
 
 
 @router.get(
-    "/status/{task_id}",
+    "/status/{suno_task_id}",
     summary="노래 생성 상태 조회",
     description="""
 Suno API를 통해 노래 생성 작업의 상태를 조회합니다.
+SUCCESS 상태인 경우 백그라운드에서 MP3 파일을 다운로드하고 Song 테이블을 업데이트합니다.
 
 ## 경로 파라미터
-- **task_id**: 노래 생성 시 반환된 작업 ID (필수)
+- **suno_task_id**: 노래 생성 시 반환된 Suno API 작업 ID (필수)
 
 ## 반환 정보
 - **success**: 조회 성공 여부
-- **status**: 작업 상태 (pending, processing, complete, failed)
+- **status**: 작업 상태 (PENDING, processing, SUCCESS, failed)
 - **message**: 상태 메시지
 - **clips**: 생성된 노래 클립 목록 (완료 시)
 - **raw_response**: Suno API 원본 응답
@@ -187,14 +194,15 @@ GET /song/status/abc123...
 ```
 
 ## 상태 값
-- **pending**: 대기 중
+- **PENDING**: 대기 중
 - **processing**: 생성 중
-- **complete**: 생성 완료
+- **SUCCESS**: 생성 완료
 - **failed**: 생성 실패
 
 ## 참고
 - 스트림 URL: 30-40초 내 생성
 - 다운로드 URL: 2-3분 내 생성
+- SUCCESS 시 백그라운드에서 MP3 다운로드 및 DB 업데이트 진행
     """,
     response_model=PollingSongResponse,
     responses={
@@ -203,16 +211,63 @@ GET /song/status/abc123...
     },
 )
 async def get_song_status(
-    task_id: str,
+    suno_task_id: str,
+    background_tasks: BackgroundTasks,
+    session: AsyncSession = Depends(get_session),
 ) -> PollingSongResponse:
-    """task_id로 노래 생성 작업의 상태를 조회합니다."""
+    """suno_task_id로 노래 생성 작업의 상태를 조회합니다.
+
+    SUCCESS 상태인 경우 백그라운드에서 MP3 파일을 다운로드하고
+    Song 테이블의 status를 completed로, song_result_url을 업데이트합니다.
+    """
+    print(f"[get_song_status] START - suno_task_id: {suno_task_id}")
     try:
         suno_service = SunoService()
-        result = await suno_service.get_task_status(task_id)
+        result = await suno_service.get_task_status(suno_task_id)
-        return _parse_suno_status_response(result)
+        parsed_response = suno_service.parse_status_response(result)
+        print(f"[get_song_status] Suno API response - suno_task_id: {suno_task_id}, status: {parsed_response.status}")
+
+        # SUCCESS 상태인 경우 백그라운드 태스크 실행
+        if parsed_response.status == "SUCCESS" and parsed_response.clips:
+            # 첫 번째 클립의 audioUrl 가져오기
+            first_clip = parsed_response.clips[0]
+            audio_url = first_clip.audio_url
+
+            if audio_url:
+                # suno_task_id로 Song 조회하여 task_id 가져오기 (여러 개 있을 경우 가장 최근 것 선택)
+                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 song:
+                    # task_id로 Project 조회하여 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"
+
+                    # 백그라운드 태스크로 MP3 다운로드 및 DB 업데이트
+                    print(f"[get_song_status] Background task args - task_id: {song.task_id}, audio_url: {audio_url}, store_name: {store_name}")
+                    background_tasks.add_task(
+                        download_and_save_song,
+                        task_id=song.task_id,
+                        audio_url=audio_url,
+                        store_name=store_name,
+                    )
+
+        print(f"[get_song_status] SUCCESS - suno_task_id: {suno_task_id}")
+        return parsed_response
+
     except Exception as e:
         import traceback
 
+        print(f"[get_song_status] EXCEPTION - suno_task_id: {suno_task_id}, error: {e}")
         return PollingSongResponse(
             success=False,
             status="error",
@@ -225,127 +280,115 @@ async def get_song_status(
 
 @router.get(
     "/download/{task_id}",
-    summary="노래 다운로드",
+    summary="노래 다운로드 상태 조회",
     description="""
-완료된 노래를 서버에 다운로드하고 접근 가능한 URL을 반환합니다.
+        task_id를 기반으로 Song 테이블의 상태를 polling하고,
+        completed인 경우 Project 정보와 노래 URL을 반환합니다.
 
-## 경로 파라미터
+        ## 경로 파라미터
-- **task_id**: 노래 생성 시 반환된 작업 ID (필수)
+        - **task_id**: 프로젝트 task_id (필수)
 
-## 반환 정보
+        ## 반환 정보
-- **success**: 다운로드 성공 여부
+        - **success**: 조회 성공 여부
-- **message**: 응답 메시지
+        - **status**: 처리 상태 (processing, completed, failed)
-- **file_path**: 저장된 파일의 상대 경로
+        - **message**: 응답 메시지
-- **file_url**: 프론트엔드에서 접근 가능한 파일 URL
+        - **store_name**: 업체명
+        - **region**: 지역명
+        - **detail_region_info**: 상세 지역 정보
+        - **task_id**: 작업 고유 식별자
+        - **language**: 언어
+        - **song_result_url**: 노래 결과 URL (completed 시)
+        - **created_at**: 생성 일시
 
-## 사용 예시
+        ## 사용 예시
-```
+        ```
-GET /song/download/abc123...
+        GET /song/download/019123ab-cdef-7890-abcd-ef1234567890
-```
+        ```
 
-## 참고
+        ## 참고
-- 노래 생성이 완료된 상태(complete)에서만 다운로드 가능합니다.
+        - processing 상태인 경우 song_result_url은 null입니다.
-- 파일은 /media/{날짜}/{uuid7}/song.mp3 경로에 저장됩니다.
+        - completed 상태인 경우 Project 정보와 함께 song_result_url을 반환합니다.
-- 반환된 file_url을 사용하여 프론트엔드에서 MP3를 재생할 수 있습니다.
     """,
     response_model=DownloadSongResponse,
     responses={
-        200: {"description": "다운로드 성공"},
+        200: {"description": "조회 성공"},
-        400: {"description": "노래 생성이 완료되지 않음"},
+        404: {"description": "Song을 찾을 수 없음"},
-        500: {"description": "다운로드 실패"},
+        500: {"description": "조회 실패"},
     },
 )
 async def download_song(
     task_id: str,
+    session: AsyncSession = Depends(get_session),
 ) -> DownloadSongResponse:
-    """완료된 노래를 다운로드하여 서버에 저장하고 접근 URL을 반환합니다."""
+    """task_id로 Song 상태를 polling하고 completed 시 Project 정보와 노래 URL을 반환합니다."""
+    print(f"[download_song] START - task_id: {task_id}")
     try:
-        suno_service = SunoService()
+        # task_id로 Song 조회 (여러 개 있을 경우 가장 최근 것 선택)
-        result = await suno_service.get_task_status(task_id)
+        song_result = await session.execute(
+            select(Song)
+            .where(Song.task_id == task_id)
+            .order_by(Song.created_at.desc())
+            .limit(1)
+        )
+        song = song_result.scalar_one_or_none()
 
-        # API 응답 확인
+        if not song:
-        if result.get("code") != 200:
+            print(f"[download_song] Song NOT FOUND - task_id: {task_id}")
             return DownloadSongResponse(
                 success=False,
-                message="Suno API 응답 오류",
+                status="not_found",
-                error_message=result.get("msg", "Unknown error"),
+                message=f"task_id '{task_id}'에 해당하는 Song을 찾을 수 없습니다.",
+                error_message="Song not found",
             )
 
-        data = result.get("data", {})
+        print(f"[download_song] Song found - task_id: {task_id}, status: {song.status}")
-        status = data.get("status", "unknown")
-
-        # 완료 상태 확인 (Suno API는 다양한 완료 상태값 반환)
-        completed_statuses = {"complete", "SUCCESS", "TEXT_SUCCESS"}
-        if status not in completed_statuses:
-            return DownloadSongResponse(
-                success=False,
-                message=f"노래 생성이 완료되지 않았습니다. 현재 상태: {status}",
-                error_message="노래 생성 완료 후 다운로드해 주세요.",
-            )
-
-        # 클립 데이터는 data.response.sunoData에 있음 (camelCase)
-        # data.get()이 None을 반환할 수 있으므로 or {}로 처리
-        response_data = data.get("response") or {}
-        clips_data = response_data.get("sunoData") or []
-        if not clips_data:
-            return DownloadSongResponse(
-                success=False,
-                message="생성된 노래 클립이 없습니다.",
-                error_message="sunoData is empty",
-            )
-
-        # 첫 번째 클립의 streamAudioUrl 가져오기 (camelCase)
-        first_clip = clips_data[0]
-        stream_audio_url = first_clip.get("streamAudioUrl")
-
-        if not stream_audio_url:
-            return DownloadSongResponse(
-                success=False,
-                message="스트리밍 오디오 URL을 찾을 수 없습니다.",
-                error_message="stream_audio_url is missing",
-            )
-
-        # 저장 경로 생성: media/{날짜}/{uuid7}/song.mp3
-        today = date.today().isoformat()
-        unique_id = uuid7str()
-        relative_dir = f"{today}/{unique_id}"
-        file_name = "song.mp3"
-
-        # 절대 경로 생성
-        media_dir = Path("media") / today / unique_id
-        media_dir.mkdir(parents=True, exist_ok=True)
-        file_path = media_dir / file_name
-
-        # 오디오 파일 다운로드 (비동기 파일 쓰기)
-        async with httpx.AsyncClient() as client:
-            response = await client.get(stream_audio_url, timeout=60.0)
-            response.raise_for_status()
-
-            # aiofiles는 Path 객체를 문자열로 변환하여 사용
-            async with aiofiles.open(str(file_path), "wb") as f:
-                await f.write(response.content)
-
-        # 프론트엔드에서 접근 가능한 URL 생성
-        relative_path = f"/media/{relative_dir}/{file_name}"
-        base_url = f"http://{prj_settings.PROJECT_DOMAIN}"
-        file_url = f"{base_url}{relative_path}"
 
+        # processing 상태인 경우
+        if song.status == "processing":
+            print(f"[download_song] PROCESSING - task_id: {task_id}")
             return DownloadSongResponse(
                 success=True,
-            message="노래 다운로드가 완료되었습니다.",
+                status="processing",
-            file_path=relative_path,
+                message="노래 생성이 진행 중입니다.",
-            file_url=file_url,
+                task_id=task_id,
             )
 
-    except httpx.HTTPError as e:
+        # failed 상태인 경우
+        if song.status == "failed":
+            print(f"[download_song] FAILED - task_id: {task_id}")
             return DownloadSongResponse(
                 success=False,
-            message="오디오 파일 다운로드에 실패했습니다.",
+                status="failed",
-            error_message=str(e),
+                message="노래 생성에 실패했습니다.",
+                task_id=task_id,
+                error_message="Song generation failed",
             )
+
+        # completed 상태인 경우 - Project 정보 조회
+        project_result = await session.execute(
+            select(Project).where(Project.id == song.project_id)
+        )
+        project = project_result.scalar_one_or_none()
+
+        print(f"[download_song] COMPLETED - task_id: {task_id}, song_result_url: {song.song_result_url}")
+        return DownloadSongResponse(
+            success=True,
+            status="completed",
+            message="노래 다운로드가 완료되었습니다.",
+            store_name=project.store_name if project else None,
+            region=project.region if project else None,
+            detail_region_info=project.detail_region_info if project else None,
+            task_id=task_id,
+            language=project.language if project else None,
+            song_result_url=song.song_result_url,
+            created_at=song.created_at,
+        )
+
     except Exception as e:
+        print(f"[download_song] EXCEPTION - task_id: {task_id}, error: {e}")
         return DownloadSongResponse(
             success=False,
-            message="노래 다운로드에 실패했습니다.",
+            status="error",
+            message="노래 다운로드 조회에 실패했습니다.",
             error_message=str(e),
         )

app/song/api/song_admin.py

@@ -15,7 +15,9 @@ class SongAdmin(ModelView, model=Song):
         "project_id",
         "lyric_id",
         "task_id",
+        "suno_task_id",
         "status",
+        "language",
         "created_at",
     ]
 
@@ -24,10 +26,11 @@ class SongAdmin(ModelView, model=Song):
         "project_id",
         "lyric_id",
         "task_id",
+        "suno_task_id",
         "status",
+        "language",
         "song_prompt",
-        "song_result_url_1",
+        "song_result_url",
-        "song_result_url_2",
         "created_at",
     ]
 
@@ -36,7 +39,9 @@ class SongAdmin(ModelView, model=Song):
 
     column_searchable_list = [
         Song.task_id,
+        Song.suno_task_id,
         Song.status,
+        Song.language,
     ]
 
     column_default_sort = (Song.created_at, True)  # True: DESC (최신순)
@@ -46,6 +51,7 @@ class SongAdmin(ModelView, model=Song):
         Song.project_id,
         Song.lyric_id,
         Song.status,
+        Song.language,
         Song.created_at,
     ]
 
@@ -54,9 +60,10 @@ class SongAdmin(ModelView, model=Song):
         "project_id": "프로젝트 ID",
         "lyric_id": "가사 ID",
         "task_id": "작업 ID",
+        "suno_task_id": "Suno 작업 ID",
         "status": "상태",
+        "language": "언어",
         "song_prompt": "프롬프트",
-        "song_result_url_1": "결과 URL 1",
+        "song_result_url": "결과 URL",
-        "song_result_url_2": "결과 URL 2",
         "created_at": "생성일시",
     }

app/song/models.py

@@ -17,17 +17,18 @@ class Song(Base):
     노래 테이블
 
     AI를 통해 생성된 노래 정보를 저장합니다.
-    가사를 기반으로 생성되며, 두 개의 결과 URL을 저장할 수 있습니다.
+    가사를 기반으로 생성됩니다.
 
     Attributes:
         id: 고유 식별자 (자동 증가)
         project_id: 연결된 Project의 id (외래키)
         lyric_id: 연결된 Lyric의 id (외래키)
         task_id: 노래 생성 작업의 고유 식별자 (UUID 형식)
+        suno_task_id: Suno API 작업 고유 식별자 (선택)
         status: 처리 상태 (pending, processing, completed, failed 등)
         song_prompt: 노래 생성에 사용된 프롬프트
-        song_result_url_1: 첫 번째 생성 결과 URL (선택)
+        song_result_url: 생성 결과 URL (선택)
-        song_result_url_2: 두 번째 생성 결과 URL (선택)
+        language: 출력 언어
         created_at: 생성 일시 (자동 설정)
 
     Relationships:
@@ -75,6 +76,12 @@ class Song(Base):
         comment="노래 생성 작업 고유 식별자 (UUID)",
     )
 
+    suno_task_id: Mapped[Optional[str]] = mapped_column(
+        String(64),
+        nullable=True,
+        comment="Suno API 작업 고유 식별자",
+    )
+
     status: Mapped[str] = mapped_column(
         String(50),
         nullable=False,
@@ -87,16 +94,17 @@ class Song(Base):
         comment="노래 생성에 사용된 프롬프트",
     )
 
-    song_result_url_1: Mapped[Optional[str]] = mapped_column(
+    song_result_url: Mapped[Optional[str]] = mapped_column(
         String(2048),
         nullable=True,
-        comment="첫 번째 노래 결과 URL",
+        comment="노래 결과 URL",
     )
 
-    song_result_url_2: Mapped[Optional[str]] = mapped_column(
+    language: Mapped[str] = mapped_column(
-        String(2048),
+        String(50),
-        nullable=True,
+        nullable=False,
-        comment="두 번째 노래 결과 URL",
+        default="Korean",
+        comment="출력 언어 (Korean, English, Chinese, Japanese, Thai, Vietnamese)",
     )
 
     created_at: Mapped[datetime] = mapped_column(

app/song/schemas/song_schema.py

@@ -15,7 +15,7 @@ class GenerateSongRequest(BaseModel):
     """노래 생성 요청 스키마
 
     Usage:
-        POST /song/generate
+        POST /song/generate/{task_id}
         Request body for generating a song via Suno API.
 
     Example Request:
@@ -51,29 +51,46 @@ class GenerateSongResponse(BaseModel):
     """노래 생성 응답 스키마
 
     Usage:
-        POST /song/generate
+        POST /song/generate/{task_id}
-        Returns the task ID for tracking song generation.
+        Returns the task IDs for tracking song generation.
 
-    Example Response:
+    Note:
+        실패 조건:
+        - task_id에 해당하는 Project가 없는 경우 (404 HTTPException)
+        - task_id에 해당하는 Lyric이 없는 경우 (404 HTTPException)
+        - Suno API 호출 실패
+
+    Example Response (Success):
         {
             "success": true,
-            "task_id": "abc123...",
+            "task_id": "019123ab-cdef-7890-abcd-ef1234567890",
-            "message": "노래 생성 요청이 접수되었습니다."
+            "suno_task_id": "abc123...",
+            "message": "노래 생성 요청이 접수되었습니다. suno_task_id로 상태를 조회하세요.",
+            "error_message": null
+        }
+
+    Example Response (Failure):
+        {
+            "success": false,
+            "task_id": "019123ab-cdef-7890-abcd-ef1234567890",
+            "suno_task_id": null,
+            "message": "노래 생성 요청에 실패했습니다.",
+            "error_message": "Suno API connection error"
         }
     """
 
     success: bool = Field(..., description="요청 성공 여부")
-    task_id: Optional[str] = Field(None, description="Suno 작업 ID")
+    task_id: Optional[str] = Field(None, description="내부 작업 ID (Project/Lyric task_id)")
+    suno_task_id: Optional[str] = Field(None, description="Suno API 작업 ID")
     message: str = Field(..., description="응답 메시지")
     error_message: Optional[str] = Field(None, description="에러 메시지 (실패 시)")
 
 
 class PollingSongRequest(BaseModel):
-    """노래 생성 상태 조회 요청 스키마
+    """노래 생성 상태 조회 요청 스키마 (Legacy)
 
-    Usage:
+    Note:
-        POST /song/polling
+        현재 사용되지 않음. GET /song/status/{suno_task_id} 엔드포인트 사용.
-        Request body for checking song generation status.
 
     Example Request:
         {
@@ -100,21 +117,66 @@ class PollingSongResponse(BaseModel):
     """노래 생성 상태 조회 응답 스키마
 
     Usage:
-        POST /song/polling 또는 GET /song/status/{task_id}
+        GET /song/status/{suno_task_id}
-        Returns the current status of song generation.
+        Suno API 작업 상태를 조회합니다.
 
-    Example Response:
+    Note:
+        상태 값:
+        - PENDING: 대기 중
+        - processing: 생성 중
+        - SUCCESS / TEXT_SUCCESS / complete: 생성 완료
+        - failed: 생성 실패
+        - error: API 조회 오류
+
+        SUCCESS 상태 시:
+        - 백그라운드에서 MP3 파일 다운로드 시작
+        - Song 테이블의 status를 completed로 업데이트
+        - song_result_url에 로컬 파일 경로 저장
+
+    Example Response (Processing):
         {
             "success": true,
-            "status": "complete",
+            "status": "processing",
+            "message": "노래를 생성하고 있습니다.",
+            "clips": null,
+            "raw_response": {...},
+            "error_message": null
+        }
+
+    Example Response (Success):
+        {
+            "success": true,
+            "status": "SUCCESS",
             "message": "노래 생성이 완료되었습니다.",
-            "clips": [...]
+            "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
+        }
+
+    Example Response (Failure):
+        {
+            "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, complete, failed)"
+        None, description="작업 상태 (PENDING, processing, SUCCESS, failed)"
     )
     message: str = Field(..., description="상태 메시지")
     clips: Optional[List[SongClipData]] = Field(None, description="생성된 노래 클립 목록")
@@ -127,21 +189,72 @@ class DownloadSongResponse(BaseModel):
 
     Usage:
         GET /song/download/{task_id}
-        Downloads the generated song and returns the local file path.
+        Polls for song completion and returns project info with song URL.
 
-    Example Response:
+    Note:
+        상태 값:
+        - processing: 노래 생성 진행 중 (song_result_url은 null)
+        - completed: 노래 생성 완료 (song_result_url 포함)
+        - failed: 노래 생성 실패
+        - not_found: task_id에 해당하는 Song 없음
+        - error: 조회 중 오류 발생
+
+    Example Response (Processing):
         {
             "success": true,
+            "status": "processing",
+            "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,
+            "status": "completed",
             "message": "노래 다운로드가 완료되었습니다.",
-            "file_path": "/media/2025-01-15/01234567-89ab-7def-0123-456789abcdef/song.mp3",
+            "store_name": "스테이 머뭄",
-            "file_url": "http://localhost:8000/media/2025-01-15/01234567-89ab-7def-0123-456789abcdef/song.mp3"
+            "region": "군산",
+            "detail_region_info": "군산 신흥동 말랭이 마을",
+            "task_id": "019123ab-cdef-7890-abcd-ef1234567890",
+            "language": "Korean",
+            "song_result_url": "http://localhost:8000/media/2025-01-15/스테이머뭄.mp3",
+            "created_at": "2025-01-15T12:00:00",
+            "error_message": null
+        }
+
+    Example Response (Not Found):
+        {
+            "success": false,
+            "status": "not_found",
+            "message": "task_id 'xxx'에 해당하는 Song을 찾을 수 없습니다.",
+            "store_name": null,
+            "region": null,
+            "detail_region_info": null,
+            "task_id": null,
+            "language": null,
+            "song_result_url": null,
+            "created_at": null,
+            "error_message": "Song not found"
         }
     """
 
     success: bool = Field(..., description="다운로드 성공 여부")
+    status: str = Field(..., description="처리 상태 (processing, completed, failed, not_found, error)")
     message: str = Field(..., description="응답 메시지")
-    file_path: Optional[str] = Field(None, description="저장된 파일 경로 (상대 경로)")
+    store_name: Optional[str] = Field(None, description="업체명")
-    file_url: Optional[str] = Field(None, description="파일 접근 URL")
+    region: Optional[str] = Field(None, description="지역명")
+    detail_region_info: Optional[str] = Field(None, description="상세 지역 정보")
+    task_id: Optional[str] = Field(None, description="작업 고유 식별자")
+    language: Optional[str] = Field(None, description="언어")
+    song_result_url: Optional[str] = Field(None, description="노래 결과 URL")
+    created_at: Optional[datetime] = Field(None, description="생성 일시")
     error_message: Optional[str] = Field(None, description="에러 메시지 (실패 시)")
 
 

app/song/worker/song_task.py

@@ -0,0 +1,101 @@
+"""
+Song Background Tasks
+
+노래 생성 관련 백그라운드 태스크를 정의합니다.
+"""
+
+from datetime import date
+from pathlib import Path
+
+import aiofiles
+import httpx
+from sqlalchemy import select
+
+from app.database.session import AsyncSessionLocal
+from app.song.models import Song
+from app.utils.common import generate_task_id
+from config import prj_settings
+
+
+async def download_and_save_song(
+    task_id: str,
+    audio_url: str,
+    store_name: str,
+) -> None:
+    """백그라운드에서 노래를 다운로드하고 Song 테이블을 업데이트합니다.
+
+    Args:
+        task_id: 프로젝트 task_id
+        audio_url: 다운로드할 오디오 URL
+        store_name: 저장할 파일명에 사용할 업체명
+    """
+    print(f"[download_and_save_song] START - task_id: {task_id}, store_name: {store_name}")
+    try:
+        # 저장 경로 생성: media/{날짜}/{uuid7}/{store_name}.mp3
+        today = date.today().isoformat()
+        unique_id = await generate_task_id()
+        # 파일명에 사용할 수 없는 문자 제거
+        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"
+
+        # 절대 경로 생성
+        media_dir = Path("media") / today / unique_id
+        media_dir.mkdir(parents=True, exist_ok=True)
+        file_path = media_dir / file_name
+        print(f"[download_and_save_song] Directory created - path: {file_path}")
+
+        # 오디오 파일 다운로드
+        print(f"[download_and_save_song] Downloading audio - task_id: {task_id}, url: {audio_url}")
+        async with httpx.AsyncClient() as client:
+            response = await client.get(audio_url, timeout=60.0)
+            response.raise_for_status()
+
+            async with aiofiles.open(str(file_path), "wb") as f:
+                await f.write(response.content)
+        print(f"[download_and_save_song] File saved - task_id: {task_id}, path: {file_path}")
+
+        # 프론트엔드에서 접근 가능한 URL 생성
+        relative_path = f"/media/{today}/{unique_id}/{file_name}"
+        base_url = f"http://{prj_settings.PROJECT_DOMAIN}"
+        file_url = f"{base_url}{relative_path}"
+        print(f"[download_and_save_song] URL generated - task_id: {task_id}, url: {file_url}")
+
+        # Song 테이블 업데이트 (새 세션 사용)
+        async with AsyncSessionLocal() as session:
+            # 여러 개 있을 경우 가장 최근 것 선택
+            result = await session.execute(
+                select(Song)
+                .where(Song.task_id == task_id)
+                .order_by(Song.created_at.desc())
+                .limit(1)
+            )
+            song = result.scalar_one_or_none()
+
+            if song:
+                song.status = "completed"
+                song.song_result_url = file_url
+                await session.commit()
+                print(f"[download_and_save_song] SUCCESS - task_id: {task_id}, status: completed")
+            else:
+                print(f"[download_and_save_song] Song NOT FOUND in DB - task_id: {task_id}")
+
+    except Exception as e:
+        print(f"[download_and_save_song] EXCEPTION - task_id: {task_id}, error: {e}")
+        # 실패 시 Song 테이블 업데이트
+        async with AsyncSessionLocal() as session:
+            # 여러 개 있을 경우 가장 최근 것 선택
+            result = await session.execute(
+                select(Song)
+                .where(Song.task_id == task_id)
+                .order_by(Song.created_at.desc())
+                .limit(1)
+            )
+            song = result.scalar_one_or_none()
+
+            if song:
+                song.status = "failed"
+                await session.commit()
+                print(f"[download_and_save_song] FAILED - task_id: {task_id}, status updated to failed")

app/utils/common.py

@@ -0,0 +1,58 @@
+"""
+Common Utility Functions
+
+공통으로 사용되는 유틸리티 함수들을 정의합니다.
+
+사용 예시:
+    from app.utils.common import generate_task_id
+
+    # task_id 생성
+    task_id = await generate_task_id(session=session, table_name=Project)
+
+Note:
+    페이지네이션 기능은 app.utils.pagination 모듈을 사용하세요:
+        from app.utils.pagination import PaginatedResponse, get_paginated
+"""
+
+from typing import Any, Optional, Type
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+from uuid_extensions import uuid7
+
+
+async def generate_task_id(
+    session: Optional[AsyncSession] = None,
+    table_name: Optional[Type[Any]] = None,
+) -> str:
+    """고유한 task_id를 생성합니다.
+
+    Args:
+        session: SQLAlchemy AsyncSession (optional)
+        table_name: task_id 컬럼이 있는 SQLAlchemy 테이블 클래스 (optional)
+
+    Returns:
+        str: 생성된 uuid7 문자열
+
+    Usage:
+        # 단순 uuid7 생성
+        task_id = await generate_task_id()
+
+        # 테이블에서 중복 검사 후 생성
+        task_id = await generate_task_id(session=session, table_name=Project)
+    """
+    task_id = str(uuid7())
+
+    if session is None or table_name is None:
+        return task_id
+
+    while True:
+        result = await session.execute(
+            select(table_name).where(table_name.task_id == task_id)
+        )
+        existing = result.scalar_one_or_none()
+
+        if existing is None:
+            return task_id
+
+        task_id = str(uuid7())

app/utils/pagination.py

@@ -0,0 +1,230 @@
+"""
+Pagination Module
+
+페이지네이션 관련 Pydantic 스키마와 유틸리티 함수를 정의합니다.
+
+사용 예시:
+    from app.utils.pagination import PaginatedResponse, get_paginated
+
+    # 라우터에서 response_model로 사용
+    @router.get("/items", response_model=PaginatedResponse[ItemModel])
+    async def get_items(
+        page: int = 1,
+        page_size: int = 20,
+        session: AsyncSession = Depends(get_session),
+    ):
+        return await get_paginated(
+            session=session,
+            model=Item,
+            item_schema=ItemModel,
+            page=page,
+            page_size=page_size,
+        )
+"""
+
+import math
+from typing import Any, Callable, Dict, Generic, List, Optional, Type, TypeVar
+
+from pydantic import BaseModel, Field
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+T = TypeVar("T")
+ModelT = TypeVar("ModelT")
+
+
+class PaginatedResponse(BaseModel, Generic[T]):
+    """페이지네이션 응답 스키마 (재사용 가능)
+
+    Usage:
+        다른 모델에서도 페이지네이션이 필요할 때 재사용 가능:
+        - PaginatedResponse[LyricListItem]
+        - PaginatedResponse[SongListItem]
+        - PaginatedResponse[VideoListItem]
+
+    Example:
+        from app.utils.pagination 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,
+        )
+
+
+async def get_paginated(
+    session: AsyncSession,
+    model: Type[ModelT],
+    item_schema: Type[T],
+    page: int = 1,
+    page_size: int = 20,
+    max_page_size: int = 100,
+    filters: Optional[Dict[str, Any]] = None,
+    order_by: Optional[str] = "created_at",
+    order_desc: bool = True,
+    transform_fn: Optional[Callable[[ModelT], T]] = None,
+) -> PaginatedResponse[T]:
+    """범용 페이지네이션 조회 함수
+
+    Args:
+        session: SQLAlchemy AsyncSession
+        model: SQLAlchemy 모델 클래스 (예: Lyric, Song, Video)
+        item_schema: Pydantic 스키마 클래스 (예: LyricListItem)
+        page: 페이지 번호 (1부터 시작, 기본값: 1)
+        page_size: 페이지당 데이터 수 (기본값: 20)
+        max_page_size: 최대 페이지 크기 (기본값: 100)
+        filters: 필터 조건 딕셔너리 (예: {"status": "completed"})
+        order_by: 정렬 기준 컬럼명 (기본값: "created_at")
+        order_desc: 내림차순 정렬 여부 (기본값: True)
+        transform_fn: 모델을 스키마로 변환하는 함수 (None이면 자동 변환)
+
+    Returns:
+        PaginatedResponse[T]: 페이지네이션된 응답
+
+    Usage:
+        # 기본 사용
+        result = await get_paginated(
+            session=session,
+            model=Lyric,
+            item_schema=LyricListItem,
+            page=1,
+            page_size=20,
+        )
+
+        # 필터링 사용
+        result = await get_paginated(
+            session=session,
+            model=Lyric,
+            item_schema=LyricListItem,
+            filters={"status": "completed"},
+        )
+
+        # 커스텀 변환 함수 사용
+        def transform(lyric: Lyric) -> LyricListItem:
+            return LyricListItem(
+                id=lyric.id,
+                task_id=lyric.task_id,
+                status=lyric.status,
+                lyric_result=lyric.lyric_result[:100] if lyric.lyric_result else None,
+                created_at=lyric.created_at,
+            )
+
+        result = await get_paginated(
+            session=session,
+            model=Lyric,
+            item_schema=LyricListItem,
+            transform_fn=transform,
+        )
+    """
+    # 페이지 크기 제한
+    page_size = min(page_size, max_page_size)
+    offset = (page - 1) * page_size
+
+    # 기본 쿼리
+    query = select(model)
+    count_query = select(func.count(model.id))
+
+    # 필터 적용
+    if filters:
+        for field, value in filters.items():
+            if value is not None and hasattr(model, field):
+                column = getattr(model, field)
+                query = query.where(column == value)
+                count_query = count_query.where(column == value)
+
+    # 전체 개수 조회
+    total_result = await session.execute(count_query)
+    total = total_result.scalar() or 0
+
+    # 정렬 적용
+    if order_by and hasattr(model, order_by):
+        order_column = getattr(model, order_by)
+        if order_desc:
+            query = query.order_by(order_column.desc())
+        else:
+            query = query.order_by(order_column.asc())
+
+    # 페이지네이션 적용
+    query = query.offset(offset).limit(page_size)
+
+    # 데이터 조회
+    result = await session.execute(query)
+    records = result.scalars().all()
+
+    # 페이지네이션 정보 계산
+    total_pages = math.ceil(total / page_size) if total > 0 else 1
+
+    # 스키마로 변환
+    if transform_fn:
+        items = [transform_fn(record) for record in records]
+    else:
+        # 자동 변환: 모델의 속성을 스키마 필드와 매칭
+        items = []
+        for record in records:
+            item_data = {}
+            for field_name in item_schema.model_fields.keys():
+                if hasattr(record, field_name):
+                    item_data[field_name] = getattr(record, field_name)
+            items.append(item_schema(**item_data))
+
+    return PaginatedResponse[T](
+        items=items,
+        total=total,
+        page=page,
+        page_size=page_size,
+        total_pages=total_pages,
+        has_next=page < total_pages,
+        has_prev=page > 1,
+    )

app/utils/suno.py

@@ -22,6 +22,9 @@ task_id = await suno.generate(
 
 # 상태 확인 (폴링 방식)
 result = await suno.get_task_status(task_id)
+
+# 상태 응답 파싱
+parsed = suno.parse_status_response(result)
 ```
 
 ## 콜백 URL 사용법
@@ -52,11 +55,12 @@ generate() 호출 시 callback_url 파라미터를 전달하면 생성 완료
 - 동일 task_id에 대해 여러 콜백 수신 가능 (멱등성 처리 필요)
 """
 
-from typing import Any
+from typing import Any, List, Optional
 
 import httpx
 
 from config import apikey_settings
+from app.song.schemas.song_schema import PollingSongResponse, SongClipData
 
 
 class SunoService:
@@ -175,3 +179,83 @@ class SunoService:
                 raise ValueError("Suno API returned empty response for task status")
 
             return data
+
+    def parse_status_response(self, result: dict | None) -> PollingSongResponse:
+        """Suno API 상태 응답을 파싱하여 PollingSongResponse로 변환합니다.
+
+        Args:
+            result: get_task_status()에서 반환된 원본 응답
+
+        Returns:
+            PollingSongResponse: 파싱된 상태 응답
+
+        Note:
+            응답 구조:
+            - PENDING 상태: data.response가 null, data.status가 "PENDING"
+            - SUCCESS 상태: data.response.sunoData에 클립 데이터 배열, data.status가 "SUCCESS"
+        """
+        if result is None:
+            return PollingSongResponse(
+                success=False,
+                status="error",
+                message="Suno API 응답이 비어있습니다.",
+                clips=None,
+                raw_response=None,
+                error_message="Suno API returned None response",
+            )
+
+        code = result.get("code", 0)
+        data = result.get("data", {})
+
+        if code != 200:
+            return PollingSongResponse(
+                success=False,
+                status="failed",
+                message="Suno API 응답 오류",
+                clips=None,
+                raw_response=result,
+                error_message=result.get("msg", "Unknown error"),
+            )
+
+        # status는 data.status에 있음 (PENDING, SUCCESS 등)
+        status = data.get("status", "unknown")
+
+        # 클립 데이터는 data.response.sunoData에 있음 (camelCase)
+        # PENDING 상태에서는 response가 null
+        response_data = data.get("response") or {}
+        clips_data = response_data.get("sunoData") or []
+
+        # 상태별 메시지
+        status_messages = {
+            "PENDING": "노래 생성 대기 중입니다.",
+            "processing": "노래를 생성하고 있습니다.",
+            "complete": "노래 생성이 완료되었습니다.",
+            "SUCCESS": "노래 생성이 완료되었습니다.",
+            "TEXT_SUCCESS": "노래 생성이 완료되었습니다.",
+            "failed": "노래 생성에 실패했습니다.",
+        }
+
+        # 클립 데이터 파싱 (Suno API는 camelCase 사용)
+        clips = None
+        if clips_data:
+            clips = [
+                SongClipData(
+                    id=clip.get("id"),
+                    audio_url=clip.get("audioUrl"),
+                    stream_audio_url=clip.get("streamAudioUrl"),
+                    image_url=clip.get("imageUrl"),
+                    title=clip.get("title"),
+                    status=clip.get("status"),
+                    duration=clip.get("duration"),
+                )
+                for clip in clips_data
+            ]
+
+        return PollingSongResponse(
+            success=True,
+            status=status,
+            message=status_messages.get(status, f"상태: {status}"),
+            clips=clips,
+            raw_response=result,
+            error_message=None,
+        )

app/video/api/routers/v1/router.py

@@ -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.lyrics.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

app/video/api/routers/v1/video.py

@@ -0,0 +1,108 @@
+"""
+Video API Endpoints (Test)
+
+프론트엔드 개발을 위한 테스트용 엔드포인트입니다.
+"""
+
+from datetime import datetime, timedelta
+from typing import Optional
+
+from fastapi import APIRouter
+from pydantic import BaseModel, Field
+
+router = APIRouter(prefix="/video", tags=["video"])
+
+# =============================================================================
+# Schemas
+# =============================================================================
+
+
+class VideoGenerateResponse(BaseModel):
+    """영상 생성 응답 스키마"""
+
+    success: bool = Field(..., description="성공 여부")
+    task_id: str = Field(..., description="작업 고유 식별자")
+    message: str = Field(..., description="응답 메시지")
+    error_message: Optional[str] = Field(None, description="에러 메시지")
+
+
+class VideoStatusResponse(BaseModel):
+    """영상 상태 조회 응답 스키마"""
+
+    task_id: str = Field(..., description="작업 고유 식별자")
+    status: str = Field(..., description="처리 상태 (processing, completed, failed)")
+    video_url: Optional[str] = Field(None, description="영상 URL")
+
+
+class VideoItem(BaseModel):
+    """영상 아이템 스키마"""
+
+    task_id: str = Field(..., description="작업 고유 식별자")
+    video_url: str = Field(..., description="영상 URL")
+    created_at: datetime = Field(..., description="생성 일시")
+
+
+class VideoListResponse(BaseModel):
+    """영상 목록 응답 스키마"""
+
+    videos: list[VideoItem] = Field(..., description="영상 목록")
+    total: int = Field(..., description="전체 개수")
+
+
+# =============================================================================
+# Test Endpoints
+# =============================================================================
+
+TEST_VIDEO_URL = "https://ado2mediastoragepublic.blob.core.windows.net/ado2-media-public-access/ado2-media-original/dev-user-idx/dev-task-idx/1a584e86-6a74-417d-8cff-270ef60c8646.mp4"
+
+
+@router.post(
+    "/generate/{task_id}",
+    summary="영상 생성 요청 (테스트)",
+    response_model=VideoGenerateResponse,
+)
+async def generate_video(task_id: str) -> VideoGenerateResponse:
+    """영상 생성 요청 테스트 엔드포인트"""
+    return VideoGenerateResponse(
+        success=True,
+        task_id=task_id,
+        message="영상 생성 요청 성공",
+        error_message=None,
+    )
+
+
+@router.get(
+    "/status/{task_id}",
+    summary="영상 상태 조회 (테스트)",
+    response_model=VideoStatusResponse,
+)
+async def get_video_status(task_id: str) -> VideoStatusResponse:
+    """영상 상태 조회 테스트 엔드포인트"""
+    return VideoStatusResponse(
+        task_id=task_id,
+        status="completed",
+        video_url=TEST_VIDEO_URL,
+    )
+
+
+@router.get(
+    "s/",
+    summary="영상 목록 조회 (테스트)",
+    response_model=VideoListResponse,
+)
+async def get_videos() -> VideoListResponse:
+    """영상 목록 조회 테스트 엔드포인트"""
+    now = datetime.now()
+    videos = [
+        VideoItem(
+            task_id=f"test-task-id-{i:03d}",
+            video_url=TEST_VIDEO_URL,
+            created_at=now - timedelta(hours=i),
+        )
+        for i in range(10)
+    ]
+
+    return VideoListResponse(
+        videos=videos,
+        total=len(videos),
+    )