OpenStreetMap の世界へようこそ！〜Pythonを使ってデータを活用してみる〜 #Python – Qiita

import requests
import json
import time
import os


def cleaning_text(text: str) -> str:
    if ";" in text:
        text = text.split(";")[0]
    text = text.strip()
    return text


def fetch_mountain_spots(output_file="mountain_peak_data.json", area="日本"):
    print(f"{area}の山頂（natural=peak）データを取得中...")

    # 出力ディレクトリの設定
    output_dir = "output"
    base_dir = os.path.dirname(os.path.abspath(__file__))
    output_path = os.path.join(base_dir, output_dir, output_file)

    # 出力ディレクトリが存在しない場合は作成
    os.makedirs(os.path.join(base_dir, output_dir), exist_ok=True)

    # Overpass APIのエンドポイント
    overpass_url = "https://overpass-api.de/api/interpreter"

    # スポットデータを保存するリスト
    peak_spots = []
    total_spots = 0
    processed_spot_ids = set()

    print(f"mountain_peaks を取得中...")

    query_body = "node[\"natural\"=\"peak\"](area.searchArea);"
    overpass_query = """
        [out:json][timeout:180];
        area[name="{}"]->.searchArea;
        (
          {}
        );
        out center body;
        """.format(
        area, query_body
    )

    try:
        # APIリクエスト
        response = requests.post(overpass_url, data={"data": overpass_query})
        response.raise_for_status()

        # レスポンスのJSONを取得
        data = response.json()
        element_list = data.get("elements", [])

        print(f"  取得: {len(element_list)}件")

        # 取得したデータを整形
        for element in element_list:
            spot_id = element.get("id")

            # 基本情報
            spot_info = {
                "id": spot_id,
                "type": element.get("type"),
                "spot_type": "mountain_peaks",
                "tags": element.get("tags", {}),
            }

            # スポットIDの重複を防ぐ
            if spot_id in processed_spot_ids:
                continue

            processed_spot_ids.add(spot_id)

            # 座標情報の取得
            if element.get("type") == "node":
                spot_info["lat"] = element.get("lat")
                spot_info["lon"] = element.get("lon")
            elif (
                element.get("type") in ["way", "relation"]
                and "center" in element
            ):
                spot_info["lat"] = element.get("center", {}).get("lat")
                spot_info["lon"] = element.get("center", {}).get("lon")

            # 名前や他の重要な情報を抽出
            tags = element.get("tags", {})
            if "name" in tags:
                spot_info["name"] = cleaning_text(tags["name"])
            if "name:ja" in tags:
                spot_info["name_ja"] = cleaning_text(tags["name:ja"])
            if "name:en" in tags:
                spot_info["name_en"] = cleaning_text(tags["name:en"])

            # 標高情報の取得
            if "ele" in tags:
                spot_info["elevation"] = tags["ele"]
            elif "altitude" in tags:
                spot_info["elevation"] = tags["altitude"]

            # 標高の単位がない場合はメートルと仮定
            if "elevation" in spot_info and isinstance(
                spot_info["elevation"], str
            ):
                elevation_str = spot_info["elevation"]
                try:
                    if elevation_str.replace("ｍ", "m").lower().endswith("m"):
                        elevation_str = elevation_str[:-1]
                    elevation_str = elevation_str.replace(",", "")
                    elevation_str = cleaning_text(elevation_str)
                    if elevation_str and elevation_str.strip().replace('.', '', 1).isdigit(): # 小数点も許容
                        spot_info["elevation"] = float(elevation_str.strip())
                    else:
                        spot_info["elevation"] = None
                except ValueError:
                    spot_info["elevation"] = None
                except Exception:
                    spot_info["elevation"] = None

            if "height" in tags:
                spot_info["height_tag"] = tags["height"]
            if "summit:type" in tags:
                spot_info["summit_type"] = tags["summit:type"]

            peak_spots.append(spot_info)

    except requests.exceptions.RequestException as e:
        print(f"APIリクエストエラー: {e}")
    except json.JSONDecodeError as e:
        print(f"JSONデコードエラー: {e}")
    except Exception as e:
        print(f"エラーが発生しました: {e}")

    print(f"{len(peak_spots)}件を取得しました")
    total_spots = len(peak_spots)

    output_data = {
        "mountain_peaks": peak_spots
    }

    with open(output_path, "w", encoding="utf-8") as f:
        json.dump(
            {
                "spots": output_data,
                "total_spots": total_spots,
                "timestamp": time.time(),
                "area": area,
            },
            f,
            ensure_ascii=False,
            indent=2,
        )

    print(
        f"取得完了: 合計{total_spots}件を{output_dir}/{output_file}に保存しました。"
    )
    return {"spots": output_data, "total_spots": total_spots}


if __name__ == "__main__":
    fetch_mountain_spots()