group_fqcd_jr/app/service/profile_assembly_service.py

"""画像组装：中期记忆 → 长期事实 → 画像 + 版本快照。

这是"记忆系统为画像服务"的落地环节。三段职责：

1. **事实提升（中期 → 长期）**：把 `memory_unit` 里证据足够的记忆提炼进 `user_facts`。
   门槛是 `evidence_count >= 2` **或** `confidence >= 0.90` —— 这条门槛就是
   "客户随口一说不能变成画像结论"的落地方式。
2. **画像组装（长期 → 画像）**：把 `user_facts` 按**白名单**映射进 `fin_customer_profile`。
   未列入白名单的事实只进 `user_facts`，不进画像，避免画像被噪声撑大。
3. **版本留痕**：每次重建写一条 `profile_snapshots`，并用 `generation_basis` 记录
   **每个字段分别来自哪里**——风控与合规复盘时要能回答"当时凭什么这么判断"。

## 两条必须由代码保证的红线

- **`investor_type` 只来自问卷测评**（`fin_risk_assessment` 最新一条）。下面的实现里
  它只从问卷查询取数，任何记忆路径都碰不到它。客户在对话里说"我是激进型"不会改变它
  ——这是合规底线，不能只靠约定。
- **按字段所有权写入**：交易侧的客观字段（`total_asset`/`trading_frequency`/`behavior_score`）
  本服务**不写**，留给交易模块，避免两个模块抢写同一列。
"""

import json as _json
from datetime import UTC, datetime
from hashlib import sha256
from typing import Any
from uuid import uuid4

from sqlalchemy import or_, select, text
from sqlalchemy.ext.asyncio import AsyncSession

from app.model.fund import FundCustomerProfile
from app.model.memory import MemoryUnit
from app.model.profile import ProfileSnapshot, UserFact

# 提升门槛
MIN_EVIDENCE = 2
HIGH_CONFIDENCE = 0.90

# 事实键 → 画像字段的**白名单**映射。没列在这里的事实（如 profile:family）只进 user_facts，
# 不进画像字段——画像要保持"能直接支撑决策"的信噪比。
FACT_TO_PROFILE_FIELD: dict[str, str] = {
    "preference:asset_class": "preferred_asset_class",
    "preference:horizon": "investment_horizon",
}

# 自述类事实（客户自己说的偏好）统一进 risk_tags，并标注来源为"自述"。
# 保留它们的价值在于：当出现「问卷 C4 / 自述稳健 / 行为买 R4」三方不一致时，
# 这种矛盾本身就是风控信号——但绝不能与问卷等级混进同一个字段。
SELF_REPORTED_PREFIXES = ("preference:risk_level", "preference:", "profile:")

# 关键事实：参与决策，标记出来便于下游优先读取
CRITICAL_FACTS = frozenset({
    "preference:risk_level", "preference:horizon", "preference:asset_class",
})

# 画像中允许本服务写入的字段（其余字段归交易/注册侧所有）
PROFILE_OWNED_FIELDS = ("investor_type", "preferred_asset_class", "investment_horizon", "risk_tags")


def _now() -> datetime:
    return datetime.now(UTC).replace(tzinfo=None)


def _fact_id() -> int:
    """`user_facts.id` 没有 auto_increment，主键由应用生成。

    用微秒时间戳：单调递增、无需额外序列、同客户同微秒重复在单进程写入下不可能发生。
    """
    return int(datetime.now(UTC).timestamp() * 1_000_000)


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

    # ---------- 中期 → 长期 ----------

    async def promote_facts(self, customer_id: int) -> list[str]:
        """把证据足够的记忆提炼为长期事实；返回本次提升的事实键。"""
        now = _now()
        rows = list(await self.session.scalars(
            select(MemoryUnit).where(
                MemoryUnit.customer_id == customer_id,
                MemoryUnit.status == "active",
                or_(
                    MemoryUnit.evidence_count >= MIN_EVIDENCE,
                    MemoryUnit.confidence >= HIGH_CONFIDENCE,
                ),
                or_(MemoryUnit.valid_until.is_(None), MemoryUnit.valid_until > now),
            )
        ))
        promoted: list[str] = []
        for memory in rows:
            key = str(memory.memory_key)
            value = self._fact_value(memory)
            existing = await self.session.scalar(
                select(UserFact).where(
                    UserFact.customer_id == customer_id, UserFact.fact_key == key
                )
            )
            if existing is None:
                self.session.add(UserFact(
                    # 主键显式赋值：该表无 auto_increment
                    id=_fact_id(),
                    customer_id=customer_id,
                    fact_key=key,
                    fact_value=value,
                    source_portal=str(memory.source_type or "conversation"),
                    source_episode_id=None,
                    confidence=float(memory.confidence or 0.0),
                    is_critical=key in CRITICAL_FACTS,
                    created_at=now,
                ))
            else:
                existing.fact_value = value
                existing.confidence = float(memory.confidence or 0.0)
                existing.is_critical = key in CRITICAL_FACTS
            promoted.append(key)
        await self.session.flush()
        return promoted

    @staticmethod
    def _fact_value(memory: MemoryUnit) -> Any:
        """事实值优先取结构化值，回退到正文；始终以 JSON 可存的形式返回。"""
        structured = memory.structured_value
        if isinstance(structured, dict) and "value" in structured:
            return structured["value"]
        if structured is not None:
            return structured
        return memory.content or ""

    # ---------- 长期 → 画像 ----------

    async def rebuild_profile(self, customer_id: int) -> dict[str, Any]:
        """用长期事实 + 问卷重建画像，并写一条版本快照。"""
        now = _now()
        facts = list(await self.session.scalars(
            select(UserFact).where(UserFact.customer_id == customer_id)
        ))
        assessment = (await self.session.execute(text(
            """
            SELECT investor_type, questionnaire_version, assessed_at, valid_until
            FROM fin_risk_assessment
            WHERE customer_id = :customer_id
            ORDER BY assessed_at DESC, id DESC
            LIMIT 1
            """
        ), {"customer_id": customer_id})).first()

        values: dict[str, Any] = {}
        basis: dict[str, Any] = {}

        # 红线：风险等级只从问卷取；记忆里哪怕有 preference:risk_level 也不写这个字段
        if assessment is not None and assessment[0]:
            values["investor_type"] = str(assessment[0])
            basis["investor_type"] = {
                "source": "fin_risk_assessment",
                "questionnaire_version": assessment[1],
                "assessed_at": str(assessment[2]),
                "valid_until": str(assessment[3]),
            }

        tags: list[str] = []
        for fact in facts:
            key = str(fact.fact_key)
            field = FACT_TO_PROFILE_FIELD.get(key)
            if field is not None:
                values[field] = self._as_text(fact.fact_value)
                basis[field] = {
                    "source": "user_facts", "fact_key": key,
                    "confidence": float(fact.confidence or 0.0),
                }
            elif key.startswith(SELF_REPORTED_PREFIXES):
                # 自述信息进标签，并显式标注"自述"，与问卷等级区分开
                tags.append(f"自述:{key}={self._as_text(fact.fact_value)}")
                basis.setdefault("risk_tags", {"source": "user_facts", "items": []})
                basis["risk_tags"]["items"].append(key)
        if tags:
            values["risk_tags"] = "；".join(tags)

        profile = await self.session.get(FundCustomerProfile, customer_id)
        if profile is None:
            # 画像行由开户流程创建：`trade_account` 等身份字段在库里是 NOT NULL，属注册／账户侧
            # 所有。本服务不代替开户去造这些数据——否则会写出一条**假的**开户记录，
            # 而画像恰恰是风控要读的东西，假数据比没有数据更危险。未开户时如实报告。
            return {
                "profile": None,
                "reason": "profile_row_not_opened",
                "generation_basis": basis,
                "promoted": len(facts),
            }
        for field in PROFILE_OWNED_FIELDS:
            if field == "investor_type":
                # 问卷是唯一权威：本轮没有问卷记录时**保持原值**（既不写入也不清空）。
                # 否则重测前的空档会把开户时的等级抹掉，而该列是 NOT NULL。
                if field in values:
                    setattr(profile, field, values[field])
                continue
            # 其余字段由本服务独占：本轮没有对应事实即清空。
            # 这不只是洁癖——记忆失效后若不清理，画像会留着一个已经作废的投资期限，
            # 投顾据此给建议，而客户从未授权这条信息继续生效（实测踩到过）。
            setattr(profile, field, values.get(field))
        profile.updated_at = now
        await self.session.flush()

        snapshot = {
            **{field: getattr(profile, field, None) for field in PROFILE_OWNED_FIELDS},
            "generated_at": now.isoformat(),
        }
        await self._write_snapshot(customer_id, snapshot, basis, now)
        return {"profile": snapshot, "generation_basis": basis, "promoted": len(facts)}

    @staticmethod
    def _as_text(value: Any) -> str:
        """把 JSON 列里取出的值渲染成可读字符串。

        字符串类型的值可能带着 JSON 序列化时的外层引号（取决于驱动如何回读 JSON 列），
        这里去掉它们——`risk_tags` 是给风控与投顾看的，多一对引号会让人以为值本身包含引号。
        """
        if isinstance(value, str):
            return value.strip().strip('"')
        return _json.dumps(value, ensure_ascii=False)

    async def _write_snapshot(
        self, customer_id: int, snapshot: dict[str, Any], basis: dict[str, Any], now: datetime
    ) -> None:
        """写入新版本快照并把旧版本置为非当前。

        唯一键 `uk_profile_snapshot_current` 建立在生成列 `current_customer_id` 上，
        保证「每个客户最多一条 current」；因此必须先清旧再写新，顺序不能反。
        """
        previous = list(await self.session.scalars(
            select(ProfileSnapshot).where(
                ProfileSnapshot.customer_id == customer_id, ProfileSnapshot.is_current.is_(True)
            )
        ))
        for row in previous:
            row.is_current = False
            row.updated_at = now
        await self.session.flush()

        latest = await self.session.scalar(text(
            "SELECT COALESCE(MAX(version), 0) FROM profile_snapshots WHERE customer_id = :cid"
        ), {"cid": customer_id})
        version = int(latest or 0) + 1
        payload = _json.dumps(snapshot, ensure_ascii=False, sort_keys=True)
        self.session.add(ProfileSnapshot(
            profile_uuid=str(uuid4()),
            customer_id=customer_id,
            version=version,
            snapshot=snapshot,
            generation_basis=basis,
            snapshot_hash=sha256(payload.encode("utf-8")).hexdigest(),
            is_current=True,
            generated_at=now,
            created_at=now,
            updated_at=now,
        ))
        await self.session.flush()

    # ---------- 完整链路 ----------

    async def rebuild(self, customer_id: int) -> dict[str, Any]:
        promoted = await self.promote_facts(customer_id)
        outcome = await self.rebuild_profile(customer_id)
        outcome["promoted_keys"] = promoted
        return outcome