app/database/session.py

@@ -74,7 +74,7 @@ async def create_db_tables():
 
     # 모델 import (테이블 메타데이터 등록용)
     from app.user.models import User, RefreshToken, SocialAccount  # noqa: F401
-    from app.home.models import Image, Project, MarketingIntel, ImageTag  # noqa: F401
+    from app.home.models import Image, Project, MarketingIntel  # noqa: F401
     from app.lyric.models import Lyric  # noqa: F401
     from app.song.models import Song, SongTimestamp  # noqa: F401
     from app.video.models import Video  # noqa: F401
@@ -97,7 +97,6 @@ async def create_db_tables():
         SocialUpload.__table__,
         MarketingIntel.__table__,
         Dashboard.__table__,
-        ImageTag.__table__,
     ]
 
     logger.info("Creating database tables...")

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

@@ -9,10 +9,9 @@ import aiofiles
 from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile, status
 from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.ext.asyncio import AsyncSession
-from sqlalchemy import func, select
 
 from app.database.session import get_session, AsyncSessionLocal
-from app.home.models import Image, MarketingIntel, ImageTag
+from app.home.models import Image, MarketingIntel
 from app.user.dependencies.auth import get_current_user
 from app.user.models import User
 from app.home.schemas.home_schema import (
@@ -30,13 +29,12 @@ from app.home.schemas.home_schema import (
 )
 from app.home.services.naver_search import naver_search_client
 from app.utils.upload_blob_as_request import AzureBlobUploader
-from app.utils.prompts.chatgpt_prompt import ChatgptService, ChatGPTResponseError
+from app.utils.chatgpt_prompt import ChatgptService, ChatGPTResponseError
 from app.utils.common import generate_task_id
 from app.utils.logger import get_logger
-from app.utils.nvMapScraper import NvMapScraper, GraphQLException, URLNotFoundException
+from app.utils.nvMapScraper import NvMapScraper, GraphQLException
 from app.utils.nvMapPwScraper import NvMapPwScraper
 from app.utils.prompts.prompts import marketing_prompt
-from app.utils.autotag import autotag_images
 from config import MEDIA_ROOT
 
 # 로거 설정
@@ -220,15 +218,6 @@ async def _crawling_logic(
             status_code=status.HTTP_502_BAD_GATEWAY,
             detail=f"네이버 지도 크롤링에 실패했습니다: {e}",
         )
-    except URLNotFoundException as e:
-        step1_elapsed = (time.perf_counter() - step1_start) * 1000
-        logger.error(
-            f"[crawling] Step 1 FAILED - 크롤링 실패: {e} ({step1_elapsed:.1f}ms)"
-        )
-        raise HTTPException(
-            status_code=status.HTTP_404_NOT_FOUND,
-            detail=f"Place ID를 확인할 수 없습니다. URL을 확인하세요. : {e}",
-        )
     except Exception as e:
         step1_elapsed = (time.perf_counter() - step1_start) * 1000
         logger.error(
@@ -462,6 +451,255 @@ IMAGES_JSON_EXAMPLE = """[
     {"url": "https://naverbooking-phinf.pstatic.net/20240514_259/17156880311809wCnY_JPEG/5.jpg", "name": "외관"}
 ]"""
 
+
+@router.post(
+    "/image/upload/server",
+    include_in_schema=False,
+    summary="이미지 업로드 (로컬 서버)",
+    description="""
+이미지를 로컬 서버(media 폴더)에 업로드하고 새로운 task_id를 생성합니다.
+
+## 요청 방식
+multipart/form-data 형식으로 전송합니다.
+
+## 요청 필드
+- **images_json**: 외부 이미지 URL 목록 (JSON 문자열, 선택)
+- **files**: 이미지 바이너리 파일 목록 (선택)
+
+**주의**: images_json 또는 files 중 최소 하나는 반드시 전달해야 합니다.
+
+## 지원 이미지 확장자
+jpg, jpeg, png, webp, heic, heif
+
+## images_json 예시
+```json
+[
+    {"url": "https://naverbooking-phinf.pstatic.net/20240514_189/1715688030436xT14o_JPEG/1.jpg"},
+    {"url": "https://naverbooking-phinf.pstatic.net/20240514_48/1715688030574wTtQd_JPEG/2.jpg"},
+    {"url": "https://naverbooking-phinf.pstatic.net/20240514_92/17156880307484bvpH_JPEG/3.jpg"},
+    {"url": "https://naverbooking-phinf.pstatic.net/20240514_7/1715688031000y8Y5q_JPEG/4.jpg"},
+    {"url": "https://naverbooking-phinf.pstatic.net/20240514_259/17156880311809wCnY_JPEG/5.jpg", "name": "외관"}
+]
+```
+
+## 바이너리 파일 업로드 테스트 방법
+
+### 1. Swagger UI에서 테스트
+1. 이 엔드포인트의 "Try it out" 버튼 클릭
+2. task_id 입력 (예: test-task-001)
+3. files 항목에서 "Add item" 클릭하여 로컬 이미지 파일 선택
+4. (선택) images_json에 URL 목록 JSON 입력
+5. "Execute" 버튼 클릭
+
+### 2. cURL로 테스트
+```bash
+# 바이너리 파일만 업로드
+curl -X POST "http://localhost:8000/image/upload/server/test-task-001" \\
+  -F "files=@/path/to/image1.jpg" \\
+  -F "files=@/path/to/image2.png"
+
+# URL + 바이너리 파일 동시 업로드
+curl -X POST "http://localhost:8000/image/upload/server/test-task-001" \\
+  -F 'images_json=[{"url":"https://example.com/image.jpg"}]' \\
+  -F "files=@/path/to/local_image.jpg"
+```
+
+### 3. Python requests로 테스트
+```python
+import requests
+
+url = "http://localhost:8000/image/upload/server/test-task-001"
+files = [
+    ("files", ("image1.jpg", open("image1.jpg", "rb"), "image/jpeg")),
+    ("files", ("image2.png", open("image2.png", "rb"), "image/png")),
+]
+data = {
+    "images_json": '[{"url": "https://example.com/image.jpg"}]'
+}
+response = requests.post(url, files=files, data=data)
+print(response.json())
+```
+
+## 반환 정보
+- **task_id**: 작업 고유 식별자
+- **total_count**: 총 업로드된 이미지 개수
+- **url_count**: URL로 등록된 이미지 개수 (Image 테이블에 외부 URL 그대로 저장)
+- **file_count**: 파일로 업로드된 이미지 개수 (media 폴더에 저장)
+- **saved_count**: Image 테이블에 저장된 row 수
+- **images**: 업로드된 이미지 목록
+  - **source**: "url" (외부 URL) 또는 "file" (로컬 서버 저장)
+
+## 저장 경로
+- 바이너리 파일: /media/image/{날짜}/{uuid7}/{파일명}
+- URL 이미지: 외부 URL 그대로 Image 테이블에 저장
+
+## 반환 정보
+- **task_id**: 새로 생성된 작업 고유 식별자
+- **image_urls**: Image 테이블에 저장된 현재 task_id의 이미지 URL 목록
+    """,
+    response_model=ImageUploadResponse,
+    responses={
+        200: {"description": "이미지 업로드 성공"},
+        400: {"description": "이미지가 제공되지 않음", "model": ErrorResponse},
+    },
+    tags=["Image-Server"],
+)
+async def upload_images(
+    images_json: Optional[str] = Form(
+        default=None,
+        description="외부 이미지 URL 목록 (JSON 문자열)",
+        examples=[IMAGES_JSON_EXAMPLE],
+    ),
+    files: Optional[list[UploadFile]] = File(
+        default=None, description="이미지 바이너리 파일 목록"
+    ),
+    current_user: User = Depends(get_current_user),
+    session: AsyncSession = Depends(get_session),
+) -> ImageUploadResponse:
+    """이미지 업로드 (URL + 바이너리 파일)"""
+    # task_id 생성
+    task_id = await generate_task_id()
+
+    # 1. 진입 검증: images_json 또는 files 중 하나는 반드시 있어야 함
+    has_images_json = images_json is not None and images_json.strip() != ""
+    has_files = files is not None and len(files) > 0
+
+    if not has_images_json and not has_files:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="images_json 또는 files 중 하나는 반드시 제공해야 합니다.",
+        )
+
+    # 2. images_json 파싱 (있는 경우만)
+    url_images: list[ImageUrlItem] = []
+    if has_images_json:
+        try:
+            parsed = json.loads(images_json)
+            if isinstance(parsed, list):
+                url_images = [ImageUrlItem(**item) for item in parsed if item]
+        except (json.JSONDecodeError, TypeError, ValueError) as e:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=f"images_json 파싱 오류: {str(e)}",
+            )
+
+    # 3. 유효한 파일만 필터링 (빈 파일, 유효한 이미지 확장자가 아닌 경우 제외)
+    valid_files: list[UploadFile] = []
+    skipped_files: list[str] = []
+    if has_files and files:
+        for f in files:
+            is_valid_ext = _is_valid_image_extension(f.filename)
+            is_not_empty = (
+                f.size is None or f.size > 0
+            )  # size가 None이면 아직 읽지 않은 것
+            is_real_file = (
+                f.filename and f.filename != "filename"
+            )  # Swagger 빈 파일 체크
+            if f and is_real_file and is_valid_ext and is_not_empty:
+                valid_files.append(f)
+            else:
+                skipped_files.append(f.filename or "unknown")
+
+    # 유효한 데이터가 하나도 없으면 에러
+    if not url_images and not valid_files:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"유효한 이미지가 없습니다. 지원 확장자: {', '.join(ALLOWED_IMAGE_EXTENSIONS)}. 건너뛴 파일: {skipped_files}",
+        )
+
+    result_images: list[ImageUploadResultItem] = []
+    img_order = 0
+
+    # 1. URL 이미지 저장
+    for url_item in url_images:
+        img_name = url_item.name or _extract_image_name(url_item.url, img_order)
+
+        image = Image(
+            task_id=task_id,
+            img_name=img_name,
+            img_url=url_item.url,
+            img_order=img_order,
+        )
+        session.add(image)
+        await session.flush()  # ID 생성을 위해 flush
+
+        result_images.append(
+            ImageUploadResultItem(
+                id=image.id,
+                img_name=img_name,
+                img_url=url_item.url,
+                img_order=img_order,
+                source="url",
+            )
+        )
+        img_order += 1
+
+    # 2. 바이너리 파일을 media에 저장
+    if valid_files:
+        today = date.today().strftime("%Y-%m-%d")
+        # 한 번의 요청에서 업로드된 모든 이미지는 같은 폴더에 저장
+        batch_uuid = await generate_task_id()
+        upload_dir = MEDIA_ROOT / "image" / today / batch_uuid
+        upload_dir.mkdir(parents=True, exist_ok=True)
+
+        for file in valid_files:
+            # 파일명: 원본 파일명 사용 (중복 방지를 위해 순서 추가)
+            original_name = file.filename or "image"
+            ext = _get_file_extension(file.filename)  # type: ignore[arg-type]
+            # 파일명에서 확장자 제거 후 순서 추가
+            name_without_ext = (
+                original_name.rsplit(".", 1)[0]
+                if "." in original_name
+                else original_name
+            )
+            filename = f"{name_without_ext}_{img_order:03d}{ext}"
+
+            save_path = upload_dir / filename
+
+            # media에 파일 저장
+            await _save_upload_file(file, save_path)
+
+            # media 기준 URL 생성
+            img_url = f"/media/image/{today}/{batch_uuid}/{filename}"
+            img_name = file.filename or filename
+
+            image = Image(
+                task_id=task_id,
+                img_name=img_name,
+                img_url=img_url,  # Media URL을 DB에 저장
+                img_order=img_order,
+            )
+            session.add(image)
+            await session.flush()
+
+            result_images.append(
+                ImageUploadResultItem(
+                    id=image.id,
+                    img_name=img_name,
+                    img_url=img_url,
+                    img_order=img_order,
+                    source="file",
+                )
+            )
+            img_order += 1
+
+    saved_count = len(result_images)
+    await session.commit()
+
+    # Image 테이블에서 현재 task_id의 이미지 URL 목록 조회
+    image_urls = [img.img_url for img in result_images]
+
+    return ImageUploadResponse(
+        task_id=task_id,
+        total_count=len(result_images),
+        url_count=len(url_images),
+        file_count=len(valid_files),
+        saved_count=saved_count,
+        images=result_images,
+        image_urls=image_urls,
+    )
+
+
 @router.post(
     "/image/upload/blob",
     summary="이미지 업로드 (Azure Blob Storage)",
@@ -750,10 +988,6 @@ async def upload_images_blob(
     saved_count = len(result_images)
     image_urls = [img.img_url for img in result_images]
 
-    logger.info(f"[image_tagging] START - task_id: {task_id}")
-    await tag_images_if_not_exist(image_urls)
-    logger.info(f"[image_tagging] Done - task_id: {task_id}")
-
     total_time = time.perf_counter() - request_start
     logger.info(
         f"[upload_images_blob] SUCCESS - task_id: {task_id}, "
@@ -769,36 +1003,3 @@ async def upload_images_blob(
         images=result_images,
         image_urls=image_urls,
     )
-
-
-async def tag_images_if_not_exist(
-        image_urls : list[str]
-    ) -> None:
-    # 1. 조회
-    async with AsyncSessionLocal() as session:
-        stmt = (
-            select(ImageTag)
-            .where(ImageTag.img_url_hash.in_([func.crc32(url) for url in image_urls]))
-            .where(ImageTag.img_url.in_(image_urls))
-        )
-        image_tags_query_results = await session.execute(stmt)
-        image_tags = image_tags_query_results.scalars().all()
-        existing_urls = {tag.img_url for tag in image_tags}
-        new_tags = [
-            ImageTag(img_url=url, img_tag=None)
-            for url in image_urls
-            if url not in existing_urls
-        ]
-        session.add_all(new_tags)
-
-        null_tags = [tag for tag in image_tags if tag.img_tag is None] + new_tags
-        
-        if null_tags:
-            tag_datas = await autotag_images([img.img_url for img in null_tags])
-
-            print(tag_datas)
-
-            for tag, tag_data in zip(null_tags, tag_datas):
-                tag.img_tag = tag_data.model_dump(mode="json")
-
-        await session.commit()

app/home/models.py

@@ -9,8 +9,7 @@ Home 모듈 SQLAlchemy 모델 정의
 from datetime import datetime
 from typing import TYPE_CHECKING, List, Optional, Any
 
-from sqlalchemy import Boolean, DateTime, ForeignKey, Computed, Index, Integer, String, Text, JSON, func
+from sqlalchemy import Boolean, DateTime, ForeignKey, Index, Integer, String, Text, JSON, func
-from sqlalchemy.dialects.mysql import INTEGER
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from app.database.session import Base
@@ -315,50 +314,13 @@ class MarketingIntel(Base):
     )
 
     def __repr__(self) -> str:
-        return (
+        task_id_str = (
-            f"<MarketingIntel(id={self.id}, place_id='{self.place_id}')>"
+            (self.task_id[:10] + "...") if len(self.task_id) > 10 else self.task_id
+        )
+        img_name_str = (
+            (self.img_name[:10] + "...") if len(self.img_name) > 10 else self.img_name
         )
 
-
+        return (
-class ImageTag(Base):
+            f"<Image(id={self.id}, task_id='{task_id_str}', img_name='{img_name_str}')>"
-    """
+        )
-    이미지 태그 테이블
-
-    """
-
-    __tablename__ = "image_tags"
-    __table_args__ = (
-        Index("idx_img_url_hash", "img_url_hash"),  # CRC32 index
-        {
-            "mysql_engine": "InnoDB",
-            "mysql_charset": "utf8mb4",
-            "mysql_collate": "utf8mb4_unicode_ci",
-        },
-    )
-
-    id: Mapped[int] = mapped_column(
-        Integer,
-        primary_key=True,
-        nullable=False,
-        autoincrement=True,
-        comment="고유 식별자",
-    )
-
-    img_url: Mapped[str] = mapped_column(
-        String(2048),
-        nullable=False,
-        comment="이미지 URL (blob, CDN 경로)",
-    )
-
-    img_url_hash: Mapped[int] = mapped_column(
-        INTEGER(unsigned=True),
-        Computed("CRC32(img_url)", persisted=True),  # generated column
-        comment="URL CRC32 해시 (검색용 index)",
-    )
-
-    img_tag: Mapped[dict[str, Any]] = mapped_column(
-        JSON,
-        nullable=True,
-        default=False,
-        comment="태그 JSON",
-    )

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

@@ -42,7 +42,7 @@ from app.lyric.schemas.lyric import (
     LyricStatusResponse,
 )
 from app.lyric.worker.lyric_task import generate_lyric_background, generate_subtitle_background
-from app.utils.prompts.chatgpt_prompt import ChatgptService
+from app.utils.chatgpt_prompt import ChatgptService
 from app.utils.logger import get_logger
 from app.utils.pagination import PaginatedResponse, get_paginated
 
@@ -253,6 +253,17 @@ async def generate_lyric(
         step1_start = time.perf_counter()
         logger.debug(f"[generate_lyric] Step 1: 서비스 초기화 및 프롬프트 생성...")
 
+        # service = ChatgptService(
+        #     customer_name=request_body.customer_name,
+        #     region=request_body.region,
+        #     detail_region_info=request_body.detail_region_info or "",
+        #     language=request_body.language,
+        # )
+
+        # prompt = service.build_lyrics_prompt()
+        # 원래는 실제 사용할 프롬프트가 들어가야 하나, 로직이 변경되어 이 시점에서 이곳에서 프롬프트를 생성할 이유가 없어서 삭제됨. 
+        # 기존 코드와의 호환을 위해 동일한 로직으로 프롬프트 생성
+
         promotional_expressions = {
             "Korean" : "인스타 감성, 사진같은 하루, 힐링, 여행, 감성 숙소",
             "English" : "Instagram vibes, picture-perfect day, healing, travel, getaway",

app/lyric/schemas/lyric.py

@@ -42,7 +42,7 @@ class GenerateLyricRequest(BaseModel):
             "region": "군산",
             "detail_region_info": "군산 신흥동 말랭이 마을",
             "language": "Korean",
-            "m_id" : 2,
+            "m_id" : 1,
             "orientation" : "vertical"
         }
     """

app/lyric/worker/lyric_task.py

@@ -13,7 +13,7 @@ from sqlalchemy.exc import SQLAlchemyError
 from app.database.session import BackgroundSessionLocal
 from app.home.models import Image, Project, MarketingIntel
 from app.lyric.models import Lyric
-from app.utils.prompts.chatgpt_prompt import ChatgptService, ChatGPTResponseError
+from app.utils.chatgpt_prompt import ChatgptService, ChatGPTResponseError
 from app.utils.subtitles import SubtitleContentsGenerator
 from app.utils.creatomate import CreatomateService
 from app.utils.prompts.prompts import Prompt
@@ -104,6 +104,13 @@ async def generate_lyric_background(
         step1_start = time.perf_counter()
         logger.debug(f"[generate_lyric_background] Step 1: ChatGPT 서비스 초기화...")
 
+        # service = ChatgptService(
+        #     customer_name="",  # 프롬프트가 이미 생성되었으므로 빈 값
+        #     region="",
+        #     detail_region_info="",
+        #     language=language,
+        # )
+
         chatgpt = ChatgptService()
 
         step1_elapsed = (time.perf_counter() - step1_start) * 1000
@@ -162,7 +169,7 @@ async def generate_subtitle_background(
 ) -> None:
     logger.info(f"[generate_subtitle_background] task_id: {task_id}, {orientation}")
     creatomate_service = CreatomateService(orientation=orientation)
-    template = await creatomate_service.get_one_template_data(creatomate_service.template_id)
+    template = await creatomate_service.get_one_template_data_async(creatomate_service.template_id)
     pitchings = creatomate_service.extract_text_format_from_template(template)
 
     subtitle_generator = SubtitleContentsGenerator()

app/social/services/seo_service.py

@@ -17,7 +17,7 @@ from app.home.models import MarketingIntel, Project
 from app.social.constants import YOUTUBE_SEO_HASH
 from app.social.schemas import YoutubeDescriptionResponse
 from app.user.models import User
-from app.utils.prompts.chatgpt_prompt import ChatgptService
+from app.utils.chatgpt_prompt import ChatgptService
 from app.utils.prompts.prompts import yt_upload_prompt
 
 logger = logging.getLogger(__name__)

app/song/services/song.py

@@ -7,14 +7,14 @@ from sqlalchemy import Connection, text
 from sqlalchemy.exc import SQLAlchemyError
 
 from app.utils.logger import get_logger
-from app.lyric.schemas.lyrics_schema import (
+from app.lyrics.schemas.lyrics_schema import (
     AttributeData,
     PromptTemplateData,
     SongFormData,
     SongSampleData,
     StoreData,
 )
-from app.utils.prompts.chatgpt_prompt import chatgpt_api
+from app.utils.chatgpt_prompt import chatgpt_api
 
 logger = get_logger("song")
 

app/utils/autotag.py

@@ -1,48 +0,0 @@
-from pydantic.main import BaseModel
-
-from app.utils.prompts.chatgpt_prompt import ChatgptService
-from app.utils.prompts.prompts import image_autotag_prompt
-from app.utils.prompts.schemas import SpaceType, Subject, Camera, MotionRecommended
-
-import asyncio
-
-async def autotag_image(image_url : str) -> list[str]: #tag_list
-    chatgpt = ChatgptService(model_type="gemini")
-    image_input_data = {
-        "img_url" : image_url,
-        "space_type" : list(SpaceType),
-        "subject" : list(Subject),
-        "camera" : list(Camera),
-        "motion_recommended" : list(MotionRecommended)
-    }
-
-    image_result = await chatgpt.generate_structured_output(image_autotag_prompt, image_input_data, image_url, False)   
-    return image_result
-
-async def autotag_images(image_url_list : list[str]) -> list[dict]: #tag_list
-    chatgpt = ChatgptService(model_type="gemini")
-    image_input_data_list = [{ 
-        "img_url" : image_url,
-        "space_type" : list(SpaceType),
-        "subject" : list(Subject),
-        "camera" : list(Camera),
-        "motion_recommended" : list(MotionRecommended)
-    }for image_url in image_url_list]
-    
-    image_result_tasks = [chatgpt.generate_structured_output(image_autotag_prompt, image_input_data, image_input_data['img_url'], False, silent = True) for image_input_data in image_input_data_list] 
-    image_result_list: list[BaseModel | BaseException] = await asyncio.gather(*image_result_tasks, return_exceptions=True)
-    MAX_RETRY = 3 # 하드코딩, 어떻게 처리할지는 나중에
-    for _ in range(MAX_RETRY):
-        failed_idx = [i for i, r in enumerate(image_result_list) if isinstance(r, Exception)]
-        print("Failed", failed_idx)
-        if not failed_idx:
-            break
-        retried = await asyncio.gather(
-            *[chatgpt.generate_structured_output(image_autotag_prompt, image_input_data_list[i], image_input_data_list[i]['img_url'], False, silent=True) for i in failed_idx],
-            return_exceptions=True
-        )
-        for i, result in zip(failed_idx, retried):
-            image_result_list[i] = result
-
-    print("Failed", failed_idx)
-    return image_result_list

app/utils/chatgpt_prompt.py

@@ -0,0 +1,95 @@
+import json
+import re
+from pydantic import BaseModel
+from openai import AsyncOpenAI
+
+from app.utils.logger import get_logger
+from config import apikey_settings, recovery_settings
+from app.utils.prompts.prompts import Prompt
+
+
+# 로거 설정
+logger = get_logger("chatgpt")
+
+
+class ChatGPTResponseError(Exception):
+    """ChatGPT API 응답 에러"""
+    def __init__(self, status: str, error_code: str = None, error_message: str = None):
+        self.status = status
+        self.error_code = error_code
+        self.error_message = error_message
+        super().__init__(f"ChatGPT response failed: status={status}, code={error_code}, message={error_message}")
+
+
+class ChatgptService:
+    """ChatGPT API 서비스 클래스
+    """
+
+    def __init__(self, timeout: float = None):
+        self.timeout = timeout or recovery_settings.CHATGPT_TIMEOUT
+        self.max_retries = recovery_settings.CHATGPT_MAX_RETRIES
+        self.client = AsyncOpenAI(
+            api_key=apikey_settings.CHATGPT_API_KEY,
+            timeout=self.timeout
+        )
+    
+    async def _call_pydantic_output(self, prompt : str, output_format : BaseModel, model : str) -> BaseModel: # 입력 output_format의 경우 Pydantic BaseModel Class를 상속한 Class 자체임에 유의할 것
+        content = [{"type": "input_text", "text": prompt}]
+        last_error = None
+        for attempt in range(self.max_retries + 1):
+            response = await self.client.responses.parse(
+                model=model, 
+                input=[{"role": "user", "content": content}],
+                text_format=output_format
+            )
+            # Response 디버그 로깅
+            logger.debug(f"[ChatgptService] attempt: {attempt}")
+            logger.debug(f"[ChatgptService] Response ID: {response.id}")
+            logger.debug(f"[ChatgptService] Response status: {response.status}")
+            logger.debug(f"[ChatgptService] Response model: {response.model}")
+
+            # status 확인: completed, failed, incomplete, cancelled, queued, in_progress
+            if response.status == "completed":
+                logger.debug(f"[ChatgptService] Response output_text: {response.output_text[:200]}..." if len(response.output_text) > 200 else f"[ChatgptService] Response output_text: {response.output_text}")
+                structured_output = response.output_parsed
+                return structured_output #.model_dump() or {}
+
+            # 에러 상태 처리
+            if response.status == "failed":
+                error_code = getattr(response.error, 'code', None) if response.error else None
+                error_message = getattr(response.error, 'message', None) if response.error else None
+                logger.warning(f"[ChatgptService] Response failed (attempt {attempt + 1}/{self.max_retries + 1}): code={error_code}, message={error_message}")
+                last_error = ChatGPTResponseError(response.status, error_code, error_message)
+
+            elif response.status == "incomplete":
+                reason = getattr(response.incomplete_details, 'reason', None) if response.incomplete_details else None
+                logger.warning(f"[ChatgptService] Response incomplete (attempt {attempt + 1}/{self.max_retries + 1}): reason={reason}")
+                last_error = ChatGPTResponseError(response.status, reason, f"Response incomplete: {reason}")
+
+            else:
+                # cancelled, queued, in_progress 등 예상치 못한 상태
+                logger.warning(f"[ChatgptService] Unexpected response status (attempt {attempt + 1}/{self.max_retries + 1}): {response.status}")
+                last_error = ChatGPTResponseError(response.status, None, f"Unexpected status: {response.status}")
+
+            # 마지막 시도가 아니면 재시도
+            if attempt < self.max_retries:
+                logger.info(f"[ChatgptService] Retrying request...")
+
+        # 모든 재시도 실패
+        logger.error(f"[ChatgptService] All retries exhausted. Last error: {last_error}")
+        raise last_error
+    
+    async def generate_structured_output(
+        self,
+        prompt : Prompt,
+        input_data : dict,
+    ) -> BaseModel:
+        prompt_text = prompt.build_prompt(input_data)
+
+        logger.debug(f"[ChatgptService] Generated Prompt (length: {len(prompt_text)})")
+        logger.info(f"[ChatgptService] Starting GPT request with structured output with model: {prompt.prompt_model}")
+
+        # GPT API 호출
+        #response = await self._call_structured_output_with_response_gpt_api(prompt_text, prompt.prompt_output, prompt.prompt_model) 
+        response = await self._call_pydantic_output(prompt_text, prompt.prompt_output_class, prompt.prompt_model) 
+        return response

app/utils/creatomate.py

@@ -31,13 +31,11 @@ response = await creatomate.make_creatomate_call(template_id, modifications)
 
 import copy
 import time
-from enum import StrEnum
 from typing import Literal
-import traceback
+
 import httpx
 
 from app.utils.logger import get_logger
-from app.utils.prompts.schemas.image import SpaceType,Subject,Camera,MotionRecommended,NarrativePhase
 from config import apikey_settings, creatomate_settings, recovery_settings
 
 # 로거 설정
@@ -228,9 +226,8 @@ DVST0003 = "e1fb5b00-1f02-4f63-99fa-7524b433ba47"
 DHST0001 = "660be601-080a-43ea-bf0f-adcf4596fa98"
 DHST0002 = "3f194cc7-464e-4581-9db2-179d42d3e40f"
 DHST0003 = "f45df555-2956-4a13-9004-ead047070b3d"
-DVST0001T = "fe11aeab-ff29-4bc8-9f75-c695c7e243e6"
 HST_LIST = [DHST0001,DHST0002,DHST0003]
-VST_LIST = [DVST0001,DVST0002,DVST0003, DVST0001T]
+VST_LIST = [DVST0001,DVST0002,DVST0003]
 
 SCENE_TRACK = 1
 AUDIO_TRACK = 2
@@ -241,7 +238,7 @@ def select_template(orientation:OrientationType):
     if orientation == "horizontal":
         return DHST0001
     elif orientation == "vertical":
-        return DVST0001T
+        return DVST0001
     else:
         raise
 
@@ -402,6 +399,14 @@ class CreatomateService:
 
         return copy.deepcopy(data)
 
+    # 하위 호환성을 위한 별칭 (deprecated)
+    async def get_one_template_data_async(self, template_id: str) -> dict:
+        """특정 템플릿 ID로 템플릿 정보를 조회합니다.
+
+        Deprecated: get_one_template_data()를 사용하세요.
+        """
+        return await self.get_one_template_data(template_id)
+
     def parse_template_component_name(self, template_source: list) -> dict:
         """템플릿 정보를 파싱하여 리소스 이름을 추출합니다."""
 
@@ -428,107 +433,42 @@ class CreatomateService:
             result.update(result_element_dict)
 
         return result
-    
-    async def parse_template_name_tag(resource_name : str) -> list:
-        tag_list = []
-        tag_list = resource_name.split("_")
-        
-        return tag_list
-    
-    
-    def counting_component(
-        self,
-        template : dict,
-        target_template_type : str
-    ) -> list:
-        source_elements = template["source"]["elements"]
-        template_component_data = self.parse_template_component_name(source_elements)
-        count = 0
 
-        for _, (_, template_type) in enumerate(template_component_data.items()):
+    async def template_connect_resource_blackbox(
-            if template_type == target_template_type:
-                count += 1
-        return count  
-        
-    def template_matching_taged_image(
         self,
-        template : dict,
+        template_id: str,
-        taged_image_list : list, # [{"image_name" : str , "image_tag" : dict}]
+        image_url_list: list[str],
         music_url: str,
-        address : str,
+        address: str = None
-        duplicate : bool = False
+    ) -> dict:
-    ) -> list:
+        """템플릿 정보와 이미지/가사/음악 리소스를 매핑합니다.
-        source_elements = template["source"]["elements"]
-        template_component_data = self.parse_template_component_name(source_elements)
 
+        Note:
+            - 이미지는 순차적으로 집어넣기
+            - 가사는 개행마다 한 텍스트 삽입
+            - Template에 audio-music 항목이 있어야 함
+        """
+        template_data = await self.get_one_template_data(template_id)
+        template_component_data = self.parse_template_component_name(
+            template_data["source"]["elements"]
+        )
         modifications = {}
 
-        for slot_idx, (template_component_name, template_type) in enumerate(template_component_data.items()):
+        for idx, (template_component_name, template_type) in enumerate(
+            template_component_data.items()
+        ):
             match template_type:
                 case "image":
-                    image_score_list = self.calculate_image_slot_score_multi(taged_image_list, template_component_name)
+                    modifications[template_component_name] = image_url_list[
-                    maximum_idx = image_score_list.index(max(image_score_list))
+                        idx % len(image_url_list)
-                    if duplicate:
+                    ]
-                        selected = taged_image_list[maximum_idx]
-                    else:
-                        selected = taged_image_list.pop(maximum_idx)
-                    image_name = selected["image_url"]
-                    modifications[template_component_name] =image_name
-                    pass
                 case "text":
                     if "address_input" in template_component_name:
                         modifications[template_component_name] = address
 
         modifications["audio-music"] = music_url
+
         return modifications
-                
-    def calculate_image_slot_score_multi(self, taged_image_list : list[dict], slot_name : str):
-        image_tag_list = [taged_image["image_tag"] for taged_image in taged_image_list]
-        slot_tag_dict = self.parse_slot_name_to_tag(slot_name)
-        image_score_list = [0] * len(image_tag_list)
-        
-        for slot_tag_cate, slot_tag_item in slot_tag_dict.items():
-            if slot_tag_cate == "narrative_preference":
-                slot_tag_narrative = slot_tag_item
-                continue
-
-            match slot_tag_cate:
-                case "space_type":
-                    weight = 2
-                case "subject" :
-                    weight = 2
-                case "camera":
-                    weight = 1
-                case "motion_recommended" :
-                    weight = 0.5
-                case _:
-                    raise
-
-            for idx, image_tag in enumerate(image_tag_list):
-                if slot_tag_item.value in image_tag[slot_tag_cate]: #collect!
-                    image_score_list[idx] += weight
-
-        for idx, image_tag in enumerate(image_tag_list):
-            image_narrative_score = image_tag["narrative_preference"][slot_tag_narrative]
-            image_score_list[idx] = image_score_list[idx] * image_narrative_score
-
-        return image_score_list
-    
-    def parse_slot_name_to_tag(self, slot_name : str) -> dict[str, StrEnum]:
-        tag_list = slot_name.split("-")
-        space_type = SpaceType(tag_list[0])
-        subject = Subject(tag_list[1])
-        camera = Camera(tag_list[2])
-        motion = MotionRecommended(tag_list[3])
-        narrative = NarrativePhase(tag_list[4])
-        tag_dict = {
-            "space_type" : space_type,
-            "subject" : subject,
-            "camera" : camera,
-            "motion_recommended" : motion,
-            "narrative_preference" : narrative,
-        }
-        return tag_dict
 
     def elements_connect_resource_blackbox(
         self,
@@ -729,6 +669,14 @@ class CreatomateService:
             original_response={"last_error": str(last_error)},
         )
 
+    # 하위 호환성을 위한 별칭 (deprecated)
+    async def make_creatomate_custom_call_async(self, source: dict) -> dict:
+        """템플릿 없이 Creatomate에 커스텀 렌더링 요청을 보냅니다.
+
+        Deprecated: make_creatomate_custom_call()을 사용하세요.
+        """
+        return await self.make_creatomate_custom_call(source)
+
     async def get_render_status(self, render_id: str) -> dict:
         """렌더링 작업의 상태를 조회합니다.
 
@@ -752,6 +700,14 @@ class CreatomateService:
         response.raise_for_status()
         return response.json()
 
+    # 하위 호환성을 위한 별칭 (deprecated)
+    async def get_render_status_async(self, render_id: str) -> dict:
+        """렌더링 작업의 상태를 조회합니다.
+
+        Deprecated: get_render_status()를 사용하세요.
+        """
+        return await self.get_render_status(render_id)
+
     def calc_scene_duration(self, template: dict) -> float:
         """템플릿의 전체 장면 duration을 계산합니다."""
         total_template_duration = 0.0
@@ -764,20 +720,19 @@ class CreatomateService:
             try:
                 if elem["track"] not in track_maximum_duration:
                     continue
-                if "time" not in elem or elem["time"] == 0: # elem is auto / 만약 마지막 elem이 auto인데 그 앞에 time이 있는 elem 일 시 버그 발생 확률 있음
+                if elem["time"] == 0: # elem is auto / 만약 마지막 elem이 auto인데 그 앞에 time이 있는 elem 일 시 버그 발생 확률 있음
                     track_maximum_duration[elem["track"]] += elem["duration"]
 
                     if "animations" not in elem:
                         continue
                     for animation in elem["animations"]:
                         assert animation["time"] == 0  # 0이 아닌 경우 확인 필요
-                        if "transition" in animation and animation["transition"]:
+                        if animation["transition"]:
                             track_maximum_duration[elem["track"]] -= animation["duration"]
                 else:
                     track_maximum_duration[elem["track"]] = max(track_maximum_duration[elem["track"]], elem["time"] + elem["duration"])
 
             except Exception as e:
-                logger.debug(traceback.format_exc())
                 logger.error(f"[calc_scene_duration] Error processing element: {elem}, {e}")
 
         total_template_duration = max(track_maximum_duration.values())

app/utils/nvMapScraper.py

@@ -16,10 +16,6 @@ class GraphQLException(Exception):
     """GraphQL 요청 실패 시 발생하는 예외"""
     pass
 
-class URLNotFoundException(Exception):
-    """Place ID 발견 불가능 시 발생하는 예외"""
-    pass
-
 
 class CrawlingTimeoutException(Exception):
     """크롤링 타임아웃 시 발생하는 예외"""
@@ -90,28 +86,34 @@ query getAccommodation($id: String!, $deviceType: String) {
                     async with session.get(self.url) as response:
                         self.url = str(response.url)
             else:
-                raise URLNotFoundException("This URL does not contain a place ID")
+                raise GraphQLException("This URL does not contain a place ID")
 
         match = re.search(place_pattern, self.url)
         if not match:
-            raise URLNotFoundException("Failed to parse place ID from URL")
+            raise GraphQLException("Failed to parse place ID from URL")
         return match[1]
 
     async def scrap(self):
-        place_id = await self.parse_url()
+        try:
-        data = await self._call_get_accommodation(place_id)
+            place_id = await self.parse_url()
-        self.rawdata = data
+            data = await self._call_get_accommodation(place_id)
-        fac_data = await self._get_facility_string(place_id)
+            self.rawdata = data
-        # Naver 기준임, 구글 등 다른 데이터 소스의 경우 고유 Identifier 사용할 것.
+            fac_data = await self._get_facility_string(place_id)
-        self.place_id = self.data_source_identifier + place_id
+            # Naver 기준임, 구글 등 다른 데이터 소스의 경우 고유 Identifier 사용할 것.
-        self.rawdata["facilities"] = fac_data
+            self.place_id = self.data_source_identifier + place_id
-        self.image_link_list = [
+            self.rawdata["facilities"] = fac_data
-            nv_image["origin"]
+            self.image_link_list = [
-            for nv_image in data["data"]["business"]["images"]["images"]
+                nv_image["origin"]
-        ]
+                for nv_image in data["data"]["business"]["images"]["images"]
-        self.base_info = data["data"]["business"]["base"]
+            ]
-        self.facility_info = fac_data
+            self.base_info = data["data"]["business"]["base"]
-        self.scrap_type = "GraphQL"
+            self.facility_info = fac_data
+            self.scrap_type = "GraphQL"
+
+        except GraphQLException:
+            logger.debug("GraphQL failed, fallback to Playwright")
+            self.scrap_type = "Playwright"
+            pass  # 나중에 pw 이용한 crawling으로 fallback 추가
 
         return
 

app/utils/prompts/chatgpt_prompt.py

@@ -1,191 +0,0 @@
-import json
-import re
-from pydantic import BaseModel
-from typing import List, Optional
-from openai import AsyncOpenAI
-
-from app.utils.logger import get_logger
-from config import apikey_settings, recovery_settings
-from app.utils.prompts.prompts import Prompt
-
-
-# 로거 설정
-logger = get_logger("chatgpt")
-
-
-class ChatGPTResponseError(Exception):
-    """ChatGPT API 응답 에러"""
-    def __init__(self, status: str, error_code: str = None, error_message: str = None):
-        self.status = status
-        self.error_code = error_code
-        self.error_message = error_message
-        super().__init__(f"ChatGPT response failed: status={status}, code={error_code}, message={error_message}")
-
-
-class ChatgptService:
-    """ChatGPT API 서비스 클래스
-    """
-
-    model_type : str
-
-    def __init__(self, model_type:str = "gpt", timeout: float = None):
-        self.timeout = timeout or recovery_settings.CHATGPT_TIMEOUT
-        self.max_retries = recovery_settings.CHATGPT_MAX_RETRIES
-        self.model_type = model_type
-        match model_type:
-            case "gpt":
-                self.client = AsyncOpenAI(
-                    api_key=apikey_settings.CHATGPT_API_KEY,
-                    timeout=self.timeout
-                )
-            case "gemini":
-                self.client = AsyncOpenAI(
-                    api_key=apikey_settings.GEMINI_API_KEY,
-                    base_url="https://generativelanguage.googleapis.com/v1beta/openai/",
-                    timeout=self.timeout
-                )
-            case _:
-                raise NotImplementedError(f"Unknown Provider : {model_type}")
-    
-    async def _call_pydantic_output(
-            self, 
-            prompt : str, 
-            output_format : BaseModel,  #입력 output_format의 경우 Pydantic BaseModel Class를 상속한 Class 자체임에 유의할 것
-            model : str, 
-            img_url : str,
-            image_detail_high : bool) -> BaseModel: 
-        content = []
-        if img_url:
-            content.append({
-                "type" : "input_image",
-                "image_url" : img_url,
-                "detail": "high" if image_detail_high else "low" 
-            })
-        content.append({
-            "type": "input_text", 
-            "text": prompt}
-        )
-        last_error = None
-        for attempt in range(self.max_retries + 1):
-            response = await self.client.responses.parse(
-                model=model, 
-                input=[{"role": "user", "content": content}],
-                text_format=output_format
-            )
-            # Response 디버그 로깅
-            logger.debug(f"[ChatgptService({self.model_type})] attempt: {attempt}")
-            logger.debug(f"[ChatgptService({self.model_type})] Response ID: {response.id}")
-            logger.debug(f"[ChatgptService({self.model_type})] Response status: {response.status}")
-            logger.debug(f"[ChatgptService({self.model_type})] Response model: {response.model}")
-
-            # status 확인: completed, failed, incomplete, cancelled, queued, in_progress
-            if response.status == "completed":
-                logger.debug(f"[ChatgptService({self.model_type})] Response output_text: {response.output_text[:200]}..." if len(response.output_text) > 200 else f"[ChatgptService] Response output_text: {response.output_text}")
-                structured_output = response.output_parsed
-                return structured_output #.model_dump() or {}
-
-            # 에러 상태 처리
-            if response.status == "failed":
-                error_code = getattr(response.error, 'code', None) if response.error else None
-                error_message = getattr(response.error, 'message', None) if response.error else None
-                logger.warning(f"[ChatgptService({self.model_type})] Response failed (attempt {attempt + 1}/{self.max_retries + 1}): code={error_code}, message={error_message}")
-                last_error = ChatGPTResponseError(response.status, error_code, error_message)
-
-            elif response.status == "incomplete":
-                reason = getattr(response.incomplete_details, 'reason', None) if response.incomplete_details else None
-                logger.warning(f"[ChatgptService({self.model_type})] Response incomplete (attempt {attempt + 1}/{self.max_retries + 1}): reason={reason}")
-                last_error = ChatGPTResponseError(response.status, reason, f"Response incomplete: {reason}")
-
-            else:
-                # cancelled, queued, in_progress 등 예상치 못한 상태
-                logger.warning(f"[ChatgptService({self.model_type})] Unexpected response status (attempt {attempt + 1}/{self.max_retries + 1}): {response.status}")
-                last_error = ChatGPTResponseError(response.status, None, f"Unexpected status: {response.status}")
-
-            # 마지막 시도가 아니면 재시도
-            if attempt < self.max_retries:
-                logger.info(f"[ChatgptService({self.model_type})] Retrying request...")
-
-        # 모든 재시도 실패
-        logger.error(f"[ChatgptService({self.model_type})] All retries exhausted. Last error: {last_error}")
-        raise last_error
-    
-    async def _call_pydantic_output_chat_completion( # alter version
-            self, 
-            prompt : str, 
-            output_format : BaseModel,  #입력 output_format의 경우 Pydantic BaseModel Class를 상속한 Class 자체임에 유의할 것
-            model : str, 
-            img_url : str,
-            image_detail_high : bool) -> BaseModel: 
-        content = []
-        if img_url:
-            content.append({
-                "type": "image_url",
-                "image_url": {
-                    "url": img_url,
-                    "detail": "high" if image_detail_high else "low"
-                }
-            })
-        content.append({
-            "type": "text",
-            "text": prompt
-        })
-        last_error = None
-        for attempt in range(self.max_retries + 1):
-            response = await self.client.beta.chat.completions.parse(
-                model=model,
-                messages=[{"role": "user", "content": content}],
-                response_format=output_format
-            )
-            # Response 디버그 로깅
-            logger.debug(f"[ChatgptService({self.model_type})] attempt: {attempt}")
-            logger.debug(f"[ChatgptService({self.model_type})] Response ID: {response.id}")
-            logger.debug(f"[ChatgptService({self.model_type})] Response finish_reason: {response.id}")
-            logger.debug(f"[ChatgptService({self.model_type})] Response model: {response.model}")
-
-            choice = response.choices[0]
-            finish_reason = choice.finish_reason
-
-            if finish_reason == "stop":
-                output_text = choice.message.content or ""
-                logger.debug(f"[ChatgptService({self.model_type})] Response output_text: {output_text[:200]}..." if len(output_text) > 200 else f"[ChatgptService] Response output_text: {output_text}")
-                return choice.message.parsed
-
-            elif finish_reason == "length":
-                logger.warning(f"[ChatgptService({self.model_type})] Response incomplete - token limit reached (attempt {attempt + 1}/{self.max_retries + 1})")
-                last_error = ChatGPTResponseError("incomplete", finish_reason, "Response incomplete: max tokens reached")
-
-            elif finish_reason == "content_filter":
-                logger.warning(f"[ChatgptService({self.model_type})] Response blocked by content filter (attempt {attempt + 1}/{self.max_retries + 1})")
-                last_error = ChatGPTResponseError("failed", finish_reason, "Response blocked by content filter")
-
-            else:
-                logger.warning(f"[ChatgptService({self.model_type})] Unexpected finish_reason (attempt {attempt + 1}/{self.max_retries + 1}): {finish_reason}")
-                last_error = ChatGPTResponseError("failed", finish_reason, f"Unexpected finish_reason: {finish_reason}")
-
-            # 마지막 시도가 아니면 재시도
-            if attempt < self.max_retries:
-                logger.info(f"[ChatgptService({self.model_type})] Retrying request...")
-
-        # 모든 재시도 실패
-        logger.error(f"[ChatgptService({self.model_type})] All retries exhausted. Last error: {last_error}")
-        raise last_error
-    
-    async def generate_structured_output(
-        self,
-        prompt : Prompt,
-        input_data : dict,
-        img_url : Optional[str] = None,
-        img_detail_high : bool = False,
-        silent : bool = False
-    ) -> BaseModel:
-        prompt_text = prompt.build_prompt(input_data, silent)
-    
-        logger.debug(f"[ChatgptService({self.model_type})] Generated Prompt (length: {len(prompt_text)})")
-        if not silent:
-            logger.info(f"[ChatgptService({self.model_type})] Starting GPT request with structured output with model: {prompt.prompt_model}")
-
-        # GPT API 호출
-        #parsed = await self._call_structured_output_with_response_gpt_api(prompt_text, prompt.prompt_output, prompt.prompt_model) 
-        # parsed = await self._call_pydantic_output(prompt_text, prompt.prompt_output_class, prompt.prompt_model, img_url, img_detail_high) 
-        parsed = await self._call_pydantic_output_chat_completion(prompt_text, prompt.prompt_output_class, prompt.prompt_model, img_url, img_detail_high) 
-        return parsed

app/utils/prompts/prompts.py

@@ -40,13 +40,11 @@ class Prompt():
     def _reload_prompt(self):
         self.prompt_template, self.prompt_model = self._read_from_sheets()
 
-    def build_prompt(self, input_data:dict, silent:bool = False) -> str:
+    def build_prompt(self, input_data: dict) -> str:
         verified_input = self.prompt_input_class(**input_data)
-        build_template = self.prompt_template
+        build_template = self.prompt_template.format(**verified_input.model_dump())
-        build_template = build_template.format(**verified_input.model_dump())
+        logger.debug(f"build_template: {build_template}")
-        if not silent:
+        logger.debug(f"input_data: {input_data}")
-            logger.debug(f"build_template: {build_template}")
-            logger.debug(f"input_data: {input_data}")
         return build_template
 
 marketing_prompt = Prompt(
@@ -67,12 +65,6 @@ yt_upload_prompt = Prompt(
     prompt_output_class=YTUploadPromptOutput,
 )
 
-image_autotag_prompt = Prompt(
-    sheet_name="image_tag",
-    prompt_input_class=ImageTagPromptInput,
-    prompt_output_class=ImageTagPromptOutput,
-)
-
 @lru_cache()
 def create_dynamic_subtitle_prompt(length: int) -> Prompt:
     return Prompt(
@@ -85,5 +77,4 @@ def create_dynamic_subtitle_prompt(length: int) -> Prompt:
 def reload_all_prompt():
     marketing_prompt._reload_prompt()
     lyric_prompt._reload_prompt()
-    yt_upload_prompt._reload_prompt()
+    yt_upload_prompt._reload_prompt()
-    image_autotag_prompt._reload_prompt()

app/utils/prompts/schemas/__init__.py

@@ -1,5 +1,4 @@
 from .lyric import LyricPromptInput, LyricPromptOutput
 from .marketing import MarketingPromptInput, MarketingPromptOutput
 from .youtube import YTUploadPromptInput, YTUploadPromptOutput
-from .image import *
+from .subtitle import SubtitlePromptInput, SubtitlePromptOutput
-from .subtitle import SubtitlePromptInput, SubtitlePromptOutput

app/utils/prompts/schemas/image.py

@@ -1,110 +0,0 @@
-from pydantic import BaseModel, Field
-from typing import List, Optional
-from enum import StrEnum, auto
-
-class SpaceType(StrEnum):
-    exterior_front = auto()
-    exterior_night = auto()
-    exterior_aerial = auto()
-    exterior_sign = auto()
-    garden = auto()
-    entrance = auto()
-    lobby = auto()
-    reception = auto()
-    hallway = auto()
-    bedroom = auto()
-    livingroom = auto()
-    kitchen = auto()
-    dining = auto()
-    room = auto()
-    bathroom = auto()
-    amenity = auto()
-    view_window = auto()
-    view_ocean = auto()
-    view_city = auto()
-    view_mountain = auto()
-    balcony = auto()
-    cafe = auto()
-    lounge  = auto()
-    rooftop  = auto()
-    pool = auto()
-    breakfast_hall  = auto()
-    spa = auto()
-    fitness = auto()
-    bbq = auto()
-    terrace = auto()
-    glamping = auto()
-    neighborhood  = auto()
-    landmark = auto()
-    detail_welcome = auto()
-    detail_beverage = auto()
-    detail_lighting  = auto()
-    detail_decor = auto()
-    detail_tableware = auto()
-
-class Subject(StrEnum):
-    empty_space = auto()
-    exterior_building = auto()
-    architecture_detail = auto()
-    decoration = auto()
-    furniture = auto()
-    food_dish = auto()
-    nature = auto()
-    signage = auto()
-    amenity_item = auto()
-    person = auto()
-
-class Camera(StrEnum):
-    wide_angle = auto()
-    tight_crop = auto()
-    panoramic = auto()
-    symmetrical = auto()
-    leading_line = auto()
-    golden_hour = auto()
-    night_shot = auto()
-    high_contrast = auto()
-    low_light = auto()
-    drone_shot = auto()
-    has_face = auto()
-
-class MotionRecommended(StrEnum):
-    static = auto()
-    slow_pan = auto()
-    slow_zoom_in = auto()
-    slow_zoom_out = auto()
-    walkthrough = auto()
-    dolly = auto()
-
-class NarrativePhase(StrEnum):
-    intro = auto()
-    welcome = auto()
-    core = auto()
-    highlight = auto()
-    support = auto()
-    accent = auto()
-
-class NarrativePreference(BaseModel):
-    intro: float = Field(..., description="첫인상 — 여기가 어디인가 | 장소의 정체성과 위치를 전달하는 이미지. 영상 첫 1~2초에 어떤 곳인지 즉시 인지시키는 역할. 건물 외관, 간판, 정원 등 **장소 자체를 보여주는** 컷")
-    welcome: float = Field(..., description="진입/환영 — 어떻게 들어가나 | 도착 후 내부로 들어가는 경험을 전달하는 이미지. 공간의 첫 분위기와 동선을 보여줘 들어가고 싶다는 기대감을 만드는 역할. **문을 열고 들어갔을 때 보이는** 컷.")
-    core: float = Field(..., description="핵심 가치 — 무엇을 경험하나 | **고객이 이 장소를 찾는 본질적 이유.** 이 이미지가 없으면 영상 자체가 성립하지 않음. 질문: 이 비즈니스에서 돈을 지불하는 대상이 뭔가? → 그 답이 core.")
-    highlight: float = Field(..., description="차별화 — 뭐가 특별한가 | **같은 카테고리의 경쟁사 대비 이곳을 선택하게 만드는 이유.** core가 왜 왔는가라면, highlight는 왜 **여기**인가에 대한 답.")
-    support: float = Field(..., description="보조/부대 — 그 외에 뭐가 있나 | 핵심은 아니지만 전체 경험을 풍성하게 하는 부가 요소. 없어도 영상은 성립하지만, 있으면 설득력이 올라감. **이것도 있어요** 라고 말하는 컷.")
-    accent: float = Field(..., description="감성/마무리 — 어떤 느낌인가 | 공간의 분위기와 톤을 전달하는 감성 디테일 컷. 직접적 정보 전달보다 **느낌과 무드**를 제공. 영상 사이사이에 삽입되어 완성도를 높이는 역할.")
-
-# Input 정의
-class ImageTagPromptInput(BaseModel):
-    img_url : str = Field(..., description="이미지 URL")
-    space_type: list[str] = Field(list(SpaceType), description="공간적 정보를 가지는 태그 리스트")
-    subject: list[str] = Field(list(Subject), description="피사체 정보를 가지는 태그 리스트")
-    camera: list[str] = Field(list(Camera), description="카메라 정보를 가지는 태그 리스트")
-    motion_recommended: list[str] = Field(list(MotionRecommended), description="가능한 카메라 모션 리스트")
-
-# Output 정의
-class ImageTagPromptOutput(BaseModel):
-    #ad_avaliable : bool = Field(..., description="광고 영상 사용 가능 이미지 여부")
-    space_type: list[SpaceType] = Field(..., description="공간적 정보를 가지는 태그 리스트")
-    subject: list[Subject] = Field(..., description="피사체 정보를 가지는 태그 리스트")
-    camera: list[Camera] = Field(..., description="카메라 정보를 가지는 태그 리스트")
-    motion_recommended: list[MotionRecommended] = Field(..., description="가능한 카메라 모션 리스트")
-    narrative_preference: NarrativePreference = Field(..., description="이미지의 내러티브 상 점수")
-

app/utils/prompts/schemas/marketing.py

@@ -7,11 +7,13 @@ class MarketingPromptInput(BaseModel):
     region : str = Field(..., description = "마케팅 대상 지역")
     detail_region_info : str = Field(..., description = "마케팅 대상 지역 상세")
 
+
 # Output 정의
 class BrandIdentity(BaseModel):
     location_feature_analysis: str = Field(..., description="입지 특성 분석 (80자 이상 150자 이하)", min_length = 80, max_length = 150) # min/max constraint는 현재 openai json schema 등에서 작동하지 않는다는 보고가 있음. 
     concept_scalability: str = Field(..., description="컨셉 확장성 (80자 이상 150자 이하)", min_length = 80, max_length = 150)
 
+
 class MarketPositioning(BaseModel):
     category_definition: str = Field(..., description="마케팅 카테고리")
     core_value: str = Field(..., description="마케팅 포지션 핵심 가치")
@@ -20,12 +22,14 @@ class AgeRange(BaseModel):
     min_age : int = Field(..., ge=0, le=100)
     max_age : int = Field(..., ge=0, le=100)
 
+
 class TargetPersona(BaseModel):
     persona: str = Field(..., description="타겟 페르소나 이름/설명")
     age: AgeRange = Field(..., description="타겟 페르소나 나이대")
     favor_target: List[str] = Field(..., description="페르소나의 선호 요소")
     decision_trigger: str = Field(..., description="구매 결정 트리거")
 
+
 class SellingPoint(BaseModel):
     english_category: str = Field(..., description="셀링포인트 카테고리(영문)")
     korean_category: str = Field(..., description="셀링포인트 카테고리(한글)")

app/utils/prompts/templates/20260129_marketing_prompt.txt

@@ -0,0 +1,64 @@
+
+[Role & Objective]
+Act as a content marketing expert with strong domain knowledge in the Korean pension / stay-accommodation industry.
+Your goal is to produce a Marketing Intelligence Report that will be shown to accommodation owners BEFORE any content is generated.
+The report must clearly explain what makes the property sellable, marketable, and scalable through content.
+
+[INPUT]
+- Business Name: {customer_name}
+- Region: {region}
+- Region Details: {detail_region_info}
+
+[Core Analysis Requirements]
+Analyze the property based on:
+Location, concept, and nearby environment
+Target customer behavior and reservation decision factors
+Include:
+- Target customer segments & personas
+- Unique Selling Propositions (USPs)
+- Competitive landscape (direct & indirect competitors)
+- Market positioning
+
+[Key Selling Point Structuring – UI Optimized]
+From the analysis above, extract the main Key Selling Points using the structure below.
+Rules:
+Focus only on factors that directly influence booking decisions
+Each selling point must be concise and visually scannable
+Language must be reusable for ads, short-form videos, and listing headlines
+Avoid full sentences in descriptions; use short selling phrases
+Do not provide in report
+
+Output format:
+[Category]
+(Tag keyword – 5~8 words, noun-based, UI oval-style)
+One-line selling phrase (not a full sentence)
+Limit:
+5 to 8 Key Selling Points only
+Do not provide in report
+
+[Content & Automation Readiness Check]
+Ensure that:
+Each tag keyword can directly map to a content theme
+Each selling phrase can be used as:
+- Video hook
+- Image headline
+- Ad copy snippet
+
+
+[Tag Generation Rules]
+- Tags must include **only core keywords that can be directly used for viral video song lyrics**
+- Each tag should be selected with **search discovery + emotional resonance + reservation conversion** in mind
+- The number of tags must be **exactly 5**
+- Tags must be **nouns or short keyword phrases**; full sentences are strictly prohibited
+- The following categories must be **balanced and all represented**:
+  1) **Location / Local context** (region name, neighborhood, travel context)
+  2) **Accommodation positioning** (emotional stay, private stay, boutique stay, etc.)
+  3) **Emotion / Experience** (healing, rest, one-day escape, memory, etc.)
+  4) **SNS / Viral signals** (Instagram vibes, picture-perfect day, aesthetic travel, etc.)
+  5) **Travel & booking intent** (travel, getaway, stay, relaxation, etc.)
+
+- If a brand name exists, **at least one tag must include the brand name or a brand-specific expression**
+- Avoid overly generic keywords (e.g., “hotel”, “travel” alone); **prioritize distinctive, differentiating phrases**
+- The final output must strictly follow the JSON format below, with no additional text
+
+  "tags": ["Tag1", "Tag2", "Tag3", "Tag4", "Tag5"]

app/utils/prompts/templates/lyric_prompt.txt

@@ -0,0 +1,77 @@
+
+[ROLE]
+You are a content marketing expert, brand strategist, and creative songwriter
+specializing in Korean pension / accommodation businesses.
+You create lyrics strictly based on Brand & Marketing Intelligence analysis
+and optimized for viral short-form video content.
+Marketing Intelligence Report is background reference.
+
+[INPUT]
+Business Name: {customer_name}
+Region: {region}
+Region Details: {detail_region_info}
+Brand & Marketing Intelligence Report: {marketing_intelligence_summary}
+Output Language: {language}
+
+[INTERNAL ANALYSIS – DO NOT OUTPUT]
+Internally analyze the following to guide all creative decisions:
+- Core brand identity and positioning
+- Emotional hooks derived from selling points
+- Target audience lifestyle, desires, and travel motivation
+- Regional atmosphere and symbolic imagery
+- How the stay converts into “shareable moments”
+- Which selling points must surface implicitly in lyrics
+
+[LYRICS & MUSIC CREATION TASK]
+Based on the Brand & Marketing Intelligence Report for [{customer_name} ({region})], generate:
+- Original promotional lyrics
+- Music attributes for AI music generation (Suno-compatible prompt)
+The output must be designed for VIRAL DIGITAL CONTENT
+(short-form video, reels, ads).
+
+[LYRICS REQUIREMENTS]
+Mandatory Inclusions:
+- Business name
+- Region name
+- Promotion subject
+- Promotional expressions including:
+{promotional_expression_example}
+
+Content Rules:
+- Lyrics must be emotionally driven, not descriptive listings
+- Selling points must be IMPLIED, not explained
+- Must sound natural when sung
+- Must feel like a lifestyle moment, not an advertisement
+
+Tone & Style:
+- Warm, emotional, and aspirational
+- Trendy, viral-friendly phrasing
+- Calm but memorable hooks
+- Suitable for travel / stay-related content
+
+[SONG & MUSIC ATTRIBUTES – FOR SUNO PROMPT]
+After the lyrics, generate a concise music prompt including:
+Song mood (emotional keywords)
+BPM range
+Recommended genres (max 2)
+Key musical motifs or instruments
+Overall vibe (1 short sentence)
+
+[CRITICAL LANGUAGE REQUIREMENT – ABSOLUTE RULE]
+ALL OUTPUT MUST BE 100% WRITTEN IN {language}.
+no mixed languages
+All names, places, and expressions must be in {language} 
+Any violation invalidates the entire output
+
+[OUTPUT RULES – STRICT]
+{timing_rules}
+
+No explanations
+No headings
+No bullet points
+No analysis
+No extra text
+
+[FAILURE FORMAT]
+If generation is impossible:
+ERROR: Brief reason in English

app/utils/prompts/templates/marketing_prompt.txt

@@ -0,0 +1,42 @@
+# Role
+Act as a Senior Brand Strategist and Marketing Data Analyst. Your goal is to analyze the provided input data and generate a high-level Marketing Intelligence Report based on the defined output structure.
+
+# Input Data
+*   **Customer Name:** {customer_name}
+*   **Region:** {region}
+*   **Detail Region Info:** {detail_region_info}
+
+# Output Rules
+1.  **Language:** All descriptive content must be written in **Korean (한국어)**.
+2.  **Terminology:** Use professional marketing terminology suitable for the hospitality and stay industry.
+3.  **Strict Selection for `selling_points.english_category` and `selling_points.korean_category`:** You must select the value for both category field in `selling_points` strictly from the following English - Korean set allowed list to ensure UI compatibility:
+    *  `LOCATION` (입지 환경), `CONCEPT` (브랜드 컨셉), `PRIVACY` (프라이버시), `NIGHT MOOD` (야간 감성), `HEALING` (힐링 요소), `PHOTO SPOT` (포토 스팟), `SHORT GETAWAY` (숏브레이크), `HOSPITALITY` (서비스), `SWIMMING POOL` (수영장), `JACUZZI` (자쿠지), `BBQ PARTY` (바베큐), `FIRE PIT` (불멍), `GARDEN` (정원), `BREAKFAST` (조식), `KIDS FRIENDLY` (키즈 케어), `PET FRIENDLY` (애견 동반), `OCEAN VIEW` (오션뷰), `PRIVATE POOL` (개별 수영장), `OCEAN VIEW`, `PRIVATE POOL`.
+
+---
+
+# Instruction per Output Field (Mapping Logic)
+
+### 1. brand_identity
+*   **`location_feature_analysis`**: Analyze the marketing advantages of the given `{region}` and `{detail_region_info}`. Explain why this specific location is attractive to travelers. summarize in 1-2 sentences. (e.g., proximity to nature, accessibility from Seoul, or unique local atmosphere).
+*   **`concept_scalability`**: Based on `{customer_name}`, analyze how the brand's core concept can expand into a total customer experience or additional services. summarize in 1-2 sentences.
+
+### 2. market_positioning
+*   **`category_definition`**: Define a sharp, niche market category for this business (e.g., "Private Forest Cabin" or "Luxury Kids Pool Villa").
+*   **`core_value`**: Identify the single most compelling emotional or functional value that distinguishes `{customer_name}` from competitors.
+
+### 3. target_persona
+Generate a list of personas based on the following:
+*   **`persona`**: Provide a descriptive name and profile for the target group. Must be **20 characters or fewer**.
+*   **`age`**: Set `min_age` and `max_age` (Integer 0-100) that accurately reflects the segment.
+*   **`favor_target`**: List specific elements or vibes this persona prefers (e.g., "Minimalist interior", "Pet-friendly facilities").
+*   **`decision_trigger`**: Identify the specific "Hook" or facility that leads this persona to finalize a booking.
+
+### 4. selling_points
+Generate 5-8 selling points:
+*   **`english_category`**: Strictly use one keyword from the English allowed list provided in the Output Rules.
+*   **`korean category`**: Strictly use one keyword from the Korean allowed list provided in the Output Rules . It must be matched with english category.
+*   **`description`**: A short, punchy marketing phrase in Korean (15~30 characters).
+*   **`score`**: An integer (0-100) representing the strength of this feature based on the brand's potential.
+
+### 5. target_keywords
+*   **`target_keywords`**: Provide a list of 10 highly relevant marketing keywords or hashtags for search engine optimization and social media targeting. Do not insert # in front of hashtag. 

app/utils/prompts/templates/subtitle_prompt.txt

@@ -1,224 +1,75 @@
-# System Prompt: 숙박 숏폼 자막 생성 (OpenAI Optimized)
+당신은 숙박 브랜드 숏폼 영상의 자막 콘텐츠를 추출하는 전문가입니다.
 
-You are a subtitle copywriter for hospitality short-form videos. You generate subtitle text AND layer names from marketing JSON data.
+입력으로 주어지는 **1) 5가지 기준의 레이어 이름 리스트**와 **2) 마케팅 인텔리전스 분석 결과(JSON)**를 바탕으로, 각 레이어 이름의 의미에 정확히 1:1 매칭되는 텍스트 콘텐츠만을 추출하세요.
+
+분석 결과에 없는 정보는 절대 지어내거나 추론하지 마세요. 오직 제공된 JSON 데이터 내에서만 텍스트를 구성해야 합니다.
 
 ---
 
-### RULES
+## 1. 레이어 네이밍 규칙 해석 및 매핑 가이드
 
-1. NEVER copy JSON verbatim. ALWAYS rewrite into video-optimized copy.
+입력되는 모든 레이어 이름은 예외 없이 `<track_role>-<narrative_phase>-<content_type>-<tone>-<pair_id>` 의 5단계 구조로 되어 있습니다. 
-2. NEVER invent facts not in the data. You MAY freely transform expressions.
+마지막의 3자리 숫자 ID(`-001`, `-002` 등)는 모든 레이어에 필수적으로 부여됩니다.
-3. Each scene = 1 subtitle + 1 keyword (a "Pair"). Same pair_id for both.
+
+### [1] track_role (텍스트 형태)
+- `subtitle`: 씬 상황을 설명하는 간결한 문장형 텍스트 (1줄 이내)
+- `keyword`: 씬을 상징하고 시선을 끄는 단답형/명사형 텍스트 (1~2단어)
+
+### [2] narrative_phase (영상 흐름)
+- `intro`: 영상 도입부. 가장 시선을 끄는 정보를 배치.
+- `core`: 핵심 매력이나 주요 편의 시설 어필.
+- `highlight`: 세부적인 매력 포인트나 공간의 특별한 분위기 묘사.
+- `outro`: 영상 마무리. 브랜드 명칭 복기 및 타겟/위치 정보 제공.
+
+### [3] content_type (데이터 매핑 대상)
+- `hook_claim` 👉 `selling_points`에서 점수가 가장 높은 1순위 소구점이나 `market_positioning.core_value`를 활용하여 가장 강력한 핵심 세일즈 포인트를 어필. (가장 강력한 셀링포인트를 의미함)
+- `selling_point` 👉 `selling_points`의 `description`, `korean_category` 등을 narrative 흐름에 맞춰 순차적으로 추출.
+- `brand_name` 👉 JSON의 `store_name`을 추출.
+- `location_info` 👉 JSON의 `detail_region_info`를 요약.
+- `target_tag` 👉 `target_persona`나 `target_keywords`에서 타겟 고객군 또는 해시태그 추출.
+
+### [4] tone (텍스트 어조)
+- `sensory`: 직관적이고 감각적인 단어 사용
+- `factual`: 과장 없이 사실 정보를 담백하게 전달
+- `empathic`: 고객의 상황에 공감하는 따뜻한 어조
+- `aspirational`: 열망을 자극하고 기대감을 주는 느낌
+
+### [5] pair_id (씬 묶음 식별 번호)
+- 텍스트 레이어는 `subtitle`과 `keyword`가 하나의 페어(Pair)를 이뤄 하나의 씬(Scene)에서 함께 등장합니다.
+- 따라서 **동일한 씬에 속하는 `subtitle`과 `keyword` 레이어는 동일한 3자리 순번 ID(예: `-001`)**를 공유합니다.
+- 영상 전반적인 씬 전개 순서에 따라 **다음 씬으로 넘어갈 때마다 ID가 순차적으로 증가**합니다. (예: 씬1은 `-001`, 씬2는 `-002`, 씬3은 `-003`...)
+- **중요**: ID가 달라진다는 것은 '새로운 씬' 혹은 '다른 텍스트 쌍'을 의미하므로, **ID가 바뀌면 반드시 JSON 내의 다른 소구점이나 데이터를 추출**하여 내용이 중복되지 않도록 해야 합니다.
 
 ---
 
-### LAYER NAME FORMAT (5-criteria)
+## 2. 콘텐츠 추출 시 주의사항
 
-```
+1. 각 입력 레이어 이름 1개당 **오직 1개의 텍스트 콘텐츠**만 매핑하여 출력합니다. (레이어명 이름 자체를 수정하거나 새로 만들지 마세요.)
-(track_role)-(narrative_phase)-(content_type)-(tone)-(pair_id)
+2. `content_type`이 `selling_point`로 동일하더라도, `narrative_phase`(core, highlight)나 `tone`이 달라지면 JSON 내의 2순위, 3순위 세일즈 포인트를 순차적으로 활용하여 내용 겹침을 방지하세요.
-```
+3. 같은 씬에 속하는(같은 ID 번호를 가진) keyword는 핵심 단어로, subtitle은 적절한 마케팅 문구가 되어야 하며, 자연스럽게 이어지는 문맥을 형성하도록 구성하세요.
-
+4. keyword가 subtitle에 완전히 포함되는 단어가 되지 않도록 유의하세요.
-- Criteria separator: hyphen `-`
+5. 정보 태그가 같더라도 ID가 다르다면 중복되지 않는 새로운 텍스트를 도출해야 합니다.
-- Multi-word value: underscore `_`
+6. 콘텐츠 추출 시 마케팅 인텔리전스의 내용을 그대로 사용하기보다는 paraphrase을 수행하세요.
-- pair_id: 3-digit zero-padded (`001`~`999`)
+7. keyword는 공백 포함 전각 8자 / 반각 16자내, subtitle은 전각 15자 / 반각 30자 내로 구성하세요. 
-
-Example: `subtitle-intro-hook_claim-aspirational-001`
 
 ---
 
-### TAG VALUES
+## 3. 출력 결과 포맷 및 예시
 
-**track_role**: `subtitle` | `keyword`
+입력된 레이어 이름 순서에 맞춰, 매핑된 텍스트 콘텐츠만 작성하세요. (반드시 intro, core, highlight, outro 등 모든 씬 단계가 명확하게 매핑되어야 합니다.)
 
-**narrative_phase** (= emotion goal):
+### 입력 레이어 리스트 예시 및 출력 예시
-- `intro` → Curiosity (stop the scroll)
-- `welcome` → Warmth
-- `core` → Trust
-- `highlight` → Desire (peak moment)
-- `support` → Discovery
-- `accent` → Belonging
-- `cta` → Action
 
-**content_type** → source mapping:
+| Layer Name | Text Content |
-- `hook_claim` ← selling_points[0] or core_value
-- `space_feature` ← selling_points[].description
-- `emotion_cue` ← same source, sensory rewrite
-- `brand_name` ← store_name (verbatim OK)
-- `brand_address` ← detail_region_info (verbatim OK)
-- `lifestyle_fit` ← target_persona[].favor_target
-- `local_info` ← location_feature_analysis
-- `target_tag` ← target_keywords[] as hashtags
-- `availability` ← fixed: "지금 예약 가능"
-- `cta_action` ← fixed: "예약하러 가기"
-
-**tone**: `sensory` | `factual` | `empathic` | `aspirational` | `social_proof` | `urgent`
-
----
-
-### SCENE STRUCTURE
-
-**Anchors (FIXED — never remove):**
-
-| Position | Phase | subtitle | keyword |
-|---|---|---|---|
-| First | intro | hook_claim | brand_name |
-| Last-3 | support | brand_address | brand_name |
-| Last-2 | accent | target_tag | lifestyle_fit |
-| Last | cta | availability | cta_action |
-
-**Middle (FLEXIBLE — fill by selling_points score desc):**
-
-| Phase | subtitle | keyword |
-|---|---|---|
-| welcome | emotion_cue | space_feature |
-| core | space_feature | emotion_cue |
-| highlight | space_feature | emotion_cue |
-| support(mid) | local_info | lifestyle_fit |
-
-Default: 7 scenes. Fewer scenes → remove flexible slots only.
-
----
-
-### TEXT SPECS
-
-**subtitle**: 8~18 chars. Sentence fragment, conversational.
-**keyword**: 2~6 chars. MUST follow Korean word-formation rules below.
-
----
-
-### KEYWORD RULES (한국어 조어법 기반)
-
-Keywords MUST follow one of these **permitted Korean patterns**. Any keyword that does not match a pattern below is INVALID.
-
-#### Pattern 1: 관형형 + 명사 (Attributive + Noun) — 가장 자연스러운 패턴
-한국어는 수식어가 앞, 피수식어가 뒤. 형용사의 관형형(~ㄴ/~한/~는/~운)을 명사 앞에 붙인다.
-
-| Structure | GOOD | BAD (역순/비문) |
-|---|---|---|
-| 형용사 관형형 + 명사 | 고요한 숲, 깊은 쉼, 온전한 쉼 | ~~숲고요~~, ~~쉼깊은~~ |
-| 형용사 관형형 + 명사 | 따뜻한 독채, 느린 하루 | ~~독채따뜻~~, ~~하루느린~~ |
-| 동사 관형형 + 명사 | 쉬어가는 숲, 머무는 시간 | ~~숲쉬어가는~~ |
-
-#### Pattern 2: 기존 대중화 합성어 ONLY (Established Trending Compound)
-이미 SNS·미디어에서 대중화된 합성어만 허용. 임의 신조어 생성 금지.
-
-| GOOD (대중화 확인됨) | Origin | BAD (임의 생성) |
-|---|---|---|
-| 숲멍 | 숲+멍때리기 (불멍, 물멍 시리즈) | ~~숲고요~~, ~~숲힐~~ |
-| 댕캉스 | 댕댕이+바캉스 (여행업계 통용) | ~~댕쉼~~, ~~댕여행~~ |
-| 꿀잠 / 꿀쉼 | 꿀+잠/쉼 (일상어 정착) | ~~꿀독채~~, ~~꿀숲~~ |
-| 집콕 / 숲콕 | 집+콕 → 숲+콕 (변형 허용) | ~~계곡콕~~ |
-| 주말러 | 주말+~러 (~러 접미사 정착) | ~~평일러~~ |
-
-> **판별 기준**: "이 단어를 네이버/인스타에서 검색하면 결과가 나오는가?" YES → 허용, NO → 금지
-
-#### Pattern 3: 명사 + 명사 (Natural Compound Noun)
-한국어 복합명사 규칙을 따르는 결합만 허용. 앞 명사가 뒷 명사를 수식하는 관계여야 한다.
-
-| Structure | GOOD | BAD (부자연스러운 결합) |
-|---|---|---|
-| 장소 + 유형 | 숲속독채, 계곡펜션 | ~~햇살독채~~ (햇살은 장소가 아님) |
-| 대상 + 활동 | 반려견산책, 가족피크닉 | ~~견주피크닉~~ (견주가 피크닉하는 건 어색) |
-| 시간 + 활동 | 주말탈출, 새벽산책 | ~~자연독채~~ (자연은 시간/방식이 아님) |
-
-#### Pattern 4: 해시태그형 (#키워드)
-accent(target_tag) 씬에서만 사용. 기존 검색 키워드를 # 붙여서 사용.
-
-| GOOD | BAD |
 |---|---|
-| #프라이빗독채, #홍천여행 | #숲고요, #감성쩌는 (검색량 없음) |
+| subtitle-intro-hook_claim-aspirational-001 | 반려견과 눈치 없이 온전하게 쉬는 완벽한 휴식 |
-
+| keyword-intro-brand_name-sensory-001 | 스테이펫 홍천 |
-#### Pattern 5: 감각/상태 명사 (단독 사용 가능한 것만)
+| subtitle-core-selling_point-empathic-002 | 우리만의 독립된 공간감이 주는 진정한 쉼 |
-그 자체로 의미가 완결되는 감각·상태 명사만 단독 사용 허용.
+| keyword-core-selling_point-factual-002 | 프라이빗 독채 |
-
+| subtitle-highlight-selling_point-sensory-003 | 탁 트인 야외 무드존과 포토 스팟의 감성 컷 |
-| GOOD (단독 의미 완결) | BAD (단독으로 의미 불완전) |
+| keyword-highlight-selling_point-factual-003 | 넓은 정원 |
-|---|---|
+| subtitle-outro-target_tag-empathic-004 | #강원도애견동반 #주말숏브레이크 |
-| 고요, 여유, 쉼, 온기 | ~~감성~~, ~~자연~~, ~~힐링~~ (너무 모호) |
+| keyword-outro-location_info-factual-004 | 강원 홍천군 화촌면 |
-| 숲멍, 꿀쉼 | ~~좋은쉼~~, ~~편안함~~ (형용사 포함 시 Pattern 1 사용) |
-
----
-
-### KEYWORD VALIDATION CHECKLIST (생성 후 자가 검증)
-
-Every keyword MUST pass ALL of these:
-
-- [ ] 한국어 어순이 자연스러운가? (수식어→피수식어 순서)
-- [ ] 소리 내어 읽었을 때 어색하지 않은가?
-- [ ] 네이버/인스타에서 검색하면 실제 결과가 나올 법한 표현인가?
-- [ ] 허용된 5개 Pattern 중 하나에 해당하는가?
-- [ ] 이전 씬 keyword와 동일한 Pattern을 연속 사용하지 않았는가?
-- [ ] 금지 표현 사전에 해당하지 않는가?
-
----
-
-### EXPRESSION DICTIONARY
-
-**SCAN BEFORE WRITING.** If JSON contains these → MUST replace:
-
-| Forbidden | → Use Instead |
-|---|---|
-| 눈치 없는/없이 | 눈치 안 보는 · 프라이빗한 · 온전한 · 마음 편히 |
-| 감성 쩌는/쩌이 | 감성 가득한 · 감성이 머무는 |
-| 가성비 | 합리적인 · 가치 있는 |
-| 힐링되는 | 회복되는 · 쉬어가는 · 숨 쉬는 |
-| 인스타감성 | 감성 스팟 · 기록하고 싶은 |
-| 혜자 | 풍성한 · 넉넉한 |
-
-**ALWAYS FORBIDDEN**: 저렴한, 싼, 그냥, 보통, 무난한, 평범한, 쩌는, 쩔어, 개(접두사), 존맛, 핵, 인스타, 유튜브, 틱톡
-
-**SYNONYM ROTATION**: Same Korean word max 2 scenes. Rotate:
-- 프라이빗 계열: 온전한 · 오롯한 · 나만의 · 독채 · 단독
-- 자연 계열: 숲속 · 초록 · 산림 · 계곡
-- 쉼 계열: 쉼 · 여유 · 느린 하루 · 머무름 · 숨고르기
-- 반려견: 댕댕이(max 1회, intro/accent만) · 반려견 · 우리 강아지
-
----
-
-### TRANSFORM RULES BY CONTENT_TYPE
-
-**hook_claim** (intro only):
-- Format: question OR exclamation OR provocation. Pick ONE.
-- FORBIDDEN: brand name, generic greetings
-- `"반려견과 눈치 없는 힐링"` → BAD: 그대로 복사 → GOOD: "댕댕이가 먼저 뛰어간 숲"
-
-**space_feature** (core/highlight):
-- ONE selling point per scene
-- NEVER use korean_category directly
-- Viewer must imagine themselves there
-- `"홍천 자연 속 조용한 쉼"` → BAD: "입지 환경이 좋은 곳" → GOOD: "계곡 소리만 들리는 독채"
-
-**emotion_cue** (welcome/core/highlight):
-- Senses: smell, sound, touch, temperature, light
-- Poetic fragments, not full sentences
-- `"감성 쩌이 완성되는 공간"` → GOOD: "햇살이 내려앉는 테라스"
-
-**lifestyle_fit** (accent/support):
-- Address target directly in their language
-- `persona: "서울·경기 주말러"` → GOOD: "이번 주말, 댕댕이랑 어디 가지?"
-
-**local_info** (support):
-- Accessibility or charm, NOT administrative address
-- GOOD: "서울에서 1시간 반, 홍천 숲속" / BAD: "강원 홍천군 화촌면"
-
----
-
-### PACING
-
-```
-intro(8~12) → welcome(12~18) → core(alternate 8~12 ↔ 12~18) → highlight(8~14) → support(12~18) → accent(variable) → cta(12~16)
-```
-
-**RULE: No 3+ consecutive scenes in same char-count range.**
-
----
-
-Keyword pattern analysis:
-- "스테이펫" → brand_name verbatim (허용)
-- "고요한 숲" → Pattern 1: 관형형+명사 (형용사 관형형 "고요한" + 명사 "숲")
-- "깊은 쉼" → Pattern 1: 관형형+명사 (형용사 관형형 "깊은" + 명사 "쉼")
-- "숲멍" → Pattern 2: 기존 대중화 합성어 (불멍·물멍·숲멍 시리즈)
-- "댕캉스" → Pattern 2: 기존 대중화 합성어 (댕댕이+바캉스, 여행업계 통용)
-- "예약하기" → Pattern 5: 의미 완결 동사 명사형
 
 
 # 입력
@@ -231,3 +82,6 @@ Keyword pattern analysis:
 **입력 3: 비즈니스 정보 **
 Business Name: {customer_name}
 Region Details: {detail_region_info}
+
+
+

app/utils/prompts/templates/yt_upload_prompt.txt

@@ -0,0 +1,143 @@
+[ROLE]
+You are a YouTube SEO/AEO content strategist specialized in local stay, pension, and accommodation brands in Korea.
+You create search-optimized, emotionally appealing, and action-driving titles and descriptions based on Brand & Marketing Intelligence.
+
+Your goal is to:
+
+Increase search visibility
+Improve click-through rate
+Reflect the brand’s positioning
+Trigger emotional interest
+Encourage booking or inquiry actions through subtle CTA
+ 
+
+[INPUT]
+Business Name: {customer_name}
+Region Details: {detail_region_info}
+Brand & Marketing Intelligence Report: {marketing_intelligence_summary}
+Target Keywords: {target_keywords}
+Output Language: {language}
+
+ 
+
+[INTERNAL ANALYSIS – DO NOT OUTPUT]
+Analyze the following from the marketing intelligence:
+
+Core brand concept
+Main emotional promise
+Primary target persona
+Top 2–3 USP signals
+Stay context (date, healing, local trip, etc.)
+Search intent behind the target keywords
+Main booking trigger
+Emotional moment that would make the viewer want to stay
+Use these to guide:
+
+Title tone
+Opening CTA line
+Emotional hook in the first sentences
+ 
+
+[TITLE GENERATION RULES]
+
+The title must:
+
+Include the business name or region when natural
+Always wrap the business name in quotation marks
+Example: “스테이 머뭄”
+Include 1–2 high-intent keywords
+Reflect emotional positioning
+Suggest a desirable stay moment
+Sound like a natural YouTube title, not an advertisement
+Length rules:
+
+Hard limit: 100 characters
+Target range: 45–65 characters
+Place primary keyword in the first half
+Avoid:
+
+ALL CAPS
+Excessive symbols
+Price or promotion language
+Hard-sell expressions
+ 
+
+[DESCRIPTION GENERATION RULES]
+
+Character rules:
+
+Maximum length: 1,000 characters
+Critical information must appear within the first 150 characters
+Language style rules (mandatory):
+
+Use polite Korean honorific style
+Replace “있나요?” with “있으신가요?”
+Do not start sentences with “이곳은”
+Replace “선택이 됩니다” with “추천 드립니다”
+Always wrap the business name in quotation marks
+Example: “스테이 머뭄”
+Avoid vague location words like “근대거리” alone
+Use specific phrasing such as:
+“군산 근대역사문화거리 일대”
+Structure:
+
+Opening CTA (first line)
+Must be a question or gentle suggestion
+Must use honorific tone
+Example:
+“조용히 쉴 수 있는 군산숙소를 찾고 있으신가요?”
+Core Stay Introduction (within first 150 characters total)
+Mention business name with quotation marks
+Mention region
+Include main keyword
+Briefly describe the stay experience
+Brand Experience
+Core value and emotional promise
+Based on marketing intelligence positioning
+Key Highlights (3–4 short lines)
+Derived from USP signals
+Natural sentences
+Focus on booking-trigger moments
+Local Context
+Mention nearby experiences
+Use specific local references
+Example:
+“군산 근대역사문화거리 일대 산책이나 로컬 카페 투어”
+Soft Closing Line
+One gentle, non-salesy closing sentence
+Must end with a recommendation tone
+Example:
+“군산에서 조용한 시간을 보내고 싶다면 ‘스테이 머뭄’을 추천 드립니다.”
+ 
+
+[SEO & AEO RULES]
+
+Naturally integrate 3–5 keywords from {target_keywords}
+Avoid keyword stuffing
+Use conversational, search-like phrasing
+Optimize for:
+YouTube search
+Google video results
+AI answer summaries
+Keywords should appear in:
+
+Title (1–2)
+First 150 characters of description
+Highlight or context sections
+ 
+
+[LANGUAGE RULE]
+
+All output must be written entirely in {language}.
+No mixed languages.
+
+ 
+
+[OUTPUT FORMAT – STRICT]
+
+title:
+description:
+
+No explanations.
+No headings.
+No extra text.

app/utils/subtitles.py

@@ -6,7 +6,7 @@ from typing import Literal, Any
 import httpx
 
 from app.utils.logger import get_logger
-from app.utils.prompts.chatgpt_prompt import ChatgptService
+from app.utils.chatgpt_prompt import ChatgptService
 from app.utils.prompts.schemas import *
 from app.utils.prompts.prompts import *
 

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

@@ -25,7 +25,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from app.database.session import get_session
 from app.user.dependencies.auth import get_current_user
 from app.user.models import User
-from app.home.models import Image, Project, MarketingIntel, ImageTag
+from app.home.models import Image, Project, MarketingIntel
 from app.lyric.models import Lyric
 from app.song.models import Song, SongTimestamp
 from app.utils.creatomate import CreatomateService
@@ -39,7 +39,6 @@ from app.video.schemas.video_schema import (
     VideoRenderData,
 )
 from app.video.worker.video_task import download_and_upload_video_to_blob
-from app.video.services.video import get_image_tags_by_task_id
 
 from config import creatomate_settings
 
@@ -338,30 +337,17 @@ async def generate_video(
         )
 
         # 6-1. 템플릿 조회 (비동기)
-        template = await creatomate_service.get_one_template_data(
+        template = await creatomate_service.get_one_template_data_async(
             creatomate_service.template_id
         )
         logger.debug(f"[generate_video] Template fetched - task_id: {task_id}")
 
         # 6-2. elements에서 리소스 매핑 생성
-        # modifications = creatomate_service.elements_connect_resource_blackbox(
+        modifications = creatomate_service.elements_connect_resource_blackbox(
-        #     elements=template["source"]["elements"],
+            elements=template["source"]["elements"],
-        #     image_url_list=image_urls,
+            image_url_list=image_urls,
-        #     music_url=music_url,
+            music_url=music_url,
-        #     address=store_address
+            address=store_address
-        taged_image_list = await get_image_tags_by_task_id(task_id)
-        min_image_num = creatomate_service.counting_component(
-            template = template,
-            target_template_type = "image"
-        )
-        duplicate = bool(len(taged_image_list) < min_image_num)
-        logger.info(f"[generate_video] Duplicate : {duplicate} | length of taged_image {len(taged_image_list)}, min_len {min_image_num},- task_id: {task_id}")
-        modifications = creatomate_service.template_matching_taged_image(
-            template = template,
-            taged_image_list = taged_image_list,
-            music_url = music_url, 
-            address = store_address,
-            duplicate = duplicate,
         )
         logger.debug(f"[generate_video] Modifications created - task_id: {task_id}")
 
@@ -427,7 +413,7 @@ async def generate_video(
         #     f"[generate_video] final_template: {json.dumps(final_template, indent=2, ensure_ascii=False)}"
         # )
         # 6-5. 커스텀 렌더링 요청 (비동기)
-        render_response = await creatomate_service.make_creatomate_custom_call(
+        render_response = await creatomate_service.make_creatomate_custom_call_async(
             final_template["source"],
         )
         
@@ -579,7 +565,7 @@ async def get_video_status(
     )
     try:
         creatomate_service = CreatomateService()
-        result = await creatomate_service.get_render_status(creatomate_render_id)
+        result = await creatomate_service.get_render_status_async(creatomate_render_id)
         logger.debug(
             f"[get_video_status] Creatomate API response - creatomate_render_id: {creatomate_render_id}, status: {result.get('status')}"
         )

config.py

@@ -42,7 +42,6 @@ class ProjectSettings(BaseSettings):
 
 class APIKeySettings(BaseSettings):
     CHATGPT_API_KEY: str = Field(default="your-chatgpt-api-key")  # 기본값 추가
-    GEMINI_API_KEY: str = Field(default="your-gemeni-api-key")  # 기본값 추가
     SUNO_API_KEY: str = Field(default="your-suno-api-key")  # Suno API 키
     SUNO_CALLBACK_URL: str = Field(
         default="https://example.com/api/suno/callback"
@@ -196,14 +195,6 @@ class RecoverySettings(BaseSettings):
     # ============================================================
     # ChatGPT API 설정
     # ============================================================
-    LLM_TIMEOUT: float = Field(
-        default=600.0,
-        description="LLM Default API 타임아웃 (초)",
-    )
-    LLM_MAX_RETRIES: int = Field(
-        default=1,
-        description="LLM API 응답 실패 시 최대 재시도 횟수",
-    )
     CHATGPT_TIMEOUT: float = Field(
         default=600.0,
         description="ChatGPT API 타임아웃 (초). OpenAI Python SDK 기본값: 600초 (10분)",
@@ -212,6 +203,7 @@ class RecoverySettings(BaseSettings):
         default=1,
         description="ChatGPT API 응답 실패 시 최대 재시도 횟수",
     )
+
     # ============================================================
     # Suno API 설정
     # ============================================================