group_fqcd_jr/app/service/profile_generation_service.py

"""画像版本生成服务（`docs/00` §6.4.6 的生成器实现）。

## 为什么需要它

底座**规划了** `profile_snapshots` 版本投影表，也**写明了规则**，但**没有生成器**：

> 「画像生成只允许读取已提交的业务事实和状态为"有效"的中期记忆。新画像和两条同步事件
> 必须在同一个 MySQL 事务中写入，旧画像的 `is_current` 同时置为 0。」
> —— `docs/00` §6.4.6

本服务把这条规则落成代码。**在它之前**，`profile_snapshots` 里的数据只能由种子脚本直接写，
"画像怎么生成的"没有可运行答案。

## 落地的四条规则

1. **同事务**：新画像版本 + **两条**跨存储同步事件（MILVUS / NEO4J）一起提交，任一失败整体回滚。
2. **旧版本同时置 0**：`is_current` 是 `profile_snapshots` 的唯一当前标记
   （`current_customer_id` 上有唯一键），必须先清旧再插新。
3. **幂等**：`snapshot_hash` 未变化时**不产生新版本**（返回 `changed=False`），
   避免每次调用都把版本号 +1、并往 outbox 灌重复事件。
4. **只读已提交事实**：数据来源限定为 `fin_customer_profile` 与 `fin_risk_assessment`
   （以及有效中期记忆的口径由 `generation_basis` 记录），不读取未提交状态。

## 与读取侧的关系

生成的 `snapshot` JSON 结构**必须与 `app/core/profile_projection.py` 的白名单字段一致** ——
投影是只读白名单，写侧多写字段不会被对外暴露，但白名单要求的字段**必须存在**，
否则端点会返回空画像。这一点由 `test_profile_generation_service.py` 固定。
"""

from __future__ import annotations

import hashlib
import json
from collections.abc import Mapping
from dataclasses import dataclass
from datetime import UTC, datetime
from typing import Any
from uuid import uuid4

from sqlalchemy.ext.asyncio import AsyncSession

from app.core.errors import ValidationAgentError
from app.core.profile_projection import PROFILE_FIELD_POLICY
from app.repository.profile_repository import ProfileRepository

#: `memory_sync_outbox` 的两个目标存储。
#:
#: ⚠️ **一律小写**：消费端按 `target_store` 的值分派 handler
#: （`MemorySyncOutboxWorker.handlers.get(event.target_store)`），且
#: `graph_projection_worker` / `projection_reconciliation_service` 的领取与重放
#: 都以 `status in {"pending","failed"}` + `"processed"` 为准。全仓（包含投顾线两处生产者）
#: 统一使用小写 `milvus`/`neo4j`、`upsert`、`pending`。
#:
#: 历史说明：`docs/00` §6.4.6 该栏曾写作大写 `MILVUS`/`NEO4J`、`UPSERT` 与中文 `待处理`，
#: 与上述实现从未对齐；本模块原先照文档写，是**全仓唯一的异类**，导致自己写的事件
#: 任何消费者都领不到。现统一为小写，并已把历史 2 行就地改齐（主键/唯一键不变）。
TARGET_MILVUS = "milvus"
TARGET_NEO4J = "neo4j"
SYNC_TARGETS: tuple[str, ...] = (TARGET_MILVUS, TARGET_NEO4J)

#: 同步操作与状态取值。
SYNC_OPERATION_UPSERT = "upsert"
SYNC_STATUS_PENDING = "pending"
AGGREGATE_TYPE_PROFILE = "profile"


@dataclass(frozen=True)
class ProfileGenerationResult:
    """生成结果；`changed=False` 表示快照未变化、**未写任何行**。"""

    customer_id: int
    version: int
    profile_uuid: str
    snapshot_hash: str
    changed: bool
    sync_events: int


def build_snapshot(
    profile_row: Mapping[str, Any],
    assessment_row: Mapping[str, Any] | None,
    *,
    now: datetime | None = None,
) -> dict[str, object]:
    """把权威事实投影成画像快照（字段与读取侧白名单对齐）。

    - `assessment_valid_until` / `assessment_expired` **由测评行实时推出**，不沿用旧快照的布尔值；
      没有测评行时两者都**不写入**（读取侧据此表现为"无测评信息"，而不是伪造一个有效期）。
    - 时间字段统一用 ISO 字符串，保证 JSON 稳定、`snapshot_hash` 可复现。
    """
    current = now or datetime.now(UTC)
    snapshot: dict[str, object] = {
        "investor_type": profile_row["investor_type"],
        "investment_horizon": profile_row["investment_horizon"],
        "trading_frequency": profile_row["trading_frequency"],
        "preferred_asset_class": _as_list(profile_row["preferred_asset_class"]),
        "risk_tags": _as_list(profile_row["risk_tags"]),
        "behavior_score": profile_row["behavior_score"],
        "total_asset": _as_money(profile_row["total_asset"]),
    }
    if profile_row["last_active_at"] is not None:
        snapshot["last_active_at"] = _iso(profile_row["last_active_at"])

    if assessment_row is not None and assessment_row["valid_until"] is not None:
        valid_until = _instant(assessment_row["valid_until"])
        snapshot["assessment_valid_until"] = valid_until.isoformat()
        snapshot["assessment_expired"] = valid_until <= current
    return snapshot


def compute_hash(snapshot: dict[str, object]) -> str:
    """`snapshot_hash`：对**键排序后**的规范化 JSON 求 SHA-256。

    排序是为了让"同一内容"总是得到同一哈希 —— 它是幂等判据，不能被字典插入顺序影响。
    """
    canonical = json.dumps(snapshot, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


class ProfileGenerationService:
    """生成客户画像版本；**不 commit**，事务归调用方。"""

    def __init__(self, session: AsyncSession) -> None:
        self._session = session
        self._repo = ProfileRepository(session)

    async def generate(
        self, customer_id: int, *, now: datetime | None = None
    ) -> ProfileGenerationResult:
        current = now or datetime.now(UTC).replace(tzinfo=None)

        profile_row = await self._repo.profile_row(customer_id)
        if profile_row is None:
            # 失败关闭：没有客户主表行就不该凭空造画像。
            raise ValidationAgentError(f"客户不存在或缺少画像主表行：{customer_id}")
        assessment_row = await self._repo.latest_assessment(customer_id)

        snapshot = build_snapshot(profile_row, assessment_row, now=current.replace(tzinfo=UTC))
        snapshot_hash = compute_hash(snapshot)

        existing = await self._repo.current_snapshot(customer_id)
        if existing is not None and str(existing["snapshot_hash"] or "") == snapshot_hash:
            # 幂等：内容没变就不升版本、不投事件。
            return ProfileGenerationResult(
                customer_id=customer_id,
                version=int(existing["version"]),
                profile_uuid=str(existing["profile_uuid"] or ""),
                snapshot_hash=snapshot_hash,
                changed=False,
                sync_events=0,
            )

        # `profile_uuid` 上有**唯一键**（跨版本也唯一），所以每个版本必须用新 uuid ——
        # 不能沿用以表示"同一实体"（实测会撞 `Duplicate entry ... for key profile_uuid`）。
        # 待接入投影消费者时，若 Milvus/Neo4j 需要**跨版本稳定的实体标识**，
        # 应另加一列（例如 `entity_uuid`）区分"版本标识"与"实体标识"，不要复用本列。
        profile_uuid = str(uuid4())
        version = await self._repo.next_version(customer_id)

        # 顺序要紧：先清旧当前标记（唯一键约束），再插新版本。
        await self._repo.clear_current(customer_id, now=current)
        await self._repo.insert_snapshot(
            customer_id=customer_id,
            version=version,
            profile_uuid=profile_uuid,
            snapshot_json=json.dumps(snapshot, ensure_ascii=False, sort_keys=True),
            generation_basis_json=json.dumps(
                self._generation_basis(assessment_row), ensure_ascii=False, sort_keys=True
            ),
            snapshot_hash=snapshot_hash,
            now=current,
        )

        # 两条同步事件共用同一个 event_uuid（唯一键是 (event_uuid, target_store)）。
        event_uuid = str(uuid4())
        payload = {
            "aggregate_uuid": profile_uuid,
            "customer_id": str(customer_id),
            "version": version,
            # 别名：投影适配器契约用 `profile_version`。两个键都写，避免消费端
            # 因生产者用词不同而取不到值（本仓三种 payload 形状的历史遗留）。
            "profile_version": version,
            "snapshot_hash": snapshot_hash,
            "snapshot": snapshot,
            "memory_sources": await self._memory_sources(customer_id),
        }
        for target in SYNC_TARGETS:
            self._repo.add_sync_event(
                event_uuid=event_uuid,
                target_store=target,
                aggregate_uuid=profile_uuid,
                aggregate_version=version,
                payload=payload,
                now=current,
                status=SYNC_STATUS_PENDING,
                operation=SYNC_OPERATION_UPSERT,
            )

        return ProfileGenerationResult(
            customer_id=customer_id,
            version=version,
            profile_uuid=profile_uuid,
            snapshot_hash=snapshot_hash,
            changed=True,
            sync_events=len(SYNC_TARGETS),
        )

    async def _memory_sources(self, customer_id: int) -> list[dict[str, Any]]:
        """组装投影适配器要的 `memory_sources`（数据源：`memory_unit` 中 `status='active'`）。

        为什么由生产端组装而不是消费端回查：消费端到时那条记忆可能已改版本，
        让它自己去查会把"投递的是哪一版"变成不确定。事件里带上当时的确定快照，
        投递语义才与 `aggregate_version` 一致（幂等判据也才有意义）。

        为什么是 `memory_unit` 而不是 `user_facts`：两者用途不同——`user_facts` 是
        **已确认的结构化事实**，喂画像与图投影；`memory_unit` 是**长期记忆条目**，
        正是长期记忆向量集合要存的东西。这不与"图投影只读 user_facts"的不变式冲突，
        因为 Milvus 存的就是记忆本身的向量，不是画像事实的另一种说法。
        """
        rows = await self._repo.active_memories(customer_id)
        return [
            {
                "memory_uuid": str(row["memory_uuid"]),
                "memory_key": str(row["memory_key"]),
                "content": str(row["content"]),
                "memory_type": str(row["memory_type"]),
                "confidence": float(row["confidence"]),
                "version": int(row["version"]),
                "valid_until": (
                    row["valid_until"].isoformat() if row["valid_until"] else None
                ),
            }
            for row in rows
        ]

    @staticmethod
    def _generation_basis(assessment_row: Mapping[str, Any] | None) -> dict[str, object]:
        """记录这次画像**依据了什么**（`docs/00`：使用的测评版本、交易窗口和记忆版本列表）。

        当前只接入了测评版本；记忆与交易窗口待接入时补键，不臆造。
        """
        basis: dict[str, object] = {"sources": ["fin_customer_profile", "fin_risk_assessment"]}
        if assessment_row is not None:
            basis["questionnaire_version"] = assessment_row["questionnaire_version"]
            basis["assessment_id"] = int(assessment_row["id"])
        return basis


# ---------------------------------------------------------------------------
# 取值归一化
# ---------------------------------------------------------------------------


def _as_list(value: object) -> list[object]:
    if value is None:
        return []
    if isinstance(value, (list, tuple)):
        return list(value)
    if isinstance(value, str):
        try:
            parsed = json.loads(value)
        except ValueError:
            return [value]
        return list(parsed) if isinstance(parsed, list) else [value]
    return [value]


def _as_money(value: object) -> str:
    """金额统一成字符串（与读取侧 `public()` 的口径一致，避免浮点误差）。"""
    return "" if value is None else str(value)


def _instant(value: object) -> datetime:
    if isinstance(value, datetime):
        return value if value.tzinfo is not None else value.replace(tzinfo=UTC)
    return datetime.fromisoformat(str(value)).replace(tzinfo=UTC)


def _iso(value: object) -> str:
    return _instant(value).isoformat()


#: 导出以便测试断言"写侧字段覆盖了读取侧白名单所需项"。
REQUIRED_SNAPSHOT_FIELDS: tuple[str, ...] = tuple(
    field for field in PROFILE_FIELD_POLICY if field != "customer_tier"
)