#!/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()