fix(crawling): homepages를 GraphQL 대신 __APOLLO_STATE__에서 추출

placeDetail(base)의 homepages 필드는 스키마에 없어 전체 크롤링이
400으로 실패했다(컨테이너 검증 완료). 쿼리에서 제거하고, WTM 우회용
브라우저가 이미 로드하는 place 페이지의 __APOLLO_STATE__ JSON을 캡처해
대표(repr) 링크를 추출한다. 브라우저 캡처 실패 시 HTML 직접 파싱으로
보충한다(직접 요청은 429로 막히는 환경이 있어 best-effort).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CVSeUT2pEmeHc6hG6gVXUS
This commit is contained in:
hbyang 2026-08-21 11:30:01 +09:00
parent 87cb3eb3cb
commit 6197fcdc91
2 changed files with 88 additions and 35 deletions

View File

@ -131,7 +131,8 @@ if (originalQuery) {
cls, cls,
place_id: str, place_id: str,
payloads: list[dict], payloads: list[dict],
) -> list[dict | None] | None: capture_apollo: bool = False,
) -> list[dict | None] | None | tuple[list[dict | None] | None, str | None]:
"""실제 브라우저로 네이버 WTM 안티봇 캡차를 통과해 GraphQL 쿼리를 실행한다. """실제 브라우저로 네이버 WTM 안티봇 캡차를 통과해 GraphQL 쿼리를 실행한다.
네이버 pcmap GraphQL은 두 헤더를 검사한다: 네이버 pcmap GraphQL은 두 헤더를 검사한다:
@ -144,12 +145,19 @@ if (originalQuery) {
Args: Args:
place_id: 네이버 place ID place_id: 네이버 place ID
payloads: GraphQL POST 본문 목록 payloads: GraphQL POST 본문 목록
capture_apollo: True면 place 페이지에 인라인된 __APOLLO_STATE__
JSON 문자열도 함께 캡처해 (results, apollo_json) 튜플로 반환.
(GraphQL base가 노출하지 않는 homepages 등 SSR 캐시 전용 필드용)
Returns: Returns:
payload별 파싱 JSON 목록(실패 항목은 None). 토큰 캡처 실패 시 None. 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: if not cls.is_ready:
logger.warning("[NvMapPwScraper] fetch_graphql: scraper가 초기화되지 않았습니다") logger.warning("[NvMapPwScraper] fetch_graphql: scraper가 초기화되지 않았습니다")
return None return _ret(None)
page = await cls._new_stealth_page() page = await cls._new_stealth_page()
captured: dict = {} captured: dict = {}
@ -185,7 +193,17 @@ if (originalQuery) {
if not captured.get("tok"): if not captured.get("tok"):
logger.warning("[NvMapPwScraper] WTM 토큰 캡처 실패") 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 = { headers = {
"Content-Type": "application/json", "Content-Type": "application/json",
@ -220,11 +238,11 @@ if (originalQuery) {
results.append(None) results.append(None)
else: else:
results.append(r) results.append(r)
return results return _ret(results, apollo_json)
except Exception as e: except Exception as e:
logger.error(f"[NvMapPwScraper] fetch_graphql 오류: {e}") logger.error(f"[NvMapPwScraper] fetch_graphql 오류: {e}")
return None return _ret(None)
finally: finally:
await page.close() await page.close()

View File

@ -56,10 +56,6 @@ query getAccommodation($id: String!, $deviceType: String) {
microReviews microReviews
conveniences conveniences
visitorReviewsTotal visitorReviewsTotal
homepages {
repr { url landingUrl isDeadUrl type }
etc { url landingUrl isDeadUrl type }
}
} }
menus { menus {
name name
@ -261,9 +257,11 @@ query getVisitorReviewStats($id: String!) {
# self.scrap_type = "GraphQL-Browser" # self.scrap_type = "GraphQL-Browser"
# ── 실제 브라우저로 WTM 캡차 우회 ── # ── 실제 브라우저로 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) # 편의시설은 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.scrap_type = "GraphQL-Browser"
self.rawdata = data self.rawdata = data
@ -302,29 +300,43 @@ query getVisitorReviewStats($id: String!) {
self.facility_info = fac_data self.facility_info = fac_data
self.voted_keyword_stats = stats_data self.voted_keyword_stats = stats_data
self.menu_info = business.get("menus") or None 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 return
@staticmethod @staticmethod
def _extract_official_site_url(base_info: dict | None) -> str | None: def _extract_homepage_from_html(html: str) -> str | None:
"""base.homepages에서 살아있는 링크 하나를 고른다 (대표 repr 우선, 그 다음 etc 순서). """플레이스 페이지 HTML의 __APOLLO_STATE__에서 홈페이지 링크를 추출한다.
네이버 플레이스의 홈페이지 항목은 자체 홈페이지 외에 인스타그램/블로그 GraphQL placeDetail(base)는 homepages 필드를 노출하지 않으므로(400),
등일 수도 있다 — 업체가 대표로 등록한 링크를 그대로 신뢰한다. SSR 페이지에 인라인된 Apollo 캐시의 "homepages" 객체를 직접 파싱한다.
대표(repr) 링크 우선, 죽은 링크(isDeadUrl)는 제외. 홈페이지 항목은
자체 홈페이지 외에 인스타그램/블로그 등일 수도 있다 — 업체가 대표로
등록한 링크를 그대로 신뢰한다.
""" """
homepages = (base_info or {}).get("homepages") or {} decoder = json.JSONDecoder()
candidates = [homepages.get("repr"), *(homepages.get("etc") or [])] search_from = 0
for item in candidates: while True:
if item and item.get("url") and not item.get("isDeadUrl"): idx = html.find('"homepages":', search_from)
return item["url"] if idx == -1:
return None 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을 호출한다. """직접 호출이 WTM 캡차에 막힌 경우, 실제 브라우저로 GraphQL을 호출한다.
Returns: 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: Raises:
GraphQLException: 브라우저 폴백마저 실패한 경우 GraphQLException: 브라우저 폴백마저 실패한 경우
@ -402,10 +414,14 @@ query getVisitorReviewStats($id: String!) {
payloads = [overview_payload, stats_payload, interior_payload, exterior_payload, review_payload, *biz_payloads] payloads = [overview_payload, stats_payload, interior_payload, exterior_payload, review_payload, *biz_payloads]
MAX_RETRY = 3 MAX_RETRY = 3
results = None results = None
apollo_json: str | None = None
last_error: Exception | None = None last_error: Exception | None = None
for attempt in range(1, MAX_RETRY + 1): for attempt in range(1, MAX_RETRY + 1):
try: 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: except Exception as e:
last_error = e last_error = e
logger.warning(f"[NvMapScraper] 브라우저 폴백 시도 {attempt}/{MAX_RETRY} 오류: {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)}" f"리뷰:{len(review_urls)} / 업체(biz):{len(biz_urls)}"
) )
logger.info(f"[NvMapScraper] 브라우저 폴백 SUCCESS - place_id: {place_id}") # 홈페이지 링크: GraphQL base는 homepages를 노출하지 않으므로(400),
return data, stats_data, extra_photo_urls, biz_urls # 브라우저가 로드한 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: async def _call_get_accommodation(self, place_id: str) -> dict:
"""GraphQL API를 호출하여 숙소 정보를 가져옵니다. """GraphQL API를 호출하여 숙소 정보를 가져옵니다.
@ -541,29 +566,39 @@ query getVisitorReviewStats($id: String!) {
logger.warning(f"[NvMapScraper] Failed to get review stats: {e}") logger.warning(f"[NvMapScraper] Failed to get review stats: {e}")
return None 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: Args:
place_id: 네이버 지도 장소 ID place_id: 네이버 지도 장소 ID
Returns: Returns:
편의시설 정보 문자열 또는 None (편의시설 정보 문자열 또는 None, 홈페이지 링크 또는 None)
""" """
facility: str | None = None
homepage: str | None = None
place_types = ["place", "accommodation", "restaurant"] place_types = ["place", "accommodation", "restaurant"]
try: try:
async with aiohttp.ClientSession() as session: async with aiohttp.ClientSession() as session:
for place_type in place_types: for place_type in place_types:
url = f"https://pcmap.place.naver.com/{place_type}/{place_id}/home" url = f"https://pcmap.place.naver.com/{place_type}/{place_id}/home"
async with session.get(url, headers=self._get_request_headers()) as response: 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="편의") c_elem = soup.find("span", "place_blind", string="편의")
if c_elem: if c_elem:
return c_elem.parent.parent.find("div").string facility = c_elem.parent.parent.find("div").string
return None if facility is not None and homepage is not None:
break
return facility, homepage
except Exception as e: except Exception as e:
logger.warning(f"[NvMapScraper] Failed to get facility info: {e}") logger.warning(f"[NvMapScraper] Failed to get facility/homepage info: {e}")
return None return facility, homepage
# if __name__ == "__main__": # if __name__ == "__main__":