Add coordinate cache to avoid redundant API calls

- Cache stored in ~/.cache/cpc_stations_coords.json
- Cache key based on station name + city + district + address
- Shows cache hit/miss statistics on stderr
- Subsequent runs are instant for cached stations

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

View File

@@ -19,6 +19,7 @@ from __future__ import annotations
import argparse import argparse
import csv import csv
import json import json
import os
import re import re
import sys import sys
import time import time
@@ -26,10 +27,12 @@ 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"
CITY_MAP = { CITY_MAP = {
"台北市": "A ", "台北市": "A ",
"新北市": "B ", "新北市": "B ",
@@ -135,6 +138,29 @@ 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 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 將地址轉換為座標。
@@ -173,8 +199,26 @@ def geocode(address: str, station_name: str, city: str, district: str) -> Option
def add_coordinates(stations: List[Dict[str, str]], delay: float = 1.0) -> List[Dict[str, str]]: def add_coordinates(stations: List[Dict[str, str]], delay: float = 1.0) -> List[Dict[str, str]]:
"""為站點資料加上座標資訊。""" """為站點資料加上座標資訊(使用快取)"""
cache = load_coords_cache()
cache_updated = False
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:
station["緯度"] = ""
station["經度"] = ""
missing_coords.append(station)
# 查詢缺少的座標
if missing_coords:
print(f"快取命中: {len(stations) - len(missing_coords)}/{len(stations)},需查詢: {len(missing_coords)}", file=sys.stderr)
for station in missing_coords:
full_address = f"{station['縣市']}{station['鄉鎮區']}{station['地址']}" full_address = f"{station['縣市']}{station['鄉鎮區']}{station['地址']}"
coords = geocode( coords = geocode(
address=full_address, address=full_address,
@@ -185,10 +229,14 @@ def add_coordinates(stations: List[Dict[str, str]], delay: float = 1.0) -> List[
if coords: if coords:
station["緯度"] = str(coords["lat"]) station["緯度"] = str(coords["lat"])
station["經度"] = str(coords["lon"]) station["經度"] = str(coords["lon"])
else: key = make_cache_key(station)
station["緯度"] = "" cache[key] = coords
station["經度"] = "" cache_updated = True
time.sleep(delay) # 避免 Nominatim 請求過快 time.sleep(delay) # 避免 Nominatim 請求過快
if cache_updated:
save_coords_cache(cache)
return stations return stations