diff --git a/app/utils/nvMapPwScraper.py b/app/utils/nvMapPwScraper.py index 0fc13f3..9dd4a26 100644 --- a/app/utils/nvMapPwScraper.py +++ b/app/utils/nvMapPwScraper.py @@ -131,7 +131,8 @@ if (originalQuery) { cls, place_id: str, payloads: list[dict], - ) -> list[dict | None] | None: + capture_apollo: bool = False, + ) -> list[dict | None] | None | tuple[list[dict | None] | None, str | None]: """실제 브라우저로 네이버 WTM 안티봇 캡차를 통과해 GraphQL 쿼리를 실행한다. 네이버 pcmap GraphQL은 두 헤더를 검사한다: @@ -144,12 +145,19 @@ if (originalQuery) { Args: place_id: 네이버 place ID payloads: GraphQL POST 본문 목록 + capture_apollo: True면 place 페이지에 인라인된 __APOLLO_STATE__ + JSON 문자열도 함께 캡처해 (results, apollo_json) 튜플로 반환. + (GraphQL base가 노출하지 않는 homepages 등 SSR 캐시 전용 필드용) Returns: payload별 파싱 JSON 목록(실패 항목은 None). 토큰 캡처 실패 시 None. + capture_apollo=True면 (위 결과, apollo_json 또는 None) 튜플. """ + def _ret(results, apollo=None): + return (results, apollo) if capture_apollo else results + if not cls.is_ready: logger.warning("[NvMapPwScraper] fetch_graphql: scraper가 초기화되지 않았습니다") - return None + return _ret(None) page = await cls._new_stealth_page() captured: dict = {} @@ -185,7 +193,17 @@ if (originalQuery) { if not captured.get("tok"): logger.warning("[NvMapPwScraper] WTM 토큰 캡처 실패") - return None + return _ret(None) + + apollo_json: str | None = None + if capture_apollo: + try: + apollo_json = await page.evaluate( + "() => { try { return JSON.stringify(window.__APOLLO_STATE__ || null); }" + " catch (e) { return null; } }" + ) + except Exception as e: + logger.warning(f"[NvMapPwScraper] __APOLLO_STATE__ 캡처 실패: {e}") headers = { "Content-Type": "application/json", @@ -220,11 +238,11 @@ if (originalQuery) { results.append(None) else: results.append(r) - return results + return _ret(results, apollo_json) except Exception as e: logger.error(f"[NvMapPwScraper] fetch_graphql 오류: {e}") - return None + return _ret(None) finally: await page.close() diff --git a/app/utils/nvMapScraper.py b/app/utils/nvMapScraper.py index a46bdc6..bfc172c 100644 --- a/app/utils/nvMapScraper.py +++ b/app/utils/nvMapScraper.py @@ -56,10 +56,6 @@ query getAccommodation($id: String!, $deviceType: String) { microReviews conveniences visitorReviewsTotal - homepages { - repr { url landingUrl isDeadUrl type } - etc { url landingUrl isDeadUrl type } - } } menus { name @@ -261,9 +257,11 @@ query getVisitorReviewStats($id: String!) { # self.scrap_type = "GraphQL-Browser" # ── 실제 브라우저로 WTM 캡차 우회 ── - data, stats_data, extra_photo_urls, biz_photo_urls = await self._scrap_via_browser(place_id) + data, stats_data, extra_photo_urls, biz_photo_urls, homepage_url = await self._scrap_via_browser(place_id) # 편의시설은 HTML 페이지 파싱이라 GraphQL(WTM) 차단과 별개로 직접 시도 (best-effort, 실패 시 None) - fac_data = await self._get_facility_string(place_id) + # 홈페이지 링크는 브라우저 캡처가 실패한 경우에만 HTML 경로로 보충한다. + fac_data, html_homepage = await self._get_facility_and_homepage(place_id) + homepage_url = homepage_url or html_homepage self.scrap_type = "GraphQL-Browser" self.rawdata = data @@ -302,29 +300,43 @@ query getVisitorReviewStats($id: String!) { self.facility_info = fac_data self.voted_keyword_stats = stats_data self.menu_info = business.get("menus") or None - self.official_site_url = self._extract_official_site_url(self.base_info) + self.official_site_url = homepage_url return @staticmethod - def _extract_official_site_url(base_info: dict | None) -> str | None: - """base.homepages에서 살아있는 링크 하나를 고른다 (대표 repr 우선, 그 다음 etc 순서). + def _extract_homepage_from_html(html: str) -> str | None: + """플레이스 페이지 HTML의 __APOLLO_STATE__에서 홈페이지 링크를 추출한다. - 네이버 플레이스의 홈페이지 항목은 자체 홈페이지 외에 인스타그램/블로그 - 등일 수도 있다 — 업체가 대표로 등록한 링크를 그대로 신뢰한다. + GraphQL placeDetail(base)는 homepages 필드를 노출하지 않으므로(400), + SSR 페이지에 인라인된 Apollo 캐시의 "homepages" 객체를 직접 파싱한다. + 대표(repr) 링크 우선, 죽은 링크(isDeadUrl)는 제외. 홈페이지 항목은 + 자체 홈페이지 외에 인스타그램/블로그 등일 수도 있다 — 업체가 대표로 + 등록한 링크를 그대로 신뢰한다. """ - homepages = (base_info or {}).get("homepages") or {} - candidates = [homepages.get("repr"), *(homepages.get("etc") or [])] - for item in candidates: - if item and item.get("url") and not item.get("isDeadUrl"): - return item["url"] - return None + decoder = json.JSONDecoder() + search_from = 0 + while True: + idx = html.find('"homepages":', search_from) + if idx == -1: + return None + search_from = idx + 1 + try: + homepages, _ = decoder.raw_decode(html, idx + len('"homepages":')) + except ValueError: + continue + if not isinstance(homepages, dict): + continue + candidates = [homepages.get("repr"), *(homepages.get("etc") or [])] + for item in candidates: + if isinstance(item, dict) and item.get("url") and not item.get("isDeadUrl"): + return item["url"] - async def _scrap_via_browser(self, place_id: str) -> tuple[dict, list[dict] | None, list[dict], list[dict]]: + async def _scrap_via_browser(self, place_id: str) -> tuple[dict, list[dict] | None, list[dict], list[dict], str | None]: """직접 호출이 WTM 캡차에 막힌 경우, 실제 브라우저로 GraphQL을 호출한다. Returns: - (overview_data, review_stats_details, extra_photo_urls, biz_photo_urls) + (overview_data, review_stats_details, extra_photo_urls, biz_photo_urls, homepage_url) Raises: GraphQLException: 브라우저 폴백마저 실패한 경우 @@ -402,10 +414,14 @@ query getVisitorReviewStats($id: String!) { payloads = [overview_payload, stats_payload, interior_payload, exterior_payload, review_payload, *biz_payloads] MAX_RETRY = 3 results = None + apollo_json: str | None = None last_error: Exception | None = None for attempt in range(1, MAX_RETRY + 1): try: - results = await NvMapPwScraper.fetch_graphql(place_id, payloads) + results, captured_apollo = await NvMapPwScraper.fetch_graphql( + place_id, payloads, capture_apollo=True + ) + apollo_json = apollo_json or captured_apollo except Exception as e: last_error = e logger.warning(f"[NvMapScraper] 브라우저 폴백 시도 {attempt}/{MAX_RETRY} 오류: {e}") @@ -459,8 +475,17 @@ query getVisitorReviewStats($id: String!) { f"리뷰:{len(review_urls)} / 업체(biz):{len(biz_urls)}" ) - logger.info(f"[NvMapScraper] 브라우저 폴백 SUCCESS - place_id: {place_id}") - return data, stats_data, extra_photo_urls, biz_urls + # 홈페이지 링크: GraphQL base는 homepages를 노출하지 않으므로(400), + # 브라우저가 로드한 place 페이지의 __APOLLO_STATE__ JSON에서 추출한다. + homepage_url = ( + self._extract_homepage_from_html(apollo_json) if apollo_json else None + ) + + logger.info( + f"[NvMapScraper] 브라우저 폴백 SUCCESS - place_id: {place_id}, " + f"homepage: {homepage_url or '없음'}" + ) + return data, stats_data, extra_photo_urls, biz_urls, homepage_url async def _call_get_accommodation(self, place_id: str) -> dict: """GraphQL API를 호출하여 숙소 정보를 가져옵니다. @@ -541,29 +566,39 @@ query getVisitorReviewStats($id: String!) { logger.warning(f"[NvMapScraper] Failed to get review stats: {e}") return None - async def _get_facility_string(self, place_id: str) -> str | None: - """장소 페이지에서 편의시설 정보를 크롤링합니다. 숙소, 음식점 순으로 시도합니다. + async def _get_facility_and_homepage(self, place_id: str) -> tuple[str | None, str | None]: + """장소 페이지에서 편의시설 정보와 홈페이지 링크를 크롤링합니다. 숙소, 음식점 순으로 시도합니다. Args: place_id: 네이버 지도 장소 ID Returns: - 편의시설 정보 문자열 또는 None + (편의시설 정보 문자열 또는 None, 홈페이지 링크 또는 None) """ + facility: str | None = None + homepage: str | None = None place_types = ["place", "accommodation", "restaurant"] try: async with aiohttp.ClientSession() as session: for place_type in place_types: url = f"https://pcmap.place.naver.com/{place_type}/{place_id}/home" async with session.get(url, headers=self._get_request_headers()) as response: - soup = bs4.BeautifulSoup(await response.read(), "html.parser") + raw = await response.read() + if homepage is None: + homepage = self._extract_homepage_from_html( + raw.decode("utf-8", errors="ignore") + ) + if facility is None: + soup = bs4.BeautifulSoup(raw, "html.parser") c_elem = soup.find("span", "place_blind", string="편의") if c_elem: - return c_elem.parent.parent.find("div").string - return None + facility = c_elem.parent.parent.find("div").string + if facility is not None and homepage is not None: + break + return facility, homepage except Exception as e: - logger.warning(f"[NvMapScraper] Failed to get facility info: {e}") - return None + logger.warning(f"[NvMapScraper] Failed to get facility/homepage info: {e}") + return facility, homepage # if __name__ == "__main__":