cpc-price-cli/cpc_price.py

#!/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
    from rich.console import Console
    from rich.table import Table
except ImportError:
    print("請先安裝依賴套件: pip install requests beautifulsoup4 rich", 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())
    table = Table(show_header=True, header_style="bold")
    for h in headers:
        table.add_column(h)

    for r in records:
        row = list(r.values())
        if use_ad:
            row[0] = roc_to_ad(row[0])
        table.add_row(*row)

    Console().print(table)


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()