Compare commits
No commits in common. "2dadb843c8a64d2712673b20400c5ffdc4197ad7" and "4d8462651a8a8a17c6c3f6030ded8bb2908c6944" have entirely different histories.
2dadb843c8
...
4d8462651a
|
|
@ -97,13 +97,11 @@ 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", "general"
|
||||
]
|
||||
industry: Literal["stay", "restaurant", "cafe", "salon", "clinic", "fitness", "academy", "attraction"]
|
||||
|
||||
|
||||
async def _resolve_industry(category: str, customer_name: str = "") -> str:
|
||||
"""업체를 통합 프롬프트의 9개 industry enum 중 하나로 AI 분류.
|
||||
"""업체를 통합 프롬프트의 8개 industry enum 중 하나로 AI 분류.
|
||||
|
||||
분류 근거는 카테고리를 우선 사용하고, 카테고리가 없으면 업체명으로 분류한다.
|
||||
근거가 둘 다 없으면 빈 문자열 반환. API 장애 시 예외 전파(하위 분석도 동일 API 의존).
|
||||
|
|
@ -117,12 +115,10 @@ 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(관광/체험/액티비티)."
|
||||
"위 8개 중 어느 것에도 명확히 해당하지 않는 경우에만 general(기타/범용 업종)로 분류하세요. "
|
||||
"유사한 업종이 있으면 general 대신 그 업종을 우선 선택하세요."
|
||||
)
|
||||
result = await chatgpt._call_pydantic_output_chat_completion(
|
||||
prompt=prompt,
|
||||
|
|
@ -316,7 +312,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="이미지 마케팅 적합성 필터링 중 오류가 발생했습니다.",
|
||||
|
|
@ -341,22 +337,11 @@ 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 호출 → 구조화된 마케팅 분석 결과 반환
|
||||
|
|
@ -471,7 +456,6 @@ 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
|
||||
|
|
@ -674,7 +658,7 @@ async def upload_images_blob(
|
|||
),
|
||||
industry: str = Form(
|
||||
default="",
|
||||
description="업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction|general). 크롤링 응답의 industry 값을 그대로 전달",
|
||||
description="업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction). 크롤링 응답의 industry 값을 그대로 전달",
|
||||
),
|
||||
current_user: User = Depends(get_current_user),
|
||||
) -> ImageUploadResponse:
|
||||
|
|
|
|||
|
|
@ -118,7 +118,7 @@ class Project(Base):
|
|||
String(50),
|
||||
nullable=False,
|
||||
default="",
|
||||
comment="업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction|general)",
|
||||
comment="업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction)",
|
||||
)
|
||||
|
||||
language: Mapped[str] = mapped_column(
|
||||
|
|
|
|||
|
|
@ -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|general)")
|
||||
industry: str = Field(default="", description="업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction)")
|
||||
|
||||
|
||||
# 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|general)")
|
||||
industry: str = Field(default="", description="업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction)")
|
||||
|
||||
|
||||
class ManualMarketingRequest(BaseModel):
|
||||
|
|
|
|||
|
|
@ -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|general)")
|
||||
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,
|
||||
|
|
|
|||
|
|
@ -35,33 +35,6 @@ logger = get_logger("song")
|
|||
|
||||
router = APIRouter(prefix="/song", tags=["Song"])
|
||||
|
||||
TARGET_SONG_DURATION_SECONDS = 60.0
|
||||
|
||||
|
||||
def _select_clip_by_duration(
|
||||
clips_data: list[dict], target_seconds: float = TARGET_SONG_DURATION_SECONDS
|
||||
) -> dict:
|
||||
"""생성된 클립 중 목표 길이(target_seconds)에 가장 가까운 클립을 선택합니다.
|
||||
|
||||
길이 차이가 동일하면 먼저 등장한 클립(더 낮은 인덱스)을 선택합니다.
|
||||
duration 정보가 없는 클립은 비교 대상에서 제외되며, 모든 클립에 duration이 없으면
|
||||
첫 번째 클립을 반환합니다.
|
||||
"""
|
||||
best_clip = None
|
||||
best_diff = None
|
||||
|
||||
for clip in clips_data:
|
||||
duration = clip.get("duration")
|
||||
if duration is None:
|
||||
continue
|
||||
|
||||
diff = abs(duration - target_seconds)
|
||||
if best_diff is None or diff < best_diff:
|
||||
best_diff = diff
|
||||
best_clip = clip
|
||||
|
||||
return best_clip if best_clip is not None else clips_data[0]
|
||||
|
||||
|
||||
@router.post(
|
||||
"/generate/{task_id}",
|
||||
|
|
@ -440,14 +413,12 @@ async def get_song_status(
|
|||
clips_data = response_data.get("sunoData") or []
|
||||
|
||||
if clips_data:
|
||||
# 생성된 클립(보통 2개) 중 목표 길이(1분)에 가장 가까운 클립 선택 (동일하면 첫 번째)
|
||||
first_clip = _select_clip_by_duration(clips_data)
|
||||
# 첫 번째 클립(clips[0])의 audioUrl과 duration 사용
|
||||
first_clip = clips_data[0]
|
||||
audio_url = first_clip.get("audioUrl")
|
||||
clip_duration = first_clip.get("duration")
|
||||
logger.debug(
|
||||
f"[get_song_status] Selected clip by duration - id: {first_clip.get('id')}, "
|
||||
f"audio_url: {audio_url}, duration: {clip_duration}, "
|
||||
f"candidates: {[(c.get('id'), c.get('duration')) for c in clips_data]}"
|
||||
f"[get_song_status] Using first clip - id: {first_clip.get('id')}, audio_url: {audio_url}, duration: {clip_duration}"
|
||||
)
|
||||
|
||||
if audio_url:
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
@ -225,48 +225,29 @@ autotext_template_h_1 = {
|
|||
"stroke_color": "#333333",
|
||||
"stroke_width": "0.2 vmin"
|
||||
}
|
||||
DVST0001 = "fe11aeab-ff29-4bc8-9f75-c695c7e243e6"
|
||||
DVRT0001 = "3c4b74c4-82d7-4742-9b05-4b999fef4cbd"
|
||||
DVCF0001 = "ebb532ec-e49a-4a7e-b0e2-a8132f1bf1f7"
|
||||
DVST0001 = "75161273-0422-4771-adeb-816bd7263fb0"
|
||||
DVST0002 = "c68cf750-bc40-485a-a2c5-3f9fe301e386"
|
||||
DVST0003 = "e1fb5b00-1f02-4f63-99fa-7524b433ba47"
|
||||
DHST0001 = "660be601-080a-43ea-bf0f-adcf4596fa98"
|
||||
DHST0002 = "3f194cc7-464e-4581-9db2-179d42d3e40f"
|
||||
DHST0003 = "f45df555-2956-4a13-9004-ead047070b3d"
|
||||
DVST0001T = "fe11aeab-ff29-4bc8-9f75-c695c7e243e6"
|
||||
HST_LIST = [DHST0001,DHST0002,DHST0003]
|
||||
VST_LIST = [DVST0001,DVRT0001,DVCF0001]
|
||||
|
||||
# industry → 세로형 템플릿 매핑 (신규 세로형 템플릿 추가 시 여기에만 등록)
|
||||
VERTICAL_INDUSTRY_TEMPLATE_MAP: dict[str, str] = {
|
||||
"stay": DVST0001,
|
||||
"restaurant": DVRT0001,
|
||||
"cafe": DVCF0001,
|
||||
}
|
||||
|
||||
# 매핑에 없는 업종(salon, clinic 등)의 폴백 템플릿
|
||||
DEFAULT_VERTICAL_TEMPLATE = DVST0001
|
||||
VST_LIST = [DVST0001T]
|
||||
|
||||
SCENE_TRACK = 1
|
||||
AUDIO_TRACK = 2
|
||||
SUBTITLE_TRACK = 3
|
||||
KEYWORD_TRACK = 4
|
||||
|
||||
def select_template(orientation: OrientationType, industry: str | None = None) -> str:
|
||||
"""orientation과 industry에 따라 Creatomate 템플릿 ID를 선택합니다.
|
||||
|
||||
Args:
|
||||
orientation: 영상 방향 ("horizontal" 또는 "vertical")
|
||||
industry: 업종 분류 (stay|restaurant|cafe 등).
|
||||
세로형에서만 사용되며, 매핑에 없으면 기본 템플릿으로 폴백합니다.
|
||||
|
||||
Returns:
|
||||
선택된 템플릿 ID
|
||||
"""
|
||||
def select_template(orientation:OrientationType):
|
||||
if orientation == "horizontal":
|
||||
template_id = DHST0001
|
||||
elif orientation == "vertical":
|
||||
template_id = VERTICAL_INDUSTRY_TEMPLATE_MAP.get(industry, DEFAULT_VERTICAL_TEMPLATE)
|
||||
template_id = random.choice(VST_LIST)
|
||||
else:
|
||||
raise
|
||||
logger.info(f"[select_template] orientation={orientation}, industry={industry}, template_id={template_id}")
|
||||
logger.info(f"[select_template] orientation={orientation}, template_id={template_id}")
|
||||
return template_id
|
||||
|
||||
async def get_shared_client() -> httpx.AsyncClient:
|
||||
|
|
@ -315,22 +296,21 @@ class CreatomateService:
|
|||
def __init__(
|
||||
self,
|
||||
api_key: str | None = None,
|
||||
orientation: OrientationType = "vertical",
|
||||
industry: str | None = None,
|
||||
orientation: OrientationType = "vertical"
|
||||
):
|
||||
"""
|
||||
Args:
|
||||
api_key: Creatomate API 키 (Bearer token으로 사용)
|
||||
None일 경우 config에서 자동으로 가져옴
|
||||
orientation: 영상 방향 ("horizontal" 또는 "vertical", 기본값: "vertical")
|
||||
industry: 업종 분류 (stay|restaurant|cafe 등). 세로형 템플릿 선택에 사용되며,
|
||||
매핑에 없거나 None이면 기본 세로형 템플릿으로 폴백
|
||||
target_duration: 목표 영상 길이 (초)
|
||||
None일 경우 orientation에 해당하는 기본값 사용
|
||||
"""
|
||||
self.api_key = api_key or apikey_settings.CREATOMATE_API_KEY
|
||||
self.orientation = orientation
|
||||
|
||||
# orientation·industry에 따른 템플릿 설정 가져오기
|
||||
self.template_id = select_template(orientation, industry=industry)
|
||||
# orientation에 따른 템플릿 설정 가져오기
|
||||
self.template_id = select_template(orientation)
|
||||
self.headers = {
|
||||
"Content-Type": "application/json",
|
||||
"Authorization": f"Bearer {self.api_key}",
|
||||
|
|
@ -915,20 +895,6 @@ 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
|
||||
|
|
@ -942,7 +908,15 @@ 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"] - self._entrance_transition_offset(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"]
|
||||
else:
|
||||
track_maximum_duration[elem["track"]] = max(track_maximum_duration[elem["track"]], elem["time"] + elem["duration"])
|
||||
|
||||
|
|
@ -954,73 +928,6 @@ 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
|
||||
|
||||
@staticmethod
|
||||
def _scale_element(elem: dict, extend_rate: float) -> None:
|
||||
"""element 하나의 time/duration/animations를 extend_rate만큼 스케일링합니다.
|
||||
|
||||
composition의 자식 element까지 재귀적으로 처리합니다. 재귀하지 않으면
|
||||
부모 컴포지션(씬 슬롯)만 늘어나고 내부 leaf 이미지/애니메이션은 원래
|
||||
길이에서 끝나버려, 슬롯이 끝나기 전 컴포지션의 fill_color(회색)가
|
||||
노출되는 버그가 생긴다 — modify_element/parse_template_component_name과
|
||||
동일하게 composition을 재귀 순회해야 함.
|
||||
|
||||
오디오 판별은 track 번호가 아닌 type == "audio"로 한다. composition
|
||||
내부의 track 번호는 그 컴포지션 안에서만 유효한 로컬 레이어 인덱스라
|
||||
전역 AUDIO_TRACK/KEYWORD_TRACK 상수와 우연히 값이 겹칠 수 있다
|
||||
(실제 템플릿에도 다중 레이어 씬의 2번째/3번째 이미지가 track=2, track=4인
|
||||
경우가 존재함).
|
||||
"""
|
||||
if elem.get("type") == "audio":
|
||||
return
|
||||
|
||||
# "time"은 숫자 오프셋 대신 "end"(컴포지션 끝에 고정) 같은 특수 키워드일 수 있음 —
|
||||
# 이런 값은 렌더 시점에 (이미 스케일링된) duration 기준으로 재계산되므로 그대로 둔다.
|
||||
if "time" in elem and isinstance(elem["time"], (int, float)):
|
||||
elem["time"] = elem["time"] * extend_rate
|
||||
if "duration" in elem and isinstance(elem["duration"], (int, float)):
|
||||
elem["duration"] = elem["duration"] * extend_rate
|
||||
|
||||
for animation in elem.get("animations", []):
|
||||
anim_time = animation.get("time", 0)
|
||||
if isinstance(anim_time, (int, float)) and anim_time != 0:
|
||||
animation["time"] = anim_time * extend_rate
|
||||
if isinstance(animation.get("duration"), (int, float)):
|
||||
animation["duration"] = animation["duration"] * extend_rate
|
||||
|
||||
if elem.get("type") == "composition":
|
||||
for child in elem.get("elements", []):
|
||||
CreatomateService._scale_element(child, extend_rate)
|
||||
|
||||
def extend_template_duration(self, template: dict, target_duration: float) -> dict:
|
||||
"""템플릿의 duration을 target_duration으로 확장합니다."""
|
||||
# template["duration"] = target_duration + 0.5 # 늘린것보단 짧게
|
||||
|
|
@ -1031,10 +938,22 @@ class CreatomateService:
|
|||
|
||||
for elem in new_template["source"]["elements"]:
|
||||
try:
|
||||
if elem["track"] == AUDIO_TRACK : # audio track(최상위 음악 트랙)은 패스
|
||||
# if elem["type"] == "audio":
|
||||
# continue
|
||||
if elem["track"] == AUDIO_TRACK : # audio track은 패스
|
||||
continue
|
||||
|
||||
self._scale_element(elem, extend_rate)
|
||||
if "time" in elem:
|
||||
elem["time"] = elem["time"] * extend_rate
|
||||
if "duration" in elem:
|
||||
elem["duration"] = elem["duration"] * extend_rate
|
||||
|
||||
if "animations" not in elem:
|
||||
continue
|
||||
for animation in elem["animations"]:
|
||||
if animation.get("time", 0) != 0:
|
||||
animation["time"] = animation["time"] * extend_rate
|
||||
animation["duration"] = animation["duration"] * extend_rate
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"[extend_template_duration] Error processing element: {elem}, {e}"
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -1,5 +1,4 @@
|
|||
import asyncio
|
||||
import random
|
||||
import re
|
||||
from html import unescape
|
||||
from difflib import SequenceMatcher
|
||||
|
|
@ -20,52 +19,23 @@ class NvMapPwScraper():
|
|||
_win_width = 1280
|
||||
_win_height = 720
|
||||
_max_retry = 3
|
||||
_timeout = 30 # place id timeout threshold seconds
|
||||
_retry_delay_ms = 1500 # 검색 실패 후 재시도 전 대기시간 (네이버 요청 밀도 완화)
|
||||
|
||||
# UA·뷰포트·sec-ch-ua를 한 세트로 묶은 브라우저 프로필 후보군 (안티봇 핑거프린트 회피용)
|
||||
_UA_PROFILES = [
|
||||
{
|
||||
"user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36",
|
||||
"sec_ch_ua": '"Not?A_Brand";v="99", "Chromium";v="130", "Google Chrome";v="130"',
|
||||
"viewport": {"width": 1280, "height": 720},
|
||||
},
|
||||
{
|
||||
"user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Safari/537.36",
|
||||
"sec_ch_ua": '"Not?A_Brand";v="99", "Chromium";v="129", "Google Chrome";v="129"',
|
||||
"viewport": {"width": 1366, "height": 768},
|
||||
},
|
||||
{
|
||||
"user_agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36",
|
||||
"sec_ch_ua": '"Not?A_Brand";v="99", "Chromium";v="130", "Google Chrome";v="130"',
|
||||
"viewport": {"width": 1440, "height": 900},
|
||||
},
|
||||
{
|
||||
"user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
|
||||
"sec_ch_ua": '"Not?A_Brand";v="99", "Chromium";v="131", "Google Chrome";v="131"',
|
||||
"viewport": {"width": 1536, "height": 864},
|
||||
},
|
||||
]
|
||||
_current_profile = None
|
||||
_timeout = 60 # place id timeout threshold seconds
|
||||
|
||||
# instance var
|
||||
page = None
|
||||
|
||||
@classmethod
|
||||
def _pick_profile(cls):
|
||||
"""현재 프로필과 다른 프로필을 무작위로 선택한다 (후보가 1개뿐이면 그대로 반환)."""
|
||||
candidates = [p for p in cls._UA_PROFILES if p is not cls._current_profile]
|
||||
return random.choice(candidates or cls._UA_PROFILES)
|
||||
|
||||
@classmethod
|
||||
def default_context_builder(cls):
|
||||
cls._current_profile = cls._pick_profile()
|
||||
profile = cls._current_profile
|
||||
|
||||
context_builder_dict = {}
|
||||
context_builder_dict['viewport'] = dict(profile['viewport'])
|
||||
context_builder_dict['screen'] = dict(profile['viewport'])
|
||||
context_builder_dict['user_agent'] = profile['user_agent']
|
||||
context_builder_dict['viewport'] = {
|
||||
'width' : cls._win_width,
|
||||
'height' : cls._win_height
|
||||
}
|
||||
context_builder_dict['screen'] = {
|
||||
'width' : cls._win_width,
|
||||
'height' : cls._win_height
|
||||
}
|
||||
context_builder_dict['user_agent'] = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36"
|
||||
context_builder_dict['locale'] = 'ko-KR'
|
||||
context_builder_dict['timezone_id']='Asia/Seoul'
|
||||
|
||||
|
|
@ -81,14 +51,6 @@ class NvMapPwScraper():
|
|||
cls._context = await cls._browser.new_context(**cls.default_context_builder())
|
||||
cls.is_ready = True
|
||||
|
||||
@classmethod
|
||||
async def _recreate_context(cls):
|
||||
"""네이버 안티봇 차단이 의심될 때 브라우저 컨텍스트를 새로 생성해 세션/핑거프린트를 초기화한다."""
|
||||
old_context = cls._context
|
||||
cls._context = await cls._browser.new_context(**cls.default_context_builder())
|
||||
if old_context:
|
||||
await old_context.close()
|
||||
|
||||
GRAPHQL_URL = "https://pcmap-api.place.naver.com/graphql"
|
||||
|
||||
@classmethod
|
||||
|
|
@ -159,7 +121,7 @@ class NvMapPwScraper():
|
|||
}
|
||||
|
||||
results: list[dict | None] = []
|
||||
for idx, payload in enumerate(payloads, start=1):
|
||||
for payload in payloads:
|
||||
r = await page.evaluate(
|
||||
"""async (a) => {
|
||||
const [url, body, hdr] = a;
|
||||
|
|
@ -177,11 +139,7 @@ class NvMapPwScraper():
|
|||
[cls.GRAPHQL_URL, payload, headers],
|
||||
)
|
||||
if isinstance(r, dict) and "__err" in r:
|
||||
op = payload.get("operationName", "?")
|
||||
logger.warning(
|
||||
f"[NvMapPwScraper] graphql fetch 실패 "
|
||||
f"({idx}/{len(payloads)} {op}): {r['__err']}"
|
||||
)
|
||||
logger.warning(f"[NvMapPwScraper] graphql fetch 실패: {r['__err']}")
|
||||
results.append(None)
|
||||
else:
|
||||
results.append(r)
|
||||
|
|
@ -232,7 +190,7 @@ patchedGetter.apply(navigator);
|
|||
patchedGetter.toString();''')
|
||||
|
||||
await self.page.set_extra_http_headers({
|
||||
'sec-ch-ua': self._current_profile['sec_ch_ua']
|
||||
'sec-ch-ua': '\"Not?A_Brand\";v=\"99\", \"Chromium\";v=\"130\"'
|
||||
})
|
||||
await self.page.goto("http://google.com")
|
||||
|
||||
|
|
@ -326,110 +284,30 @@ patchedGetter.toString();''')
|
|||
logger.debug(f"[DEBUG] 목록 후보 {len(candidates)}개 추출")
|
||||
return candidates
|
||||
|
||||
@staticmethod
|
||||
def _parse_allsearch_candidates(body: dict) -> list[dict]:
|
||||
"""allSearch 응답 JSON에서 후보 목록을 추출한다."""
|
||||
place_list = (((body.get("result") or {}).get("place") or {}).get("list")) or []
|
||||
return [
|
||||
{
|
||||
"title": p.get("name") or "",
|
||||
"place_url": f"https://map.naver.com/p/entry/place/{p['id']}",
|
||||
"roadAddress": p.get("roadAddress") or "",
|
||||
"address": p.get("address") or "",
|
||||
}
|
||||
for p in place_list
|
||||
if p.get("id")
|
||||
]
|
||||
|
||||
def _select_best_candidate(self, candidates: list[dict], title: str, address: str) -> dict | None:
|
||||
"""이름 유사도(70%)와 주소 유사도(30%)를 함께 고려해 최적 후보를 선택한다.
|
||||
|
||||
주소 없이 업체명만으로 검색한 경우(3차 폴백)는 이름 유사도만 사용한다.
|
||||
"""
|
||||
if not candidates:
|
||||
return None
|
||||
|
||||
for c in candidates:
|
||||
name_score = self._similarity(title, self._clean_title(c["title"]))
|
||||
cand_addr = c.get("roadAddress") or c.get("address") or ""
|
||||
addr_score = self._similarity(address, cand_addr) if address and cand_addr else 0.0
|
||||
c["_name_score"] = name_score
|
||||
c["_addr_score"] = addr_score
|
||||
c["_total_score"] = name_score * 0.7 + addr_score * 0.3 if address else name_score
|
||||
|
||||
return max(candidates, key=lambda c: c["_total_score"])
|
||||
|
||||
async def _capture_allsearch(self, url: str, timeout_s: float = 5.0) -> list[dict]:
|
||||
"""검색 페이지가 자연 발생시키는 allSearch API 응답을 캡처해 후보 목록으로 변환한다.
|
||||
|
||||
pcmap iframe 렌더링을 기다릴 필요가 없어 경쟁 조건이 없고, 업체명 단독
|
||||
검색처럼 iframe 자체가 생성되지 않는 케이스에서도 동작한다.
|
||||
"""
|
||||
captured: list[dict] = []
|
||||
|
||||
def on_response(resp):
|
||||
# GET 요청의 실제 응답만 캡처 (OPTIONS preflight 등 오탐 방지)
|
||||
if "allSearch" in resp.url and resp.request.method == "GET":
|
||||
async def _consume():
|
||||
try:
|
||||
captured.append(await resp.json())
|
||||
except Exception:
|
||||
pass
|
||||
asyncio.create_task(_consume())
|
||||
|
||||
self.page.on("response", on_response)
|
||||
try:
|
||||
try:
|
||||
await self.goto_url(url, wait_until="domcontentloaded", timeout=self._timeout * 1000)
|
||||
except Exception:
|
||||
logger.error("[ERROR] Can't Finish domcontentloaded")
|
||||
|
||||
deadline = time.monotonic() + timeout_s
|
||||
while time.monotonic() < deadline:
|
||||
if "/place/" in self.page.url or captured:
|
||||
break
|
||||
await self.page.wait_for_timeout(200)
|
||||
finally:
|
||||
self.page.remove_listener("response", on_response)
|
||||
|
||||
if not captured:
|
||||
return []
|
||||
return self._parse_allsearch_candidates(captured[0])
|
||||
|
||||
async def _try_search(self, address: str, title: str) -> str | None:
|
||||
"""주어진 주소+업체명으로 검색해서 place URL을 반환한다. 실패 시 None."""
|
||||
encoded_query = parse.quote(f"{address} {title}".strip())
|
||||
url = f"https://map.naver.com/p/search/{encoded_query}"
|
||||
|
||||
# 1순위: allSearch API 응답 캡처 (iframe 렌더링 대기 불필요, 업체명 단독 검색도 지원)
|
||||
candidates = await self._capture_allsearch(url)
|
||||
try:
|
||||
await self.goto_url(url, wait_until="networkidle", timeout=self._timeout * 1000)
|
||||
except:
|
||||
if "/place/" in self.page.url:
|
||||
return self.page.url
|
||||
logger.error("[ERROR] Can't Finish networkidle")
|
||||
|
||||
if "/place/" in self.page.url:
|
||||
return self.page.url
|
||||
|
||||
candidates = await self._extract_candidates_from_list_page()
|
||||
if candidates:
|
||||
best = self._select_best_candidate(candidates, title, address)
|
||||
logger.info(
|
||||
f"[AUTO-SELECT] '{title}' → '{best['title']}' "
|
||||
f"(name={best['_name_score']:.2f}, addr={best['_addr_score']:.2f}) {best['place_url']}"
|
||||
best = max(
|
||||
candidates,
|
||||
key=lambda c: self._similarity(title, self._clean_title(c['title']))
|
||||
)
|
||||
return best['place_url']
|
||||
|
||||
# 2순위(안전망): allSearch 캡처 실패 시 기존 iframe HTML 파싱 방식으로 폴백
|
||||
logger.warning("[FALLBACK] allSearch 캡처 실패 → iframe 파싱 방식 시도")
|
||||
iframe_candidates = []
|
||||
for _ in range(3): # 500ms × 3 = 최대 1.5초
|
||||
if "/place/" in self.page.url:
|
||||
return self.page.url
|
||||
iframe_candidates = await self._extract_candidates_from_list_page()
|
||||
if iframe_candidates:
|
||||
break
|
||||
await self.page.wait_for_timeout(500)
|
||||
|
||||
if iframe_candidates:
|
||||
best = self._select_best_candidate(iframe_candidates, title, address)
|
||||
best_score = self._similarity(title, self._clean_title(best['title']))
|
||||
logger.info(
|
||||
f"[AUTO-SELECT-IFRAME] '{title}' → '{best['title']}' "
|
||||
f"(name={best['_name_score']:.2f}, addr={best['_addr_score']:.2f}) {best['place_url']}"
|
||||
f"[AUTO-SELECT] '{title}' → '{best['title']}' (score={best_score:.2f}) {best['place_url']}"
|
||||
)
|
||||
return best['place_url']
|
||||
|
||||
|
|
@ -460,27 +338,16 @@ patchedGetter.toString();''')
|
|||
# 2차 시도: 정제 주소 + 업체명
|
||||
refined = self._refine_address(address)
|
||||
if refined != address:
|
||||
await self.page.wait_for_timeout(self._retry_delay_ms)
|
||||
logger.info(f"[REFINE] 주소 정제: '{address}' → '{refined}'")
|
||||
result = await self._try_search(refined, title)
|
||||
if result:
|
||||
return result
|
||||
|
||||
# 3차 시도: 업체명만으로 검색
|
||||
await self.page.wait_for_timeout(self._retry_delay_ms)
|
||||
logger.info(f"[RETRY] 업체명만으로 재시도: '{title}'")
|
||||
result = await self._try_search("", title)
|
||||
if result:
|
||||
return result
|
||||
|
||||
# 1~3차 모두 실패 → 네이버 안티봇 차단 의심, 컨텍스트 재생성 후 원본 조건으로 1회 재시도
|
||||
logger.warning(f"[BLOCK-SUSPECTED] 3회 시도 모두 실패 → 컨텍스트 재생성 후 재시도: '{title}'")
|
||||
await self.page.close()
|
||||
await self._recreate_context()
|
||||
await self.create_page()
|
||||
result = await self._try_search(address, title)
|
||||
if result:
|
||||
return result
|
||||
|
||||
logger.error(f"[ERROR] Not found url for {selected}")
|
||||
return None
|
||||
|
|
|
|||
|
|
@ -52,12 +52,6 @@ query getAccommodation($id: String!, $deviceType: String) {
|
|||
conveniences
|
||||
visitorReviewsTotal
|
||||
}
|
||||
menus {
|
||||
name
|
||||
price
|
||||
description
|
||||
recommend
|
||||
}
|
||||
images { images { origin url } }
|
||||
}
|
||||
}"""
|
||||
|
|
@ -106,7 +100,6 @@ 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()
|
||||
|
|
@ -216,8 +209,7 @@ query getVisitorReviewStats($id: String!) {
|
|||
|
||||
# ── 실제 브라우저로 WTM 캡차 우회 ──
|
||||
data, stats_data, extra_photo_urls = await self._scrap_via_browser(place_id)
|
||||
# 편의시설은 HTML 페이지 파싱이라 GraphQL(WTM) 차단과 별개로 직접 시도 (best-effort, 실패 시 None)
|
||||
fac_data = await self._get_facility_string(place_id)
|
||||
fac_data = None # 편의시설(HTML)은 브라우저 경로에서 생략 (best-effort)
|
||||
self.scrap_type = "GraphQL-Browser"
|
||||
|
||||
self.rawdata = data
|
||||
|
|
@ -251,7 +243,6 @@ 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
|
||||
|
||||
|
|
|
|||
|
|
@ -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":
|
||||
|
|
|
|||
|
|
@ -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|general). 통합 프롬프트가 이 값으로 내부 분기")
|
||||
industry : str = Field(default="", description = "업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction). 통합 프롬프트가 이 값으로 내부 분기")
|
||||
space_type: list[str] = Field(list(SpaceType), description="공간적 정보를 가지는 태그 리스트")
|
||||
subject: list[str] = Field(list(Subject), description="피사체 정보를 가지는 태그 리스트")
|
||||
camera: list[str] = Field(list(Camera), description="카메라 정보를 가지는 태그 리스트")
|
||||
|
|
|
|||
|
|
@ -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|general). 통합 프롬프트가 이 값으로 내부 분기")
|
||||
industry : str = Field(default="", description = "업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction). 통합 프롬프트가 이 값으로 내부 분기")
|
||||
|
||||
# Output 정의
|
||||
class LyricPromptOutput(BaseModel):
|
||||
|
|
|
|||
|
|
@ -6,12 +6,7 @@ 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|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 = "메뉴/상품 목록 (네이버 지도 크롤링, 쉼표 구분, 없으면 빈 문자열). 실제 판매 품목 기반 셀링포인트 작성에 사용")
|
||||
industry : str = Field(default="", description = "업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction). 통합 프롬프트가 이 값으로 내부 분기")
|
||||
|
||||
# Output 정의
|
||||
class BrandIdentity(BaseModel):
|
||||
|
|
|
|||
|
|
@ -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|general). 통합 프롬프트가 이 값으로 내부 분기")
|
||||
industry : str = Field(default="", description = "업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction). 통합 프롬프트가 이 값으로 내부 분기")
|
||||
|
||||
# Output 정의
|
||||
class PitchingOutput(BaseModel):
|
||||
|
|
|
|||
|
|
@ -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|general). 통합 프롬프트가 이 값으로 내부 분기")
|
||||
industry : str = Field(default="", description = "업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction). 통합 프롬프트가 이 값으로 내부 분기")
|
||||
|
||||
# Output 정의
|
||||
class YTUploadPromptOutput(BaseModel):
|
||||
|
|
|
|||
|
|
@ -191,7 +191,6 @@ async def generate_video(
|
|||
store_address = project.detail_region_info
|
||||
brand_name = project.store_name
|
||||
region = project.region
|
||||
industry = project.industry
|
||||
|
||||
# MarketingIntel 조회
|
||||
marketing_result = await session.execute(
|
||||
|
|
@ -358,8 +357,7 @@ async def generate_video(
|
|||
f"[generate_video] Stage 2 START - Creatomate API - task_id: {task_id}"
|
||||
)
|
||||
creatomate_service = CreatomateService(
|
||||
orientation=orientation,
|
||||
industry=industry,
|
||||
orientation=orientation
|
||||
)
|
||||
logger.debug(
|
||||
f"[generate_video] Using template_id: {creatomate_service.template_id}, (song duration: {song_duration})"
|
||||
|
|
|
|||
|
|
@ -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]
|
||||
|
|
@ -9,12 +9,7 @@ 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,
|
||||
SCENE_TRACK,
|
||||
SUBTITLE_TRACK,
|
||||
KEYWORD_TRACK,
|
||||
)
|
||||
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
|
||||
|
|
@ -57,8 +52,10 @@ async def generate_creative_assets_background(
|
|||
|
||||
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)
|
||||
|
||||
# ── 프로젝트 / 마케팅 인텔 로드 ──────────────────────────────────
|
||||
# 템플릿 선택에 project.industry가 필요하므로 서비스 생성보다 먼저 로드
|
||||
async with BackgroundSessionLocal() as session:
|
||||
project_result = await session.execute(
|
||||
select(Project)
|
||||
|
|
@ -72,9 +69,6 @@ async def generate_creative_assets_background(
|
|||
)
|
||||
marketing_intelligence = marketing_result.scalar_one_or_none()
|
||||
|
||||
creatomate_service = CreatomateService(orientation=orientation, industry=project.industry)
|
||||
template = await creatomate_service.get_one_template_data(creatomate_service.template_id)
|
||||
|
||||
store_address = project.detail_region_info
|
||||
customer_name = project.store_name
|
||||
language = project.language or "Korean"
|
||||
|
|
@ -127,8 +121,6 @@ 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()
|
||||
|
|
@ -142,9 +134,8 @@ async def generate_creative_assets_background(
|
|||
)
|
||||
pitching_output_list = generated_subtitles.pitching_results
|
||||
|
||||
# LLM이 pitching_tag에 " | 화면: ..." 컨텍스트 접미를 그대로 되돌려줄 수 있으므로 제거
|
||||
subtitle_modifications = {
|
||||
pitching_output.pitching_tag.split(IMAGE_CONTEXT_SEP)[0].strip(): pitching_output.pitching_data
|
||||
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}")
|
||||
|
|
@ -179,113 +170,49 @@ 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 레이블 리스트에 배정 이미지 컨텍스트를 포함하여 반환합니다.
|
||||
|
||||
텍스트 레이어(SUBTITLE/KEYWORD_TRACK)와 이미지 컴포지션(SCENE_TRACK)의
|
||||
시간 범위를 계산해, 각 텍스트가 재생되는 동안 화면에 가장 오래 보이는
|
||||
(시간 겹침이 최대인) 컴포지션의 이미지 태그를 컨텍스트로 붙입니다.
|
||||
narrative_phase 토큰 매칭과 달리 phase 이름이 어긋나거나(자막 outro vs
|
||||
이미지 accent) 같은 phase에 씬이 여러 개여도 올바른 짝을 찾습니다.
|
||||
|
||||
겹치는 씬이 없거나 해당 씬에 배정된 이미지가 없으면(CTA 등) 컨텍스트를
|
||||
생략합니다. 렌더 시 extend_template_duration은 전 트랙 균등 배율이므로
|
||||
원본 템플릿 기준 겹침 관계는 스케일 후에도 유지됩니다.
|
||||
슬롯명(`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) 단계"
|
||||
"""
|
||||
comp_tags = _aggregate_tags_by_composition(template, assigned_image_tags, creatomate_service)
|
||||
if not comp_tags:
|
||||
return pitching_label_list
|
||||
# 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
|
||||
|
||||
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)
|
||||
if not phase_to_tag:
|
||||
return pitching_label_list
|
||||
|
||||
result = []
|
||||
for label in pitching_label_list:
|
||||
text_range = text_ranges.get(label)
|
||||
context_str = _context_for_label(text_range, scene_ranges, comp_tags) if text_range else ""
|
||||
# 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
|
||||
|
|
|
|||
Loading…
Reference in New Issue