From 8415e1c3a712d0e9c5f08686f413b660f746f3d2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EA=B9=80=EC=84=B1=EA=B2=BD?= Date: Mon, 29 Jun 2026 10:36:53 +0900 Subject: [PATCH] =?UTF-8?q?=EB=84=A4=EC=9D=B4=EB=B2=84=20=ED=81=AC?= =?UTF-8?q?=EB=A1=A4=EB=A7=81=20=EC=9A=B0=ED=9A=8C=20=EC=BD=94=EB=93=9C=20?= =?UTF-8?q?=EC=B6=94=EA=B0=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/utils/nvMapPwScraper.py | 105 ++++++++++++++++++++++++++++ app/utils/nvMapScraper.py | 133 ++++++++++++++++++++++++++++++++++-- 2 files changed, 232 insertions(+), 6 deletions(-) diff --git a/app/utils/nvMapPwScraper.py b/app/utils/nvMapPwScraper.py index 9268125..0872dbe 100644 --- a/app/utils/nvMapPwScraper.py +++ b/app/utils/nvMapPwScraper.py @@ -50,6 +50,111 @@ class NvMapPwScraper(): if not cls._context: cls._context = await cls._browser.new_context(**cls.default_context_builder()) cls.is_ready = True + + GRAPHQL_URL = "https://pcmap-api.place.naver.com/graphql" + _PLACE_TYPES = ("restaurant", "accommodation") + + @classmethod + async def fetch_graphql( + cls, + place_id: str, + payloads: list[dict], + place_type: str | None = None, + ) -> list[dict | None] | None: + """실제 브라우저로 네이버 WTM 안티봇 캡차를 통과해 GraphQL 쿼리를 실행한다. + + 네이버 pcmap GraphQL은 두 헤더를 검사한다: + - x-wtm-graphql : base64({"arg": place_id, "type": ..., "source": "place"}) + - x-wtm-ncaptcha-token : 캡차 JS가 생성 (요청마다 발급, 직접 생성 불가) + 둘 다 없으면 405(캡차)로 막힌다. 따라서 place 페이지를 실제로 로드해 + 페이지가 자연 발생시키는 GraphQL 요청에서 두 헤더를 캡처한 뒤, + 같은 토큰으로 우리 쿼리들을 in-page fetch로 실행한다. + + Args: + place_id: 네이버 place ID + payloads: GraphQL POST 본문 목록 + place_type: "restaurant" | "accommodation" (없으면 둘 다 시도) + + Returns: + payload별 파싱 JSON 목록(실패 항목은 None). 토큰 캡처 실패 시 None. + """ + if not cls.is_ready: + logger.warning("[NvMapPwScraper] fetch_graphql: scraper가 초기화되지 않았습니다") + return None + + page = await cls._context.new_page() + captured: dict = {} + + def on_request(req): + if "/graphql" in req.url and req.method == "POST" and "tok" not in captured: + tok = req.headers.get("x-wtm-ncaptcha-token") + if tok: + captured["tok"] = tok + captured["wtm"] = req.headers.get("x-wtm-graphql") + + page.on("request", on_request) + + try: + types = [place_type] if place_type in cls._PLACE_TYPES else list(cls._PLACE_TYPES) + for t in types: + try: + await page.goto( + f"https://pcmap.place.naver.com/{t}/{place_id}/home", + wait_until="domcontentloaded", + timeout=30000, + ) + except Exception as e: + logger.debug(f"[NvMapPwScraper] goto {t} 실패: {e}") + continue + # 페이지가 GraphQL 요청을 보내며 토큰이 헤더에 실릴 때까지 대기 (최대 ~9초) + for _ in range(30): + if captured.get("tok"): + break + await page.wait_for_timeout(300) + if captured.get("tok"): + logger.info(f"[NvMapPwScraper] WTM 토큰 캡처 성공 (type={t})") + break + + if not captured.get("tok"): + logger.warning("[NvMapPwScraper] WTM 토큰 캡처 실패") + return None + + headers = { + "Content-Type": "application/json", + "x-wtm-graphql": captured["wtm"], + "x-wtm-ncaptcha-token": captured["tok"], + } + + results: list[dict | None] = [] + for payload in payloads: + r = await page.evaluate( + """async (a) => { + const [url, body, hdr] = a; + try { + const r = await fetch(url, { + method: 'POST', + headers: hdr, + body: JSON.stringify(body), + credentials: 'include', + }); + if (r.status !== 200) return {__err: r.status}; + return await r.json(); + } catch (e) { return {__err: String(e)}; } + }""", + [cls.GRAPHQL_URL, payload, headers], + ) + if isinstance(r, dict) and "__err" in r: + logger.warning(f"[NvMapPwScraper] graphql fetch 실패: {r['__err']}") + results.append(None) + else: + results.append(r) + return results + + except Exception as e: + logger.error(f"[NvMapPwScraper] fetch_graphql 오류: {e}") + return None + finally: + await page.close() def __init__(self): if not self.is_ready: diff --git a/app/utils/nvMapScraper.py b/app/utils/nvMapScraper.py index df0b4b8..ee6985e 100644 --- a/app/utils/nvMapScraper.py +++ b/app/utils/nvMapScraper.py @@ -55,6 +55,21 @@ query getAccommodation($id: String!, $deviceType: String) { } }""" + REVIEW_STATS_QUERY: str = """ +query getVisitorReviewStats($id: String!) { + visitorReviewStats(input: {businessId: $id}) { + analysis { + votedKeyword { + details { + code + displayName + count + } + } + } + } +}""" + DEFAULT_HEADERS: dict = { "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", "Referer": "https://map.naver.com/", @@ -73,6 +88,7 @@ query getAccommodation($id: String!, $deviceType: String) { self.base_info: dict | None = None self.facility_info: str | None = None self.place_type: str | None = None # 크롤링으로 감지된 업종 (accommodation | restaurant) + self.voted_keyword_stats: list[dict] | None = None # 키워드 투표 집계 (displayName, count) def _get_request_headers(self) -> dict: headers = self.DEFAULT_HEADERS.copy() @@ -108,9 +124,31 @@ query getAccommodation($id: String!, $deviceType: String) { async def scrap(self): place_id = await self.parse_url() - data = await self._call_get_accommodation(place_id) + + # ── 빠른 경로: 직접 aiohttp 호출 (현재 비활성화) ── + # 데이터센터 IP에서는 네이버 WTM 안티봇이 대부분 405(캡차)로 차단해 + # 시간만 버리므로 비활성화하고 곧바로 브라우저 경로를 탄다. + # 네이버가 차단을 완화하거나 다른 IP에서 운영해 직접 호출 성공률이 + # 높아지면 아래 try 블록을 되살려 빠른 경로를 우선 시도하면 된다. + # try: + # data, fac_data, stats_data = await asyncio.gather( + # self._call_get_accommodation(place_id), + # self._get_facility_string(place_id), + # self._call_get_review_stats(place_id), + # ) + # self.scrap_type = "GraphQL" + # except (GraphQLException, CrawlingTimeoutException) as e: + # logger.info(f"[NvMapScraper] 직접 호출 실패({e}) → 브라우저 폴백 시도") + # data, stats_data = await self._scrap_via_browser(place_id) + # fac_data = None # 편의시설(HTML)은 폴백 경로에서 생략 (best-effort) + # self.scrap_type = "GraphQL-Browser" + + # ── 실제 브라우저로 WTM 캡차 우회 ── + data, stats_data = await self._scrap_via_browser(place_id) + fac_data = None # 편의시설(HTML)은 브라우저 경로에서 생략 (best-effort) + self.scrap_type = "GraphQL-Browser" + self.rawdata = data - fac_data = await self._get_facility_string(place_id) # Naver 기준임, 구글 등 다른 데이터 소스의 경우 고유 Identifier 사용할 것. self.place_id = self.data_source_identifier + place_id self.rawdata["facilities"] = fac_data @@ -120,10 +158,58 @@ query getAccommodation($id: String!, $deviceType: String) { ] self.base_info = data["data"]["business"]["base"] self.facility_info = fac_data - self.scrap_type = "GraphQL" + self.voted_keyword_stats = stats_data return + async def _scrap_via_browser(self, place_id: str) -> tuple[dict, list[dict] | None]: + """직접 호출이 WTM 캡차에 막힌 경우, 실제 브라우저로 GraphQL을 호출한다. + + Returns: + (overview_data, review_stats_details) + + Raises: + GraphQLException: 브라우저 폴백마저 실패한 경우 + """ + from app.utils.nvMapPwScraper import NvMapPwScraper + + overview_payload = { + "operationName": "getAccommodation", + "variables": {"id": place_id, "deviceType": "pc"}, + "query": self.OVERVIEW_QUERY, + } + stats_payload = { + "operationName": "getVisitorReviewStats", + "variables": {"id": place_id}, + "query": self.REVIEW_STATS_QUERY, + } + + try: + results = await NvMapPwScraper.fetch_graphql( + place_id, [overview_payload, stats_payload], self.place_type + ) + except Exception as e: + logger.error(f"[NvMapScraper] 브라우저 폴백 오류: {e}") + raise GraphQLException(f"브라우저 폴백 실패: {e}") + + if not results or results[0] is None or "data" not in results[0]: + raise GraphQLException("브라우저 폴백 크롤링 실패 (WTM 토큰 또는 응답 없음)") + + data = results[0] + stats_data = None + stats_raw = results[1] if len(results) > 1 else None + if stats_raw: + stats_data = ( + stats_raw.get("data", {}) + .get("visitorReviewStats", {}) + .get("analysis", {}) + .get("votedKeyword", {}) + .get("details") + ) or None + + logger.info(f"[NvMapScraper] 브라우저 폴백 SUCCESS - place_id: {place_id}") + return data, stats_data + async def _call_get_accommodation(self, place_id: str) -> dict: """GraphQL API를 호출하여 숙소 정보를 가져옵니다. @@ -152,13 +238,14 @@ query getAccommodation($id: String!, $deviceType: String) { async with session.post( self.GRAPHQL_URL, data=json_payload, - headers=self._get_request_headers() + headers=self._get_request_headers(), ) as response: if response.status == 200: logger.info(f"[NvMapScraper] SUCCESS - place_id: {place_id}") return await response.json() - # 실패 상태 코드 + # 405/429 등 실패: 네이버 WTM 안티봇 캡차 차단 (데이터센터 IP/지문 기반). + # 직접 호출로는 통과 불가 → scrap()에서 브라우저 폴백으로 전환. logger.error(f"[NvMapScraper] Failed with status {response.status} - place_id: {place_id}") raise GraphQLException( f"Request failed with status {response.status}" @@ -167,11 +254,45 @@ query getAccommodation($id: String!, $deviceType: String) { except (TimeoutError, asyncio.TimeoutError): logger.error(f"[NvMapScraper] Timeout - place_id: {place_id}") raise CrawlingTimeoutException(f"Request timed out after {self.REQUEST_TIMEOUT}s") - except aiohttp.ClientError as e: logger.error(f"[NvMapScraper] Client error: {e}") raise GraphQLException(f"Client error: {e}") + async def _call_get_review_stats(self, place_id: str) -> list[dict] | None: + """방문자 키워드 투표 집계를 가져옵니다. + + Returns: + [{"code": ..., "displayName": ..., "count": ...}, ...] 또는 None + """ + payload = { + "operationName": "getVisitorReviewStats", + "variables": {"id": place_id}, + "query": self.REVIEW_STATS_QUERY, + } + timeout = aiohttp.ClientTimeout(total=self.REQUEST_TIMEOUT) + try: + async with aiohttp.ClientSession(timeout=timeout) as session: + async with session.post( + self.GRAPHQL_URL, + json=payload, + headers=self._get_request_headers(), + ) as response: + if response.status != 200: + logger.warning(f"[NvMapScraper] review stats failed: {response.status}") + return None + result = await response.json() + details = ( + result.get("data", {}) + .get("visitorReviewStats", {}) + .get("analysis", {}) + .get("votedKeyword", {}) + .get("details") + ) + return details or None + except Exception as e: + logger.warning(f"[NvMapScraper] Failed to get review stats: {e}") + return None + async def _get_facility_string(self, place_id: str) -> str | None: """장소 페이지에서 편의시설 정보를 크롤링합니다. 숙소, 음식점 순으로 시도합니다.