スクレイピングで食べログを抽出し、オフィス周辺の飲食店をコンプリートする話。

1.はじめに

この記事は「KDDI Advent Calendar 2024」の11日目の記事です。

私の職場は飯田橋・水道橋にあるのですが、どこの店が美味しいのか、どこが評判なのか簡単には覚えられないほど、飲食店が立ち並んでおります。

せっかくなら効率よく、会社周辺の飲食店を行きつくしたいと思い、
食べログのスクレイピングで全掲載店をリストで抽出しました。

飲み会の幹事の店探しや、ランチ探しで困ることが減りましたね。
皆さんの職場やご自宅の周りの飲食店リストも作ってみてください

2.スクレイピング先を食べログにした理由

飲食店サイトはリクルートさんが運営されているhotpepperグルメなど他にも多数ありますが、食べログを選定理由した理由は以下になります。

①　使いやすいし慣れているから。
②　Webサイトが構造化されているから。
③　KDDIの資本提携先である同社サービスを信頼したいから。

3.完成物イメージ

数千店舗を抽出しました、職場の周りの飲食店多い！
食べログのスコアで降順に表示して、一定の口コミ数以上で絞ったら、お店選びで失敗しなくなりました。

4.実装

4.1.ライブラリの読み込み

#必要なライブラリのインポート
import requests
from bs4 import BeautifulSoup
import re
import pandas as pd
import time
import random
import os
import logging
import geopy.distance
import urllib3

4.2.ロギング設定とCSVファイルの初期化

ここではログファイルと店舗出力用のcsvファイルを作っています。
ログファイルは有っても無くても良いのですが、食べログのwebサーバーにどれぐらいのアクセスをしているかを確認した方が安心なので作成しています。

アクセス先のWebサーバーを落とした場合、訴訟を起こされる可能性があります。
アクセス先がスクレイピングOKがどうか、規約を先に読むようにしましょう。

# ロギング設定とCSVファイルの初期化

# ログファイルの設定
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("tabelog_scraping.log"),
        logging.StreamHandler()
    ]
)

# CSVファイルの初期化
output_csv = 'tabelog_Suidobashi.csv'
columns = ['store_id', 'store_name', 'hyperlink', 'genre', 'score', 'review_cnt', 'nearest_station', 'open_year', 'walking_time']

# 初回実行時にCSVファイルが存在しない場合はヘッダーを書き込む
if not os.path.exists(output_csv):
    df_header = pd.DataFrame(columns=columns)
    df_header.to_csv(output_csv, index=False)
    logging.info(f'CSVファイルを新規作成しました: {output_csv}')
else:
    logging.info(f'既存のCSVファイルを使用します: {output_csv}')

4.3.クラスの作成

class Tabelog:
    """
    食べログスクレイピングクラス。
    test_mode=Trueで動作させると、最初のページの2店舗のデータのみを取得できる。
    """
    def __init__(self, base_url, test_mode=False, begin_page=1, end_page=30, output_csv='tabelog_SuidoubashiIIdabashi.csv'):
        self.base_url = base_url
        self.test_mode = test_mode
        self.begin_page = begin_page
        self.end_page = end_page
        self.output_csv = output_csv
        self.store_id_num = 0
        self.columns = ['store_id', 'store_name', 'genre', 'hyperlink', 'score', 'review_cnt', 'nearest_station', 'open_year', 'walking_time']
        
        # リクエストヘッダーの設定
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ' 
                          '(KHTML, like Gecko) Chrome/90.0.4430.93 Safari/537.36'
        }
        
    def scrape(self):
        """
        スクレイピングを開始するメソッド。
        """
        page_num = self.begin_page
        while True:
            if page_num == 1:
                # 1ページ目のURL
                list_url = "https://tabelog.com/rstLst/?pcd=13&LstPrf=A1310&LstAre=A131003&station_id=5352&Cat=&RdoCosTp=2&LstCos=0&LstCosT=8&vac_net=0&search_date=2024%2F11%2F1%28%E9%87%91%29&svt=1900&svps=2&svd=20241101&LstRev=0&LstSitu=0&LstReserve=0&LstSmoking=0&PG=1&from_search=&voluntary_search=1&SrtT=rt&Srt=&sort_mode=1&LstRange=&keyword=&from_search_form=1&lid=&ChkNewOpen=&hfc=1"
            else:
                # 2ページ目以降のURL
                list_url = f"https://tabelog.com/tokyo/A1310/A131003/R5352/rstLst/{page_num}/?Srt=D&SrtT=rt&sort_mode=1&LstReserve=0&LstSmoking=0&svd=20241030&svt=1900&svps=2&vac_net=0&LstCosT=8&RdoCosTp=2"
            logging.info(f'ページ {page_num} のスクレイピングを開始します: {list_url}')
            success = self.scrape_list(list_url)
            if not success:
                logging.info('これ以上店舗が見つからないため、スクレイピングを終了します。')
                break
            if self.test_mode and page_num >= self.begin_page:
                logging.info('テストモードが有効なので、スクレイピングを終了します。')
                break
            if page_num >= self.end_page:
                logging.info('指定された終了ページに達したため、スクレイピングを終了します。')
                break
            page_num += 1
            time.sleep(4)
    
    def scrape_list(self, list_url):
        try:
            response = requests.get(list_url, headers=self.headers, timeout=10, verify=False)
            response.raise_for_status()
        except requests.exceptions.RequestException as e:
            logging.error(f'リストページの取得に失敗しました: {list_url}\nエラー内容: {e}')
            return False
        
        soup = BeautifulSoup(response.content, 'html.parser')
        soup_a_list = soup.find_all('a', class_='list-rst__rst-name-target')  

        if len(soup_a_list) == 0:
            logging.info('店舗が見つかりませんでした。')
            return False

        # テストモードの場合、最初の2店舗のみ処理
        if self.test_mode:
            target_restaurants = soup_a_list[:2]
        else:
            target_restaurants = soup_a_list

        for soup_a in target_restaurants:
            item_url = soup_a.get('href')  
            if not item_url:
                continue
            self.store_id_num += 1
            self.scrape_item(item_url)
            # リクエスト間隔をランダムに設定（1秒～3秒）
            time.sleep(random.uniform(1, 3))

        return True

    def scrape_item(self, item_url):
        """
        個別店舗情報ページのパーシング。
        """
        logging.info(f'店舗ページのスクレイピングを開始します: {item_url}')
        start_time = time.time()
        
        try:
            response = requests.get(item_url, headers=self.headers, timeout=10, verify=False)
            response.raise_for_status()
        except requests.exceptions.RequestException as e:
            logging.error(f'店舗ページの取得に失敗しました: {item_url}\nエラー内容: {e}')
            self.store_id_num -= 1
            return
    
        soup = BeautifulSoup(response.content, 'html.parser')

        # 店名を取得
        name_wrap = soup.find('div', class_='rstinfo-table__name-wrap')
        if not name_wrap:
            logging.warning(f'店名が見つかりませんでした: {item_url}')
            self.store_id_num -= 1
            return
        store_name_tag = name_wrap.find('span')
        if not store_name_tag:
            logging.warning(f'店名のタグが見つかりませんでした: {item_url}')
            self.store_id_num -= 1
            return
        store_name = store_name_tag.get_text(strip=True)
        logging.info(f'{self.store_id_num} → 店名: {store_name}')

        # ジャンルを取得
        genre = '不明'
        genre_header = soup.find('th', text='ジャンル')
        if genre_header:
            genre_tag = genre_header.find_next_sibling('td').find('span')
            if genre_tag:
                genre = genre_tag.get_text(strip=True)
        logging.info(f'  ジャンル: {genre}')

        #食べログスコア
        rating_score_tag = soup.find('b', class_='c-rating__val')
        if not rating_score_tag or not rating_score_tag.span:
            logging.warning(f'総合点数が見つかりませんでした: {item_url}')
            self.store_id_num -= 1
            return
        rating_score = rating_score_tag.span.get_text(strip=True)
        logging.info(f'  総合点数: {rating_score} 点')
        
        if rating_score == '-':
            logging.info('  総合点数が存在しないため、処理対象外です。')
            self.store_id_num -= 1
            return
        
        review_tag_id = soup.find('li', id="rdnavi-review")
        if not review_tag_id:
            logging.warning(f'レビューナビゲーションが見つかりませんでした: {item_url}')
            review_cnt = '0'
        else:
            review_count_em = review_tag_id.find('span', class_='rstdtl-navi__total-count').em
            review_cnt = review_count_em.get_text(strip=True) if review_count_em else '0'
            logging.info(f'  レビュー件数: {review_cnt}')
        
        nearest_station = self.get_nearest_station(soup)
        logging.info(f'  最寄駅: {nearest_station}')
        
        open_year = self.get_open_year(soup)
        logging.info(f'  オープン年: {open_year}')

        # 緯度経度の取得
        store_coords = self.get_store_coordinates(soup)
        station_coords = (35.70041655842002, 139.75041644557353)  # 出発地点の緯度経度
        
        # 徒歩時間の計算
        walking_time = self.calculate_walking_time(store_coords, station_coords)
        logging.info(f'  徒歩時間: {walking_time}')
        
        # make_dfメソッドの呼び出しで、すべての引数を正しく渡す
        self.make_df(store_name, item_url, genre, rating_score, review_cnt, nearest_station, open_year, walking_time)
        
        process_time = time.time() - start_time
        logging.info(f'  スクレイピング完了までの時間: {process_time:.2f} 秒')

    def get_nearest_station(self, soup):
        """
        最寄駅の取得
        """
        nearest_station = '不明'
        station_tag = soup.find('span', class_='linktree__parent-target-text')
        if station_tag:
            nearest_station = station_tag.get_text(strip=True)
        return nearest_station

    def get_open_year(self, soup):
        
        open_year = '不明'
        open_tag = soup.find('p', class_='rstinfo-opened-date')
        if open_tag:
            # 年を抽出するための正規表現
            match = re.search(r'(\d{4})年', open_tag.get_text(strip=True))
            if match:
                open_year = match.group(1) 
        return open_year

    def get_store_coordinates(self, soup):
        """
        Googleマップの静的画像から緯度経度を取得する。
        """
        # `data-original`属性からURLを取得
        map_image = soup.find('img', class_='js-map-lazyload')
        if not map_image:
            logging.warning('Googleマップの画像が見つかりませんでした')
            return (None, None)
    
        map_url = map_image.get('data-original')
    
        # 緯度経度をURLから抽出
        import re
        match = re.search(r'center=([-.\d]+),([-.\d]+)', map_url)
        if match:
            lat = float(match.group(1))
            lng = float(match.group(2))
            return (lat, lng)
        
        logging.warning('緯度経度がURLから抽出できませんでした')
        return (None, None)    

    def calculate_walking_time(self, store_coords, station_coords):
        """
        2つの緯度経度間の距離を計算し、徒歩時間を返す。
        """
        if None in store_coords or None in station_coords:
            return '不明'
        
        # 距離を計算（キロメートル）
        distance_km = geopy.distance.distance(store_coords, station_coords).km
        
        # 徒歩速度を80メートル/分として徒歩時間を計算
        walking_speed_m_per_min = 80
        walking_time_min = (distance_km * 1000) / walking_speed_m_per_min
        
        return f'{int(walking_time_min)}分'

    def make_df(self, store_name, hyperlink, genre, score, review_cnt, nearest_station, open_year, walking_time):
        """
        取得したデータをCSVに追加で保存するメソッド。
        """
        store_id = str(self.store_id_num)
        data = {
            'store_id': store_id,
            'store_name': store_name,
            'hyperlink': hyperlink,
            'genre': genre, 
            'score': score,
            'review_cnt': review_cnt,
            'nearest_station': nearest_station,
            'open_year': open_year,
            'walking_time': walking_time  
        }
        df_row = pd.DataFrame([data])

        # 初回書き込み時にヘッダーを追加
        write_header = not os.path.exists(self.output_csv)
    
        # デバッグ用: データ内容を確認
        # logging.debug(f'CSVに追加するデータ: {data}')
        
        df_row.to_csv(self.output_csv, mode='a', header=write_header, index=False)
        logging.info(f'CSVにデータを追加しました: {store_name}')
        return

4.4.スクレイピングの実行

# 警告を無視する
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

# スクレイピング対象のURL
base_url = 'https://tabelog.com/rstLst/?pcd=13&LstPrf=A1310&LstAre=A131003&station_id=5352&Cat=&RdoCosTp=2&LstCos=0&LstCosT=8&vac_net=0&search_date=2024%2F11%2F1%28%E9%87%91%29&svt=1900&svps=2&svd=20241101&LstRev=0&LstSitu=0&LstReserve=0&LstSmoking=0&PG=1&from_search=&voluntary_search=1&SrtT=rt&Srt=&sort_mode=1&LstRange=&keyword=&from_search_form=1&lid=&ChkNewOpen=&hfc=1'
# Tabelogクラスのインスタンス化
tabelog = Tabelog(
    base_url=base_url,
    test_mode=False,         
    begin_page=1,
    end_page=100,            
    output_csv='tabelog_suidobashi.csv'
)

# スクレイピングの開始
tabelog.scrape()

5.簡単に解説

食べログのWebサイトにはgoogleマップが差し込まれているので、F12で観察してみましょう。リンクに緯度経度があるのが分かると思います。

   def get_store_coordinates(self, soup):
        """
        Googleマップの静的画像から緯度経度を取得する。
        """
        # `data-original`属性からURLを取得
        map_image = soup.find('img', class_='js-map-lazyload')
        if not map_image:
            logging.warning('Googleマップの画像が見つかりませんでした')
            return (None, None)
    
        map_url = map_image.get('data-original')
    
        # 緯度経度をURLから抽出
        import re
        match = re.search(r'center=([-.\d]+),([-.\d]+)', map_url)
        if match:
            lat = float(match.group(1))
            lng = float(match.group(2))
            return (lat, lng)
        
        logging.warning('緯度経度がURLから抽出できませんでした')
        return (None, None)    

    def calculate_walking_time(self, store_coords, station_coords):
        """
        2つの緯度経度間の距離を計算し、徒歩時間を返す。
        """
        if None in store_coords or None in station_coords:
            return '不明'
        
        # 距離を計算（キロメートル）
        distance_km = geopy.distance.distance(store_coords, station_coords).km
        
        # 徒歩速度を80メートル/分として徒歩時間を計算
        walking_speed_m_per_min = 80
        walking_time_min = (distance_km * 1000) / walking_speed_m_per_min
        
        return f'{int(walking_time_min)}分'

クラスの中で職場の緯度経度を指定し、地点間の距離について計算しています。
私の場合は職場の緯度経度を指定し、出力CSVの列に徒歩時間として出力しています。

# 緯度経度の取得
store_coords = self.get_store_coordinates(soup)
station_coords = (35.70041655842002, 139.75041644557353)  # 出発地点の緯度経度

緯度経度を変えると、その地点からの徒歩時間を出力できます。
私は自宅周辺の飲食店リストも作ってみました。
その場合、スクレイピングするサイトのUrlも変えてください

def scrape(self):
        """
        スクレイピングを開始するメソッド。
        """
        page_num = self.begin_page
        while True:
            if page_num == 1:
                # 1ページ目のURL
                list_url = "https://tabelog.com/rstLst/?pcd=13&LstPrf=A1310&LstAre=A131003&station_id=5352&Cat=&RdoCosTp=2&LstCos=0&LstCosT=8&vac_net=0&search_date=2024%2F11%2F1%28%E9%87%91%29&svt=1900&svps=2&svd=20241101&LstRev=0&LstSitu=0&LstReserve=0&LstSmoking=0&PG=1&from_search=&voluntary_search=1&SrtT=rt&Srt=&sort_mode=1&LstRange=&keyword=&from_search_form=1&lid=&ChkNewOpen=&hfc=1"
            else:
                # 2ページ目以降のURL
                list_url = f"https://tabelog.com/tokyo/A1310/A131003/R5352/rstLst/{page_num}/?Srt=D&SrtT=rt&sort_mode=1&LstReserve=0&LstSmoking=0&svd=20241030&svt=1900&svps=2&vac_net=0&LstCosT=8&RdoCosTp=2"
            logging.info(f'ページ {page_num} のスクレイピングを開始します: {list_url}')
            success = self.scrape_list(list_url)
            if not success:
                logging.info('これ以上店舗が見つからないため、スクレイピングを終了します。')
                break
            if self.test_mode and page_num >= self.begin_page:
                logging.info('テストモードが有効なので、スクレイピングを終了します。')
                break
            if page_num >= self.end_page:
                logging.info('指定された終了ページに達したため、スクレイピングを終了します。')
                break
            page_num += 1
            time.sleep(4)

# スクレイピング対象のURL
base_url = 'https://tabelog.com/rstLst/?pcd=13&LstPrf=A1310&LstAre=A131003&station_id=5352&Cat=&RdoCosTp=2&LstCos=0&LstCosT=8&vac_net=0&search_date=2024%2F11%2F1%28%E9%87%91%29&svt=1900&svps=2&svd=20241101&LstRev=0&LstSitu=0&LstReserve=0&LstSmoking=0&PG=1&from_search=&voluntary_search=1&SrtT=rt&Srt=&sort_mode=1&LstRange=&keyword=&from_search_form=1&lid=&ChkNewOpen=&hfc=1'
# Tabelogクラスのインスタンス化
tabelog = Tabelog(
    base_url=base_url,
    test_mode=False,       
    begin_page=1,
    end_page=100,          
    output_csv='tabelog_suidobashi.csv'
)

# スクレイピングの開始
tabelog.scrape()

6.まとめ

スクレイピングはやはり便利ですね。
前述した通り、アクセス先の規約をよく確認したうえで試してみてください。
リクエスト間隔もちゃんと開けるのを忘れないようにしてください。

7.参考