feat: 图投影对账（数据层）并补运维工具，第 2 步收尾

与既有 `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 文件无错。
2026-09-10 21:55:03 +08:00
parent d7f6ef7ddc
commit 89f889b350
2 changed files with 219 additions and 0 deletions
@@ -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):