From 451aa4e915dbbb461f43af8086e30c56775e6957 Mon Sep 17 00:00:00 2001 From: Windows <19353512109@163.com> Date: Sat, 12 Sep 2026 14:20:38 +0800 Subject: [PATCH] =?UTF-8?q?fix(profile):=20ProfileAssemblyService=20?= =?UTF-8?q?=E6=BC=8F=E5=86=99=20current=5Fcustomer=5Fid=EF=BC=88=E5=94=AF?= =?UTF-8?q?=E4=B8=80=E9=94=AE=E5=A4=B1=E6=95=88=E7=9A=84=E5=8F=A6=E4=B8=80?= =?UTF-8?q?=E5=A4=84=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 现象（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 = "低风险"`，候选态） --- app/service/profile_assembly_service.py | 14 +++- tools/fix_profile_snapshot_current.py | 88 +++++++++++++++++++++++++ 2 files changed, 101 insertions(+), 1 deletion(-) create mode 100644 tools/fix_profile_snapshot_current.py diff --git a/app/service/profile_assembly_service.py b/app/service/profile_assembly_service.py index ce68ecc..09b739b 100644 --- a/app/service/profile_assembly_service.py +++ b/app/service/profile_assembly_service.py @@ -226,8 +226,16 @@ class ProfileAssemblyService: ) -> None: """写入新版本快照并把旧版本置为非当前。 - 唯一键 `uk_profile_snapshot_current` 建立在生成列 `current_customer_id` 上， + 唯一键 `uk_profile_snapshot_current` 建立在 `current_customer_id` 上， 保证「每个客户最多一条 current」；因此必须先清旧再写新，顺序不能反。 + + ⚠️ **该列不是生成列**（订正 2026-09-12）：它由 `alembic/baseline_generated.sql` + 与真实库确认为**普通可空列 + 唯一键**，`app/model/profile.py` 的模块 docstring + 第 2 条已写明"当前版本必须由写入方**显式写入**客户 ID（历史版本写 NULL）"。 + 此处原先按"生成列"理解而两处都不赋值，后果实测到： + 旧版本 `is_current=0` 却仍留着 `current_customer_id`，新版本 `is_current=1` + 却是 NULL —— **唯一键形同虚设**（多个 NULL 不冲突）， + 而且旧值一旦残留、新值再补写就会直接撞键。 """ previous = list(await self.session.scalars( select(ProfileSnapshot).where( @@ -236,6 +244,8 @@ class ProfileAssemblyService: )) for row in previous: row.is_current = False + # 必须同时清空，否则与即将插入的新当前版本撞唯一键（见 docstring）。 + row.current_customer_id = None row.updated_at = now await self.session.flush() @@ -252,6 +262,8 @@ class ProfileAssemblyService: generation_basis=basis, snapshot_hash=sha256(payload.encode("utf-8")).hexdigest(), is_current=True, + # 当前版本必须显式写入客户 ID（历史版本为 NULL），见 docstring。 + current_customer_id=customer_id, generated_at=now, created_at=now, updated_at=now, diff --git a/tools/fix_profile_snapshot_current.py b/tools/fix_profile_snapshot_current.py new file mode 100644 index 0000000..287241b --- /dev/null +++ b/tools/fix_profile_snapshot_current.py @@ -0,0 +1,88 @@ +"""数据订正：`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)))