diff --git a/app/service/profile_graph_projection_service.py b/app/service/profile_graph_projection_service.py index fc9a12e..448b120 100644 --- a/app/service/profile_graph_projection_service.py +++ b/app/service/profile_graph_projection_service.py @@ -43,6 +43,28 @@ class ProjectionOutcome: reason: str = "" +@dataclass(frozen=True) +class ReconciliationOutcome: + """一次对账的结果：图里的边与权威画像应有的边是否一致。 + + `missing` = 画像里有、图里没有（投影漏了或落后）；`orphaned` = 图里有、画像已无 + （记忆失效后没清理）。两者都为空才算一致。 + """ + + customer_id: int + expected: tuple[tuple[str, str, str], ...] = () + actual: tuple[tuple[str, str, str], ...] = () + missing: tuple[tuple[str, str, str], ...] = () + orphaned: tuple[tuple[str, str, str], ...] = () + repaired: bool = False + degraded: bool = False + reason: str = "" + + @property + def consistent(self) -> bool: + return not self.missing and not self.orphaned + + class ProfileGraphProjectionService: def __init__( self, session: AsyncSession, relationships: RelationshipService | None @@ -142,6 +164,105 @@ class ProfileGraphProjectionService: updated_at=_now_iso(), ) + # ---------- 对账 ---------- + + async def reconcile_customer( + self, customer_id: int, *, repair: bool = False + ) -> ReconciliationOutcome: + """比对图里的边与权威画像应有的边，可选自动修复。 + + 与 `ProjectionReconciliationService` 的分工：那个服务做**事件层**对账（哪些投影 + 事件还没投递、需要重放），本方法做**数据层**对账（投递完成之后，图里的内容与权威 + 源是否一致）。两者互补——事件层保证"该投的都投了"，数据层保证"投进去的是对的、 + 不该留的没留"。 + + 图是投影、MySQL 是唯一真相，因此差异一律**以画像为准**：`missing` 补写、 + `orphaned` 删除，而不是反过来去改画像。 + """ + if self.relationships is None: + return ReconciliationOutcome( + customer_id, degraded=True, reason="graph_client_unavailable" + ) + expected = set(await self._edges(customer_id)) + try: + actual = await self._actual_edges(customer_id) + except Exception as exc: + return ReconciliationOutcome( + customer_id, expected=tuple(sorted(expected)), degraded=True, + reason=f"graph_read_failed:{type(exc).__name__}", + ) + missing = expected - actual + orphaned = actual - expected + repaired = False + if repair and (missing or orphaned): + try: + for relation, target_type, target_id in sorted(missing): + await self._merge_edge(customer_id, relation, target_type, target_id) + for relation, target_type, target_id in sorted(orphaned): + await self._delete_edge(customer_id, relation, target_type, target_id) + repaired = True + except Exception as exc: + return ReconciliationOutcome( + customer_id, expected=tuple(sorted(expected)), actual=tuple(sorted(actual)), + missing=tuple(sorted(missing)), orphaned=tuple(sorted(orphaned)), + degraded=True, reason=f"graph_repair_failed:{type(exc).__name__}", + ) + # 修复后重新核对：不凭"操作没报错"就宣布修好了 + actual = await self._actual_edges(customer_id) + missing = expected - actual + orphaned = actual - expected + return ReconciliationOutcome( + customer_id, + expected=tuple(sorted(expected)), + actual=tuple(sorted(actual)), + missing=tuple(sorted(missing)), + orphaned=tuple(sorted(orphaned)), + repaired=repaired, + ) + + async def _actual_edges(self, customer_id: int) -> set[tuple[str, str, str]]: + """读出该客户在图中现有的边：(关系, 目标类型, 目标标识)。""" + # 用 properties(t) 而不是 coalesce(t.tag_key, ...)：后者会引用当前图中尚不存在的 + # 属性名，Neo4j 每次执行都抛出 UnknownPropertyKeyWarning，把日志刷成噪音。 + rows = await self.relationships.driver.execute_query( # type: ignore[union-attr] + "MATCH (c:Customer {customer_id: $customer_id})-[r]->(t) " + "RETURN type(r) AS rel, labels(t) AS labels, properties(t) AS props", + customer_id=int(customer_id), + ) + edges: set[tuple[str, str, str]] = set() + for row in rows: + labels = row.get("labels") or [] + # 节点标签转回 payload 用的小写类型名（Tag -> tag），与 NODE_SPECS 对齐 + target_type = str(labels[0]).lower() if labels else "" + if not target_type: + continue + props = row.get("props") or {} + target_id = next( + (props[key] for key in ("tag_key", "product_code", "category", "event_id") + if props.get(key) is not None), + None, + ) + if target_id is None: + continue + edges.add((str(row.get("rel")), target_type, str(target_id))) + return edges + + async def _delete_edge( + self, customer_id: int, relation: str, target_type: str, target_id: str + ) -> None: + """删除一条边；节点本身保留——它可能被其他客户或产品共享。""" + if relation not in RelationshipService.ALLOWED_RELATIONSHIPS: + # 关系名取自图中的既有边，属于外部数据，必须过白名单再拼进 Cypher + raise ValueError(f"relationship is not allowed: {relation!r}") + target = node_spec(target_type) + query = ( + f"MATCH (c:Customer {{customer_id: $customer_id}})-[r:{relation}]->" + f"(t:{target.label} {{{target.property}: $target_id}}) DELETE r" + ) + await self.relationships.driver.execute_query( # type: ignore[union-attr] + query, customer_id=int(customer_id), target_id=target_id + ) + @staticmethod def _text(value: Any) -> str: if isinstance(value, str): diff --git a/tools/reconcile_graph.py b/tools/reconcile_graph.py new file mode 100644 index 0000000..41f4846 --- /dev/null +++ b/tools/reconcile_graph.py @@ -0,0 +1,98 @@ +"""图投影对账：检查图里的关系与权威画像是否一致，可选自动修复。 + +图是投影（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()))