Yahoo!ローカルサーチAPIから情報をCSVで取得する【Python】

Yahoo!ローカルサーチAPIとは

• 公式サイト：リンク

Yahoo!ローカルサーチAPI（以下、ローカルサーチAPI）は、店舗、イベント、クチコミ情報などの地域・拠点情報（POI）を検索するためのAPIです。
検索対象は、全国の店舗を網羅した電話帳データおよび店舗オーナーなどからの投稿データです。
ローカルサーチAPIの主な機能は以下の通りです。
・地域・拠点情報の名称および業種をキーワードで検索できます。
・中心位置を指定して円範囲で絞り込めます。
・矩形範囲を指定して絞り込めます。

本記事の作成動機

• 調べてみたいことがありデータ取得スクリプトを作成したのですが、特に地理情報を勉強されている初学者の方にとってお店の位置情報を可視化するのは楽しい気がします
• データのハンドリングに慣れていない人がなるべく手軽に試せるように、APIからデータ取得するサンプルスクリプトを共有しようと思いました

前準備

Client IDの取得

このAPIを利用するには、事前にClient IDを取得する必要があります。
下記リンクを参考に取得してください。

ご利用ガイド：公式リンク

Client IDが取得できたら、後述のPythonスクリプトでデータ取得をしてみます。

Pythonによるデータ取得スクリプト

実行環境

• MacOS 14.4.1
• Python 3.9.6

スクリプト

Loader_YahooLocalSearch.py

import requests
import time
import glob
import sys
import csv
import os

#ベースとなるURL
base_url = "https://map.yahooapis.jp/search/local/V1/localSearch"

api_key = input("APIキー（Client ID）を入力: ") #コマンドライン上でAPIキーの入力を求める

#都道府県の住所コードを辞書で定義（都道府県や市区町村の住所コードを入力しない場合、全国分を繰り返し処理で取得するため）
prefectures = {
    "01": "北海道", "02": "青森県", "03": "岩手県", "04": "宮城県", "05": "秋田県", 
    "06": "山形県", "07": "福島県", "08": "茨城県", "09": "栃木県", "10": "群馬県",
    "11": "埼玉県", "12": "千葉県", "13": "東京都", "14": "神奈川県", "15": "新潟県",
    "16": "富山県", "17": "石川県", "18": "福井県", "19": "山梨県", "20": "長野県",
    "21": "岐阜県", "22": "静岡県", "23": "愛知県", "24": "三重県", "25": "滋賀県",
    "26": "京都府", "27": "大阪府", "28": "兵庫県", "29": "奈良県", "30": "和歌山県",
    "31": "鳥取県", "32": "島根県", "33": "岡山県", "34": "広島県", "35": "山口県",
    "36": "徳島県", "37": "香川県", "38": "愛媛県", "39": "高知県", "40": "福岡県",
    "41": "佐賀県", "42": "長崎県", "43": "熊本県", "44": "大分県", "45": "宮崎県",
    "46": "鹿児島県", "47": "沖縄県"
}

#URLパラメータ用の辞書を用意し、後からパラメータを順次格納する。01はヒット総数の確認用、02はデータ取得用。
params_01 = {"appid":api_key, "results":1, "output":"json"} 
params_02 = {"appid":api_key, "sort":"kana", "output":"json"} #繰り返し処理でID重複が生じる可能性を低減するためソート順を「かな」に設定

#パラメータを対話的に入力する
p_query = input('検索する文字列を入力（Enterキーでスキップ） >> ')
if p_query != "":
    params_01['query'] = str(p_query)
    params_02['query'] = str(p_query)
else:
    pass

p_gc = input('検索する業種コードを入力（Enterキーでスキップ） >> ')
if p_gc != "":
    params_01['gc'] = str(p_gc)
    params_02['gc'] = str(p_gc)
else:
    pass

p_ac = input('検索する住所コードを入力（Enterキーでスキップ） >> ')
if p_ac != "":   
    params_01['ac'] = str(p_ac)
    params_02['ac'] = str(p_ac)
else:
    pass

#ヒット件数取得用の関数
def count_data(params_01):
    response_01 = requests.get(base_url, params=params_01) #ヒット件数の確認用のリクエストを投げる処理
    jsonData_01 = response_01.json()
    time.sleep(0.5) #リクエスト１回ごとに若干時間をあけてAPI側への負荷を軽減する
    return jsonData_01["ResultInfo"]["Total"]

#データ取得処理用の関数
def fetch_data(params_02, total_num, pref_name, output_dir):
    max_return = 100 #APIの仕様では一回のリクエストにつき100件まで取得可能なので、その上限値を一回の取得数として設定
    pages = (int(total_num) // int(max_return)) + 1 #全件を取得するために必要なリクエスト回数を算定

    params_02['results'] = max_return #全件取得用のパラメータを設定

    Records = [] #取得データを格納するための空リストを用意

    #全件取得するためのループ処理
    for i in range(pages):
        i_startRecord = 1 + (i * int(max_return))
        params_02['start'] = i_startRecord
        response_02 = requests.get(base_url, params=params_02)

        #レスポンスのステータスが200＝正常取得だった場合の処理
        if response_02.status_code == 200:
            try:
                jsonData_02 = response_02.json() #レスポンスをJSONデータとして格納する
            except ValueError:
                print("エラー: レスポンスデータの解析処理に失敗しました。")
                sys.exit() #ここでエラーが生じた場合は処理を終了させる。ここをcontinueに変えて、この100件分だけスキップして処理続行させることも可能。
        else:
            print("エラー:", response_02.status_code)
            sys.exit() #レスポンスが正常に取得できなかった場合は処理を終了させる。

        #JSONデータ内の各要素から必要項目を指定してリストに格納する
        for poi in jsonData_02.get('Feature', []):
            poi_id = poi.get('Id', "") #FeatureにId項目があればその値を、ない場合は空欄を返す
            poi_name = poi.get('Name', "")
            coordinates = poi.get('Geometry', {}).get('Coordinates', "").split(",") #Coordinatesの座標値はカンマ区切りで緯度経度に分割する
            poi_lat = coordinates[1] if len(coordinates) > 1 else ""
            poi_lng = coordinates[0] if len(coordinates) > 0 else ""
            Records.append([poi_id, poi_name, poi_lat, poi_lng])

        sys.stdout.write(f"\r{pref_name}: {i+1}/{pages} is done.") #進捗状況を表示する
        sys.stdout.flush() #進捗状況を強制的に変更する
        time.sleep(0.5) #リクエスト１回ごとに若干時間をあけてAPI側への負荷を軽減する

    #CSVへの書き出し
    csv_file_path = os.path.join(output_dir, f"poi_result_{pref_name}_{total_num}.csv")
    with open(csv_file_path, 'w', newline='', encoding='utf-8') as f:
        csvwriter = csv.writer(f, delimiter=',', quotechar='"', quoting=csv.QUOTE_NONNUMERIC) #CSVの書き出し方式を適宜指定
        csvwriter.writerow(['ID', 'name', 'lat', 'lng'])
        for record in Records:
            csvwriter.writerow(record)

    print(f"\nデータ（{pref_name}）がCSV形式で出力されました。ファイル名： {csv_file_path}")

#住所コードが入力されなかった場合、全都道府県を対象に繰り返し処理を行う
if not p_ac:
    #ヒット件数の確認用のリクエストを投げる処理
    try:
        total_num_all = count_data(params_01)
    except ValueError:
        print(f'ヒット件数取得に失敗したため、プログラムを終了します。入力パラメータをご確認ください。')
        sys.exit()
    #総ヒット件数を表示し、処理を続行するか確認
    next_input = input("検索結果は " + str(total_num_all) + "件です。\nキャンセルする場合は 1 を、データを取得するにはEnterキーまたはその他を押してください。 >> ")
    if next_input == "1":
        print(f'プログラムをキャンセルしました')
        sys.exit()
    else:
        pass

    #出力先フォルダの設定
    output_dir = "output_" + str(total_num_all)
    os.makedirs(output_dir, exist_ok=True)

    #処理を続行する場合は、都道府県コードごとにデータ取得の繰り返し処理を実施する
    for pref_code, pref_name in prefectures.items(): #都道府県コードと名称をそれぞれ変数に格納
        params_01['ac'] = pref_code
        params_02['ac'] = pref_code

        #ヒット件数の確認用のリクエストを投げる処理
        try:
            total_num_each = count_data(params_01)
        except ValueError:
            print(f'ヒット件数取得に失敗しました。 ({pref_name})')
            continue #ここで失敗した場合、プログラム自体は終了せず、該当する都道府県の処理だけスキップ

        #ヒット件数が0件以上かつ取得条件の3100件以内だった場合は取得処理を実行、それ以外はメッセージを出して終了させる。なお、パラメータに何らかの問題があると大量のヒット件数が返されることがある。
        if total_num_each > 3100:
            print(f"データ取得上限の件数を超えているか入力パラメータが不適なため、取得処理をスキップします ({pref_name})。この都道府県では市区町村コードなどで条件を細分化してください。")
        elif total_num_each > 0:
            fetch_data(params_02, total_num_each, pref_name, output_dir)
        else:
            print(f"該当するデータがありません。（{pref_name}）")

    #取得した複数のCSVを一つにマージしたCSVファイルも作成するか確認
    merge_input = input("取得したCSVを一つに統合する必要がなければ 1 を、統合したCSVも作成する場合はEnterキーまたはその他を押してください。 >> ")
    if merge_input == "1":
        pass
    else:
        #マージする場合の処理
        merged_csv_path = os.path.join(output_dir, f"poi_merged_{total_num_all}.csv")
        csv_files = glob.glob(os.path.join(output_dir, "poi_result_*.csv"))

        with open(merged_csv_path, 'w', newline='', encoding='utf-8') as outputfile:
            csvwriter = csv.writer(outputfile, delimiter=',', quotechar='"', quoting=csv.QUOTE_NONNUMERIC)
            csvwriter.writerow(['ID', 'Name', 'Lat', 'Lng']) #ヘッダーを書き込む

            for filename in csv_files:
                with open(filename, 'r', newline='', encoding='utf-8') as inputfile:
                    csvreader = csv.reader(inputfile)
                    next(csvreader) #各CSVファイルのヘッダーをスキップ
                    for row in csvreader:
                        csvwriter.writerow(row)
        print(f"\n統合データをCSV形式で出力しました。ファイル名： {merged_csv_path}")

#住所コードが入力された場合、その住所コードとパラメータに従って処理を行う
else:
    #ヒット件数の確認用のリクエストを投げる処理
    try:
        total_num = count_data(params_01)
    except ValueError:
        print(f'ヒット件数取得に失敗したため、プログラムを終了します。入力パラメータをご確認ください。')
        sys.exit()
    #総ヒット件数を表示し、処理を続行するか確認
    next_input = input("検索結果は " + str(total_num) + "件です。\nキャンセルする場合は 1 を、データを取得するにはEnterキーまたはその他を押してください。 >> ")
    if next_input == "1":
        print(f'プログラムをキャンセルしました')
        sys.exit()
    else:
        pass

    #出力先フォルダの設定
    output_dir = "output_" + str(total_num)
    os.makedirs(output_dir, exist_ok=True)
    
    #ヒット件数が0件以上かつ取得条件の3100件以内だった場合は取得処理を実行、それ以外はメッセージを出して終了させる。なお、パラメータに何らかの問題があると大量のヒット件数が返されることがある。
    if total_num > 3100:
        print(f"データ取得上限の件数を超えているか入力パラメータが不適なため、取得処理をスキップします。市区町村コードなどで条件を細分化してください。")
    elif total_num > 0:
        fetch_data(params_02, total_num, p_ac, output_dir)
    else:
        print(f"該当するデータがありません。")

解説

• 説明はスクリプト内のコメントを参照ください。Pythonの外部モジュールとしてrequestsを使っているので、これは必要に応じてpipでインストールしてください

• スクリプトを実行すると、コマンドライン上でClient ID、検索ワード、業種コード、住所コードが順番に要求されますので、必要な情報を対話的に入力していってください。また、Client ID以外は必須ではないので、Enterキーを押して入力をスキップできます

• APIの仕様では、1回のリクエストで取得できるデータは100件までとなります。そのため、例えばヒット件数が1000件だったら10回分繰り返し処理が必要です。そこで、まずはヒット総数を["ResultInfo"]["Total"]で確認して何回繰り返し処理を行うかを計算し、さらにn回目でどこを取得開始位置としてデータ取得するかを['start']で判断しています

• また、['start']で取得できる開始位置の上限が3000までのようなので、実質的に一つの条件で3100件以上ヒットした場合、それ以降は取得できないと理解しています。そのため、3100件以上がヒットした場合は条件を絞るようにメッセージで示す形にしています

• 住所コード（特定の都道府県や市区町村を検索対象とする場合のコード。例：東京=13）を指定しなかった場合は、全国分を各都道府県単位での繰り返し処理によりデータ取得します。これは、全国分を一気に取得しようとすると3100件を超えるケースが多いと想定したためです

• APIからのレスポンスはJSON形式ですが、その後の扱いやすさを考えて出力結果はCSVで整形しなおしています。ここは好みかと思います

• 出力結果を確認すると、IDの重複が見られたり、Yahoo!地図に登録があるにもかかわらずデータ取得できないものがあったり、というケースが見られます。データ内容と網羅性は念の為ご確認ください。

出力結果の例

CSVデータ

全国のラーメン・つけ麺店を業種コードから取得した例です。約2万7千件取得できました。CSV出力の結果をExcelで読み込んだのが下記となります。
取得結果の一部でID重複があったり、地元のラーメン屋をチェックしたらYahoo！地図にはラーメン店として掲載されているのにこれでは拾えなかった店舗もあり、若干の抜け漏れや重複には留意が必要です。

GISで表示した図

取得したデータをQGISで表示した図です。概ね良い感じで取得できてることがわかります。

ラーメン激戦区を調べてみる

QGISを使って1km圏内に10件以上のラーメン店が密集している箇所を特定した図です。
手法はこちらの記事（リンク）を参照ください。

私が住んでいる千葉県柏市は、特に柏駅周辺はラーメン激戦区かな〜？という印象でしたが、数の勝負だと都内はやっぱり桁違いですね。ただ、柏市も周辺に密集地が少ないので相対的に激戦区っぽく感じるのだろう、という結論に至りました。

おわりに

以上、Yahoo!ローカルサーチAPIからデータ取得するためのPythonスクリプトについて説明しました。
お店などの情報を手軽に取得できるのは楽しいですね。また、業種コードなどでデータが整理されているのもありがたいです。
すでに多くの人が可視化などで使っているこのAPIですが、私のほうでも色々試してみたいと思います。