commit 36b12dbe816d5cb2015b44aba253234e51c0d9b2 Author: Timmy Date: Fri Mar 27 08:15:17 2026 +0800 Initial commit: Add CPC direct-operated stations scraper Script for scraping Taiwan CPC (中油) official direct-operated gas station information from vipmbr.cpc.com.tw Co-Authored-By: Claude Opus 4.6 (1M context) diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..026eec8 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,65 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +## Overview + +This skill contains a single standalone script for scraping Taiwan CPC (中油) official direct-operated gas station information. The script fetches data from `https://vipmbr.cpc.com.tw/mbwebs/service_search.aspx` using form POST requests and parses the HTML response. + +## Architecture + +The script uses only Python stdlib (no external dependencies): + +- `HTMLParser` - Custom `StationTableParser` class extracts data from the `#MyGridView1` table +- `urllib` - Handles HTTP requests with proper User-Agent and form data encoding +- `__VIEWSTATE`/`__EVENTVALIDATION` - ASP.NET hidden fields are extracted and submitted in POST requests + +The query flow: +1. GET initial page to extract ASP.NET hidden fields +2. POST with `TypeGroup=rbGroup2` to select "direct-operated stations" (直營站) +3. Extract hidden fields again +4. POST with `btnQuery=查詢` to submit the query +5. Parse HTML response table + +## Common Commands + +### Get all direct-operated stations (default text output) +```bash +python3 skills/c/cpc_direct_stations.py +``` + +### Query by city +```bash +python3 skills/c/cpc_direct_stations.py --city 台北市 +``` + +### Filter by keyword (station name or address) +```bash +python3 skills/c/cpc_direct_stations.py --city 新北市 --keyword 中山 +``` + +### Export as JSON +```bash +python3 skills/c/cpc_direct_stations.py --city 台北市 --format json +``` + +### Export as CSV (with BOM for Excel compatibility) +```bash +python3 skills/c/cpc_direct_stations.py --format csv --out cpc_stations.csv +``` + +## City Codes + +The `CITY_MAP` dict maps Chinese city names to CPC internal codes (e.g., `"台北市": "A "`). Supported cities include all Taiwan counties and cities. + +## Output Schema + +Each station record contains: +- `縣市` - County/City +- `鄉鎮區` - District +- `類別` - Category (e.g., "直營站") +- `站名` - Station name +- `站代號` - Station code +- `地址` - Address +- `電話` - Phone number +- `營業時間` - Business hours diff --git a/cpc_direct_stations.py b/cpc_direct_stations.py new file mode 100755 index 0000000..87e32e6 --- /dev/null +++ b/cpc_direct_stations.py @@ -0,0 +1,273 @@ +#!/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 urllib.parse +import urllib.request +from html import unescape +from html.parser import HTMLParser +from typing import Dict, List + +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 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): + lines.append( + f"{idx}. {station['站名']}（{station['站代號']}）｜{station['縣市']}{station['鄉鎮區']}｜{station['地址']}｜{station['電話']}｜{station['營業時間']}" + ) + 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="輸出到檔案路徑") + 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.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()