wangjianlong_0626
451aa4e915
## 现象（2026-09-12 端到端跑通后查数据时发现） 客户 9102 的画像快照状态**彼此矛盾**： | version | is_current | current_customer_id | |---|---|---| | 2 | `0` | `9102` ← 清旧时没清空 | | 3 | `1` | `NULL` ← 建新时没写入 | ## 根因 `ProfileAssemblyService._write_snapshot` 把 `current_customer_id` **当成了生成列**： - 方法 docstring 原文写着"唯一键 `uk_profile_snapshot_current` 建立在**生成列** `current_customer_id` 上" - 因此两处都不赋值（以为 DB 会自动填） **但该列不是生成列** —— `alembic/baseline_generated.sql` 与真实库都是**普通可空列 + 唯一键**， `app/model/profile.py` 的模块 docstring 第 2 条已明确："当前版本必须由写入方**显式写入**客户 ID （历史版本写 NULL），才能保证「每个客户最多一条当前快照」"。 后果：唯一键**形同虚设**（多个 NULL 不冲突）⇒ 不变式失效；且旧版本残留的值 一旦与新版本补上的值相同，就会**直接撞唯一键**。 > 这与本线先前修的 `CustomerProfileCandidateService._write_profile_snapshot` 是**同一个缺陷的另一处** > —— 当时只找到一处，这次是靠真实链路跑出数据后核对才暴露出来。 ## 改动 `app/service/profile_assembly_service.py`： - 旧版本：`is_current = False` 的同时 `current_customer_id = None` - 新版本：`is_current=True` 的同时 `current_customer_id=customer_id` - 订正方法 docstring 的错误认知（"生成列"→ 普通可空列 + 唯一键），并写明后果 ## 已有数据订正 新增 `tools/fix_profile_snapshot_current.py`（**默认 dry-run**、幂等、`--apply` 才提交）： 1. 先清空 `is_current=0` 却残留值的行 2. 再补写 `is_current=1` 却是 NULL 的行 3. **顺序要紧**：反过来的话第 2 步会与残留值撞唯一键 本机实测：清空 1 行、补写 1 行，复核两类异常均归零。 ## 验证 - `mypy app` → 0 错 / 245 文件 - `pytest tests`（全量）→ `2 failed, 1418 passed, 1 skipped` （2 个失败为既有环境项：httpx 把中文序列化成 `\uXXXX`，非本次引入） - 端到端：真实对话 → 记忆抽取 → `memory_unit` 落库已实测通过（客户 9102 `preference:risk_level = "低风险"`，候选态）
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)))
|