系列：30 天用 Google AI 打造臺灣防災速報 App（Day 16/30）
前兩週的下載程式都要手動執行。今天讓它自動化：Cloud Functions 每 10 分鐘下載一次 NCDR 的示警，整理後寫進 Firestore。這篇只談示警的排程；中央氣象署的地震報告另有一個排程，每 2 分鐘一次，9/26 上線。
這個排程的定位是每 10 分鐘彙整一次的防災資訊，不是地震速報。地震發生後幾秒內的警報，由手機的國家級警報負責；這個 App 提供的是之後的整理、查詢與說明。
Cloud Functions 需要 Firebase 的 Blaze 方案（用多少付多少，每月仍有免費額度）。升級前先在 Google Cloud 設了預算快訊：每月 40 美元，超過就通知。
排程本身的量很小：每 10 分鐘一次，一個月約 4,320 次執行，Cloud Functions 每月的免費額度是 200 萬次呼叫。真正要注意的是每次執行讀寫了多少次資料庫，這一點後面會看到。
第二代 Cloud Functions 支援 Python。函式上方以 @ 開頭的那一行是 Python 的裝飾器，用來替函式加上設定，這裡設定的是排程：
# functions/main.py
@scheduler_fn.on_schedule(schedule="*/10 * * * *", region="asia-east1",
timezone=scheduler_fn.Timezone("Asia/Taipei"), timeout_sec=120)
def poll_alerts(event: scheduler_fn.ScheduledEvent) -> None:
db = firestore.client()
try:
counts = sync_alerts(db, fetch_alerts())
except Exception as e:
_write_status(db, "sync_status", ok=False, error=e) # 記錄這一輪失敗
raise
_write_status(db, "sync_status", ok=True, fields={"active_count": counts.get("active", 0)})
print("poll_alerts", counts)
*/10 * * * * 是排程的寫法，意思是每 10 分鐘一次。fetch_alerts() 與整理用的 pipeline() 沿用第一週的程式。sync_alerts() 決定每一則示警該放在哪裡，下面的三個決定都在這個函式裡。每一輪結束時把各狀態的筆數印到日誌，在 Google Cloud 的網頁管理介面（Cloud Console）就看得到每次執行的結果；同時把同步的時間與成功與否寫進 meta/sync_status，App 首頁的「資料同步」時間讀的就是這份文件。
8 月寫的第一版設計是這樣：這一輪下載的資料裡沒有出現、資料庫裡卻還是生效中的示警，就當作已經結案。
實際資料推翻了這個設計。9/25 第一次排程執行時，NCDR 的資料整理後有 831 則示警，其中 731 則早就過期了，卻還留在資料裡。NCDR 不會在示警過期時立刻移除，所以「消失」不是判斷結案的好依據。
更危險的是下載失敗。如果某一輪網路出錯、只取得部分資料，照第一版的設計，沒拿到的示警會全部被當成結案，生效中的示警就從 App 上消失了。
所以 active、closed、expired 這三種狀態，一律由示警本身的內容決定，這是第一週 normalize.py 的規則：
closed）。expires）：過期（expired）。active）。沒有到期時間的也算生效中，寧可多顯示，也不要漏掉示警。下載失敗時，fetch_alerts() 會回報錯誤並中止這一輪。alerts 與 history 都不會變動，只在同步狀態裡記下這一輪失敗，等下一輪再試。
那麼真的從資料裡消失的示警怎麼辦？alerts 裡的每一份文件都記著自己的到期時間。這一輪沒出現、而且到期時間已經過了，才搬到 history，日誌裡記成 expired_in_db：
# alerts 裡已過期、但這一輪資料沒有列出的舊文件
for uid, d in in_db.items():
exp = d.get("expires")
if uid not in handled and exp and datetime.fromisoformat(exp) < now:
ops.append(("set", db.collection("history").document(uid), _to_history(d, "expired", now)))
ops.append(("delete", db.collection("alerts").document(uid), None))
counts["expired_in_db"] += 1
判斷依據是到期時間，不是「這一輪有沒有出現」。
Day 5 發現 NCDR 的示警 id 不唯一，當時的結論是：只剔除 id、更新時間、摘要完全相同的資料，寧可多顯示、不可漏掉；「哪幾則是同一事件的新舊版本」要等 CAP 電文的 references 欄位（官方標明這則取代哪一則）才能判斷。
9/25 寫排程時，我沒有照這個結論做。第一版的 sync_alerts() 依 id 分組，每組只留更新時間最新的一則，其他標成「被新版取代」（superseded）放進 history。想法是避免同一則示警的新舊版本同時出現。
審稿時發現這和 Day 5 的結論衝突。以 9/26 03:03 的正式資料檢查：生效中的示警有 59 則，其中 4 個 id 底下各有好幾則，合計 25 則，全部是彼此獨立的通知：
照第一版的寫法，這 4 組 25 則只會留下 4 則，另外 21 則生效中的示警被放進 history，App 上看不到。對使用者來說，這就是漏報。
修正後移除分組，每一則（每個 uid）各自判斷，生效中的全部留在 alerts。先用同一份資料離線驗證：生效中 59 則，寫進 alerts 也是 59 則。9/26 凌晨部署，前後兩輪的正式紀錄如下。時間是 UTC（世界標準時間，比臺灣時間慢 8 小時）；new_active 是這一輪新增的生效示警數，ops 是資料庫寫入與刪除的操作次數：
2026-09-25 19:10 UTC {'active': 33, 'expired': 753, 'closed': 22, 'superseded': 26, 'ops': 0}
2026-09-25 19:20 UTC {'active': 58, 'expired': 754, 'moved': 1, 'closed': 22, 'new_active': 26, 'ops': 28}
修正前 alerts 只有 33 則，另有 26 則被標成 superseded；修正後的第一輪把這 26 則寫回 alerts，同一輪剛好有 1 則到期搬到 history，生效中的示警變成 58 則（33＋26－1）。被藏起來的是 26 則，比上面算的 21 則多。原因在第一版挑「最新一則」時，連已經過期的版本也算進去：某個 id 最新的一則如果已經過期，同一個 id 底下仍生效的示警，也會被標成被新版取代。同一事件的新舊版本可能同時出現，這是 Day 5 就接受的代價；真正的新舊版本判斷，之後用 CAP 電文的 references 來做。
第一次執行的結果：
{'active': 34, 'closed': 22, 'expired': 731, 'superseded': 44, 'ops': 1628}
第一次執行要把所有示警放到正確的位置，1,628 次是合理的：生效中 34 則寫入 alerts，其餘 797 則寫入 history，同時對 alerts 刪除這 797 則，合計 34＋797＋797＝1,628。程式把它分成 500、500、500、128 四批送出。
問題是之後的每一輪都一樣。第一版的程式每一輪都把 800 則左右已結束的示警重新寫進 history 一次，再把它們從 alerts 刪除一次，即使它們本來就不在 alerts。以 UTC 18:00 那一輪為例：
2026-09-25 18:00 UTC {'active': 34, 'expired': 751, 'closed': 22, 'superseded': 26, 'ops': 1632}
這一輪寫入 34＋799＝833 次，刪除 799 次。一天 144 輪，約寫入 12 萬次、刪除 11.5 萬次。Firestore 的免費額度是每天寫入 2 萬次、刪除 2 萬次，這個寫法分別是免費額度的 6 倍與 5.8 倍。Day 14 估算費用時只算了 Gemini 的呼叫，並寫明資料庫的讀寫費用要上線後再核對，這就是核對出來的問題。
改法是先讀出 alerts 目前有哪些文件，再只處理有變化的部分：
in_db = {snap.id: snap.to_dict() for snap in db.collection("alerts").stream()}
ops, handled = [], set()
for a in alerts:
status = a.status
handled.add(a.uid)
if status == "active":
if a.uid not in in_db: # 新出現的生效示警
doc = a.to_dict()
doc.update(status="active", synced_at=now)
ops.append(("set", db.collection("alerts").document(a.uid), doc))
elif a.uid in in_db: # 原本生效、這一輪結束：連同 AI 欄位搬到 history
ops.append(("set", db.collection("history").document(a.uid), _to_history(in_db[a.uid], status, now)))
ops.append(("delete", db.collection("alerts").document(a.uid), None))
elif now - a.updated <= NEW_WINDOW: # 剛發布就已結束的示警（NEW_WINDOW 是 30 分鐘）
doc = a.to_dict()
doc["synced_at"] = now
ops.append(("set", db.collection("history").document(a.uid), _to_history(doc, status, now)))
uid 包含摘要的雜湊值，所以已經在 alerts 裡的就是同一則，不用重寫。搬到 history 時用的是資料庫裡的那一份，AI 補上的摘要與譯文會一起保存；_to_history() 同時記下結束時間，並設定 90 天後自動刪除。
上線前先用記憶體裡的假資料庫測試：同一份資料連續同步兩次，第二次是 0 次操作；把其中一則示警改成已到期，只有 2 次（寫入 history、從 alerts 刪除）。以下是這項修正部署後的正式紀錄，當時決定二的分組修正還沒部署，所以紀錄裡仍有 superseded：
2026-09-25 18:30 UTC {'active': 34, 'expired': 752, 'closed': 22, 'superseded': 26, 'ops': 1634}
2026-09-25 18:40 UTC {'active': 34, 'expired': 752, 'closed': 22, 'superseded': 26, 'ops': 0}
讀取也算一下。改法每一輪要讀一次 alerts，生效中的示警有幾十份，一天 144 輪，約 5 千到 9 千次讀取，Firestore 的免費額度是每天 5 萬次。App 開啟時讀取示警、AI 補欄位時的寫入，另外計算，要看實際使用量，之後用帳單核對。
示警寫進 alerts 之後，還要加上 AI 的分級、摘要和譯文。這一步不放在排程函式裡，另外寫一個函式，在 alerts 新增文件時觸發：
@firestore_fn.on_document_created(document="alerts/{uid}", region=REGION, secrets=[GEMINI_SECRET],
timeout_sec=120, memory=options.MemoryOption.MB_512)
def enrich_alert(event):
...
下載與同步要快、要穩定，呼叫 Gemini 比較慢，也可能暫時失敗。分成兩個函式，AI 加值函式出錯時，示警照樣準時寫進資料庫，App 照樣顯示原文。Gemini 的金鑰存在 Secret Manager（Google Cloud 存放金鑰等機密資料的服務），部署時才提供給函式，不寫在程式裡。
因為只在新增文件時觸發，這個函式上線前就已經在 alerts 裡的 37 則示警不會被處理，另外用一支腳本補上。
第一次使用 Firestore 觸發器要等權限生效。 enrich_alert 第一次部署失敗：
Permission denied while using the Eventarc Service Agent. If you recently started to use Eventarc,
it may take a few minutes before all necessary permissions are propagated to the Service Agent.
Eventarc 是 Google Cloud 轉送事件的服務，Firestore 觸發器靠它運作。專案第一次用到時，權限要幾分鐘才會生效。約十分鐘後重新部署就成功了。
references。明天 Day 17 換到前端：用 Google Stitch 從一段文字產生 App 的設計稿。
iThome鐵人賽
看影片追技術