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) <noreply@anthropic.com>
This commit is contained in:
65
CLAUDE.md
Normal file
65
CLAUDE.md
Normal file
@@ -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
|
||||
273
cpc_direct_stations.py
Executable file
273
cpc_direct_stations.py
Executable file
@@ -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()
|
||||
Reference in New Issue
Block a user