개요
이번에 새로운 프로젝트를 시작하게 되었습니다.
프로젝트 주제는 "현재 자신의 상황에 맞는 식당 추천 시스템"입니다.
식당 추천 시스템을 만들기 위해 네이버 지도의 식당 데이터를 수집하기로 결정했고 네이버 지도 정보를 크롤링하는 코드를 작성하기로 결정했습니다.
처음 목표는 간단했습니다.
- 네이버 지도에 특정 지역을 검색
- 각 식당의 place ID 추출
- 상세 정보(이름, 주소...) 수집
네이버 지도 식당 페이지 분석
먼저 네이버 지도에서 "공덕역 식당"을 검색했을 때 나오는 페이지를 분석해봅시다.
위에서 빨간색 테두리의 식당 리스트와 파란색 테두리의 식당 상세 정보는 모두 Iframe으로 이루어져 있습니다.
또한 식당 리스트의 경우 무한 스크롤과 페이지네이션이 조합된 형태입니다.
스크롤을 내릴 때마다 새로운 식당이 로딩되고 모두 로딩되면 페이지를 이동하여 새로운 식당 정보를 얻을 수 있습니다.
또한 식당을 클릭하여 새로운 창이 열리면 여기서 식당 상세 정보를 확인할 수 있습니다.
그렇다면 공덕역 식당 정보를 얻기 위해 다음과 같은 과정을 거쳐야 합니다.
- 네이버 지도에서 "공덕역 식당" 검색
- 모든 식당 정보 크롤링
- 각 식당의 상세 정보 크롤링
🚧 첫 번째 어려움 : 동적 데이터 수집
처음엔 당연히 requests와 BeautifulSoup로 시작했습니다.
import requests
from bs4 import BeautifulSoup
response = requests.get("https://map.naver.com/...")
soup = BeautifulSoup(response.text, 'html.parser')
네이버 지도는 무한 스크롤 방식으로 구현되어 있었습니다.
스크롤을 내려야만 JavaScript가 새로운 식당 데이터를 불러오기 때문에 단순히 html을 파싱하는 과정으로는 모든 식당 정보를 얻을 수 없습니다.
✅ 해결책: 가상 브라우저
이에 대한 해결책으로 가상 브라우저를 사용하기로 결정했습니다.
대부분 크롤링용 가상 브라우저로는 Selenium을 많이 사용하지만 저는 Playwright를 선택했습니다.
이유는 간단했습니다. Playwright는 기본적으로 비동기식 프로그래밍이고 Selenium은 기본적으로 동기식 프로그래밍을 지원하기 때문입니다.
물론 Selenium도 비동기 방식으로 실행할 수 있긴 하지만 라이브러리 자체에서 제공해주지 않기 때문에 Playwright를 선택했습니다.
특히, 크롤링의 경우 시간 지연의 대부분 원인이 Network I/O 작업이기 때문에 시간 단축을 위해서라도 비동기 프로그래밍을 적용해야 했습니다.
Network I/O와 논블록킹 I/O에 대한 글을 작성한 적이 있는데 이 글을 읽어보면 비동기 프로그래밍을 적용해야하는 이유에 대해 알 수 있을 것 같습니다.
https://growth-coder.tistory.com/350
I/O multiplexing (Non Blocking I/O vs Blocking I/O)
개요논블록킹 network I/O 모델이 등장하게 된 배경에 대해 알아봅시다.C10K 문제와 논블록킹 I/O 모델의 필요성C10K 문제는 network I/O와 연관이 있는 문제입니다. socket 프로그래밍을 통해 사용자의 요
growth-coder.tistory.com
https://growth-coder.tistory.com/301
동기(Synchronous) vs 비동기(Asynchronous), 블로킹(Blocking) vs 논블로킹(Non Blocking)
동기, 비동기동기 비동기는 작업의 순서에 초점을 맞춘다. 동기 작업은 요청한 작업이 완료된 후 다음 작업을 진행하기 때문에 작업의 순서가 보장된다. 비동기 작업은 작업을 요청한 후 완료
growth-coder.tistory.com
그럼 가상 브라우저를 활용해서 스크롤을 조작해서 식당 리스트를 크롤링 해보겠습니다.
저는 스크롤링을 하다가 더 이상 새로운 식당이 로드되지 않을 때까지 반복했습니다.
🚧 두 번째 어려움 : 식당의 고유 값 추출
식당 데이터를 중복해서 저장하지 않기 위해 식당의 고유 값을 추출해야 했습니다.
대부분의 사이트와 마찬가지로 식당 리스트의 각 요소들에서 a 태그의 href 속성의 값으로 들어있을 것이라고 생각했습니다.
하지만 네이버 지도의 경우 a 태그에 식당의 고유 값이 존재하지 않았습니다.
개발자 도구를 켜서 검색을 해도 식당의 고유 값을 찾을 수 없었기 때문에 다른방법을 찾아야 했습니다.
❌ 첫 번째 해결책: 직접 네이버 API 호출
네트워크 탭에서 HTTP 요청을 확인해 본 결과 모든 식당 리스트의 결과를 반환하는 API를 발견했습니다.
이 API를 활용한다면 번거롭게 가상 브라우저를 띄울 필요는 없어보이지만 url 그대로 HTTP 요청을 날려보면 403 forbidden 에러가 뜨는 것을 확인하실 수 있습니다.
인증 정보가 필요하지 않은 단순 GET 요청이라 문제가 없을 줄 알았는데 아마 네이버에서 비정상적인 접근을 차단하는 시스템이 있는 것으로 보입니다.
❌ 두 번째 해결책: url 변화 감지
다시 가상 브라우저 방식을 사용해 봅시다.
식당을 클릭하게 되면 새로운 iframe이 뜨게 되고 url이 바뀌게 되는데 바뀐 url을 확인해보면 아래와 같은 형태로 고유 값을 찾을 수 있었습니다.
https://map.naver.com/p/search/%EA%B3%B5%EB%8D%95%EC%97%AD%20%EC%8B%9D%EB%8B%B9/place/1126017750
# 각 식당을 클릭하면 URL이 변경된다
# https://map.naver.com/p/search/.../place/1234567890
restaurant_element.click()
await page.wait_for_url("**/place/**")
# 정규표현식으로 place ID 추출
match = re.search(r"/place/(\d+)", page.url)
place_id = match.group(1)✅ 세 번째 해결책: iframe 활용
마지막 해결책으로 iframe을 활용해보기로 했습니다.
앞서 말씀드렸다싶이 식당 상세 정보는 iframe으로 이루어져 있습니다.
iframe은 html 파일 안에 다른 html 파일을 삽입할 수 있는 요소입니다.
일반적으로 다른 웹페이지를 띄울 때 사용하곤 합니다.
네이버 지도 페이지를 확인해 본 결과 식당 상세 정보 iframe을 직접 브라우저를 통해 접속할 수 있었습니다.
https://pcmap.place.naver.com/restaurant/1235412532/home
이 페이지를 활용해서 단순하게 새로운 탭에 식당 상세 정보 페이지를 띄우기로 결정했습니다.
이렇게 되면 url 변화를 감지하면서 생기는 동기화 문제를 전혀 신경 쓸 필요가 없어집니다.
각 식당 상세 정보 크롤링은 독립적인 탭에서 실행되기 때문입니다.
코드
import asyncio
from playwright.async_api import async_playwright
from typing import List, Dict, Optional, Tuple
from storage_manager import RestaurantStorageManager
import os
import re
# 타임아웃 상수 (ms)
TIMEOUT = 10000
class NaverMapRestaurantCrawler:
def __init__(self, headless: bool = True):
self.headless = headless
self.launch_options = self._get_launch_options()
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",
},
}
async def _perform_search(self, page, search_query: str):
"""검색 수행"""
await page.goto("https://httpbin.org/ip")
await page.goto("https://map.naver.com/", wait_until="domcontentloaded")
search_input = await page.wait_for_selector(
"input.input_search", state="visible", timeout=TIMEOUT
)
await search_input.click()
await search_input.fill(search_query)
await search_input.press("Enter")
await page.wait_for_selector(
"iframe#searchIframe", state="visible", timeout=TIMEOUT
)
async def _get_search_frame(self, page):
"""검색 결과 iframe 가져오기"""
iframe_element = await page.query_selector("iframe#searchIframe")
return await iframe_element.content_frame()
async def _scroll_to_load_all(self, frame):
"""모든 결과가 로드될 때까지 스크롤"""
previous_count = 0
no_change_count = 0
max_no_change = 3
while True:
current_restaurants = await frame.query_selector_all("li.UEzoS")
current_count = len(current_restaurants)
if current_count == previous_count:
no_change_count += 1
if no_change_count >= max_no_change:
print("더 이상 로드할 데이터가 없습니다.")
break
else:
no_change_count = 0
previous_count = current_count
await frame.evaluate(
"""
() => {
const scrollContainer = document.querySelector('.Ryr1F') ||
document.querySelector('[role="main"]') ||
document.body;
if (scrollContainer) {
scrollContainer.scrollTop = scrollContainer.scrollHeight;
} else {
window.scrollTo(0, document.body.scrollHeight);
}
}
"""
)
await asyncio.sleep(2)
async def _extract_basic_info(self, restaurant):
"""식당 기본 정보 추출"""
name_elem = await restaurant.query_selector("span.TYaxT")
name = await name_elem.inner_text() if name_elem else "이름 없음"
category_elem = await restaurant.query_selector("span.KCMnt")
category = await category_elem.inner_text() if category_elem else ""
return name, category
async def _extract_place_id(self, restaurant, page):
"""place_id 추출"""
link_elem = await restaurant.query_selector("a.place_bluelink")
if not link_elem:
return None
await link_elem.click()
await page.wait_for_url(lambda url: "/place/" in url, timeout=TIMEOUT)
new_url = page.url
match = re.search(r"/place/(\d+)", new_url)
return match.group(1) if match else None
async def _extract_address_info(self, place_id: str, context):
"""주소 및 좌표 정보 추출"""
place_detail_url = f"https://pcmap.place.naver.com/place/{place_id}"
detail_page = await context.new_page()
address = None
await detail_page.goto(place_detail_url)
await detail_page.wait_for_selector("span.LDgIH", timeout=TIMEOUT)
address_elem = await detail_page.query_selector("span.LDgIH")
address = await address_elem.inner_text()
await detail_page.close()
return address
async def _extract_restaurant_data(self, restaurants, frame, page, context):
"""식당 데이터 추출"""
results = []
for restaurant in restaurants:
name, category = await self._extract_basic_info(restaurant)
place_id = await self._extract_place_id(restaurant, page)
address = None
if place_id:
address = await self._extract_address_info(place_id, context)
results.append(
{
"place_id": place_id,
"name": name,
"category": category,
"origin_address": address,
}
)
await page.go_back()
return results
async def crawl_single_page(self, search_query: str) -> List[Dict]:
"""특정 페이지 하나만 크롤링"""
async with async_playwright() as p:
browser = await p.chromium.launch(**self.launch_options)
context = await browser.new_context(**self._get_context_options())
page = await context.new_page()
await page.route(
"**/*.{png,jpg,jpeg,gif,svg,webp}", lambda route: route.abort()
)
results = []
try:
# 검색 수행
await self._perform_search(page, search_query)
# iframe 가져오기
frame = await self._get_search_frame(page)
if not frame:
return results
# 모든 결과 로드
await self._scroll_to_load_all(frame)
await frame.wait_for_selector(
"li.UEzoS", state="visible", timeout=TIMEOUT
)
# 데이터 추출
restaurants = await frame.query_selector_all("li.UEzoS")
results = await self._extract_restaurant_data(
restaurants, frame, page, context
)
print(f"{len(restaurants)}개 수집")
except Exception as e:
print(f"크롤링 중 오류: {str(e)}")
finally:
await browser.close()
return results
def merge_and_dedupe_results(
all_results: List[List[Dict]], existing_place_ids: set
) -> List[Dict]:
"""결과 병합 및 중복 제거"""
merged_results = []
for page_results in all_results:
merged_results.extend(page_results)
return [
item for item in merged_results if item["place_id"] not in existing_place_ids
]
def print_results_summary(results: List[Dict]):
"""결과 요약 출력"""
print(f"\n총 {len(results)}개 신규 식당 수집")
for i, restaurant in enumerate(results, 1):
print(
f"{i}. {restaurant['place_id']} [{restaurant['name']}] "
f"[{restaurant['category']}] [{restaurant['page']}] "
f"[origin_address: {restaurant['origin_address']}] "
f"[address: {restaurant['address']}] "
f"[latitude: {restaurant['latitude']}, longitude: {restaurant['longitude']}]"
)
async def main():
try:
search_query = input(
"식당 크롤링 할 위치를 입력하세요 (공덕역 식당 등등...) : "
)
print(f"search_query: {search_query}")
# 크롤러 생성 및 실행
crawler = NaverMapRestaurantCrawler(headless=False)
all_results = await crawler.crawl_single_page(search_query)
# 결과 처리
print_results_summary(all_results)
except Exception as e:
print(f"프로그램 실행 중 오류 발생: {str(e)}")
raise
if __name__ == "__main__":
asyncio.run(main())
마무리
이렇게 네이버 지도 페이지를 분석하며 식당 상세 정보 크롤링을 진행해보았습니다.
긴 글 읽어주셔서 감사합니다!