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

#!/usr/bin/env python3
"""
用 Browserless 抓 Yahoo 台股大盤（加權指數）即時行情。

用法：
  python3 stocks/yahoo-market-browserless.py
  python3 stocks/yahoo-market-browserless.py ^TWII

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

import os
import re
import sys
import json
import urllib.request
import urllib.error
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")
SYMBOL = sys.argv[1] if len(sys.argv) > 1 else "^TWII"


def fetch_html(symbol: str) -> str:
    url = f"https://tw.stock.yahoo.com/quote/{symbol}"
    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 to_float(num: str) -> float | None:
    try:
        if not num or num == "-":
            return None
        return float(num.replace(",", ""))
    except Exception:
        return None


def parse_market(html: str, symbol: str) -> dict:
    # 名稱：優先取 title（加權指數(^TWII) 走勢圖 - Yahoo股市）
    name = pick(html, r"<title>([^<(]+)\([^)]*\)\s*走勢圖")
    if name == "-":
        name = pick(html, r'property="og:title" content="([^<(]+)\([^)]*\)')

    price = pick(html, r"Fz\(32px\)[^>]*>([\d,]+\.?\d*)<")

    # 漲跌值常出現在百分比前一個 span，且前面有紅綠三角形 span
    updown = pick(
        html,
        r'border-width:[^"]+\"></span>([+\-]?[\d,]+\.?\d*)</span><span[^>]*>\([+\-]?[0-9]+\.[0-9]+%\)',
    )
    pct = pick(html, r"\(([+\-]?[0-9]+\.[0-9]+%)\)</span>")

    open_price = pick(html, r">開盤</span><span[^>]*>([\d,]+\.?\d*)")
    high = pick(html, r">最高</span><span[^>]*>([\d,]+\.?\d*)")
    low = pick(html, r">最低</span><span[^>]*>([\d,]+\.?\d*)")
    prev = pick(html, r">昨收</span><span[^>]*>([\d,]+\.?\d*)")

    amount_100m = pick(html, r">成交金額\(億\)</span><span[^>]*>([\d,]+\.?\d*)")
    time_raw = pick(html, r"(?:開盤|收盤) \| ([0-9/ :]+) 更新")
    time_fmt = "-" if time_raw == "-" else f"{time_raw.replace('/', '-')}（台北）"

    # 以指數與昨收重算漲跌，避免頁面符號誤判
    p = to_float(price)
    pv = to_float(prev)
    if p is not None and pv is not None:
        diff = p - pv
        updown = f"{diff:+.2f}"
        if pv != 0:
            pct = f"{(diff / pv) * 100:+.2f}%"

    return {
        "name": name,
        "symbol": symbol,
        "price": price,
        "updown": updown,
        "pct": pct,
        "open": open_price,
        "high": high,
        "low": low,
        "prev": prev,
        "amount_100m": amount_100m,
        "time": time_fmt,
    }


def sign_num(v: str) -> str:
    if v in ("-", ""):
        return "-"
    return v if v.startswith("-") or v.startswith("+") else f"+{v}"


def print_summary(q: dict) -> None:
    print(f"{q['name']}（{q['symbol']}）")
    print(f"• 指數：{q['price']}")
    print(f"• 漲跌：{sign_num(q['updown'])}（{sign_num(q['pct'])}）")
    print(f"• 開盤／最高／最低：{q['open']}／{q['high']}／{q['low']}")
    print(f"• 昨收：{q['prev']}")
    print(f"• 成交金額：{q['amount_100m']} 億")
    print(f"• 資料時間：{q['time']}")


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