167 lines
4.5 KiB
Python
167 lines
4.5 KiB
Python
#!/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()
|