114 lines
3.2 KiB
Python
114 lines
3.2 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
AwoooP Phase 1 Batch 1 回填腳本
|
||
================================
|
||
對 incidents / knowledge_entries / playbooks / audit_logs 四張表
|
||
分批將 project_id IS NULL 的列回填為 'awoooi'。
|
||
|
||
前置條件：
|
||
awooop_phase1_batch1_rls_2026-05-04.sql Step A（ADD COLUMN nullable）已執行
|
||
|
||
執行方式：
|
||
從 secret manager / operator vault 設定 DATABASE_URL，禁止在指令或檔案中寫入 URL。
|
||
cd apps/api && python scripts/awooop_phase1_batch1_backfill.py
|
||
|
||
2026-05-04 ogt + Claude Sonnet 4.6（ADR-118 Batch 1 C-3 修正）
|
||
"""
|
||
|
||
import asyncio
|
||
import os
|
||
import time
|
||
|
||
from sqlalchemy import text
|
||
from sqlalchemy.ext.asyncio import create_async_engine
|
||
|
||
DATABASE_URL = os.environ["DATABASE_URL"]
|
||
|
||
TABLES = [
|
||
("incidents", "incident_id"),
|
||
("knowledge_entries", "id"),
|
||
("playbooks", "id"),
|
||
("audit_logs", "id"),
|
||
]
|
||
|
||
BATCH_SIZE = 5000
|
||
SLEEP_MS = 100 # 批次間休眠 ms，降低對正常流量的影響
|
||
|
||
|
||
async def count_nulls(conn, table: str) -> int:
|
||
result = await conn.execute(
|
||
text(f"SELECT count(*) FROM {table} WHERE project_id IS NULL") # noqa: S608
|
||
)
|
||
return result.scalar()
|
||
|
||
|
||
async def backfill_table(engine, table: str, pk_col: str) -> int:
|
||
total_updated = 0
|
||
print(f"\n[{table}] 開始回填...")
|
||
|
||
while True:
|
||
async with engine.begin() as conn:
|
||
result = await conn.execute(text(f"""
|
||
UPDATE {table}
|
||
SET project_id = 'awoooi'
|
||
WHERE {pk_col} IN (
|
||
SELECT {pk_col} FROM {table}
|
||
WHERE project_id IS NULL
|
||
LIMIT :batch_size
|
||
FOR UPDATE SKIP LOCKED
|
||
)
|
||
"""), {"batch_size": BATCH_SIZE})
|
||
rows = result.rowcount
|
||
|
||
total_updated += rows
|
||
if rows == 0:
|
||
break
|
||
|
||
print(f" [{table}] 已回填 {total_updated} 筆...")
|
||
await asyncio.sleep(SLEEP_MS / 1000)
|
||
|
||
print(f" [{table}] 回填完成，共 {total_updated} 筆")
|
||
return total_updated
|
||
|
||
|
||
async def verify(engine) -> bool:
|
||
print("\n=== 驗收確認 ===")
|
||
ok = True
|
||
async with engine.connect() as conn:
|
||
for table, _ in TABLES:
|
||
null_count = await count_nulls(conn, table)
|
||
status = "✅" if null_count == 0 else "❌"
|
||
print(f" {status} {table}: {null_count} 筆 NULL project_id")
|
||
if null_count != 0:
|
||
ok = False
|
||
return ok
|
||
|
||
|
||
async def main():
|
||
print("=" * 60)
|
||
print("AwoooP Phase 1 Batch 1 Backfill")
|
||
print("=" * 60)
|
||
|
||
engine = create_async_engine(DATABASE_URL, echo=False)
|
||
t0 = time.monotonic()
|
||
|
||
for table, pk_col in TABLES:
|
||
await backfill_table(engine, table, pk_col)
|
||
|
||
passed = await verify(engine)
|
||
elapsed = time.monotonic() - t0
|
||
|
||
print(f"\n{'✅ 所有表回填完成' if passed else '❌ 仍有 NULL，請重跑'}")
|
||
print(f"耗時：{elapsed:.1f}s")
|
||
print()
|
||
if passed:
|
||
print("下一步：執行 awooop_phase1_batch1_rls_2026-05-04.sql 的 Step C")
|
||
else:
|
||
print("⚠️ 請確認無長 transaction 持有 SKIP LOCKED 的列後重跑")
|
||
|
||
await engine.dispose()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|