네이버 크롤링 우회 코드 추가
parent
b908408b4a
commit
8415e1c3a7
|
|
@ -51,6 +51,111 @@ class NvMapPwScraper():
|
|||
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:
|
||||
raise Exception("nvMapScraper is not initiated")
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
"""장소 페이지에서 편의시설 정보를 크롤링합니다. 숙소, 음식점 순으로 시도합니다.
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue