first commit

.gitignore

@@ -0,0 +1,15 @@
+# Python virtual environment
+.venv/
+
+# Python bytecode cache
+__pycache__/
+
+# Environment variables
+.env
+
+# Claude AI related files
+.claude/
+.claudeignore
+
+# VSCode settings
+.vscode/

.python-version

@@ -0,0 +1 @@
+3.13

app/admin_manager.py

@@ -0,0 +1,34 @@
+from fastapi import FastAPI
+from sqladmin import Admin
+
+from app.database.session import engine
+from app.lyrics.api.lyrics_admin import (
+    LyricsAttributeAdmin,
+    LyricsPromptTemplateAdmin,
+    LyricsSongResultsAllAdmin,
+    LyricsSongSampleAdmin,
+    LyricsStoreDefaultInfoAdmin,
+)
+from config import prj_settings
+
+# https://github.com/aminalaee/sqladmin
+
+
+def init_admin(
+    app: FastAPI,
+    db_engine: engine,
+    base_url: str = prj_settings.ADMIN_BASE_URL,
+) -> Admin:
+    admin = Admin(
+        app,
+        db_engine,
+        base_url=base_url,
+    )
+
+    admin.add_view(LyricsStoreDefaultInfoAdmin)
+    admin.add_view(LyricsAttributeAdmin)
+    admin.add_view(LyricsSongSampleAdmin)
+    admin.add_view(LyricsPromptTemplateAdmin)
+    admin.add_view(LyricsSongResultsAllAdmin)
+
+    return admin

app/core/common.py

@@ -0,0 +1,41 @@
+# app/main.py
+import asyncio
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """FastAPI 애플리케이션 생명주기 관리"""
+    # Startup - 애플리케이션 시작 시
+    print("Starting up...")
+
+    try:
+        pass
+        # # 데이터베이스 테이블 생성
+        # from app.database.session import create_db_tables
+
+        # await create_db_tables()
+        # print("Database tables ready")
+    except asyncio.TimeoutError:
+        print("Database initialization timed out")
+        # 타임아웃 시 앱 시작 중단하려면 raise, 계속하려면 pass
+        raise
+    except Exception as e:
+        print(f"Database initialization failed: {e}")
+        # 에러 시 앱 시작 중단하려면 raise, 계속하려면 pass
+        raise
+
+    yield  # 애플리케이션 실행 중
+
+    # Shutdown - 애플리케이션 종료 시
+    print("Shutting down...")
+    from app.database.session import engine
+
+    await engine.dispose()
+    print("Database engine disposed")
+
+
+# FastAPI 앱 생성 (lifespan 적용)
+app = FastAPI(title="CastAD", lifespan=lifespan)

app/core/exceptions.py

@@ -0,0 +1,114 @@
+from fastapi import FastAPI, HTTPException, Request, Response, status
+from fastapi.responses import JSONResponse
+
+
+class FastShipError(Exception):
+    """Base exception for all exceptions in fastship api"""
+    # status_code to be returned for this exception
+    # when it is handled
+    status = status.HTTP_400_BAD_REQUEST
+
+
+class EntityNotFound(FastShipError):
+    """Entity not found in database"""
+
+    status = status.HTTP_404_NOT_FOUND
+
+
+class BadPassword(FastShipError):
+    """Password is not strong enough or invalid"""
+
+    status = status.HTTP_400_BAD_REQUEST
+
+
+class ClientNotAuthorized(FastShipError):
+    """Client is not authorized to perform the action"""
+
+    status = status.HTTP_401_UNAUTHORIZED
+
+
+class ClientNotVerified(FastShipError):
+    """Client is not verified"""
+
+    status = status.HTTP_401_UNAUTHORIZED
+
+
+class NothingToUpdate(FastShipError):
+    """No data provided to update"""
+
+
+class BadCredentials(FastShipError):
+    """User email or password is incorrect"""
+
+    status = status.HTTP_401_UNAUTHORIZED
+
+
+class InvalidToken(FastShipError):
+    """Access token is invalid or expired"""
+
+    status = status.HTTP_401_UNAUTHORIZED
+
+
+class DeliveryPartnerNotAvailable(FastShipError):
+    """Delivery partner/s do not service the destination"""
+
+    status = status.HTTP_406_NOT_ACCEPTABLE
+
+
+class DeliveryPartnerCapacityExceeded(FastShipError):
+    """Delivery partner has reached their max handling capacity"""
+
+    status = status.HTTP_406_NOT_ACCEPTABLE
+
+
+def _get_handler(status: int, detail: str):
+    # Define
+    def handler(request: Request, exception: Exception) -> Response:
+        # DEBUG PRINT STATEMENT 👇
+        from rich import print, panel
+        print(
+            panel.Panel(
+                exception.__class__.__name__,
+                title="Handled Exception",
+                border_style="red",
+            ),
+        )
+        # DEBUG PRINT STATEMENT 👆
+        
+        # Raise HTTPException with given status and detail
+        # can return JSONResponse as well
+        raise HTTPException(
+            status_code=status,
+            detail=detail,
+        )
+    # Return ExceptionHandler required with given
+    # status and detail for HTTPExcetion above
+    return handler
+
+
+def add_exception_handlers(app: FastAPI):
+    # Get all subclass of 👇, our custom exceptions
+    exception_classes = FastShipError.__subclasses__()
+
+    for exception_class in exception_classes:
+        # Add exception handler
+        app.add_exception_handler(
+            # Custom exception class
+            exception_class,
+            # Get handler function
+            _get_handler(
+                status=exception_class.status,
+                detail=exception_class.__doc__,
+            ),
+        )
+
+    @app.exception_handler(status.HTTP_500_INTERNAL_SERVER_ERROR)
+    def internal_server_error_handler(request, exception):
+        return JSONResponse(
+            content={"detail": "Something went wrong..."},
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            headers={
+                "X-Error": f"{exception}",
+            }
+        )
+    

app/database/redis.py

@@ -0,0 +1,30 @@
+from uuid import UUID
+from redis.asyncio import Redis
+
+from app.config import db_settings
+
+
+_token_blacklist = Redis(
+    host=db_settings.REDIS_HOST,
+    port=db_settings.REDIS_PORT,
+    db=0,
+)
+_shipment_verification_codes = Redis(
+    host=db_settings.REDIS_HOST,
+    port=db_settings.REDIS_PORT,
+    db=1,
+    decode_responses=True,
+)
+
+async def add_jti_to_blacklist(jti: str):
+    await _token_blacklist.set(jti, "blacklisted")
+
+
+async def is_jti_blacklisted(jti: str) -> bool:
+    return await _token_blacklist.exists(jti)
+
+async def add_shipment_verification_code(id: UUID, code: int):
+    await _shipment_verification_codes.set(str(id), code)
+
+async def get_shipment_verification_code(id: UUID) -> str:
+    return str(await _shipment_verification_codes.get(str(id)))

app/database/session-prod.py

@@ -0,0 +1,97 @@
+from asyncio import current_task
+from typing import AsyncGenerator
+
+from sqlalchemy.ext.asyncio import (
+    AsyncSession,
+    async_sessionmaker,
+    create_async_engine,
+)
+from sqlalchemy.orm import DeclarativeBase
+from sqlalchemy.pool import AsyncQueuePool  # 비동기 풀 클래스
+
+from config import db_settings
+
+
+# Base 클래스 정의
+class Base(DeclarativeBase):
+    pass
+
+
+engine = create_async_engine(
+    # MySQL async URL (asyncmy 드라이버)
+    url=db_settings.MYSQL_URL,  # 예: "mysql+asyncmy://test:test@host:3306/poc"
+    # === Connection Pool 설정 ===
+    pool_size=10,  # 기본 풀 크기: 10개 연결 유지
+    max_overflow=10,  # 최대 증가: 10개 (총 20개까지 가능)
+    poolclass=AsyncQueuePool,  # 비동기 큐 풀 사용 (기본값, 명시적 지정)
+    pool_timeout=30,  # 풀에서 연결 대기 시간: 30초 (기본 30초)
+    pool_recycle=3600,  # 연결 재사용 주기: 1시간 (기본 3600초)
+    pool_pre_ping=True,  # 연결 사용 전 유효성 검사: True로 설정
+    pool_reset_on_return="rollback",  # 연결 반환 시 자동 롤백
+    # === MySQL 특화 설정 ===
+    echo=False,  # SQL 쿼리 로깅 (디버깅 시 True)
+    # === 연결 타임아웃 및 재시도 ===
+    connect_args={
+        "connect_timeout": 10,  # MySQL 연결 타임아웃: 10초
+        "read_timeout": 30,  # 읽기 타임아웃: 30초
+        "write_timeout": 30,  # 쓰기 타임아웃: 30초
+        "charset": "utf8mb4",  # 문자셋 (이모지 지원)
+        "sql_mode": "STRICT_TRANS_TABLES,NO_ZERO_DATE,NO_ZERO_IN_DATE",
+        "init_command": "SET SESSION time_zone = '+00:00'",  # 초기 연결 시 실행
+    },
+)
+
+# Async 세션 팩토리 생성
+async_session_factory = async_sessionmaker(
+    bind=engine,
+    class_=AsyncSession,
+    expire_on_commit=False,  # 커밋 후 객체 상태 유지
+    autoflush=True,  # 변경 감지 자동 플러시
+)
+
+# async_scoped_session 생성
+AsyncScopedSession = async_session_factory(
+    async_session_factory,
+    scopefunc=current_task,
+)
+
+
+# 테이블 생성 함수
+async def create_db_tables() -> None:
+    async with engine.begin() as conn:
+        # from app.database.models import Shipment, Seller  # noqa: F401
+        await conn.run_sync(Base.metadata.create_all)
+        print("MySQL tables created successfully")
+
+
+# 세션 제너레이터 (FastAPI Depends에 사용)
+async def get_session() -> AsyncGenerator[AsyncSession, None]:
+    """
+    Async 세션 컨텍스트 매니저
+    - FastAPI dependency로 사용
+    - Connection Pool에서 연결 획득/반환 자동 관리
+    """
+    async with async_session_factory() as session:
+        # pre-commit 훅 (선택적: 트랜잭션 시작 전 실행)
+        # await session.begin()  # async_sessionmaker에서 자동 begin
+
+        try:
+            yield session
+            # FastAPI 요청 완료 시 자동 commit (예외 발생 시 rollback)
+        except Exception as e:
+            await session.rollback()  # 명시적 롤백 (선택적)
+            print(f"Session rollback due to: {e}")  # 로깅
+            raise
+        finally:
+            # 명시적 세션 종료 (Connection Pool에 반환)
+            # context manager가 자동 처리하지만, 명시적으로 유지
+            await session.close()
+            print("session closed successfully")
+            # 또는 session.aclose() - Python 3.10+
+
+
+# 애플리케이션 종료 시 엔진 정리 (선택적)
+async def dispose_engine() -> None:
+    """애플리케이션 종료 시 모든 연결 해제"""
+    await engine.dispose()
+    print("Database engine disposed")

app/database/session.py

@@ -0,0 +1,75 @@
+from typing import AsyncGenerator
+
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+from sqlalchemy.orm import DeclarativeBase
+
+from config import db_settings
+
+
+class Base(DeclarativeBase):
+    pass
+
+
+# 데이터베이스 엔진 생성
+engine = create_async_engine(
+    url=db_settings.MYSQL_URL,
+    echo=False,
+    pool_size=10,
+    max_overflow=10,
+    pool_timeout=5,
+    pool_recycle=3600,
+    pool_pre_ping=True,
+    pool_reset_on_return="rollback",
+    connect_args={
+        "connect_timeout": 3,
+        "charset": "utf8mb4",
+        # "allow_public_key_retrieval": True,
+    },
+)
+
+# Async sessionmaker 생성
+AsyncSessionLocal = async_sessionmaker(
+    bind=engine,
+    class_=AsyncSession,
+    expire_on_commit=False,
+    autoflush=False,  # 명시적 flush 권장
+)
+
+
+async def create_db_tables():
+    async with engine.begin() as connection:
+        from app.lyrics.models import (  # noqa: F401
+            Attribute,
+            PromptTemplate,
+            SongResultsAll,
+            SongSample,
+            StoreDefaultInfo,
+        )
+
+        print("Creating database tables...")
+
+        import asyncio
+
+        async with asyncio.timeout(10):
+            async with engine.begin() as connection:
+                await connection.run_sync(Base.metadata.create_all)
+
+
+# FastAPI 의존성용 세션 제너레이터
+async def get_session() -> AsyncGenerator[AsyncSession, None]:
+    async with AsyncSessionLocal() as session:
+        try:
+            yield session
+            # print("Session commited")
+            # await session.commit()
+        except Exception as e:
+            await session.rollback()
+            print(f"Session rollback due to: {e}")
+            raise e
+    # async with 종료 시 session.close()가 자동 호출됨
+
+
+# 앱 종료 시 엔진 리소스 정리 함수
+async def dispose_engine() -> None:
+    await engine.dispose()
+    print("Database engine disposed")

app/home/api/routers/v1/__init__.py

@@ -0,0 +1,15 @@
+"""API 1 Version Router Module."""
+
+# from fastapi import APIRouter, Depends
+
+# API 버전 1 라우터를 정의합니다.
+# router = APIRouter(
+#     prefix="/api/v1",
+#     dependencies=[Depends(check_use_api), Depends(set_current_connect)],
+# )
+# router = APIRouter(
+#     prefix="/api/v1",
+#     dependencies=[Depends(check_use_api), Depends(set_current_connect)],
+# )
+# router.include_router(auth.router, tags=[Tags.AUTH])
+# router.include_router(board.router, prefix="/boards", tags=[Tags.BOARD])

app/home/api/routers/v1/home.py

@@ -0,0 +1,463 @@
+from datetime import datetime
+from typing import Optional
+
+from fastapi import APIRouter, Body, Depends, status
+from pydantic import BaseModel, EmailStr, Field
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.database.session import get_session
+
+router = APIRouter()
+
+
+# ============================================================
+# Pydantic Models for Request/Response
+# ============================================================
+
+
+class SigninRequest(BaseModel):
+    """회원가입 요청 스키마"""
+
+    email: EmailStr = Field(
+        ..., description="사용자 이메일 주소", example="user@example.com"
+    )
+    password: str = Field(
+        ..., min_length=8, description="비밀번호 (최소 8자)", example="password123"
+    )
+    name: str = Field(
+        ..., min_length=2, max_length=50, description="사용자 이름", example="홍길동"
+    )
+    phone: Optional[str] = Field(
+        None,
+        pattern=r"^\d{3}-\d{4}-\d{4}$",
+        description="전화번호 (형식: 010-1234-5678)",
+        example="010-1234-5678",
+    )
+
+
+class SigninResponse(BaseModel):
+    """회원가입 응답 스키마"""
+
+    success: bool = Field(..., description="요청 성공 여부")
+    message: str = Field(..., description="응답 메시지")
+    user_id: int = Field(..., description="생성된 사용자 ID")
+    email: EmailStr = Field(..., description="등록된 이메일")
+    created_at: datetime = Field(..., description="계정 생성 시간")
+
+
+class LoginRequest(BaseModel):
+    """로그인 요청 스키마"""
+
+    email: EmailStr = Field(
+        ..., description="사용자 이메일 주소", example="user@example.com"
+    )
+    password: str = Field(..., description="비밀번호", example="password123")
+
+
+class LoginResponse(BaseModel):
+    """로그인 응답 스키마"""
+
+    success: bool = Field(..., description="로그인 성공 여부")
+    message: str = Field(..., description="응답 메시지")
+    access_token: str = Field(..., description="JWT 액세스 토큰")
+    refresh_token: str = Field(..., description="JWT 리프레시 토큰")
+    token_type: str = Field(default="bearer", description="토큰 타입")
+    expires_in: int = Field(..., description="토큰 만료 시간 (초)")
+
+
+class LogoutResponse(BaseModel):
+    """로그아웃 응답 스키마"""
+
+    success: bool = Field(..., description="로그아웃 성공 여부")
+    message: str = Field(..., description="응답 메시지")
+
+
+class ProfileResponse(BaseModel):
+    """프로필 조회 응답 스키마"""
+
+    user_id: int = Field(..., description="사용자 ID")
+    email: EmailStr = Field(..., description="이메일 주소")
+    name: str = Field(..., description="사용자 이름")
+    phone: Optional[str] = Field(None, description="전화번호")
+    profile_image: Optional[str] = Field(None, description="프로필 이미지 URL")
+    created_at: datetime = Field(..., description="계정 생성 시간")
+    last_login: Optional[datetime] = Field(None, description="마지막 로그인 시간")
+
+
+class HomeResponse(BaseModel):
+    """홈 응답 스키마"""
+
+    message: str = Field(..., description="환영 메시지")
+    version: str = Field(..., description="API 버전")
+    status: str = Field(..., description="서비스 상태")
+    timestamp: datetime = Field(..., description="응답 시간")
+
+
+class ErrorResponse(BaseModel):
+    """에러 응답 스키마"""
+
+    success: bool = Field(default=False, description="요청 성공 여부")
+    error_code: str = Field(..., description="에러 코드")
+    message: str = Field(..., description="에러 메시지")
+    detail: Optional[str] = Field(None, description="상세 에러 정보")
+
+
+# ============================================================
+# Dummy Data
+# ============================================================
+
+DUMMY_USER = {
+    "user_id": 1,
+    "email": "user@example.com",
+    "name": "홍길동",
+    "phone": "010-1234-5678",
+    "profile_image": "https://example.com/images/profile/default.png",
+    "created_at": datetime(2024, 1, 15, 10, 30, 0),
+    "last_login": datetime(2024, 12, 18, 9, 0, 0),
+}
+
+DUMMY_TOKENS = {
+    "access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxIiwiZXhwIjoxNzM0NTAwMDAwfQ.dummy_signature",
+    "refresh_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxIiwiZXhwIjoxNzM1MDAwMDAwfQ.dummy_refresh",
+    "token_type": "bearer",
+    "expires_in": 3600,
+}
+
+
+# ============================================================
+# Endpoints
+# ============================================================
+
+
+@router.get(
+    "/db",
+    summary="데이터베이스 상태 확인",
+    description="데이터베이스 연결 상태를 확인합니다. 간단한 쿼리를 실행하여 DB 연결이 정상인지 테스트합니다.",
+    response_description="데이터베이스 연결 상태 정보",
+    responses={
+        200: {
+            "description": "데이터베이스 연결 정상",
+            "content": {
+                "application/json": {
+                    "example": {
+                        "status": "healthy",
+                        "database": "connected",
+                        "test_query": 1,
+                    }
+                }
+            },
+        },
+        500: {"description": "데이터베이스 연결 실패", "model": ErrorResponse},
+    },
+    tags=["health"],
+)
+async def db_health_check(session: AsyncSession = Depends(get_session)):
+    """DB 연결 상태 확인"""
+    try:
+        result = await session.execute(text("SELECT 1"))
+        return {
+            "status": "healthy",
+            "database": "connected",
+            "test_query": result.scalar(),
+        }
+    except Exception as e:
+        return {"status": "unhealthy", "database": "disconnected", "error": str(e)}
+
+
+@router.get(
+    "/",
+    summary="홈 엔드포인트",
+    description="API 서비스의 기본 정보를 반환합니다. 서비스 상태, 버전, 현재 시간 등의 정보를 확인할 수 있습니다.",
+    response_model=HomeResponse,
+    response_description="서비스 기본 정보",
+    responses={
+        200: {
+            "description": "성공적으로 홈 정보 반환",
+            "content": {
+                "application/json": {
+                    "example": {
+                        "message": "CASTAD API 서비스에 오신 것을 환영합니다.",
+                        "version": "0.1.0",
+                        "status": "running",
+                        "timestamp": "2024-12-18T10:00:00",
+                    }
+                }
+            },
+        }
+    },
+    tags=["home"],
+)
+async def home() -> HomeResponse:
+    """홈 페이지 - API 기본 정보 반환"""
+    return HomeResponse(
+        message="CASTAD API 서비스에 오신 것을 환영합니다.",
+        version="0.1.0",
+        status="running",
+        timestamp=datetime.now(),
+    )
+
+
+@router.post(
+    "/signin",
+    summary="회원가입",
+    description="""
+새로운 사용자 계정을 생성합니다.
+
+## 요청 필드
+- **email**: 유효한 이메일 주소 (필수)
+- **password**: 최소 8자 이상의 비밀번호 (필수)
+- **name**: 2~50자 사이의 사용자 이름 (필수)
+- **phone**: 전화번호 (선택, 형식: 010-1234-5678)
+
+## 비밀번호 정책
+- 최소 8자 이상
+- 영문, 숫자 조합 권장
+    """,
+    response_model=SigninResponse,
+    response_description="회원가입 결과",
+    status_code=status.HTTP_201_CREATED,
+    responses={
+        201: {"description": "회원가입 성공", "model": SigninResponse},
+        400: {
+            "description": "잘못된 요청 (유효성 검사 실패)",
+            "model": ErrorResponse,
+            "content": {
+                "application/json": {
+                    "example": {
+                        "success": False,
+                        "error_code": "VALIDATION_ERROR",
+                        "message": "입력값이 유효하지 않습니다.",
+                        "detail": "이메일 형식이 올바르지 않습니다.",
+                    }
+                }
+            },
+        },
+        409: {
+            "description": "이미 존재하는 이메일",
+            "model": ErrorResponse,
+            "content": {
+                "application/json": {
+                    "example": {
+                        "success": False,
+                        "error_code": "EMAIL_EXISTS",
+                        "message": "이미 등록된 이메일입니다.",
+                        "detail": None,
+                    }
+                }
+            },
+        },
+    },
+    tags=["auth"],
+)
+async def signin(
+    request_body: SigninRequest = Body(
+        ...,
+        description="회원가입에 필요한 사용자 정보",
+        openapi_examples={
+            "기본 예시": {
+                "summary": "필수 필드만 입력",
+                "description": "이메일, 비밀번호, 이름만 입력하는 경우",
+                "value": {
+                    "email": "newuser@example.com",
+                    "password": "securepass123",
+                    "name": "김철수",
+                },
+            },
+            "전체 필드 예시": {
+                "summary": "모든 필드 입력",
+                "description": "선택 필드를 포함한 전체 입력",
+                "value": {
+                    "email": "newuser@example.com",
+                    "password": "securepass123",
+                    "name": "김철수",
+                    "phone": "010-9876-5432",
+                },
+            },
+        },
+    ),
+) -> SigninResponse:
+    """새로운 사용자 회원가입 처리"""
+    return SigninResponse(
+        success=True,
+        message="회원가입이 완료되었습니다.",
+        user_id=2,
+        email=request_body.email,
+        created_at=datetime.now(),
+    )
+
+
+@router.post(
+    "/login",
+    summary="로그인",
+    description="""
+사용자 인증을 수행하고 JWT 토큰을 발급합니다.
+
+## 인증 방식
+이메일과 비밀번호를 사용한 기본 인증을 수행합니다.
+인증 성공 시 액세스 토큰과 리프레시 토큰이 발급됩니다.
+
+## 토큰 정보
+- **access_token**: API 요청 시 사용 (유효기간: 1시간)
+- **refresh_token**: 액세스 토큰 갱신 시 사용 (유효기간: 7일)
+    """,
+    response_model=LoginResponse,
+    response_description="로그인 결과 및 토큰 정보",
+    responses={
+        200: {"description": "로그인 성공", "model": LoginResponse},
+        401: {
+            "description": "인증 실패",
+            "model": ErrorResponse,
+            "content": {
+                "application/json": {
+                    "example": {
+                        "success": False,
+                        "error_code": "INVALID_CREDENTIALS",
+                        "message": "이메일 또는 비밀번호가 올바르지 않습니다.",
+                        "detail": None,
+                    }
+                }
+            },
+        },
+        403: {
+            "description": "계정 비활성화",
+            "model": ErrorResponse,
+            "content": {
+                "application/json": {
+                    "example": {
+                        "success": False,
+                        "error_code": "ACCOUNT_DISABLED",
+                        "message": "비활성화된 계정입니다.",
+                        "detail": "관리자에게 문의하세요.",
+                    }
+                }
+            },
+        },
+    },
+    tags=["auth"],
+)
+async def login(
+    request_body: LoginRequest = Body(
+        ...,
+        description="로그인 인증 정보",
+        openapi_examples={
+            "로그인 예시": {
+                "summary": "일반 로그인",
+                "description": "이메일과 비밀번호로 로그인",
+                "value": {"email": "user@example.com", "password": "password123"},
+            }
+        },
+    ),
+) -> LoginResponse:
+    """사용자 로그인 및 토큰 발급"""
+    return LoginResponse(
+        success=True,
+        message="로그인에 성공했습니다.",
+        access_token=DUMMY_TOKENS["access_token"],
+        refresh_token=DUMMY_TOKENS["refresh_token"],
+        token_type=DUMMY_TOKENS["token_type"],
+        expires_in=DUMMY_TOKENS["expires_in"],
+    )
+
+
+@router.post(
+    "/logout",
+    summary="로그아웃",
+    description="""
+현재 세션을 종료하고 토큰을 무효화합니다.
+
+## 동작 방식
+- 서버 측 토큰 블랙리스트에 현재 토큰 등록
+- 클라이언트 측 토큰 삭제 권장
+
+## 주의사항
+로그아웃 후에는 동일한 토큰으로 API 요청이 불가능합니다.
+    """,
+    response_model=LogoutResponse,
+    response_description="로그아웃 결과",
+    responses={
+        200: {"description": "로그아웃 성공", "model": LogoutResponse},
+        401: {
+            "description": "인증되지 않은 요청",
+            "model": ErrorResponse,
+            "content": {
+                "application/json": {
+                    "example": {
+                        "success": False,
+                        "error_code": "UNAUTHORIZED",
+                        "message": "인증이 필요합니다.",
+                        "detail": "유효한 토큰을 제공해주세요.",
+                    }
+                }
+            },
+        },
+    },
+    tags=["auth"],
+)
+async def logout() -> LogoutResponse:
+    """사용자 로그아웃 처리"""
+    return LogoutResponse(success=True, message="로그아웃되었습니다.")
+
+
+@router.get(
+    "/profile",
+    summary="프로필 조회",
+    description="""
+현재 로그인한 사용자의 프로필 정보를 조회합니다.
+
+## 반환 정보
+- 기본 정보: 사용자 ID, 이메일, 이름
+- 연락처 정보: 전화번호
+- 프로필 이미지 URL
+- 계정 정보: 생성일, 마지막 로그인 시간
+
+## 인증 필요
+이 엔드포인트는 유효한 액세스 토큰이 필요합니다.
+Authorization 헤더에 Bearer 토큰을 포함해주세요.
+    """,
+    response_model=ProfileResponse,
+    response_description="사용자 프로필 정보",
+    responses={
+        200: {"description": "프로필 조회 성공", "model": ProfileResponse},
+        401: {
+            "description": "인증되지 않은 요청",
+            "model": ErrorResponse,
+            "content": {
+                "application/json": {
+                    "example": {
+                        "success": False,
+                        "error_code": "UNAUTHORIZED",
+                        "message": "인증이 필요합니다.",
+                        "detail": "유효한 토큰을 제공해주세요.",
+                    }
+                }
+            },
+        },
+        404: {
+            "description": "사용자를 찾을 수 없음",
+            "model": ErrorResponse,
+            "content": {
+                "application/json": {
+                    "example": {
+                        "success": False,
+                        "error_code": "USER_NOT_FOUND",
+                        "message": "사용자를 찾을 수 없습니다.",
+                        "detail": None,
+                    }
+                }
+            },
+        },
+    },
+    tags=["user"],
+)
+async def profile() -> ProfileResponse:
+    """현재 사용자 프로필 조회"""
+    return ProfileResponse(
+        user_id=DUMMY_USER["user_id"],
+        email=DUMMY_USER["email"],
+        name=DUMMY_USER["name"],
+        phone=DUMMY_USER["phone"],
+        profile_image=DUMMY_USER["profile_image"],
+        created_at=DUMMY_USER["created_at"],
+        last_login=DUMMY_USER["last_login"],
+    )

app/home/services/base.py

@@ -0,0 +1,24 @@
+from uuid import UUID
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlmodel import SQLModel
+
+
+class BaseService:
+    def __init__(self, model, session: AsyncSession):
+        self.model = model
+        self.session = session
+
+    async def _get(self, id: UUID):
+        return await self.session.get(self.model, id)
+    
+    async def _add(self, entity):
+        self.session.add(entity)
+        await self.session.commit()
+        await self.session.refresh(entity)
+        return entity
+    
+    async def _update(self, entity):
+        return await self._add(entity)
+    
+    async def _delete(self, entity):
+        await self.session.delete(entity)

app/home/tests/home/conftest.py

@@ -0,0 +1,48 @@
+from typing import AsyncGenerator
+
+import pytest_asyncio
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+from sqlalchemy.pool import NullPool
+
+from app.database.session import Base
+from config import db_settings
+
+# 테스트 전용 DB URL
+TEST_DB_URL = db_settings.MYSQL_URL.replace(
+    f"/{db_settings.MYSQL_DB}",
+    "/test_db",  # 별도 테스트 DB 사용
+)
+
+
+@pytest_asyncio.fixture
+async def test_engine():
+    """각 테스트마다 생성되는 테스트 엔진"""
+    engine = create_async_engine(
+        TEST_DB_URL,
+        poolclass=NullPool,  # 테스트에서는 풀 비활성화
+        echo=True,  # SQL 쿼리 로깅
+    )
+
+    # 테스트 테이블 생성
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+
+    yield engine
+
+    # 테스트 테이블 삭제
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.drop_all)
+
+    await engine.dispose()
+
+
+@pytest_asyncio.fixture
+async def db_session(test_engine) -> AsyncGenerator[AsyncSession, None]:
+    """각 테스트마다 새로운 세션 (격리 보장)"""
+    async_session = async_sessionmaker(
+        test_engine, class_=AsyncSession, expire_on_commit=False
+    )
+
+    async with async_session() as session:
+        yield session
+        await session.rollback()  # 테스트 후 롤백

app/home/tests/home/test_db.py

@@ -0,0 +1,17 @@
+import pytest
+from sqlalchemy import text
+
+
+@pytest.mark.asyncio
+async def test_database_connection(test_engine):
+    """테스트 엔진을 사용한 연결 테스트"""
+    async with test_engine.begin() as connection:
+        result = await connection.execute(text("SELECT 1"))
+        assert result.scalar() == 1
+
+
+@pytest.mark.asyncio
+async def test_session_usage(db_session):
+    """세션을 사용한 테스트"""
+    result = await db_session.execute(text("SELECT 1 as num"))
+    assert result.scalar() == 1

app/home/tests/test_db.py

@@ -0,0 +1,30 @@
+import pytest
+from sqlalchemy import text
+
+from app.database.session import AsyncSessionLocal, engine
+
+
+@pytest.mark.asyncio
+async def test_database_connection():
+    """데이터베이스 연결 테스트"""
+    async with engine.begin() as connection:
+        result = await connection.execute(text("SELECT 1"))
+        assert result.scalar() == 1
+
+
+@pytest.mark.asyncio
+async def test_session_creation():
+    """세션 생성 테스트"""
+    async with AsyncSessionLocal() as session:
+        result = await session.execute(text("SELECT 1"))
+        assert result.scalar() == 1
+
+
+@pytest.mark.asyncio
+async def test_database_version():
+    """MySQL 버전 확인 테스트"""
+    async with AsyncSessionLocal() as session:
+        result = await session.execute(text("SELECT VERSION()"))
+        version = result.scalar()
+        assert version is not None
+        print(f"MySQL Version: {version}")

app/lyrics/api/lyrics_admin.py

@@ -0,0 +1,135 @@
+from sqladmin import ModelView
+
+from app.lyrics.models import PromptTemplate  # noqa: F401
+from app.lyrics.models import Attribute, SongResultsAll, SongSample, StoreDefaultInfo
+
+
+class LyricsStoreDefaultInfoAdmin(ModelView, model=StoreDefaultInfo):
+    name = "상가 기본 정보"
+    name_plural = "상가 정보 목록"
+    icon = "fa-solid fa-store"
+    category = "상가 정보 관리"
+    page_size = 20
+
+    column_list = ["id", "store_name"]
+
+    # 폼(생성/수정)에서 제외
+    form_excluded_columns = ["created_at"]
+
+    column_searchable_list = [
+        StoreDefaultInfo.store_name,
+        StoreDefaultInfo.store_phone_number,
+    ]
+
+    column_default_sort = (StoreDefaultInfo.store_name, False)  # False: ASC, True: DESC
+
+    column_sortable_list = [
+        StoreDefaultInfo.created_at,
+        StoreDefaultInfo.store_phone_number,
+    ]
+
+    # 폼 컬럼 (기존 유지, user_posts가 관계라면 모델에 정의 필요)
+    # form_columns = [
+    #     Post.user_id,
+    #     Post.title,
+    #     Post.content,
+    #     Post.is_published,
+    #     Post.user_posts,  # Many-to-Many 또는 One-to-Many 관계 가정
+    # ]
+
+
+class LyricsAttributeAdmin(ModelView, model=Attribute):
+    name = "속성"
+    name_plural = "속성 목록"
+    icon = "fa-solid fa-tags"
+    category = "속성 관리"
+    page_size = 20
+
+    column_list = ["id", "attr_category"]
+
+    # 폼(생성/수정)에서 제외
+    form_excluded_columns = ["created_at"]
+
+    column_searchable_list = [
+        Attribute.attr_category,
+        Attribute.attr_value,
+    ]
+
+    column_default_sort = (Attribute.created_at, False)  # False: ASC, True: DESC
+
+    column_sortable_list = [
+        Attribute.created_at,
+    ]
+
+
+class LyricsSongSampleAdmin(ModelView, model=SongSample):
+    name = "가사 샘플"
+    name_plural = "가사 샘플 목록"
+    icon = "fa-solid fa-flask"
+    category = "가사 샘플 관리"
+    page_size = 20
+
+    column_list = [
+        "id",
+        "ai_model",
+        "season",
+        "num_of_people",
+        "people_category",
+        "genre",
+    ]
+
+    # 폼(생성/수정)에서 제외
+    form_excluded_columns = ["created_at"]
+
+    column_default_sort = (SongSample.created_at, False)  # False: ASC, True: DESC
+
+
+class LyricsPromptTemplateAdmin(ModelView, model=PromptTemplate):
+    name = "프롬프트 템플릿"
+    name_plural = "프롬프트 템플릿 목록"
+    icon = "fa-solid fa-file-alt"
+    category = "프롬프트 템플릿 관리"
+    page_size = 20
+
+    column_list = [
+        "id",
+        "description",
+    ]
+
+    # 폼(생성/수정)에서 제외
+    form_excluded_columns = ["created_at"]
+
+    column_default_sort = (PromptTemplate.created_at, False)  # False: ASC, True: DESC
+
+
+class LyricsSongResultsAllAdmin(ModelView, model=SongResultsAll):
+    name = "가사 결과"
+    name_plural = "가사 결과 목록"
+    icon = "fa-solid fa-music"
+    category = "가사 결과 관리"
+    page_size = 20
+
+    column_list = [
+        "id",
+        "store_name",
+    ]
+
+    # 폼(생성/수정)에서 제외
+    form_excluded_columns = ["created_at"]
+
+    column_searchable_list = [
+        SongResultsAll.ai,
+        SongResultsAll.ai_model,
+    ]
+
+    column_default_sort = (SongResultsAll.created_at, False)  # False: ASC, True: DESC
+
+    column_sortable_list = [
+        SongResultsAll.store_name,
+        SongResultsAll.store_category,
+        SongResultsAll.ai,
+        SongResultsAll.ai_model,
+        SongResultsAll.num_of_people,
+        SongResultsAll.genre,
+        SongResultsAll.created_at,
+    ]

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

@@ -0,0 +1,146 @@
+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/lyrics/dependencies.py

@@ -0,0 +1,8 @@
+from typing import Annotated
+
+from fastapi import Depends
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.database.session import get_session
+
+SessionDep = Annotated[AsyncSession, Depends(get_session)]

app/lyrics/models-refer.py

@@ -0,0 +1,158 @@
+from app.core.database import Base
+from sqlalchemy import (
+    Boolean,
+    DateTime,
+    Enum,
+    ForeignKey,
+    Index,
+    Integer,
+    PrimaryKeyConstraint,
+    String,
+    func,
+)
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+from starlette.authentication import BaseUser
+
+
+class User(Base, BaseUser):
+    __tablename__ = "users"
+
+    id: Mapped[int] = mapped_column(
+        Integer, primary_key=True, nullable=False, autoincrement=True
+    )
+    username: Mapped[str] = mapped_column(
+        String(255), unique=True, nullable=False, index=True
+    )
+    email: Mapped[str] = mapped_column(
+        String(255), unique=True, nullable=False, index=True
+    )
+    hashed_password: Mapped[str] = mapped_column(String(60), nullable=False)
+    # age_level 컬럼을 Enum으로 정의
+    age_level_choices = ["10", "20", "30", "40", "50", "60", "70", "80"]
+    age_level: Mapped[str] = mapped_column(
+        Enum(*age_level_choices, name="age_level_enum"),
+        nullable=False,
+        default="10",
+    )
+    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
+    created_at: Mapped[DateTime] = mapped_column(DateTime, server_default=func.now())
+
+    # One-to-many relationship with Post (DynamicMapped + lazy="dynamic")
+    posts_user: Mapped[list["Post"]] = relationship("Post", back_populates="user_posts")
+
+    # # Many-to-many relationship with Group
+    # user_groups: DynamicMapped["UserGroupAssociation"] = relationship(
+    #     "UserGroupAssociation", back_populates="user", lazy="dynamic"
+    # )
+    # n:m 관계 (Group) – 최적의 lazy 옵션: selectin
+    group_user: Mapped[list["Group"]] = relationship(
+        "Group",
+        secondary="user_group_association",
+        back_populates="user_group",
+        lazy="selectin",
+    )
+
+    def __repr__(self) -> str:
+        return f"id={self.id}, username={self.username}"
+
+    @property
+    def is_authenticated(self) -> bool:
+        return self.is_active
+
+    @property
+    def display_name(self) -> str:
+        return self.username
+
+    @property
+    def identity(self) -> str:
+        return self.username
+
+
+# 1:N Relationship - Posts
+class Post(Base):
+    __tablename__ = "posts"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    user_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False
+    )
+    title: Mapped[str] = mapped_column(String(255), nullable=False)
+    content: Mapped[str] = mapped_column(String(10000), nullable=False)
+    is_published: Mapped[bool] = mapped_column(Boolean, default=False)
+    created_at: Mapped[DateTime] = mapped_column(DateTime, server_default=func.now())
+    updated_at: Mapped[DateTime] = mapped_column(
+        DateTime, server_default=func.now(), onupdate=func.now()
+    )
+    # tags: Mapped[dict] = mapped_column(MutableDict.as_mutable(JSONB), default=[]) // sqlite 지원 안함
+    view_count: Mapped[int] = mapped_column(Integer, default=0)
+
+    # Many-to-one relationship with User (using dynamic loading)
+    user_posts: Mapped["User"] = relationship("User", back_populates="posts_user")
+
+    def __repr__(self) -> str:
+        return f"Post(id={self.id}, user_id={self.user_id}, title={self.title})"
+
+    __table_args__ = (
+        Index("idx_posts_user_id", "user_id"),
+        Index("idx_posts_created_at", "created_at"),
+        Index(
+            "idx_posts_user_id_created_at", "user_id", "created_at"
+        ),  # Composite index
+    )
+
+
+# N:M Relationship - Users and Groups
+# Association table for many-to-many relationship
+# N:M Association Table (중간 테이블)
+class UserGroupAssociation(Base):
+    __tablename__ = "user_group_association"
+
+    user_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("users.id", ondelete="CASCADE"), primary_key=True
+    )
+    group_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("groups.id", ondelete="CASCADE"), primary_key=True
+    )
+
+    # # 관계 정의
+    # user: Mapped["User"] = relationship("User", back_populates="user_groups")
+    # group: Mapped["Group"] = relationship("Group", back_populates="group_users")
+    # # 복합 기본 키 설정
+
+    # 기본 키 설정을 위한 __table_args__ 추가
+    __table_args__ = (PrimaryKeyConstraint("user_id", "group_id"),)
+
+    def __repr__(self) -> str:
+        return f"UserGroupAssociation(user_id={self.user_id}, group_id={self.group_id})"
+
+
+# Group 테이블
+class Group(Base):
+    __tablename__ = "groups"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    name: Mapped[str] = mapped_column(String(255), nullable=False, unique=True)
+    description: Mapped[str] = mapped_column(String(1000))
+    is_public: Mapped[bool] = mapped_column(Boolean, default=True)
+    created_at: Mapped[DateTime] = mapped_column(DateTime, server_default=func.now())
+    updated_at: Mapped[DateTime] = mapped_column(
+        DateTime, server_default=func.now(), onupdate=func.now()
+    )
+
+    user_group: Mapped[list["User"]] = relationship(
+        "User",
+        secondary="user_group_association",
+        back_populates="group_user",
+        lazy="selectin",
+    )
+
+    # Group을 만든 사용자와 관계 (일반적인 1:N 관계)
+    def __repr__(self) -> str:
+        return f"Group(id={self.id}, name={self.name})"
+
+    __table_args__ = (
+        Index("idx_groups_name", "name"),
+        Index("idx_groups_is_public", "is_public"),
+        Index("idx_groups_created_at", "created_at"),
+        Index("idx_groups_composite", "is_public", "created_at"),
+    )

app/lyrics/models.py

@@ -0,0 +1,273 @@
+from sqlalchemy import DateTime, Integer, String, Text, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.database.session import Base
+
+# from sqlalchemy import (
+#     Column,
+#     Integer,
+#     String,
+#     Boolean,
+#     DateTime,
+#     ForeignKey,
+#     Numeric,
+#     Table,
+#     Index,
+#     UniqueConstraint,
+#     CheckConstraint,
+#     text,
+#     func,
+#     PrimaryKeyConstraint,
+#     Enum,
+# )
+
+
+class StoreDefaultInfo(Base):
+    __tablename__ = "store_default_info"
+
+    id: Mapped[int] = mapped_column(
+        Integer, primary_key=True, nullable=False, autoincrement=True
+    )
+
+    store_info: Mapped[str] = mapped_column(
+        Text,
+        unique=False,
+        nullable=True,
+    )
+
+    store_name: Mapped[str] = mapped_column(
+        String(255),
+        unique=False,
+        nullable=False,
+    )
+
+    store_category: Mapped[str] = mapped_column(
+        String(255),
+        unique=False,
+        nullable=True,
+    )
+
+    store_region: Mapped[str] = mapped_column(
+        String(255),
+        unique=False,
+        nullable=True,
+    )
+
+    store_address: Mapped[str] = mapped_column(
+        String(255),
+        unique=False,
+        nullable=True,
+    )
+
+    store_phone_number: Mapped[str] = mapped_column(
+        String(255),
+        unique=True,
+        nullable=True,
+    )
+
+    created_at: Mapped[DateTime] = mapped_column(DateTime, server_default=func.now())
+
+    def __repr__(self) -> str:
+        return f"id={self.id}, store_name={self.store_name}"
+
+
+class PromptTemplate(Base):
+    __tablename__ = "prompt_template"
+
+    id: Mapped[int] = mapped_column(
+        Integer, primary_key=True, nullable=False, autoincrement=True
+    )
+
+    description: Mapped[str] = mapped_column(
+        Text,
+        unique=False,
+        nullable=True,
+    )
+
+    prompt: Mapped[str] = mapped_column(
+        Text,
+        unique=False,
+        nullable=False,
+    )
+
+    created_at: Mapped[DateTime] = mapped_column(DateTime, server_default=func.now())
+
+    def __repr__(self) -> str:
+        return f"id={self.id}, description={self.description}"
+
+
+class Attribute(Base):
+    __tablename__ = "attribute"
+
+    id: Mapped[int] = mapped_column(
+        Integer, primary_key=True, nullable=False, autoincrement=True
+    )
+
+    attr_category: Mapped[str] = mapped_column(
+        String(255),
+        unique=True,
+        nullable=False,
+    )
+
+    attr_value: Mapped[str] = mapped_column(
+        String(255),
+        unique=True,
+        nullable=False,
+    )
+
+    created_at: Mapped[DateTime] = mapped_column(DateTime, server_default=func.now())
+
+    def __repr__(self) -> str:
+        return f"id={self.id}, attr_category={self.attr_category}"
+
+
+class SongSample(Base):
+    __tablename__ = "song_sample"
+
+    id: Mapped[int] = mapped_column(
+        Integer, primary_key=True, nullable=False, autoincrement=True
+    )
+
+    ai: Mapped[str] = mapped_column(
+        String(255),
+        unique=False,
+        nullable=False,
+    )
+
+    ai_model: Mapped[str] = mapped_column(
+        String(255),
+        unique=False,
+        nullable=False,
+    )
+
+    genre: Mapped[str] = mapped_column(
+        String(255),
+        unique=False,
+        nullable=True,
+    )
+
+    sample_song: Mapped[str] = mapped_column(
+        Text,
+        unique=False,
+        nullable=False,
+    )
+
+    created_at: Mapped[DateTime] = mapped_column(DateTime, server_default=func.now())
+
+    def __repr__(self) -> str:
+        return f"id={self.id}, sample_song={self.sample_song}"
+
+
+class SongResultsAll(Base):
+    __tablename__ = "song_results_all"
+
+    id: Mapped[int] = mapped_column(
+        Integer, primary_key=True, nullable=False, autoincrement=True
+    )
+
+    store_info: Mapped[str] = mapped_column(
+        Text,
+        unique=False,
+        nullable=True,
+    )
+
+    store_name: Mapped[str] = mapped_column(
+        String(255),
+        unique=False,
+        nullable=False,
+    )
+
+    store_category: Mapped[str] = mapped_column(
+        String(255),
+        unique=False,
+        nullable=True,
+    )
+
+    store_address: Mapped[str] = mapped_column(
+        String(255),
+        unique=False,
+        nullable=True,
+    )
+
+    store_phone_number: Mapped[str] = mapped_column(
+        String(255),
+        unique=False,
+        nullable=True,
+    )
+
+    description: Mapped[str] = mapped_column(
+        Text,
+        unique=False,
+        nullable=True,
+    )
+
+    prompt: Mapped[str] = mapped_column(
+        String(255),
+        unique=False,
+        nullable=False,
+    )
+
+    attr_category: Mapped[str] = mapped_column(
+        String(255),
+        unique=False,
+        nullable=False,
+    )
+
+    attr_value: Mapped[str] = mapped_column(
+        String(255),
+        unique=False,
+        nullable=False,
+    )
+
+    ai: Mapped[str] = mapped_column(
+        String(255),
+        unique=False,
+        nullable=False,
+    )
+
+    ai_model: Mapped[str] = mapped_column(
+        String(255),
+        unique=False,
+        nullable=False,
+    )
+
+    season: Mapped[str] = mapped_column(
+        String(255),
+        unique=False,
+        nullable=True,
+    )
+
+    num_of_people: Mapped[int] = mapped_column(
+        Integer,
+        unique=False,
+        nullable=True,
+    )
+
+    people_category: Mapped[str] = mapped_column(
+        String(255),
+        unique=False,
+        nullable=True,
+    )
+
+    genre: Mapped[str] = mapped_column(
+        String(255),
+        unique=False,
+        nullable=True,
+    )
+
+    sample_song: Mapped[str] = mapped_column(
+        Text,
+        unique=False,
+        nullable=False,
+    )
+
+    result_song: Mapped[str] = mapped_column(
+        Text,
+        unique=False,
+        nullable=False,
+    )
+
+    created_at: Mapped[DateTime] = mapped_column(DateTime, server_default=func.now())
+
+    def __repr__(self) -> str:
+        return f"id={self.id}, result_song={self.result_song}"

app/lyrics/schemas/lyrics_schema.py

@@ -0,0 +1,91 @@
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import Dict, List
+
+from fastapi import Request
+
+
+@dataclass
+class StoreData:
+    id: int
+    created_at: datetime
+    store_name: str
+    store_category: str | None = None
+    store_region: str | None = None
+    store_address: str | None = None
+    store_phone_number: str | None = None
+    store_info: str | None = None
+
+
+@dataclass
+class AttributeData:
+    id: int
+    attr_category: str
+    attr_value: str
+    created_at: datetime
+
+
+@dataclass
+class SongSampleData:
+    id: int
+    ai: str
+    ai_model: str
+    sample_song: str
+    season: str | None = None
+    num_of_people: int | None = None
+    people_category: str | None = None
+    genre: str | None = None
+
+
+@dataclass
+class PromptTemplateData:
+    id: int
+    prompt: str
+    description: str | None = None
+
+
+@dataclass
+class SongFormData:
+    store_name: str
+    store_id: str
+    prompts: str
+    attributes: Dict[str, str] = field(default_factory=dict)
+    attributes_str: str = ""
+    lyrics_ids: List[int] = field(default_factory=list)
+    llm_model: str = "gpt-4o"
+
+    @classmethod
+    async def from_form(cls, request: Request):
+        """Request의 form 데이터로부터 dataclass 인스턴스 생성"""
+        form_data = await request.form()
+
+        # 고정 필드명들
+        fixed_keys = {"store_info_name", "store_id", "llm_model", "prompts"}
+
+        # lyrics-{id} 형태의 모든 키를 찾아서 ID 추출
+        lyrics_ids = []
+        attributes = {}
+
+        for key, value in form_data.items():
+            if key.startswith("lyrics-"):
+                lyrics_id = key.split("-")[1]
+                lyrics_ids.append(int(lyrics_id))
+            elif key not in fixed_keys:
+                attributes[key] = value
+
+        # attributes를 문자열로 변환
+        attributes_str = (
+            "\r\n\r\n".join([f"{key} : {value}" for key, value in attributes.items()])
+            if attributes
+            else ""
+        )
+
+        return cls(
+            store_name=form_data.get("store_info_name", ""),
+            store_id=form_data.get("store_id", ""),
+            attributes=attributes,
+            attributes_str=attributes_str,
+            lyrics_ids=lyrics_ids,
+            llm_model=form_data.get("llm_model", "gpt-4o"),
+            prompts=form_data.get("prompts", ""),
+        )

app/lyrics/services/base.py

@@ -0,0 +1,24 @@
+from uuid import UUID
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlmodel import SQLModel
+
+
+class BaseService:
+    def __init__(self, model, session: AsyncSession):
+        self.model = model
+        self.session = session
+
+    async def _get(self, id: UUID):
+        return await self.session.get(self.model, id)
+    
+    async def _add(self, entity):
+        self.session.add(entity)
+        await self.session.commit()
+        await self.session.refresh(entity)
+        return entity
+    
+    async def _update(self, entity):
+        return await self._add(entity)
+    
+    async def _delete(self, entity):
+        await self.session.delete(entity)

app/lyrics/services/lyrics.py

@@ -0,0 +1,852 @@
+import random
+from typing import List
+
+from fastapi import Request, status
+from fastapi.exceptions import HTTPException
+from sqlalchemy import Connection, text
+from sqlalchemy.exc import SQLAlchemyError
+
+from app.lyrics.schemas.lyrics_schema import (
+    AttributeData,
+    PromptTemplateData,
+    SongFormData,
+    SongSampleData,
+    StoreData,
+)
+from app.utils.chatgpt_prompt import chatgpt_api
+
+
+async def get_store_info(conn: Connection) -> List[StoreData]:
+    try:
+        query = """SELECT * FROM store_default_info;"""
+        result = await conn.execute(text(query))
+
+        all_store_info = [
+            StoreData(
+                id=row[0],
+                store_info=row[1],
+                store_name=row[2],
+                store_category=row[3],
+                store_region=row[4],
+                store_address=row[5],
+                store_phone_number=row[6],
+                created_at=row[7],
+            )
+            for row in result
+        ]
+
+        result.close()
+        return all_store_info
+    except SQLAlchemyError as e:
+        print(e)
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="요청하신 서비스가 잠시 내부적으로 문제가 발생하였습니다.",
+        )
+    except Exception as e:
+        print(e)
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="알수없는 이유로 서비스 오류가 발생하였습니다",
+        )
+
+
+async def get_attribute(conn: Connection) -> List[AttributeData]:
+    try:
+        query = """SELECT * FROM attribute;"""
+        result = await conn.execute(text(query))
+
+        all_attribute = [
+            AttributeData(
+                id=row[0],
+                attr_category=row[1],
+                attr_value=row[2],
+                created_at=row[3],
+            )
+            for row in result
+        ]
+
+        result.close()
+        return all_attribute
+    except SQLAlchemyError as e:
+        print(e)
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="요청하신 서비스가 잠시 내부적으로 문제가 발생하였습니다.",
+        )
+    except Exception as e:
+        print(e)
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="알수없는 이유로 서비스 오류가 발생하였습니다",
+        )
+
+
+async def get_attribute(conn: Connection) -> List[AttributeData]:
+    try:
+        query = """SELECT * FROM attribute;"""
+        result = await conn.execute(text(query))
+
+        all_attribute = [
+            AttributeData(
+                id=row[0],
+                attr_category=row[1],
+                attr_value=row[2],
+                created_at=row[3],
+            )
+            for row in result
+        ]
+
+        result.close()
+        return all_attribute
+    except SQLAlchemyError as e:
+        print(e)
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="요청하신 서비스가 잠시 내부적으로 문제가 발생하였습니다.",
+        )
+    except Exception as e:
+        print(e)
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="알수없는 이유로 서비스 오류가 발생하였습니다",
+        )
+
+
+async def get_sample_song(conn: Connection) -> List[SongSampleData]:
+    try:
+        query = """SELECT * FROM song_sample;"""
+        result = await conn.execute(text(query))
+
+        all_sample_song = [
+            SongSampleData(
+                id=row[0],
+                ai=row[1],
+                ai_model=row[2],
+                genre=row[3],
+                sample_song=row[4],
+            )
+            for row in result
+        ]
+
+        result.close()
+        return all_sample_song
+    except SQLAlchemyError as e:
+        print(e)
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="요청하신 서비스가 잠시 내부적으로 문제가 발생하였습니다.",
+        )
+    except Exception as e:
+        print(e)
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="알수없는 이유로 서비스 오류가 발생하였습니다",
+        )
+
+
+async def get_prompt_template(conn: Connection) -> List[PromptTemplateData]:
+    try:
+        query = """SELECT * FROM prompt_template;"""
+        result = await conn.execute(text(query))
+
+        all_prompt_template = [
+            PromptTemplateData(
+                id=row[0],
+                description=row[1],
+                prompt=row[2],
+            )
+            for row in result
+        ]
+
+        result.close()
+        return all_prompt_template
+    except SQLAlchemyError as e:
+        print(e)
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="요청하신 서비스가 잠시 내부적으로 문제가 발생하였습니다.",
+        )
+    except Exception as e:
+        print(e)
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="알수없는 이유로 서비스 오류가 발생하였습니다",
+        )
+
+
+async def get_song_result(conn: Connection) -> List[PromptTemplateData]:
+    try:
+        query = """SELECT * FROM prompt_template;"""
+        result = await conn.execute(text(query))
+
+        all_prompt_template = [
+            PromptTemplateData(
+                id=row[0],
+                description=row[1],
+                prompt=row[2],
+            )
+            for row in result
+        ]
+
+        result.close()
+        return all_prompt_template
+    except SQLAlchemyError as e:
+        print(e)
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="요청하신 서비스가 잠시 내부적으로 문제가 발생하였습니다.",
+        )
+    except Exception as e:
+        print(e)
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="알수없는 이유로 서비스 오류가 발생하였습니다",
+        )
+
+
+async def make_song_result(request: Request, conn: Connection):
+    try:
+        # 1. Form 데이터 파싱
+        form_data = await SongFormData.from_form(request)
+
+        print(f"\n{'=' * 60}")
+        print(f"Store ID: {form_data.store_id}")
+        print(f"Lyrics IDs: {form_data.lyrics_ids}")
+        print(f"Prompt IDs: {form_data.prompts}")
+        print(f"{'=' * 60}\n")
+
+        # 2. Store 정보 조회
+        store_query = """
+            SELECT * FROM store_default_info WHERE id=:id;
+        """
+        store_result = await conn.execute(text(store_query), {"id": form_data.store_id})
+
+        all_store_info = [
+            StoreData(
+                id=row[0],
+                store_info=row[1],
+                store_name=row[2],
+                store_category=row[3],
+                store_region=row[4],
+                store_address=row[5],
+                store_phone_number=row[6],
+                created_at=row[7],
+            )
+            for row in store_result
+        ]
+
+        if not all_store_info:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail=f"Store not found: {form_data.store_id}",
+            )
+
+        store_info = all_store_info[0]
+        print(f"Store: {store_info.store_name}")
+
+        # 3. 속성 조회 -- 단계별 선택 프로세서시 구현 필요 없음
+
+        # 4. Sample Song 조회 및 결합
+        combined_sample_song = None
+
+        if form_data.lyrics_ids:
+            print(f"\n[샘플 가사 조회] - {len(form_data.lyrics_ids)}개")
+
+            lyrics_query = """
+                SELECT sample_song FROM song_sample
+                WHERE id IN :ids
+                ORDER BY created_at;
+            """
+            lyrics_result = await conn.execute(
+                text(lyrics_query), {"ids": tuple(form_data.lyrics_ids)}
+            )
+
+            sample_songs = [
+                row.sample_song for row in lyrics_result.fetchall() if row.sample_song
+            ]
+
+            if sample_songs:
+                combined_sample_song = "\n\n".join(
+                    [f"[샘플 {i + 1}]\n{song}" for i, song in enumerate(sample_songs)]
+                )
+                print(f"{len(sample_songs)}개의 샘플 가사 조회 완료")
+            else:
+                print("샘플 가사가 비어있습니다")
+        else:
+            print("선택된 lyrics가 없습니다")
+
+        # 5. 템플릿 가져오기
+        if not form_data.prompts:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="프롬프트 ID가 필요합니다",
+            )
+
+        print("템플릿 가져오기")
+
+        prompts_query = """
+            SELECT * FROM prompt_template WHERE id=:id;
+        """
+
+        # ✅ 수정: store_query → prompts_query
+        prompts_result = await conn.execute(
+            text(prompts_query), {"id": form_data.prompts}
+        )
+
+        prompts_info = [
+            PromptTemplateData(
+                id=row[0],
+                description=row[1],
+                prompt=row[2],
+            )
+            for row in prompts_result
+        ]
+
+        if not prompts_info:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail=f"Prompt not found: {form_data.prompts}",
+            )
+
+        prompt = prompts_info[0]
+        print(f"Prompt Template: {prompt.prompt}")
+
+        # ✅ 6. 프롬프트 조합
+        updated_prompt = prompt.prompt.replace("###", form_data.attributes_str).format(
+            name=store_info.store_name or "",
+            address=store_info.store_address or "",
+            category=store_info.store_category or "",
+            description=store_info.store_info or "",
+        )
+
+        updated_prompt += f"""
+
+            다음은 참고해야 하는 샘플 가사 정보입니다.
+
+            샘플 가사를 참고하여 작곡을 해주세요.
+
+            {combined_sample_song}
+            """
+
+        print(f"\n[업데이트된 프롬프트]\n{updated_prompt}\n")
+
+        # 7. 모델에게 요청
+        generated_lyrics = await chatgpt_api.generate_lyrics(prompt=updated_prompt)
+
+        # 글자 수 계산
+        total_chars_with_space = len(generated_lyrics)
+        total_chars_without_space = len(
+            generated_lyrics.replace(" ", "")
+            .replace("\n", "")
+            .replace("\r", "")
+            .replace("\t", "")
+        )
+
+        # final_lyrics 생성
+        final_lyrics = f"""속성 {form_data.attributes_str}
+        전체 글자 수 (공백 포함): {total_chars_with_space}자
+        전체 글자 수 (공백 제외): {total_chars_without_space}자\r\n\r\n{generated_lyrics}"""
+
+        print("=" * 40)
+        print("[translate:form_data.attributes_str:] ", form_data.attributes_str)
+        print("[translate:total_chars_with_space:] ", total_chars_with_space)
+        print("[translate:total_chars_without_space:] ", total_chars_without_space)
+        print("[translate:final_lyrics:]")
+        print(final_lyrics)
+        print("=" * 40)
+
+        # 8. DB 저장
+        insert_query = """
+            INSERT INTO song_results_all (
+                store_info, store_name, store_category, store_address, store_phone_number,
+                description, prompt, attr_category, attr_value,
+                ai, ai_model, genre,
+                sample_song, result_song, created_at
+            ) VALUES (
+                :store_info, :store_name, :store_category, :store_address, :store_phone_number,
+                :description, :prompt, :attr_category, :attr_value,
+                :ai, :ai_model, :genre,
+                :sample_song, :result_song, NOW()
+            );
+        """
+
+        # ✅ attr_category, attr_value 추가
+        insert_params = {
+            "store_info": store_info.store_info or "",
+            "store_name": store_info.store_name,
+            "store_category": store_info.store_category or "",
+            "store_address": store_info.store_address or "",
+            "store_phone_number": store_info.store_phone_number or "",
+            "description": store_info.store_info or "",
+            "prompt": form_data.prompts,
+            "attr_category": ", ".join(form_data.attributes.keys())
+            if form_data.attributes
+            else "",
+            "attr_value": ", ".join(form_data.attributes.values())
+            if form_data.attributes
+            else "",
+            "ai": "ChatGPT",
+            "ai_model": form_data.llm_model,
+            "genre": "후크송",
+            "sample_song": combined_sample_song or "없음",
+            "result_song": final_lyrics,
+        }
+
+        await conn.execute(text(insert_query), insert_params)
+        await conn.commit()
+
+        print("결과 저장 완료")
+
+        print("\n전체 결과 조회 중...")
+
+        # 9. 생성 결과 가져오기 (created_at 역순)
+        select_query = """
+            SELECT * FROM song_results_all 
+            ORDER BY created_at DESC;
+        """
+
+        all_results = await conn.execute(text(select_query))
+
+        results_list = [
+            {
+                "id": row.id,
+                "store_info": row.store_info,
+                "store_name": row.store_name,
+                "store_category": row.store_category,
+                "store_address": row.store_address,
+                "store_phone_number": row.store_phone_number,
+                "description": row.description,
+                "prompt": row.prompt,
+                "attr_category": row.attr_category,
+                "attr_value": row.attr_value,
+                "ai": row.ai,
+                "ai_model": row.ai_model,
+                "genre": row.genre,
+                "sample_song": row.sample_song,
+                "result_song": row.result_song,
+                "created_at": row.created_at.isoformat() if row.created_at else None,
+            }
+            for row in all_results.fetchall()
+        ]
+
+        print(f"전체 {len(results_list)}개의 결과 조회 완료\n")
+
+        return results_list
+
+    except HTTPException:
+        raise
+    except SQLAlchemyError as e:
+        print(f"Database Error: {e}")
+        import traceback
+
+        traceback.print_exc()
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="데이터베이스 연결에 문제가 발생했습니다.",
+        )
+    except Exception as e:
+        print(f"Unexpected Error: {e}")
+        import traceback
+
+        traceback.print_exc()
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="서비스 처리 중 오류가 발생했습니다.",
+        )
+
+
+async def get_song_result(conn: Connection):  # 반환 타입 수정
+    try:
+        select_query = """
+            SELECT * FROM song_results_all 
+            ORDER BY created_at DESC;
+        """
+
+        all_results = await conn.execute(text(select_query))
+
+        results_list = [
+            {
+                "id": row.id,
+                "store_info": row.store_info,
+                "store_name": row.store_name,
+                "store_category": row.store_category,
+                "store_address": row.store_address,
+                "store_phone_number": row.store_phone_number,
+                "description": row.description,
+                "prompt": row.prompt,
+                "attr_category": row.attr_category,
+                "attr_value": row.attr_value,
+                "ai": row.ai,
+                "ai_model": row.ai_model,
+                "season": row.season,
+                "num_of_people": row.num_of_people,
+                "people_category": row.people_category,
+                "genre": row.genre,
+                "sample_song": row.sample_song,
+                "result_song": row.result_song,
+                "created_at": row.created_at.isoformat() if row.created_at else None,
+            }
+            for row in all_results.fetchall()
+        ]
+
+        print(f"전체 {len(results_list)}개의 결과 조회 완료\n")
+
+        return results_list
+    except HTTPException:  # HTTPException은 그대로 raise
+        raise
+    except SQLAlchemyError as e:
+        print(f"Database Error: {e}")
+        import traceback
+
+        traceback.print_exc()
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="데이터베이스 연결에 문제가 발생했습니다.",
+        )
+    except Exception as e:
+        print(f"Unexpected Error: {e}")
+        import traceback
+
+        traceback.print_exc()
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="서비스 처리 중 오류가 발생했습니다.",
+        )
+
+
+async def make_automation(request: Request, conn: Connection):
+    try:
+        # 1. Form 데이터 파싱
+        form_data = await SongFormData.from_form(request)
+
+        print(f"\n{'=' * 60}")
+        print(f"Store ID: {form_data.store_id}")
+        print(f"{'=' * 60}\n")
+
+        # 2. Store 정보 조회
+        store_query = """
+            SELECT * FROM store_default_info WHERE id=:id;
+        """
+        store_result = await conn.execute(text(store_query), {"id": form_data.store_id})
+
+        all_store_info = [
+            StoreData(
+                id=row[0],
+                store_info=row[1],
+                store_name=row[2],
+                store_category=row[3],
+                store_region=row[4],
+                store_address=row[5],
+                store_phone_number=row[6],
+                created_at=row[7],
+            )
+            for row in store_result
+        ]
+
+        if not all_store_info:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail=f"Store not found: {form_data.store_id}",
+            )
+
+        store_info = all_store_info[0]
+        print(f"Store: {store_info.store_name}")
+
+        # 3. 속성 조회 -- 단계별 선택 프로세서시 구현 필요 없음
+        attribute_query = """
+            SELECT * FROM attribute;
+        """
+
+        attribute_results = await conn.execute(text(attribute_query))
+
+        # 결과 가져오기
+        attribute_rows = attribute_results.fetchall()
+
+        formatted_attributes = ""
+        selected_categories = []
+        selected_values = []
+
+        if attribute_rows:
+            attribute_list = [
+                AttributeData(
+                    id=row[0],
+                    attr_category=row[1],
+                    attr_value=row[2],
+                    created_at=row[3],
+                )
+                for row in attribute_rows
+            ]
+
+            # ✅ 각 category에서 하나의 value만 랜덤 선택
+            formatted_pairs = []
+            for attr in attribute_list:
+                # 쉼표로 분리 및 공백 제거
+                values = [v.strip() for v in attr.attr_value.split(",") if v.strip()]
+
+                if values:
+                    # 랜덤하게 하나만 선택
+                    selected_value = random.choice(values)
+                    formatted_pairs.append(f"{attr.attr_category} : {selected_value}")
+
+                    # ✅ 선택된 category와 value 저장
+                    selected_categories.append(attr.attr_category)
+                    selected_values.append(selected_value)
+
+            # 최종 문자열 생성
+            formatted_attributes = "\n".join(formatted_pairs)
+
+            print(f"\n[포맷팅된 문자열 속성 정보]\n{formatted_attributes}\n")
+        else:
+            print("속성 데이터가 없습니다")
+            formatted_attributes = ""
+
+        # 4. 템플릿 가져오기
+        print("템플릿 가져오기 (ID=1)")
+
+        prompts_query = """
+            SELECT * FROM prompt_template WHERE id=1;
+        """
+
+        prompts_result = await conn.execute(text(prompts_query))
+
+        row = prompts_result.fetchone()
+
+        if not row:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="Prompt ID 1 not found",
+            )
+
+        prompt = PromptTemplateData(
+            id=row[0],
+            description=row[1],
+            prompt=row[2],
+        )
+
+        print(f"Prompt Template: {prompt.prompt}")
+
+        # 5. 템플릿 조합
+
+        updated_prompt = prompt.prompt.replace("###", formatted_attributes).format(
+            name=store_info.store_name or "",
+            address=store_info.store_address or "",
+            category=store_info.store_category or "",
+            description=store_info.store_info or "",
+        )
+
+        print("\n" + "=" * 80)
+        print("업데이트된 프롬프트")
+        print("=" * 80)
+        print(updated_prompt)
+        print("=" * 80 + "\n")
+
+        # 4. Sample Song 조회 및 결합
+        combined_sample_song = None
+
+        if form_data.lyrics_ids:
+            print(f"\n[샘플 가사 조회] - {len(form_data.lyrics_ids)}개")
+
+            lyrics_query = """
+                SELECT sample_song FROM song_sample
+                WHERE id IN :ids
+                ORDER BY created_at;
+            """
+            lyrics_result = await conn.execute(
+                text(lyrics_query), {"ids": tuple(form_data.lyrics_ids)}
+            )
+
+            sample_songs = [
+                row.sample_song for row in lyrics_result.fetchall() if row.sample_song
+            ]
+
+            if sample_songs:
+                combined_sample_song = "\n\n".join(
+                    [f"[샘플 {i + 1}]\n{song}" for i, song in enumerate(sample_songs)]
+                )
+                print(f"{len(sample_songs)}개의 샘플 가사 조회 완료")
+            else:
+                print("샘플 가사가 비어있습니다")
+        else:
+            print("선택된 lyrics가 없습니다")
+
+        # 1. song_sample 테이블의 모든 ID 조회
+        print("\n[샘플 가사 랜덤 선택]")
+
+        all_ids_query = """
+            SELECT id FROM song_sample;
+        """
+        ids_result = await conn.execute(text(all_ids_query))
+        all_ids = [row.id for row in ids_result.fetchall()]
+
+        print(f"전체 샘플 가사 개수: {len(all_ids)}개")
+
+        # 2. 랜덤하게 3개 선택 (또는 전체 개수가 3개 미만이면 전체)
+        combined_sample_song = None
+
+        if all_ids:
+            # 3개 또는 전체 개수 중 작은 값 선택
+            sample_count = min(3, len(all_ids))
+            selected_ids = random.sample(all_ids, sample_count)
+
+            print(f"랜덤 선택된 ID: {selected_ids}")
+
+            # 3. 선택된 ID로 샘플 가사 조회
+            lyrics_query = """
+                SELECT sample_song FROM song_sample
+                WHERE id IN :ids
+                ORDER BY created_at;
+            """
+            lyrics_result = await conn.execute(
+                text(lyrics_query), {"ids": tuple(selected_ids)}
+            )
+
+            sample_songs = [
+                row.sample_song for row in lyrics_result.fetchall() if row.sample_song
+            ]
+
+            # 4. combined_sample_song 생성
+            if sample_songs:
+                combined_sample_song = "\n\n".join(
+                    [f"[샘플 {i + 1}]\n{song}" for i, song in enumerate(sample_songs)]
+                )
+                print(f"{len(sample_songs)}개의 샘플 가사 조회 완료")
+            else:
+                print("샘플 가사가 비어있습니다")
+        else:
+            print("song_sample 테이블에 데이터가 없습니다")
+
+        # 5. 프롬프트에 샘플 가사 추가
+        if combined_sample_song:
+            updated_prompt += f"""
+
+                다음은 참고해야 하는 샘플 가사 정보입니다.
+
+                샘플 가사를 참고하여 작곡을 해주세요.
+
+                {combined_sample_song}
+            """
+            print("샘플 가사 정보가 프롬프트에 추가되었습니다")
+        else:
+            print("샘플 가사가 없어 기본 프롬프트만 사용합니다")
+
+        print(f"\n[최종 프롬프트 길이: {len(updated_prompt)} 자]\n")
+
+        # 7. 모델에게 요청
+        generated_lyrics = await chatgpt_api.generate_lyrics(prompt=updated_prompt)
+
+        # 글자 수 계산
+        total_chars_with_space = len(generated_lyrics)
+        total_chars_without_space = len(
+            generated_lyrics.replace(" ", "")
+            .replace("\n", "")
+            .replace("\r", "")
+            .replace("\t", "")
+        )
+
+        # final_lyrics 생성
+        final_lyrics = f"""속성 {formatted_attributes}
+        전체 글자 수 (공백 포함): {total_chars_with_space}자
+        전체 글자 수 (공백 제외): {total_chars_without_space}자\r\n\r\n{generated_lyrics}"""
+
+        # 8. DB 저장
+        insert_query = """
+            INSERT INTO song_results_all (
+                store_info, store_name, store_category, store_address, store_phone_number,
+                description, prompt, attr_category, attr_value,
+                ai, ai_model, genre,
+                sample_song, result_song, created_at
+            ) VALUES (
+                :store_info, :store_name, :store_category, :store_address, :store_phone_number,
+                :description, :prompt, :attr_category, :attr_value,
+                :ai, :ai_model, :genre,
+                :sample_song, :result_song, NOW()
+            );
+        """
+        print("\n[insert_params 선택된 속성 확인]")
+        print(f"Categories: {selected_categories}")
+        print(f"Values: {selected_values}")
+        print()
+
+        # attr_category, attr_value
+        insert_params = {
+            "store_info": store_info.store_info or "",
+            "store_name": store_info.store_name,
+            "store_category": store_info.store_category or "",
+            "store_address": store_info.store_address or "",
+            "store_phone_number": store_info.store_phone_number or "",
+            "description": store_info.store_info or "",
+            "prompt": prompt.id,
+            # 랜덤 선택된 category와 value 사용
+            "attr_category": ", ".join(selected_categories)
+            if selected_categories
+            else "",
+            "attr_value": ", ".join(selected_values) if selected_values else "",
+            "ai": "ChatGPT",
+            "ai_model": "gpt-4o",
+            "genre": "후크송",
+            "sample_song": combined_sample_song or "없음",
+            "result_song": final_lyrics,
+        }
+
+        await conn.execute(text(insert_query), insert_params)
+        await conn.commit()
+
+        print("결과 저장 완료")
+
+        print("\n전체 결과 조회 중...")
+
+        # 9. 생성 결과 가져오기 (created_at 역순)
+        select_query = """
+            SELECT * FROM song_results_all
+            ORDER BY created_at DESC;
+        """
+
+        all_results = await conn.execute(text(select_query))
+
+        results_list = [
+            {
+                "id": row.id,
+                "store_info": row.store_info,
+                "store_name": row.store_name,
+                "store_category": row.store_category,
+                "store_address": row.store_address,
+                "store_phone_number": row.store_phone_number,
+                "description": row.description,
+                "prompt": row.prompt,
+                "attr_category": row.attr_category,
+                "attr_value": row.attr_value,
+                "ai": row.ai,
+                "ai_model": row.ai_model,
+                "genre": row.genre,
+                "sample_song": row.sample_song,
+                "result_song": row.result_song,
+                "created_at": row.created_at.isoformat() if row.created_at else None,
+            }
+            for row in all_results.fetchall()
+        ]
+
+        print(f"전체 {len(results_list)}개의 결과 조회 완료\n")
+
+        return results_list
+
+    except HTTPException:
+        raise
+    except SQLAlchemyError as e:
+        print(f"Database Error: {e}")
+        import traceback
+
+        traceback.print_exc()
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="데이터베이스 연결에 문제가 발생했습니다.",
+        )
+    except Exception as e:
+        print(f"Unexpected Error: {e}")
+        import traceback
+
+        traceback.print_exc()
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="서비스 처리 중 오류가 발생했습니다.",
+        )

app/utils/chatgpt_prompt.py

@@ -0,0 +1,93 @@
+from openai import AsyncOpenAI
+
+from config import apikey_settings
+
+
+class ChatgptService:
+    def __init__(self):
+        self.model = "gpt-4o"
+        self.client = AsyncOpenAI(api_key=apikey_settings.CHATGPT_API_KEY)
+
+    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,
+        # )
+
+        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()

app/utils/cors.py

@@ -0,0 +1,24 @@
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+
+from config import cors_settings
+
+# sys.path.append(
+#     os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+# )  # root 경로 추가
+
+
+class CustomCORSMiddleware:
+    def __init__(self, app: FastAPI):
+        self.app = app
+
+    def configure_cors(self):
+        self.app.add_middleware(
+            CORSMiddleware,
+            allow_origins=cors_settings.CORS_ALLOW_ORIGINS,
+            allow_credentials=cors_settings.CORS_ALLOW_CREDENTIALS,
+            allow_methods=cors_settings.CORS_ALLOW_METHODS,
+            allow_headers=cors_settings.CORS_ALLOW_HEADERS,
+            expose_headers=cors_settings.CORS_EXPOSE_HEADERS,
+            max_age=cors_settings.CORS_MAX_AGE,
+        )

config.py

@@ -0,0 +1,123 @@
+from pathlib import Path
+
+from pydantic import Field
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+PROJECT_DIR = Path(__file__).resolve().parent
+
+_base_config = SettingsConfigDict(
+    env_file=PROJECT_DIR / ".env",
+    env_ignore_empty=True,
+    extra="ignore",
+)
+
+
+class ProjectSettings(BaseSettings):
+    PROJECT_NAME: str = Field(default="CastAD")
+    PROJECT_DOMAIN: str = Field(default="localhost:8000")
+    VERSION: str = Field(default="0.1.0")
+    DESCRIPTION: str = Field(default="FastAPI 기반 CastAD 프로젝트")
+    ADMIN_BASE_URL: str = Field(default="/admin")
+
+    model_config = _base_config
+
+
+class APIKeySettings(BaseSettings):
+    CHATGPT_API_KEY: str = Field(default="your-chatgpt-api-key")  # 기본값 추가
+
+    model_config = _base_config
+
+
+class CORSSettings(BaseSettings):
+    # CORS (Cross-Origin Resource Sharing) 설정
+
+    # 요청을 허용할 출처(Origin) 목록
+    # ["*"]: 모든 출처 허용 (개발 환경용, 프로덕션에서는 구체적인 도메인 지정 권장)
+    # 예: ["https://example.com", "https://app.example.com"]
+    CORS_ALLOW_ORIGINS: list[str] = ["*"]
+
+    # 자격 증명(쿠키, Authorization 헤더 등) 포함 요청 허용 여부
+    # True: 클라이언트가 credentials: 'include'로 요청 시 쿠키/인증 정보 전송 가능
+    # 주의: CORS_ALLOW_ORIGINS가 ["*"]일 때는 보안상 False 권장
+    CORS_ALLOW_CREDENTIALS: bool = True
+
+    # 허용할 HTTP 메서드 목록
+    # ["*"]: 모든 메서드 허용 (GET, POST, PUT, DELETE, PATCH, OPTIONS 등)
+    # 구체적 지정 예: ["GET", "POST", "PUT", "DELETE"]
+    CORS_ALLOW_METHODS: list[str] = ["*"]
+
+    # 클라이언트가 요청 시 사용할 수 있는 HTTP 헤더 목록
+    # ["*"]: 모든 헤더 허용
+    # 구체적 지정 예: ["Content-Type", "Authorization", "X-Custom-Header"]
+    CORS_ALLOW_HEADERS: list[str] = ["*"]
+
+    # 브라우저의 JavaScript에서 접근 가능한 응답 헤더 목록
+    # []: 기본 안전 헤더(Cache-Control, Content-Language, Content-Type,
+    #     Expires, Last-Modified, Pragma)만 접근 가능
+    # 추가 노출 필요 시: ["X-Total-Count", "X-Request-Id", "X-Custom-Header"]
+    CORS_EXPOSE_HEADERS: list[str] = []
+
+    # Preflight 요청(OPTIONS) 결과를 캐시하는 시간(초)
+    # 600: 10분간 캐시 (이 시간 동안 동일 요청에 대해 preflight 생략)
+    # 0으로 설정 시 매번 preflight 요청 발생
+    CORS_MAX_AGE: int = 600
+
+    model_config = _base_config
+
+
+class DatabaseSettings(BaseSettings):
+    # MySQL 연결 설정 (기본값: 테스트 계정 및 poc DB)
+    MYSQL_HOST: str = Field(default="localhost")
+    MYSQL_PORT: int = Field(default=3306)
+    MYSQL_USER: str = Field(default="test")
+    MYSQL_PASSWORD: str = Field(default="")  # 환경변수에서 로드
+    MYSQL_DB: str = Field(default="poc")
+
+    # Redis 설정
+    REDIS_HOST: str = "localhost"
+    REDIS_PORT: int = 6379
+
+    model_config = _base_config
+
+    @property
+    def MYSQL_URL(self) -> str:
+        """비동기 MySQL URL 생성 (asyncmy 드라이버 사용, SQLAlchemy 통합 최적화)"""
+        return f"mysql+asyncmy://{self.MYSQL_USER}:{self.MYSQL_PASSWORD}@{self.MYSQL_HOST}:{self.MYSQL_PORT}/{self.MYSQL_DB}"
+
+    def REDIS_URL(self, db: int = 0) -> str:
+        """Redis URL 생성 (db 인수로 기본값 지원)"""
+        return f"redis://{self.REDIS_HOST}:{self.REDIS_PORT}/{db}"
+
+
+class SecuritySettings(BaseSettings):
+    JWT_SECRET: str = "your-jwt-secret-key"  # 기본값 추가 (필수 필드 안전)
+    JWT_ALGORITHM: str = "HS256"  # 기본값 추가 (필수 필드 안전)
+
+    model_config = _base_config
+
+
+class NotificationSettings(BaseSettings):
+    MAIL_USERNAME: str = "your-email@example.com"  # 기본값 추가
+    MAIL_PASSWORD: str = "your-email-password"  # 기본값 추가
+    MAIL_FROM: str = "your-email@example.com"  # 기본값 추가
+    MAIL_PORT: int = 587  # 기본값 추가
+    MAIL_SERVER: str = "smtp.gmail.com"  # 기본값 추가
+    MAIL_FROM_NAME: str = "FastPOC App"  # 기본값 추가
+    MAIL_STARTTLS: bool = True
+    MAIL_SSL_TLS: bool = False
+    USE_CREDENTIALS: bool = True
+    VALIDATE_CERTS: bool = True
+
+    TWILIO_SID: str = "your-twilio-sid"  # 기본값 추가
+    TWILIO_AUTH_TOKEN: str = "your-twilio-token"  # 기본값 추가
+    TWILIO_NUMBER: str = "+1234567890"  # 기본값 추가
+
+    model_config = _base_config
+
+
+prj_settings = ProjectSettings()
+apikey_settings = APIKeySettings()
+db_settings = DatabaseSettings()
+security_settings = SecuritySettings()
+notification_settings = NotificationSettings()
+cors_settings = CORSSettings()

main.py

@@ -0,0 +1,50 @@
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.staticfiles import StaticFiles
+from scalar_fastapi import get_scalar_api_reference
+
+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.lyrics.api.routers.v1.router import router as lyrics_router
+from app.utils.cors import CustomCORSMiddleware
+from config import prj_settings
+
+app = FastAPI(
+    title=prj_settings.PROJECT_NAME,
+    version=prj_settings.VERSION,
+    description=prj_settings.DESCRIPTION,
+    lifespan=lifespan,
+    docs_url=None,  # 기본 Swagger UI 비활성화
+    redoc_url=None,  # 기본 ReDoc 비활성화
+)
+
+init_admin(app, engine)
+
+custom_cors_middleware = CustomCORSMiddleware(app)
+custom_cors_middleware.configure_cors()
+
+app.mount("/static", StaticFiles(directory="static"), name="static")
+app.mount("/media", StaticFiles(directory="media"), name="media")
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_methods=["*"],
+    allow_headers=["*"],
+    allow_credentials=True,
+    max_age=-1,
+)
+
+
+@app.get("/docs", include_in_schema=False)
+def get_scalar_docs():
+    return get_scalar_api_reference(
+        openapi_url=app.openapi_url,
+        title="Scalar API",
+    )
+
+
+app.include_router(home_router)
+app.include_router(lyrics_router)

pyproject.toml

@@ -0,0 +1,25 @@
+[project]
+name = "castad-ver0-1"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.13"
+dependencies = [
+    "aiomysql>=0.3.2",
+    "asyncmy>=0.2.10",
+    "fastapi-cli>=0.0.16",
+    "fastapi[standard]>=0.125.0",
+    "openai>=2.13.0",
+    "pydantic-settings>=2.12.0",
+    "redis>=7.1.0",
+    "ruff>=0.14.9",
+    "scalar-fastapi>=1.5.0",
+    "sqladmin[full]>=0.22.0",
+    "sqlalchemy[asyncio]>=2.0.45",
+]
+
+[dependency-groups]
+dev = [
+    "pytest>=9.0.2",
+    "pytest-asyncio>=1.3.0",
+]