diff --git a/cpc_direct_stations.py b/cpc_direct_stations.py index 87e32e6..9aa3a00 100755 --- a/cpc_direct_stations.py +++ b/cpc_direct_stations.py @@ -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.0,Nominatim 限制)") 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":