92 lines
3.2 KiB
Python
92 lines
3.2 KiB
Python
|
"""数据订正：把 `memory_sync_outbox` 的历史取值改齐为全仓小写口径。
|
|||
|
|
|
|||
|
|
**默认 dry-run**（只报告将影响几行），加 `--apply` 才真正提交。幂等，可重复运行。
|
|||
|
|
|
|||
|
|
### 背景
|
|||
|
|
|
|||
|
|
`docs/00` §6.4.6 的取值栏曾写作大写 `MILVUS`/`NEO4J`、`UPSERT` 与中文 `待处理`，
|
|||
|
|
与全仓实现从未对齐。按那份文档写入的事件会**静默失效**：
|
|||
|
|
|
|||
|
|
- 消费端 `MemorySyncOutboxWorker` 按 `handlers.get(event.target_store)` 分派 handler；
|
|||
|
|
- 领取条件是 `status.in_({"pending","failed"})`。
|
|||
|
|
|
|||
|
|
大写 + 中文两个条件都不满足 ⇒ 事件任何消费者都领不到、永久滞留且不报错
|
|||
|
|
（唯一键 `(event_uuid, target_store)` 对大小写没有约束，MySQL 也不会报错）。
|
|||
|
|
|
|||
|
|
因此凡是在本仓写入过 `memory_sync_outbox` 的环境，都可能有这类脏行。
|
|||
|
|
|
|||
|
|
### 安全口径
|
|||
|
|
|
|||
|
|
- 只改 `target_store` / `operation` / `status` 的**值**；
|
|||
|
|
- **不触碰**主键、唯一键 `uk_memory_sync_event (event_uuid, target_store)`、
|
|||
|
|
`payload`、时间戳、`retry_count`；
|
|||
|
|
- 就地改值不会新增行（两组取值大小写不同，本会各自成行，这里把它们并到正确的一组）；
|
|||
|
|
- 中文字符不受 `LOWER()` 影响（非 ASCII 字节不变），故 `LOWER()` 是安全的。
|
|||
|
|
|
|||
|
|
用法：
|
|||
|
|
.\\.venv\\Scripts\\python.exe tools\\normalize_memory_sync_outbox.py # 先看
|
|||
|
|
.\\.venv\\Scripts\\python.exe tools\\normalize_memory_sync_outbox.py --apply # 再改
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
import asyncio
|
|||
|
|
import sys
|
|||
|
|
|
|||
|
|
from sqlalchemy import text
|
|||
|
|
|
|||
|
|
from app.infrastructure.db import engine
|
|||
|
|
|
|||
|
|
_STATUS_CASE = """
|
|||
|
|
CASE status
|
|||
|
|
WHEN '待处理' THEN 'pending'
|
|||
|
|
WHEN '处理中' THEN 'processing'
|
|||
|
|
WHEN '已完成' THEN 'processed'
|
|||
|
|
WHEN '失败' THEN 'failed'
|
|||
|
|
ELSE LOWER(status)
|
|||
|
|
END
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
_UPDATE = text(f"""
|
|||
|
|
UPDATE memory_sync_outbox
|
|||
|
|
SET target_store = LOWER(target_store),
|
|||
|
|
operation = LOWER(operation),
|
|||
|
|
status = {_STATUS_CASE}
|
|||
|
|
WHERE target_store <> LOWER(target_store)
|
|||
|
|
OR operation <> LOWER(operation)
|
|||
|
|
OR status <> {_STATUS_CASE}
|
|||
|
|
""")
|
|||
|
|
|
|||
|
|
_GROUP = text(
|
|||
|
|
"SELECT target_store, operation, status, COUNT(*) n "
|
|||
|
|
"FROM memory_sync_outbox GROUP BY target_store, operation, status"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
_DETAIL = text(
|
|||
|
|
"SELECT id, event_uuid, target_store, operation, status, retry_count "
|
|||
|
|
"FROM memory_sync_outbox ORDER BY id"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
|
|||
|
|
async def main(apply: bool) -> int:
|
|||
|
|
async with engine.connect() as conn:
|
|||
|
|
before = (await conn.execute(_GROUP)).mappings().all()
|
|||
|
|
print("改前:", [dict(row) for row in before] or "（空表）")
|
|||
|
|
|
|||
|
|
affected = (await conn.execute(_UPDATE)).rowcount
|
|||
|
|
print("将订正行数:", affected)
|
|||
|
|
|
|||
|
|
if not apply:
|
|||
|
|
print("dry-run：未提交。确认无误后加 --apply 再运行。")
|
|||
|
|
await conn.rollback()
|
|||
|
|
return 0
|
|||
|
|
|
|||
|
|
await conn.commit()
|
|||
|
|
after = (await conn.execute(_GROUP)).mappings().all()
|
|||
|
|
print("改后:", [dict(row) for row in after] or "（空表）")
|
|||
|
|
for row in (await conn.execute(_DETAIL)).mappings().all():
|
|||
|
|
print(" ", dict(row))
|
|||
|
|
return 0
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
sys.exit(asyncio.run(main("--apply" in sys.argv)))
|