diff --git a/app/song/api/routers/v1/song.py b/app/song/api/routers/v1/song.py index f593042..506c073 100644 --- a/app/song/api/routers/v1/song.py +++ b/app/song/api/routers/v1/song.py @@ -35,6 +35,33 @@ 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}", @@ -413,12 +440,14 @@ async def get_song_status( clips_data = response_data.get("sunoData") or [] if clips_data: - # 첫 번째 클립(clips[0])의 audioUrl과 duration 사용 - first_clip = clips_data[0] + # 생성된 클립(보통 2개) 중 목표 길이(1분)에 가장 가까운 클립 선택 (동일하면 첫 번째) + first_clip = _select_clip_by_duration(clips_data) audio_url = first_clip.get("audioUrl") clip_duration = first_clip.get("duration") logger.debug( - f"[get_song_status] Using first clip - id: {first_clip.get('id')}, audio_url: {audio_url}, duration: {clip_duration}" + 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]}" ) if audio_url: diff --git a/app/utils/creatomate.py b/app/utils/creatomate.py index 61e3573..e081a8e 100644 --- a/app/utils/creatomate.py +++ b/app/utils/creatomate.py @@ -964,6 +964,43 @@ class CreatomateService: 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 # 늘린것보단 짧게 @@ -974,22 +1011,10 @@ class CreatomateService: for elem in new_template["source"]["elements"]: try: - # if elem["type"] == "audio": - # continue - if elem["track"] == AUDIO_TRACK : # audio track은 패스 + if elem["track"] == AUDIO_TRACK : # audio track(최상위 음악 트랙)은 패스 continue - - 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 + self._scale_element(elem, extend_rate) except Exception as e: logger.error( f"[extend_template_duration] Error processing element: {elem}, {e}" diff --git a/app/utils/nvMapPwScraper.py b/app/utils/nvMapPwScraper.py index e369b96..039e85f 100644 --- a/app/utils/nvMapPwScraper.py +++ b/app/utils/nvMapPwScraper.py @@ -121,7 +121,7 @@ class NvMapPwScraper(): } results: list[dict | None] = [] - for payload in payloads: + for idx, payload in enumerate(payloads, start=1): r = await page.evaluate( """async (a) => { const [url, body, hdr] = a; @@ -139,7 +139,11 @@ class NvMapPwScraper(): [cls.GRAPHQL_URL, payload, headers], ) if isinstance(r, dict) and "__err" in r: - logger.warning(f"[NvMapPwScraper] graphql fetch 실패: {r['__err']}") + op = payload.get("operationName", "?") + logger.warning( + f"[NvMapPwScraper] graphql fetch 실패 " + f"({idx}/{len(payloads)} {op}): {r['__err']}" + ) results.append(None) else: results.append(r) @@ -284,30 +288,110 @@ 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 = 8.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}" - 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") - + # 1순위: allSearch API 응답 캡처 (iframe 렌더링 대기 불필요, 업체명 단독 검색도 지원) + candidates = await self._capture_allsearch(url) if "/place/" in self.page.url: return self.page.url - candidates = await self._extract_candidates_from_list_page() if candidates: - best = max( - candidates, - key=lambda c: self._similarity(title, self._clean_title(c['title'])) - ) - best_score = self._similarity(title, self._clean_title(best['title'])) + best = self._select_best_candidate(candidates, title, address) logger.info( - f"[AUTO-SELECT] '{title}' → '{best['title']}' (score={best_score:.2f}) {best['place_url']}" + f"[AUTO-SELECT] '{title}' → '{best['title']}' " + f"(name={best['_name_score']:.2f}, addr={best['_addr_score']:.2f}) {best['place_url']}" + ) + 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) + logger.info( + f"[AUTO-SELECT-IFRAME] '{title}' → '{best['title']}' " + f"(name={best['_name_score']:.2f}, addr={best['_addr_score']:.2f}) {best['place_url']}" ) return best['place_url']