castad
/
o2o-castad-backend
7
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'main' of https://gitea.o2o.kr/castad/o2o-castad-backend into feature-dashboard

feature-dashboard
김성경 2026-04-03 11:39:49 +09:00
parent ae9f0b3c62 162e5d699d
commit d955ac80f1
23 changed files with 1885 additions and 1349 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
app/database/session.py
View File

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

303
app/home/api/routers/v1/home.py
View File

@ -9,9 +9,10 @@ import aiofiles
from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile, status from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile, status
from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import func, select
from app.database.session import get_session, AsyncSessionLocal from app.database.session import get_session, AsyncSessionLocal
from app.home.models import Image, MarketingIntel from app.home.models import Image, MarketingIntel, ImageTag
from app.user.dependencies.auth import get_current_user from app.user.dependencies.auth import get_current_user
from app.user.models import User from app.user.models import User
from app.home.schemas.home_schema import ( from app.home.schemas.home_schema import (
@ -29,12 +30,13 @@ from app.home.schemas.home_schema import (
) )
from app.home.services.naver_search import naver_search_client from app.home.services.naver_search import naver_search_client
from app.utils.upload_blob_as_request import AzureBlobUploader from app.utils.upload_blob_as_request import AzureBlobUploader
from app.utils.chatgpt_prompt import ChatgptService, ChatGPTResponseError from app.utils.prompts.chatgpt_prompt import ChatgptService, ChatGPTResponseError
from app.utils.common import generate_task_id from app.utils.common import generate_task_id
from app.utils.logger import get_logger from app.utils.logger import get_logger
from app.utils.nvMapScraper import NvMapScraper, GraphQLException from app.utils.nvMapScraper import NvMapScraper, GraphQLException, URLNotFoundException
from app.utils.nvMapPwScraper import NvMapPwScraper from app.utils.nvMapPwScraper import NvMapPwScraper
from app.utils.prompts.prompts import marketing_prompt from app.utils.prompts.prompts import marketing_prompt
from app.utils.autotag import autotag_images
from config import MEDIA_ROOT from config import MEDIA_ROOT
# 로거 설정 # 로거 설정
@ -218,6 +220,15 @@ async def _crawling_logic(
status_code=status.HTTP_502_BAD_GATEWAY, status_code=status.HTTP_502_BAD_GATEWAY,
detail=f"네이버 지도 크롤링에 실패했습니다: {e}", 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: except Exception as e:
step1_elapsed = (time.perf_counter() - step1_start) * 1000 step1_elapsed = (time.perf_counter() - step1_start) * 1000
logger.error( logger.error(
@ -451,255 +462,6 @@ IMAGES_JSON_EXAMPLE = """[
{"url": "https://naverbooking-phinf.pstatic.net/20240514_259/17156880311809wCnY_JPEG/5.jpg", "name": "외관"} {"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( @router.post(
"/image/upload/blob", "/image/upload/blob",
summary="이미지 업로드 (Azure Blob Storage)", summary="이미지 업로드 (Azure Blob Storage)",
@ -988,6 +750,10 @@ async def upload_images_blob(
saved_count = len(result_images) saved_count = len(result_images)
image_urls = [img.img_url for img in 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 total_time = time.perf_counter() - request_start
logger.info( logger.info(
f"[upload_images_blob] SUCCESS - task_id: {task_id}, " f"[upload_images_blob] SUCCESS - task_id: {task_id}, "
@ -1003,3 +769,36 @@ async def upload_images_blob(
images=result_images, images=result_images,
image_urls=image_urls, 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()

54
app/home/models.py
View File

@ -9,7 +9,8 @@ Home 모듈 SQLAlchemy 모델 정의
from datetime import datetime from datetime import datetime
from typing import TYPE_CHECKING, List, Optional, Any from typing import TYPE_CHECKING, List, Optional, Any
from sqlalchemy import Boolean, DateTime, ForeignKey, Index, Integer, String, Text, JSON, func from sqlalchemy import Boolean, DateTime, ForeignKey, Computed, Index, Integer, String, Text, JSON, func
from sqlalchemy.dialects.mysql import INTEGER
from sqlalchemy.orm import Mapped, mapped_column, relationship from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database.session import Base from app.database.session import Base
@ -314,13 +315,50 @@ class MarketingIntel(Base):
) )
def __repr__(self) -> str: def __repr__(self) -> str:
task_id_str = ( return (
(self.task_id[:10] + "...") if len(self.task_id) > 10 else self.task_id f"<MarketingIntel(id={self.id}, place_id='{self.place_id}')>"
)
img_name_str = (
(self.img_name[:10] + "...") if len(self.img_name) > 10 else self.img_name
) )
return (
f"<Image(id={self.id}, task_id='{task_id_str}', img_name='{img_name_str}')>" class ImageTag(Base):
"""
이미지 태그 테이블
"""
__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",
) )

13
app/lyric/api/routers/v1/lyric.py
View File

@ -42,7 +42,7 @@ from app.lyric.schemas.lyric import (
LyricStatusResponse, LyricStatusResponse,
) )
from app.lyric.worker.lyric_task import generate_lyric_background, generate_subtitle_background from app.lyric.worker.lyric_task import generate_lyric_background, generate_subtitle_background
from app.utils.chatgpt_prompt import ChatgptService from app.utils.prompts.chatgpt_prompt import ChatgptService
from app.utils.logger import get_logger from app.utils.logger import get_logger
from app.utils.pagination import PaginatedResponse, get_paginated from app.utils.pagination import PaginatedResponse, get_paginated
@ -253,17 +253,6 @@ async def generate_lyric(
step1_start = time.perf_counter() step1_start = time.perf_counter()
logger.debug(f"[generate_lyric] Step 1: 서비스 초기화 및 프롬프트 생성...") 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 = { promotional_expressions = {
"Korean" : "인스타 감성, 사진같은 하루, 힐링, 여행, 감성 숙소", "Korean" : "인스타 감성, 사진같은 하루, 힐링, 여행, 감성 숙소",
"English" : "Instagram vibes, picture-perfect day, healing, travel, getaway", "English" : "Instagram vibes, picture-perfect day, healing, travel, getaway",

2
app/lyric/schemas/lyric.py
View File

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

11
app/lyric/worker/lyric_task.py
View File

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

2
app/social/services/seo_service.py
View File

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

4
app/song/services/song.py
View File

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

48
app/utils/autotag.py Normal file
View File

@ -0,0 +1,48 @@
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

95
app/utils/chatgpt_prompt.py
View File

@ -1,95 +0,0 @@
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

137
app/utils/creatomate.py
View File

@ -31,11 +31,13 @@ response = await creatomate.make_creatomate_call(template_id, modifications)
import copy import copy
import time import time
from enum import StrEnum
from typing import Literal from typing import Literal
import traceback
import httpx import httpx
from app.utils.logger import get_logger 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 from config import apikey_settings, creatomate_settings, recovery_settings
# 로거 설정 # 로거 설정
@ -226,8 +228,9 @@ DVST0003 = "e1fb5b00-1f02-4f63-99fa-7524b433ba47"
DHST0001 = "660be601-080a-43ea-bf0f-adcf4596fa98" DHST0001 = "660be601-080a-43ea-bf0f-adcf4596fa98"
DHST0002 = "3f194cc7-464e-4581-9db2-179d42d3e40f" DHST0002 = "3f194cc7-464e-4581-9db2-179d42d3e40f"
DHST0003 = "f45df555-2956-4a13-9004-ead047070b3d" DHST0003 = "f45df555-2956-4a13-9004-ead047070b3d"
DVST0001T = "fe11aeab-ff29-4bc8-9f75-c695c7e243e6"
HST_LIST = [DHST0001,DHST0002,DHST0003] HST_LIST = [DHST0001,DHST0002,DHST0003]
VST_LIST = [DVST0001,DVST0002,DVST0003] VST_LIST = [DVST0001,DVST0002,DVST0003, DVST0001T]
SCENE_TRACK = 1 SCENE_TRACK = 1
AUDIO_TRACK = 2 AUDIO_TRACK = 2
@ -238,7 +241,7 @@ def select_template(orientation:OrientationType):
if orientation == "horizontal": if orientation == "horizontal":
return DHST0001 return DHST0001
elif orientation == "vertical": elif orientation == "vertical":
return DVST0001 return DVST0001T
else: else:
raise raise
@ -399,14 +402,6 @@ class CreatomateService:
return copy.deepcopy(data) 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: def parse_template_component_name(self, template_source: list) -> dict:
"""템플릿 정보를 파싱하여 리소스 이름을 추출합니다.""" """템플릿 정보를 파싱하여 리소스 이름을 추출합니다."""
@ -434,42 +429,93 @@ class CreatomateService:
return result return result
async def template_connect_resource_blackbox( async def parse_template_name_tag(resource_name : str) -> list:
self, tag_list = []
template_id: str, tag_list = resource_name.split("_")
image_url_list: list[str],
music_url: str, return tag_list
address: str = None
) -> dict:
"""템플릿 정보와 이미지/가사/음악 리소스를 매핑합니다. def template_matching_taged_image(
self,
template : dict,
taged_image_list : list, # [{"image_name" : str , "image_tag" : dict}]
music_url: str,
address : str,
duplicate : bool = False
) -> 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 = {} modifications = {}
for idx, (template_component_name, template_type) in enumerate( for slot_idx, (template_component_name, template_type) in enumerate(template_component_data.items()):
template_component_data.items()
):
match template_type: match template_type:
case "image": case "image":
modifications[template_component_name] = image_url_list[ image_score_list = self.calculate_image_slot_score_multi(taged_image_list, template_component_name)
idx % len(image_url_list) maximum_idx = image_score_list.index(max(image_score_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": case "text":
if "address_input" in template_component_name: if "address_input" in template_component_name:
modifications[template_component_name] = address modifications[template_component_name] = address
modifications["audio-music"] = music_url modifications["audio-music"] = music_url
return modifications 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( def elements_connect_resource_blackbox(
self, self,
elements: list, elements: list,
@ -669,14 +715,6 @@ class CreatomateService:
original_response={"last_error": str(last_error)}, 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: async def get_render_status(self, render_id: str) -> dict:
"""렌더링 작업의 상태를 조회합니다. """렌더링 작업의 상태를 조회합니다.
@ -700,14 +738,6 @@ class CreatomateService:
response.raise_for_status() response.raise_for_status()
return response.json() 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: def calc_scene_duration(self, template: dict) -> float:
"""템플릿의 전체 장면 duration을 계산합니다.""" """템플릿의 전체 장면 duration을 계산합니다."""
total_template_duration = 0.0 total_template_duration = 0.0
@ -720,19 +750,20 @@ class CreatomateService:
try: try:
if elem["track"] not in track_maximum_duration: if elem["track"] not in track_maximum_duration:
continue continue
if elem["time"] == 0: # elem is auto / 만약 마지막 elem이 auto인데 그 앞에 time이 있는 elem 일 시 버그 발생 확률 있음 if "time" not in elem or elem["time"] == 0: # elem is auto / 만약 마지막 elem이 auto인데 그 앞에 time이 있는 elem 일 시 버그 발생 확률 있음
track_maximum_duration[elem["track"]] += elem["duration"] track_maximum_duration[elem["track"]] += elem["duration"]
if "animations" not in elem: if "animations" not in elem:
continue continue
for animation in elem["animations"]: for animation in elem["animations"]:
assert animation["time"] == 0 # 0이 아닌 경우 확인 필요 assert animation["time"] == 0 # 0이 아닌 경우 확인 필요
if animation["transition"]: if "transition" in animation and animation["transition"]:
track_maximum_duration[elem["track"]] -= animation["duration"] track_maximum_duration[elem["track"]] -= animation["duration"]
else: else:
track_maximum_duration[elem["track"]] = max(track_maximum_duration[elem["track"]], elem["time"] + elem["duration"]) track_maximum_duration[elem["track"]] = max(track_maximum_duration[elem["track"]], elem["time"] + elem["duration"])
except Exception as e: except Exception as e:
logger.debug(traceback.format_exc())
logger.error(f"[calc_scene_duration] Error processing element: {elem}, {e}") logger.error(f"[calc_scene_duration] Error processing element: {elem}, {e}")
total_template_duration = max(track_maximum_duration.values()) total_template_duration = max(track_maximum_duration.values())

14
app/utils/nvMapScraper.py
View File

@ -16,6 +16,10 @@ class GraphQLException(Exception):
"""GraphQL 요청 실패 시 발생하는 예외""" """GraphQL 요청 실패 시 발생하는 예외"""
pass pass
class URLNotFoundException(Exception):
"""Place ID 발견 불가능 시 발생하는 예외"""
pass
class CrawlingTimeoutException(Exception): class CrawlingTimeoutException(Exception):
"""크롤링 타임아웃 시 발생하는 예외""" """크롤링 타임아웃 시 발생하는 예외"""
@ -86,15 +90,14 @@ query getAccommodation($id: String!, $deviceType: String) {
async with session.get(self.url) as response: async with session.get(self.url) as response:
self.url = str(response.url) self.url = str(response.url)
else: else:
raise GraphQLException("This URL does not contain a place ID") raise URLNotFoundException("This URL does not contain a place ID")
match = re.search(place_pattern, self.url) match = re.search(place_pattern, self.url)
if not match: if not match:
raise GraphQLException("Failed to parse place ID from URL") raise URLNotFoundException("Failed to parse place ID from URL")
return match[1] return match[1]
async def scrap(self): async def scrap(self):
try:
place_id = await self.parse_url() place_id = await self.parse_url()
data = await self._call_get_accommodation(place_id) data = await self._call_get_accommodation(place_id)
self.rawdata = data self.rawdata = data
@ -110,11 +113,6 @@ query getAccommodation($id: String!, $deviceType: String) {
self.facility_info = fac_data self.facility_info = fac_data
self.scrap_type = "GraphQL" self.scrap_type = "GraphQL"
except GraphQLException:
logger.debug("GraphQL failed, fallback to Playwright")
self.scrap_type = "Playwright"
pass # 나중에 pw 이용한 crawling으로 fallback 추가
return return
async def _call_get_accommodation(self, place_id: str) -> dict: async def _call_get_accommodation(self, place_id: str) -> dict:

191
app/utils/prompts/chatgpt_prompt.py Normal file
View File

@ -0,0 +1,191 @@
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

21
app/utils/prompts/prompts.py
View File

@ -40,9 +40,11 @@ class Prompt():
def _reload_prompt(self): def _reload_prompt(self):
self.prompt_template, self.prompt_model = self._read_from_sheets() self.prompt_template, self.prompt_model = self._read_from_sheets()
def build_prompt(self, input_data: dict) -> str: def build_prompt(self, input_data:dict, silent:bool = False) -> str:
verified_input = self.prompt_input_class(**input_data) verified_input = self.prompt_input_class(**input_data)
build_template = self.prompt_template.format(**verified_input.model_dump()) build_template = self.prompt_template
build_template = build_template.format(**verified_input.model_dump())
if not silent:
logger.debug(f"build_template: {build_template}") logger.debug(f"build_template: {build_template}")
logger.debug(f"input_data: {input_data}") logger.debug(f"input_data: {input_data}")
return build_template return build_template
@ -65,6 +67,20 @@ yt_upload_prompt = Prompt(
prompt_output_class=YTUploadPromptOutput, prompt_output_class=YTUploadPromptOutput,
) )
image_autotag_prompt = Prompt(
prompt_template_path=os.path.join(prompt_settings.PROMPT_FOLDER_ROOT, prompt_settings.IMAGE_TAG_PROMPT_FILE_NAME),
prompt_input_class = ImageTagPromptInput,
prompt_output_class = ImageTagPromptOutput,
prompt_model = prompt_settings.IMAGE_TAG_PROMPT_MODEL
)
image_autotag_prompt = Prompt(
prompt_template_path=os.path.join(prompt_settings.PROMPT_FOLDER_ROOT, prompt_settings.IMAGE_TAG_PROMPT_FILE_NAME),
prompt_input_class = ImageTagPromptInput,
prompt_output_class = ImageTagPromptOutput,
prompt_model = prompt_settings.IMAGE_TAG_PROMPT_MODEL
)
@lru_cache() @lru_cache()
def create_dynamic_subtitle_prompt(length: int) -> Prompt: def create_dynamic_subtitle_prompt(length: int) -> Prompt:
return Prompt( return Prompt(
@ -78,3 +94,4 @@ def reload_all_prompt():
marketing_prompt._reload_prompt() marketing_prompt._reload_prompt()
lyric_prompt._reload_prompt() lyric_prompt._reload_prompt()
yt_upload_prompt._reload_prompt() yt_upload_prompt._reload_prompt()
image_autotag_prompt._reload_prompt()

1
app/utils/prompts/schemas/__init__.py
View File

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

110
app/utils/prompts/schemas/image.py Normal file
View File

@ -0,0 +1,110 @@
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="이미지의 내러티브 상 점수")

4
app/utils/prompts/schemas/marketing.py
View File

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

24
app/utils/prompts/templates/image_autotag_prompt.txt Normal file
View File

@ -0,0 +1,24 @@
You are an expert image analyst and content tagger.
Analyze the provided image carefully and select the most appropriate tags for each category below.
and lastly, check this image is acceptable for marketing advertisement video.
and set NarrativePreference score for each narrative phase.
## Rules
- For each category, select between 0 and the specified maximum number of tags.
- Only select tags from the exact list provided for each category. Do NOT invent or modify tags.
- If no tags in a category are applicable, return an empty list for that category.
- Base your selections solely on what is visually present or strongly implied in the image.
-
## Tag Categories
space_type:{space_type}
subject: {subject}
camera:{camera}
motion_recommended: {motion_recommended}
## Output
Return a JSON object where each key is a category name and the value is a list of selected tags.
Selected tags must be chosen from the available tags of that category only.
and NarrativePreference score for each narrative phase.

461
app/utils/prompts/templates/subtitle_prompt.txt
View File

@ -1,76 +1,431 @@
당신은 숙박 브랜드 숏폼 영상의 자막 콘텐츠를 추출하는 전문가입니다. ---
name: Creatomate-subtitle-naming_v01
description: "Generate and name subtitle layers in Creatomate video templates using a structured 5-criteria tag convention: track_role - narrative_phase - content_type - tone - pair_id. Use this skill whenever the user mentions subtitle naming, caption tagging, or wants to create/rename subtitle or keyword text layers in a Creatomate template based on marketing intelligence data. Also trigger when the user provides marketing analysis data and asks to generate subtitle content for a hospitality/stay brand video. This skill covers two text tracks: subtitle (scene description) and keyword (core emotional keyword). Within a single tag value, multi-word terms use underscores; between tag criteria, hyphens are used."
---
입력으로 주어지는 **1) 5가지 기준의 레이어 이름 리스트**와 **2) 마케팅 인텔리전스 분석 결과(JSON)**를 바탕으로, 각 레이어 이름의 의미에 정확히 1:1 매칭되는 텍스트 콘텐츠만을 추출하세요. # Creatomate Subtitle Layer Naming & Copywriting — 5-Criteria Tag Convention
분석 결과에 없는 정보는 절대 지어내거나 추론하지 마세요. 오직 제공된 JSON 데이터 내에서만 텍스트를 구성해야 합니다. You are a **subtitle copywriter and video structure director** for hospitality brand short-form videos. You name subtitle layers using a structured tagging system AND generate compelling subtitle text by transforming marketing intelligence data into viewer-engaging copy.
This skill is a companion to the image layer naming skill. While image layers describe *what you see*, subtitle layers describe *what you read* — and each subtitle maps 1:1 to an image scene.
## Core Principles
1. **Transform (Rewrite)**: NEVER copy JSON data verbatim. ALWAYS rewrite into video-optimized copy.
2. **Fact-based**: NEVER invent information not present in the analysis. BUT freely transform HOW data is expressed.
3. **Emotion-designed**: Each scene MUST evoke a specific emotion in the viewer.
--- ---
## 1. 레이어 네이밍 규칙 해석 및 매핑 가이드 ## PHASE 1. The Naming Format
입력되는 모든 레이어 이름은 예외 없이 `<track_role>-<narrative_phase>-<content_type>-<tone>-<pair_id>` 의 5단계 구조로 되어 있습니다. ```
마지막의 3자리 숫자 ID(`-001`, `-002` 등)는 모든 레이어에 필수적으로 부여됩니다. (track_role)-(narrative_phase)-(content_type)-(tone)-(pair_id)
```
### [1] track_role (텍스트 형태) **Separator rules:**
- `subtitle`: 씬 상황을 설명하는 간결한 문장형 텍스트 (1줄 이내) - Between criteria (the 5 slots): **hyphen `-`**
- `keyword`: 씬을 상징하고 시선을 끄는 단답형/명사형 텍스트 (1~2단어) - Within a multi-word tag value: **underscore `_`**
- pair_id format: **3-digit zero-padded number** (`001`, `002`, ... `999`)
### [2] narrative_phase (영상 흐름) Example: `subtitle-intro-hook_claim-aspirational-001`
- `intro`: 영상 도입부. 가장 시선을 끄는 정보를 배치.
- `core`: 핵심 매력이나 주요 편의 시설 어필.
- `highlight`: 세부적인 매력 포인트나 공간의 특별한 분위기 묘사.
- `outro`: 영상 마무리. 브랜드 명칭 복기 및 타겟/위치 정보 제공.
### [3] content_type (데이터 매핑 대상) > A `subtitle` and `keyword` layer sharing the same scene MUST have the **same pair_id**. This is a developer requirement for programmatic pairing.
- `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 내의 다른 소구점이나 데이터를 추출**하여 내용이 중복되지 않도록 해야 합니다.
--- ---
## 2. 콘텐츠 추출 시 주의사항 ## PHASE 2. Tag Vocabulary
1. 각 입력 레이어 이름 1개당 **오직 1개의 텍스트 콘텐츠**만 매핑하여 출력합니다. (레이어명 이름 자체를 수정하거나 새로 만들지 마세요.) ### [1] track_role — Which text track?
2. `content_type`이 `selling_point`로 동일하더라도, `narrative_phase`(core, highlight)나 `tone`이 달라지면 JSON 내의 2순위, 3순위 세일즈 포인트를 순차적으로 활용하여 내용 겹침을 방지하세요.
3. 같은 씬에 속하는(같은 ID 번호를 가진) keyword는 핵심 단어로, subtitle은 적절한 마케팅 문구가 되어야 하며, 자연스럽게 이어지는 문맥을 형성하도록 구성하세요. | Value | Meaning | Visual Treatment |
4. keyword가 subtitle에 완전히 포함되는 단어가 되지 않도록 유의하세요. |---|---|---|
5. 정보 태그가 같더라도 ID가 다르다면 중복되지 않는 새로운 텍스트를 도출해야 합니다. | `subtitle` | Scene description / sub-headline | Smaller text (Track 3) |
6. 콘텐츠 추출 시 마케팅 인텔리전스의 내용을 그대로 사용하기보다는 paraphrase을 수행하세요. | `keyword` | Core emotional keyword | Larger, bolder text (Track 4) |
7. keyword는 공백 포함 전각 8자 / 반각 16자내, subtitle은 전각 15자 / 반각 30자 내로 구성하세요.
Every scene gets **both** a `subtitle` and a `keyword` layer. They form a **Pair** sharing the same `pair_id`.
### [2] narrative_phase — Where in the story?
Uses the **same vocabulary as image layers** for 1:1 mapping.
| Value | Meaning | Position | Emotion Goal |
|---|---|---|---|
| `intro` | Brand first impression, hook | Opening scene | Curiosity — stop the scroll |
| `welcome` | Location / check-in introduction | 2nd~3rd scene | Warmth — "I want to go there" |
| `core` | Key space features, repeated delivery | Mid-section | Trust — "This place is legit" |
| `highlight` | Signature space/emotion emphasis | Mid-section | Desire — "I need this" |
| `support` | Surrounding environment, local curation | Later section | Discovery — "There's even more" |
| `accent` | Emotional wrap-up, target, pre-CTA | Near the end | Belonging — "This is for me" |
| `cta` | Call To Action | Final scene | Action — "Book now" |
### [3] content_type — What kind of text?
| Value | Meaning | Source Field | Example |
|---|---|---|---|
| `brand_name` | Business name | `store_name` | 스테이펫 홍천 |
| `brand_address` | Business address | `detail_region_info` | 강원 홍천군 화촌면 담연발길 5-2 |
| `hook_claim` | 1-line hook copy | `selling_points[0]` or `core_value` | 댕댕이가 먼저 뛰어간 숲 |
| `space_feature` | Space characteristic | `selling_points[].description` | 프라이빗 독채에서 자연 그대로 |
| `emotion_cue` | Emotional trigger phrase | `selling_points[].description` (sensory rewrite) | 숲 향기 가득한 테라스 |
| `lifestyle_fit` | Lifestyle empathy | `target_persona[].favor_target` | 주말마다 어디 갈지 고민하는 견주님 |
| `local_info` | Nearby local information | `location_feature_analysis` | 서울에서 1시간 반, 홍천 숲속 |
| `target_tag` | Target audience hashtags | `target_keywords[]` | #펫프렌들리 #강원여행 |
| `availability` | Booking status | (fixed text) | 지금 예약 가능 |
| `cta_action` | CTA button text | (fixed text) | 예약하러 가기 |
### [4] tone — What emotional register?
| Value | Characteristics | Recommended | Forbidden | Example (same content) |
|---|---|---|---|---|
| `sensory` | Poetic, sense-evoking | 향기, 소리, 촉감, 온도 | 추상적 형용사 (좋은, 멋진) | 이끼 향 가득한 숲속 테라스 |
| `factual` | Informational, neutral | 숫자, 거리, 시설명 | 감탄사, 과장 수식어 | 객실 내 전용 마당 30평 |
| `empathic` | Empathetic, warm | ~하는 분, ~하고 싶은 | 명령형, 단정형 | 반려견과 마음 편히 쉬고 싶을 때 |
| `aspirational` | Desire-triggering | ~같은, ~처럼, 꿈꾸던 | 부정형, 비교급 | 내가 꿈꾸던 반려견과의 여행 |
| `social_proof` | Credibility, target-based | 타겟 명시, 해시태그 | 과장된 추천 | 2030 커플·견주 추천 |
| `urgent` | Action-prompting | 지금, 바로, 확인 | 위협적 표현 | 지금 예약 가능 |
**Hospitality brand forbidden words (ALL tones):** 저렴한, 싼, 그냥, 보통, 무난한, 평범한
### [5] pair_id — Scene pair identifier
- Format: 3-digit zero-padded number (`001` ~ `999`)
- A `subtitle` and its matching `keyword` in the same scene share the **identical pair_id**
- Assigned sequentially from `001` in video playback order
- Example: Scene 1 → `001`, Scene 2 → `002`, ...
--- ---
## 3. 출력 결과 포맷 및 예시 ## PHASE 3. Anchor Position Rules (Fixed Slots)
입력된 레이어 이름 순서에 맞춰, 매핑된 텍스트 콘텐츠만 작성하세요. (반드시 intro, core, highlight, outro 등 모든 씬 단계가 명확하게 매핑되어야 합니다.) Regardless of total scene count, **the first and last 3 positions are always fixed content**. Only middle scenes are flexible based on marketing data.
### 입력 레이어 리스트 예시 및 출력 예시 ```
[First] ──── Middle Scenes (flexible) ──── [Last-3] [Last-2] [Last]
↓ ↓ ↓ ↓
intro address hashtag CTA
```
| Layer Name | Text Content | ### Anchor 1: First Scene (intro)
| Track | content_type | Content Rule |
|---|---|---|
| subtitle | `hook_claim` | Transform `selling_points[0]` or `core_value` into a scroll-stopping hook |
| keyword | `brand_name` | `store_name` — brand recognition |
Layer names:
- `subtitle-intro-hook_claim-aspirational-001`
- `keyword-intro-brand_name-sensory-001`
### Anchor 2: Last Scene (cta)
| Track | content_type | Content Rule |
|---|---|---|
| subtitle | `availability` | Booking status (fixed: 지금 예약 가능) |
| keyword | `cta_action` | CTA button text (fixed: 예약하러 가기) |
### Anchor 3: Second to Last (accent) — Hashtags & Target
| Track | content_type | Content Rule |
|---|---|---|
| subtitle | `target_tag` | Extract from `target_keywords[]` as hashtags |
| keyword | `lifestyle_fit` | Transform `target_persona[].favor_target` into aspirational keyword |
### Anchor 4: Third to Last (support) — Address & Brand
| Track | content_type | Content Rule |
|---|---|---|
| subtitle | `brand_address` | Full address from `detail_region_info` |
| keyword | `brand_name` | `store_name` — brand reinforcement |
---
## PHASE 4. Middle Scenes (Flexible Slots)
Fill in order based on available marketing data. Use `selling_points[]` sorted by `score` descending.
| narrative_phase | subtitle content_type | keyword content_type | Data Source |
|---|---|---|---|
| `welcome` | `emotion_cue` | `space_feature` | `selling_points[0]` (highest score) |
| `core` | `space_feature` | `emotion_cue` | `selling_points[1~3]` (next items) |
| `highlight` | `space_feature` | `emotion_cue` | Signature/unique feature from data |
| `support` (mid) | `local_info` | `lifestyle_fit` | `location_feature_analysis` |
> If fewer scenes are available, reduce flexible slots. Anchor positions are NEVER removed.
---
## PHASE 5. Brand Expression Dictionary (표현 변환 사전)
Before writing ANY subtitle text, scan the source data against this dictionary. If a listed expression appears in the JSON data, it MUST be replaced with one of the approved alternatives. NEVER use the original expression verbatim.
### 5-0a. Expression Refinement Rules
**WHY this matters:** Marketing analysis data is written in analytical language, not consumer-facing language. Some expressions carry unintended negative nuance, sound unnatural in video subtitles, or feel like jargon. This dictionary ensures every subtitle reads as polished, brand-safe copy.
**HOW to apply:**
1. Before writing each subtitle, check if ANY word/phrase from the source data matches the "Raw Expression" column
2. Replace with the most contextually appropriate "Approved Alternative"
3. If multiple alternatives exist, choose based on the scene's `tone` tag
### 5-0b. Mandatory Expression Replacements
| Raw Expression (JSON 원본) | Problem | Approved Alternatives | Tone Guidance |
|---|---|---|---|
| 눈치 없는 | "센스 없는/무례한"으로 오해 가능 | **눈치 안 보는** · **프라이빗한** · **온전한** · **자유로운** | sensory→"온전한", empathic→"눈치 안 보는", aspirational→"프라이빗한" |
| 눈치 없이 | 동일 문제 (부사형) | **눈치 안 보고** · **마음 편히** · **자유롭게** | sensory→"마음 편히", empathic→"눈치 안 보고", aspirational→"자유롭게" |
| 감성 쩌는 / 쩌이 | 과도한 속어, 브랜드 품격 저하 | **감성 가득한** · **감성이 머무는** · **분위기 있는** | sensory→"감성이 머무는", aspirational→"감성 가득한" |
| 가성비 | 저가 이미지 연상 | **합리적인** · **가치 있는** | factual→"합리적인", aspirational→"가치 있는" |
| 인스타감성 / 인스타 | 플랫폼 종속 표현 | **감성 스팟** · **포토 스팟** · **기록하고 싶은** | sensory→"기록하고 싶은", social_proof→"감성 스팟" |
| ~맛집 | 숙박 브랜드에 부적합 | **~명소** · **~스팟** | factual→"명소", sensory→"스팟" |
| 힐링되는 | 과잉 사용으로 진부 | **회복되는** · **쉬어가는** · **숨 쉬는** | sensory→"숨 쉬는", empathic→"쉬어가는", factual→"회복되는" |
| 혜자 | 속어, 브랜드 부적합 | **풍성한** · **넉넉한** | factual→"넉넉한", aspirational→"풍성한" |
### 5-0c. Contextual Synonym Expansion
When the same concept appears in multiple scenes, use **synonyms** to avoid repetition. Each concept has a synonym pool — cycle through them across scenes.
| Concept | Synonym Pool (rotate across scenes) |
|---|---| |---|---|
| subtitle-intro-hook_claim-aspirational-001 | 반려견과 눈치 없이 온전하게 쉬는 완벽한 휴식 | | 프라이빗/독립 | 온전한 · 프라이빗한 · 오롯한 · 나만의 · 독채 · 단독 |
| keyword-intro-brand_name-sensory-001 | 스테이펫 홍천 | | 자연/숲 | 숲속 · 자연 속 · 초록 · 산림 · 계곡 · 숲 |
| subtitle-core-selling_point-empathic-002 | 우리만의 독립된 공간감이 주는 진정한 쉼 | | 쉼/휴식 | 쉼 · 쉬어감 · 여유 · 느린 하루 · 머무름 · 숨 고르기 |
| keyword-core-selling_point-factual-002 | 프라이빗 독채 | | 반려견 | 댕댕이 · 반려견 · 우리 강아지 · 반려동물 · 우리 아이 |
| subtitle-highlight-selling_point-sensory-003 | 탁 트인 야외 무드존과 포토 스팟의 감성 컷 |
| keyword-highlight-selling_point-factual-003 | 넓은 정원 |
| subtitle-outro-target_tag-empathic-004 | #강원도애견동반 #주말숏브레이크 |
| keyword-outro-location_info-factual-004 | 강원 홍천군 화촌면 |
> **"댕댕이" 사용 가이드**: 28~49세 타깃 숏폼 자막에서 사용 적합. 단, 영상 전체에서 **최대 1회**만 사용하고 나머지는 "반려견"/"우리 강아지" 등으로 로테이션. intro(hook_claim)이나 accent(lifestyle_fit)처럼 **감성 후킹이 필요한 씬에서 사용**하고, factual/urgent 톤의 씬에서는 "반려견"을 사용할 것. 5성급 럭셔리 포지셔닝 브랜드라면 "반려견"으로 대체.
| 감성/분위기 | 감성 · 무드 · 온기 · 따스함 · 분위기 |
| 예약/행동 | 예약하기 · 지금 바로 · 확인하기 · 만나러 가기 |
> **RULE: The same Korean word MUST NOT appear in more than 2 scenes across the entire video.** Use the synonym pool to rotate expressions.
### 5-0d. Forbidden Expressions (Global)
These words MUST NEVER appear in any subtitle, regardless of tone:
| Category | Forbidden Words |
|---|---|
| 저가 연상 | 저렴한, 싼, 싸게, 할인, 가성비, 혜자 |
| 무성의 | 그냥, 보통, 무난한, 평범한, 괜찮은 |
| 과잉 속어 | 쩌는, 쩔어, 개(접두사), 존맛, 핵 |
| 부정 뉘앙스 | 눈치 없는, 눈치 없이, 질리지 않는 |
| 플랫폼 종속 | 인스타, 유튜브, 틱톡 (브랜드명 직접 언급) |
---
## PHASE 6. Copywriting Transformation Rules
### 6-1. Text Specifications
| track_role | Character Limit | Style | Example |
|---|---|---|---|
| `subtitle` | **8~18 chars** (incl. spaces) | Sentence fragment, conversational | 숲 향기 가득한 프라이빗 공간 |
| `keyword` | **2~6 chars** | Noun phrase, hashtag-like | 자연독채 |
### 6-2. Transformation Rules by content_type
#### `hook_claim` — The scroll-stopper (intro only)
- **Source**: `selling_points[0].description` or `market_positioning.core_value`
- **Transform rules**:
- Choose ONE format: question ("여기 진짜 있어?"), exclamation ("이런 곳이 있었다니"), provocation ("아직도 호텔만 가세요?")
- Use specific numbers if available (ratings, reviews, distance)
- **FORBIDDEN**: Brand name in hook, generic greetings
- **Transform example**:
- Source: `"반려견과 눈치 없는 힐링"` + `core_value: "자연 속 프라이빗 애견동반 힐링 스테이"`
- BAD: "반려견과 눈치 없는 힐링" (verbatim copy)
- BAD: "애견 동반 가능한 숙소" (generic extraction)
- GOOD: "댕댕이가 먼저 뛰어간 숲" (sensory rewrite, avoids "눈치 없이" per Expression Dictionary)
#### `space_feature` — Core appeal (core/highlight)
- **Source**: `selling_points[]` by score descending
- **Transform rules**:
- ONE selling point per scene (NEVER combine 2+)
- Do NOT use `korean_category` directly — transform `description` into sensory copy
- Write so the viewer can **imagine themselves there**
- **Transform example**:
- Source: `("description": "홍천 자연 속 조용한 쉼", "korean_category": "입지 환경")`
- BAD subtitle: "입지 환경이 좋은 곳" (used category name)
- GOOD subtitle: "계곡 소리만 들리는 독채"
- GOOD keyword: "자연독채"
#### `emotion_cue` — Feeling trigger (welcome/core/highlight)
- **Source**: Same `selling_points[]` item as its paired `space_feature`, but rewritten for emotion
- **Transform rules**:
- Appeal to senses: smell, sound, touch, temperature, light
- Use poetic fragments, not full sentences
- **Transform example**:
- Source: `("description": "감성 쩌이 완성되는 공간", "korean_category": "포토 스팟")`
- GOOD subtitle: "햇살이 내려앉는 테라스"
- GOOD keyword: "감성 가득"
#### `lifestyle_fit` — "This is for me" (accent/support)
- **Source**: `target_persona[].favor_target` or `decision_trigger`
- **Transform rules**:
- Write as if addressing the target directly
- Use their language, not marketing language
- **Transform example**:
- Source: `favor_target: "조용한 자연 뷰", persona: "서울·경기 주말러"`
- GOOD subtitle: "이번 주말, 댕댕이랑 어디 가지?"
- GOOD keyword: "주말탈출"
#### `local_info` — Location appeal (support)
- **Source**: `detail_region_info`, `location_feature_analysis`
- **Transform rules**:
- Express as **accessibility or regional charm**, NOT administrative address
- GOOD: "서울에서 1시간 반, 홍천 숲속" / BAD: "강원 홍천군 화촌면"
- keyword: Region name or travel keyword ("홍천", "#강원여행")
#### `brand_name` — Brand presence (intro keyword, support keyword)
- **Source**: `store_name`
- Present as-is. This is the ONE content_type where verbatim extraction is correct.
#### `brand_address` — Full address (support subtitle)
- **Source**: `detail_region_info`
- Present as-is. Factual, no transformation needed.
#### `target_tag` — Hashtags (accent subtitle)
- **Source**: `target_keywords[]`
- Format as SNS hashtags: `#홍천애견동반숙소 #스테이펫`
- Select 3~5 most relevant keywords
#### `availability` + `cta_action` — CTA (last scene)
- Fixed text. No transformation from data.
- subtitle: "지금 예약 가능" / keyword: "예약하러 가기"
### 6-3. Pacing Rules
Maintain **rhythm** between scenes by alternating subtitle character length:
```
intro → Short & punchy (8~12 chars) : curiosity burst
welcome → Medium (12~18 chars) : warm introduction
core → Alternate: short(8~12) ↔ medium(12~18)
highlight → Short & sensory (8~14 chars) : lingering impact
support → Medium (12~18 chars) : information delivery
accent → Short hashtags (variable)
cta → Medium (12~16 chars) : clear action
```
> **RULE: NEVER have 3+ consecutive scenes with the same character count range** — prevents monotony.
---
## PHASE 7. Emotional Arc
```
Emotion Intensity
▲
│ ★ highlight
│ ╱ ╲
│ ╱───╱ ╲───╲
│ ╱ core support╲
│ ╱welcome accent╲
│ ╱ intro cta ╲
└────────────────────────────── ► Time
Curiosity → Trust → Desire → Belonging → Action
```
Each `narrative_phase` maps to a specific emotional goal. The subtitle text MUST serve that emotion:
| Phase | Emotion | Subtitle's Job |
|---|---|---|
| `intro` | Curiosity | "What is this?" — stop the scroll |
| `welcome` | Warmth | "I want to see more" — gentle pull |
| `core` | Trust | "This place is real" — concrete appeal |
| `highlight` | Desire | "I need this" — peak sensory moment |
| `support` | Discovery | "There's even more" — added value |
| `accent` | Belonging | "This is for me" — target identification |
| `cta` | Action | "Book now" — clear next step |
---
## PHASE 8. Scene Assembly Examples
### Example A: 7-Scene Video (Standard)
```
Scene 1 [ANCHOR-First] intro-001 → subtitle: hook_claim / keyword: brand_name
Scene 2 [Flexible] welcome-002 → subtitle: emotion_cue / keyword: space_feature
Scene 3 [Flexible] core-003 → subtitle: space_feature / keyword: emotion_cue
Scene 4 [Flexible] highlight-004 → subtitle: space_feature / keyword: emotion_cue
Scene 5 [ANCHOR-Last-3] support-005 → subtitle: brand_address / keyword: brand_name
Scene 6 [ANCHOR-Last-2] accent-006 → subtitle: target_tag / keyword: lifestyle_fit
Scene 7 [ANCHOR-Last] cta-007 → subtitle: availability / keyword: cta_action
```
### Example B: 5-Scene Video (Compact)
```
Scene 1 [ANCHOR-First] intro-001 → subtitle: hook_claim / keyword: brand_name
Scene 2 [Flexible] core-002 → subtitle: space_feature / keyword: emotion_cue
Scene 3 [ANCHOR-Last-3] support-003 → subtitle: brand_address / keyword: brand_name
Scene 4 [ANCHOR-Last-2] accent-004 → subtitle: target_tag / keyword: lifestyle_fit
Scene 5 [ANCHOR-Last] cta-005 → subtitle: availability / keyword: cta_action
```
### Example C: 10-Scene Video (Extended)
```
Scene 1 [ANCHOR-First] intro-001 → subtitle: hook_claim / keyword: brand_name
Scene 2 [Flexible] welcome-002 → subtitle: emotion_cue / keyword: space_feature
Scene 3 [Flexible] core-003 → subtitle: space_feature / keyword: emotion_cue
Scene 4 [Flexible] core-004 → subtitle: space_feature / keyword: emotion_cue
Scene 5 [Flexible] highlight-005 → subtitle: space_feature / keyword: emotion_cue
Scene 6 [Flexible] highlight-006 → subtitle: space_feature / keyword: emotion_cue
Scene 7 [Flexible] support-007 → subtitle: local_info / keyword: lifestyle_fit
Scene 8 [ANCHOR-Last-3] support-008 → subtitle: brand_address / keyword: brand_name
Scene 9 [ANCHOR-Last-2] accent-009 → subtitle: target_tag / keyword: lifestyle_fit
Scene 10 [ANCHOR-Last] cta-010 → subtitle: availability / keyword: cta_action
```
> Fewer scenes → fewer flexible slots. Anchor positions are NEVER removed.
---
## PHASE 9. How to Generate (Step-by-Step)
### Step 1 — Parse marketing intelligence data
Scan for these key fields:
- `store_name` → brand_name, brand_address source
- `detail_region_info` → address, location appeal
- `selling_points[]` → sort by `score` descending; primary content source
- `market_positioning.core_value` → hook_claim alternative
- `target_persona[]` → lifestyle_fit, target_tag source
- `target_keywords[]` → hashtag source
- `location_feature_analysis` → local_info source
### Step 2 — Determine scene count and assign pair_ids
Based on video length or template structure:
- Count total scenes → assign `001` through `NNN`
- Lock anchor positions (first, last 3)
- Fill flexible middle slots
### Step 3 — Transform text for each layer
For each scene:
1. Identify the `content_type` from the scene map
2. Find the source data field
3. Apply the transformation rules from Phase 5
4. Verify character count limits
5. Check pacing rhythm against adjacent scenes
### Step 4 — Present as table for review
| # | pair_id | Phase | Layer Name | Track | Text | Chars | Emotion |
|---|---------|-------|------------|-------|------|-------|---------|
| 1 | 001 | intro | `subtitle-intro-hook_claim-aspirational-001` | subtitle | 댕댕이가 먼저 뛰어간 숲 | 12 | Curiosity |
| 2 | 001 | intro | `keyword-intro-brand_name-sensory-001` | keyword | 스테이펫 | 4 | Curiosity |
| ... | ... | ... | ... | ... | ... | ... | ... |
# 입력 # 입력
**입력 1: 레이어 이름 리스트** **입력 1: 레이어 이름 리스트**
@ -83,5 +438,3 @@
Business Name: {customer_name} Business Name: {customer_name}
Region Details: {detail_region_info} Region Details: {detail_region_info}

2
app/utils/subtitles.py
View File

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

26
app/video/api/routers/v1/video.py
View File

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

1662
app/video/services/video.py
View File

File diff suppressed because it is too large Load Diff

10
config.py
View File

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