group_fqcd_jr/app/service/profile_graph_projection_service.py

"""画像 → 图投影：把 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 = ""


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(),
        )

    @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()