Timmy
1c432a251e
- Move cpc_direct_stations.py to scripts/ - Rename CLAUDE.md to SKILL.md - Add .gitignore for Python cache files Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
340 lines
11 KiB
Python
Executable File
340 lines
11 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""抓取台灣中油官方查詢系統中的直營加油站資料。
|
||
|
||
資料來源：
|
||
- https://www.cpc.com.tw/cp.aspx?n=34
|
||
- 實際查詢頁：https://vipmbr.cpc.com.tw/mbwebs/service_search.aspx
|
||
|
||
用法：
|
||
python3 scripts/cpc_direct_stations.py
|
||
python3 scripts/cpc_direct_stations.py --format json
|
||
python3 scripts/cpc_direct_stations.py --format csv
|
||
python3 scripts/cpc_direct_stations.py --city 台北市
|
||
python3 scripts/cpc_direct_stations.py --city 新北市 --keyword 中山
|
||
python3 scripts/cpc_direct_stations.py --out /tmp/cpc-direct.json
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
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, Optional
|
||
|
||
BASE_URL = "https://vipmbr.cpc.com.tw/mbwebs/service_search.aspx"
|
||
USER_AGENT = "Mozilla/5.0 (OpenClaw; cpc-direct-stations)"
|
||
CITY_MAP = {
|
||
"台北市": "A ",
|
||
"新北市": "B ",
|
||
"基隆市": "C ",
|
||
"宜蘭縣": "D ",
|
||
"桃園市": "E ",
|
||
"新竹市": "F ",
|
||
"新竹縣": "G ",
|
||
"苗栗縣": "H ",
|
||
"台中市": "I ",
|
||
"彰化縣": "K ",
|
||
"南投縣": "L ",
|
||
"雲林縣": "M ",
|
||
"嘉義市": "N ",
|
||
"嘉義縣": "O ",
|
||
"台南市": "P ",
|
||
"高雄市": "R ",
|
||
"屏東縣": "T ",
|
||
"台東縣": "U ",
|
||
"花蓮縣": "V ",
|
||
"澎湖縣": "W ",
|
||
"連江縣": "X ",
|
||
"金門縣": "Y ",
|
||
"郵政信箱": "Z ",
|
||
"全部縣市": "全部縣市",
|
||
}
|
||
|
||
|
||
class StationTableParser(HTMLParser):
|
||
def __init__(self) -> None:
|
||
super().__init__()
|
||
self.in_table = False
|
||
self.in_row = False
|
||
self.in_cell = False
|
||
self.current_cell_parts: List[str] = []
|
||
self.current_row: List[str] = []
|
||
self.rows: List[List[str]] = []
|
||
self._table_depth = 0
|
||
self._skip_header = True
|
||
|
||
def handle_starttag(self, tag: str, attrs):
|
||
attrs_dict = dict(attrs)
|
||
if tag == "table" and attrs_dict.get("id") == "MyGridView1":
|
||
self.in_table = True
|
||
self._table_depth = 1
|
||
return
|
||
if self.in_table and tag == "table":
|
||
self._table_depth += 1
|
||
if not self.in_table:
|
||
return
|
||
if tag == "tr":
|
||
self.in_row = True
|
||
self.current_row = []
|
||
elif tag in {"td", "th"} and self.in_row:
|
||
self.in_cell = True
|
||
self.current_cell_parts = []
|
||
elif tag == "br" and self.in_cell:
|
||
self.current_cell_parts.append("\n")
|
||
|
||
def handle_endtag(self, tag: str):
|
||
if not self.in_table:
|
||
return
|
||
if tag in {"td", "th"} and self.in_cell:
|
||
value = unescape("".join(self.current_cell_parts))
|
||
value = re.sub(r"\s+", " ", value.replace("\xa0", " ")).strip()
|
||
value = value.replace(" \n ", "\n").replace("\n ", "\n").replace(" \n", "\n")
|
||
self.current_row.append(value)
|
||
self.in_cell = False
|
||
self.current_cell_parts = []
|
||
elif tag == "tr" and self.in_row:
|
||
if self.current_row:
|
||
if self._skip_header:
|
||
self._skip_header = False
|
||
else:
|
||
self.rows.append(self.current_row)
|
||
self.in_row = False
|
||
elif tag == "table":
|
||
self._table_depth -= 1
|
||
if self._table_depth <= 0:
|
||
self.in_table = False
|
||
|
||
def handle_data(self, data: str):
|
||
if self.in_table and self.in_cell:
|
||
self.current_cell_parts.append(data)
|
||
|
||
|
||
def fetch(url: str, data: Dict[str, str] | None = None) -> str:
|
||
payload = None
|
||
if data is not None:
|
||
payload = urllib.parse.urlencode(data).encode("utf-8")
|
||
req = urllib.request.Request(url, data=payload, headers={"User-Agent": USER_AGENT})
|
||
with urllib.request.urlopen(req, timeout=60) as response:
|
||
return response.read().decode("utf-8", "ignore")
|
||
|
||
|
||
def extract_hidden_fields(html: str) -> Dict[str, str]:
|
||
fields = {}
|
||
for name in ["__VIEWSTATE", "__EVENTVALIDATION", "__VIEWSTATEGENERATOR"]:
|
||
match = re.search(rf'name="{re.escape(name)}"[^>]*value="([^"]*)"', html)
|
||
if not match:
|
||
raise RuntimeError(f"找不到必要欄位：{name}")
|
||
fields[name] = match.group(1)
|
||
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 != "全部縣市":
|
||
raise RuntimeError(f"不支援的縣市：{city}")
|
||
|
||
html = fetch(BASE_URL)
|
||
fields = extract_hidden_fields(html)
|
||
|
||
fields.update(
|
||
{
|
||
"__EVENTTARGET": "rbGroup2",
|
||
"__EVENTARGUMENT": "",
|
||
"TypeGroup": "rbGroup2",
|
||
"ddlCity": city_value,
|
||
"ddlSubCity": "全部鄉鎮區",
|
||
"tbKWQuery": keyword,
|
||
"TimeGroup": "rbGroup4",
|
||
}
|
||
)
|
||
html = fetch(BASE_URL, fields)
|
||
fields = extract_hidden_fields(html)
|
||
|
||
fields.update(
|
||
{
|
||
"__EVENTTARGET": "",
|
||
"__EVENTARGUMENT": "",
|
||
"TypeGroup": "rbGroup2",
|
||
"ddlCity": city_value,
|
||
"ddlSubCity": "全部鄉鎮區",
|
||
"tbKWQuery": keyword,
|
||
"TimeGroup": "rbGroup4",
|
||
"btnQuery": "查 詢",
|
||
}
|
||
)
|
||
return fetch(BASE_URL, fields)
|
||
|
||
|
||
def parse_stations(html: str) -> List[Dict[str, str]]:
|
||
parser = StationTableParser()
|
||
parser.feed(html)
|
||
|
||
stations = []
|
||
for row in parser.rows:
|
||
if len(row) < 7:
|
||
continue
|
||
station_name = row[3].split("\n")[0].strip()
|
||
station_code = ""
|
||
row3_parts = [part.strip() for part in row[3].split("\n") if part.strip()]
|
||
if len(row3_parts) > 1:
|
||
station_code = row3_parts[1]
|
||
|
||
stations.append(
|
||
{
|
||
"縣市": row[0],
|
||
"鄉鎮區": row[1],
|
||
"類別": row[2],
|
||
"站名": station_name,
|
||
"站代號": station_code,
|
||
"地址": row[4],
|
||
"電話": row[5],
|
||
"營業時間": row[6],
|
||
}
|
||
)
|
||
return stations
|
||
|
||
|
||
def filter_stations(stations: List[Dict[str, str]], keyword: str = "") -> List[Dict[str, str]]:
|
||
if not keyword:
|
||
return stations
|
||
needle = keyword.strip().lower()
|
||
return [
|
||
station
|
||
for station in stations
|
||
if needle in json.dumps(station, ensure_ascii=False).lower()
|
||
]
|
||
|
||
|
||
def output_json(stations: List[Dict[str, str]], out_path: str | None) -> None:
|
||
text = json.dumps(stations, ensure_ascii=False, indent=2)
|
||
if out_path:
|
||
with open(out_path, "w", encoding="utf-8") as f:
|
||
f.write(text)
|
||
else:
|
||
print(text)
|
||
|
||
|
||
def output_csv(stations: List[Dict[str, str]], out_path: str | None) -> None:
|
||
fieldnames = ["縣市", "鄉鎮區", "類別", "站名", "站代號", "地址", "電話", "營業時間", "緯度", "經度"]
|
||
if out_path:
|
||
f = open(out_path, "w", encoding="utf-8-sig", newline="")
|
||
else:
|
||
f = sys.stdout
|
||
try:
|
||
writer = csv.DictWriter(f, fieldnames=fieldnames)
|
||
writer.writeheader()
|
||
writer.writerows(stations)
|
||
finally:
|
||
if out_path:
|
||
f.close()
|
||
|
||
|
||
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['營業時間']}{coord}"
|
||
)
|
||
text = "\n".join(lines)
|
||
if out_path:
|
||
with open(out_path, "w", encoding="utf-8") as f:
|
||
f.write(text)
|
||
else:
|
||
print(text)
|
||
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(description="抓取台灣中油直營加油站資料")
|
||
parser.add_argument("--city", default="全部縣市", help="縣市，例如：台北市、新北市、台中市")
|
||
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":
|
||
output_csv(stations, args.out)
|
||
else:
|
||
output_text(stations, args.out)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|