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

#!/usr/bin/env python3
"""
用 Browserless 抓 Yahoo 股市頁面，輸出精簡報價摘要（省 token 版）

用法：
  python3 stocks/yahoo-quote-browserless.py 3481
  python3 stocks/yahoo-quote-browserless.py 2330

環境變數（可覆蓋）：
  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")
STOCK_NO = sys.argv[1] if len(sys.argv) > 1 else "3481"


def fetch_html(stock_no: str) -> str:
    url = f"https://tw.stock.yahoo.com/quote/{stock_no}.TW"
    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_quote(html: str, stock_no: str) -> dict:
    # 名稱容錯：有些股票 h1 結構會不一樣
    name = pick(html, r"<h1[^>]*>([^<(]+)", "-").strip()
    if not name or name in ("-", "Yahoo股市"):
        name = pick(html, r'property="og:title" content="([^"(]+)\([0-9]+\.TW\)', "").strip()
    if not name or name in ("-", "Yahoo股市"):
        name = pick(html, r"<title>([^<(]+)\([0-9]+\.TW\)", "-").strip()

    # 價格容錯：支援千分位與不同 class 變化
    num_pat = r"([\d,]+\.?\d*)"
    price = pick(html, rf"Fz\(32px\)[^>]*>{num_pat}<")

    # 漲跌先抓頁面值，後續再用（現價-昨收）覆寫，避免符號抓錯
    updown = pick(html, rf"Fz\(20px\)[^>]*>([\d\.\+\-]+)<", "-")
    if updown == "-":
        updown = pick(html, rf"Fz\(20px\)[^>]*>[\s\S]*?</span>([\d\.\+\-]+)<", "0.00")

    pct = pick(html, r"\(([+-]?[0-9]+\.[0-9]+%)\)</span>", "0.00%")

    open_price = pick(html, rf">開盤</span><span[^>]*>{num_pat}")
    high = pick(html, rf">最高</span><span[^>]*>{num_pat}")
    low = pick(html, rf">最低</span><span[^>]*>{num_pat}")
    prev = pick(html, rf">昨收</span><span[^>]*>{num_pat}")
    avg = pick(html, rf">均價</span><span[^>]*>{num_pat}")

    volume_lots = pick(
        html,
        r'Mb\(4px\)">([0-9,]+)</span><span class="Fz\(12px\) C\(\$c-icon\)">成交量',
        "0",
    )
    volume_shares = "{:,.0f}".format(float(volume_lots.replace(",", "")) * 1000)

    bids = re.findall(r"Fw\(n\)[^>]*>([\d,]+\.?\d*)</span>", html)
    best_bid = bids[0] if len(bids) >= 1 else "-"
    best_ask = bids[5] if len(bids) >= 6 else "-"

    time_raw = pick(html, r"開盤 \| ([0-9/ :]+) 更新")
    time_fmt = "-" if time_raw == "-" else f"{time_raw.replace('/', '-')}（台北）"
    market = "未知" if time_raw == "-" else "開盤中"

    # 以現價與昨收重算漲跌，避免頁面箭頭造成正負號誤判
    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,
        "stock_no": stock_no,
        "price": price,
        "updown": updown,
        "pct": pct,
        "open": open_price,
        "high": high,
        "low": low,
        "prev": prev,
        "volume_lots": volume_lots,
        "volume_shares": volume_shares,
        "avg": avg,
        "best_bid": best_bid,
        "best_ask": best_ask,
        "market": market,
        "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['stock_no']}）")
    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['volume_shares']}（Yahoo 顯示 {q['volume_lots']} 張，換算股數）")
    print(f"• 均價：{q['avg']}")
    print(f"• 最佳買賣：{q['best_bid']}／{q['best_ask']}")
    print(f"• 市場狀態：{q['market']}")
    print(f"• 資料時間：{q['time']}")


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