89 lines
3.2 KiB
Python
89 lines
3.2 KiB
Python
|
"""数据订正：`profile_snapshots.current_customer_id` 与 `is_current` 对齐。
|
|||
|
|
|
|||
|
|
**默认 dry-run**，加 `--apply` 才提交。幂等，可重复运行。
|
|||
|
|
|
|||
|
|
### 背景
|
|||
|
|
|
|||
|
|
`current_customer_id` 是**普通可空列 + 唯一键** `uk_profile_snapshot_current`（不是生成列），
|
|||
|
|
契约是「当前版本写客户 ID、历史版本写 NULL」。但写入方曾把它误当生成列，
|
|||
|
|
两处路径都未赋值（`ProfileAssemblyService._write_snapshot`、
|
|||
|
|
`CustomerProfileCandidateService._write_profile_snapshot`），造成两类脏数据：
|
|||
|
|
|
|||
|
|
1. 历史版本 `is_current=0` 却**仍留着** `current_customer_id`；
|
|||
|
|
2. 当前版本 `is_current=1` 却是 NULL。
|
|||
|
|
|
|||
|
|
后果：唯一键形同虚设（多个 NULL 不冲突）⇒「每个客户最多一条 current」失效；
|
|||
|
|
而且 (1) 的值一旦与 (2) 补上的值相同就会**撞唯一键**。
|
|||
|
|
|
|||
|
|
### 顺序要紧
|
|||
|
|
|
|||
|
|
先清 (1) 再补 (2)：反过来的话，第 2 步写入时旧值还在，会直接撞键。
|
|||
|
|
|
|||
|
|
用法：
|
|||
|
|
.\\.venv\\Scripts\\python.exe tools\\fix_profile_snapshot_current.py # 先看
|
|||
|
|
.\\.venv\\Scripts\\python.exe tools\\fix_profile_snapshot_current.py --apply # 再改
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
import asyncio
|
|||
|
|
import sys
|
|||
|
|
|
|||
|
|
from sqlalchemy import text
|
|||
|
|
|
|||
|
|
from app.infrastructure.db import engine
|
|||
|
|
|
|||
|
|
#: (1) 历史版本不该持有该列 → 清空。必须先做。
|
|||
|
|
_CLEAR_STALE = text("""
|
|||
|
|
UPDATE profile_snapshots
|
|||
|
|
SET current_customer_id = NULL
|
|||
|
|
WHERE is_current = 0 AND current_customer_id IS NOT NULL
|
|||
|
|
""")
|
|||
|
|
|
|||
|
|
#: (2) 当前版本必须持有该列 → 补上。放在 (1) 之后，避免撞唯一键。
|
|||
|
|
_FILL_CURRENT = text("""
|
|||
|
|
UPDATE profile_snapshots
|
|||
|
|
SET current_customer_id = customer_id
|
|||
|
|
WHERE is_current = 1 AND current_customer_id IS NULL
|
|||
|
|
""")
|
|||
|
|
|
|||
|
|
_COUNT_STALE = text(
|
|||
|
|
"SELECT COUNT(*) FROM profile_snapshots "
|
|||
|
|
"WHERE is_current = 0 AND current_customer_id IS NOT NULL"
|
|||
|
|
)
|
|||
|
|
_COUNT_MISSING = text(
|
|||
|
|
"SELECT COUNT(*) FROM profile_snapshots "
|
|||
|
|
"WHERE is_current = 1 AND current_customer_id IS NULL"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
|
|||
|
|
async def main(apply: bool) -> int:
|
|||
|
|
async with engine.connect() as conn:
|
|||
|
|
stale = await conn.scalar(_COUNT_STALE)
|
|||
|
|
missing = await conn.scalar(_COUNT_MISSING)
|
|||
|
|
print(f"待清空（is_current=0 却留着值）: {stale}")
|
|||
|
|
print(f"待补写（is_current=1 却是 NULL）: {missing}")
|
|||
|
|
|
|||
|
|
if not apply:
|
|||
|
|
print("dry-run：未提交。确认无误后加 --apply 再运行。")
|
|||
|
|
await conn.rollback()
|
|||
|
|
return 0
|
|||
|
|
|
|||
|
|
cleared = (await conn.execute(_CLEAR_STALE)).rowcount
|
|||
|
|
filled = (await conn.execute(_FILL_CURRENT)).rowcount
|
|||
|
|
await conn.commit()
|
|||
|
|
print(f"已清空 {cleared} 行、已补写 {filled} 行")
|
|||
|
|
|
|||
|
|
# 复核：两类异常都应归零
|
|||
|
|
print("复核 待清空:", await conn.scalar(_COUNT_STALE))
|
|||
|
|
print("复核 待补写:", await conn.scalar(_COUNT_MISSING))
|
|||
|
|
print("\n各客户的当前快照：")
|
|||
|
|
for row in (await conn.execute(text(
|
|||
|
|
"SELECT customer_id, version, is_current, current_customer_id "
|
|||
|
|
"FROM profile_snapshots ORDER BY customer_id, version"
|
|||
|
|
))).mappings().all():
|
|||
|
|
print(" ", dict(row))
|
|||
|
|
return 0
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
sys.exit(asyncio.run(main("--apply" in sys.argv)))
|