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 文件无错。
277 lines
12 KiB
Python
277 lines
12 KiB
Python
"""画像 → 图投影：把 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()
|