이미지 크롤링 로직 변경 및 자막 이미지 매칭 보완
parent
04d8614919
commit
46449a7a3f
|
|
@ -41,6 +41,8 @@ from app.utils.nvMapPwScraper import NvMapPwScraper
|
|||
from app.utils.prompts.prompts import marketing_prompt
|
||||
from app.utils.address_parser import extract_region_from_address
|
||||
from app.utils.autotag import autotag_images
|
||||
from app.utils.image_filter import filter_marketing_images, assemble_images
|
||||
from app.video.services.video import get_image_tags_by_task_id
|
||||
from config import MEDIA_ROOT
|
||||
|
||||
logger = get_logger("home")
|
||||
|
|
@ -102,7 +104,7 @@ async def _resolve_industry(category: str, customer_name: str = "") -> str:
|
|||
"""업체를 통합 프롬프트의 8개 industry enum 중 하나로 AI 분류.
|
||||
|
||||
분류 근거는 카테고리를 우선 사용하고, 카테고리가 없으면 업체명으로 분류한다.
|
||||
근거가 둘 다 없거나 분류 실패 시 빈 문자열 반환(프롬프트가 범용 처리).
|
||||
근거가 둘 다 없으면 빈 문자열 반환. API 장애 시 예외 전파(하위 분석도 동일 API 의존).
|
||||
"""
|
||||
if category:
|
||||
basis_label, basis_value = "네이버 지도 카테고리", category
|
||||
|
|
@ -110,7 +112,6 @@ async def _resolve_industry(category: str, customer_name: str = "") -> str:
|
|||
basis_label, basis_value = "업체명", customer_name
|
||||
else:
|
||||
return ""
|
||||
try:
|
||||
chatgpt = ChatgptService()
|
||||
prompt = (
|
||||
f"{basis_label}: '{basis_value}'\n"
|
||||
|
|
@ -127,9 +128,6 @@ async def _resolve_industry(category: str, customer_name: str = "") -> str:
|
|||
image_detail_high=False,
|
||||
)
|
||||
return result.industry
|
||||
except Exception as e:
|
||||
logger.warning(f"[industry] AI 분류 실패, 빈 값 반환: {e}")
|
||||
return ""
|
||||
|
||||
|
||||
@router.post(
|
||||
|
|
@ -247,12 +245,12 @@ async def _crawling_logic(
|
|||
)
|
||||
|
||||
step1_elapsed = (time.perf_counter() - step1_start) * 1000
|
||||
image_count = len(scraper.image_link_list) if scraper.image_link_list else 0
|
||||
logger.info(
|
||||
f"[crawling] Step 1 완료 - 이미지 {image_count}개 ({step1_elapsed:.1f}ms)"
|
||||
f"[crawling] Step 1 완료 - 업체사진 {len(scraper.owner_images or [])}개, "
|
||||
f"보충사진 {len(scraper.extra_photo_urls or [])}개 ({step1_elapsed:.1f}ms)"
|
||||
)
|
||||
|
||||
# ========== Step 2: 정보 가공 ==========
|
||||
# ========== Step 2: 정보 가공 (industry 선행 계산) ==========
|
||||
step2_start = time.perf_counter()
|
||||
logger.info("[crawling] Step 2: 정보 가공 시작...")
|
||||
|
||||
|
|
@ -265,7 +263,21 @@ async def _crawling_logic(
|
|||
customer_name = scraper.base_info.get("name", "")
|
||||
category = scraper.base_info.get("category", "")
|
||||
region = _extract_region_from_address(road_address, jibun_address)
|
||||
try:
|
||||
# industry는 마케팅 적합성 필터(Step 3)와 ProcessedInfo 양쪽에 필요하므로
|
||||
# 이미지 필터링보다 먼저 계산한다.
|
||||
industry = await _resolve_industry(category, customer_name)
|
||||
except ChatGPTResponseError as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_502_BAD_GATEWAY,
|
||||
detail=f"업종 분류 중 ChatGPT 오류가 발생했습니다: {e.error_message}",
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"[crawling] Step 2 FAILED - 업종 분류 오류: {e}")
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_502_BAD_GATEWAY,
|
||||
detail="업종 분류 중 오류가 발생했습니다.",
|
||||
)
|
||||
|
||||
processed_info = ProcessedInfo(
|
||||
customer_name=customer_name,
|
||||
|
|
@ -277,15 +289,53 @@ async def _crawling_logic(
|
|||
step2_elapsed = (time.perf_counter() - step2_start) * 1000
|
||||
logger.info(
|
||||
f"[crawling] Step 2 완료 - {customer_name}, {region}, "
|
||||
f"category={category!r}, place_type={scraper.place_type!r}, industry={industry!r} ({step2_elapsed:.1f}ms)"
|
||||
f"category={category!r}, industry={industry!r} ({step2_elapsed:.1f}ms)"
|
||||
)
|
||||
|
||||
# ========== Step 3: ChatGPT 마케팅 분석 ==========
|
||||
# ========== Step 3: 이미지 마케팅 적합성 필터링 ==========
|
||||
# 업체 사진이 SUPPLEMENT_THRESHOLD(30장) 이상이면 보충이 불필요하므로
|
||||
# 방문자 사진 필터링(Gemini 호출) 자체를 건너뛴다.
|
||||
step3_start = time.perf_counter()
|
||||
logger.info("[crawling] Step 3: ChatGPT 마케팅 분석 시작...")
|
||||
owner_images = scraper.owner_images or []
|
||||
extra_photo_urls = scraper.extra_photo_urls or []
|
||||
|
||||
if len(owner_images) >= NvMapScraper.SUPPLEMENT_THRESHOLD:
|
||||
scraper.image_link_list = owner_images[: NvMapScraper.MAX_IMAGES]
|
||||
step3_elapsed = (time.perf_counter() - step3_start) * 1000
|
||||
logger.info(
|
||||
f"[crawling] Step 3 SKIP - 업체 사진 {len(owner_images)}장 "
|
||||
f"≥ {NvMapScraper.SUPPLEMENT_THRESHOLD}장 → 방문자 사진 필터링 생략"
|
||||
)
|
||||
else:
|
||||
try:
|
||||
extra_pass_flags = await filter_marketing_images(
|
||||
[img["original"] for img in extra_photo_urls], industry
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"[crawling] Step 3 FAILED - 이미지 필터링 중 오류: {e}")
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_502_BAD_GATEWAY,
|
||||
detail="이미지 마케팅 적합성 필터링 중 오류가 발생했습니다.",
|
||||
)
|
||||
scraper.image_link_list = assemble_images(
|
||||
owner_images, extra_photo_urls, extra_pass_flags, NvMapScraper.MAX_IMAGES
|
||||
)
|
||||
passed_count = sum(extra_pass_flags)
|
||||
step3_elapsed = (time.perf_counter() - step3_start) * 1000
|
||||
logger.info(
|
||||
f"[crawling] Step 3 완료 - 방문자 사진 {len(extra_photo_urls)}장 중 "
|
||||
f"{passed_count}장 통과 → 최종 이미지 {len(scraper.image_link_list)}장 "
|
||||
f"({step3_elapsed:.1f}ms)"
|
||||
)
|
||||
if not scraper.image_link_list:
|
||||
logger.warning("[crawling] Step 3 - 필터링 후 사용 가능 이미지가 0장입니다.")
|
||||
|
||||
# ========== Step 4: ChatGPT 마케팅 분석 ==========
|
||||
step4_start = time.perf_counter()
|
||||
logger.info("[crawling] Step 4: ChatGPT 마케팅 분석 시작...")
|
||||
|
||||
try:
|
||||
# Step 3-1: ChatGPT 서비스 초기화 및 입력 데이터 구성
|
||||
# Step 4-1: ChatGPT 서비스 초기화 및 입력 데이터 구성
|
||||
chatgpt_service = ChatgptService()
|
||||
input_marketing_data = {
|
||||
"customer_name": customer_name,
|
||||
|
|
@ -294,12 +344,12 @@ async def _crawling_logic(
|
|||
"industry": industry, # 통합 프롬프트 내부 분기용 업종 enum
|
||||
}
|
||||
|
||||
# Step 3-2: GPT API 호출 → 구조화된 마케팅 분석 결과 반환
|
||||
# Step 4-2: GPT API 호출 → 구조화된 마케팅 분석 결과 반환
|
||||
marketing_analysis = await chatgpt_service.generate_structured_output(
|
||||
marketing_prompt, input_marketing_data
|
||||
)
|
||||
|
||||
# Step 3-3: 분석 결과 DB 저장 (industry는 Project로 흐르므로 여기엔 미저장)
|
||||
# Step 4-3: 분석 결과 DB 저장 (industry는 Project로 흐르므로 여기엔 미저장)
|
||||
marketing_intel = MarketingIntel(
|
||||
place_id=scraper.place_id,
|
||||
intel_result=marketing_analysis.model_dump(),
|
||||
|
|
@ -310,26 +360,26 @@ async def _crawling_logic(
|
|||
m_id = marketing_intel.id
|
||||
logger.debug(f"[MarketingPrompt] INSERT place_id={marketing_intel.place_id} id={marketing_intel.id}")
|
||||
|
||||
step3_elapsed = (time.perf_counter() - step3_start) * 1000
|
||||
step4_elapsed = (time.perf_counter() - step4_start) * 1000
|
||||
logger.info(
|
||||
f"[crawling] Step 3 완료 - 마케팅 분석 성공 ({step3_elapsed:.1f}ms)"
|
||||
f"[crawling] Step 4 완료 - 마케팅 분석 성공 ({step4_elapsed:.1f}ms)"
|
||||
)
|
||||
|
||||
except ChatGPTResponseError as e:
|
||||
step3_elapsed = (time.perf_counter() - step3_start) * 1000
|
||||
step4_elapsed = (time.perf_counter() - step4_start) * 1000
|
||||
logger.error(
|
||||
f"[crawling] Step 3 FAILED - ChatGPT Error: status={e.status}, "
|
||||
f"code={e.error_code}, message={e.error_message} ({step3_elapsed:.1f}ms)"
|
||||
f"[crawling] Step 4 FAILED - ChatGPT Error: status={e.status}, "
|
||||
f"code={e.error_code}, message={e.error_message} ({step4_elapsed:.1f}ms)"
|
||||
)
|
||||
marketing_analysis = None
|
||||
gpt_status = "failed"
|
||||
|
||||
except Exception as e:
|
||||
step3_elapsed = (time.perf_counter() - step3_start) * 1000
|
||||
step4_elapsed = (time.perf_counter() - step4_start) * 1000
|
||||
logger.error(
|
||||
f"[crawling] Step 3 FAILED - GPT 마케팅 분석 중 오류: {e} ({step3_elapsed:.1f}ms)"
|
||||
f"[crawling] Step 4 FAILED - GPT 마케팅 분석 중 오류: {e} ({step4_elapsed:.1f}ms)"
|
||||
)
|
||||
logger.exception("[crawling] Step 3 상세 오류:")
|
||||
logger.exception("[crawling] Step 4 상세 오류:")
|
||||
marketing_analysis = None
|
||||
gpt_status = "failed"
|
||||
else:
|
||||
|
|
@ -346,7 +396,9 @@ async def _crawling_logic(
|
|||
if scraper.base_info:
|
||||
logger.info(f"[crawling] - Step 2 (정보가공): {step2_elapsed:.1f}ms")
|
||||
if "step3_elapsed" in locals():
|
||||
logger.info(f"[crawling] - Step 3 (GPT 분석): {step3_elapsed:.1f}ms")
|
||||
logger.info(f"[crawling] - Step 3 (이미지 필터링): {step3_elapsed:.1f}ms")
|
||||
if "step4_elapsed" in locals():
|
||||
logger.info(f"[crawling] - Step 4 (GPT 분석): {step4_elapsed:.1f}ms")
|
||||
|
||||
return {
|
||||
"status": gpt_status if 'gpt_status' in locals() else "completed",
|
||||
|
|
@ -603,6 +655,10 @@ async def upload_images_blob(
|
|||
default=None,
|
||||
description="이미지 바이너리 파일 목록",
|
||||
),
|
||||
industry: str = Form(
|
||||
default="",
|
||||
description="업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction). 크롤링 응답의 industry 값을 그대로 전달",
|
||||
),
|
||||
current_user: User = Depends(get_current_user),
|
||||
) -> ImageUploadResponse:
|
||||
"""이미지 업로드 (URL + Azure Blob Storage)
|
||||
|
|
@ -804,9 +860,21 @@ async def upload_images_blob(
|
|||
image_urls = [img.img_url for img in result_images]
|
||||
|
||||
logger.info(f"[image_tagging] START - task_id: {task_id}")
|
||||
await tagging_images(image_urls, clear_old_tags=True)
|
||||
await tagging_images(image_urls, industry=industry, clear_old_tags=True)
|
||||
logger.info(f"[image_tagging] Done - task_id: {task_id}")
|
||||
|
||||
# 태깅 직후 영상 생성에 사용 가능한 이미지가 하나도 없으면 조기에 실패시킨다.
|
||||
# (여기서 걸러지지 않으면 훨씬 나중인 영상 생성 단계에서야 슬롯 미배정으로 발견됨)
|
||||
# marketing_acceptable 필터링은 크롤링 단계에서 이미 완료되었으므로 여기서는 재필터링하지 않는다.
|
||||
taged_image_list = await get_image_tags_by_task_id(task_id)
|
||||
logger.info(f"태깅된 이미지: {len(taged_image_list)}개 - task_id: {task_id}")
|
||||
if not taged_image_list:
|
||||
logger.error(f"[image_tagging] 영상 생성에 적합한 이미지가 없음 - task_id: {task_id}")
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="영상 생성에 적합한 이미지가 없습니다. 다른 이미지로 다시 업로드해주세요.",
|
||||
)
|
||||
|
||||
total_time = time.perf_counter() - request_start
|
||||
logger.info(
|
||||
f"[upload_images_blob] SUCCESS - task_id: {task_id}, "
|
||||
|
|
@ -826,6 +894,7 @@ async def upload_images_blob(
|
|||
|
||||
async def tagging_images(
|
||||
image_urls : list[str],
|
||||
industry: str = "",
|
||||
clear_old_tags : bool = False
|
||||
) -> None:
|
||||
# 1. 조회
|
||||
|
|
@ -851,8 +920,8 @@ async def tagging_images(
|
|||
await session.commit()
|
||||
|
||||
if null_imts:
|
||||
tag_datas = await autotag_images([img.img_url for img in null_imts])
|
||||
print(tag_datas)
|
||||
tag_datas = await autotag_images([img.img_url for img in null_imts], industry=industry)
|
||||
# print(tag_datas)
|
||||
|
||||
async with AsyncSessionLocal() as session:
|
||||
for tag, tag_data in zip(null_imts, tag_datas):
|
||||
|
|
|
|||
|
|
@ -116,7 +116,7 @@ class Project(Base):
|
|||
|
||||
industry: Mapped[str] = mapped_column(
|
||||
String(50),
|
||||
nullable=True,
|
||||
nullable=False,
|
||||
default="",
|
||||
comment="업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction)",
|
||||
)
|
||||
|
|
@ -314,6 +314,12 @@ class MarketingIntel(Base):
|
|||
comment="자막 정보 생성 결과물",
|
||||
)
|
||||
|
||||
image_match : Mapped[dict[str, Any]] = mapped_column(
|
||||
JSON,
|
||||
nullable=True,
|
||||
comment="이미지 슬롯 배정 결과물 — {슬롯명: image_url, 'audio-music': music_url, ...}",
|
||||
)
|
||||
|
||||
created_at: Mapped[datetime] = mapped_column(
|
||||
DateTime,
|
||||
nullable=False,
|
||||
|
|
|
|||
|
|
@ -98,6 +98,12 @@ class ProcessedInfo(BaseModel):
|
|||
# selling_points: list[str] = Field(default_factory=list, description="추천 부대시설 목록")
|
||||
|
||||
|
||||
class ImageListItem(BaseModel):
|
||||
"""크롤링 이미지 아이템 (미리보기 URL + 원본 URL)"""
|
||||
preview: str = Field(..., description="미리보기용 CDN URL")
|
||||
original: str = Field(..., description="업로드용 원본 URL")
|
||||
|
||||
|
||||
class CrawlingResponse(BaseModel):
|
||||
"""크롤링 응답 스키마"""
|
||||
|
||||
|
|
@ -105,7 +111,7 @@ class CrawlingResponse(BaseModel):
|
|||
json_schema_extra={
|
||||
"example": {
|
||||
"status": "completed",
|
||||
"image_list": ["https://example.com/image1.jpg", "https://example.com/image2.jpg"],
|
||||
"image_list": [{"preview": "https://example.com/image1.jpg", "original": "https://example.com/image1.jpg"}],
|
||||
"image_count": 2,
|
||||
"processed_info": {
|
||||
"customer_name": "스테이 머뭄",
|
||||
|
|
@ -234,7 +240,7 @@ class CrawlingResponse(BaseModel):
|
|||
default="completed",
|
||||
description="처리 상태 (completed: 성공, failed: ChatGPT 분석 실패)"
|
||||
)
|
||||
image_list: Optional[list[str]] = Field(None, description="이미지 URL 목록")
|
||||
image_list: Optional[list[ImageListItem]] = Field(None, description="이미지 목록 (preview/original URL)")
|
||||
image_count: Optional[int] = Field(None, description="이미지 개수")
|
||||
processed_info: Optional[ProcessedInfo] = Field(
|
||||
None, description="가공된 장소 정보 (customer_name, region, detail_region_info)"
|
||||
|
|
|
|||
|
|
@ -42,7 +42,8 @@ from app.lyric.schemas.lyric import (
|
|||
LyricStatusResponse,
|
||||
SubtitleStatusResponse,
|
||||
)
|
||||
from app.lyric.worker.lyric_task import generate_lyric_background, generate_subtitle_background
|
||||
from app.lyric.worker.lyric_task import generate_lyric_background
|
||||
from app.video.worker.creative_assets_task import generate_subtitle_background
|
||||
from app.utils.prompts.chatgpt_prompt import ChatgptService
|
||||
from app.utils.logger import get_logger
|
||||
from app.utils.pagination import PaginatedResponse, get_paginated
|
||||
|
|
@ -160,6 +161,7 @@ async def get_lyric_by_task_id(
|
|||
project_id=lyric.project_id,
|
||||
status=lyric.status,
|
||||
lyric_result=lyric.lyric_result,
|
||||
genre=lyric.genre,
|
||||
created_at=lyric.created_at,
|
||||
)
|
||||
|
||||
|
|
@ -269,15 +271,21 @@ async def generate_lyric(
|
|||
step1_start = time.perf_counter()
|
||||
logger.debug(f"[generate_lyric] Step 1: 서비스 초기화 및 프롬프트 생성...")
|
||||
|
||||
timing_rules = {
|
||||
"60s" : """
|
||||
8–12 lines
|
||||
Full verse flow, immersive mood
|
||||
"""
|
||||
}
|
||||
marketing_intel_result = await session.execute(select(MarketingIntel).where(MarketingIntel.id == request_body.m_id))
|
||||
marketing_intel = marketing_intel_result.scalar_one_or_none()
|
||||
|
||||
# 자동 선택(genre 미지정) 재생성 시, 직전에 사용된 장르를 조회해 이번엔 다른 장르가 나오도록 제외 처리
|
||||
exclude_genre = ""
|
||||
if not request_body.genre:
|
||||
previous_lyric_result = await session.execute(
|
||||
select(Lyric)
|
||||
.where(Lyric.task_id == task_id, Lyric.genre.is_not(None))
|
||||
.order_by(Lyric.created_at.desc())
|
||||
.limit(1)
|
||||
)
|
||||
previous_lyric = previous_lyric_result.scalar_one_or_none()
|
||||
if previous_lyric:
|
||||
exclude_genre = previous_lyric.genre
|
||||
|
||||
lyric_input_data = {
|
||||
"customer_name" : request_body.customer_name,
|
||||
|
|
@ -285,7 +293,8 @@ async def generate_lyric(
|
|||
"detail_region_info" : request_body.detail_region_info or "",
|
||||
"marketing_intelligence_summary" : json.dumps(marketing_intel.intel_result, ensure_ascii = False),
|
||||
"language" : request_body.language,
|
||||
"timing_rules" : timing_rules["60s"], # 아직은 선택지 하나
|
||||
"genre": request_body.genre or "", # 비어있으면 GPT가 가사 무드에 맞춰 장르를 직접 추천
|
||||
"exclude_genre": exclude_genre, # 자동 선택 재생성 시 직전 장르 제외
|
||||
"industry": request_body.industry, # 크롤 응답에서 받아 전달된 업종 enum
|
||||
}
|
||||
|
||||
|
|
@ -340,6 +349,7 @@ async def generate_lyric(
|
|||
lyric_prompt=estimated_prompt,
|
||||
lyric_result=None,
|
||||
language=request_body.language,
|
||||
genre=request_body.genre or None, # 자동 선택인 경우 GPT 응답 완료 후 채워짐
|
||||
)
|
||||
session.add(lyric)
|
||||
await session.commit()
|
||||
|
|
@ -575,19 +585,29 @@ async def get_subtitle_status(
|
|||
detail=f"task_id '{task_id}'에 해당하는 마케팅 인텔리전스를 찾을 수 없습니다.",
|
||||
)
|
||||
|
||||
if intel.subtitle:
|
||||
logger.info(f"[get_subtitle_status] completed - task_id: {task_id}")
|
||||
# 자막과 이미지 배정 모두 완료돼야 영상 생성 가능
|
||||
subtitle_done = bool(intel.subtitle)
|
||||
image_match_done = bool(intel.image_match)
|
||||
|
||||
if subtitle_done and image_match_done:
|
||||
logger.info(f"[get_subtitle_status] completed (subtitle+image_match) - task_id: {task_id}")
|
||||
return SubtitleStatusResponse(
|
||||
task_id=task_id,
|
||||
status="completed",
|
||||
message="자막 생성이 완료되었습니다.",
|
||||
message="자막 생성 및 이미지 배정이 완료되었습니다.",
|
||||
)
|
||||
|
||||
logger.info(f"[get_subtitle_status] pending - task_id: {task_id}")
|
||||
pending_parts = []
|
||||
if not subtitle_done:
|
||||
pending_parts.append("자막")
|
||||
if not image_match_done:
|
||||
pending_parts.append("이미지 배정")
|
||||
pending_msg = ", ".join(pending_parts)
|
||||
logger.info(f"[get_subtitle_status] pending ({pending_msg}) - task_id: {task_id}")
|
||||
return SubtitleStatusResponse(
|
||||
task_id=task_id,
|
||||
status="pending",
|
||||
message="자막 생성이 진행 중입니다. 잠시 후 다시 확인해주세요.",
|
||||
message=f"{pending_msg} 처리가 진행 중입니다. 잠시 후 다시 확인해주세요.",
|
||||
)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -93,6 +93,13 @@ class Lyric(Base):
|
|||
comment="가사 출력 언어 (Korean, English, Chinese, Japanese, Thai, Vietnamese)",
|
||||
)
|
||||
|
||||
genre: Mapped[str | None] = mapped_column(
|
||||
String(20),
|
||||
nullable=True,
|
||||
comment="음악 장르 (kpop, pop, ballad, hip-hop, rnb, edm, jazz, rock). "
|
||||
"수동 선택 시 요청값, 자동 선택 시 GPT 추천값",
|
||||
)
|
||||
|
||||
is_deleted: Mapped[bool] = mapped_column(
|
||||
Boolean,
|
||||
nullable=False,
|
||||
|
|
|
|||
|
|
@ -78,6 +78,16 @@ class GenerateLyricRequest(BaseModel):
|
|||
m_id : Optional[int] = Field(None, description="마케팅 인텔리전스 ID 값")
|
||||
industry: str = Field(default="", description="업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction)")
|
||||
instrumental: bool = Field(default=False, description="BGM 전용 모드 (가사 생성 안 함)")
|
||||
genre: Optional[str] = Field(
|
||||
None,
|
||||
description="사용자가 수동으로 고른 음악 장르 (kpop|pop|ballad|hip-hop|rnb|edm|jazz|rock). "
|
||||
"'자동 선택'인 경우 None으로 전달하면 GPT가 가사 무드에 맞춰 추천",
|
||||
)
|
||||
genre: Optional[str] = Field(
|
||||
None,
|
||||
description="사용자가 수동으로 고른 음악 장르 (kpop|pop|ballad|hip-hop|rnb|edm|jazz|rock). "
|
||||
"'자동 선택'인 경우 None으로 전달하면 GPT가 가사 무드에 맞춰 추천",
|
||||
)
|
||||
|
||||
|
||||
class GenerateLyricResponse(BaseModel):
|
||||
|
|
@ -200,6 +210,10 @@ class LyricDetailResponse(BaseModel):
|
|||
project_id: int = Field(..., description="프로젝트 ID")
|
||||
status: str = Field(..., description="처리 상태 (processing, completed, failed)")
|
||||
lyric_result: Optional[str] = Field(None, description="생성된 가사 또는 에러 메시지 (실패 시)")
|
||||
genre: Optional[str] = Field(
|
||||
None,
|
||||
description="확정된 음악 장르. 수동 선택 시 요청값, 자동 선택 시 GPT 추천값 (완료 전에는 None)",
|
||||
)
|
||||
created_at: Optional[datetime] = Field(None, description="생성 일시")
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -4,30 +4,24 @@ Lyric Background Tasks
|
|||
가사 생성 관련 백그라운드 태스크를 정의합니다.
|
||||
"""
|
||||
|
||||
import traceback
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy.exc import SQLAlchemyError
|
||||
|
||||
from app.database.session import BackgroundSessionLocal
|
||||
from app.home.models import Image, Project, MarketingIntel
|
||||
from app.lyric.models import Lyric
|
||||
from app.utils.prompts.chatgpt_prompt import ChatgptService, ChatGPTResponseError
|
||||
from app.utils.subtitles import SubtitleContentsGenerator
|
||||
from app.utils.creatomate import CreatomateService
|
||||
from app.utils.prompts.prompts import Prompt
|
||||
from app.utils.logger import get_logger
|
||||
|
||||
# 로거 설정
|
||||
logger = get_logger("lyric")
|
||||
|
||||
|
||||
async def _update_lyric_status(
|
||||
task_id: str,
|
||||
status: str,
|
||||
result: str | None = None,
|
||||
lyric_id: int | None = None,
|
||||
genre: str | None = None,
|
||||
) -> bool:
|
||||
"""Lyric 테이블의 상태를 업데이트합니다.
|
||||
|
||||
|
|
@ -36,6 +30,7 @@ async def _update_lyric_status(
|
|||
status: 변경할 상태 ("processing", "completed", "failed")
|
||||
result: 가사 결과 또는 에러 메시지
|
||||
lyric_id: 특정 Lyric 레코드 ID (재생성 시 정확한 레코드 식별용)
|
||||
genre: 확정된 음악 장르 (자동 선택이었던 경우 GPT 추천값으로 갱신)
|
||||
|
||||
Returns:
|
||||
bool: 업데이트 성공 여부
|
||||
|
|
@ -61,6 +56,8 @@ async def _update_lyric_status(
|
|||
lyric.status = status
|
||||
if result is not None:
|
||||
lyric.lyric_result = result
|
||||
if genre is not None:
|
||||
lyric.genre = genre
|
||||
await session.commit()
|
||||
logger.info(f"[Lyric] Status updated - task_id: {task_id}, lyric_id: {lyric_id}, status: {status}")
|
||||
return True
|
||||
|
|
@ -117,14 +114,18 @@ async def generate_lyric_background(
|
|||
#result = await service.generate(prompt=prompt)
|
||||
result_response = await chatgpt.generate_structured_output(prompt, lyric_input_data)
|
||||
result = result_response.lyric
|
||||
confirmed_genre = result_response.recommended_genre
|
||||
step2_elapsed = (time.perf_counter() - step2_start) * 1000
|
||||
logger.info(f"[generate_lyric_background] Step 2 완료 - 응답 {len(result)}자 ({step2_elapsed:.1f}ms)")
|
||||
logger.info(
|
||||
f"[generate_lyric_background] Step 2 완료 - 응답 {len(result)}자, "
|
||||
f"genre: {confirmed_genre} ({step2_elapsed:.1f}ms)"
|
||||
)
|
||||
|
||||
# ========== Step 3: DB 상태 업데이트 ==========
|
||||
step3_start = time.perf_counter()
|
||||
logger.debug(f"[generate_lyric_background] Step 3: DB 상태 업데이트...")
|
||||
|
||||
await _update_lyric_status(task_id, "completed", result, lyric_id)
|
||||
await _update_lyric_status(task_id, "completed", result, lyric_id, genre=confirmed_genre)
|
||||
|
||||
step3_elapsed = (time.perf_counter() - step3_start) * 1000
|
||||
logger.debug(f"[generate_lyric_background] Step 3 완료 ({step3_elapsed:.1f}ms)")
|
||||
|
|
@ -155,73 +156,3 @@ async def generate_lyric_background(
|
|||
elapsed = (time.perf_counter() - task_start) * 1000
|
||||
logger.error(f"[generate_lyric_background] EXCEPTION - task_id: {task_id}, error: {e} ({elapsed:.1f}ms)", exc_info=True)
|
||||
await _update_lyric_status(task_id, "failed", f"Error: {str(e)}", lyric_id)
|
||||
|
||||
async def generate_subtitle_background(
|
||||
orientation: str,
|
||||
task_id: str,
|
||||
max_retries: int = 3,
|
||||
) -> None:
|
||||
logger.info(f"[generate_subtitle_background] START - task_id: {task_id}, orientation: {orientation}")
|
||||
|
||||
for attempt in range(1, max_retries + 1):
|
||||
try:
|
||||
creatomate_service = CreatomateService(orientation=orientation)
|
||||
template = await creatomate_service.get_one_template_data(creatomate_service.template_id)
|
||||
pitchings = creatomate_service.extract_text_format_from_template(template)
|
||||
|
||||
subtitle_generator = SubtitleContentsGenerator()
|
||||
|
||||
async with BackgroundSessionLocal() as session:
|
||||
project_result = await session.execute(
|
||||
select(Project)
|
||||
.where(Project.task_id == task_id)
|
||||
.order_by(Project.created_at.desc())
|
||||
.limit(1)
|
||||
)
|
||||
project = project_result.scalar_one_or_none()
|
||||
marketing_result = await session.execute(
|
||||
select(MarketingIntel).where(MarketingIntel.id == project.marketing_intelligence)
|
||||
)
|
||||
marketing_intelligence = marketing_result.scalar_one_or_none()
|
||||
|
||||
store_address = project.detail_region_info
|
||||
customer_name = project.store_name
|
||||
language = project.language or "Korean"
|
||||
logger.info(f"[generate_subtitle_background] customer_name: {customer_name}, store_address: {store_address}, language: {language}")
|
||||
|
||||
generated_subtitles = await subtitle_generator.generate_subtitle_contents(
|
||||
marketing_intelligence=marketing_intelligence.intel_result,
|
||||
pitching_label_list=pitchings,
|
||||
customer_name=customer_name,
|
||||
detail_region_info=store_address,
|
||||
language=language,
|
||||
industry=project.industry,
|
||||
)
|
||||
pitching_output_list = generated_subtitles.pitching_results
|
||||
|
||||
subtitle_modifications = {
|
||||
pitching_output.pitching_tag: pitching_output.pitching_data
|
||||
for pitching_output in pitching_output_list
|
||||
}
|
||||
logger.info(f"[generate_subtitle_background] subtitle_modifications: {subtitle_modifications}")
|
||||
|
||||
async with BackgroundSessionLocal() as session:
|
||||
marketing_result = await session.execute(
|
||||
select(MarketingIntel).where(MarketingIntel.id == project.marketing_intelligence)
|
||||
)
|
||||
marketing_intelligence = marketing_result.scalar_one_or_none()
|
||||
marketing_intelligence.subtitle = subtitle_modifications
|
||||
await session.commit()
|
||||
|
||||
logger.info(f"[generate_subtitle_background] DONE - task_id: {task_id} (attempt {attempt}/{max_retries})")
|
||||
return
|
||||
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"[generate_subtitle_background] FAILED (attempt {attempt}/{max_retries}) - task_id: {task_id}, error: {e}",
|
||||
exc_info=True,
|
||||
)
|
||||
if attempt < max_retries:
|
||||
logger.info(f"[generate_subtitle_background] 재시도 중... ({attempt + 1}/{max_retries}) - task_id: {task_id}")
|
||||
|
||||
logger.error(f"[generate_subtitle_background] 모든 재시도 실패 - task_id: {task_id}")
|
||||
|
|
|
|||
|
|
@ -6,10 +6,11 @@ from app.utils.prompts.schemas import SpaceType, Subject, Camera, MotionRecommen
|
|||
|
||||
import asyncio
|
||||
|
||||
async def autotag_image(image_url : str) -> list[str]: #tag_list
|
||||
async def autotag_image(image_url : str, industry: str = "") -> list[str]: #tag_list
|
||||
chatgpt = ChatgptService(model_type="gemini")
|
||||
image_input_data = {
|
||||
"img_url" : image_url,
|
||||
"industry" : industry,
|
||||
"space_type" : list(SpaceType),
|
||||
"subject" : list(Subject),
|
||||
"camera" : list(Camera),
|
||||
|
|
@ -19,10 +20,11 @@ async def autotag_image(image_url : str) -> list[str]: #tag_list
|
|||
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
|
||||
async def autotag_images(image_url_list : list[str], industry: str = "") -> list[dict]: #tag_list
|
||||
chatgpt = ChatgptService(model_type="gemini")
|
||||
image_input_data_list = [{
|
||||
"img_url" : image_url,
|
||||
"industry" : industry,
|
||||
"space_type" : list(SpaceType),
|
||||
"subject" : list(Subject),
|
||||
"camera" : list(Camera),
|
||||
|
|
|
|||
|
|
@ -233,7 +233,7 @@ DHST0002 = "3f194cc7-464e-4581-9db2-179d42d3e40f"
|
|||
DHST0003 = "f45df555-2956-4a13-9004-ead047070b3d"
|
||||
DVST0001T = "fe11aeab-ff29-4bc8-9f75-c695c7e243e6"
|
||||
HST_LIST = [DHST0001,DHST0002,DHST0003]
|
||||
VST_LIST = [DVST0001,DVST0002,DVST0003, DVST0001T]
|
||||
VST_LIST = [DVST0001T]
|
||||
|
||||
SCENE_TRACK = 1
|
||||
AUDIO_TRACK = 2
|
||||
|
|
@ -242,11 +242,13 @@ KEYWORD_TRACK = 4
|
|||
|
||||
def select_template(orientation:OrientationType):
|
||||
if orientation == "horizontal":
|
||||
return DHST0001
|
||||
template_id = DHST0001
|
||||
elif orientation == "vertical":
|
||||
return DVST0001T
|
||||
template_id = random.choice(VST_LIST)
|
||||
else:
|
||||
raise
|
||||
logger.info(f"[select_template] orientation={orientation}, template_id={template_id}")
|
||||
return template_id
|
||||
|
||||
async def get_shared_client() -> httpx.AsyncClient:
|
||||
"""공유 HTTP 클라이언트를 반환합니다. 없으면 생성합니다."""
|
||||
|
|
@ -456,50 +458,155 @@ class CreatomateService:
|
|||
def template_matching_taged_image(
|
||||
self,
|
||||
template: dict,
|
||||
taged_image_list : list, # [{"image_name" : str , "image_tag" : dict}]
|
||||
taged_image_list: list, # [{"image_url": str, "image_tag": dict}] — 이미 마케팅 적합성 필터를 통과한 이미지만 전달할 것
|
||||
music_url: str,
|
||||
address: str,
|
||||
duplicate: bool = False
|
||||
) -> list:
|
||||
) -> tuple[dict, dict]:
|
||||
"""템플릿 슬롯에 이미지를 배정합니다.
|
||||
|
||||
Returns:
|
||||
(modifications, assigned) 튜플
|
||||
- modifications: {슬롯명: image_url, "audio-music": music_url, ...} — modify_element에 전달
|
||||
- assigned: {슬롯명: image_tag} — 자막 생성 컨텍스트에 전달 (modify_element에 넣지 않음)
|
||||
|
||||
배정 전략 (duplicate=False):
|
||||
- 순수 greedy pop 대신 "난이도 우선 greedy" 적용
|
||||
- 사전에 전체 슬롯 점수를 계산해 최고 달성 점수가 낮은(=까다로운) 슬롯을 먼저 배정
|
||||
- 이미지 배정 직전 점수를 재계산 (pop 이후 인덱스 변동 대응)
|
||||
|
||||
배정 전략 (duplicate=True, 이미지 < 슬롯):
|
||||
- 1단계(최소 커버리지 보장): 이미지별로 가장 잘 맞는 슬롯을 찾아 pool의 모든 이미지를
|
||||
서로 다른 슬롯에 먼저 배정한다 (이미지 수 <= 슬롯 수이므로 전량 배정 가능).
|
||||
슬롯을 먼저 소진시키는 것이 아니라 "이미지"를 기준으로 난이도 우선 배정하여,
|
||||
특정 이미지 몇 장이 여러 슬롯을 독점하는 것을 방지한다.
|
||||
- 2단계: 1단계에서 못 채운 나머지 슬롯은 기존처럼 pool 전체에서 최고점 이미지를
|
||||
배정한다 (여기서부터는 중복 재사용 불가피).
|
||||
|
||||
Note:
|
||||
marketing_acceptable 필터링은 크롤링 단계에서 이미 완료되므로 여기서는 재필터링하지 않는다.
|
||||
taged_image_list를 그대로 pool로 사용한다.
|
||||
"""
|
||||
source_elements = template["source"]["elements"]
|
||||
template_component_data = self.parse_template_component_name(source_elements)
|
||||
|
||||
taged_image_list = [img for img in taged_image_list if img.get("image_tag") is not None]
|
||||
modifications = {}
|
||||
pool = taged_image_list
|
||||
modifications: dict = {}
|
||||
assigned: dict = {} # {슬롯명: image_tag} — 자막 컨텍스트용
|
||||
|
||||
for slot_idx, (template_component_name, template_type) in enumerate(template_component_data.items()):
|
||||
match template_type:
|
||||
case "image":
|
||||
image_score_list = self.calculate_image_slot_score_multi(taged_image_list, template_component_name)
|
||||
if duplicate:
|
||||
# 이미지 부족(슬롯 수 > 이미지 수): 재사용이 불가피하므로 슬롯에 가장 적합한 이미지(최고점) 선택
|
||||
maximum_idx = image_score_list.index(max(image_score_list))
|
||||
selected = taged_image_list[maximum_idx]
|
||||
# 이미지 슬롯과 텍스트 슬롯 분리
|
||||
image_slots = [name for name, t in template_component_data.items() if t == "image"]
|
||||
text_slots = [(name, t) for name, t in template_component_data.items() if t == "text"]
|
||||
|
||||
if not pool:
|
||||
logger.warning("[template_matching_taged_image] 태그된 이미지가 없음 — 이미지 슬롯 배정 불가")
|
||||
elif duplicate:
|
||||
# 이미지 부족(슬롯 수 > 이미지 수)
|
||||
# Step 1: 최소 커버리지 보장 — 이미지마다 가장 잘 맞는 슬롯을 찾아 서로 다른 슬롯에 배정
|
||||
remaining_slots = list(image_slots)
|
||||
|
||||
def _best_slot_for_image(image: dict, slots: list) -> tuple[str | None, float]:
|
||||
best_slot, best_score = None, -1.0
|
||||
for slot in slots:
|
||||
score = self.calculate_image_slot_score_multi([image], slot)[0]
|
||||
if score > best_score:
|
||||
best_slot, best_score = slot, score
|
||||
return best_slot, best_score
|
||||
|
||||
# 사전 점수 계산 (최고 달성 점수가 낮을수록 배정처가 마땅찮은=까다로운 이미지)
|
||||
prelim_best_score = [
|
||||
_best_slot_for_image(image, remaining_slots)[1] for image in pool
|
||||
]
|
||||
ordered_indices = sorted(
|
||||
range(len(pool)),
|
||||
key=lambda i: prelim_best_score[i]
|
||||
)
|
||||
|
||||
for idx in ordered_indices:
|
||||
if not remaining_slots:
|
||||
break
|
||||
image = pool[idx]
|
||||
# 슬롯이 이전 배정으로 줄어들었으므로 현재 remaining_slots 기준 재계산
|
||||
best_slot, _ = _best_slot_for_image(image, remaining_slots)
|
||||
if best_slot is None:
|
||||
continue
|
||||
modifications[best_slot] = image["image_url"]
|
||||
assigned[best_slot] = image["image_tag"]
|
||||
remaining_slots.remove(best_slot)
|
||||
|
||||
# Step 2: 커버리지 배정 후 남은 슬롯은 pool 전체에서 최고점 이미지 배정 (재사용 불가피)
|
||||
for slot in remaining_slots:
|
||||
scores = self.calculate_image_slot_score_multi(pool, slot)
|
||||
if not scores:
|
||||
continue
|
||||
best_idx = scores.index(max(scores))
|
||||
selected = pool[best_idx]
|
||||
modifications[slot] = selected["image_url"]
|
||||
assigned[slot] = selected["image_tag"]
|
||||
else:
|
||||
# 이미지 충분(슬롯 수 <= 이미지 수): 상위 3장 중 점수 비례 확률로 선택해 영상 다양화
|
||||
# 동일 요청을 반복할 때 매번 같은 영상이 생성되는 문제 방지
|
||||
sorted_indices = sorted(range(len(image_score_list)), key=lambda i: image_score_list[i], reverse=True)
|
||||
# 이미지 충분(슬롯 수 <= 이미지 수): 난이도 우선 greedy 배정
|
||||
# Step 1: 정렬용 사전 점수 계산 (최고 달성 점수가 낮을수록 배정이 어려운 슬롯)
|
||||
prelim = {slot: self.calculate_image_slot_score_multi(pool, slot) for slot in image_slots}
|
||||
ordered_slots = sorted(
|
||||
image_slots,
|
||||
key=lambda s: max(prelim[s]) if prelim[s] else 0.0
|
||||
# 오름차순 → 달성 최대값이 낮은(어려운) 슬롯이 앞으로
|
||||
)
|
||||
|
||||
# Step 2: 난이도 순서로 배정, 배정마다 pool에서 pop + 다음 슬롯은 점수 재계산
|
||||
for slot in ordered_slots:
|
||||
if not pool:
|
||||
logger.warning(f"[template_matching_taged_image] 이미지 풀 소진 — 남은 슬롯 배정 불가: {slot}")
|
||||
break
|
||||
# pop 후 인덱스 변동이 있으므로 현재 pool로 재계산 (사전 점수 재사용 금지)
|
||||
scores = self.calculate_image_slot_score_multi(pool, slot)
|
||||
if not scores:
|
||||
continue
|
||||
|
||||
# 상위 3장 중 점수 비례 확률로 선택 (영상 다양화)
|
||||
sorted_indices = sorted(range(len(scores)), key=lambda i: scores[i], reverse=True)
|
||||
top_indices = sorted_indices[:min(3, len(sorted_indices))]
|
||||
weights = [image_score_list[i] for i in top_indices]
|
||||
weights = [scores[i] for i in top_indices]
|
||||
if sum(weights) > 0:
|
||||
maximum_idx = random.choices(top_indices, weights=weights, k=1)[0]
|
||||
chosen_pool_idx = random.choices(top_indices, weights=weights, k=1)[0]
|
||||
else:
|
||||
maximum_idx = random.choice(top_indices)
|
||||
selected = taged_image_list.pop(maximum_idx)
|
||||
image_name = selected["image_url"]
|
||||
modifications[template_component_name] =image_name
|
||||
pass
|
||||
case "text":
|
||||
if "address_input" in template_component_name:
|
||||
modifications[template_component_name] = address
|
||||
chosen_pool_idx = random.choice(top_indices)
|
||||
|
||||
selected = pool.pop(chosen_pool_idx)
|
||||
modifications[slot] = selected["image_url"]
|
||||
assigned[slot] = selected["image_tag"]
|
||||
|
||||
# 텍스트 슬롯 처리
|
||||
for name, _ in text_slots:
|
||||
if "address_input" in name:
|
||||
modifications[name] = address
|
||||
|
||||
modifications["audio-music"] = music_url
|
||||
return modifications
|
||||
return modifications, assigned
|
||||
|
||||
# 점수 계산 상수 (모듈 레벨로 빼면 더 좋지만 기존 코드 스타일 따라 클래스 내부에 위치)
|
||||
_NARR_FLOOR = 0.25 # narrative 최솟값 modulator: base 점수가 완전히 0이 되는 것을 방지
|
||||
_NARR_BONUS = 1.5 # narrative 독립 보너스: base=0이어도 narrative 신호가 점수에 남음
|
||||
_NARR_MIN = 0.0 # narrative_preference 값 clamp 하한 (무경계 스키마 방어)
|
||||
_NARR_MAX = 100.0 # narrative_preference 값 clamp 상한. NarrativePreference 스키마(0.0~100.0)와 스케일 일치
|
||||
|
||||
def calculate_image_slot_score_multi(self, taged_image_list : list[dict], slot_name : str):
|
||||
"""이미지 슬롯에 대한 각 이미지의 매칭 점수를 계산합니다.
|
||||
|
||||
점수 = base * narr_mod + NARR_BONUS * narr
|
||||
- base: 태그 카테고리 매칭 합산 (최대 5.5)
|
||||
- narr_mod: [NARR_FLOOR, 1.0] 구간으로 완화된 narrative 계수
|
||||
- NARR_BONUS * narr: narrative 독립 보너스 (base=0이어도 신호 유지)
|
||||
이 구조로 "base=0, narrative가 높음"과 "완전히 틀린 이미지(0,0)"를 구별할 수 있음.
|
||||
"""
|
||||
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)
|
||||
if slot_tag_dict is None:
|
||||
# parse 실패 시 전부 0점 반환 (슬롯 skip)
|
||||
return [0.0] * len(image_tag_list)
|
||||
|
||||
base_score_list = [0.0] * len(image_tag_list)
|
||||
# slot_tag_narrative = NarrativePhase.accent # 기본값 (narrative 토큰 없는 슬롯 대비)
|
||||
|
||||
for slot_tag_cate, slot_tag_item in slot_tag_dict.items():
|
||||
if slot_tag_cate == "narrative_preference":
|
||||
|
|
@ -508,28 +615,45 @@ class CreatomateService:
|
|||
|
||||
match slot_tag_cate:
|
||||
case "space_type":
|
||||
weight = 2
|
||||
weight = 2.0
|
||||
case "subject":
|
||||
weight = 2
|
||||
weight = 2.0
|
||||
case "camera":
|
||||
weight = 1
|
||||
weight = 1.0
|
||||
case "motion_recommended":
|
||||
weight = 0.5
|
||||
case _:
|
||||
raise
|
||||
raise ValueError(f"[calculate_image_slot_score_multi] unknown slot tag category: {slot_tag_cate}")
|
||||
|
||||
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
|
||||
base_score_list[idx] += weight
|
||||
|
||||
# narrative 점수 적용: 순수 곱셈 대신 "완화 modulator + 독립 보너스"
|
||||
# - narr_mod: base를 흐릴 수 있지만 NARR_FLOOR 이하로는 안 떨어짐
|
||||
# - NARR_BONUS * narr: base=0이어도 narrative가 높으면 양수 점수 확보
|
||||
image_score_list = []
|
||||
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
|
||||
raw_narr = image_tag.get("narrative_preference", {}).get(slot_tag_narrative, 0.0)
|
||||
clamped_narr = min(self._NARR_MAX, max(self._NARR_MIN, float(raw_narr)))
|
||||
narr = clamped_narr / self._NARR_MAX # 0.0~1.0로 정규화
|
||||
narr_mod = self._NARR_FLOOR + (1.0 - self._NARR_FLOOR) * narr
|
||||
score = base_score_list[idx] * narr_mod + self._NARR_BONUS * narr
|
||||
image_score_list.append(score)
|
||||
|
||||
return image_score_list
|
||||
|
||||
def parse_slot_name_to_tag(self, slot_name : str) -> dict[str, StrEnum]:
|
||||
def parse_slot_name_to_tag(self, slot_name: str) -> dict[str, StrEnum] | None:
|
||||
"""슬롯 이름을 파싱하여 태그 딕셔너리를 반환합니다.
|
||||
|
||||
슬롯 이름 형식: {space_type}-{subject}-{camera}-{motion}-{narrative}
|
||||
파싱 실패 시 None을 반환합니다 (호출자가 해당 슬롯을 skip+log 처리).
|
||||
"""
|
||||
try:
|
||||
tag_list = slot_name.split("-")
|
||||
if len(tag_list) < 5:
|
||||
logger.warning(f"[parse_slot_name_to_tag] 슬롯명 토큰 부족 ({len(tag_list)}개): '{slot_name}' — 슬롯 skip")
|
||||
return None
|
||||
space_type = SpaceType(tag_list[0])
|
||||
subject = Subject(tag_list[1])
|
||||
camera = Camera(tag_list[2])
|
||||
|
|
@ -543,6 +667,9 @@ class CreatomateService:
|
|||
"narrative_preference": narrative,
|
||||
}
|
||||
return tag_dict
|
||||
except (ValueError, IndexError) as e:
|
||||
logger.warning(f"[parse_slot_name_to_tag] 슬롯명 파싱 실패: '{slot_name}' — {e} — 슬롯 skip")
|
||||
return None
|
||||
|
||||
def elements_connect_resource_blackbox(
|
||||
self,
|
||||
|
|
@ -788,8 +915,7 @@ class CreatomateService:
|
|||
for animation in elem["animations"]:
|
||||
if "reversed" in animation:
|
||||
continue
|
||||
assert animation.get("time",0) == 0 # 0이 아닌 경우 확인 필요
|
||||
if "transition" in animation and animation["transition"]:
|
||||
if "transition" in animation and animation["transition"] and animation.get("time", 0) == 0:
|
||||
track_maximum_duration[elem["track"]] -= animation["duration"]
|
||||
else:
|
||||
track_maximum_duration[elem["track"]] = max(track_maximum_duration[elem["track"]], elem["time"] + elem["duration"])
|
||||
|
|
@ -825,7 +951,8 @@ class CreatomateService:
|
|||
if "animations" not in elem:
|
||||
continue
|
||||
for animation in elem["animations"]:
|
||||
assert animation["time"] == 0 # 0이 아닌 경우 확인 필요
|
||||
if animation.get("time", 0) != 0:
|
||||
animation["time"] = animation["time"] * extend_rate
|
||||
animation["duration"] = animation["duration"] * extend_rate
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
|
|
@ -890,7 +1017,7 @@ class CreatomateService:
|
|||
|
||||
len_keywords = len(target_keywords) if len(target_keywords) < 3 else 3
|
||||
|
||||
hashtaged_target_keywords = [f"#{tk}" for tk in target_keywords[len_keywords]]
|
||||
hashtaged_target_keywords = [f"#{tk}" for tk in target_keywords[:len_keywords]]
|
||||
|
||||
mod_dict = {
|
||||
"thumb-hashtag-primary" : ' '.join(hashtaged_target_keywords),
|
||||
|
|
|
|||
|
|
@ -0,0 +1,125 @@
|
|||
"""크롤링 단계 이미지 마케팅 적합성 필터.
|
||||
|
||||
마케팅 적합성(marketing_acceptable) 판정은 이 모듈이 전담한다. 크롤링 시점에
|
||||
방문자/AI View 사진(extra_photo_urls)만을 대상으로 판정하며, 업로드 이후 단계의
|
||||
태깅(app/utils/autotag.py의 autotag_images(), sheet 기반 image_autotag_prompt)은
|
||||
공간/피사체/카메라/모션/내러티브 태그만 다루고 marketing_acceptable은 포함하지 않는다
|
||||
(이미 이 단계에서 걸러진 이미지만 태깅 대상이 되기 때문). 프롬프트가 짧아
|
||||
시트(Prompt) 대신 코드에 하드코딩한다.
|
||||
|
||||
업체 등록 사진(owner_images)은 이 필터의 대상이 아니며, 호출측(라우터)에서
|
||||
owner_images 개수가 NvMapScraper.SUPPLEMENT_THRESHOLD(30) 이상이면 이 모듈을
|
||||
아예 호출하지 않는 것이 원칙이다.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
|
||||
from app.utils.logger import get_logger
|
||||
from app.utils.prompts.chatgpt_prompt import ChatgptService
|
||||
from app.utils.prompts.schemas import MarketingFilterOutput
|
||||
|
||||
logger = get_logger("image_filter")
|
||||
|
||||
# 크롤링 단계 필터링에 사용할 Gemini 모델. 시트(prompts.py)와 달리 코드에 고정한다.
|
||||
MARKETING_FILTER_MODEL = "gemini-3.5-flash"
|
||||
|
||||
MAX_RETRY = 2
|
||||
|
||||
MARKETING_FILTER_PROMPT_TEMPLATE = """\
|
||||
당신은 광고 영상 제작을 위한 이미지 심사자입니다. 아래 업종의 광고 영상에 이 이미지를 \
|
||||
사용해도 되는지 판정하세요.
|
||||
|
||||
업종(industry): {industry}
|
||||
|
||||
## MARKETING SUITABILITY CHECK (industry-aware)
|
||||
- marketing_acceptable: true | false
|
||||
- reject_reason: short reason if false (e.g., 저화질, 인물 신원 노출, 업종 부적합, 민감/의료 장면, 미성년자 식별 노출). Empty if acceptable.
|
||||
- Person identity protection: 식별 가능한 얼굴이 주피사체이고 동의 여부가 불확실하면 보수적으로 처리.
|
||||
|
||||
주어진 스키마에 맞춰 marketing_acceptable과 reject_reason만 반환하세요.\
|
||||
"""
|
||||
|
||||
|
||||
def _build_prompt(industry: str) -> str:
|
||||
return MARKETING_FILTER_PROMPT_TEMPLATE.format(industry=industry or "미상")
|
||||
|
||||
|
||||
async def filter_marketing_images(image_url_list: list[str], industry: str = "") -> list[bool]:
|
||||
"""이미지 URL마다 마케팅 적합성(marketing_acceptable)만 판정해 통과 여부 리스트를 반환한다.
|
||||
|
||||
이미지 1장당 API 호출 1회를 asyncio.gather로 병렬 실행한다(배치 호출 아님).
|
||||
실패한 이미지는 최대 MAX_RETRY회 재시도하며, 그래도 실패하면 보수적으로
|
||||
False(제외)로 처리한다.
|
||||
|
||||
Args:
|
||||
image_url_list: 판정할 이미지 URL 목록 (extra_photo_urls의 original만 전달할 것)
|
||||
industry: 업종 분류값 (_resolve_industry 결과). 필터 프롬프트 앞에서 먼저 계산되어야 함.
|
||||
|
||||
Returns:
|
||||
image_url_list와 같은 순서의 통과 여부(bool) 리스트.
|
||||
"""
|
||||
if not image_url_list:
|
||||
return []
|
||||
|
||||
chatgpt = ChatgptService(model_type="gemini")
|
||||
prompt_text = _build_prompt(industry)
|
||||
|
||||
async def _call(url: str):
|
||||
return await chatgpt._call_pydantic_output_chat_completion(
|
||||
prompt=prompt_text,
|
||||
output_format=MarketingFilterOutput,
|
||||
model=MARKETING_FILTER_MODEL,
|
||||
img_url=url,
|
||||
image_detail_high=False,
|
||||
)
|
||||
|
||||
results: list[MarketingFilterOutput | BaseException] = await asyncio.gather(
|
||||
*[_call(url) for url in image_url_list], return_exceptions=True
|
||||
)
|
||||
|
||||
for _ in range(MAX_RETRY):
|
||||
failed_idx = [i for i, r in enumerate(results) if isinstance(r, Exception)]
|
||||
if not failed_idx:
|
||||
break
|
||||
logger.warning(f"[image_filter] 재시도 대상 인덱스: {failed_idx}")
|
||||
retried = await asyncio.gather(
|
||||
*[_call(image_url_list[i]) for i in failed_idx], return_exceptions=True
|
||||
)
|
||||
for i, result in zip(failed_idx, retried):
|
||||
results[i] = result
|
||||
|
||||
final_failed = [i for i, r in enumerate(results) if isinstance(r, Exception)]
|
||||
if final_failed:
|
||||
logger.warning(
|
||||
f"[image_filter] {len(final_failed)}건 최종 실패 → 보수적으로 제외 처리: {final_failed}"
|
||||
)
|
||||
|
||||
return [
|
||||
(not isinstance(r, Exception)) and r.marketing_acceptable
|
||||
for r in results
|
||||
]
|
||||
|
||||
|
||||
def assemble_images(
|
||||
owner_images: list[dict],
|
||||
extra_images: list[dict],
|
||||
extra_pass_flags: list[bool],
|
||||
max_images: int,
|
||||
) -> list[dict]:
|
||||
"""owner_images 전량 + 필터 통과한 extra_images를 합쳐 최대 max_images장을 만든다.
|
||||
|
||||
owner_images는 무조건 포함(필터링 대상 아님). extra_images는 extra_pass_flags와
|
||||
같은 순서로 매칭되며, 통과분만 순서대로 이어붙인다. original URL 기준으로 중복을
|
||||
제거한다(nvMapScraper._extract_origins의 dedup 방식과 동일).
|
||||
"""
|
||||
passed_extra = [img for img, ok in zip(extra_images, extra_pass_flags) if ok]
|
||||
|
||||
seen: set[str] = set()
|
||||
combined: list[dict] = []
|
||||
for img in owner_images + passed_extra:
|
||||
original = img.get("original")
|
||||
if original and original not in seen:
|
||||
seen.add(original)
|
||||
combined.append(img)
|
||||
|
||||
return combined[:max_images]
|
||||
|
|
@ -52,14 +52,12 @@ class NvMapPwScraper():
|
|||
cls.is_ready = True
|
||||
|
||||
GRAPHQL_URL = "https://pcmap-api.place.naver.com/graphql"
|
||||
_PLACE_TYPES = ("restaurant", "accommodation")
|
||||
|
||||
@classmethod
|
||||
async def fetch_graphql(
|
||||
cls,
|
||||
place_id: str,
|
||||
payloads: list[dict],
|
||||
place_type: str | None = None,
|
||||
) -> list[dict | None] | None:
|
||||
"""실제 브라우저로 네이버 WTM 안티봇 캡차를 통과해 GraphQL 쿼리를 실행한다.
|
||||
|
||||
|
|
@ -73,8 +71,6 @@ class NvMapPwScraper():
|
|||
Args:
|
||||
place_id: 네이버 place ID
|
||||
payloads: GraphQL POST 본문 목록
|
||||
place_type: "restaurant" | "accommodation" (없으면 둘 다 시도)
|
||||
|
||||
Returns:
|
||||
payload별 파싱 JSON 목록(실패 항목은 None). 토큰 캡처 실패 시 None.
|
||||
"""
|
||||
|
|
@ -95,8 +91,7 @@ class NvMapPwScraper():
|
|||
page.on("request", on_request)
|
||||
|
||||
try:
|
||||
types = [place_type] if place_type in cls._PLACE_TYPES else list(cls._PLACE_TYPES)
|
||||
for t in types:
|
||||
for t in ("place", "restaurant", "accommodation"):
|
||||
try:
|
||||
await page.goto(
|
||||
f"https://pcmap.place.naver.com/{t}/{place_id}/home",
|
||||
|
|
|
|||
|
|
@ -35,6 +35,8 @@ class NvMapScraper:
|
|||
GRAPHQL_URL: str = "https://pcmap-api.place.naver.com/graphql"
|
||||
REQUEST_TIMEOUT = 120 # 초
|
||||
data_source_identifier = "nv"
|
||||
SUPPLEMENT_THRESHOLD = 30 # 업체 사진이 이 수 미만일 때 방문자 사진으로 보충
|
||||
MAX_IMAGES = 50 # 업체+방문자 사진 합산 최대 장수
|
||||
OVERVIEW_QUERY: str = """
|
||||
query getAccommodation($id: String!, $deviceType: String) {
|
||||
business: placeDetail(input: {id: $id, isNx: true, deviceType: $deviceType}) {
|
||||
|
|
@ -51,7 +53,15 @@ query getAccommodation($id: String!, $deviceType: String) {
|
|||
visitorReviewsTotal
|
||||
}
|
||||
images { images { origin url } }
|
||||
cpImages(source: [ugcImage]) { images { origin url } }
|
||||
}
|
||||
}"""
|
||||
|
||||
PHOTO_VIEWER_QUERY: str = """
|
||||
query getPhotoViewerItems($input: PhotoViewerInput) {
|
||||
photoViewer(input: $input) {
|
||||
photos {
|
||||
originalUrl
|
||||
}
|
||||
}
|
||||
}"""
|
||||
|
||||
|
|
@ -84,10 +94,11 @@ query getVisitorReviewStats($id: String!) {
|
|||
)
|
||||
self.scrap_type: str | None = None
|
||||
self.rawdata: dict | None = None
|
||||
self.image_link_list: list[str] | None = None
|
||||
self.image_link_list: list[dict] | None = None # [{"preview": str, "original": str}]
|
||||
self.owner_images: list[dict] | None = None # 업체 등록 사진 (마케팅 필터 제외 대상)
|
||||
self.extra_photo_urls: list[dict] | None = None # 방문자/AI View 보충 사진 (마케팅 필터 대상)
|
||||
self.base_info: dict | None = None
|
||||
self.facility_info: str | None = None
|
||||
self.place_type: str | None = None # 크롤링으로 감지된 업종 (accommodation | restaurant)
|
||||
self.voted_keyword_stats: list[dict] | None = None # 키워드 투표 집계 (displayName, count)
|
||||
|
||||
def _get_request_headers(self) -> dict:
|
||||
|
|
@ -96,6 +107,62 @@ query getVisitorReviewStats($id: String!) {
|
|||
headers["Cookie"] = self.cookies
|
||||
return headers
|
||||
|
||||
_GIF_PATTERN = re.compile(r"\.gif(?:/|$)", re.IGNORECASE)
|
||||
|
||||
@classmethod
|
||||
def _is_gif_url(cls, url: str) -> bool:
|
||||
"""URL의 확장자가 gif인지 확인한다.
|
||||
|
||||
네이버 CDN은 리사이즈 파라미터를 확장자 뒤에 경로 세그먼트로 붙인다
|
||||
(예: ".../3982000924.gif/500x500") — endswith(".gif")로는 잡히지 않으므로
|
||||
경로 세그먼트 단위로 ".gif" 뒤에 "/" 또는 문자열 끝이 오는지 검사한다.
|
||||
"""
|
||||
return bool(cls._GIF_PATTERN.search(url))
|
||||
|
||||
@staticmethod
|
||||
def _extract_origins(image_node: dict) -> list[dict]:
|
||||
"""GraphQL 이미지 노드({ images: [{origin, url}, ...] })에서 {preview, original} 목록을 추출한다.
|
||||
origin = 원본(업로드용), url = CDN 리사이즈(미리보기용)
|
||||
"""
|
||||
items = image_node.get("images") or []
|
||||
return [
|
||||
{"preview": item.get("url") or item["origin"], "original": item["origin"]}
|
||||
for item in items
|
||||
if item.get("origin") and not NvMapScraper._is_gif_url(item["origin"])
|
||||
]
|
||||
|
||||
@staticmethod
|
||||
def _extract_photo_viewer_urls(photo_viewer_raw: dict | None) -> list[dict]:
|
||||
"""getPhotoViewerItems 응답에서 {preview, original} 목록을 추출한다.
|
||||
photoViewer는 썸네일 필드가 없으므로 preview = original 동일하게 사용.
|
||||
"""
|
||||
photos = ((photo_viewer_raw or {}).get("data") or {}).get("photoViewer") or {}
|
||||
items = photos.get("photos") or []
|
||||
return [
|
||||
{"preview": p["originalUrl"], "original": p["originalUrl"]}
|
||||
for p in items
|
||||
if p.get("originalUrl") and not NvMapScraper._is_gif_url(p["originalUrl"])
|
||||
]
|
||||
|
||||
@staticmethod
|
||||
def _dedup_by_original(images: list[dict]) -> list[dict]:
|
||||
"""{"preview","original"} dict 리스트를 original URL 기준으로 순서를 유지한 채 중복 제거한다."""
|
||||
seen: set[str] = set()
|
||||
result = []
|
||||
for img in images:
|
||||
original = img.get("original")
|
||||
if original and original not in seen:
|
||||
seen.add(original)
|
||||
result.append(img)
|
||||
return result
|
||||
|
||||
@staticmethod
|
||||
def _interleave(a: list, b: list) -> list:
|
||||
"""두 리스트를 1:1 교차 병합한다. 남은 항목은 뒤에 붙인다."""
|
||||
result = [x for pair in zip(a, b) for x in pair]
|
||||
longer = a[len(b):] if len(a) > len(b) else b[len(a):]
|
||||
return result + longer
|
||||
|
||||
async def parse_url(self) -> str:
|
||||
"""URL에서 place ID를 추출합니다. 단축 URL인 경우 실제 URL로 변환합니다."""
|
||||
place_pattern = r"/place/(\d+)"
|
||||
|
|
@ -114,9 +181,6 @@ query getVisitorReviewStats($id: String!) {
|
|||
# place.naver.com/{type}/{id} 형식 (예: m.place.naver.com/restaurant/11667161)
|
||||
type_match = re.search(r"place\.naver\.com/([a-zA-Z]+)/(\d+)", self.url)
|
||||
if type_match:
|
||||
url_place_type = type_match.group(1)
|
||||
if url_place_type in ("accommodation", "restaurant"):
|
||||
self.place_type = url_place_type
|
||||
return type_match.group(2)
|
||||
if not match:
|
||||
raise URLNotFoundException("Failed to parse place ID from URL")
|
||||
|
|
@ -144,7 +208,7 @@ query getVisitorReviewStats($id: String!) {
|
|||
# self.scrap_type = "GraphQL-Browser"
|
||||
|
||||
# ── 실제 브라우저로 WTM 캡차 우회 ──
|
||||
data, stats_data = await self._scrap_via_browser(place_id)
|
||||
data, stats_data, extra_photo_urls = await self._scrap_via_browser(place_id)
|
||||
fac_data = None # 편의시설(HTML)은 브라우저 경로에서 생략 (best-effort)
|
||||
self.scrap_type = "GraphQL-Browser"
|
||||
|
||||
|
|
@ -152,21 +216,41 @@ query getVisitorReviewStats($id: String!) {
|
|||
# Naver 기준임, 구글 등 다른 데이터 소스의 경우 고유 Identifier 사용할 것.
|
||||
self.place_id = self.data_source_identifier + place_id
|
||||
self.rawdata["facilities"] = fac_data
|
||||
self.image_link_list = [
|
||||
nv_image["origin"]
|
||||
for nv_image in data["data"]["business"]["images"]["images"]
|
||||
]
|
||||
business = data["data"]["business"]
|
||||
|
||||
# 업체 등록 이미지 (마케팅 적합성 필터 제외 대상 — 자체 dedup)
|
||||
self.owner_images = self._dedup_by_original(self._extract_origins(business.get("images") or {}))
|
||||
|
||||
# 방문자/AI View 보충 사진 (마케팅 적합성 필터 대상). owner에 이미 있는 원본은 제외.
|
||||
owner_originals = {img["original"] for img in self.owner_images}
|
||||
self.extra_photo_urls = self._dedup_by_original(
|
||||
[img for img in extra_photo_urls if img["original"] not in owner_originals]
|
||||
)
|
||||
|
||||
# 업체 사진이 임계값 미만이면 내부/외부/리뷰 사진으로 보충
|
||||
# (필터링 없는 기본 조립. 마케팅 적합성 필터를 적용하려면 호출측에서
|
||||
# owner_images / extra_photo_urls를 직접 사용해 image_filter.assemble_images로 재조립할 것.)
|
||||
if len(self.owner_images) < self.SUPPLEMENT_THRESHOLD:
|
||||
combined = self._dedup_by_original(self.owner_images + self.extra_photo_urls)
|
||||
logger.info(
|
||||
f"[NvMapScraper] 업체 사진 {len(self.owner_images)}장 < {self.SUPPLEMENT_THRESHOLD}장 "
|
||||
f"→ 보충 사진 {len(self.extra_photo_urls)}장 추가 (합산 {len(combined)}장, 상한 {self.MAX_IMAGES}장)"
|
||||
)
|
||||
else:
|
||||
combined = self.owner_images
|
||||
|
||||
self.image_link_list = combined[: self.MAX_IMAGES]
|
||||
self.base_info = data["data"]["business"]["base"]
|
||||
self.facility_info = fac_data
|
||||
self.voted_keyword_stats = stats_data
|
||||
|
||||
return
|
||||
|
||||
async def _scrap_via_browser(self, place_id: str) -> tuple[dict, list[dict] | None]:
|
||||
async def _scrap_via_browser(self, place_id: str) -> tuple[dict, list[dict] | None, list[str]]:
|
||||
"""직접 호출이 WTM 캡차에 막힌 경우, 실제 브라우저로 GraphQL을 호출한다.
|
||||
|
||||
Returns:
|
||||
(overview_data, review_stats_details)
|
||||
(overview_data, review_stats_details, extra_photo_urls)
|
||||
|
||||
Raises:
|
||||
GraphQLException: 브라우저 폴백마저 실패한 경우
|
||||
|
|
@ -183,32 +267,97 @@ query getVisitorReviewStats($id: String!) {
|
|||
"variables": {"id": place_id},
|
||||
"query": self.REVIEW_STATS_QUERY,
|
||||
}
|
||||
interior_payload = {
|
||||
"operationName": "getPhotoViewerItems",
|
||||
"variables": {
|
||||
"input": {
|
||||
"businessId": place_id,
|
||||
"cursors": [{"id": "aiView"}],
|
||||
"filter": "AI View",
|
||||
"subFilter": "INTERIOR",
|
||||
"dateRange": "",
|
||||
"excludeAuthorIds": [],
|
||||
"excludeClipIds": [],
|
||||
"excludeSection": [],
|
||||
}
|
||||
},
|
||||
"query": self.PHOTO_VIEWER_QUERY,
|
||||
}
|
||||
exterior_payload = {
|
||||
"operationName": "getPhotoViewerItems",
|
||||
"variables": {
|
||||
"input": {
|
||||
"businessId": place_id,
|
||||
"cursors": [{"id": "aiView"}],
|
||||
"filter": "AI View",
|
||||
"subFilter": "EXTERIOR",
|
||||
"dateRange": "",
|
||||
"excludeAuthorIds": [],
|
||||
"excludeClipIds": [],
|
||||
"excludeSection": [],
|
||||
}
|
||||
},
|
||||
"query": self.PHOTO_VIEWER_QUERY,
|
||||
}
|
||||
review_payload = {
|
||||
"operationName": "getPhotoViewerItems",
|
||||
"variables": {
|
||||
"input": {
|
||||
"businessId": place_id,
|
||||
"cursors": [{"id": "placeReview"}],
|
||||
"dateRange": "",
|
||||
"excludeAuthorIds": [],
|
||||
"excludeClipIds": [],
|
||||
"excludeSection": [],
|
||||
}
|
||||
},
|
||||
"query": self.PHOTO_VIEWER_QUERY,
|
||||
}
|
||||
|
||||
payloads = [overview_payload, stats_payload, interior_payload, exterior_payload, review_payload]
|
||||
MAX_RETRY = 3
|
||||
results = None
|
||||
last_error: Exception | None = None
|
||||
for attempt in range(1, MAX_RETRY + 1):
|
||||
try:
|
||||
results = await NvMapPwScraper.fetch_graphql(
|
||||
place_id, [overview_payload, stats_payload], self.place_type
|
||||
)
|
||||
results = await NvMapPwScraper.fetch_graphql(place_id, payloads)
|
||||
except Exception as e:
|
||||
logger.error(f"[NvMapScraper] 브라우저 폴백 오류: {e}")
|
||||
raise GraphQLException(f"브라우저 폴백 실패: {e}")
|
||||
last_error = e
|
||||
logger.warning(f"[NvMapScraper] 브라우저 폴백 시도 {attempt}/{MAX_RETRY} 오류: {e}")
|
||||
if attempt < MAX_RETRY:
|
||||
await asyncio.sleep(1)
|
||||
continue
|
||||
if results and results[0] is not None and "data" in results[0]:
|
||||
break
|
||||
logger.warning(f"[NvMapScraper] 브라우저 폴백 시도 {attempt}/{MAX_RETRY} 실패(405 등), 재시도")
|
||||
results = None
|
||||
if attempt < MAX_RETRY:
|
||||
await asyncio.sleep(1)
|
||||
|
||||
if not results or results[0] is None or "data" not in results[0]:
|
||||
if last_error:
|
||||
raise GraphQLException(f"브라우저 폴백 실패: {last_error}")
|
||||
raise GraphQLException("브라우저 폴백 크롤링 실패 (WTM 토큰 또는 응답 없음)")
|
||||
|
||||
data = results[0]
|
||||
stats_data = None
|
||||
stats_raw = results[1] if len(results) > 1 else None
|
||||
if stats_raw:
|
||||
stats_data = (
|
||||
stats_raw.get("data", {})
|
||||
.get("visitorReviewStats", {})
|
||||
.get("analysis", {})
|
||||
.get("votedKeyword", {})
|
||||
.get("details")
|
||||
) or None
|
||||
_vrs = (stats_raw.get("data") or {}).get("visitorReviewStats") or {}
|
||||
_analysis = _vrs.get("analysis") or {}
|
||||
_voted = _analysis.get("votedKeyword") or {}
|
||||
stats_data = _voted.get("details") or None
|
||||
|
||||
interior_urls = self._extract_photo_viewer_urls(results[2] if len(results) > 2 else None)
|
||||
exterior_urls = self._extract_photo_viewer_urls(results[3] if len(results) > 3 else None)
|
||||
review_urls = self._extract_photo_viewer_urls(results[4] if len(results) > 4 else None)
|
||||
extra_photo_urls = self._interleave(interior_urls, exterior_urls) + review_urls
|
||||
logger.info(
|
||||
f"[NvMapScraper] 보충 이미지 - 내부:{len(interior_urls)} 외부:{len(exterior_urls)} 리뷰:{len(review_urls)}"
|
||||
)
|
||||
|
||||
logger.info(f"[NvMapScraper] 브라우저 폴백 SUCCESS - place_id: {place_id}")
|
||||
return data, stats_data
|
||||
return data, stats_data, extra_photo_urls
|
||||
|
||||
async def _call_get_accommodation(self, place_id: str) -> dict:
|
||||
"""GraphQL API를 호출하여 숙소 정보를 가져옵니다.
|
||||
|
|
@ -281,14 +430,10 @@ query getVisitorReviewStats($id: String!) {
|
|||
logger.warning(f"[NvMapScraper] review stats failed: {response.status}")
|
||||
return None
|
||||
result = await response.json()
|
||||
details = (
|
||||
result.get("data", {})
|
||||
.get("visitorReviewStats", {})
|
||||
.get("analysis", {})
|
||||
.get("votedKeyword", {})
|
||||
.get("details")
|
||||
)
|
||||
return details or None
|
||||
_vrs = (result.get("data") or {}).get("visitorReviewStats") or {}
|
||||
_analysis = _vrs.get("analysis") or {}
|
||||
_voted = _analysis.get("votedKeyword") or {}
|
||||
return _voted.get("details") or None
|
||||
except Exception as e:
|
||||
logger.warning(f"[NvMapScraper] Failed to get review stats: {e}")
|
||||
return None
|
||||
|
|
@ -302,7 +447,7 @@ query getVisitorReviewStats($id: String!) {
|
|||
Returns:
|
||||
편의시설 정보 문자열 또는 None
|
||||
"""
|
||||
place_types = ["accommodation", "restaurant"]
|
||||
place_types = ["place", "accommodation", "restaurant"]
|
||||
try:
|
||||
async with aiohttp.ClientSession() as session:
|
||||
for place_type in place_types:
|
||||
|
|
|
|||
|
|
@ -176,7 +176,7 @@ class ChatgptService:
|
|||
input_data : dict,
|
||||
img_url : Optional[str] = None,
|
||||
img_detail_high : bool = False,
|
||||
silent : bool = False
|
||||
silent : bool = True
|
||||
) -> BaseModel:
|
||||
prompt_text = prompt.build_prompt(input_data, silent)
|
||||
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ from typing import List, Optional
|
|||
from enum import StrEnum, auto
|
||||
|
||||
class SpaceType(StrEnum):
|
||||
"""이미지가 촬영된 공간·장소 유형. LLM이 이미지를 보고 해당하는 공간 태그를 선택하는 데 사용."""
|
||||
exterior_front = auto()
|
||||
exterior_night = auto()
|
||||
exterior_aerial = auto()
|
||||
|
|
@ -43,6 +44,7 @@ class SpaceType(StrEnum):
|
|||
detail_tableware = auto()
|
||||
|
||||
class Subject(StrEnum):
|
||||
"""이미지 내 주요 피사체 유형. 화면에 무엇이 담겨 있는지를 분류하는 태그."""
|
||||
empty_space = auto()
|
||||
exterior_building = auto()
|
||||
architecture_detail = auto()
|
||||
|
|
@ -55,6 +57,7 @@ class Subject(StrEnum):
|
|||
person = auto()
|
||||
|
||||
class Camera(StrEnum):
|
||||
"""이미지의 촬영 기법·구도·조명 특성. 영상 편집 시 컷의 시각적 스타일을 구분하는 태그."""
|
||||
wide_angle = auto()
|
||||
tight_crop = auto()
|
||||
panoramic = auto()
|
||||
|
|
@ -68,6 +71,7 @@ class Camera(StrEnum):
|
|||
has_face = auto()
|
||||
|
||||
class MotionRecommended(StrEnum):
|
||||
"""해당 이미지에 적용 가능한 카메라 모션 유형. 영상 제작 시 각 컷에 어울리는 움직임을 추천하는 데 사용."""
|
||||
static = auto()
|
||||
slow_pan = auto()
|
||||
slow_zoom_in = auto()
|
||||
|
|
@ -76,23 +80,41 @@ class MotionRecommended(StrEnum):
|
|||
dolly = auto()
|
||||
|
||||
class NarrativePhase(StrEnum):
|
||||
"""광고 영상 내러티브 단계. 이미지가 스토리 흐름에서 어느 위치에 배치되어야 하는지를 나타냄."""
|
||||
intro = auto()
|
||||
welcome = auto()
|
||||
core = auto()
|
||||
highlight = auto()
|
||||
support = auto()
|
||||
accent = auto()
|
||||
cta = auto()
|
||||
|
||||
NARRATIVE_PHASE_LABEL: dict[NarrativePhase, str] = {
|
||||
NarrativePhase.intro: "첫인상(intro)",
|
||||
NarrativePhase.welcome: "진입(welcome)",
|
||||
NarrativePhase.core: "핵심(core)",
|
||||
NarrativePhase.highlight: "차별화(highlight)",
|
||||
NarrativePhase.support: "보조(support)",
|
||||
NarrativePhase.accent: "감성(accent)",
|
||||
NarrativePhase.cta: "행동유도(cta)",
|
||||
}
|
||||
|
||||
class NarrativePreference(BaseModel):
|
||||
"""이미지가 각 내러티브 단계에 얼마나 적합한지를 나타내는 점수 모델 (0.0 ~ 100.0).
|
||||
LLM이 이미지를 분석한 뒤 각 단계별 적합도를 수치로 반환하며,
|
||||
영상 편집기가 이 점수를 기반으로 컷 배치 순서를 자동 결정하는 데 사용."""
|
||||
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="감성/마무리 — 어떤 느낌인가 | 공간의 분위기와 톤을 전달하는 감성 디테일 컷. 직접적 정보 전달보다 **느낌과 무드**를 제공. 영상 사이사이에 삽입되어 완성도를 높이는 역할.")
|
||||
cta: float = Field(..., description="행동 유도 — 지금 예약/방문하고 싶게 만드는 컷 | 가격·혜택·한정 오퍼 등 직접적 행동을 촉구하거나, 마지막 장면에서 강한 인상을 남기는 이미지.")
|
||||
|
||||
# Input 정의
|
||||
class ImageTagPromptInput(BaseModel):
|
||||
"""이미지 태깅 프롬프트에 전달되는 입력 스키마.
|
||||
분석할 이미지 URL과 업종 정보, LLM이 선택 가능한 태그 후보 리스트를 포함."""
|
||||
img_url : str = Field(..., description="이미지 URL")
|
||||
industry : str = Field(default="", description = "업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction). 통합 프롬프트가 이 값으로 내부 분기")
|
||||
space_type: list[str] = Field(list(SpaceType), description="공간적 정보를 가지는 태그 리스트")
|
||||
|
|
@ -100,9 +122,18 @@ class ImageTagPromptInput(BaseModel):
|
|||
camera: list[str] = Field(list(Camera), description="카메라 정보를 가지는 태그 리스트")
|
||||
motion_recommended: list[str] = Field(list(MotionRecommended), description="가능한 카메라 모션 리스트")
|
||||
|
||||
class MarketingFilterOutput(BaseModel):
|
||||
"""크롤링 단계에서 이미지의 마케팅(광고 영상) 사용 가능 여부만 판정하는 프롬프트의 출력 스키마.
|
||||
ImageTagPromptOutput과 달리 태깅(공간/피사체/카메라/모션/내러티브)은 포함하지 않는다."""
|
||||
marketing_acceptable: bool = Field(..., description="광고 영상 사용 가능 여부")
|
||||
reject_reason: str = Field(default="", description="거부 사유 (marketing_acceptable=false 시). 예: 저화질, 인물 신원 노출, 업종 부적합, 민감/의료 장면, 미성년자 식별 노출")
|
||||
|
||||
|
||||
# Output 정의
|
||||
class ImageTagPromptOutput(BaseModel):
|
||||
#ad_avaliable : bool = Field(..., description="광고 영상 사용 가능 이미지 여부")
|
||||
"""이미지 태깅 프롬프트의 LLM 출력 스키마.
|
||||
공간·피사체·카메라·모션 태그와 내러티브 적합도 점수를 담아 반환.
|
||||
마케팅 사용 가능 여부는 크롤링 단계(MarketingFilterOutput)에서 이미 판정되므로 포함하지 않는다."""
|
||||
space_type: list[SpaceType] = Field(..., description="공간적 정보를 가지는 태그 리스트")
|
||||
subject: list[Subject] = Field(..., description="피사체 정보를 가지는 태그 리스트")
|
||||
camera: list[Camera] = Field(..., description="카메라 정보를 가지는 태그 리스트")
|
||||
|
|
|
|||
|
|
@ -1,5 +1,7 @@
|
|||
from pydantic import BaseModel, Field
|
||||
from typing import List, Optional
|
||||
from typing import List, Literal, Optional
|
||||
|
||||
GENRE_VALUES = Literal["kpop", "pop", "ballad", "hip-hop", "rnb", "edm", "jazz", "rock"]
|
||||
|
||||
# Input 정의
|
||||
class LyricPromptInput(BaseModel):
|
||||
|
|
@ -8,10 +10,16 @@ class LyricPromptInput(BaseModel):
|
|||
detail_region_info : str = Field(..., description = "마케팅 대상 지역 상세")
|
||||
marketing_intelligence_summary : Optional[str] = Field(None, description = "마케팅 분석 정보 보고서")
|
||||
language : str= Field(..., description = "가사 언어")
|
||||
timing_rules : str = Field(..., description = "시간 제어문")
|
||||
genre : str = Field(default="", description = "지정된 음악 장르 (kpop|pop|ballad|hip-hop|rnb|edm|jazz|rock). 비어있으면 GPT가 무드에 맞는 장르를 자유롭게 추천")
|
||||
exclude_genre : str = Field(default="", description = "재생성 시 직전에 사용된 장르. genre가 비어있는 경우(자동 선택) 이 장르는 추천에서 제외")
|
||||
industry : str = Field(default="", description = "업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction). 통합 프롬프트가 이 값으로 내부 분기")
|
||||
|
||||
# Output 정의
|
||||
class LyricPromptOutput(BaseModel):
|
||||
lyric: str = Field(..., description="생성된 가사")
|
||||
suno_prompt: str = Field(..., description="Suno AI용 프롬프트")
|
||||
recommended_genre: GENRE_VALUES = Field(
|
||||
...,
|
||||
description="가사에 사용된(또는 가장 잘 맞는) 음악 장르. genre 입력이 주어졌다면 그 값을 그대로 반환하고, "
|
||||
"비어있었다면 exclude_genre를 제외한 나머지 중 가사 무드에 가장 잘 맞는 장르를 직접 선택해 반환",
|
||||
)
|
||||
|
|
@ -114,7 +114,7 @@ class SunoService:
|
|||
Args:
|
||||
prompt: 가사 (customMode=true일 때 가사로 사용)
|
||||
1분 이내 길이의 노래에 적합한 가사여야 함
|
||||
genre: 음악 장르 (예: "K-Pop", "Pop", "R&B", "Hip-Hop", "Ballad", "EDM")
|
||||
genre: 음악 장르 (예: "K-Pop", "Pop", "R&B", "Hip-Hop", "Ballad", "EDM", "Rock", "Jazz")
|
||||
None일 경우 style 파라미터를 전송하지 않음
|
||||
callback_url: 생성 완료 시 알림 받을 URL (None일 경우 config에서 기본값 사용)
|
||||
instrumental: True이면 BGM 전용 — 더미 가사로 60초 길이를 유도하고 보컬 없이 생성
|
||||
|
|
@ -137,7 +137,7 @@ class SunoService:
|
|||
logger.info(f"[Suno] BGM 더미 가사 장르 {normalized_genre} 선택됨")
|
||||
else:
|
||||
formatted_prompt = (
|
||||
f"[Song Duration: Exactly 1 minute - Must be around 60 seconds]\n{prompt}"
|
||||
f"[Song Duration: Around 1 minute - Must be around 60 seconds]\n{prompt}"
|
||||
)
|
||||
|
||||
payload: dict[str, Any] = {
|
||||
|
|
|
|||
|
|
@ -26,13 +26,12 @@ from app.dependencies.pagination import PaginationParams, get_pagination_params
|
|||
from app.user.dependencies.auth import get_current_user, get_current_user_optional
|
||||
from app.user.models import User
|
||||
from app.utils.pagination import PaginatedResponse
|
||||
from app.home.models import Image, Project, MarketingIntel, ImageTag
|
||||
from app.home.models import Image, Project, MarketingIntel
|
||||
from app.home.api.routers.v1.home import _extract_region_from_address
|
||||
from app.utils.address_parser import SIDO_CITIES, SIDO_SEARCH_ALIASES
|
||||
from app.lyric.models import Lyric
|
||||
from app.song.models import Song, SongTimestamp
|
||||
from app.utils.creatomate import CreatomateService
|
||||
from app.utils.subtitles import SubtitleContentsGenerator
|
||||
|
||||
from app.comment.models import Comment
|
||||
from app.database.like_cache import (
|
||||
|
|
@ -59,7 +58,7 @@ from app.video.schemas.video_schema import (
|
|||
VideoThumbnailItem,
|
||||
)
|
||||
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
|
||||
|
||||
|
|
@ -199,16 +198,22 @@ async def generate_video(
|
|||
)
|
||||
marketing_intelligence: MarketingIntel = marketing_result.scalar_one_or_none()
|
||||
|
||||
# subtitle 미완료 시 즉시 반환 — Lyric/Song/Image 쿼리 전에 체크하여 불필요한 조회 방지
|
||||
# 자막 + 이미지 배정 미완료 시 즉시 반환 — Lyric/Song/Image 쿼리 전에 체크하여 불필요한 조회 방지
|
||||
# 클라이언트가 /lyric/subtitle/status/{task_id} 폴링 후 재시도
|
||||
if not marketing_intelligence.subtitle or not marketing_intelligence.image_match:
|
||||
pending_what = []
|
||||
if not marketing_intelligence.subtitle:
|
||||
logger.info(f"[generate_video] subtitle pending - task_id: {task_id}")
|
||||
pending_what.append("자막")
|
||||
if not marketing_intelligence.image_match:
|
||||
pending_what.append("이미지 배정")
|
||||
pending_msg = ", ".join(pending_what)
|
||||
logger.info(f"[generate_video] 사전 준비 미완료 ({pending_msg}) - task_id: {task_id}")
|
||||
return GenerateVideoResponse(
|
||||
success=False,
|
||||
status="subtitle_pending",
|
||||
task_id=task_id,
|
||||
creatomate_render_id=None,
|
||||
message="자막 생성이 아직 완료되지 않았습니다. /lyric/subtitle/status/{task_id}로 완료 확인 후 재요청하세요.",
|
||||
message=f"{pending_msg} 생성이 아직 완료되지 않았습니다. /lyric/subtitle/status/{{task_id}}로 완료 확인 후 재요청하세요.",
|
||||
error_message=None,
|
||||
)
|
||||
|
||||
|
|
@ -365,25 +370,15 @@ async def generate_video(
|
|||
logger.debug(f"[generate_video] Template fetched - task_id: {task_id}")
|
||||
|
||||
# 6-2. elements에서 리소스 매핑 생성
|
||||
# modifications = creatomate_service.elements_connect_resource_blackbox(
|
||||
# elements=template["source"]["elements"],
|
||||
# image_url_list=image_urls,
|
||||
# music_url=music_url,
|
||||
# address=store_address
|
||||
taged_image_list = await get_image_tags_by_task_id(task_id)
|
||||
min_image_num = creatomate_service.counting_component(
|
||||
template = template,
|
||||
target_template_type = "image"
|
||||
)
|
||||
duplicate = bool(len(taged_image_list) < min_image_num)
|
||||
logger.info(f"[generate_video] Duplicate : {duplicate} | length of taged_image {len(taged_image_list)}, min_len {min_image_num},- task_id: {task_id}")
|
||||
modifications = creatomate_service.template_matching_taged_image(
|
||||
template = template,
|
||||
taged_image_list = taged_image_list,
|
||||
music_url = music_url,
|
||||
address = store_address,
|
||||
duplicate = duplicate,
|
||||
)
|
||||
# 이미지 배정은 /lyric/generate 사전 단계에서 이미 수행되어 marketing_intelligence.image_match에 저장됨.
|
||||
# 여기서는 사전 산출물을 읽어 음악 URL과 주소만 보완한다.
|
||||
modifications: dict = dict(marketing_intelligence.image_match) # 슬롯→image_url 사전 배정 결과
|
||||
modifications["audio-music"] = music_url
|
||||
# address_input 슬롯은 사전 단계에서도 채워지지만 영상 시점에 재확인
|
||||
for key in list(modifications.keys()):
|
||||
if "address_input" in key:
|
||||
modifications[key] = store_address
|
||||
logger.info(f"[generate_video] image_match loaded from DB (slots: {len(modifications)}) - task_id: {task_id}")
|
||||
logger.debug(f"[generate_video] Modifications created - task_id: {task_id}")
|
||||
|
||||
subtitle_modifications = marketing_intelligence.subtitle
|
||||
|
|
@ -391,14 +386,14 @@ async def generate_video(
|
|||
modifications.update(subtitle_modifications)
|
||||
|
||||
# revert thumbnail scene
|
||||
# thumbnail_modifications = creatomate_service.make_thumbnail_modification(
|
||||
# brand_name =brand_name,
|
||||
# region = region,
|
||||
# brand_concept = brand_concept,
|
||||
# category_definition= category_definition,
|
||||
# target_keywords=target_keywords)
|
||||
thumbnail_modifications = creatomate_service.make_thumbnail_modification(
|
||||
brand_name =brand_name,
|
||||
region = region,
|
||||
brand_concept = brand_concept,
|
||||
category_definition= category_definition,
|
||||
target_keywords=target_keywords)
|
||||
|
||||
# modifications.update(thumbnail_modifications)
|
||||
modifications.update(thumbnail_modifications)
|
||||
|
||||
# 6-3. elements 수정
|
||||
new_elements = creatomate_service.modify_element(
|
||||
|
|
|
|||
|
|
@ -0,0 +1,218 @@
|
|||
"""
|
||||
Creative Assets Background Tasks
|
||||
|
||||
영상 생성 전 사전 준비(이미지 배정 + 자막 생성)를 수행하는 백그라운드 태스크를 정의합니다.
|
||||
"""
|
||||
|
||||
from sqlalchemy import select
|
||||
|
||||
from app.database.session import BackgroundSessionLocal
|
||||
from app.home.models import Project, MarketingIntel
|
||||
from app.utils.subtitles import SubtitleContentsGenerator
|
||||
from app.utils.creatomate import CreatomateService
|
||||
from app.utils.logger import get_logger
|
||||
from app.video.services.video import get_image_tags_by_task_id
|
||||
from app.utils.prompts.schemas.image import NARRATIVE_PHASE_LABEL
|
||||
|
||||
# 로거 설정
|
||||
logger = get_logger("video")
|
||||
|
||||
|
||||
async def generate_subtitle_background(
|
||||
orientation: str,
|
||||
task_id: str,
|
||||
max_retries: int = 3,
|
||||
) -> None:
|
||||
"""자막 + 이미지 배정을 사전에 수행하는 백그라운드 태스크.
|
||||
|
||||
내부적으로 `generate_creative_assets_background`를 호출한다.
|
||||
"""
|
||||
await generate_creative_assets_background(
|
||||
orientation=orientation,
|
||||
task_id=task_id,
|
||||
max_retries=max_retries,
|
||||
)
|
||||
|
||||
|
||||
async def generate_creative_assets_background(
|
||||
orientation: str,
|
||||
task_id: str,
|
||||
max_retries: int = 3,
|
||||
) -> None:
|
||||
"""이미지 배정 + 자막 생성을 사전에 수행하는 백그라운드 태스크.
|
||||
|
||||
실행 순서:
|
||||
1. 템플릿 조회 (읽기 전용, 렌더 없음)
|
||||
2. 이미지 매칭 — get_image_tags_by_task_id → template_matching_taged_image
|
||||
→ MarketingIntel.image_match 저장
|
||||
3. 자막 생성 — 배정된 이미지 컨텍스트(슬롯별 공간/단계 정보)를 pitching 리스트에 포함
|
||||
→ MarketingIntel.subtitle 저장
|
||||
"""
|
||||
logger.info(f"[generate_creative_assets_background] START - task_id: {task_id}, orientation: {orientation}")
|
||||
|
||||
for attempt in range(1, max_retries + 1):
|
||||
try:
|
||||
creatomate_service = CreatomateService(orientation=orientation)
|
||||
template = await creatomate_service.get_one_template_data(creatomate_service.template_id)
|
||||
|
||||
# ── 프로젝트 / 마케팅 인텔 로드 ──────────────────────────────────
|
||||
async with BackgroundSessionLocal() as session:
|
||||
project_result = await session.execute(
|
||||
select(Project)
|
||||
.where(Project.task_id == task_id)
|
||||
.order_by(Project.created_at.desc())
|
||||
.limit(1)
|
||||
)
|
||||
project = project_result.scalar_one_or_none()
|
||||
marketing_result = await session.execute(
|
||||
select(MarketingIntel).where(MarketingIntel.id == project.marketing_intelligence)
|
||||
)
|
||||
marketing_intelligence = marketing_result.scalar_one_or_none()
|
||||
|
||||
store_address = project.detail_region_info
|
||||
customer_name = project.store_name
|
||||
language = project.language or "Korean"
|
||||
logger.info(
|
||||
f"[generate_creative_assets_background] customer: {customer_name}, "
|
||||
f"address: {store_address}, language: {language} - task_id: {task_id}"
|
||||
)
|
||||
|
||||
# ── Step 1: 이미지 매칭 ─────────────────────────────────────────
|
||||
# marketing_acceptable 필터링은 크롤링 단계에서 이미 완료되었으므로 재필터링하지 않는다.
|
||||
taged_image_list = await get_image_tags_by_task_id(task_id)
|
||||
min_image_num = creatomate_service.counting_component(
|
||||
template=template,
|
||||
target_template_type="image",
|
||||
)
|
||||
duplicate = len(taged_image_list) < min_image_num
|
||||
logger.info(
|
||||
f"[generate_creative_assets_background] image matching — "
|
||||
f"pool={len(taged_image_list)}, slots={min_image_num}, "
|
||||
f"duplicate={duplicate} - task_id: {task_id}"
|
||||
)
|
||||
|
||||
image_modifications, assigned_image_tags = creatomate_service.template_matching_taged_image(
|
||||
template=template,
|
||||
taged_image_list=taged_image_list,
|
||||
music_url="", # 음악 URL은 영상 생성 시점에 결정 — 사전 단계에서는 빈 값
|
||||
address=store_address,
|
||||
duplicate=duplicate,
|
||||
)
|
||||
# audio-music은 영상 생성 시점에 덮어씌워지므로 image_match에서 제거
|
||||
image_match = {k: v for k, v in image_modifications.items() if k != "audio-music"}
|
||||
|
||||
logger.info(f"[generate_creative_assets_background] image_match: {list(image_match.keys())} - task_id: {task_id}")
|
||||
|
||||
# image_match 저장
|
||||
async with BackgroundSessionLocal() as session:
|
||||
marketing_result = await session.execute(
|
||||
select(MarketingIntel).where(MarketingIntel.id == project.marketing_intelligence)
|
||||
)
|
||||
mi = marketing_result.scalar_one_or_none()
|
||||
mi.image_match = image_match
|
||||
await session.commit()
|
||||
|
||||
logger.info(f"[generate_creative_assets_background] image_match saved - task_id: {task_id}")
|
||||
|
||||
# ── Step 2: 자막 생성 (이미지 컨텍스트 포함) ──────────────────────
|
||||
# 슬롯별 배정된 이미지의 공간/내러티브 정보를 pitching 레이블에 포함하여
|
||||
# LLM이 "이 자막이 나올 때 화면에 보이는 이미지"를 인지하고 자막을 작성할 수 있게 함.
|
||||
pitchings_raw = creatomate_service.extract_text_format_from_template(template)
|
||||
pitchings_with_context = _build_pitching_list_with_image_context(
|
||||
pitching_label_list=pitchings_raw,
|
||||
assigned_image_tags=assigned_image_tags,
|
||||
)
|
||||
|
||||
subtitle_generator = SubtitleContentsGenerator()
|
||||
generated_subtitles = await subtitle_generator.generate_subtitle_contents(
|
||||
marketing_intelligence=marketing_intelligence.intel_result,
|
||||
pitching_label_list=pitchings_with_context,
|
||||
customer_name=customer_name,
|
||||
detail_region_info=store_address,
|
||||
language=language,
|
||||
industry=project.industry,
|
||||
)
|
||||
pitching_output_list = generated_subtitles.pitching_results
|
||||
|
||||
subtitle_modifications = {
|
||||
pitching_output.pitching_tag: pitching_output.pitching_data
|
||||
for pitching_output in pitching_output_list
|
||||
}
|
||||
logger.info(f"[generate_creative_assets_background] subtitle_modifications: {subtitle_modifications}")
|
||||
|
||||
# subtitle 저장
|
||||
async with BackgroundSessionLocal() as session:
|
||||
marketing_result = await session.execute(
|
||||
select(MarketingIntel).where(MarketingIntel.id == project.marketing_intelligence)
|
||||
)
|
||||
mi = marketing_result.scalar_one_or_none()
|
||||
mi.subtitle = subtitle_modifications
|
||||
await session.commit()
|
||||
|
||||
logger.info(
|
||||
f"[generate_creative_assets_background] DONE - task_id: {task_id} "
|
||||
f"(attempt {attempt}/{max_retries})"
|
||||
)
|
||||
return
|
||||
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"[generate_creative_assets_background] FAILED (attempt {attempt}/{max_retries}) "
|
||||
f"- task_id: {task_id}, error: {e}",
|
||||
exc_info=True,
|
||||
)
|
||||
if attempt < max_retries:
|
||||
logger.info(
|
||||
f"[generate_creative_assets_background] 재시도 중... "
|
||||
f"({attempt + 1}/{max_retries}) - task_id: {task_id}"
|
||||
)
|
||||
|
||||
logger.error(f"[generate_creative_assets_background] 모든 재시도 실패 - task_id: {task_id}")
|
||||
|
||||
|
||||
def _build_pitching_list_with_image_context(
|
||||
pitching_label_list: list[str],
|
||||
assigned_image_tags: dict[str, dict],
|
||||
) -> list[str]:
|
||||
"""pitching 레이블 리스트에 배정 이미지 컨텍스트를 포함하여 반환합니다.
|
||||
|
||||
슬롯명(`bedroom-furniture-tight_crop-slow_zoom_in-welcome`)의 5번째 토큰과
|
||||
pitching 레이어명(`subtitle-welcome-emotion_cue-empathic-001`)의 2번째 토큰이
|
||||
동일한 narrative_phase이므로 시간 겹침 계산 없이 phase로 직접 매칭합니다.
|
||||
|
||||
출력 예시:
|
||||
"subtitle-welcome-emotion_cue-empathic-001 | 화면: lobby / furniture / 진입(welcome) 단계"
|
||||
"""
|
||||
# narrative_phase → 해당 phase에 배정된 image_tag 매핑 구성
|
||||
# 슬롯명 형식: {space_type}-{subject}-{camera}-{motion}-{narrative_phase}
|
||||
phase_to_tag: dict[str, dict] = {}
|
||||
for slot_name, tag in assigned_image_tags.items():
|
||||
tokens = slot_name.split("-")
|
||||
if len(tokens) >= 5:
|
||||
phase = tokens[4]
|
||||
# 같은 phase에 슬롯이 여러 개면 첫 번째만 사용
|
||||
if phase not in phase_to_tag:
|
||||
phase_to_tag[phase] = tag
|
||||
|
||||
if not phase_to_tag:
|
||||
return pitching_label_list
|
||||
|
||||
result = []
|
||||
for label in pitching_label_list:
|
||||
# pitching 레이어명 형식: {track_role}-{narrative_phase}-{content_type}-{tone}-{pair_id}
|
||||
tokens = label.split("-")
|
||||
context_str = ""
|
||||
if len(tokens) >= 2:
|
||||
phase = tokens[1]
|
||||
tag = phase_to_tag.get(phase)
|
||||
if tag:
|
||||
space_types = tag.get("space_type", [])
|
||||
subjects = tag.get("subject", [])
|
||||
space_str = "/".join(space_types[:2]) if space_types else "?"
|
||||
subject_str = subjects[0] if subjects else "?"
|
||||
phase_str = NARRATIVE_PHASE_LABEL.get(phase, phase)
|
||||
context_str = f" | 화면: {space_str} / {subject_str} / {phase_str} 단계"
|
||||
|
||||
result.append(label + context_str)
|
||||
|
||||
return result
|
||||
|
|
@ -8,4 +8,5 @@ ALTER TABLE project
|
|||
ADD COLUMN biz_type VARCHAR(50) NULL DEFAULT 'place'
|
||||
COMMENT '업종 분류 (place: 기본, accommodation: 숙박, restaurant: 음식점)' AFTER id;
|
||||
|
||||
ALTER TABLE project CHANGE biz_type industry VARCHAR(50) DEFAULT '';
|
||||
ALTER TABLE project CHANGE biz_type industry VARCHAR(50) DEFAULT ''
|
||||
COMMENT '업종 분류' AFTER id;
|
||||
|
|
|
|||
|
|
@ -0,0 +1,12 @@
|
|||
-- ============================================================
|
||||
-- Migration: lyric 테이블에 genre 컬럼 추가
|
||||
-- Date: 2026-07-02
|
||||
-- Description: 가사 생성 시점에 확정된 음악 장르 저장
|
||||
-- - 수동 선택: 사용자가 고른 장르
|
||||
-- - 자동 선택: GPT가 가사 무드에 맞춰 추천한 장르
|
||||
-- 장르-BPM 불일치로 인한 곡 길이 편차 문제 해결을 위해 도입
|
||||
-- ============================================================
|
||||
|
||||
ALTER TABLE lyric
|
||||
ADD COLUMN genre VARCHAR(20) NULL
|
||||
COMMENT '음악 장르 (kpop, pop, ballad, hip-hop, rnb, edm, jazz, rock). 수동 선택 시 요청값, 자동 선택 시 GPT 추천값' AFTER language;
|
||||
Loading…
Reference in New Issue