lzf_0626
7635014d9e
第 2 步（Neo4j）主体，三件事：
1. 新增 app/service/graph_model.py：图模型的**单一来源**。
节点标签与主属性名会被拼进 Cypher（Neo4j 的标签不能用参数占位），因此必须来自受控常量，
绝不能是调用方传入的字符串——否则就是 Cypher 注入；同时读服务与投影侧必须就"客户节点
长什么样"达成一致。此前两边各写各的（投影写 :Entity{entity_id}、读服务查
:Customer{customer_id}），结果写进去的关系永远读不出来。
另用 RELATION_SEMANTICS 补上"谁指向谁"的方向校验：底座已有关系白名单（8 种），
但白名单只约束关系名，不约束方向，这里补上，避免把 TRADED 写成 Customer→Tag。
2. 改 app/worker/graph_projection_worker.py：按 payload 的 source_type/target_type 写具体标签，
投影前校验关系方向。标签与属性名全部取自 graph_model，调用方传不进任意字符串。
3. 新增 app/service/profile_graph_projection_service.py：把画像事实投影成节点与关系，
并补上此前缺失的**投影删除客户端**（销户或记忆失效时 DETACH DELETE，同时清悬挂边）。
只投影 user_facts（与画像同源），不直接读原始记忆——否则同一件事在画像与图里会有两种说法。
图库故障一律返回 degraded 而不抛异常：投顾推荐可以暂时没有图，但不该因为图挂了
导致画像更新失败。
背景（核查发现）：MemorySyncOutbox 与 GraphProjectionWorker 此前是**孤儿代码**——表建了、
worker 也实现了（含幂等、重试、死信、MERGE），但没有任何代码往 outbox 写、也没有任何地方
实例化这个 worker，整条图投影链路从未接上过。这正是图中只有 Neo4j 自带 Person/Movie
示例数据的根因。本次把"画像 → 图"这条链路接通；领域事件驱动的投影仍待接入（worker 已修好可用）。
实测验证：
· 投影客户 9001 → relations=1，随后 neighbors(PREFERS) 直接读到
{'tag_key': 'preference:risk_level=稳健型'} —— 写后读通，标签已对齐；
· delete_customer → 关系归零（degraded=False），重新投影恢复为 1，再次投影仍为 1（幂等）；
· 图中标签为 Customer/Tag、关系为 PREFERS；早先探针残留的 Entity 节点已清理；
· ruff 通过、mypy 112 文件无错。
81 lines
3.4 KiB
Python
81 lines
3.4 KiB
Python
"""图模型定义：节点类型、主属性与关系语义（**单一来源**）。
|
|
|
|
为什么必须集中定义：
|
|
|
|
1. **标签与属性名要拼进 Cypher**（Neo4j 的标签不能用查询参数占位），因此它们**必须**来自
|
|
受控常量，绝不能接受调用方传入的任意字符串，否则就是 Cypher 注入。
|
|
2. 读服务（`RelationshipService`）与投影侧需要就"客户节点长什么样"达成一致。此前投影写
|
|
`:Entity {entity_id}`、读服务查 `:Customer {customer_id}`，两边各写各的，结果是**写进去的
|
|
关系永远读不出来**。把节点规格放在一处，两边都从这里取，才不会再次漂移。
|
|
3. `RELATION_SEMANTICS` 记录每种关系连接哪两类节点，用于在投影时拒绝无意义的边
|
|
（例如把 `TRADED` 写成 Customer→Tag）。底座已有关系白名单（8 种），这里补的是
|
|
"谁指向谁"的语义约束。
|
|
"""
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class NodeSpec:
|
|
"""一个节点类型的规格。`cast` 决定主键类型：客户 id 是整数（与读服务查询一致）。"""
|
|
|
|
type_key: str
|
|
label: str
|
|
property: str
|
|
cast: type
|
|
|
|
|
|
# payload 里使用的类型名 → 节点规格。未登记的类型一律拒绝。
|
|
NODE_SPECS: dict[str, NodeSpec] = {
|
|
"customer": NodeSpec("customer", "Customer", "customer_id", int),
|
|
"product": NodeSpec("product", "Product", "product_code", str),
|
|
"tag": NodeSpec("tag", "Tag", "tag_key", str),
|
|
"industry": NodeSpec("industry", "Industry", "category", str),
|
|
"event": NodeSpec("event", "Event", "event_id", str),
|
|
}
|
|
|
|
# 关系 → (允许的源节点类型, 允许的目标节点类型)
|
|
RELATION_SEMANTICS: dict[str, tuple[tuple[str, ...], tuple[str, ...]]] = {
|
|
"PREFERS": (("customer",), ("tag",)),
|
|
"HAS_GOAL": (("customer",), ("tag",)),
|
|
"INTERESTED_IN": (("customer",), ("product", "industry")),
|
|
"TRADED": (("customer",), ("product",)),
|
|
"HOLDS": (("customer",), ("product",)),
|
|
"TRIGGERED_RISK": (("customer",), ("event",)),
|
|
"BELONGS_TO_CATEGORY": (("product",), ("tag",)),
|
|
"EXPOSED_TO_INDUSTRY": (("product", "customer"), ("industry",)),
|
|
}
|
|
|
|
|
|
def node_spec(type_key: object) -> NodeSpec:
|
|
"""按类型名取节点规格；未登记的类型抛 `ValueError`（宁可失败也不拼出任意标签）。"""
|
|
key = str(type_key or "").strip().lower()
|
|
spec = NODE_SPECS.get(key)
|
|
if spec is None:
|
|
raise ValueError(f"node type is not allowed: {type_key!r}")
|
|
return spec
|
|
|
|
|
|
def node_id(spec: NodeSpec, value: Any) -> Any:
|
|
"""把主键值转成该节点应有的类型。"""
|
|
return value if spec.cast is str else spec.cast(value)
|
|
|
|
|
|
def relation_allowed(relation: str, source_type: str, target_type: str) -> bool:
|
|
"""该关系是否允许连接这两类节点。"""
|
|
expected = RELATION_SEMANTICS.get(relation)
|
|
if expected is None:
|
|
return False
|
|
sources, targets = expected
|
|
return source_type in sources and target_type in targets
|
|
|
|
|
|
def merge_node_clause(variable: str, spec: NodeSpec, id_parameter: str) -> str:
|
|
"""生成 `MERGE (a:Label {prop: $param})` 子句。
|
|
|
|
只有 `spec` 里的标签与属性名会进入查询文本（它们来自本模块常量），`$param` 走参数绑定，
|
|
因此调用方无法通过数据影响 Cypher 结构。
|
|
"""
|
|
return f"MERGE ({variable}:{spec.label} {{{spec.property}: ${id_parameter}}})"
|