업로드 시 이미지 태깅 추가. 이미지 태그 DB 추가. 이미지 태그 prompt 추가
parent
ce79cb5d04
commit
c72736c334
|
|
@ -73,7 +73,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 # noqa: F401
|
||||
from app.home.models import Image, Project, MarketingIntel, ImageTag # 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
|
||||
|
|
@ -96,6 +96,7 @@ async def create_db_tables():
|
|||
SocialUpload.__table__,
|
||||
MarketingIntel.__table__,
|
||||
Dashboard.__table__,
|
||||
ImageTag.__table__,
|
||||
]
|
||||
|
||||
logger.info("Creating database tables...")
|
||||
|
|
|
|||
|
|
@ -9,9 +9,10 @@ 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
|
||||
from app.home.models import Image, MarketingIntel, ImageTag
|
||||
from app.user.dependencies.auth import get_current_user
|
||||
from app.user.models import User
|
||||
from app.home.schemas.home_schema import (
|
||||
|
|
@ -35,6 +36,7 @@ from app.utils.logger import get_logger
|
|||
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
|
||||
|
||||
# 로거 설정
|
||||
|
|
@ -451,255 +453,6 @@ 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)",
|
||||
|
|
@ -988,6 +741,10 @@ 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}, "
|
||||
|
|
@ -1003,3 +760,34 @@ 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])
|
||||
|
||||
for tag, tag_data in zip(null_tags, tag_datas):
|
||||
tag.img_tag = tag_data.model_dump(mode="json")
|
||||
|
||||
await session.commit()
|
||||
|
|
|
|||
|
|
@ -9,7 +9,8 @@ Home 모듈 SQLAlchemy 모델 정의
|
|||
from datetime import datetime
|
||||
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 app.database.session import Base
|
||||
|
|
@ -308,13 +309,50 @@ class MarketingIntel(Base):
|
|||
)
|
||||
|
||||
def __repr__(self) -> str:
|
||||
task_id_str = (
|
||||
(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 (
|
||||
f"<MarketingIntel(id={self.id}, place_id='{self.place_id}')>"
|
||||
)
|
||||
|
||||
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",
|
||||
)
|
||||
|
|
@ -0,0 +1,33 @@
|
|||
from app.utils.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()
|
||||
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()
|
||||
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) for image_input_data in image_input_data_list]
|
||||
image_result_list = await asyncio.gather(*image_result_tasks)
|
||||
|
||||
return image_result_list
|
||||
|
|
@ -1,6 +1,7 @@
|
|||
import json
|
||||
import re
|
||||
from pydantic import BaseModel
|
||||
from typing import List, Optional
|
||||
from openai import AsyncOpenAI
|
||||
|
||||
from app.utils.logger import get_logger
|
||||
|
|
@ -33,8 +34,24 @@ class ChatgptService:
|
|||
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}]
|
||||
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(
|
||||
|
|
@ -83,6 +100,8 @@ class ChatgptService:
|
|||
self,
|
||||
prompt : Prompt,
|
||||
input_data : dict,
|
||||
img_url : Optional[str] = None,
|
||||
img_detail_high : bool = False
|
||||
) -> BaseModel:
|
||||
prompt_text = prompt.build_prompt(input_data)
|
||||
|
||||
|
|
@ -91,5 +110,5 @@ class ChatgptService:
|
|||
|
||||
# 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
|
||||
response = await self._call_pydantic_output(prompt_text, prompt.prompt_output_class, prompt.prompt_model, img_url, img_detail_high)
|
||||
return response
|
||||
|
|
@ -432,6 +432,37 @@ 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
|
||||
|
||||
|
||||
async def template_matching_taged_image(
|
||||
self,
|
||||
template_id : str,
|
||||
taged_image_list : list,
|
||||
address : str
|
||||
) -> list:
|
||||
|
||||
template_data = await self.get_one_template_data(template_id)
|
||||
source_elements = template_data["source"]["elements"]
|
||||
template_component_data = self.parse_template_component_name(source_elements)
|
||||
|
||||
modifications = {}
|
||||
|
||||
for idx, (template_component_name, template_type) in enumerate(template_component_data.items()):
|
||||
match template_type:
|
||||
case "image":
|
||||
# modifications[template_component_name] = somethingtagedimage()
|
||||
pass
|
||||
case "text":
|
||||
if "address_input" in template_component_name:
|
||||
modifications[template_component_name] = address
|
||||
|
||||
# modifications["audio-music"] = music_url
|
||||
|
||||
async def template_connect_resource_blackbox(
|
||||
self,
|
||||
|
|
|
|||
|
|
@ -59,7 +59,15 @@ yt_upload_prompt = Prompt(
|
|||
prompt_model = prompt_settings.YOUTUBE_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
|
||||
)
|
||||
|
||||
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()
|
||||
|
|
@ -1,3 +1,4 @@
|
|||
from .lyric import LyricPromptInput, LyricPromptOutput
|
||||
from .marketing import MarketingPromptInput, MarketingPromptOutput
|
||||
from .youtube import YTUploadPromptInput, YTUploadPromptOutput
|
||||
from .youtube import YTUploadPromptInput, YTUploadPromptOutput
|
||||
from .image import *
|
||||
|
|
@ -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="이미지의 내러티브 상 점수")
|
||||
|
||||
|
|
@ -7,13 +7,11 @@ 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="마케팅 포지션 핵심 가치")
|
||||
|
|
@ -22,14 +20,12 @@ 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="셀링포인트 카테고리(한글)")
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
@ -186,6 +186,9 @@ class PromptSettings(BaseSettings):
|
|||
|
||||
YOUTUBE_PROMPT_FILE_NAME : str = Field(default="yt_upload_prompt.txt")
|
||||
YOUTUBE_PROMPT_MODEL : str = Field(default="gpt-5-mini")
|
||||
|
||||
IMAGE_TAG_PROMPT_FILE_NAME : str = Field(default="yt_upload_prompt.txt")
|
||||
IMAGE_TAG_PROMPT_MODEL : str = Field(default="gpt-5-mini")
|
||||
model_config = _base_config
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue