wecom-fullstack
9bbe3001bf
ci/woodpecker/push/main Pipeline was successful
- 列 dedupe_key VARCHAR(160) NULL + 唯一索引 (entity_id,kpi_id,alert_type,dedupe_key) - write_alerts 三级判重：dedupe_key 命中 / 历史 NULL 行反解兼容 / 无命中插入 - 并发兜底：savepoint 捕获 IntegrityError 计跳过 - dry-run would_insert 改用 count_would_insert（修复 len(hits) 虚高） - DDL 幂等执行器 scripts/ddl_kpi_alerts_dedupe.py（check/apply/rollback） - tests/test_dedupe_key.py 7 passed（T1-T7b）；alert/overdue 回归 173 passed
114 lines
4.1 KiB
Python
114 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
||
"""kpi_alerts 去重键结构变更（幂等执行器，A2 方案）
|
||
|
||
用法：
|
||
ddl_kpi_alerts_dedupe.py --check # 只读：打印列/索引是否存在（默认）
|
||
ddl_kpi_alerts_dedupe.py --apply # 幂等执行：缺列则加列，缺索引则建索引
|
||
ddl_kpi_alerts_dedupe.py --rollback # 回退：删索引 + 删列（数据无需恢复：本改造无 UPDATE/DELETE）
|
||
|
||
特性：
|
||
* information_schema 判存在后再执行 ⇒ 重复跑安全（第二次全 NOOP）
|
||
* 只加可空列 + 唯一索引；历史行 dedupe_key 全 NULL ⇒ 建索引必然无冲突
|
||
* root 本机 socket 连接（与 migrate_map_connections.py 同口径），不读 .env
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import sys
|
||
|
||
from sqlalchemy import create_engine, text
|
||
|
||
URL = "mysql+pymysql://root@localhost/cma?unix_socket=/var/run/mysqld/mysqld.sock&charset=utf8mb4"
|
||
TABLE = "kpi_alerts"
|
||
COLUMN = "dedupe_key"
|
||
INDEX = "uk_kpi_alerts_dedupe"
|
||
|
||
ADD_COLUMN_SQL = (
|
||
"ALTER TABLE kpi_alerts "
|
||
"ADD COLUMN dedupe_key VARCHAR(160) NULL "
|
||
"COMMENT '生成侧幂等键 period|scope|ident，仅 alert_type=overdue 填充；历史行为 NULL' "
|
||
"AFTER alert_type"
|
||
)
|
||
ADD_INDEX_SQL = (
|
||
"CREATE UNIQUE INDEX uk_kpi_alerts_dedupe "
|
||
"ON kpi_alerts (entity_id, kpi_id, alert_type, dedupe_key)"
|
||
)
|
||
DROP_INDEX_SQL = "DROP INDEX uk_kpi_alerts_dedupe ON kpi_alerts"
|
||
DROP_COLUMN_SQL = "ALTER TABLE kpi_alerts DROP COLUMN dedupe_key"
|
||
|
||
|
||
def col_exists(eng) -> bool:
|
||
with eng.connect() as c:
|
||
n = c.execute(text(
|
||
"SELECT COUNT(*) FROM information_schema.columns "
|
||
"WHERE table_schema=DATABASE() AND table_name=:t AND column_name=:c"
|
||
), {"t": TABLE, "c": COLUMN}).scalar()
|
||
return int(n or 0) > 0
|
||
|
||
|
||
def idx_exists(eng) -> bool:
|
||
with eng.connect() as c:
|
||
n = c.execute(text(
|
||
"SELECT COUNT(*) FROM information_schema.statistics "
|
||
"WHERE table_schema=DATABASE() AND table_name=:t AND index_name=:i"
|
||
), {"t": TABLE, "i": INDEX}).scalar()
|
||
return int(n or 0) > 0
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser()
|
||
g = ap.add_mutually_exclusive_group()
|
||
g.add_argument("--check", action="store_true", help="只读检查（默认）")
|
||
g.add_argument("--apply", action="store_true", help="幂等执行")
|
||
g.add_argument("--rollback", action="store_true", help="回退结构")
|
||
a = ap.parse_args()
|
||
if not (a.check or a.apply or a.rollback):
|
||
a.check = True
|
||
|
||
eng = create_engine(URL, future=True)
|
||
col, idx = col_exists(eng), idx_exists(eng)
|
||
print("[state] %s.%s=%s index %s=%s" % (TABLE, COLUMN, col, INDEX, idx))
|
||
|
||
if a.check:
|
||
print("[check] %s" % ("结构就绪（无操作）" if (col and idx) else
|
||
("缺列" if not col else "") + ("缺索引" if not idx else "")))
|
||
return 0
|
||
|
||
if a.rollback:
|
||
with eng.begin() as c:
|
||
if idx:
|
||
c.execute(text(DROP_INDEX_SQL))
|
||
print("[rollback] DROP INDEX %s ✅" % INDEX)
|
||
if col:
|
||
c.execute(text(DROP_COLUMN_SQL))
|
||
print("[rollback] DROP COLUMN %s ✅" % COLUMN)
|
||
print("[state] 回退后 col=%s idx=%s" % (col_exists(eng), idx_exists(eng)))
|
||
return 0
|
||
|
||
# --apply
|
||
done = []
|
||
if not col:
|
||
with eng.begin() as c:
|
||
c.execute(text(ADD_COLUMN_SQL))
|
||
done.append("ADD COLUMN %s" % COLUMN)
|
||
print("[apply] ADD COLUMN %s ✅" % COLUMN)
|
||
else:
|
||
print("[apply] column %s 已存在 → NOOP" % COLUMN)
|
||
if not idx:
|
||
with eng.begin() as c:
|
||
c.execute(text(ADD_INDEX_SQL))
|
||
done.append("CREATE INDEX %s" % INDEX)
|
||
print("[apply] CREATE UNIQUE INDEX %s ✅" % INDEX)
|
||
else:
|
||
print("[apply] index %s 已存在 → NOOP" % INDEX)
|
||
|
||
col2, idx2 = col_exists(eng), idx_exists(eng)
|
||
print("[state] 执行后 col=%s idx=%s" % (col2, idx2))
|
||
payload = {"changed": done, "column_exists": col2, "index_exists": idx2}
|
||
print("[result] %s" % payload)
|
||
return 0 if (col2 and idx2) else 5
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|