"""画像 → 图投影：把 MySQL 里权威的画像与事实投影成 Neo4j 的节点与关系。 **为什么要有图**（这是它存在的理由，不是技术偏好）：投顾要靠"买过 A 的客户还买过什么" 这类多跳关系做组合推荐，风控要靠关系网络发现关联账户与资金链路；这些查询用关系库写 既别扭又慢，用图是顺手的事。 设计约束： 1. **单向投影**：MySQL 是唯一真相，图库只是派生视图。图数据损坏或模型变更时可以直接重建 （拿同一份画像重投影即可），因此这里**不需要与 MySQL 保持事务一致**。 2. **幂等**：全部用 MERGE，重复投影不会产生重复节点或关系。 3. **降级**：图库不可用时返回 `degraded`，**绝不把异常抛给画像重建这类主链路** —— 投顾推荐可以暂时没有图，但不能因为图挂了就让画像更新失败。 4. **只投影"已确认"的事实**：与画像同源（`user_facts`），不直接读原始记忆， 保证图里的偏好标签与画像口径一致；否则同一件事在画像和图里会有两种说法。 """ from dataclasses import dataclass from typing import Any from sqlalchemy import select, text from sqlalchemy.ext.asyncio import AsyncSession from app.model.profile import UserFact from app.service.graph_model import merge_node_clause, node_spec, relation_allowed from app.service.relationship_service import RelationshipService # 事实键 → (关系类型, 目标节点类型) FACT_PROJECTION: dict[str, tuple[str, str]] = { "preference:risk_level": ("PREFERS", "tag"), "preference:asset_class": ("PREFERS", "tag"), "preference:horizon": ("HAS_GOAL", "tag"), "profile:family": ("PREFERS", "tag"), } MAX_EDGES_PER_CUSTOMER = 200 @dataclass(frozen=True) class ProjectionOutcome: relations: int = 0 degraded: bool = False 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 ) -> None: self.session = session self.relationships = relationships async def project_customer(self, customer_id: int) -> ProjectionOutcome: """把该客户的画像事实投影到图中。""" if self.relationships is None: return ProjectionOutcome(degraded=True, reason="graph_client_unavailable") edges = await self._edges(customer_id) written = 0 try: for relation, target_type, target_id in edges[:MAX_EDGES_PER_CUSTOMER]: await self._merge_edge(customer_id, relation, target_type, target_id) written += 1 except Exception as exc: # 图库故障：如实标记降级，不伪装成功（与 Milvus/Neo4j 投影清理同一取向） return ProjectionOutcome(relations=written, degraded=True, reason=f"graph_write_failed:{type(exc).__name__}") return ProjectionOutcome(relations=written) async def delete_customer(self, customer_id: int) -> ProjectionOutcome: """删除该客户在图中的节点及其全部关系。 这是此前缺失的**投影删除客户端**：记忆失效或客户销户后，若图里仍留着旧关系， 投顾会拿过期偏好做推荐、风控会看到已不存在的关系网络。 `DETACH DELETE` 同时清理关系，避免留下悬挂边。 """ if self.relationships is None: return ProjectionOutcome(degraded=True, reason="graph_client_unavailable") try: await self.relationships.driver.execute_query( "MATCH (c:Customer {customer_id: $customer_id}) DETACH DELETE c", customer_id=int(customer_id), ) except Exception as exc: return ProjectionOutcome(degraded=True, reason=f"graph_delete_failed:{type(exc).__name__}") return ProjectionOutcome() # ---------- 内部 ---------- async def _edges(self, customer_id: int) -> list[tuple[str, str, str]]: """收集要投影的边：(关系, 目标节点类型, 目标节点标识)。""" edges: list[tuple[str, str, str]] = [] facts = list(await self.session.scalars( select(UserFact).where(UserFact.customer_id == customer_id) )) for fact in facts: key = str(fact.fact_key) projection = FACT_PROJECTION.get(key) if projection is None: continue relation, target_type = projection edges.append((relation, target_type, f"{key}={self._text(fact.fact_value)}")) for product_code in await self._holdings(customer_id): edges.append(("HOLDS", "product", product_code)) return edges async def _holdings(self, customer_id: int) -> list[str]: """客户当前持仓的产品代码。 交易模块尚未产生数据，或表结构与预期不符时返回空列表——持仓投影自动跳过， 而不是让整次投影失败。等交易侧就绪后无需改这里的代码。 """ try: rows = await self.session.execute(text( "SELECT p.product_code FROM fin_holding h " "JOIN fin_product p ON p.id = h.product_id " "WHERE h.customer_id = :customer_id LIMIT 50" ), {"customer_id": customer_id}) except Exception: return [] return [str(row[0]) for row in rows if row[0]] async def _merge_edge( self, customer_id: int, relation: str, target_type: str, target_id: str ) -> None: if not relation_allowed(relation, "customer", target_type): raise ValueError(f"relationship {relation} cannot connect customer -> {target_type}") source = node_spec("customer") target = node_spec(target_type) query = ( f"{merge_node_clause('a', source, 'source_id')} " f"{merge_node_clause('b', target, 'target_id')} " f"MERGE (a)-[r:{relation}]->(b) " "SET r.source = $source, r.updated_at = $updated_at" ) await self.relationships.driver.execute_query( # type: ignore[union-attr] query, source_id=node_spec("customer").cast(customer_id), target_id=target_id, source="user_facts", 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): return value.strip().strip('"') return str(value) def _now_iso() -> str: from datetime import UTC, datetime return datetime.now(UTC).replace(tzinfo=None).isoformat()