475 lines
20 KiB
Python
475 lines
20 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 # 업체+방문자 사진 합산 최대 장수
|
|
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
|
|
}
|
|
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)
|
|
|
|
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"])
|
|
]
|
|
|
|
@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 = await self._scrap_via_browser(place_id)
|
|
fac_data = None # 편의시설(HTML)은 브라우저 경로에서 생략 (best-effort)
|
|
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)
|
|
self.owner_images = self._dedup_by_original(self._extract_origins(business.get("images") or {}))
|
|
|
|
# 방문자/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로 재조립할 것.)
|
|
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}장)"
|
|
)
|
|
else:
|
|
combined = self.owner_images
|
|
|
|
self.image_link_list = combined[: self.MAX_IMAGES]
|
|
self.base_info = data["data"]["business"]["base"]
|
|
self.facility_info = fac_data
|
|
self.voted_keyword_stats = stats_data
|
|
|
|
return
|
|
|
|
async def _scrap_via_browser(self, place_id: str) -> tuple[dict, list[dict] | None, list[str]]:
|
|
"""직접 호출이 WTM 캡차에 막힌 경우, 실제 브라우저로 GraphQL을 호출한다.
|
|
|
|
Returns:
|
|
(overview_data, review_stats_details, extra_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,
|
|
}
|
|
|
|
payloads = [overview_payload, stats_payload, interior_payload, exterior_payload, review_payload]
|
|
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)
|
|
extra_photo_urls = self._interleave(interior_urls, exterior_urls) + review_urls
|
|
logger.info(
|
|
f"[NvMapScraper] 보충 이미지 - 내부:{len(interior_urls)} 외부:{len(exterior_urls)} 리뷰:{len(review_urls)}"
|
|
)
|
|
|
|
logger.info(f"[NvMapScraper] 브라우저 폴백 SUCCESS - place_id: {place_id}")
|
|
return data, stats_data, extra_photo_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)
|