group_fqcd_jr/app/service/customer_benefit_service.py

"""客户权益：按可投资资产判定层级，并展开该层级（含以下各层）应享有的权益。

### 口径来源

`knowledge/product/高净值客户服务规范.md`（公司内部服务标准，四级分层 + 各层权益）：

| 层级 | 名称 | 可投资资产门槛 |
|---|---|---|
| `gold` | 金卡 | 50 万 - 200 万 |
| `platinum` | 白金 | 200 万 - 600 万 |
| `diamond` | 钻石 | 600 万 - 1000 万 |
| `private` | 私行 | 1000 万以上 |

低于 50 万为**普通客户**（无层级）。文档写的是「以客户可投资金融资产（不含自住房产）
为主要分层依据」，本实现取 `fin_customer_profile.total_asset` —— 基线
（`docs/00` L213/L220）把它定为「风控研判所用资产快照」且「按统一口径计算」，
是系统里唯一可用的资产口径；**不另造口径**。

### 两条设计约束

1. **权益按层级累积**（文档原文"含全部金卡权益，新增以下"）：
   白金含金卡全部条目、依此类推。展开在**服务层**做，表里不重复存父级条目
   —— 否则同一权益要改多处。
2. **不把"某客户享有哪些权益"落库**：它可由层级实时推出。
   基线的同一取向见 `docs/00` L159（不保留 `net_worth_flag`，因为可算）。
"""

from dataclasses import dataclass
from decimal import Decimal
from typing import Any

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.contracts import RequestContext
from app.model.benefit import CustomerBenefit
from app.model.fund import FundCustomerProfile
from app.service.authorization_service import AuthorizationService

#: 本模块使用的权限码；定义源是 `tools/seed_test_rbac.py` 的 `PERMISSIONS`。
PERMISSION_READ_SELF = "benefit:read:self"


@dataclass(frozen=True)
class TierSpec:
    """一个层级的码、中文名与门槛（含）。"""

    code: str
    label: str
    min_total_asset: Decimal


#: **从高到低**排列：判定时取第一个满足门槛的层级。
#: 顺序即累积顺序，`_tier_chain` 依赖它，不要随意重排。
TIERS: tuple[TierSpec, ...] = (
    TierSpec("private", "私行", Decimal("10000000")),
    TierSpec("diamond", "钻石", Decimal("6000000")),
    TierSpec("platinum", "白金", Decimal("2000000")),
    TierSpec("gold", "金卡", Decimal("500000")),
)

HEADLINE = "普通客户"


def _spec(code: str) -> TierSpec | None:
    return next((t for t in TIERS if t.code == code), None)


def resolve_tier(total_asset: Decimal | int | float | None) -> TierSpec | None:
    """按可投资资产判定层级；低于最低门槛（50 万）或资产缺失时返回 None。"""
    if total_asset is None:
        return None
    amount = Decimal(str(total_asset))
    return next((t for t in TIERS if amount >= t.min_total_asset), None)


def tier_chain(spec: TierSpec) -> tuple[str, ...]:
    """该层级**及以下**所有层级码（用于累积展开）。

    例：钻石 → `("gold", "platinum", "diamond")`。
    `TIERS` 是从高到低，故取其尾部到该层级为止。
    """
    codes_low_to_high = [t.code for t in reversed(TIERS)]
    return tuple(codes_low_to_high[: codes_low_to_high.index(spec.code) + 1])


class CustomerBenefitService:
    """权益查询；**只读**，不写任何表（含不写 `sys_user.customer_tier`）。"""

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

    async def entitlements_for(self, context: RequestContext) -> dict[str, Any]:
        """端点入口：先鉴权，再按客户资产判定层级并展开权益。

        ⚠️ 权限检查放在 **Service 层**（项目惯例：26 个 service 都这么做，
        见 `AuthorizationService.require`）—— `trading.py` 的 §T 端点目前**没有**
        这一步，只有测评门槛，属既有缺口，不在本次改动范围内。
        """
        await AuthorizationService.require(context, PERMISSION_READ_SELF)
        total_asset = await self._total_asset(int(context.user_id))
        return await self.entitlements(total_asset=total_asset)

    async def _total_asset(self, customer_id: int) -> Decimal | None:
        """读客户画像的资产快照。

        基线把 `fin_customer_profile.total_asset` 定为「风控研判所用资产快照」，
        是系统里唯一的资产口径；客户未开户（无画像行）时返回 None，
        由 `resolve_tier` 判为无层级，而不是抛错 —— "还没开户"不是异常。
        """
        row = await self.session.scalar(
            select(FundCustomerProfile.total_asset).where(
                FundCustomerProfile.customer_id == customer_id
            )
        )
        return row

    async def entitlements(self, *, total_asset: Decimal | None) -> dict[str, Any]:
        """返回客户当前层级与应享权益。

        `total_asset` 由调用方从 `fin_customer_profile` 读入并传入，
        避免本服务再查一次客户主表（也在测试里便于给定资产直接断言分层）。
        """
        spec = resolve_tier(total_asset)
        if spec is None:
            return {
                "tier": None,
                "tier_label": HEADLINE,
                "min_total_asset": None,
                "total_asset": self._amount(total_asset),
                "next_tier": self._next_tier(None),
                "benefits": [],
            }

        codes = tier_chain(spec)
        rows = list(await self.session.scalars(
            select(CustomerBenefit)
            .where(
                CustomerBenefit.customer_tier.in_(codes),
                CustomerBenefit.status == "active",
            )
            .order_by(CustomerBenefit.display_order, CustomerBenefit.id)
        ))
        return {
            "tier": spec.code,
            "tier_label": spec.label,
            "min_total_asset": self._amount(spec.min_total_asset),
            "total_asset": self._amount(total_asset),
            "next_tier": self._next_tier(spec),
            "benefits": [self._view(row) for row in rows],
        }

    @staticmethod
    def _next_tier(spec: TierSpec | None) -> dict[str, Any] | None:
        """下一层级与还差多少 —— 前端可直接渲染"再投 X 元升级"。

        已是最高的私行返回 None。
        """
        if spec is None:
            target = TIERS[-1]  # 金卡
        else:
            higher = [t for t in TIERS if t.min_total_asset > spec.min_total_asset]
            if not higher:
                return None
            target = min(higher, key=lambda t: t.min_total_asset)
        return {
            "tier": target.code,
            "tier_label": target.label,
            "min_total_asset": CustomerBenefitService._amount(target.min_total_asset),
        }

    @staticmethod
    def _amount(value: Decimal | int | float | None) -> str | None:
        """金额统一用字符串输出（与项目其他接口一致，避免浮点误差）。"""
        if value is None:
            return None
        return str(Decimal(str(value)).quantize(Decimal("0.01")))

    @staticmethod
    def _view(row: CustomerBenefit) -> dict[str, Any]:
        spec = _spec(row.customer_tier)
        return {
            "benefit_code": row.benefit_code,
            # 权益所属层级（用于前端按层级分组；累积展开后可能低于客户自身层级）
            "tier": row.customer_tier,
            "tier_label": spec.label if spec else None,
            "category": row.category,
            "name": row.name,
            "description": row.description,
        }