개요
이번에 새로운 프로젝트를 시작하게 되었습니다.
프로젝트 주제는 "현재 자신의 상황에 맞는 식당 추천 시스템"입니다.
이 시스템을 만들기 위해 해당 식당에 대한 분위기, 특징과 같은 정보가 필요합니다.
단순한 식당 정보만 가지고는 이러한 정보를 수집하기가 어렵다고 판단했고 식당에 대한 리뷰 데이터를 기반으로 식당에 대한 정보를 수집하기로 결정했습니다.
이렇게 수집한 리뷰 데이터를 기반으로 카테고리 별로 분류하여 임베딩 벡터로 만들어 식당 특징 정보를 만드려고 합니다.
이를 위해 식당 정보와 리뷰 정보가 필요했기 때문에 이전 포스팅에서 식당 정보 크롤링을 구현했습니다.
https://growth-coder.tistory.com/355
식당 추천 시스템 개발기 #1 - 네이버 식당 정보 크롤링
개요이번에 새로운 프로젝트를 시작하게 되었습니다. 프로젝트 주제는 "현재 자신의 상황에 맞는 식당 추천 시스템"입니다. 식당 추천 시스템을 만들기 위해 네이버 지도의 식당 데이터를 수집
growth-coder.tistory.com
이번 포스팅에서는 식당 리뷰 정보를 크롤링하는 코드를 작성해보려고 합니다.
네이버 지도 리뷰 페이지 분석
이전 포스팅에서 네이버 지도 식당 페이지를 분석한 결과는 아래와 같습니다.
식당 리스트인 빨간색 테두리와 식당 상세 정보인 파란색 테두리는 모두 iframe으로 이루어져 있습니다.
그리고 식당 상세 정보 페이지는 iframe이 아닌 별도의 페이지로 접근할 수도 있습니다.
이 식당 상세 정보 페이지에서 리뷰 정보 페이지를 분석해보겠습니다.
식당 리뷰 페이지의 리뷰들은 식당 페이지와 다르게 페이지네이션이 존재하지 않고 오직 더보기 버튼으로 리뷰를 불러올 수 있습니다.
스크롤을 하고 계속 더보기 버튼을 누르면서 리뷰를 불러올 수 있습니다.
즉, 저의 리뷰 크롤링 전략은 계속 스크롤링을 하며 더보기 버튼을 누르면서 새로 로딩되는 리뷰 데이터를 수집하는 것입니다.
🚧 첫 번째 어려움 : 페이지네이션
우선 일반적으로 무한 스크롤 방식은 페이지네이션 방식을 사용하기 때문에 가장 먼저 네트워크 탭에서 API 요청을 확인해보았습니다.
네이버의 경우 graphql을 사용하고 있고 상당히 복잡한 request body를 보내고 있습니다.
한 번 해당 요청 그대로 요청을 보내보았으나 식당 API와 마찬가지로 접근을 제한하고 있습니다.
API를 직접 보내는 방식은 아무래도 어렵다고 판단이 되어 식당 크롤링과 마찬가지로 가상 브라우저를 활용하기로 결정했습니다.
✅ 해결책: 가상 브라우저 활용
식당 크롤링을 할 때 playwright를 사용하였는데요.
식당의 경우 페이지네이션과 무한 스크롤 방식을 사용하기 때문에 여러 페이지를 동시에 수집하기 위해 라이브러리 자체에서 비동기 함수를 제공해주는 playwright를 선택했습니다.
그러나 리뷰의 경우 페이지네이션이 존재하지 않고 스크롤 방식을 활용하기 때문에 비동기의 장점을 제대로 활용하기 어려웠습니다.
물론 여러 식당에 대한 리뷰를 수집하는 방식으로 비동기의 장점을 활용할 수 있겠지만 리뷰가 많은 식당의 경우 수천개의 리뷰를 가지고 있습니다.
"공덕역 식당"을 검색했을 때 기준으로 페이지마다 약 50개의 식당이 존재하고 5페이지 정도 존재하기 때문에 5페이지 동시 수집의 경우 컴퓨팅 리소스를 많이 잡아먹지 않았습니다.
하지만 리뷰의 경우 많으면 수천개의 리뷰를 가진 식당이 수백개 존재하기 때문에 식당 크롤링 보다는 컴퓨팅 리소스를 많이 잡아먹을 것이라고 생각했습니다.
그래서 리뷰 크롤링의 경우 동기 방식으로 한 번에 하나의 식당에 대한 리뷰를 크롤링하기로 결정했습니다.
이렇게 보면 굳이 playwright를 사용해야 하는 이유는 없을 것 같지만 그래도 식당 크롤링과 같은 기술을 사용하는 편이 크롤링 시스템 관리에 편할 것 같아서 리뷰 크롤링 또한 playwright를 사용하기로 결정했습니다.
🚧 두 번째 어려움 : 실시간 리뷰 데이터 수집
저는 리뷰 데이터가 새로 로드될 때마다 직접 element를 선택해서 정보를 가져오는 방식을 선택했습니다.
더보기 버튼을 눌러 새로운 리뷰 데이터가 로드되었을 때 이전에 수집했던 내용은 제외하고 새로운 리뷰 데이터를 수집해야 했습니다.
제가 선택한 방식은 수집한 리뷰 데이터는 DOM에서 제거하는 방식입니다.
✅ 해결책: 수집한 리뷰 DOM에서 제거
네이버 식당 리뷰의 경우 한 번 더보기 버튼을 클릭할 때마다 10개의 새로운 리뷰 정보가 로드됩니다.
새로운 리뷰가 로드되어 모두 수집하고 나면 수집한 리뷰를 모두 DOM에서 제거하고 더보기 버튼을 눌러 새로운 리뷰 정보를 크롤링하는 방식을 사용했습니다.
🚧 세 번째 어려움 : 리뷰 중복 처리
저는 식당에 대한 사용자들의 리뷰를 주기적으로 수집하여 계속 식당에 대한 임베딩 정보를 갱신하려고 했습니다.
그러기 위해서 이미 수집한 데이터를 중복해서 수집하는 일을 방지해야 했습니다.
가장 쉬운 방법은 고유 값을 사용해서 중복 처리를 하는 방법이었지만 리뷰 페이지에서는 리뷰마다 고유 값을 찾기가 어려웠습니다.
✅ 해결책: 작성자, 작성 날짜, 내용 기반으로 해시 적용
그렇기 때문에 고유 값을 직접 만들기로 결정했고 작성자, 작성 날짜, 내용을 바탕으로 해시 알고리즘을 적용하여 고유 값을 만들었습니다.
이미 기존 리뷰 데이터를 저장한 저장소에서 고유 값을 가져오고 set에 넣고 중복 체크를 하는 방식을 적용했습니다.
그리고 리뷰를 크롤링 할 때 최신순으로 정렬해서 이미 수집한 고유 값이 나오는 순간 리뷰 데이터 수집을 멈추는 방식을 사용했습니다.
🚧 네 번째 어려움 : 리뷰 로드 속도가 급격하게 느려지는 현상
위와 같이 구현해서 크롤링을 진행하였더니 초반에는 꽤나 순조롭게 리뷰 크롤링이 진행되었습니다.
하지만 특정 시점, 약 1,250개의 리뷰를 수집했을 때부터 크롤링 속도가 급격하게 느려지기 시작했습니다.
원인을 파악해 본 결과 더보기 버튼을 눌러 새로운 리뷰 정보를 로드할 때 views API를 호출하는데 이 API의 응답 시간이 갑자기 1분 이상 걸리기 시작했습니다.
원래 views API는 API 응답 시간이 500ms 내외였는데 어느 1,250개 리뷰를 수집한 이후부터 급격하게 증가하더니 1분 이상 걸리는 모습을 확인할 수 있습니다.
❌ 해결책: Views API 차단
처음 시도한 방식은 fetch 함수를 오버라이드 해서 views API 호출 자체를 막아버리는 것이었습니다.
하지만 이 방식을 사용하더라도 시간이 오래 걸리는 문제는 여전했습니다.
views API로 인해 응답 시간이 오래 걸리는 문제는 없어졌지만 더보기 버튼을 눌렀을 때 페이지가 1분 정도 동작 자체가 멈춰버렸습니다.
위 네트워크 탭을 보면 아시다싶이 페이지 차원에서 동작을 멈춰버리고 1분 정도 지난 후에 API 호출을 시작했습니다.
이 문제도 해결하려고 했지만 이 이상은 문제가 될 수도 있다고 생각이 되어 포기하게 되었습니다.
그래도 1,000여개의 리뷰 정보를 사용하면 식당에 대한 충분한 정보를 얻을 수 있다고 생각하였고 이 정도만 리뷰를 수집하기로 결정했습니다.
마무리
이렇게 리뷰 크롤링 시스템 구축을 마무리하였지만 아직 개선해야 할 부분이 많이 남아있는 것 같습니다.
리뷰 중복 처리에 대한 부분도 완벽하지 않은 것 같고 아직 playwright에 대한 사용 방법이 미숙하다고 느꼈습니다.
그래도 다행인 점은 네이버는 크롤링에 대해 엄격하게 관리하고 막으려 하는 것 같지는 않아서 쉽게 크롤링 구현을 할 수 있었던 것 같습니다.
언제나 크롤링으로 인해 타겟 사이트에 대한 공격이 되지 않도록 주의하는 것이 중요한 것 같습니다.
긴 글 읽어주셔서 감사합니다.
코드
from playwright.sync_api import sync_playwright
from typing import List, Dict, Set
import time
import hashlib
import os
import logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger(__name__)
class NaverMapReviewCrawler:
def __init__(self, headless: bool = True):
self.headless = headless
def _get_launch_options(self) -> dict:
"""브라우저 실행 옵션 반환"""
return {
"headless": self.headless,
"args": [
"--no-sandbox",
"--disable-dev-shm-usage",
"--disable-blink-features=AutomationControlled",
"--disable-features=IsolateOrigins,site-per-process",
"--disable-web-security",
"--disable-site-isolation-trials",
"--no-first-run",
"--no-default-browser-check",
"--disable-gpu",
"--disable-extensions",
"--disable-default-apps",
"--disable-sync",
"--disable-translate",
"--hide-scrollbars",
"--metrics-recording-only",
"--mute-audio",
"--safebrowsing-disable-auto-update",
"--ignore-certificate-errors",
"--ignore-ssl-errors",
"--ignore-certificate-errors-spki-list",
"--disable-setuid-sandbox",
"--window-size=1920,1080",
"--start-maximized",
],
}
def _get_context_options(self) -> dict:
"""브라우저 컨텍스트 옵션 반환"""
return {
"viewport": {"width": 1920, "height": 1080},
"user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36",
"locale": "ko-KR",
"timezone_id": "Asia/Seoul",
"permissions": ["geolocation"],
"geolocation": {"latitude": 37.5665, "longitude": 126.9780}, # 서울
"color_scheme": "light",
"device_scale_factor": 1,
"is_mobile": False,
"has_touch": False,
"extra_http_headers": {
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8",
"Accept-Language": "ko-KR,ko;q=0.9,en-US;q=0.8,en;q=0.7",
"Accept-Encoding": "gzip, deflate, br",
"Cache-Control": "max-age=0",
"Sec-Ch-Ua": '"Not_A Brand";v="8", "Chromium";v="120", "Google Chrome";v="120"',
"Sec-Ch-Ua-Mobile": "?0",
"Sec-Ch-Ua-Platform": '"Windows"',
"Sec-Fetch-Site": "none",
"Sec-Fetch-Mode": "navigate",
"Sec-Fetch-User": "?1",
"Sec-Fetch-Dest": "document",
"Upgrade-Insecure-Requests": "1",
},
}
def _sort_by_latest(self, page):
"""리뷰를 최신순으로 정렬"""
sort_buttons = page.query_selector_all("a.place_btn_option")
for btn in sort_buttons:
btn_text = btn.inner_text()
if "최신순" in btn_text:
btn.click()
time.sleep(2) # 정렬 완료 대기
logger.info("최신순으로 정렬됨")
break
def _generate_review_id(
self, author_name: str, review_text: str, visit_date: str
) -> str:
"""리뷰 고유 ID 생성"""
hash_input = f"{author_name}|{review_text}|{visit_date}"
return hashlib.sha256(hash_input.encode("utf-8")).hexdigest()
def _extract_review_data(self, elem, place_id: str) -> dict:
"""단일 리뷰 요소에서 데이터 추출"""
# 작성자
author = elem.query_selector("span.pui__NMi-Dp")
author_name = author.inner_text() if author else "익명"
# 리뷰 내용
content = elem.query_selector("div.pui__vn15t2 > a")
review_text = content.inner_text() if content else ""
# 방문날짜
date = elem.query_selector("time")
visit_date = date.inner_text() if date else ""
# 고유 ID 생성
review_id = self._generate_review_id(author_name, review_text, visit_date)
return {
"id": review_id,
"place_id": place_id,
"author": author_name,
"content": review_text,
"visit_date": visit_date,
}
def _load_more_reviews(self, page):
"""더 많은 리뷰 로드"""
more_button = page.query_selector("div.NSTUp a.fvwqf")
if more_button and more_button.is_visible():
more_button.scroll_into_view_if_needed()
more_button.click()
time.sleep(2) # 새 리뷰 로딩 대기
return True
else:
# 더보기 버튼이 없으면 스크롤
page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
time.sleep(2)
return False
def _process_reviews_on_page(
self,
page,
place_id: str,
existing_ids: Set[str],
already_appended_ids: Set[str],
) -> tuple:
"""현재 페이지의 리뷰 처리"""
reviews = []
stop_crawling = False
review_elements = page.query_selector_all("ul#_review_list > li.EjjAW")
for elem in review_elements:
review_data = self._extract_review_data(elem, place_id)
print(review_data)
review_id = review_data["id"]
# 이미 존재하는 id라면 즉시 중단
if review_id in existing_ids:
logger.info(f"이미 존재하는 리뷰(id={review_id}) 발견, 크롤링 중단")
stop_crawling = True
break
# 중복 체크
if review_id not in already_appended_ids:
reviews.append(review_data)
already_appended_ids.add(review_id)
page.evaluate("(element) => element.remove()", elem)
return reviews, stop_crawling
def crawl_all_reviews(self, place_id: str, existing_ids: Set[str]) -> List[Dict]:
"""네이버 지도의 모든 리뷰 크롤링
existing_ids: 이미 존재하는 리뷰 id의 set. 발견 시 즉시 중단 (필수)."""
with sync_playwright() as p:
browser = p.chromium.launch(**self._get_launch_options())
context = browser.new_context(**self._get_context_options())
page = context.new_page()
reviews = []
already_appended_ids = set()
# 리뷰 페이지로 이동
page.goto("https://httpbin.org/headers")
page.wait_for_timeout(2000)
url = f"https://pcmap.place.naver.com/restaurant/{place_id}/review/visitor"
page.goto(url)
page.wait_for_selector("ul#_review_list", timeout=10000)
# 최신순 정렬
self._sort_by_latest(page)
no_new_reviews_count = 0
stop_crawling = False
while not stop_crawling:
current_count = len(reviews)
# 현재 페이지의 리뷰 수집
page_reviews, stop_crawling = self._process_reviews_on_page(
page, place_id, existing_ids, already_appended_ids
)
reviews.extend(page_reviews)
logger.info(f"현재까지 {len(reviews)}개 리뷰 수집")
if stop_crawling:
break
# 새로운 리뷰가 없으면 카운트
if len(reviews) == current_count:
no_new_reviews_count += 1
if no_new_reviews_count >= 3:
logger.info("더 이상 새로운 리뷰가 없습니다.")
break
else:
no_new_reviews_count = 0
# 더 많은 리뷰 로드
self._load_more_reviews(page)
browser.close()
return reviews
# 사용 예시
def main():
try:
crawler = NaverMapReviewCrawler(headless=False)
# 예시: 특정 장소의 리뷰 크롤링
place_id = "1234567890" # 실제 place_id로 변경
existing_review_ids = set() # 기존 리뷰 ID들
reviews = crawler.crawl_all_reviews(place_id, existing_review_ids)
print(f"\n총 {len(reviews)}개의 리뷰를 수집했습니다.")
for i, review in enumerate(reviews[:5], 1): # 처음 5개만 출력
print(f"\n{i}. 작성자: {review['author']}")
print(f" 내용: {review['content'][:50]}...")
print(f" 방문일: {review['visit_date']}")
except Exception as e:
logger.error(f"크롤링 중 오류 발생: {str(e)}")
raise
if __name__ == "__main__":
main()