import asyncio import json import re import aiohttp import bs4 from app.utils.logger import get_logger from config import crawler_settings # 로거 설정 logger = get_logger("scraper") class GraphQLException(Exception): """GraphQL 요청 실패 시 발생하는 예외""" pass class URLNotFoundException(Exception): """Place ID 발견 불가능 시 발생하는 예외""" pass class CrawlingTimeoutException(Exception): """크롤링 타임아웃 시 발생하는 예외""" pass class NvMapScraper: """네이버 지도 GraphQL API 스크래퍼 네이버 지도에서 숙소/장소 정보를 크롤링합니다. """ GRAPHQL_URL: str = "https://pcmap-api.place.naver.com/graphql" REQUEST_TIMEOUT = 120 # 초 data_source_identifier = "nv" SUPPLEMENT_THRESHOLD = 30 # 업체 사진이 이 수 미만일 때 방문자 사진으로 보충 # 방문자 사진 보충 시의 합산 상한 (필터링·정책상 제한). # 업체 제공 사진은 필터링 면제 대상이므로 이 상한과 무관하게 수집분을 전부 사용한다 # (수집 자체는 BIZ_MAX_PAGES가 상한 — 초과 시 warning 로그 발생). MAX_IMAGES = 50 BIZ_PAGE_SIZE = 20 # getPhotoViewerItems 'biz' 커서의 페이지당 사진 수 BIZ_MAX_PAGES = 5 # 업체 사진 수집 상한 (5×20=100장, 도달 시 warning) OVERVIEW_QUERY: str = """ query getAccommodation($id: String!, $deviceType: String) { business: placeDetail(input: {id: $id, isNx: true, deviceType: $deviceType}) { base { id name category roadAddress address phone virtualPhone microReviews conveniences visitorReviewsTotal } menus { name price description recommend } images { images { origin url } } } }""" PHOTO_VIEWER_QUERY: str = """ query getPhotoViewerItems($input: PhotoViewerInput) { photoViewer(input: $input) { photos { originalUrl } } }""" 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/", "Origin": "https://map.naver.com", "Content-Type": "application/json", } def __init__(self, url: str, cookies: str | None = None): self.url = url self.cookies = ( cookies if cookies is not None else crawler_settings.NAVER_COOKIES ) self.scrap_type: str | None = None self.rawdata: dict | None = None self.image_link_list: list[dict] | None = None # [{"preview": str, "original": str}] self.owner_images: list[dict] | None = None # 업체 등록 사진 (마케팅 필터 제외 대상) self.extra_photo_urls: list[dict] | None = None # 방문자/AI View 보충 사진 (마케팅 필터 대상) 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) self.official_site_url: str | None = None # 업체 공식 링크 (base.homepages 대표 URL) def _get_request_headers(self) -> dict: headers = self.DEFAULT_HEADERS.copy() if self.cookies: headers["Cookie"] = self.cookies return headers _GIF_PATTERN = re.compile(r"\.gif(?:/|$)", re.IGNORECASE) @classmethod def _is_gif_url(cls, url: str) -> bool: """URL의 확장자가 gif인지 확인한다. 네이버 CDN은 리사이즈 파라미터를 확장자 뒤에 경로 세그먼트로 붙인다 (예: ".../3982000924.gif/500x500") — endswith(".gif")로는 잡히지 않으므로 경로 세그먼트 단위로 ".gif" 뒤에 "/" 또는 문자열 끝이 오는지 검사한다. """ return bool(cls._GIF_PATTERN.search(url)) @staticmethod def _extract_origins(image_node: dict) -> list[dict]: """GraphQL 이미지 노드({ images: [{origin, url}, ...] })에서 {preview, original} 목록을 추출한다. origin = 원본(업로드용), url = CDN 리사이즈(미리보기용) """ items = image_node.get("images") or [] return [ {"preview": item.get("url") or item["origin"], "original": item["origin"]} for item in items if item.get("origin") and not NvMapScraper._is_gif_url(item["origin"]) ] @staticmethod def _extract_photo_viewer_urls(photo_viewer_raw: dict | None) -> list[dict]: """getPhotoViewerItems 응답에서 {preview, original} 목록을 추출한다. photoViewer는 썸네일 필드가 없으므로 preview = original 동일하게 사용. """ photos = ((photo_viewer_raw or {}).get("data") or {}).get("photoViewer") or {} items = photos.get("photos") or [] return [ {"preview": p["originalUrl"], "original": p["originalUrl"]} for p in items if p.get("originalUrl") and not NvMapScraper._is_gif_url(p["originalUrl"]) ] @classmethod def _build_biz_payload(cls, place_id: str, page: int) -> dict: """'biz' 커서(업체 등록 사진)의 page번째(0-base) 페이지 요청 payload를 만든다.""" start_index = page * cls.BIZ_PAGE_SIZE cursor: dict = {"id": "biz"} if start_index > 0: cursor.update({ "startIndex": start_index, "hasNext": True, "lastCursor": str(start_index), }) return { "operationName": "getPhotoViewerItems", "variables": { "input": { "businessId": place_id, "cursors": [cursor], "dateRange": "", "excludeAuthorIds": [], "excludeClipIds": [], "excludeSection": [], } }, "query": cls.PHOTO_VIEWER_QUERY, } @staticmethod def _raw_photo_count(photo_viewer_raw: dict | None) -> int: """getPhotoViewerItems 응답의 사진 수(gif 필터링 전 원본 기준)를 반환한다. 페이지네이션 계속 여부 판단용 — gif가 걸러진 뒤의 수로 판단하면 마지막 페이지가 아닌데도 조기 종료할 수 있어 원본 개수를 사용한다. """ photos = ((photo_viewer_raw or {}).get("data") or {}).get("photoViewer") or {} return len(photos.get("photos") or []) @staticmethod def _dedup_by_original(images: list[dict]) -> list[dict]: """{"preview","original"} dict 리스트를 original URL 기준으로 순서를 유지한 채 중복 제거한다.""" seen: set[str] = set() result = [] for img in images: original = img.get("original") if original and original not in seen: seen.add(original) result.append(img) return result @staticmethod def _interleave(a: list, b: list) -> list: """두 리스트를 1:1 교차 병합한다. 남은 항목은 뒤에 붙인다.""" result = [x for pair in zip(a, b) for x in pair] longer = a[len(b):] if len(a) > len(b) else b[len(a):] return result + longer async def parse_url(self) -> str: """URL에서 place ID를 추출합니다. 단축 URL인 경우 실제 URL로 변환합니다.""" place_pattern = r"/place/(\d+)" # URL에 place가 없는 경우 단축 URL 처리 if "place" not in self.url: if "naver.me" in self.url: async with aiohttp.ClientSession() as session: async with session.get(self.url) as response: self.url = str(response.url) else: raise URLNotFoundException("This URL does not contain a place ID") match = re.search(place_pattern, self.url) if not match: # place.naver.com/{type}/{id} 형식 (예: m.place.naver.com/restaurant/11667161) type_match = re.search(r"place\.naver\.com/([a-zA-Z]+)/(\d+)", self.url) if type_match: return type_match.group(2) if not match: raise URLNotFoundException("Failed to parse place ID from URL") return match[1] async def scrap(self): place_id = await self.parse_url() # ── 빠른 경로: 직접 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, extra_photo_urls, biz_photo_urls, homepage_url = await self._scrap_via_browser(place_id) # 편의시설은 HTML 페이지 파싱이라 GraphQL(WTM) 차단과 별개로 직접 시도 (best-effort, 실패 시 None) # 홈페이지 링크는 브라우저 캡처가 실패한 경우에만 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 # Naver 기준임, 구글 등 다른 데이터 소스의 경우 고유 Identifier 사용할 것. self.place_id = self.data_source_identifier + place_id self.rawdata["facilities"] = fac_data business = data["data"]["business"] # 업체 등록 이미지 (마케팅 적합성 필터 제외 대상 — 자체 dedup). # placeDetail.images가 대표 사진 일부만 반환하는 경우가 있어 biz 커서 결과를 병합한다. self.owner_images = self._dedup_by_original( self._extract_origins(business.get("images") or {}) + biz_photo_urls ) # 방문자/AI View 보충 사진 (마케팅 적합성 필터 대상). owner에 이미 있는 원본은 제외. owner_originals = {img["original"] for img in self.owner_images} self.extra_photo_urls = self._dedup_by_original( [img for img in extra_photo_urls if img["original"] not in owner_originals] ) # 업체 사진이 임계값 미만이면 내부/외부/리뷰 사진으로 보충 # (필터링 없는 기본 조립. 마케팅 적합성 필터를 적용하려면 호출측에서 # owner_images / extra_photo_urls를 직접 사용해 image_filter.assemble_images로 재조립할 것.) # MAX_IMAGES 상한은 방문자 사진이 섞이는 보충 경로에만 적용하며(필터링·정책상 제한), # 업체 제공 사진만으로 구성되는 경우는 수집분(최대 BIZ_MAX_PAGES 페이지)을 전부 사용한다. if len(self.owner_images) < self.SUPPLEMENT_THRESHOLD: combined = self._dedup_by_original(self.owner_images + self.extra_photo_urls) logger.info( f"[NvMapScraper] 업체 사진 {len(self.owner_images)}장 < {self.SUPPLEMENT_THRESHOLD}장 " f"→ 보충 사진 {len(self.extra_photo_urls)}장 추가 (합산 {len(combined)}장, 상한 {self.MAX_IMAGES}장)" ) self.image_link_list = combined[: self.MAX_IMAGES] else: self.image_link_list = list(self.owner_images) 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 self.official_site_url = homepage_url return @staticmethod def _extract_homepage_from_html(html: str) -> str | None: """플레이스 페이지 HTML의 __APOLLO_STATE__에서 홈페이지 링크를 추출한다. GraphQL placeDetail(base)는 homepages 필드를 노출하지 않으므로(400), SSR 페이지에 인라인된 Apollo 캐시의 "homepages" 객체를 직접 파싱한다. 대표(repr) 링크 우선, 죽은 링크(isDeadUrl)는 제외. 홈페이지 항목은 자체 홈페이지 외에 인스타그램/블로그 등일 수도 있다 — 업체가 대표로 등록한 링크를 그대로 신뢰한다. """ 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], str | None]: """직접 호출이 WTM 캡차에 막힌 경우, 실제 브라우저로 GraphQL을 호출한다. Returns: (overview_data, review_stats_details, extra_photo_urls, biz_photo_urls, homepage_url) 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, } interior_payload = { "operationName": "getPhotoViewerItems", "variables": { "input": { "businessId": place_id, "cursors": [{"id": "aiView"}], "filter": "AI View", "subFilter": "INTERIOR", "dateRange": "", "excludeAuthorIds": [], "excludeClipIds": [], "excludeSection": [], } }, "query": self.PHOTO_VIEWER_QUERY, } exterior_payload = { "operationName": "getPhotoViewerItems", "variables": { "input": { "businessId": place_id, "cursors": [{"id": "aiView"}], "filter": "AI View", "subFilter": "EXTERIOR", "dateRange": "", "excludeAuthorIds": [], "excludeClipIds": [], "excludeSection": [], } }, "query": self.PHOTO_VIEWER_QUERY, } review_payload = { "operationName": "getPhotoViewerItems", "variables": { "input": { "businessId": place_id, "cursors": [{"id": "placeReview"}], "dateRange": "", "excludeAuthorIds": [], "excludeClipIds": [], "excludeSection": [], } }, "query": self.PHOTO_VIEWER_QUERY, } # 업체 등록 사진. placeDetail.images는 대표 사진 일부(1~수 장)만 반환하는 # 경우가 있어, 사진 탭이 실제 사용하는 'biz' 커서로 별도 조회해 보강한다. # biz 커서는 페이지당 BIZ_PAGE_SIZE(20)장이며 lastCursor가 단순 인덱스 문자열 # ("20", "40")이라 이전 응답 없이도 모든 페이지를 미리 만들 수 있고, 범위를 # 벗어난 페이지는 빈 목록을 반환하므로 블라인드 요청해도 안전하다. # 따라서 상한(BIZ_MAX_PAGES)까지의 전 페이지를 첫 배치에 한꺼번에 실어 보낸다 # — 추가 왕복(페이지 재탐색 + WTM 토큰 재캡처)이 없고, 개별 페이지 실패가 # 이후 페이지 수집을 막지 못한다. biz_payloads = [ self._build_biz_payload(place_id, page) for page in range(self.BIZ_MAX_PAGES) ] 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, 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}") if attempt < MAX_RETRY: await asyncio.sleep(1) continue if results and results[0] is not None and "data" in results[0]: break logger.warning(f"[NvMapScraper] 브라우저 폴백 시도 {attempt}/{MAX_RETRY} 실패(405 등), 재시도") results = None if attempt < MAX_RETRY: await asyncio.sleep(1) if not results or results[0] is None or "data" not in results[0]: if last_error: raise GraphQLException(f"브라우저 폴백 실패: {last_error}") raise GraphQLException("브라우저 폴백 크롤링 실패 (WTM 토큰 또는 응답 없음)") data = results[0] stats_data = None stats_raw = results[1] if len(results) > 1 else None if stats_raw: _vrs = (stats_raw.get("data") or {}).get("visitorReviewStats") or {} _analysis = _vrs.get("analysis") or {} _voted = _analysis.get("votedKeyword") or {} stats_data = _voted.get("details") or None interior_urls = self._extract_photo_viewer_urls(results[2] if len(results) > 2 else None) exterior_urls = self._extract_photo_viewer_urls(results[3] if len(results) > 3 else None) review_urls = self._extract_photo_viewer_urls(results[4] if len(results) > 4 else None) biz_pages = results[5:] biz_urls = [url for page_result in biz_pages for url in self._extract_photo_viewer_urls(page_result)] # 개별 페이지 실패(None)는 해당 20장만 누락되고 나머지 페이지는 영향 없다 (best-effort). failed_biz_pages = sum(1 for r in biz_pages if r is None) if failed_biz_pages: logger.warning( f"[NvMapScraper] 업체 사진 {failed_biz_pages}개 페이지 응답 실패 " f"— 페이지당 최대 {self.BIZ_PAGE_SIZE}장 누락 가능 (수집분으로 진행)" ) # 마지막 페이지까지 가득 차 있으면 상한 밖에 사진이 더 있을 수 있다. if biz_pages and self._raw_photo_count(biz_pages[-1]) >= self.BIZ_PAGE_SIZE: logger.warning( f"[NvMapScraper] 업체 사진이 수집 상한(BIZ_MAX_PAGES={self.BIZ_MAX_PAGES}, " f"{self.BIZ_MAX_PAGES * self.BIZ_PAGE_SIZE}장)까지 가득 참 — 초과분은 수집되지 않음" ) extra_photo_urls = self._interleave(interior_urls, exterior_urls) + review_urls logger.info( f"[NvMapScraper] 보충 이미지 - 내부:{len(interior_urls)} 외부:{len(exterior_urls)} " f"리뷰:{len(review_urls)} / 업체(biz):{len(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를 호출하여 숙소 정보를 가져옵니다. Args: place_id: 네이버 지도 장소 ID Returns: GraphQL 응답 데이터 Raises: GraphQLException: API 호출 실패 시 CrawlingTimeoutException: 타임아웃 발생 시 """ payload = { "operationName": "getAccommodation", "variables": {"id": place_id, "deviceType": "pc"}, "query": self.OVERVIEW_QUERY, } json_payload = json.dumps(payload) timeout = aiohttp.ClientTimeout(total=self.REQUEST_TIMEOUT) try: logger.info(f"[NvMapScraper] Requesting place_id: {place_id}") async with aiohttp.ClientSession(timeout=timeout) as session: async with session.post( self.GRAPHQL_URL, data=json_payload, 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}" ) 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() _vrs = (result.get("data") or {}).get("visitorReviewStats") or {} _analysis = _vrs.get("analysis") or {} _voted = _analysis.get("votedKeyword") or {} return _voted.get("details") or None except Exception as e: logger.warning(f"[NvMapScraper] Failed to get review stats: {e}") return None async def _get_facility_and_homepage(self, place_id: str) -> tuple[str | None, str | None]: """장소 페이지에서 편의시설 정보와 홈페이지 링크를 크롤링합니다. 숙소, 음식점 순으로 시도합니다. Args: place_id: 네이버 지도 장소 ID Returns: (편의시설 정보 문자열 또는 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: 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: 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/homepage info: {e}") return facility, homepage # if __name__ == "__main__": # import asyncio # url = "https://map.naver.com/p/search/%EC%8A%A4%ED%85%8C%EC%9D%B4%EB%A8%B8%EB%AD%84/place/1133638931?c=14.70,0,0,0,dh&placePath=/photo?businessCategory=pension&fromPanelNum=2&locale=ko&searchText=%EC%8A%A4%ED%85%8C%EC%9D%B4%EB%A8%B8%EB%AD%84&svcName=map_pcv5×tamp=202512191123&fromPanelNum=2&locale=ko&searchText=%EC%8A%A4%ED%85%8C%EC%9D%B4%EB%A8%B8%EB%AD%84&svcName=map_pcv5×tamp=202512191007&from=map&entry=bmp&filterType=%EC%97%85%EC%B2%B4&businessCategory=pension" # scraper = NvMapScraper(url) # asyncio.run(scraper.scrap()) # print(scraper.image_link_list) # print(len(scraper.image_link_list) if scraper.image_link_list else 0) # print(scraper.base_info)