Remove coordinate feature due to poor UX

- Revert to initial clean version without coordinates
- Nominatim rate limiting makes the feature unusable
- Keep core functionality: fetching CPC station data

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-03-27 09:47:10 +08:00
parent 876b7483ba
commit 3ddbf259ae
3 changed files with 18 additions and 6054 deletions

View File

@@ -17,10 +17,8 @@
from __future__ import annotations from __future__ import annotations
import argparse import argparse
import concurrent.futures
import csv import csv
import json import json
import os
import re import re
import sys import sys
import time import time
@@ -28,13 +26,10 @@ import urllib.parse
import urllib.request import urllib.request
from html import unescape from html import unescape
from html.parser import HTMLParser from html.parser import HTMLParser
from pathlib import Path
from typing import Dict, List, Optional from typing import Dict, List, Optional
BASE_URL = "https://vipmbr.cpc.com.tw/mbwebs/service_search.aspx" BASE_URL = "https://vipmbr.cpc.com.tw/mbwebs/service_search.aspx"
USER_AGENT = "Mozilla/5.0 (OpenClaw; cpc-direct-stations)" USER_AGENT = "Mozilla/5.0 (OpenClaw; cpc-direct-stations)"
CACHE_FILE = Path.home() / ".cache" / "cpc_stations_coords.json"
FUEL_DB_FILE = Path(__file__).parent / "fuel_stations.json" # 離線加油站座標庫
CITY_MAP = { CITY_MAP = {
"台北市": "A ", "台北市": "A ",
"新北市": "B ", "新北市": "B ",
@@ -140,76 +135,6 @@ def extract_hidden_fields(html: str) -> Dict[str, str]:
return fields return fields
def load_coords_cache() -> Dict[str, Dict[str, float]]:
"""載入座標快取。"""
if CACHE_FILE.exists():
try:
with open(CACHE_FILE, "r", encoding="utf-8") as f:
return json.load(f)
except Exception:
pass
return {}
def save_coords_cache(cache: Dict[str, Dict[str, float]]) -> None:
"""儲存座標快取。"""
CACHE_FILE.parent.mkdir(parents=True, exist_ok=True)
with open(CACHE_FILE, "w", encoding="utf-8") as f:
json.dump(cache, f, ensure_ascii=False, indent=2)
def load_fuel_db() -> Dict[str, Dict[str, float]]:
"""載入離線加油站座標資料庫來源OSM"""
if FUEL_DB_FILE.exists():
try:
with open(FUEL_DB_FILE, "r", encoding="utf-8") as f:
return json.load(f)
except Exception:
pass
return {}
def query_fuel_db(station_name: str, city: str, district: str, fuel_db: Dict) -> Optional[Dict[str, float]]:
"""從離線資料庫查詢加油站座標。"""
if not fuel_db:
return None
# 清理站名(移除代號)
clean_name = re.sub(r'\s+[A-Z]\d+[A-Z]?\s*$', '', station_name).strip()
# 嘗試多種查詢鍵
queries = [
clean_name,
f"台灣中油 {clean_name}",
f"中油 {clean_name}",
f"{clean_name}",
f"台灣中油 {clean_name}",
f"中油 {clean_name}",
]
if city:
queries.extend([
f"{city}{clean_name}",
f"{city}{clean_name}",
f"{clean_name} ({city})",
])
for q in queries:
if q in fuel_db:
return fuel_db[q]
# 模糊匹配
for key, value in fuel_db.items():
if clean_name in key or key in clean_name:
return value
return None
def make_cache_key(station: Dict[str, str]) -> str:
"""產生快取鍵值:使用站名+縣市+完整地址。"""
return f"{station['站名']}|{station['縣市']}|{station['鄉鎮區']}|{station['地址']}"
def geocode(address: str, station_name: str, city: str, district: str) -> Optional[Dict[str, float]]: def geocode(address: str, station_name: str, city: str, district: str) -> Optional[Dict[str, float]]:
"""使用 Nominatim (OpenStreetMap) API 將地址轉換為座標。 """使用 Nominatim (OpenStreetMap) API 將地址轉換為座標。
@@ -238,7 +163,7 @@ def geocode(address: str, station_name: str, city: str, district: str) -> Option
} }
url = f"https://nominatim.openstreetmap.org/search?{urllib.parse.urlencode(params)}" url = f"https://nominatim.openstreetmap.org/search?{urllib.parse.urlencode(params)}"
req = urllib.request.Request(url, headers={"User-Agent": user_agent}) req = urllib.request.Request(url, headers={"User-Agent": user_agent})
with urllib.request.urlopen(req, timeout=15) as response: with urllib.request.urlopen(req, timeout=10) as response:
data = json.loads(response.read().decode("utf-8")) data = json.loads(response.read().decode("utf-8"))
if data: if data:
return {"lat": float(data[0]["lat"]), "lon": float(data[0]["lon"])} return {"lat": float(data[0]["lat"]), "lon": float(data[0]["lon"])}
@@ -247,45 +172,9 @@ def geocode(address: str, station_name: str, city: str, district: str) -> Option
return None return None
def add_coordinates(stations: List[Dict[str, str]], delay: float = 1.0, workers: int = 2) -> List[Dict[str, str]]: def add_coordinates(stations: List[Dict[str, str]], delay: float = 1.0) -> List[Dict[str, str]]:
"""為站點資料加上座標資訊(快取 -> OSM資料庫 -> Nominatim """為站點資料加上座標資訊"""
Args:
stations: 站點列表
delay: 每個查詢間隔秒數Nominatim 限制約 1 秒)
workers: 並發查詢執行緒數(預設 2避免 Nominatim 限制)
"""
cache = load_coords_cache()
fuel_db = load_fuel_db()
if fuel_db:
print(f"載入離線資料庫: {len(fuel_db)}", file=sys.stderr)
missing_coords = []
# 先從快取讀取,再試離線資料庫
for station in stations: for station in stations:
key = make_cache_key(station)
if key in cache:
station["緯度"] = str(cache[key]["lat"])
station["經度"] = str(cache[key]["lon"])
else:
# 嘗試離線資料庫
coords = query_fuel_db(station["站名"], station["縣市"], station["鄉鎮區"], fuel_db)
if coords:
station["緯度"] = str(coords["lat"])
station["經度"] = str(coords["lon"])
# 寫入快取
cache[key] = coords
else:
station["緯度"] = ""
station["經度"] = ""
missing_coords.append(station)
# 並發查詢缺少的座標(使用 Nominatim
if missing_coords:
print(f"快取命中: {len(stations) - len(missing_coords)}/{len(stations)},需 API 查詢: {len(missing_coords)}", file=sys.stderr)
def query_one(station: Dict[str, str]) -> tuple[Dict[str, str], Optional[Dict[str, float]]]:
full_address = f"{station['縣市']}{station['鄉鎮區']}{station['地址']}" full_address = f"{station['縣市']}{station['鄉鎮區']}{station['地址']}"
coords = geocode( coords = geocode(
address=full_address, address=full_address,
@@ -293,21 +182,13 @@ def add_coordinates(stations: List[Dict[str, str]], delay: float = 1.0, workers:
city=station["縣市"], city=station["縣市"],
district=station["鄉鎮區"], district=station["鄉鎮區"],
) )
time.sleep(delay) # 避免 Nominatim 請求過快
return (station, coords)
with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as executor:
futures = {executor.submit(query_one, s): s for s in missing_coords}
for future in concurrent.futures.as_completed(futures):
station, coords = future.result()
if coords: if coords:
station["緯度"] = str(coords["lat"]) station["緯度"] = str(coords["lat"])
station["經度"] = str(coords["lon"]) station["經度"] = str(coords["lon"])
key = make_cache_key(station) else:
cache[key] = coords station["緯度"] = ""
station["經度"] = ""
save_coords_cache(cache) time.sleep(delay) # 避免 Nominatim 請求過快
return stations return stations
@@ -436,8 +317,7 @@ def main() -> None:
parser.add_argument("--format", choices=["text", "json", "csv"], default="text") parser.add_argument("--format", choices=["text", "json", "csv"], default="text")
parser.add_argument("--out", help="輸出到檔案路徑") parser.add_argument("--out", help="輸出到檔案路徑")
parser.add_argument("--coords", action="store_true", help="加上座標資訊(使用 Nominatim OSM API") parser.add_argument("--coords", action="store_true", help="加上座標資訊(使用 Nominatim OSM API")
parser.add_argument("--delay", type=float, default=1.0, help="座標查詢間隔秒數(預設 1.0") parser.add_argument("--delay", type=float, default=1.0, help="座標查詢間隔秒數(預設 1.0Nominatim 限制")
parser.add_argument("--workers", type=int, default=2, help="並發查詢執行緒數(預設 2避免 Nominatim 限制)")
args = parser.parse_args() args = parser.parse_args()
html = build_query_html(city=args.city, keyword=args.keyword) html = build_query_html(city=args.city, keyword=args.keyword)
@@ -445,7 +325,7 @@ def main() -> None:
stations = filter_stations(stations, args.keyword) stations = filter_stations(stations, args.keyword)
if args.coords: if args.coords:
stations = add_coordinates(stations, delay=args.delay, workers=args.workers) stations = add_coordinates(stations, delay=args.delay)
if args.format == "json": if args.format == "json":
output_json(stations, args.out) output_json(stations, args.out)

View File

@@ -1,189 +0,0 @@
#!/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()

File diff suppressed because it is too large Load Diff