不靠搶票外掛、不洗版、不吵人,只在「有感」的時候提醒你。這篇把 Day 1–7 的重點 一次打包:
架構圖、爬蟲解析、差異比對(diff)、通知策略、訊息排版、部署 Runbook、常見雷坑與最佳實務。
直接複製本文即可貼到 iT 邦幫忙(支援 Markdown + 內嵌 SVG 圖),無須另行下載圖片。
把 ibon 的「票區 / 價格 / 當前狀態」抽出來。重點:穩定定位元素、清洗中文字串、保留「熱賣中/售完」等非數字狀態。
import requests
from bs4 import BeautifulSoup
def extract_ticket_info(url: str):
html = requests.get(url, timeout=10).text
soup = BeautifulSoup(html, "lxml")
event = {
"title": soup.select_one(".title").get_text(strip=True),
"venue": soup.select_one(".venue").get_text(strip=True),
"datetime": soup.select_one(".time").get_text(strip=True),
}
areas = []
for row in soup.select(".ticket-area-row"):
name = row.select_one(".name").get_text(strip=True)
status = row.select_one(".status").get_text(strip=True) # 可能是數字/熱賣中/已售完
code = row.get("data-code") or name
areas.append({"code": code, "name": name, "status": status})
return {"event": event, "areas": areas}
實務小撇步
requests
設 逾時 + 重試(backoff),避免網路抖動。import time, random
import requests
def fetch(url, retries=3):
for i in range(retries):
try:
return requests.get(url, timeout=8, headers={"User-Agent":"Mozilla/5.0"}).text
except Exception:
time.sleep(1.5 * (i+1) + random.random())
raise RuntimeError("fetch failed")
核心:只在真正「有變化」時通知,並把短時間的多次變化合併成一則。
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}
import time
def should_notify(change, policy="coalesce", last_sent=None, coalesce_sec=30):
now = time.time()
active = bool(change["changed"] or change["added"] or change["removed"])
if policy == "always": return True, now
if policy == "on_change": return active, (now if active else last_sent)
if policy == "coalesce":
if not active: return False, last_sent
if not last_sent or now - last_sent >= coalesce_sec: # 開新視窗
return True, now
return False, last_sent # 視窗內:合併
return active, (now if active else last_sent)
def apply_preferences(areas, prices=None, zones=None, regex=None):
import re
def keep(a):
nm = a["name"]; ok = True
if prices: ok &= any(p in nm for p in prices)
if zones: ok &= any(z in nm for z in zones)
if regex: ok &= re.search(regex, nm or "") is not None
return ok
return [a for a in areas if keep(a)]
def format_text_message(info, change=None):
ev = info["event"]; out = [f"🎫 {ev['title']}", f"📍 {ev['venue']} 🗓 {ev['datetime']}", ""]
avail = [a for a in info["areas"] if str(a["status"]).isdigit()]
hot = [a for a in info["areas"] if a["status"] in ("熱賣中","Hot","hot")]
sold = [a for a in info["areas"] if a["status"] in ("已售完","SoldOut","sold")]
if avail: out.append("✅ 可售:" + " ".join(f"{a['name']} {a['status']}" for a in avail[:8]) + (f" …(+{len(avail)-8})" if len(avail)>8 else ""))
if hot: out.append("🟢 熱賣中:" + "、".join(a["name"] for a in hot[:10]) + (f" …(+{len(hot)-10})" if len(hot)>10 else ""))
if sold: out.append("🔴 售完:" + "、".join(a["name"] for a in sold[:10]) + (f" …(+{len(sold)-10})" if len(sold)>10 else ""))
if change:
lines = []
for c in change["changed"][:5]: lines.append(f"- {c['to']['name']}: {c['from']['status']} → {c['to']['status']}")
if lines: out += ["", "🔔 變化摘要:"] + lines
return "\n".join(out)
/cron/tick
/webhook
,Logging/Monitoring 開告警建議預設 60 秒;手刀期可短到 20–30 秒;深夜可放到 120–300 秒,搭配合併視窗減噪音。
情境 A|等補票的小芸
on-change + coalesce(30s)
情境 B|演唱會開票 5 分鐘前的阿傑
always
(短時間)on-change
。情境 C|維運值班的阿豪
requirements.txt
flask==3.0.3
gunicorn==21.2.0
requests==2.32.3
beautifulsoup4==4.12.3
lxml==5.2.1
google-cloud-firestore==2.16.1
google-cloud-logging==3.10.0
line-bot-sdk==3.11.0
Dockerfile
FROM python:3.11-slim
WORKDIR /app
RUN apt-get update && apt-get install -y --no-install-recommends ca-certificates curl && rm -rf /var/lib/apt/lists/*
COPY requirements.txt . && pip install --no-cache-dir -r requirements.txt
COPY . .
ENV PORT=8080
CMD exec gunicorn -b :$PORT -w 2 --timeout 90 app:app
gcloud(摘要)
# 推鏡像
PROJECT_ID=$(gcloud config get-value project)
REGION=asia-east1
IMAGE=asia-east1-docker.pkg.dev/$PROJECT_ID/ibon-repo/ibon-watch:latest
docker build -t "$IMAGE" . && docker push "$IMAGE"
# 部署 Cloud Run
gcloud run deploy ibon-watch --image "$IMAGE" --region $REGION --platform managed --allow-unauthenticated \
--set-env-vars "DEFAULT_PERIOD_SEC=60,MAX_PER_TICK=5,TICK_SOFT_DEADLINE_SEC=25,LINE_CHANNEL_ACCESS_TOKEN=XXX,LINE_CHANNEL_SECRET=YYY"
# 每 60s 觸發
SERVICE_URL=$(gcloud run services describe ibon-watch --region $REGION --format='value(status.url)')
gcloud scheduler jobs create http ibon-cron --schedule="* * * * *" --uri="$SERVICE_URL/cron/tick" --http-method=POST
coalesce
視窗,或加上價位/區域偏好過濾。