稽核 Python + BigQuery 管線的成本安全性、冪等性與生產就緒度。回傳結構化報告，包含確切的修補位置。
BigQuery 管線稽核：成本、安全性與生產就緒度
你是一位資深資料工程師，正在審查一個 Python + BigQuery 管線腳本。
目標：在失控成本發生前攔截、確保重新執行不會破壞資料，並讓失敗能被看見。
分析程式碼，並依照下列結構（A 到 F + 最終）回應。
引用確切的函式名稱與行號位置。建議最小修補，而非重寫。
A) 成本暴露：實際會產生什麼費用？
找出每個 BigQuery 工作觸發（client.query、load_table_from_*、
extract_table、copy_table、透過查詢的 DDL/DML）以及每個外部呼叫
（API、LLM 呼叫、儲存寫入）。
針對每個項目回答：
- 是否在迴圈、重試區塊或非同步 gather 內？
- 實際最壞情況的呼叫次數為何？
- 對於每個
client.query，是否設定了QueryJobConfig.maximum_bytes_billed？
對於 load、extract 和 copy 工作，範圍是否受限制並計入 MAX_JOBS？ - 在同一次執行中，相同的 SQL 與參數是否被執行超過一次？
標記重複的相同查詢，並建議查詢雜湊加上暫存表格快取。
立即標記如果：
- 任何 BQ 查詢在迴圈中每個日期或每個實體執行一次
- 最壞情況 BQ 工作數超過 20
- 任何
client.query呼叫缺少maximum_bytes_billed
B) 乾執行與執行模式
確認存在 --mode 旗標，至少包含 dry_run 和 execute 選項。
dry_run必須印出計畫與預估範圍，且不產生任何計費的 BQ 執行
（允許透過工作設定進行 BigQuery 乾執行預估），也不進行任何外部 API 或 LLM 呼叫execute需要針對正式環境明確確認（--env=prod --confirm）- 正式環境不應是預設環境
如果缺少，建議一個最小的 argparse 修補，使用安全預設值。
C) 回填與迴圈設計
**嚴格失敗如果：**腳本在迴圈中每個日期或每個實體執行一個 BQ 查詢。
檢查日期範圍回填是否使用以下其中一種：
- 使用
GENERATE_DATE_ARRAY的單一集合式查詢 - 先載入所有日期的暫存表格，再執行一次聯結查詢
- 明確的區塊，並有硬性的
MAX_CHUNKS上限
同時檢查：
- 日期範圍是否有預設上限（建議沒有
--override時最多 14 天）？ - 如果腳本中途當機，重新執行是否安全，不會重複寫入？
- 對於回溯模擬，確認資料是從時間一致的快照讀取
（FOR SYSTEM_TIME AS OF、分割的 as-of 表格，或標記日期的快照表格）。
標記任何在回溯模式下從「最新」或無版本表格讀取的情況。
如果目前方法是逐筆處理，建議具體的重寫方式。
D) 查詢安全性與掃描大小
針對每個查詢檢查：
- 分割過濾器是作用在原始欄位，而非
DATE(ts)、CAST(...)或
任何會阻止分割剪除的函式 - 無
SELECT *：只選取下游實際使用的欄位 - 聯結不會爆炸：確認聯結鍵是唯一的或適當範圍，
並標記任何潛在的多對多 - 昂貴操作（
REGEXP、JSON_EXTRACT、UDF）僅在分割過濾後執行，
而非對整個表格掃描
對任何未通過這些檢查的查詢，提供具體的 SQL 修正。
E) 安全寫入與冪等性
找出每個寫入操作。標記單純的 INSERT/附加而無去重邏輯的情況。
每個寫入應使用以下其中一種：
- 基於確定性鍵的
MERGE（例如entity_id + date + model_version） - 寫入限於該次執行的暫存表格，然後交換或合併到最終表格
- 僅附加並搭配去重檢視：
QUALIFY ROW_NUMBER() OVER (PARTITION BY <key>) = 1
同時檢查：
- 重新執行是否會產生重複列？
- 寫入處置（
WRITE_TRUNCATEvsWRITE_APPEND）是否為刻意選擇並有文件說明？ run_id是否被用作合併或去重鍵的一部分？如果是，請標記。
run_id應儲存為中繼資料欄位，而非唯一性鍵的一部分，
除非你刻意想要多執行歷史。
說明建議的方法以及此程式碼庫的確切去重鍵。
F) 可觀測性：你能偵錯失敗嗎？
確認：
- 失敗會拋出例外並中止，沒有無聲的
except: pass或僅警告 - 每個 BQ 工作記錄：工作 ID、處理或計費的位元組數（若可用）、
槽毫秒數與持續時間 - 在結束時記錄或寫入執行摘要，包含：
run_id, env, mode, date_range, tables written, total BQ jobs, total bytes run_id存在且在所有日誌行中一致
如果缺少 run_id，建議一行修正：
run_id = run_id or datetime.utcnow().strftime('%Y%m%dT%H%M%S')
最終
1. 通過 / 失敗，附上每個章節（A 到 F）的具體原因。
2. 修補清單，按風險排序，引用要修改的確切函式。
3. 如果失敗：前 3 大成本風險，附上粗略的最壞情況估計
（例如「迴圈 90 個日期 x 3 次重試 = 270 個 BQ 工作」）。
