stocks-query/scripts/yahoo-dividend-browserless.py

#!/usr/bin/env python3
"""
用 Browserless 抓 Yahoo 股利政策頁，輸出精簡摘要

用法：
  python3 skills/stocks-query/scripts/yahoo-dividend-browserless.py 2330
  python3 skills/stocks-query/scripts/yahoo-dividend-browserless.py 0050

環境變數（可覆蓋）：
  BROWSERLESS_ENDPOINT=http://192.168.42.124:13000
  BROWSERLESS_TOKEN=6R0W53R135510
"""

import json
import os
import re
import sys
import urllib.error
import urllib.request
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/dividend"
    api = f"{ENDPOINT}/content?token={TOKEN}"
    payload = {
        "url": url,
        "gotoOptions": {"waitUntil": "domcontentloaded", "timeout": 60000},
    }

    req = urllib.request.Request(
        api,
        data=json.dumps(payload).encode("utf-8"),
        headers={"Content-Type": "application/json"},
        method="POST",
    )

    try:
        with urllib.request.urlopen(req, timeout=90) as resp:
            text = resp.read().decode("utf-8", errors="replace")
            status = resp.getcode()
    except urllib.error.HTTPError as e:
        body = e.read().decode("utf-8", errors="replace") if hasattr(e, "read") else str(e)
        raise RuntimeError(f"Browserless HTTP {e.code}: {body[:200]}") from e

    if status != 200:
        raise RuntimeError(f"Browserless HTTP {status}: {text[:200]}")
    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_dividend(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\)", "-").strip()

    # 殖利率
    dividend_yield = pick(html, r">現金殖利率</span><span[^>]*>([\d.]+%?)")

    # 年度股利統計（從 JSON-like 嵌入或 table 抓取）
    # 嘗試從頁面的 script data 抓股利歷史
    rows = []

    # 方法1：抓 dividendData JSON
    json_match = re.search(r'"dividendData"\s*:\s*(\[[\s\S]*?\])\s*[,}]', html)
    if json_match:
        try:
            raw_data = json.loads(json_match.group(1))
            for item in raw_data[:5]:  # 最近 5 年
                year = item.get("year", item.get("formattedYear", "-"))
                cash = item.get("cashDividend", item.get("cash", 0))
                stock = item.get("stockDividend", item.get("stock", 0))
                total = item.get("totalDividend", 0)
                if total == 0:
                    try:
                        total = float(cash) + float(stock)
                    except (ValueError, TypeError):
                        total = "-"
                rows.append({
                    "year": str(year),
                    "cash": str(cash),
                    "stock": str(stock),
                    "total": str(total),
                })
        except (json.JSONDecodeError, KeyError):
            pass

    # 方法2：用 regex 抓 table rows
    if not rows:
        # 常見結構：年度、現金股利、股票股利、合計
        row_pattern = re.compile(
            r'<tr[^>]*>.*?<td[^>]*>(1\d{2,3})</td>'
            r'.*?<td[^>]*>([\d.]+)</td>'
            r'.*?<td[^>]*>([\d.]+)</td>'
            r'.*?<td[^>]*>([\d.]+)</td>',
            re.S,
        )
        for m in row_pattern.finditer(html):
            rows.append({
                "year": m.group(1),
                "cash": m.group(2),
                "stock": m.group(3),
                "total": m.group(4),
            })
            if len(rows) >= 5:
                break

    # 方法3：抓除息日相關
    ex_date = pick(html, r">除息日</span><span[^>]*>([^<]+)")
    pay_date = pick(html, r">發放日</span><span[^>]*>([^<]+)")
    cash_div = pick(html, r">現金股利</span><span[^>]*>([\d.]+)")
    stock_div = pick(html, r">股票股利</span><span[^>]*>([\d.]+)")

    return {
        "name": name,
        "stock_no": stock_no,
        "dividend_yield": dividend_yield,
        "cash_div": cash_div,
        "stock_div": stock_div,
        "ex_date": ex_date,
        "pay_date": pay_date,
        "history": rows[:5],
    }


def print_summary(d: dict) -> None:
    print(f"{d['name']}（{d['stock_no']}）｜股利摘要")
    if d["dividend_yield"] != "-":
        print(f"• 現金殖利率：{d['dividend_yield']}")
    if d["cash_div"] != "-":
        print(f"• 現金股利：{d['cash_div']}")
    if d["stock_div"] != "-":
        print(f"• 股票股利：{d['stock_div']}")
    if d["ex_date"] != "-":
        print(f"• 除息日：{d['ex_date']}")
    if d["pay_date"] != "-":
        print(f"• 發放日：{d['pay_date']}")

    if d["history"]:
        print("• 近年股利：")
        for r in d["history"]:
            print(f"  - {r['year']}年：現金 {r['cash']}／股票 {r['stock']}／合計 {r['total']}")
    else:
        print("• 近年股利：無資料")


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