에브리 타임 크롤링
요기교 리뷰 크롤링
코드
import argparse
import os
import re
import time
import pickle
import pandas as pd
from tqdm import tqdm
from tqdm import trange
import warnings
warnings.filterwarnings("ignore")
import requests
from urllib.request import urlopen, Request
from fake_useragent import UserAgent
import json
import selenium
from selenium import webdriver
from selenium.webdriver import ActionChains
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.support.ui import Select
from selenium.webdriver.support.ui import WebDriverWait
# 스크롤 내리기
def scroll_bottom():
driver.execute_script("window.scrollTo(0,document.body.scrollHeight);")
# 1. 해당 카테고리 음식점 리스트 리턴
def get_restaurant_list(lat, lng, items=500):
"""
>>> get_restaurant_list(37.56, 126.93, 20)
'Request api for 20 restaurants
that fit a specific option
based on latitude 37.56 and longitude 126.93'
"""
restaurant_list = []
# 헤더 선언 및 referer, User-Agent 전송
headers = {
"referer": "https://www.yogiyo.co.kr/mobile/",
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/88.0.4324.182 Safari/537.36",
"Accept": "application/json",
"x-apikey": "iphoneap",
"x-apisecret": "fe5183cc3dea12bd0ce299cf110a75a2",
}
params = {
"items": items,
"lat": lat,
"lng": lng,
"order": ORDER_OPTION,
"page": 0,
"search": "",
}
host = "https://www.yogiyo.co.kr"
path = "/api/v2/restaurants"
url = host+path
# url = f"https://www.yogiyo.co.kr/api/v2/restaurants?items=60&lat=35.981742&lng=126.71587056&order=rank&page=0&search=&serving_type=vd"
response = requests.get(url, headers=headers, params=params)
print(response.json())
count = 0
for item in response.json()["restaurants"]:
restaurant_list.append(item["id"])
count += 1
print(restaurant_list)
return list(set(restaurant_list))
# 2. 검색한 음식점 페이지 들어가기
def go_to_restaurant(id):
try:
restaurant_url = "https://www.yogiyo.co.kr/mobile/#/{}/".format(id)
driver.get(url=restaurant_url)
print(driver.current_url)
except Exception as e:
print("go_to_restaurant 에러")
time.sleep(5)
# 3-1. 해당 음식점의 정보 페이지로 넘어가기
def go_to_info():
print("정보 페이지 로드중...")
driver.find_element(By.XPATH,'//*[@id="content"]/div[2]/div[1]/ul/li[3]/a').click()
time.sleep(2)
print("정보 페이지 로드 완료!")
# 3-2. 정보 더보기 클릭하기
def get_info():
op_time = driver.find_element(By.XPATH,'//*[@id="info"]/div[2]/p[1]/span').text
addr = driver.find_element(By.XPATH,'//*[@id="info"]/div[2]/p[3]/span').text
try:
number = driver.find_element(By.XPATH,'//*[@id="info"]/div[2]/p[2]/span').text
except Exception as e:
number = "정보 없음"
print(f"영업시간: {op_time}, 주소: {addr}, 번호: {number}")
return op_time, addr, number
# 4-1. 해당 음식점의 리뷰 페이지로 넘어가기
def go_to_review():
print("리뷰 페이지 로드중...")
driver.find_element(By.XPATH,'//*[@id="content"]/div[2]/div[1]/ul/li[2]/a').click()
time.sleep(2)
print("리뷰 페이지 로드 완료!")
# 4-2. 리뷰 더보기 클릭하기
def click_more_review():
driver.find_element(By.CLASS_NAME, "btn-more").click()
time.sleep(2)
# 5. 리뷰 페이지 모두 펼치기
def stretch_review_page():
review_count = int(
driver.find_element(By.XPATH,'//*[@id="content"]/div[2]/div[1]/ul/li[2]/a/span').text
)
click_count = int((review_count / 10))
print("모든 리뷰 불러오기 시작...")
for _ in trange(click_count):
try:
scroll_bottom()
click_more_review()
except Exception as e:
pass
scroll_bottom()
print("모든 리뷰 불러오기 완료!")
# 6. 해당 음식점의 모든 리뷰 객체 리턴
def get_all_review_elements():
reviews = driver.find_elements(By.CSS_SELECTOR,
"#review > li.list-group-item.star-point.ng-scope"
)
return reviews
# 7. 페이지 뒤로 가기 (한 음식점 리뷰를 모두 모았으면 다시 음식점 리스트 페이지로 돌아감)
def go_back_page():
print("페이지 돌아가기중...")
driver.execute_script("window.history.go(-1)")
time.sleep(2)
print("페이지 돌아가기 완료!" + "\n")
# 8. 크롤링과 결과 데이터를 pickle 파일과 csv파일로 저장
def save_pickle_csv(location, yogiyo_df):
# 'data' 디렉토리가 없으면 생성
if not os.path.exists('data'):
os.makedirs('data')
yogiyo_df.to_csv("./data/위도{}_경도{}_df.csv".format(location[0], location[1]))
yogiyo_df.to_excel("./data/위도{}_경도{}_df.xlsx".format(location[0], location[1]))
pickle.dump(yogiyo_df, open("./data/위도{}_경도{}_df.pkl".format(location[0], location[1]), "wb"))
print("{} {} pikcle save complete!".format(location[0], location[1]))
# 9. 크롤링 메인 함수 (카테고리, 거리, 가격 정보를 추가한 데이터 프레임 구조)
def yogiyo_crawling(location):
df = pd.DataFrame(
columns=[
"Restaurant", "UserID", "Menu", "Review", "Total", "Taste",
"Quantity", "Delivery", "Date", "OperationTime", "Address","Number"
]
)
try:
restaurant_list = get_restaurant_list(location[0], location[1], RESTAURANT_COUNT)
for restaurant_id in restaurant_list:
try:
print(f"********** {restaurant_list.index(restaurant_id) + 1}/{len(restaurant_list)} 번째 **********")
go_to_restaurant(restaurant_id)
go_to_info()
op_time, addr, number = get_info()
go_to_review()
stretch_review_page()
for review in tqdm(get_all_review_elements()):
try:
df.loc[len(df)] = {
"Restaurant": driver.find_element(By.CLASS_NAME,"restaurant-name").text,
"UserID": review.find_element(By.CSS_SELECTOR,"span.review-id.ng-binding").text,
"Menu": review.find_element(By.CSS_SELECTOR,"div.order-items.default.ng-binding").text,
"Review": review.find_element(By.CSS_SELECTOR,"p").text,
"Total": str(len(review.find_elements(By.CSS_SELECTOR,"div > span.total > span.full.ng-scope"))),
"Taste": review.find_element(By.CSS_SELECTOR,"div:nth-child(2) > div > span.category > span:nth-child(3)").text,
"Quantity": review.find_element(By.CSS_SELECTOR,"div:nth-child(2) > div > span.category > span:nth-child(6)").text,
"Delivery": review.find_element(By.CSS_SELECTOR,"div:nth-child(2) > div > span.category > span:nth-child(9)").text,
"Date": review.find_element(By.CSS_SELECTOR,"div:nth-child(1) > span.review-time.ng-binding").text,
"OperationTime": op_time,
"Address": addr,
"Number" : number
}
except Exception as e:
print("리뷰 페이지 에러")
print(e)
pass
except Exception as e:
print(f"*** 음식점 ID: {restaurant_id} *** 음식점 페이지 에러")
go_back_page()
print(e)
pass
print("음식점 리스트 페이지로 돌아가는중...")
go_back_page()
except Exception as e:
print("음식점 리스트 페이지 에러")
print(e)
pass
print(f"End of [ {location[0]} - {location[1]} ] Crawling!")
save_pickle_csv(location, df)
print(f"{location[0]} {location[1]} crawling finish!!!")
return df
# 10. 요기요 크롤링 실행 함수 (사용자에 맞게 수정 가능)
def start_yogiyo_crawling():
locations = [LAT, LON](LAT,%20LON)
for location in locations:
try:
yogiyo = yogiyo_crawling(location)
except Exception as e:
print(e)
pass
parser = argparse.ArgumentParser(description="Arguments for Crawler")
parser.add_argument(
"--order",
required=False,
default="distance",
help="option for restaurant list order / choose one \
-> [rank, review_avg, review_count, min_order_value, distance, estimated_delivery_time]",
)
parser.add_argument("--num", required=False, default=500, help="option for restaurant number")
parser.add_argument("--lat", required=False, default=35.9470221, help="latitude for search")
parser.add_argument("--lon", required=False, default=126.6815184, help="longitude for search")
args = parser.parse_args()
ORDER_OPTION = args.order
RESTAURANT_COUNT = int(args.num)
LAT = float(args.lat)
LON = float(args.lon)
# 크롬 드라이버 경로 설정 (절대경로로 설정하는 것이 좋음)
chromedriver = "C:\\Users\\Multi 03\\Documents\\chromedriver-win64\\chromedriver.exe"
chrome_options = Options()
chrome_options.add_argument('--disable-deprecated-web-platform-features')
service = Service(executable_path=chromedriver)
driver = webdriver.Chrome(service=service, options=chrome_options)
# driver = webdriver.Chrome(chromedriver)
url = "https://www.yogiyo.co.kr/mobile/#/"
# fake_useragent 모듈을 통한 User-Agent 정보 생성
useragent = UserAgent()
print(useragent.chrome)
print(useragent.ie)
print(useragent.safari)
print(useragent.random)
driver.get(url=url)
print(driver.current_url)
start_yogiyo_crawling()
설명
요기요(Yogiyo) 웹사이트에서 음식점 정보와 리뷰를 자동으로 수집하는 크롤링 프로그램입니다.
주요 기능
1. 음식점 목록 수집
- 위도/경도 기반으로 주변 음식점 API 호출
- 요기요의 내부 API (
/api/v2/restaurants) 사용 - API 키와 시크릿으로 인증
2. 음식점 상세 정보 수집
3. 리뷰 데이터 수집
# 수집하는 리뷰 정보
- Restaurant: 음식점명
- UserID: 리뷰어 ID
- Menu: 주문 메뉴
- Review: 리뷰 내용
- Total: 총점
- Taste/Quantity/Delivery: 맛/양/배달 점수
- Date: 리뷰 작성일
- OperationTime/Address/Number: 음식점 정보
크롤링 프로세스
1. 위치 기반 음식점 리스트 API 호출
2. 각 음식점 페이지 방문
3. 정보 탭에서 기본 정보 수집
4. 리뷰 탭에서 모든 리뷰 로드 (스크롤 + "더보기" 클릭)
5. 리뷰 데이터 파싱 및 저장
6. 다음 음식점으로 이동
7. CSV, Excel, Pickle 파일로 저장
사용법
파라미터:
- --lat: 위도 (기본값: 35.9470221)
- --lon: 경도 (기본값: 126.6815184)
- --num: 수집할 음식점 수 (기본값: 500)
- --order: 정렬 옵션 (rank, review_avg, distance 등)
기술 스택
- Selenium: 웹페이지 자동화 및 동적 콘텐츠 처리
- Requests: API 호출
- Pandas: 데이터 처리 및 저장
- BeautifulSoup: HTML 파싱 (간접적)
최종 코드
코드
from flask import Flask, request, jsonify, make_response
from flask_cors import CORS
import pandas as pd
import numpy as np
from gensim.models import Word2Vec
import re
import requests
from bs4 import BeautifulSoup
from datetime import datetime
from waitress import serve
import json
app = Flask(__name__)
app.config['JSON_AS_ASCII'] = False
CORS(app, resources={r"/*": {"origins": "*"}})
file_path = r'C:\Users\Multi 03\Desktop\code\reviewdata.xlsx'
try:
review_data = pd.read_excel(file_path)
except FileNotFoundError:
raise FileNotFoundError(f"File not found: {file_path}")
except Exception as e:
raise Exception(f"Error loading file: {e}")
stop_words_ko = [
"수", "것", "들", "점", "등", "더", "이", "그", "저",
"때", "거", "왜", "이런", "저런", "그런", "너무", "정말",
"진짜", "좀", "많이", "안", "못", "매우", "아주"
]
def preprocess_text(text):
text = re.sub(r'[^a-zA-Z가-힣\s]', '', str(text))
text = text.lower()
tokens = [word for word in text.split() if word not in stop_words_ko]
return tokens
review_data['Processed_Text'] = review_data['Review'].fillna('').astype(str) + ' ' + review_data['Menu'].fillna('')
review_data['Tokens'] = review_data['Processed_Text'].apply(preprocess_text)
sentences = review_data['Tokens'].tolist()
model = Word2Vec(sentences, vector_size=100, window=5, min_count=1, workers=4)
def text_to_vector(tokens, model):
vectors = [model.wv[token] for token in tokens if token in model.wv]
if vectors:
return np.mean(vectors, axis=0)
else:
return np.zeros(model.vector_size)
review_data['Vector'] = review_data['Tokens'].apply(lambda x: text_to_vector(x, model))
def recommend_specific_menu_and_restaurants(base_word, top_n=5):
if base_word not in model.wv:
return {"error": f"The word '{base_word}' is not in the vocabulary."}
similar_words = model.wv.most_similar(base_word, topn=top_n)
results = []
for similar_word, similarity_score in similar_words:
matching_rows = review_data[review_data['Tokens'].apply(lambda tokens: similar_word in tokens)]
for _, row in matching_rows.iterrows():
results.append({
"Similar Word": similar_word,
"Similarity Score": float(similarity_score),
"Restaurant": row['Restaurant'],
"Menu": row['Menu'],
"Review": row['Review']
})
results_df = pd.DataFrame(results)
results_df = results_df.drop_duplicates(subset=["Similar Word", "Restaurant"])
results_df = results_df.sort_values(by="Similarity Score", ascending=False)
return results_df.to_dict(orient='records')
@app.after_request
def apply_cors(response):
response.headers["Content-Type"] = "application/json; charset=utf-8"
response.headers['Content-Type'] = 'application/json; charset=utf-8'
response.charset = 'utf-8'
return response
@app.route('/', methods=['GET'])
def home():
return "Welcome to the Flask API!"
@app.route('/recommend', methods=['GET'])
def recommend():
base_word = request.args.get('word')
if not base_word:
return make_response(jsonify({"error": "The 'word' field is required."}), 400)
recommendations = recommend_specific_menu_and_restaurants(base_word, top_n=5)
recommendations = pd.DataFrame(recommendations).fillna("N/A").to_dict(orient='records')
response = make_response(
json.dumps(recommendations, ensure_ascii=False, indent=2,allow_nan=False),
200
)
response.headers['Content-Type'] = 'application/json; charset=utf-8'
return response
#return make_response(jsonify(recommendations, ensure_ascii=False), 200)
@app.route('/health', methods=['GET'])
def health_check():
return jsonify({"status": "OK", "message": "The server is running!"})
url = "https://www.kunsan.ac.kr/dormi/index.kunsan?menuCd=DOM_000000704006003000&&cpath=%2Fdormi"
def fetch_and_parse():
response = requests.get(url)
response.encoding = 'utf-8'
html_content = response.text
soup = BeautifulSoup(html_content, "html.parser")
return soup
def parse_operating_hours(soup):
hours = {}
table = soup.find("table", class_="ctable01")
rows = table.find_all("tr")[1:]
for row in rows:
cells = row.find_all("td")
if len(cells) > 0:
day_type = row.find("th", scope="row").get_text(strip=True)
weekdays_hours = cells[0].get_text(strip=True)
holidays_hours = cells[1].get_text(strip=True) if len(cells) > 1 else ""
dinner_hours = cells[2].get_text(strip=True) if len(cells) > 2 else ""
hours[day_type] = {"평일": weekdays_hours, "공휴일": holidays_hours, "저녁": dinner_hours}
return hours
def parse_weekly_menu(soup):
menu = {"아침": {}, "점심": {}, "저녁": {}}
table = soup.find("table", style="-ms-word-break: break-all;")
rows = table.find_all("tr")[1:]
meal_times = ["아침", "점심", "저녁"]
for i, row in enumerate(rows):
meal_type = meal_times[i]
cells = row.find_all("td")
days = ["월", "화", "수", "목", "금", "토", "일"]
for j, cell in enumerate(cells):
day = days[j]
menu[meal_type][day] = cell.get_text(separator=", ", strip=True)
return menu
def get_today_menu(menu):
days = ["월", "화", "수", "목", "금", "토", "일"]
today_day = days[datetime.today().weekday()]
day_menu = {}
for meal_time in menu:
if today_day in menu[meal_time]:
day_menu[meal_time] = menu[meal_time][today_day]
else:
day_menu[meal_time] = "메뉴 없음"
return today_day, day_menu
@app.route('/time', methods=['GET'])
def time():
soup=fetch_and_parse()
operating_hours = parse_operating_hours(soup)
response = make_response(
json.dumps(operating_hours, ensure_ascii=False, indent=2),
200
)
response.headers['Content-Type'] = 'application/json; charset=utf-8'
return response
@app.route('/weekly', methods=['GET'])
def weekly_Menu():
soup=fetch_and_parse()
operating_hours = parse_weekly_menu(soup)
response = make_response(
json.dumps(operating_hours, ensure_ascii=False, indent=2),
200
)
response.headers['Content-Type'] = 'application/json; charset=utf-8'
return response
def split_menu_items(menu_dict):
result = {}
for meal_time, menu_string in menu_dict.items():
menu_items = [item.strip() for item in menu_string.split(',')]
result[meal_time] = menu_items
return result
@app.route('/today', methods=['GET'])
def today_Menu():
soup=fetch_and_parse()
weekly_menu = parse_weekly_menu(soup)
today_day, today_menu = get_today_menu(weekly_menu)
operating_hours = parse_operating_hours(soup)
menu= split_menu_items(today_menu)
response = make_response(
json.dumps({
"day": today_day,
"menu": menu,
"time": operating_hours
}, ensure_ascii=False, indent=2),
200
)
response.headers['Content-Type'] = 'application/json; charset=utf-8'
return response
# 애플리케이션 실행
if __name__ == '__main__':
# app.run(host='0.0.0.0', port=43306,debug=True)
serve(app,host='0.0.0.0', port=43306)
설명
주요 기능
1. Word2Vec 기반 식당 추천 시스템
- 리뷰 데이터에서 단어 임베딩 모델 생성
- 100차원 벡터, 윈도우 크기 5, 최소 출현 빈도 1
2. 웹 크롤링을 통한 실시간 메뉴 정보
- 군산대학교 기숙사 식당 메뉴 크롤링
- 운영시간, 주간 메뉴, 오늘 메뉴 제공