feat(投顾): 客户主动申报投顾方案 + 投顾受理自动生成方案草稿
客户在自己主页提交申报 → 投顾工作台受理 → 自动跑既有推荐逻辑生成一份待审核草稿 → 投顾再走既有的「审核通过 → 发送给客户」。补上原先「客户只能被动等方案」的缺口。 后端 - 新增表 advisor_service_request（本轮新建，带 AUTO_INCREMENT）+ 迁移 20260916_advisor_service_request（幂等：先查表再建，兼容本库 alembic 指针滞后） - 新增 AdvisorServiceRequestService：create / list_mine / queue / review · 申报前置：风险测评必须存在且未失效（FM-03，12 个月），服务端判 · 队列复用 ProductRecommendationService._visible_customer_ids（本人 + 归属），待受理排前 · 受理即调用推荐逻辑生成草稿并把 content_id 回填；生成前置失败给出人话原因并保持待受理 · 「已发送客户」不落库，由关联方案 published_at 推导，避免两处状态各写各的 - 权限码 9071-9074（客户 write/read:self、投顾 read/review），已进种子与授权工具 前端 - 投顾工作台新增「客户申报」面板（受理 / 驳回，驳回理由客户可见） - 客户页新增申报表单（金额/期限/风险偏好/备注）与「我的申报」列表 客户自助路由刻意不挂投顾灰度闸门：那是投顾业务的灰度，客户提交自己的申请不该被它拦下。
This commit is contained in:
Windows
@@ -0,0 +1,303 @@
|
||||
"""客户主动申报投顾方案 → 投顾受理审批。
|
||||
|
||||
## 流程
|
||||
|
||||
客户「我的投顾方案」页提交申报（金额/期限/风险偏好/备注）
|
||||
→ 落 `advisor_service_request`（status=pending）
|
||||
→ 投顾工作台「客户申报」看到待办
|
||||
→ 投顾受理：**自动跑一次推荐**，生成一份 pending_review 的方案草稿，
|
||||
并把草稿 id 记进 `result_content_id`（status=accepted）
|
||||
→ 投顾随后走既有的「审核通过 → 发送给客户」
|
||||
→ 客户在「我的投顾方案」看到方案（status 呈现为 delivered）
|
||||
|
||||
## 两条前置/边界
|
||||
|
||||
1. **申报前置**：必须已完成风险测评且未失效（FM-03：12 个月）。服务端判，
|
||||
不依赖前端闸门 —— 前端只是体验层。
|
||||
2. **归属**：投顾只能处理**名下客户**的申报，复用
|
||||
`ProductRecommendationService._visible_customer_ids`（本人 + `sys_customer_assignment`），
|
||||
与已发布方案/历史记录同一把尺子。
|
||||
|
||||
## 状态不重复存
|
||||
|
||||
「已发送给客户」不单独存状态，而是由 `result_content_id` 指向的方案是否已发布
|
||||
**推导**（见 `_view`）—— 两处各存一份必然对不上。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import uuid
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from decimal import Decimal
|
||||
from typing import Any
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, Field
|
||||
from sqlalchemy import select
|
||||
|
||||
from app.core.contracts import RequestContext
|
||||
from app.core.errors import (
|
||||
ForbiddenAgentError,
|
||||
GenericResourceNotFoundError,
|
||||
InvalidStateError,
|
||||
ValidationAgentError,
|
||||
)
|
||||
from app.core.product_recommendation_contracts import ProductRecommendationQuery
|
||||
from app.infrastructure.db import SessionFactory
|
||||
from app.model.advisor_service_request import (
|
||||
ACCEPTED,
|
||||
DELIVERED,
|
||||
PENDING,
|
||||
REJECTED,
|
||||
AdvisorServiceRequest,
|
||||
)
|
||||
from app.model.investment_goal import ClientFacingContent
|
||||
from app.service.api_transaction_service import ApiTransactionService
|
||||
from app.service.authorization_service import AuthorizationService
|
||||
from app.service.product_recommendation_service import ProductRecommendationService
|
||||
from app.service.suitability_service import SuitabilityService
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
#: 与投顾工作台左栏表单的选项一致；服务端仍要校验一次（前端只是体验层）。
|
||||
HORIZONS: tuple[str, ...] = ("<1年", "1-3年", "3-5年", ">5年")
|
||||
RISK_PREFERENCES: tuple[str, ...] = ("稳健", "平衡", "进取")
|
||||
|
||||
#: FM-03：风险测评 12 个月失效。
|
||||
ASSESSMENT_VALID_DAYS = 365
|
||||
|
||||
|
||||
def _naive_utc(value: datetime) -> datetime:
|
||||
"""归一成 **UTC naive** 再比较。
|
||||
|
||||
适配器返回的时间可能带 tz（`valid_until`），而 `now` 是 naive ——
|
||||
直接比较会 `TypeError: can't compare offset-naive and offset-aware datetimes`。
|
||||
"""
|
||||
if value.tzinfo is None:
|
||||
return value
|
||||
return value.astimezone(UTC).replace(tzinfo=None)
|
||||
|
||||
#: 受理时推荐逻辑的前置失败 → 给投顾看得懂的话（`generate` 的 early return 状态码）。
|
||||
GENERATE_FAILURES: dict[str, str] = {
|
||||
"profile_required": "客户尚未完成风险测评，无法生成方案",
|
||||
"investment_goal_required": "客户暂无已确认投资目标，请先录入并确认目标",
|
||||
"recommendation_input_invalid": "客户投资目标数据不完整，无法生成方案",
|
||||
}
|
||||
|
||||
|
||||
class AdvisorServiceRequestCreate(BaseModel):
|
||||
model_config = ConfigDict(extra="forbid", frozen=True)
|
||||
|
||||
amount_wan: float = Field(gt=0, le=1_000_000, description="拟投资金额（万元）")
|
||||
horizon: str = Field(description="投资期限")
|
||||
risk_preference: str = Field(description="风险偏好")
|
||||
note: str | None = Field(default=None, max_length=500, description="补充说明")
|
||||
|
||||
|
||||
class AdvisorServiceRequestService:
|
||||
# ---- 客户侧 ----
|
||||
|
||||
async def create(
|
||||
self, payload: AdvisorServiceRequestCreate, context: RequestContext, key: str | None
|
||||
) -> dict[str, object]:
|
||||
await AuthorizationService.require(context, "advisor-request:write:self")
|
||||
if payload.horizon not in HORIZONS:
|
||||
raise ValidationAgentError(f"投资期限只能是 {list(HORIZONS)} 之一")
|
||||
if payload.risk_preference not in RISK_PREFERENCES:
|
||||
raise ValidationAgentError(f"风险偏好只能是 {list(RISK_PREFERENCES)} 之一")
|
||||
customer_id = int(context.user_id)
|
||||
await self._require_valid_assessment(customer_id)
|
||||
|
||||
async def operation(session: Any) -> dict[str, object]:
|
||||
now = datetime.now(UTC).replace(tzinfo=None)
|
||||
# 同一客户同一秒可能连点两次：后缀加短随机码，避免撞 `request_no` 唯一键。
|
||||
request_no = f"ASR-{customer_id}-{now:%y%m%d%H%M%S}-{uuid.uuid4().hex[:4].upper()}"
|
||||
row = AdvisorServiceRequest(
|
||||
request_no=request_no,
|
||||
customer_id=customer_id,
|
||||
amount_wan=Decimal(str(payload.amount_wan)),
|
||||
horizon=payload.horizon,
|
||||
risk_preference=payload.risk_preference,
|
||||
note=payload.note,
|
||||
status=PENDING,
|
||||
created_at=now,
|
||||
updated_at=now,
|
||||
)
|
||||
session.add(row)
|
||||
await session.flush()
|
||||
return {
|
||||
"data": {"request_no": request_no, "status": PENDING},
|
||||
"meta": {"trace_id": context.trace_id},
|
||||
}
|
||||
|
||||
return await ApiTransactionService().execute(
|
||||
context,
|
||||
f"advisor-request:create:{customer_id}",
|
||||
key,
|
||||
payload.model_dump(mode="json"),
|
||||
operation,
|
||||
)
|
||||
|
||||
async def list_mine(self, context: RequestContext) -> dict[str, object]:
|
||||
await AuthorizationService.require(context, "advisor-request:read:self")
|
||||
rows = await self._fetch(customer_ids=[int(context.user_id)])
|
||||
rows.sort(key=lambda row: row.created_at, reverse=True)
|
||||
return {"data": await self._views(rows), "meta": {"trace_id": context.trace_id}}
|
||||
|
||||
# ---- 投顾侧 ----
|
||||
|
||||
async def queue(self, context: RequestContext) -> dict[str, object]:
|
||||
await AuthorizationService.require(context, "advisor-request:read")
|
||||
customer_ids = list(ProductRecommendationService._visible_customer_ids(context))
|
||||
rows = await self._fetch(customer_ids=customer_ids)
|
||||
# 待受理排前面，其余按时间倒序 —— 投顾打开面板先看到"要动手的"。
|
||||
rows.sort(key=lambda row: (row.status != PENDING, -row.created_at.timestamp()))
|
||||
return {"data": await self._views(rows), "meta": {"trace_id": context.trace_id}}
|
||||
|
||||
async def review(
|
||||
self,
|
||||
request_no: str,
|
||||
decision: str,
|
||||
comment: str,
|
||||
context: RequestContext,
|
||||
key: str | None,
|
||||
) -> dict[str, object]:
|
||||
await AuthorizationService.require(context, "advisor-request:review")
|
||||
if decision not in {"accepted", "rejected"}:
|
||||
raise ValidationAgentError("decision 必须为 accepted 或 rejected")
|
||||
|
||||
async with SessionFactory() as session:
|
||||
existing = await session.scalar(
|
||||
select(AdvisorServiceRequest).where(
|
||||
AdvisorServiceRequest.request_no == request_no
|
||||
)
|
||||
)
|
||||
if existing is None:
|
||||
raise GenericResourceNotFoundError("申报单不存在")
|
||||
visible = set(ProductRecommendationService._visible_customer_ids(context))
|
||||
if existing.customer_id not in visible:
|
||||
raise ForbiddenAgentError("无权处理该客户的申报")
|
||||
if existing.status != PENDING:
|
||||
raise InvalidStateError("该申报单已被处理")
|
||||
|
||||
content_id: int | None = None
|
||||
if decision == "accepted":
|
||||
content_id = await self._generate_draft_plan(request_no, existing.customer_id, context)
|
||||
|
||||
async def operation(session: Any) -> dict[str, object]:
|
||||
row = await session.get(AdvisorServiceRequest, existing.id, with_for_update=True)
|
||||
if row is None or row.status != PENDING:
|
||||
raise InvalidStateError("该申报单已被处理")
|
||||
now = datetime.now(UTC).replace(tzinfo=None)
|
||||
row.status = ACCEPTED if decision == "accepted" else REJECTED
|
||||
row.handled_by = int(context.user_id)
|
||||
row.handled_at = now
|
||||
row.advisor_note = comment or None
|
||||
row.result_content_id = content_id
|
||||
row.updated_at = now
|
||||
await session.flush()
|
||||
return {
|
||||
"data": {
|
||||
"request_no": row.request_no,
|
||||
"status": row.status,
|
||||
"result_content_id": str(content_id) if content_id else None,
|
||||
},
|
||||
"meta": {"trace_id": context.trace_id},
|
||||
}
|
||||
|
||||
return await ApiTransactionService().execute(
|
||||
context,
|
||||
f"advisor-request:{request_no}:review",
|
||||
key,
|
||||
{"decision": decision, "comment": comment},
|
||||
operation,
|
||||
)
|
||||
|
||||
# ---- 内部 ----
|
||||
|
||||
async def _require_valid_assessment(self, customer_id: int) -> None:
|
||||
"""申报前置：必须已完成风险测评且未失效（FM-03：12 个月）。"""
|
||||
profile = await SuitabilityService().authority_for_customer(customer_id)
|
||||
if profile.customer_risk_level is None:
|
||||
raise InvalidStateError("请先完成风险测评，再申报投顾方案")
|
||||
now = datetime.now(UTC).replace(tzinfo=None)
|
||||
if profile.valid_until is not None:
|
||||
if _naive_utc(profile.valid_until) < now:
|
||||
raise InvalidStateError("风险测评已过期，请重新测评后再申报")
|
||||
return
|
||||
# 没写有效期时按 FM-03 的 12 个月口径兜底 —— 与前端熔断规则同一口径。
|
||||
if (
|
||||
profile.assessed_at is not None
|
||||
and now - _naive_utc(profile.assessed_at) > timedelta(days=ASSESSMENT_VALID_DAYS)
|
||||
):
|
||||
raise InvalidStateError("风险测评已超过 12 个月，请重新测评后再申报")
|
||||
|
||||
async def _generate_draft_plan(
|
||||
self, request_no: str, customer_id: int, context: RequestContext
|
||||
) -> int:
|
||||
"""受理时自动出草稿：沿用既有推荐逻辑（硬约束 + 适当性 + 排序 + LLM 依据）。"""
|
||||
result = await ProductRecommendationService(
|
||||
enforce_profile_governance=True
|
||||
).generate(
|
||||
ProductRecommendationQuery(customer_id=customer_id, limit=3),
|
||||
context,
|
||||
f"advisor-request-{request_no}-generate",
|
||||
)
|
||||
status = result.get("status")
|
||||
if isinstance(status, str) and status in GENERATE_FAILURES:
|
||||
raise InvalidStateError(GENERATE_FAILURES[status])
|
||||
payload = result.get("data")
|
||||
content_id = str((payload or {}).get("content_id") or "") if isinstance(payload, dict) else ""
|
||||
if not content_id.isdigit():
|
||||
raise InvalidStateError("方案生成未返回方案编号，请稍后重试")
|
||||
return int(content_id)
|
||||
|
||||
@staticmethod
|
||||
async def _fetch(customer_ids: list[int]) -> list[AdvisorServiceRequest]:
|
||||
if not customer_ids:
|
||||
return []
|
||||
async with SessionFactory() as session:
|
||||
return list(
|
||||
await session.scalars(
|
||||
select(AdvisorServiceRequest)
|
||||
.where(AdvisorServiceRequest.customer_id.in_(customer_ids))
|
||||
.order_by(AdvisorServiceRequest.created_at.desc())
|
||||
.limit(50)
|
||||
)
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
async def _views(rows: list[AdvisorServiceRequest]) -> list[dict[str, object]]:
|
||||
"""补上「方案是否已发布」——`delivered` 是**推导**出来的，不单独存状态。"""
|
||||
content_ids = [row.result_content_id for row in rows if row.result_content_id]
|
||||
published: dict[int, bool] = {}
|
||||
if content_ids:
|
||||
async with SessionFactory() as session:
|
||||
found = (
|
||||
await session.execute(
|
||||
select(ClientFacingContent.id, ClientFacingContent.published_at).where(
|
||||
ClientFacingContent.id.in_(content_ids)
|
||||
)
|
||||
)
|
||||
).all()
|
||||
published = {int(cid): published_at is not None for cid, published_at in found}
|
||||
views: list[dict[str, object]] = []
|
||||
for row in rows:
|
||||
delivered = bool(
|
||||
row.result_content_id and published.get(int(row.result_content_id))
|
||||
)
|
||||
status = DELIVERED if delivered else row.status
|
||||
views.append({
|
||||
"request_no": row.request_no,
|
||||
"customer_id": str(row.customer_id),
|
||||
"amount_wan": float(row.amount_wan),
|
||||
"horizon": row.horizon,
|
||||
"risk_preference": row.risk_preference,
|
||||
"note": row.note,
|
||||
"status": status,
|
||||
"advisor_note": row.advisor_note,
|
||||
"handled_at": row.handled_at.isoformat() if row.handled_at else None,
|
||||
"result_content_id": str(row.result_content_id) if row.result_content_id else None,
|
||||
"created_at": row.created_at.isoformat() if row.created_at else None,
|
||||
})
|
||||
return views
|
||||
Reference in New Issue
Block a user