566 lines
26 KiB
Python
566 lines
26 KiB
Python
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 # 업체 사진이 이 수 미만일 때 방문자 사진으로 보충
|
||
# 방문자 사진 보충 시의 합산 상한 (필터링·정책상 제한).
|
||
# 업체 제공 사진은 필터링 면제 대상이므로 이 상한과 무관하게 전량 사용한다.
|
||
MAX_IMAGES = 50
|
||
BIZ_PAGE_SIZE = 20 # getPhotoViewerItems 'biz' 커서의 페이지당 사진 수
|
||
BIZ_MAX_PAGES = 5 # 업체 사진 페이지네이션 안전 상한 (5×20=100장)
|
||
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)
|
||
|
||
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 = await self._scrap_via_browser(place_id)
|
||
# 편의시설은 HTML 페이지 파싱이라 GraphQL(WTM) 차단과 별개로 직접 시도 (best-effort, 실패 시 None)
|
||
fac_data = await self._get_facility_string(place_id)
|
||
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 상한은 방문자 사진이 섞이는 보충 경로에만 적용하며(필터링·정책상 제한),
|
||
# 업체 제공 사진만으로 구성되는 경우는 전량 사용한다.
|
||
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
|
||
|
||
return
|
||
|
||
async def _scrap_via_browser(self, place_id: str) -> tuple[dict, list[dict] | None, list[dict], list[dict]]:
|
||
"""직접 호출이 WTM 캡차에 막힌 경우, 실제 브라우저로 GraphQL을 호출한다.
|
||
|
||
Returns:
|
||
(overview_data, review_stats_details, extra_photo_urls, biz_photo_urls)
|
||
|
||
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")이라 이전 응답 없이도 다음 페이지를 미리 요청할 수 있고, 범위를
|
||
# 벗어난 페이지는 빈 목록을 반환하므로 블라인드 요청해도 안전하다.
|
||
# 첫 배치에 3페이지를 실어 보내고, 마지막 페이지가 가득 차 있으면(=더 있을 수
|
||
# 있음) 추가 배치를 반복 요청해 업체 사진을 전량 수집한다.
|
||
BIZ_INITIAL_PAGES = 3
|
||
biz_payloads = [
|
||
self._build_biz_payload(place_id, page) for page in range(BIZ_INITIAL_PAGES)
|
||
]
|
||
|
||
payloads = [overview_payload, stats_payload, interior_payload, exterior_payload, review_payload, *biz_payloads]
|
||
MAX_RETRY = 3
|
||
results = None
|
||
last_error: Exception | None = None
|
||
for attempt in range(1, MAX_RETRY + 1):
|
||
try:
|
||
results = await NvMapPwScraper.fetch_graphql(place_id, payloads)
|
||
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 = list(results[5:])
|
||
biz_urls = [url for page_result in biz_pages for url in self._extract_photo_viewer_urls(page_result)]
|
||
# 마지막 페이지가 가득 차 있으면 다음 배치를 계속 요청해 업체 사진을 전량 수집한다
|
||
# (업체 사진은 필터링 면제 대상이라 많을수록 영상 소재 다양성이 좋아짐).
|
||
# 후속 배치 실패는 warning만 남기고 이미 수집한 분량으로 진행한다 (best-effort).
|
||
next_page = BIZ_INITIAL_PAGES
|
||
while (
|
||
biz_pages
|
||
and self._raw_photo_count(biz_pages[-1]) >= self.BIZ_PAGE_SIZE
|
||
and next_page < self.BIZ_MAX_PAGES
|
||
):
|
||
chunk_pages = range(next_page, min(next_page + 3, self.BIZ_MAX_PAGES))
|
||
chunk_payloads = [self._build_biz_payload(place_id, p) for p in chunk_pages]
|
||
try:
|
||
chunk_results = await NvMapPwScraper.fetch_graphql(place_id, chunk_payloads)
|
||
except Exception as e:
|
||
logger.warning(f"[NvMapScraper] 업체 사진 추가 페이지 수집 실패(수집분으로 진행): {e}")
|
||
break
|
||
if not chunk_results:
|
||
logger.warning("[NvMapScraper] 업체 사진 추가 페이지 응답 없음(수집분으로 진행)")
|
||
break
|
||
biz_pages = [r for r in chunk_results if r is not None]
|
||
biz_urls += [url for page_result in biz_pages for url in self._extract_photo_viewer_urls(page_result)]
|
||
next_page += len(chunk_payloads)
|
||
|
||
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)}"
|
||
)
|
||
|
||
logger.info(f"[NvMapScraper] 브라우저 폴백 SUCCESS - place_id: {place_id}")
|
||
return data, stats_data, extra_photo_urls, biz_urls
|
||
|
||
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_string(self, place_id: str) -> str | None:
|
||
"""장소 페이지에서 편의시설 정보를 크롤링합니다. 숙소, 음식점 순으로 시도합니다.
|
||
|
||
Args:
|
||
place_id: 네이버 지도 장소 ID
|
||
|
||
Returns:
|
||
편의시설 정보 문자열 또는 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")
|
||
c_elem = soup.find("span", "place_blind", string="편의")
|
||
if c_elem:
|
||
return c_elem.parent.parent.find("div").string
|
||
return None
|
||
except Exception as e:
|
||
logger.warning(f"[NvMapScraper] Failed to get facility info: {e}")
|
||
return None
|
||
|
||
|
||
# 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)
|