Initial commit: CPC fuel price CLI scraper
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
30
CLAUDE.md
Normal file
30
CLAUDE.md
Normal file
@@ -0,0 +1,30 @@
|
||||
# CLAUDE.md
|
||||
|
||||
This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
|
||||
|
||||
## Overview
|
||||
|
||||
This is a single-file Python CLI (`cpc_price.py`) that scrapes Taiwan CPC's (中油) historical fuel price page and outputs the data in various formats.
|
||||
|
||||
## Running the Script
|
||||
|
||||
```bash
|
||||
python3 cpc_price.py # table output (default)
|
||||
python3 cpc_price.py --ad # convert ROC dates to AD
|
||||
python3 cpc_price.py --latest # show only the most recent record
|
||||
python3 cpc_price.py -f csv # CSV output
|
||||
python3 cpc_price.py -f json --ad # JSON output with AD dates
|
||||
```
|
||||
|
||||
## Dependencies
|
||||
|
||||
```bash
|
||||
pip install requests beautifulsoup4
|
||||
```
|
||||
|
||||
## Key Notes
|
||||
|
||||
- **SSL**: The CPC website has a certificate missing a Subject Key Identifier, so `verify=False` is used with `urllib3` warnings suppressed.
|
||||
- **Date format**: The site uses ROC (Republic of China) calendar dates (e.g. `115/03/16`). The `--ad` flag converts them by adding 1911 to the year.
|
||||
- **Data source**: `https://www.cpc.com.tw/historyprice.aspx?n=2890` — shows roughly the last 7 weekly price adjustments.
|
||||
- **Parsing**: The page has a single `<table>` with headers in the first `<tr>`. If the site restructures its HTML, the `fetch_prices()` function is the place to fix parsing.
|
||||
166
cpc_price.py
Normal file
166
cpc_price.py
Normal file
@@ -0,0 +1,166 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
中油歷史油價查詢 CLI
|
||||
資料來源: https://www.cpc.com.tw/historyprice.aspx?n=2890
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import csv
|
||||
import json
|
||||
import sys
|
||||
from datetime import datetime
|
||||
|
||||
try:
|
||||
import requests
|
||||
from bs4 import BeautifulSoup
|
||||
except ImportError:
|
||||
print("請先安裝依賴套件: pip install requests beautifulsoup4", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
URL = "https://www.cpc.com.tw/historyprice.aspx?n=2890"
|
||||
HEADERS = {
|
||||
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
|
||||
"AppleWebKit/537.36 (KHTML, like Gecko) "
|
||||
"Chrome/120.0.0.0 Safari/537.36"
|
||||
}
|
||||
|
||||
|
||||
def roc_to_ad(roc_date: str) -> str:
|
||||
"""將民國年份 (115/03/16) 轉換為西元年份 (2026/03/16)"""
|
||||
parts = roc_date.strip().split("/")
|
||||
if len(parts) != 3:
|
||||
return roc_date
|
||||
year_ad = int(parts[0]) + 1911
|
||||
return f"{year_ad}/{parts[1]}/{parts[2]}"
|
||||
|
||||
|
||||
def fetch_prices() -> list[dict]:
|
||||
"""抓取油價資料,回傳 list of dict"""
|
||||
try:
|
||||
import urllib3
|
||||
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
||||
resp = requests.get(URL, headers=HEADERS, timeout=15, verify=False)
|
||||
resp.raise_for_status()
|
||||
resp.encoding = "utf-8"
|
||||
except requests.RequestException as e:
|
||||
print(f"連線失敗: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
soup = BeautifulSoup(resp.text, "html.parser")
|
||||
table = soup.find("table")
|
||||
if not table:
|
||||
print("找不到價格表,頁面結構可能已變更。", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
rows = table.find_all("tr")
|
||||
headers = [th.get_text(strip=True) for th in rows[0].find_all(["th", "td"])]
|
||||
if not headers:
|
||||
print("找不到表頭。", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
records = []
|
||||
for row in rows[1:]:
|
||||
cells = [td.get_text(strip=True) for td in row.find_all("td")]
|
||||
if len(cells) == len(headers):
|
||||
records.append(dict(zip(headers, cells)))
|
||||
|
||||
return records
|
||||
|
||||
|
||||
def print_table(records: list[dict], use_ad: bool) -> None:
|
||||
"""以對齊表格格式輸出"""
|
||||
if not records:
|
||||
print("無資料")
|
||||
return
|
||||
|
||||
headers = list(records[0].keys())
|
||||
col_widths = [max(len(h), max(len(r[h]) for r in records)) + 2 for h in headers]
|
||||
|
||||
def fmt_row(values):
|
||||
return " ".join(v.ljust(w) for v, w in zip(values, col_widths))
|
||||
|
||||
separator = " ".join("-" * w for w in col_widths)
|
||||
|
||||
print(fmt_row(headers))
|
||||
print(separator)
|
||||
for r in records:
|
||||
row = list(r.values())
|
||||
if use_ad:
|
||||
row[0] = roc_to_ad(row[0])
|
||||
print(fmt_row(row))
|
||||
|
||||
|
||||
def print_csv(records: list[dict], use_ad: bool) -> None:
|
||||
"""以 CSV 格式輸出"""
|
||||
if not records:
|
||||
return
|
||||
writer = csv.DictWriter(sys.stdout, fieldnames=records[0].keys())
|
||||
writer.writeheader()
|
||||
for r in records:
|
||||
if use_ad:
|
||||
r = dict(r)
|
||||
date_key = list(r.keys())[0]
|
||||
r[date_key] = roc_to_ad(r[date_key])
|
||||
writer.writerow(r)
|
||||
|
||||
|
||||
def print_json(records: list[dict], use_ad: bool) -> None:
|
||||
"""以 JSON 格式輸出"""
|
||||
if use_ad:
|
||||
out = []
|
||||
for r in records:
|
||||
r = dict(r)
|
||||
date_key = list(r.keys())[0]
|
||||
r[date_key] = roc_to_ad(r[date_key])
|
||||
out.append(r)
|
||||
else:
|
||||
out = records
|
||||
print(json.dumps(out, ensure_ascii=False, indent=2))
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="查詢台灣中油歷史油價",
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
epilog="""
|
||||
輸出格式範例:
|
||||
%(prog)s # 表格(預設)
|
||||
%(prog)s -f csv # CSV
|
||||
%(prog)s -f json # JSON
|
||||
%(prog)s --ad # 日期轉西元年
|
||||
%(prog)s -f json --ad # JSON + 西元年
|
||||
""",
|
||||
)
|
||||
parser.add_argument(
|
||||
"-f", "--format",
|
||||
choices=["table", "csv", "json"],
|
||||
default="table",
|
||||
help="輸出格式 (預設: table)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--ad",
|
||||
action="store_true",
|
||||
help="將民國年份轉換為西元年份",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--latest",
|
||||
action="store_true",
|
||||
help="只顯示最新一筆資料",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
records = fetch_prices()
|
||||
|
||||
if args.latest:
|
||||
records = records[:1]
|
||||
|
||||
if args.format == "table":
|
||||
print_table(records, args.ad)
|
||||
elif args.format == "csv":
|
||||
print_csv(records, args.ad)
|
||||
elif args.format == "json":
|
||||
print_json(records, args.ad)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user