group_fqcd_jr/app/repository/profile_repository.py

"""画像生成的数据访问层（Repository）。

按 MVC+S 边界：**Service 不得直接建 Session / 直接查 Model**，所有库交互收口在这里。

只做两件事：
1. **只读**组装画像所需的权威事实（`docs/00` §6.4.6：画像生成只允许读取**已提交的**业务事实
   和状态为**有效**的中期记忆）；
2. **写入**画像版本与两条跨存储同步事件（同事务，由调用方管理事务）。
"""

from __future__ import annotations

from datetime import datetime
from typing import Any

from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

from app.model.memory import MemorySyncOutbox

#: 画像快照：查询客户主表 + 客户分层。
#:
#: **为什么必须 JOIN `sys_user`**：`customer_tier`（客户分层：金卡 / 白金 / 钻石 / 尊享）的
#: 权威列在 `sys_user` 上，`fin_customer_profile` **没有**这一列。而 `D3.1` §3.5 与 `DEC-19`
#: 裁定 (a) 都把「读 `customer_level`」列为客服侧的**受准只读字段**（供确定性规则与画像问答），
#: `PROFILE_FIELD_POLICY` / `customer_profile_service.ALLOWED_FIELDS` 也早已把它写进白名单 ——
#: 唯独没有取数来源，于是它在"白名单里躺了很久却永远是空"。这里补上取数，字段才真正可达。
#:
#: 用 `LEFT JOIN`（不是 `INNER JOIN`）：客户可能没有 `sys_user` 行（历史数据），
#: 此时画像仍应生成，只是分层为空 —— 不能因为缺一行账号把整张画像判成"不存在"。
_SQL_PROFILE = text("""
    SELECT p.customer_id, p.investor_type, p.investment_horizon, p.preferred_asset_class,
           p.trading_frequency, p.total_asset, p.behavior_score, p.risk_tags, p.last_active_at,
           u.customer_tier
    FROM fin_customer_profile AS p
    LEFT JOIN sys_user AS u ON u.id = p.customer_id
    WHERE p.customer_id = :customer_id
""")

#: 风险测评：取**最新一条**，含有效期（用于判定是否过期）。
_SQL_ASSESSMENT = text("""
    SELECT id, questionnaire_version, investor_type, total_score, assessed_at, valid_until
    FROM fin_risk_assessment
    WHERE customer_id = :customer_id
    ORDER BY assessed_at DESC, id DESC
    LIMIT 1
""")

#: 当前画像版本（取 `is_current=1` 那一条）。
_SQL_CURRENT_SNAPSHOT = text("""
    SELECT id, profile_uuid, version, snapshot, snapshot_hash, generation_basis
    FROM profile_snapshots
    WHERE customer_id = :customer_id AND is_current = 1
    LIMIT 1
""")

#: 下一版本号 = 当前最大版本 + 1
#: （`uk_profile_snapshot_version (customer_id, version)` 要求版本号按客户唯一）。
_SQL_NEXT_VERSION = text("""
    SELECT COALESCE(MAX(version), 0) + 1
    FROM profile_snapshots WHERE customer_id = :customer_id
""")

#: 该客户**有效**的长期记忆（投影到 Milvus 长期记忆向量集合的数据源）。
#:
#: 只取 `status='active'`：失效/被取代的记忆不应再进入向量召回，否则投顾会召回
#: 已过期偏好。字段与 `MilvusProfileProjection` 的 `memory_sources` 契约一一对应。
_SQL_ACTIVE_MEMORIES = text("""
    SELECT memory_uuid, memory_key, content, memory_type, confidence, version, valid_until
    FROM memory_unit
    WHERE customer_id = :customer_id AND status = 'active'
    ORDER BY id
""")

#: 旧画像置非当前（`current_customer_id` 有唯一键，必须先置 0 才能插新的当前版本）。
_SQL_CLEAR_CURRENT = text("""
    UPDATE profile_snapshots
    SET is_current = 0, current_customer_id = NULL, updated_at = :now
    WHERE customer_id = :customer_id AND is_current = 1
""")

_SQL_INSERT_SNAPSHOT = text("""
    INSERT INTO profile_snapshots
        (profile_uuid, customer_id, version, snapshot, generation_basis, snapshot_hash,
         is_current, current_customer_id, generated_at, created_at, updated_at)
    VALUES
        (:profile_uuid, :customer_id, :version, :snapshot, :generation_basis, :snapshot_hash,
         1, :customer_id, :now, :now, :now)
""")


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

    # --- 只读事实 ---------------------------------------------------------

    async def profile_row(self, customer_id: int) -> dict[str, Any] | None:
        result = await self._session.execute(_SQL_PROFILE, {"customer_id": customer_id})
        row = result.mappings().first()
        return dict(row) if row is not None else None

    async def latest_assessment(self, customer_id: int) -> dict[str, Any] | None:
        result = await self._session.execute(_SQL_ASSESSMENT, {"customer_id": customer_id})
        row = result.mappings().first()
        return dict(row) if row is not None else None

    async def current_snapshot(self, customer_id: int) -> dict[str, Any] | None:
        result = await self._session.execute(_SQL_CURRENT_SNAPSHOT, {"customer_id": customer_id})
        row = result.mappings().first()
        return dict(row) if row is not None else None

    async def active_memories(self, customer_id: int) -> list[dict[str, Any]]:
        """该客户所有 `status='active'` 的长期记忆，按 id 稳定排序。"""
        result = await self._session.execute(
            _SQL_ACTIVE_MEMORIES, {"customer_id": customer_id}
        )
        return [dict(row) for row in result.mappings().all()]

    async def next_version(self, customer_id: int) -> int:
        return int(await self._session.scalar(_SQL_NEXT_VERSION, {"customer_id": customer_id}))

    # --- 写入（调用方管事务） ---------------------------------------------

    async def clear_current(self, customer_id: int, *, now: datetime) -> None:
        """把旧画像的 `is_current` 置 0 —— 必须在插入新版本**之前**调用。

        `uk_profile_snapshot_current` 建在 `current_customer_id` 上（唯一），
        所以同一客户同时只能有一条 `is_current=1`；顺序颠倒会撞唯一键。
        """
        await self._session.execute(_SQL_CLEAR_CURRENT, {"customer_id": customer_id, "now": now})

    async def insert_snapshot(
        self,
        *,
        customer_id: int,
        version: int,
        profile_uuid: str,
        snapshot_json: str,
        generation_basis_json: str,
        snapshot_hash: str,
        now: datetime,
    ) -> None:
        await self._session.execute(_SQL_INSERT_SNAPSHOT, {
            "profile_uuid": profile_uuid,
            "customer_id": customer_id,
            "version": version,
            "snapshot": snapshot_json,
            "generation_basis": generation_basis_json,
            "snapshot_hash": snapshot_hash,
            "now": now,
        })

    def add_sync_event(
        self,
        *,
        event_uuid: str,
        target_store: str,
        aggregate_uuid: str,
        aggregate_version: int,
        payload: dict[str, Any],
        now: datetime,
        status: str,
        operation: str,
    ) -> None:
        """追加一条跨存储同步事件。

        唯一键 `uk_memory_sync_event (event_uuid, target_store)`：**同一 `event_uuid` 允许
        对两个目标库各写一条**，因此 Milvus 与 Neo4j 共用同一个 `event_uuid`。
        """
        self._session.add(MemorySyncOutbox(
            event_uuid=event_uuid,
            aggregate_type="profile",
            aggregate_uuid=aggregate_uuid,
            aggregate_version=aggregate_version,
            target_store=target_store,
            operation=operation,
            payload=payload,
            status=status,
            retry_count=0,
            created_at=now,
        ))