cpc-direct-stations/fuel_coords_db.py

#!/usr/bin/env python3
"""建立台灣加油站座標資料庫（來源：OpenStreetMap Overpass API）

用法：
  # 下載 OSM 加油站資料並建立資料庫
  python3 fuel_coords_db.py --download

  # 查詢座標
  python3 fuel_coords_db.py --query "台灣中油 社子"
"""

from __future__ import annotations

import argparse
import json
import math
import os
import urllib.request
from pathlib import Path
from typing import Dict, List, Optional

DB_FILE = Path(__file__).parent / "fuel_stations.json"
OVERPASS_URL = "https://overpass-api.de/api/interpreter"


def download_osm_fuel_stations() -> List[Dict]:
    """從 OpenStreetMap Overpass API 下載台灣加油站資料。"""
    query = """
[out:json][timeout:300];
area["name:zh-Hant"="臺灣"]->.searchArea;
(
  node["amenity"="fuel"]["operator"~"CPC|中油|台灣中油"](area.searchArea);
  way["amenity"="fuel"]["operator"~"CPC|中油|台灣中油"](area.searchArea);
  relation["amenity"="fuel"]["operator"~"CPC|中油|台灣中油"](area.searchArea);
  node["amenity"="fuel"]["brand"~"CPC|中油|台灣中油"](area.searchArea);
  way["amenity"="fuel"]["brand"~"CPC|中油|台灣中油"](area.searchArea);
  relation["amenity"="fuel"]["brand"~"CPC|中油|台灣中油"](area.searchArea);
  node["amenity"="fuel"]["name"~"台塑|台亞"](area.searchArea);
  way["amenity"="fuel"]["name"~"台塑|台亞"](area.searchArea);
);
out center;
"""
    print("下載 OSM 加油站資料...", flush=True)
    req = urllib.request.Request(
        OVERPASS_URL,
        data=query.encode("utf-8"),
        headers={"Content-Type": "text/plain"},
    )
    with urllib.request.urlopen(req, timeout=300) as response:
        data = json.loads(response.read().decode("utf-8"))
        print(f"找到 {len(data.get('elements', []))} 個加油站", flush=True)
        return data.get("elements", [])


def build_database(elements: List[Dict]) -> Dict[str, Dict]:
    """建立站名 -> 座標 的對應資料庫。"""
    db = {}
    addresses = {}  # 地址 -> 座標 (用於地址匹配)

    for elem in elements:
        tags = elem.get("tags", {})
        name = tags.get("name") or tags.get("name:zh") or tags.get("name:zh-Hant")
        if not name:
            name = "加油站"

        # 取得座標
        if elem["type"] == "node":
            lat, lon = elem["lat"], elem["lon"]
        else:
            lat, lon = elem.get("center", {}).get("lat"), elem.get("center", {}).get("lon")
            if lat is None or lon is None:
                continue

        city = tags.get("addr:city") or tags.get("addr:city:zh") or ""
        street = tags.get("addr:street") or tags.get("addr:street:zh") or ""
        number = tags.get("addr:housenumber") or ""

        # 建立多種查詢鍵
        keys = [name]
        if city:
            keys.extend([f"{city}{name}", f"{name} ({city})"])

        brand = tags.get("brand") or tags.get("operator")
        if brand and brand not in name:
            keys.append(f"{brand} {name}")

        # 如果有完整地址，也加入地址鍵
        if street:
            full_addr = f"{city}{street}{number}".strip()
            if full_addr:
                addresses[full_addr] = {"lat": lat, "lon": lon, "source": "osm"}
                keys.append(full_addr)

        for key in keys:
            key = key.strip()
            if key and key not in db:
                db[key] = {"lat": lat, "lon": lon, "source": "osm"}

    # 合併地址庫
    db.update(addresses)
    return db


def save_database(db: Dict) -> None:
    """儲存資料庫。"""
    with open(DB_FILE, "w", encoding="utf-8") as f:
        json.dump(db, f, ensure_ascii=False, indent=2)
    print(f"資料庫已儲存至 {DB_FILE}")


def load_database() -> Dict:
    """載入資料庫。"""
    if DB_FILE.exists():
        with open(DB_FILE, "r", encoding="utf-8") as f:
            return json.load(f)
    return {}


def haversine_distance(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
    """計算兩點間的距離（公里）。"""
    R = 6371  # 地球半徑（公里）
    dlat = math.radians(lat2 - lat1)
    dlon = math.radians(lon2 - lon1)
    a = (math.sin(dlat / 2) ** 2 +
         math.cos(math.radians(lat1)) * math.cos(math.radians(lat2)) *
         math.sin(dlon / 2) ** 2)
    c = 2 * math.asin(math.sqrt(a))
    return R * c


def query_coords(name: str, city: str = "", db: Dict | None = None) -> Optional[Dict[str, float]]:
    """查詢加油站座標。"""
    if db is None:
        db = load_database()

    if not db:
        return None

    # 精確匹配
    queries = [name]
    if city:
        queries.extend([f"{city}{name}", f"{name} ({city})"])

    for q in queries:
        if q in db:
            return db[q]

    # 模糊匹配（包含站名）
    for key, value in db.items():
        if name in key or key in name:
            return value

    return None


def main() -> None:
    parser = argparse.ArgumentParser(description="台灣加油站座標資料庫")
    parser.add_argument("--download", action="store_true", help="下載 OSM 資料並建立資料庫")
    parser.add_argument("--query", help="查詢加油站座標")
    parser.add_argument("--city", default="", help="查詢時指定縣市")
    args = parser.parse_args()

    if args.download:
        elements = download_osm_fuel_stations()
        db = build_database(elements)
        save_database(db)
        print(f"總共 {len(db)} 個查詢鍵")

    elif args.query:
        db = load_database()
        if not db:
            print("資料庫不存在，請先執行 --download", file=__import__("sys").stderr)
            return
        result = query_coords(args.query, args.city, db)
        if result:
            print(f"座標: {result['lat']}, {result['lon']}")
        else:
            print(f"找不到: {args.query}")

    else:
        db = load_database()
        if db:
            print(f"資料庫包含 {len(db)} 個查詢鍵")
        else:
            print("資料庫不存在，請先執行 --download", file=__import__("sys").stderr)


if __name__ == "__main__":
    main()