lzf_0626
89f889b350
与既有 `ProjectionReconciliationService` 的分工（两者互补，不是重复实现）： · 那个服务做**事件层**对账：哪些投影事件还没投递、需要重放； · 本次新增的是**数据层**对账：投递完成之后，图里的内容与权威画像是否一致 （投影漏投、记忆失效后未清理、图库故障都会造成漂移）。 实现 `ProfileGraphProjectionService.reconcile_customer`： · 从 user_facts 与持仓算出"应有的边"，从图中读出"实际的边"，求差集得到 missing （画像有、图里没有）与 orphaned（图里有、画像已无）； · 图是投影、MySQL 是唯一真相，因此差异一律**以画像为准**：missing 补写、orphaned 删除， 而不是反过来去改画像； · `repair=True` 时修复，并**修复后重新核对**再回报——不凭"操作没报错"就宣布修好了； · 删除边前对关系名做白名单校验：关系名读自图中既有边，属外部数据， 必须过白名单才允许拼进 Cypher。 新增 tools/reconcile_graph.py：支持单客户与 `--all`、可选 `--repair`，退出码可直接用于巡检。 顺带修掉一处日志噪音：读边时原用 `coalesce(t.tag_key, t.product_code, ...)`，会引用当前 图中尚不存在的属性名，Neo4j 每次执行都抛 UnknownPropertyKeyWarning，把日志刷成噪音。 改用 `properties(t)` 后在应用侧按键取值，功能不变、日志干净。 实测（人为制造漂移再修复）： · 初始对账 一致=True、应有 2 条 / 实际 2 条； · 删掉图中的 HAS_GOAL 边后 一致=False，缺失被准确报出； · repair=True → 已修复=True、一致=True、缺失为空； · 复验 一致=True，图中恢复 HAS_GOAL 与 PREFERS 两条关系； · tools/reconcile_graph.py 单客户与 --all 均 EXIT=0，输出无警告； · ruff 通过、mypy 112 文件无错。
99 lines
3.8 KiB
Python
99 lines
3.8 KiB
Python
"""图投影对账：检查图里的关系与权威画像是否一致，可选自动修复。
|
|
|
|
图是投影（MySQL 才是唯一真相），所以它可能因为投影漏投、记忆失效后未清理、或图库故障
|
|
而与画像不一致。本工具用**数据层对账**回答"图里现在的内容对不对"，与
|
|
`ProjectionReconciliationService` 的**事件层对账**（哪些投影事件还没投递）互补。
|
|
|
|
用法：
|
|
|
|
```powershell
|
|
# 只检查，不改动
|
|
python tools/reconcile_graph.py 9001
|
|
|
|
# 检查并修复（缺失的补写、多余的删除）
|
|
python tools/reconcile_graph.py 9001 --repair
|
|
|
|
# 检查所有有记忆数据的客户
|
|
python tools/reconcile_graph.py --all
|
|
```
|
|
|
|
退出码：0 = 全部一致（或已修复），1 = 存在不一致（未修复）或对账降级。
|
|
"""
|
|
|
|
import argparse
|
|
import asyncio
|
|
import sys
|
|
|
|
from sqlalchemy import select
|
|
|
|
from app.infrastructure.db import SessionFactory
|
|
from app.infrastructure.graph import build_graph_driver
|
|
from app.model.memory import MemoryUnit
|
|
from app.service.profile_graph_projection_service import ProfileGraphProjectionService
|
|
from app.service.relationship_service import RelationshipService
|
|
|
|
# GBK 控制台下图里的中文标签可能含不可编码字符，替换而不是让脚本自身崩掉
|
|
sys.stdout.reconfigure(errors="replace")
|
|
|
|
|
|
async def reconcile_one(customer_id: int, *, repair: bool) -> bool:
|
|
driver = build_graph_driver()
|
|
if driver is None:
|
|
print("图驱动不可用：检查 .env 中的 NEO4J_URI / NEO4J_PASSWORD")
|
|
return False
|
|
graph = RelationshipService(driver)
|
|
try:
|
|
async with SessionFactory() as session:
|
|
service = ProfileGraphProjectionService(session, graph)
|
|
outcome = await service.reconcile_customer(customer_id, repair=repair)
|
|
finally:
|
|
await driver.close()
|
|
|
|
print(f"\n== 客户 {customer_id} ==")
|
|
if outcome.degraded:
|
|
print(f" 对账降级：{outcome.reason}")
|
|
return False
|
|
print(f" 应有 {len(outcome.expected)} 条 / 图里 {len(outcome.actual)} 条")
|
|
if outcome.missing:
|
|
print(f" 缺失（画像有、图里没有）{len(outcome.missing)} 条：")
|
|
for relation, target_type, target_id in outcome.missing:
|
|
print(f" + {relation} -> {target_type}:{target_id}")
|
|
if outcome.orphaned:
|
|
print(f" 多余（图里有、画像已无）{len(outcome.orphaned)} 条：")
|
|
for relation, target_type, target_id in outcome.orphaned:
|
|
print(f" - {relation} -> {target_type}:{target_id}")
|
|
if outcome.consistent:
|
|
print(" 一致")
|
|
elif repair and outcome.repaired:
|
|
print(" 已按画像修复（缺失已补写、多余已删除）")
|
|
else:
|
|
print(" 不一致：加 --repair 可按画像修复")
|
|
return outcome.consistent
|
|
|
|
|
|
async def main() -> int:
|
|
parser = argparse.ArgumentParser(description="图投影对账")
|
|
parser.add_argument("customer_id", nargs="?", type=int, help="客户 id")
|
|
parser.add_argument("--all", action="store_true", help="对账所有有记忆数据的客户")
|
|
parser.add_argument("--repair", action="store_true", help="按画像修复差异")
|
|
args = parser.parse_args()
|
|
|
|
if args.all:
|
|
async with SessionFactory() as session:
|
|
rows = list(await session.scalars(select(MemoryUnit.customer_id).distinct()))
|
|
targets = sorted({int(row) for row in rows})
|
|
if not targets:
|
|
print("没有任何客户有记忆数据，无需对账")
|
|
return 0
|
|
print(f"将对账 {len(targets)} 个客户：{targets}")
|
|
results = [await reconcile_one(cid, repair=args.repair) for cid in targets]
|
|
return 0 if all(results) else 1
|
|
|
|
if args.customer_id is None:
|
|
parser.print_help()
|
|
return 2
|
|
return 0 if await reconcile_one(args.customer_id, repair=args.repair) else 1
|
|
|
|
|
|
sys.exit(asyncio.run(main()))
|