diff --git a/app/home/api/routers/v1/home.py b/app/home/api/routers/v1/home.py index 26e04ea..1a5d951 100644 --- a/app/home/api/routers/v1/home.py +++ b/app/home/api/routers/v1/home.py @@ -97,11 +97,13 @@ def _extract_region_from_address(road_address: str | None, jibun_address: str | class _IndustryOutput(BaseModel): - industry: Literal["stay", "restaurant", "cafe", "salon", "clinic", "fitness", "academy", "attraction"] + industry: Literal[ + "stay", "restaurant", "cafe", "salon", "clinic", "fitness", "academy", "attraction", "general" + ] async def _resolve_industry(category: str, customer_name: str = "") -> str: - """업체를 통합 프롬프트의 8개 industry enum 중 하나로 AI 분류. + """업체를 통합 프롬프트의 9개 industry enum 중 하나로 AI 분류. 분류 근거는 카테고리를 우선 사용하고, 카테고리가 없으면 업체명으로 분류한다. 근거가 둘 다 없으면 빈 문자열 반환. API 장애 시 예외 전파(하위 분석도 동일 API 의존). @@ -115,10 +117,12 @@ async def _resolve_industry(category: str, customer_name: str = "") -> str: chatgpt = ChatgptService() prompt = ( f"{basis_label}: '{basis_value}'\n" - "위 정보를 바탕으로 이 업체를 다음 8개 업종 중 가장 적합한 하나로 분류하세요: " + "위 정보를 바탕으로 이 업체를 다음 업종 중 가장 적합한 하나로 분류하세요: " "stay(숙박/펜션/호텔), restaurant(음식점), cafe(카페/디저트), " "salon(미용실/네일/뷰티), clinic(병원/의원/치과), fitness(헬스/필라테스/요가), " - "academy(학원/교습소), attraction(관광/체험/액티비티)." + "academy(학원/교습소), attraction(관광/체험/액티비티). " + "위 8개 중 어느 것에도 명확히 해당하지 않는 경우에만 general(기타/범용 업종)로 분류하세요. " + "유사한 업종이 있으면 general 대신 그 업종을 우선 선택하세요." ) result = await chatgpt._call_pydantic_output_chat_completion( prompt=prompt, @@ -312,7 +316,7 @@ async def _crawling_logic( [img["original"] for img in extra_photo_urls], industry ) except Exception as e: - logger.error(f"[crawling] Step 3 FAILED - 이미지 필터링 중 오류: {e}") + # logger.error(f"[crawling] Step 3 FAILED - 이미지 필터링 중 오류: {e}") raise HTTPException( status_code=status.HTTP_502_BAD_GATEWAY, detail="이미지 마케팅 적합성 필터링 중 오류가 발생했습니다.", @@ -337,11 +341,22 @@ async def _crawling_logic( try: # Step 4-1: ChatGPT 서비스 초기화 및 입력 데이터 구성 chatgpt_service = ChatgptService() + # 메뉴 전달은 잠정 보류 — 사이드 메뉴(볶음밥·냉면 등)가 분석에 과도한 영향을 주는 + # 문제로 프롬프트에서 {menu_info}를 제거함. 크롤링(scraper.menu_info)은 유지 중이므로 + # 주력/사이드 구분 방안이 정리되면 아래 주석을 되살려 재전달할 것. + # menus = scraper.menu_info or [] + # menus = sorted(menus, key=lambda m: not m.get("recommend"))[:15] + # menu_info = ", ".join(m["name"] for m in menus if m.get("name")) + top_keywords = (scraper.voted_keyword_stats or [])[:5] input_marketing_data = { "customer_name": customer_name, "region": region, "detail_region_info": road_address or "", "industry": industry, # 통합 프롬프트 내부 분기용 업종 enum + "category": category, # 네이버 지도 원본 카테고리 텍스트 + "facility_info": scraper.facility_info or "", + "voted_keywords": ", ".join(kw["displayName"] for kw in top_keywords), + # "menu_info": menu_info, # 실제 판매 품목 (셀링포인트 구체화용) } # Step 4-2: GPT API 호출 → 구조화된 마케팅 분석 결과 반환 @@ -456,6 +471,7 @@ async def manual_marketing( "region": region, "detail_region_info": request_body.address, "industry": industry, # 통합 프롬프트 내부 분기용 업종 enum + "category": request_body.category, # 사용자 입력 원본 카테고리 텍스트 (세부 묘사 참고용) } marketing_analysis = await chatgpt_service.generate_structured_output( marketing_prompt, input_marketing_data @@ -658,7 +674,7 @@ async def upload_images_blob( ), industry: str = Form( default="", - description="업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction). 크롤링 응답의 industry 값을 그대로 전달", + description="업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction|general). 크롤링 응답의 industry 값을 그대로 전달", ), current_user: User = Depends(get_current_user), ) -> ImageUploadResponse: diff --git a/app/home/models.py b/app/home/models.py index 6a0d9d7..0417c39 100644 --- a/app/home/models.py +++ b/app/home/models.py @@ -118,7 +118,7 @@ class Project(Base): String(50), nullable=False, default="", - comment="업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction)", + comment="업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction|general)", ) language: Mapped[str] = mapped_column( diff --git a/app/home/schemas/home_schema.py b/app/home/schemas/home_schema.py index 1fcf753..a6635e5 100644 --- a/app/home/schemas/home_schema.py +++ b/app/home/schemas/home_schema.py @@ -78,7 +78,7 @@ class ProcessedInfo(BaseModel): customer_name: str = Field(..., description="고객명/가게명 (base_info.name)") region: str = Field(..., description="지역명 (roadAddress에서 추출한 시 이름)") detail_region_info: str = Field(..., description="상세 지역 정보 (roadAddress)") - industry: str = Field(default="", description="업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction)") + industry: str = Field(default="", description="업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction|general)") # class MarketingAnalysisDetail(BaseModel): @@ -249,7 +249,7 @@ class CrawlingResponse(BaseModel): None, description="마케팅 분석 결과 . 실패 시 null" ) m_id : int = Field(..., description="마케팅 분석 결과 ID") - industry: str = Field(default="", description="업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction)") + industry: str = Field(default="", description="업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction|general)") class ManualMarketingRequest(BaseModel): diff --git a/app/lyric/schemas/lyric.py b/app/lyric/schemas/lyric.py index 59f31bf..ff476d6 100644 --- a/app/lyric/schemas/lyric.py +++ b/app/lyric/schemas/lyric.py @@ -76,7 +76,7 @@ class GenerateLyricRequest(BaseModel): description="영상 방향 (horizontal: 가로형, vertical: 세로형)", ), m_id : Optional[int] = Field(None, description="마케팅 인텔리전스 ID 값") - industry: str = Field(default="", description="업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction)") + industry: str = Field(default="", description="업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction|general)") instrumental: bool = Field(default=False, description="BGM 전용 모드 (가사 생성 안 함)") genre: Optional[str] = Field( None, diff --git a/app/utils/autotag.py b/app/utils/autotag.py index aa526cf..8b66a06 100644 --- a/app/utils/autotag.py +++ b/app/utils/autotag.py @@ -36,7 +36,7 @@ async def autotag_images(image_url_list : list[str], industry: str = "") -> list MAX_RETRY = 2 for _ in range(MAX_RETRY): failed_idx = [i for i, r in enumerate(image_result_list) if isinstance(r, Exception)] - print("Failed", failed_idx) + # print("Failed", failed_idx) if not failed_idx: break retried = await asyncio.gather( @@ -46,5 +46,5 @@ async def autotag_images(image_url_list : list[str], industry: str = "") -> list for i, result in zip(failed_idx, retried): image_result_list[i] = result - print("Failed", failed_idx) + # print("Failed", failed_idx) return image_result_list \ No newline at end of file diff --git a/app/utils/creatomate.py b/app/utils/creatomate.py index 26cee72..61e3573 100644 --- a/app/utils/creatomate.py +++ b/app/utils/creatomate.py @@ -895,6 +895,20 @@ class CreatomateService: response.raise_for_status() return response.json() + @staticmethod + def _entrance_transition_offset(elem: dict) -> float: + """요소의 entrance transition(transition=true, time==0) duration 합을 반환합니다. + + 순차 배치 시 이 값만큼 이전 요소와 겹치도록 시작점이 앞당겨집니다. + """ + offset = 0.0 + for animation in elem.get("animations", []): + if "reversed" in animation: + continue + if "transition" in animation and animation["transition"] and animation.get("time", 0) == 0: + offset += animation["duration"] + return offset + def calc_scene_duration(self, template: dict) -> float: """템플릿의 전체 장면 duration을 계산합니다.""" total_template_duration = 0.0 @@ -908,15 +922,7 @@ class CreatomateService: if elem["track"] not in track_maximum_duration: continue if "time" not in elem or elem["time"] == 0: # elem is auto / 만약 마지막 elem이 auto인데 그 앞에 time이 있는 elem 일 시 버그 발생 확률 있음 - track_maximum_duration[elem["track"]] += elem["duration"] - - if "animations" not in elem: - continue - for animation in elem["animations"]: - if "reversed" in animation: - continue - if "transition" in animation and animation["transition"] and animation.get("time", 0) == 0: - track_maximum_duration[elem["track"]] -= animation["duration"] + track_maximum_duration[elem["track"]] += elem["duration"] - self._entrance_transition_offset(elem) else: track_maximum_duration[elem["track"]] = max(track_maximum_duration[elem["track"]], elem["time"] + elem["duration"]) @@ -928,6 +934,36 @@ class CreatomateService: return total_template_duration + def calc_track_time_ranges(self, template: dict, track: int) -> list[tuple[str, float, float]]: + """최상위 요소 중 지정 track의 (name, start, end) 시간 범위를 계산합니다. + + Creatomate 규칙에 따라 `time`이 없거나 0인 요소는 같은 트랙의 이전 요소 뒤에 + 순차 배치되며, entrance transition duration만큼 이전 요소와 겹치도록 + 시작점을 앞당깁니다(calc_scene_duration의 차감 규칙과 동치). + `time`이 명시된 요소는 해당 시각에서 시작하고, 커서는 그 요소의 끝으로 이동합니다. + """ + ranges: list[tuple[str, float, float]] = [] + cursor = 0.0 + for elem in template["source"]["elements"]: + try: + if elem.get("track") != track: + continue + duration = elem["duration"] + + if "time" not in elem or elem["time"] == 0: # 순차 배치(auto) + start = max(0.0, cursor - self._entrance_transition_offset(elem)) + else: + start = elem["time"] + + end = start + duration + cursor = end + ranges.append((elem.get("name", ""), start, end)) + except Exception as e: + logger.debug(traceback.format_exc()) + logger.error(f"[calc_track_time_ranges] Error processing element: {elem}, {e}") + + return ranges + def extend_template_duration(self, template: dict, target_duration: float) -> dict: """템플릿의 duration을 target_duration으로 확장합니다.""" # template["duration"] = target_duration + 0.5 # 늘린것보단 짧게 diff --git a/app/utils/image_filter.py b/app/utils/image_filter.py index 4db8e58..b32abfe 100644 --- a/app/utils/image_filter.py +++ b/app/utils/image_filter.py @@ -81,7 +81,7 @@ async def filter_marketing_images(image_url_list: list[str], industry: str = "") 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}") + # logger.warning(f"[image_filter] 재시도 대상 인덱스: {failed_idx}") retried = await asyncio.gather( *[_call(image_url_list[i]) for i in failed_idx], return_exceptions=True ) @@ -89,10 +89,10 @@ async def filter_marketing_images(image_url_list: list[str], industry: str = "") 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}" - ) + # if final_failed: + # logger.warning( + # f"[image_filter] {len(final_failed)}건 최종 실패 → 보수적으로 제외 처리: {final_failed}" + # ) return [ (not isinstance(r, Exception)) and r.marketing_acceptable diff --git a/app/utils/nvMapScraper.py b/app/utils/nvMapScraper.py index 99fe1ba..7bc12fc 100644 --- a/app/utils/nvMapScraper.py +++ b/app/utils/nvMapScraper.py @@ -52,6 +52,12 @@ query getAccommodation($id: String!, $deviceType: String) { conveniences visitorReviewsTotal } + menus { + name + price + description + recommend + } images { images { origin url } } } }""" @@ -100,6 +106,7 @@ query getVisitorReviewStats($id: String!) { self.base_info: dict | None = None self.facility_info: str | None = None self.voted_keyword_stats: list[dict] | None = None # 키워드 투표 집계 (displayName, count) + self.menu_info: list[dict] | None = None # 메뉴 목록 (name, price, description, recommend) def _get_request_headers(self) -> dict: headers = self.DEFAULT_HEADERS.copy() @@ -209,7 +216,8 @@ query getVisitorReviewStats($id: String!) { # ── 실제 브라우저로 WTM 캡차 우회 ── data, stats_data, extra_photo_urls = await self._scrap_via_browser(place_id) - fac_data = None # 편의시설(HTML)은 브라우저 경로에서 생략 (best-effort) + # 편의시설은 HTML 페이지 파싱이라 GraphQL(WTM) 차단과 별개로 직접 시도 (best-effort, 실패 시 None) + fac_data = await self._get_facility_string(place_id) self.scrap_type = "GraphQL-Browser" self.rawdata = data @@ -243,6 +251,7 @@ query getVisitorReviewStats($id: String!) { self.base_info = data["data"]["business"]["base"] self.facility_info = fac_data self.voted_keyword_stats = stats_data + self.menu_info = business.get("menus") or None return diff --git a/app/utils/prompts/chatgpt_prompt.py b/app/utils/prompts/chatgpt_prompt.py index 9581ce4..224b7a0 100644 --- a/app/utils/prompts/chatgpt_prompt.py +++ b/app/utils/prompts/chatgpt_prompt.py @@ -149,17 +149,17 @@ class ChatgptService: logger.info(f"[ChatgptService({self.model_type})] Retrying request...") continue # Response 디버그 로깅 - logger.debug(f"[ChatgptService({self.model_type})] attempt: {attempt}") - logger.debug(f"[ChatgptService({self.model_type})] Response ID: {response.id}") - logger.debug(f"[ChatgptService({self.model_type})] Response finish_reason: {response.id}") - logger.debug(f"[ChatgptService({self.model_type})] Response model: {response.model}") + # logger.debug(f"[ChatgptService({self.model_type})] attempt: {attempt}") + # logger.debug(f"[ChatgptService({self.model_type})] Response ID: {response.id}") + # logger.debug(f"[ChatgptService({self.model_type})] Response finish_reason: {response.id}") + # logger.debug(f"[ChatgptService({self.model_type})] Response model: {response.model}") choice = response.choices[0] finish_reason = choice.finish_reason if finish_reason == "stop": - output_text = choice.message.content or "" - logger.debug(f"[ChatgptService({self.model_type})] Response output_text: {output_text[:200]}..." if len(output_text) > 200 else f"[ChatgptService] Response output_text: {output_text}") + # output_text = choice.message.content or "" + # logger.debug(f"[ChatgptService({self.model_type})] Response output_text: {output_text[:200]}..." if len(output_text) > 200 else f"[ChatgptService] Response output_text: {output_text}") return choice.message.parsed elif finish_reason == "length": diff --git a/app/utils/prompts/schemas/image.py b/app/utils/prompts/schemas/image.py index e08cc97..3a177b1 100644 --- a/app/utils/prompts/schemas/image.py +++ b/app/utils/prompts/schemas/image.py @@ -116,7 +116,7 @@ class ImageTagPromptInput(BaseModel): """이미지 태깅 프롬프트에 전달되는 입력 스키마. 분석할 이미지 URL과 업종 정보, LLM이 선택 가능한 태그 후보 리스트를 포함.""" img_url : str = Field(..., description="이미지 URL") - industry : str = Field(default="", description = "업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction). 통합 프롬프트가 이 값으로 내부 분기") + industry : str = Field(default="", description = "업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction|general). 통합 프롬프트가 이 값으로 내부 분기") space_type: list[str] = Field(list(SpaceType), description="공간적 정보를 가지는 태그 리스트") subject: list[str] = Field(list(Subject), description="피사체 정보를 가지는 태그 리스트") camera: list[str] = Field(list(Camera), description="카메라 정보를 가지는 태그 리스트") diff --git a/app/utils/prompts/schemas/lyric.py b/app/utils/prompts/schemas/lyric.py index 2cd9aa4..d42f0d4 100644 --- a/app/utils/prompts/schemas/lyric.py +++ b/app/utils/prompts/schemas/lyric.py @@ -12,7 +12,7 @@ class LyricPromptInput(BaseModel): language : 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). 통합 프롬프트가 이 값으로 내부 분기") + industry : str = Field(default="", description = "업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction|general). 통합 프롬프트가 이 값으로 내부 분기") # Output 정의 class LyricPromptOutput(BaseModel): diff --git a/app/utils/prompts/schemas/marketing.py b/app/utils/prompts/schemas/marketing.py index 5dbc906..2788c14 100644 --- a/app/utils/prompts/schemas/marketing.py +++ b/app/utils/prompts/schemas/marketing.py @@ -6,7 +6,12 @@ class MarketingPromptInput(BaseModel): customer_name : str = Field(..., description = "마케팅 대상 사업체 이름") region : str = Field(..., description = "마케팅 대상 지역") detail_region_info : str = Field(..., description = "마케팅 대상 지역 상세") - industry : str = Field(default="", description = "업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction). 통합 프롬프트가 이 값으로 내부 분기") + industry : str = Field(default="", description = "업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction|general). 통합 프롬프트가 이 값으로 내부 분기") + category : str = Field(default="", description = "네이버 지도 원본 카테고리 텍스트 (예: 펜션, 한식). industry 모듈 선택에는 관여하지 않고 세부 묘사의 참고 정보로만 사용") + facility_info : str = Field(default="", description = "편의시설 정보 (네이버 지도 크롤링, 없으면 빈 문자열)") + voted_keywords : str = Field(default="", description = "방문자 키워드 투표 상위 5개 (쉼표 구분, 없으면 빈 문자열)") + # 메뉴 전달은 잠정 보류 (사이드 메뉴가 분석에 과도한 영향을 주는 문제). 재도입 시 주석 해제. + # menu_info : str = Field(default="", description = "메뉴/상품 목록 (네이버 지도 크롤링, 쉼표 구분, 없으면 빈 문자열). 실제 판매 품목 기반 셀링포인트 작성에 사용") # Output 정의 class BrandIdentity(BaseModel): diff --git a/app/utils/prompts/schemas/subtitle.py b/app/utils/prompts/schemas/subtitle.py index 46491f7..e593bbd 100644 --- a/app/utils/prompts/schemas/subtitle.py +++ b/app/utils/prompts/schemas/subtitle.py @@ -10,7 +10,7 @@ class SubtitlePromptInput(BaseModel): customer_name : str = Field(..., description = "마케팅 대상 사업체 이름") detail_region_info : str = Field(..., description = "마케팅 대상 지역 상세") language : str = Field(default="Korean", description="자막 출력 언어 (Korean, English, Chinese, Japanese, Thai, Vietnamese)") - industry : str = Field(default="", description = "업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction). 통합 프롬프트가 이 값으로 내부 분기") + industry : str = Field(default="", description = "업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction|general). 통합 프롬프트가 이 값으로 내부 분기") # Output 정의 class PitchingOutput(BaseModel): diff --git a/app/utils/prompts/schemas/youtube.py b/app/utils/prompts/schemas/youtube.py index 2acae39..2fe0564 100644 --- a/app/utils/prompts/schemas/youtube.py +++ b/app/utils/prompts/schemas/youtube.py @@ -8,7 +8,7 @@ class YTUploadPromptInput(BaseModel): marketing_intelligence_summary : Optional[str] = Field(None, description = "마케팅 분석 정보 보고서") language : str= Field(..., description = "영상 언어") target_keywords: List[str] = Field(..., description="태그 키워드 리스트") - industry : str = Field(default="", description = "업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction). 통합 프롬프트가 이 값으로 내부 분기") + industry : str = Field(default="", description = "업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction|general). 통합 프롬프트가 이 값으로 내부 분기") # Output 정의 class YTUploadPromptOutput(BaseModel): diff --git a/app/video/services/video.py b/app/video/services/video.py index 460d487..db35fe3 100644 --- a/app/video/services/video.py +++ b/app/video/services/video.py @@ -839,7 +839,7 @@ logger = get_logger("video") # ) from sqlalchemy.dialects import mysql async def get_image_tags_by_task_id(task_id: str) -> list[dict]: - print("taskid", task_id) + # print("taskid", task_id) async with AsyncSessionLocal() as session: stmt = ( select(Image.img_url, ImageTag.img_tag) @@ -854,9 +854,9 @@ async def get_image_tags_by_task_id(task_id: str) -> list[dict]: ImageTag.img_tag.is_not(None), ) ) - print(stmt.compile(dialect=mysql.dialect(), compile_kwargs={"literal_binds": True})) + # print(stmt.compile(dialect=mysql.dialect(), compile_kwargs={"literal_binds": True})) rows = (await session.execute(stmt)).all() - print("rows", rows) - print(rows) - print("image" , [{"image_url": row.img_url, "image_tag": row.img_tag} for row in rows]) + # print("rows", rows) + # print(rows) + # print("image" , [{"image_url": row.img_url, "image_tag": row.img_tag} for row in rows]) return [{"image_url": row.img_url, "image_tag": row.img_tag} for row in rows] \ No newline at end of file diff --git a/app/video/worker/creative_assets_task.py b/app/video/worker/creative_assets_task.py index 9eedfd4..5c20e89 100644 --- a/app/video/worker/creative_assets_task.py +++ b/app/video/worker/creative_assets_task.py @@ -9,7 +9,12 @@ 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.creatomate import ( + CreatomateService, + SCENE_TRACK, + SUBTITLE_TRACK, + KEYWORD_TRACK, +) 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 @@ -121,6 +126,8 @@ async def generate_creative_assets_background( pitchings_with_context = _build_pitching_list_with_image_context( pitching_label_list=pitchings_raw, assigned_image_tags=assigned_image_tags, + template=template, + creatomate_service=creatomate_service, ) subtitle_generator = SubtitleContentsGenerator() @@ -134,8 +141,9 @@ async def generate_creative_assets_background( ) pitching_output_list = generated_subtitles.pitching_results + # LLM이 pitching_tag에 " | 화면: ..." 컨텍스트 접미를 그대로 되돌려줄 수 있으므로 제거 subtitle_modifications = { - pitching_output.pitching_tag: pitching_output.pitching_data + pitching_output.pitching_tag.split(IMAGE_CONTEXT_SEP)[0].strip(): pitching_output.pitching_data for pitching_output in pitching_output_list } logger.info(f"[generate_creative_assets_background] subtitle_modifications: {subtitle_modifications}") @@ -170,49 +178,113 @@ async def generate_creative_assets_background( logger.error(f"[generate_creative_assets_background] 모든 재시도 실패 - task_id: {task_id}") +# 이미지 컨텍스트 접미 구분자 — 생성(_context_for_label)과 제거(subtitle_modifications) 양쪽에서 사용 +IMAGE_CONTEXT_SEP = " | " + + +def _merge_unique(tags: list[dict], key: str) -> list: + """태그들의 key 리스트를 순서 유지하며 중복 제거해 병합합니다.""" + return list(dict.fromkeys(item for tag in tags for item in tag.get(key, []))) + + +def _aggregate_tags_by_composition( + template: dict, + assigned_image_tags: dict[str, dict], + creatomate_service: CreatomateService, +) -> dict[str, dict]: + """SCENE_TRACK 최상위 컴포지션별로 배정된 이미지 태그를 집계합니다. + + 컴포지션 하위 leaf image가 여러 개면(다중 이미지 씬) space_type/subject를 + 순서 유지 중복 제거로 병합합니다. 배정된 태그가 없는 컴포지션(CTA 등)은 + 결과에 포함하지 않습니다. + """ + comp_tags: dict[str, dict] = {} + for elem in template["source"]["elements"]: + if elem.get("track") != SCENE_TRACK or not elem.get("name"): + continue + name_to_type = creatomate_service.parse_template_component_name([elem]) + leaf_tags = [ + assigned_image_tags[name] + for name, elem_type in name_to_type.items() + if elem_type == "image" and name in assigned_image_tags + ] + if not leaf_tags: + continue + comp_tags[elem["name"]] = { + "space_type": _merge_unique(leaf_tags, "space_type"), + "subject": _merge_unique(leaf_tags, "subject"), + } + return comp_tags + + +def _context_for_label( + text_range: tuple[float, float], + scene_ranges: list[tuple[str, float, float]], + comp_tags: dict[str, dict], +) -> str: + """텍스트 재생 구간과 시간 겹침이 최대인 씬의 이미지 컨텍스트 접미를 만듭니다. + + 태그가 배정된 컴포지션 중 겹침이 최대인 씬을 선택하며(동률 시 이른 시작 우선), + 겹치는 씬이 없으면 빈 문자열을 반환합니다. + """ + text_start, text_end = text_range + candidates = [ + (min(text_end, comp_end) - max(text_start, comp_start), -comp_start, comp_name) + for comp_name, comp_start, comp_end in scene_ranges + if comp_name in comp_tags + ] + candidates = [c for c in candidates if c[0] > 0] + if not candidates: + return "" + + best_comp = max(candidates)[2] + tag = comp_tags[best_comp] + 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 "?" + # 컴포지션명 형식: {space_type}-{subject}-{camera}-{motion}-{narrative_phase} + comp_tokens = best_comp.split("-") + phase = comp_tokens[4] if len(comp_tokens) >= 5 else "" + phase_str = NARRATIVE_PHASE_LABEL.get(phase, phase) + return f"{IMAGE_CONTEXT_SEP}화면: {space_str} / {subject_str} / {phase_str} 단계" + + def _build_pitching_list_with_image_context( pitching_label_list: list[str], assigned_image_tags: dict[str, dict], + template: dict, + creatomate_service: CreatomateService, ) -> list[str]: """pitching 레이블 리스트에 배정 이미지 컨텍스트를 포함하여 반환합니다. - 슬롯명(`bedroom-furniture-tight_crop-slow_zoom_in-welcome`)의 5번째 토큰과 - pitching 레이어명(`subtitle-welcome-emotion_cue-empathic-001`)의 2번째 토큰이 - 동일한 narrative_phase이므로 시간 겹침 계산 없이 phase로 직접 매칭합니다. + 텍스트 레이어(SUBTITLE/KEYWORD_TRACK)와 이미지 컴포지션(SCENE_TRACK)의 + 시간 범위를 계산해, 각 텍스트가 재생되는 동안 화면에 가장 오래 보이는 + (시간 겹침이 최대인) 컴포지션의 이미지 태그를 컨텍스트로 붙입니다. + narrative_phase 토큰 매칭과 달리 phase 이름이 어긋나거나(자막 outro vs + 이미지 accent) 같은 phase에 씬이 여러 개여도 올바른 짝을 찾습니다. + + 겹치는 씬이 없거나 해당 씬에 배정된 이미지가 없으면(CTA 등) 컨텍스트를 + 생략합니다. 렌더 시 extend_template_duration은 전 트랙 균등 배율이므로 + 원본 템플릿 기준 겹침 관계는 스케일 후에도 유지됩니다. 출력 예시: "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: + comp_tags = _aggregate_tags_by_composition(template, assigned_image_tags, creatomate_service) + if not comp_tags: return pitching_label_list + scene_ranges = creatomate_service.calc_track_time_ranges(template, SCENE_TRACK) + text_ranges: dict[str, tuple[float, float]] = {} + for track in (SUBTITLE_TRACK, KEYWORD_TRACK): + for name, start, end in creatomate_service.calc_track_time_ranges(template, track): + text_ranges[name] = (start, end) + 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} 단계" - + text_range = text_ranges.get(label) + context_str = _context_for_label(text_range, scene_ranges, comp_tags) if text_range else "" result.append(label + context_str) return result