stocks-query/scripts/twse-monthly-revenue.py

#!/usr/bin/env python3
"""
查詢個股月營收（Yahoo Finance TW + Browserless）

用法：
  python3 skills/stocks-query/scripts/twse-monthly-revenue.py 2330
  python3 skills/stocks-query/scripts/twse-monthly-revenue.py 3481

資料來源：Yahoo 奇摩股市（營收頁），營收單位：千元
"""

from __future__ import annotations

import json
import os
import re
import subprocess
import sys
from pathlib import Path


def load_dotenv() -> None:
    """載入 .env（優先目前目錄，其次 workspace 根目錄）。"""
    candidates = [
        Path.cwd() / ".env",
        Path(__file__).resolve().parents[3] / ".env",
    ]
    for env_path in candidates:
        if not env_path.exists():
            continue
        for raw in env_path.read_text(encoding="utf-8").splitlines():
            line = raw.strip()
            if not line or line.startswith("#") or "=" not in line:
                continue
            k, v = line.split("=", 1)
            k = k.strip()
            v = v.strip().strip('"').strip("'")
            if k and k not in os.environ:
                os.environ[k] = v
        break


load_dotenv()
ENDPOINT = os.getenv("BROWSERLESS_ENDPOINT")
TOKEN = os.getenv("BROWSERLESS_TOKEN")
if not ENDPOINT or not TOKEN:
    raise RuntimeError("缺少必要環境變數：BROWSERLESS_ENDPOINT / BROWSERLESS_TOKEN")
STOCK_NO = sys.argv[1] if len(sys.argv) > 1 else "2330"


def fetch_html(stock_no: str) -> str:
    url = f"https://tw.stock.yahoo.com/quote/{stock_no}.TW/revenue"
    api = f"{ENDPOINT}/content?token={TOKEN}"
    payload = json.dumps(
        {
            "url": url,
            "gotoOptions": {"waitUntil": "domcontentloaded", "timeout": 60000},
        },
        ensure_ascii=False,
    )

    cmd = [
        "curl", "-sS", "--max-time", "90",
        "-X", "POST", api,
        "-H", "Content-Type: application/json",
        "-d", payload,
    ]
    try:
        text = subprocess.check_output(cmd, text=True, timeout=95)
    except subprocess.CalledProcessError as e:
        raise RuntimeError(f"Browserless 呼叫失敗: {e}") from e
    except subprocess.TimeoutExpired as e:
        raise RuntimeError("Browserless 呼叫逾時") from e

    if "TimeoutError" in text:
        raise RuntimeError("Browserless 導航逾時（TimeoutError）")
    return text


def pick(html: str, pattern: str, default: str = "-") -> str:
    m = re.search(pattern, html, flags=re.S)
    return m.group(1) if m else default


def parse_revenue(html: str, stock_no: str) -> dict:
    name = pick(html, r'property="og:title" content="([^"(]+)\([0-9]+\.TW\)', "").strip()
    if not name:
        name = pick(html, r"<title>([^<(]+)\([0-9]+\.TW\)", stock_no).strip()

    # 找 JSON 中的月營收資料（data 陣列內含 revenueMoM 欄位的物件）
    obj_pat = re.compile(r'\{[^{}]*"revenueMoM":[^{}]*\}')
    matches = obj_pat.findall(html)

    if not matches:
        raise RuntimeError("抓不到月營收資料（頁面結構可能變更）")

    rows = []
    for m in matches:
        try:
            obj = json.loads(m.replace("\\u002F", "/"))
            rows.append(obj)
        except json.JSONDecodeError:
            continue

    if not rows:
        raise RuntimeError("無法解析月營收 JSON 資料")

    # 依日期排序（最新在前）
    rows.sort(key=lambda r: r.get("date", ""), reverse=True)

    return {
        "name": name,
        "stock_no": stock_no,
        "latest": rows[0],
        "history": rows[:6],
    }


def fmt_pct(val) -> str:
    """將小數比例轉為百分比字串。"""
    try:
        return f"{float(val) * 100:+.2f}%"
    except (ValueError, TypeError):
        return "-"


def fmt_rev(val: str) -> str:
    """千元營收加上千分位格式。"""
    cleaned = str(val).replace(",", "")
    try:
        return f"{int(cleaned):,}"
    except (ValueError, TypeError):
        return val


def print_summary(d: dict) -> None:
    name = d["name"]
    stock_no = d["stock_no"]
    latest = d["latest"]
    date = latest.get("date", "-")

    print(f"{name}（{stock_no}）｜月營收摘要")
    print(f"• 期間：{date}")
    print(f"• 當月營收：{fmt_rev(latest.get('revenue', '-'))} 千元")
    print(f"• 去年同月營收：{fmt_rev(latest.get('lastYearRevenue', '-'))} 千元")
    print(f"• MoM（月增率）：{fmt_pct(latest.get('revenueMoM'))}")
    print(f"• YoY（年增率）：{fmt_pct(latest.get('revenueYoY'))}")
    print(f"• 當月累計營收：{fmt_rev(latest.get('revenueAcc', '-'))} 千元")
    print(f"• 累計 YoY：{fmt_pct(latest.get('revenueYoYAcc'))}")

    # 近期趨勢
    history = d.get("history", [])
    if len(history) > 1:
        print(f"\n近 {len(history)} 月趨勢：")
        for row in history:
            rd = row.get("date", "-")
            rv = fmt_rev(row.get("revenue", "-"))
            mom = fmt_pct(row.get("revenueMoM"))
            yoy = fmt_pct(row.get("revenueYoY"))
            print(f"  {rd}｜{rv} 千元｜MoM {mom}｜YoY {yoy}")


if __name__ == "__main__":
    try:
        html = fetch_html(STOCK_NO)
        data = parse_revenue(html, STOCK_NO)
        print_summary(data)
    except Exception as e:
        print(f"取得失敗：{e}", file=sys.stderr)
        sys.exit(1)