Day 3｜技術不嚇人：整體架構與雲端低維運怎麼做到

把難的留給雲端，把爽的留給你：自動擴縮、低維運、穩穩推播 👍

先放一句人話版摘要：Scheduler 定時叫醒 → Cloud Run 抓 ibon & 比對 → 有變化就 LINE 推播。
指令互動（/watch、/list、/check）走 Webhook，任務狀態放 Firestore。
下面用圖、例子、以及幾段輕量 Python，幫你把「技術名詞」變成「啊原來就這樣」。

一圖看懂：雲端怎麼幫你省心力

[Day3 架構（cron + webhook 強化版）]

• Cloud Scheduler：固定頻率（例如 60s）呼叫 Cloud Run 的 /cron/tick。
• Cloud Run (Flask)：負責兩件事
  1. cron handler：抓取 ibon → 解析 jsonData → 與前次結果比對（diff）→ 視策略推播
  2. LINE webhook：處理 /watch、/list、/check 等指令
• Firestore：保存任務（URL、間隔、狀態、上次快取）與使用者資訊。
• LINE API：推送通知給你（含座位圖連結與各區票況）。
• ibon 活動頁：資料來源（HTML + 內嵌 jsonData）。

口訣：「cron 負責掃、webhook 負責聊、Firestore 記事情、LINE 發報告」。

雲端低維運的秘密（用廚房比喻來說）

• Cloud Run = 智慧廚房：來單才開火（自動擴縮），沒有就休息（省錢）。
• Scheduler = 鬧鐘：固定時間叫你出餐（定時掃頁）。
• Webhook = 點餐窗口：客人跟你說「我要這道」（/watch），你就把菜單記起來（Firestore）。
• Firestore = 訂單板：有哪些桌在等、上次端了什麼菜（上次票況），都寫在上面。
• LINE 推播 = 出菜鈴：菜好了（有變化），叮！請你來拿。

節奏很重要：「不洗版、但不漏接」

[排程時間軸（示意）]

• 只在變化時通知（建議）：避免「每 60 秒就吵你一次」的洗版地獄。
• 設軟性截止（TICK_SOFT_DEADLINE_SEC）：例如一輪最多跑 25 秒、每輪最多 N 個任務，避免超時。
• 任務分批：很多活動一起盯？分散到不同 tick，讓每輪都輕鬆。

容量直覺：越多請求，Cloud Run 自己加人手

[Auto-scaling 直覺圖（示意）]

• 圖上是假設每台容器大約能吃下 240 req/min（純示意）。
• 當整體請求（cron + webhook）增加，Cloud Run 會自動擴增容器（圖中的「階梯」）。
• 你不用管機器；你只要管頻率與策略（例如只在變化時通知，壓低不必要的負載）。

延遲預算：一次通知大概花在哪些環節？

[Latency Budget（示意）]

一個端到端 ≈ 720ms（示意）：
Ingress/Cold-start（80ms）→ HTTP（40ms）→ 抓 ibon（160ms）→ 解析（120ms）→ Firestore 讀（110ms）→ 差異與格式化（70ms）→ LINE push（140ms）

• 啟發：
  • 抓網頁 + 解析通常是大宗；能重用快取就重用。
  • Firestore若只讀必要欄位會更快。
  • 推播屬於外部呼叫，不必等到它完全返回才處理下一個任務（可用非阻塞設計）。

動手玩（超迷你 Smoke Test）

真的只有幾行，先確認「路通不通」。

# 1) Cloud Run /health（確認服務醒著）
curl -fsS "$SERVICE_URL/health" || echo "health check failed"

# 2) 假裝 Scheduler 的 cron（GET/POST 都示意）
curl -fsS -X POST "$SERVICE_URL/cron/tick" || echo "cron tick failed"

# 3) 假裝 LINE webhook（/watch 指令）
curl -fsS -X POST "$SERVICE_URL/webhook" \
  -H "Content-Type: application/json" \
  -d '{"user_id":"Uxxx","text":"/watch https://ibon.example 60"}' | jq .

關鍵程式骨架（Flask）

這些 endpoint 名稱只是示意，對應你自己的路由即可。

from flask import Flask, request, jsonify
import os, time

app = Flask(__name__)

# --- Firestore 封裝（示意） ---
def fetch_jobs(limit=50):
    # return [{"id":"W1","user_id":"U1","url":"https://ibon...","interval":60,"areas_cache":[...]}]
    ...

def update_job_cache(job_id, areas):
    ...

def should_notify(job, change, always=os.getenv("ALWAYS_NOTIFY","0")=="1"):
    return always or bool(change["changed"] or change["added"] or change["removed"])

# --- 核心功能（與 Day 4 解析會接起來） ---
def extract_ticket_info(url: str) -> dict:
    # 解析 ibon：活動資訊、座位圖、各區狀態（數字/熱賣中/售完）
    ...

def diff_areas(old: list, new: list):
    old_map = {a["code"]: a for a in old}
    new_map = {a["code"]: a for a in new}
    changed, added, removed = [], [], []
    for code, a in new_map.items():
        if code not in old_map:
            added.append(a); continue
        if old_map[code]["status"] != a["status"]:
            changed.append({"from": old_map[code], "to": a})
    for code, a in old_map.items():
        if code not in new_map:
            removed.append(a)
    return {"changed": changed, "added": added, "removed": removed}

def format_line_message(info, change=None):
    ev = info["event"]
    lines = [f"🎫 {ev['title']}", f"📍 {ev['venue']}  🗓 {ev['datetime']}", ""]
    avail = [a for a in info["areas"] if a["status"].isdigit()]
    hot   = [a for a in info["areas"] if a["status"] == "熱賣中"]
    sold  = [a for a in info["areas"] if a["status"] == "已售完"]
    if avail: lines.append("✅ 可售： " + "  ".join(f"{a['name']} {a['status']}" for a in avail[:8]))
    if hot:   lines.append("🟢 熱賣中： " + "、".join(a["name"] for a in hot[:10]))
    if sold:  lines.append("🔴 售完： " + "、".join(a["name"] for a in sold[:10]))
    return "\n".join(lines)

# --- cron：被 Scheduler 叫醒 ---
@app.route("/cron/tick", methods=["GET","POST"])
def cron_tick():
    start = time.time()
    jobs = fetch_jobs()
    processed = 0
    for job in jobs:
        info = extract_ticket_info(job["url"])
        change = diff_areas(job.get("areas_cache", []), info["areas"])
        if should_notify(job, change):
            # push_line(os.environ["LINE_CHANNEL_ACCESS_TOKEN"], job["user_id"], format_line_message(info, change))
            ...
        update_job_cache(job["id"], info["areas"])
        processed += 1
        if time.time() - start > int(os.getenv("TICK_SOFT_DEADLINE_SEC","25")):
            break
    return jsonify(ok=True, processed=processed, elapsed_ms=int((time.time()-start)*1000))

# --- webhook：處理 /watch /list /check ---
@app.route("/webhook", methods=["POST"])
def webhook():
    body = request.get_json(force=True)
    text = (body.get("text") or "").strip()
    # 解析文字，對應到你的 /watch /list /check...
    ...
    return jsonify(ok=True, echo=text)

@app.route("/health")
def health():
    return jsonify(status="ok")

Firestore「最小可用」資料結構（示意）

// collection: jobs
{
  id: "W123456",
  user_id: "Uxxxxx",
  url: "https://tickets.ibon.com/Show/Index/...",
  interval: 60,                // 秒
  enabled: true,
  last_run: 1737800000,        // unix
  areas_cache: [               // 上次解析的各區狀態
    { code: "B09P2J33", name: "5F B區 3800", status: "25" },
    { code: "C02X...",  name: "3F C區 2800", status: "熱賣中" }
  ]
}

常用環境變數（懶人卡）

今天作業（30 秒）

1. 開啟你的 LINE，傳送 /watch 你的活動頁URL 60。

2. 五分鐘後看一下通知有沒有跳。

3. 覺得吵就把 ALWAYS_NOTIFY=0；覺得太安靜就把間隔改短（30s）。

明天預告（Day 4）

要進洞穴了：如何用 DevTools & BeautifulSoup 把 ibon 頁面解析乾淨（jsonData 在哪、代碼怎麼翻成中文區名）。
放心，我會放超好懂的圖和可以直接跑的程式小片段。