Add coordinate lookup via Nominatim OSM API

- Add --coords flag to enable geocoding
- Add --delay option for rate limiting (default 1.0s)
- Try multiple query formats: full address, station+city, district
- Include latitude/longitude in JSON and CSV output

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-03-27 08:18:26 +08:00
parent 36b12dbe81
commit ee396828ab

View File

@@ -21,11 +21,12 @@ import csv
import json
import re
import sys
import time
import urllib.parse
import urllib.request
from html import unescape
from html.parser import HTMLParser
from typing import Dict, List
from typing import Dict, List, Optional
BASE_URL = "https://vipmbr.cpc.com.tw/mbwebs/service_search.aspx"
USER_AGENT = "Mozilla/5.0 (OpenClaw; cpc-direct-stations)"
@@ -134,6 +135,63 @@ def extract_hidden_fields(html: str) -> Dict[str, str]:
return fields
def geocode(address: str, station_name: str, city: str, district: str) -> Optional[Dict[str, float]]:
"""使用 Nominatim (OpenStreetMap) API 將地址轉換為座標。
API: https://nominatim.openstreetmap.org/search
台灣地址格式支援有限,依序嘗試:完整地址、站名+縣市、行政區
"""
# Nominatim 要求完整的 email 在 User-Agent
user_agent = f"{USER_AGENT} (geocoding)"
# 嘗試多種查詢格式
queries = []
if address:
queries.append(address)
queries.append(f"{city}{address}") # 縣市 + 地址
queries.append(f"{station_name} {city}") # 站名 + 縣市
if district:
queries.append(f"{city}{district}") # 縣市 + 行政區(作為備案)
for query in queries[:4]: # 最多試 4 種格式
try:
params = {
"q": query,
"format": "jsonv2",
"limit": "1",
"countrycodes": "tw",
}
url = f"https://nominatim.openstreetmap.org/search?{urllib.parse.urlencode(params)}"
req = urllib.request.Request(url, headers={"User-Agent": user_agent})
with urllib.request.urlopen(req, timeout=10) as response:
data = json.loads(response.read().decode("utf-8"))
if data:
return {"lat": float(data[0]["lat"]), "lon": float(data[0]["lon"])}
except Exception:
continue
return None
def add_coordinates(stations: List[Dict[str, str]], delay: float = 1.0) -> List[Dict[str, str]]:
"""為站點資料加上座標資訊。"""
for station in stations:
full_address = f"{station['縣市']}{station['鄉鎮區']}{station['地址']}"
coords = geocode(
address=full_address,
station_name=station["站名"],
city=station["縣市"],
district=station["鄉鎮區"],
)
if coords:
station["緯度"] = str(coords["lat"])
station["經度"] = str(coords["lon"])
else:
station["緯度"] = ""
station["經度"] = ""
time.sleep(delay) # 避免 Nominatim 請求過快
return stations
def build_query_html(city: str = "全部縣市", keyword: str = "") -> str:
city_value = CITY_MAP.get(city, city)
if city not in CITY_MAP and city != "全部縣市":
@@ -221,7 +279,7 @@ def output_json(stations: List[Dict[str, str]], out_path: str | None) -> None:
def output_csv(stations: List[Dict[str, str]], out_path: str | None) -> None:
fieldnames = ["縣市", "鄉鎮區", "類別", "站名", "站代號", "地址", "電話", "營業時間"]
fieldnames = ["縣市", "鄉鎮區", "類別", "站名", "站代號", "地址", "電話", "營業時間", "緯度", "經度"]
if out_path:
f = open(out_path, "w", encoding="utf-8-sig", newline="")
else:
@@ -238,8 +296,11 @@ def output_csv(stations: List[Dict[str, str]], out_path: str | None) -> None:
def output_text(stations: List[Dict[str, str]], out_path: str | None) -> None:
lines = [f"{len(stations)}"]
for idx, station in enumerate(stations, start=1):
coord = ""
if station.get("緯度") and station.get("經度"):
coord = f" ({station['緯度']},{station['經度']})"
lines.append(
f"{idx}. {station['站名']}{station['站代號']}{station['縣市']}{station['鄉鎮區']}{station['地址']}{station['電話']}{station['營業時間']}"
f"{idx}. {station['站名']}{station['站代號']}{station['縣市']}{station['鄉鎮區']}{station['地址']}{station['電話']}{station['營業時間']}{coord}"
)
text = "\n".join(lines)
if out_path:
@@ -255,12 +316,17 @@ def main() -> None:
parser.add_argument("--keyword", default="", help="關鍵字,可篩站名或地址")
parser.add_argument("--format", choices=["text", "json", "csv"], default="text")
parser.add_argument("--out", help="輸出到檔案路徑")
parser.add_argument("--coords", action="store_true", help="加上座標資訊(使用 Nominatim OSM API")
parser.add_argument("--delay", type=float, default=1.0, help="座標查詢間隔秒數(預設 1.0Nominatim 限制)")
args = parser.parse_args()
html = build_query_html(city=args.city, keyword=args.keyword)
stations = parse_stations(html)
stations = filter_stations(stations, args.keyword)
if args.coords:
stations = add_coordinates(stations, delay=args.delay)
if args.format == "json":
output_json(stations, args.out)
elif args.format == "csv":