453 lines
18 KiB
Python
453 lines
18 KiB
Python
"""客户评级业务编排（等级映射配置 + 每客户评级落库）。
|
||
|
||
评级定位（对照表 #25）：**不影响绩效金额**。自 2026-10-19 起评分每客户
|
||
落库（customer_ratings，一行一客户快照），等级映射仍是本服务承载的
|
||
**判定表**，判定规则为按 sort_order 升序取首个 min_score ≤ 总分的等级，
|
||
max_score 仅展示。两条硬约束（违反会算错等级）：
|
||
|
||
- 分值全程 ``Decimal``：入参出参 str、schema 拦格式，service 层负责字符串
|
||
到 Decimal 的转换与落库；值域校验（等级区间 0≤min≤max≤100、单项评分
|
||
0≤分≤20）在**审计上下文内**完成，失败的 400 同样留痕（2026-10-13 定稿
|
||
口径）
|
||
- 区间互不重叠：重叠会让同一总分命中多个等级（或两段都跳过），判定表
|
||
失效。判定为**闭区间两两相交**：任一其他等级满足
|
||
``other.min_score <= 新 max 且 新 min <= other.max_score`` 即重叠。
|
||
边界相邻不算重叠（前段 max=89、后段 min=90）—— 种子 A/B/C/D 即按
|
||
相邻铺满 0-100，且判定本身是单边 min_score ≤ 总分，边界归 sort_order
|
||
靠前的等级，相邻无歧义
|
||
|
||
删除允许清空：等级表对客户评级**无外键引用**（评级快照固化等级名、不
|
||
回指本表），物理删除无悬空风险，也不伤及已评级客户的历史结论；全部删空
|
||
后新评级只记总分、等级为 NULL（前端已对应提示），故不强制保留任何一条。
|
||
"""
|
||
|
||
import logging
|
||
from decimal import Decimal
|
||
|
||
from sqlalchemy.exc import IntegrityError
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.constants import (
|
||
ACTION_CREATE,
|
||
ACTION_DELETE,
|
||
ACTION_UPDATE,
|
||
MODULE_RATING,
|
||
)
|
||
from app.core.db_errors import is_duplicate_entry
|
||
from app.core.exceptions import BadRequestError, NotFoundError
|
||
from app.models import CustomerRating, CustomerRatingGrade, User
|
||
from app.repositories import customer_rating_repository, customer_repository, rating_repository
|
||
from app.schemas.rating import (
|
||
CustomerRatingOut,
|
||
CustomerRatingSaveIn,
|
||
RatingGradeCreateIn,
|
||
RatingGradeOut,
|
||
RatingGradeUpdateIn,
|
||
)
|
||
from app.services import audit_service
|
||
from app.services.audit_service import AuditRecord, RequestMeta
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 分值域上下界（Decimal 常量，禁用 float 参与比较）
|
||
_SCORE_FLOOR = Decimal("0")
|
||
_SCORE_CEIL = Decimal("100")
|
||
# 单项评分上界（五项各 0-20，总分 0-100）
|
||
_ITEM_SCORE_CEIL = Decimal("20")
|
||
|
||
|
||
def _to_out(grade: CustomerRatingGrade) -> RatingGradeOut:
|
||
"""ORM 对象转响应模型。分值 Decimal → str（分值红线出参口径）。"""
|
||
return RatingGradeOut(
|
||
id=grade.id,
|
||
grade_name=grade.grade_name,
|
||
min_score=str(grade.min_score),
|
||
max_score=str(grade.max_score),
|
||
sort_order=grade.sort_order,
|
||
remark=grade.remark,
|
||
create_time=grade.create_time,
|
||
update_time=grade.update_time,
|
||
)
|
||
|
||
|
||
def _to_decimal(value: str) -> Decimal:
|
||
"""数字字符串转 Decimal。
|
||
|
||
入参已 str 化再进 Decimal：即使上游口径变化传入数字类型，也不经 float
|
||
二进制近似，分值红线在最后一道转换处仍然成立。schema 已保证可解析，
|
||
此处不再重复校验。
|
||
"""
|
||
return Decimal(str(value))
|
||
|
||
|
||
def _ensure_score_range(min_score: Decimal, max_score: Decimal) -> None:
|
||
"""区间合法性：0 ≤ min_score ≤ max_score ≤ 100。
|
||
|
||
超界/倒挂任一不满足即「分值区间不合法」——统一一条消息不区分具体
|
||
越界方式，前端据此整体提示。须在审计上下文内调用，失败也要留痕。
|
||
"""
|
||
if not (_SCORE_FLOOR <= min_score <= max_score <= _SCORE_CEIL):
|
||
raise BadRequestError("分值区间不合法")
|
||
|
||
|
||
def _ensure_no_overlap(
|
||
grades: list[CustomerRatingGrade],
|
||
new_min: Decimal,
|
||
new_max: Decimal,
|
||
*,
|
||
exclude_id: int | None = None,
|
||
) -> None:
|
||
"""新区间与既有等级**不得重叠**（闭区间两两相交判定）。
|
||
|
||
任一其他等级满足 ``other.min_score <= new_max 且 new_min <= other.max_score``
|
||
即重叠，400 并点名冲突等级，便于前端定位到具体行。边界相邻不算重叠
|
||
（见模块 docstring）。exclude_id 供编辑时排除自身。
|
||
|
||
:raises BadRequestError: 分值区间与等级 X 重叠
|
||
"""
|
||
for other in grades:
|
||
if exclude_id is not None and other.id == exclude_id:
|
||
continue
|
||
if other.min_score <= new_max and new_min <= other.max_score:
|
||
raise BadRequestError(f"分值区间与等级 {other.grade_name} 重叠")
|
||
|
||
|
||
def _is_duplicate_entry(exc: IntegrityError) -> bool:
|
||
"""判断 IntegrityError 是否为 MySQL 1062（唯一键冲突）。
|
||
|
||
只有 1062 是「已存在」语义，可转业务提示；其余完整性错误语义未知，
|
||
原样上抛 —— 由全局 IntegrityError handler 收敛为 400（通用文案），
|
||
500 只留给程序缺陷。
|
||
"""
|
||
return is_duplicate_entry(exc)
|
||
|
||
|
||
def _raise_if_duplicate(exc: IntegrityError) -> None:
|
||
"""唯一键冲突统一转业务提示；其余完整性错误原样重抛。"""
|
||
if _is_duplicate_entry(exc):
|
||
raise BadRequestError("该等级名称已存在") from exc
|
||
raise exc
|
||
|
||
|
||
def list_grades(session: Session) -> list[RatingGradeOut]:
|
||
"""全量等级列表，按判定顺序（sort_order 升序）直出。
|
||
|
||
不分页：等级映射是判定表，数量个位数，分页无意义。
|
||
"""
|
||
return [_to_out(grade) for grade in rating_repository.list_all(session)]
|
||
|
||
|
||
def create_grade(
|
||
session: Session,
|
||
payload: RatingGradeCreateIn,
|
||
*,
|
||
operator: User,
|
||
meta: RequestMeta | None = None,
|
||
) -> RatingGradeOut:
|
||
"""新增评级等级。
|
||
|
||
区间须 0≤min≤max≤100 且与既有等级不重叠（新增同样判重叠：判定表
|
||
语义下重叠即脏配置，没有「先停用旧区间」之类的两阶段配置手段）。
|
||
|
||
:raises BadRequestError: 分值区间不合法 / 分值区间与等级 X 重叠 / 该等级名称已存在
|
||
"""
|
||
with audit_service.audit(
|
||
session,
|
||
operator=operator,
|
||
record=AuditRecord(
|
||
module=MODULE_RATING,
|
||
action=ACTION_CREATE,
|
||
summary=f"新增评级等级 {payload.grade_name}",
|
||
request_data=payload.model_dump(exclude_none=True),
|
||
),
|
||
meta=meta,
|
||
):
|
||
# 值域校验在审计上下文内：分值非法的 400 也要留痕（成功与失败都记）
|
||
new_min = _to_decimal(payload.min_score)
|
||
new_max = _to_decimal(payload.max_score)
|
||
_ensure_score_range(new_min, new_max)
|
||
_ensure_no_overlap(rating_repository.list_all(session, for_update=True), new_min, new_max)
|
||
if rating_repository.get_by_name(session, payload.grade_name) is not None:
|
||
raise BadRequestError("该等级名称已存在")
|
||
|
||
grade = CustomerRatingGrade(
|
||
grade_name=payload.grade_name,
|
||
min_score=new_min,
|
||
max_score=new_max,
|
||
sort_order=payload.sort_order,
|
||
remark=payload.remark,
|
||
)
|
||
try:
|
||
rating_repository.create(session, grade)
|
||
# 响应与后置动作全部在 commit 之前完成（见 product_service 同处说明）
|
||
out = _to_out(grade)
|
||
session.commit()
|
||
except IntegrityError as exc:
|
||
# 先查后写不是原子操作：并发提交同名等级会撞 uk_customer_rating_grades_grade_name
|
||
session.rollback()
|
||
_raise_if_duplicate(exc)
|
||
|
||
logger.info("新增评级等级 | grade_id=%s | name=%s", grade.id, payload.grade_name)
|
||
return out
|
||
|
||
|
||
def update_grade(
|
||
session: Session,
|
||
grade_id: int,
|
||
payload: RatingGradeUpdateIn,
|
||
*,
|
||
operator: User,
|
||
meta: RequestMeta | None = None,
|
||
) -> RatingGradeOut:
|
||
"""编辑评级等级（全量替换）。
|
||
|
||
区间与其他等级不得重叠 —— 重叠判定以**修改后的区间**对全部既有等级
|
||
（排除自身）两两相交，命中即 400 并点名冲突等级。
|
||
|
||
:raises NotFoundError: 评级等级不存在
|
||
:raises BadRequestError: 分值区间不合法 / 分值区间与等级 X 重叠 / 该等级名称已存在
|
||
"""
|
||
grade = rating_repository.get_by_id(session, grade_id)
|
||
with audit_service.audit(
|
||
session,
|
||
operator=operator,
|
||
record=AuditRecord(
|
||
module=MODULE_RATING,
|
||
action=ACTION_UPDATE,
|
||
summary=f"修改评级等级 {grade.grade_name if grade else f'#{grade_id}'}",
|
||
request_data=payload.model_dump(exclude_none=True),
|
||
),
|
||
meta=meta,
|
||
):
|
||
# 值域校验在审计上下文内（同 create 口径）
|
||
new_min = _to_decimal(payload.min_score)
|
||
new_max = _to_decimal(payload.max_score)
|
||
_ensure_score_range(new_min, new_max)
|
||
if grade is None:
|
||
raise NotFoundError("评级等级不存在")
|
||
|
||
_ensure_no_overlap(
|
||
rating_repository.list_all(session, for_update=True),
|
||
new_min,
|
||
new_max,
|
||
exclude_id=grade_id,
|
||
)
|
||
existed = rating_repository.get_by_name(session, payload.grade_name)
|
||
if existed is not None and existed.id != grade_id:
|
||
raise BadRequestError("该等级名称已存在")
|
||
|
||
grade.grade_name = payload.grade_name
|
||
grade.min_score = new_min
|
||
grade.max_score = new_max
|
||
grade.sort_order = payload.sort_order
|
||
grade.remark = payload.remark
|
||
out = _to_out(grade)
|
||
try:
|
||
session.commit()
|
||
except IntegrityError as exc:
|
||
# 改名可能撞 uk_customer_rating_grades_grade_name（本行改值与他人既有数据重合）
|
||
session.rollback()
|
||
_raise_if_duplicate(exc)
|
||
|
||
logger.info("修改评级等级 | grade_id=%s | name=%s", grade_id, payload.grade_name)
|
||
return out
|
||
|
||
|
||
def delete_grade(
|
||
session: Session,
|
||
grade_id: int,
|
||
*,
|
||
operator: User,
|
||
meta: RequestMeta | None = None,
|
||
) -> None:
|
||
"""物理删除评级等级。
|
||
|
||
等级表对客户评级**无外键引用** —— 评级行快照固化等级名（见
|
||
``models/customer_rating.py`` 的论证），判定只读本表现值，故可物理删除
|
||
且无需锁行防并发引用；删除等级不改变已评级客户的历史结论。
|
||
|
||
**允许清空**：全部删空后新评级只记总分、等级为 NULL（前端已对应提示），
|
||
不强制保留任何一条。
|
||
|
||
:raises NotFoundError: 评级等级不存在
|
||
"""
|
||
grade = rating_repository.get_by_id(session, grade_id)
|
||
with audit_service.audit(
|
||
session,
|
||
operator=operator,
|
||
record=AuditRecord(
|
||
module=MODULE_RATING,
|
||
action=ACTION_DELETE,
|
||
summary=f"删除评级等级 {grade.grade_name if grade else f'#{grade_id}'}",
|
||
),
|
||
meta=meta,
|
||
):
|
||
if grade is None:
|
||
raise NotFoundError("评级等级不存在")
|
||
|
||
grade_name = grade.grade_name # 固化：对象随删除失效
|
||
rating_repository.delete(session, grade)
|
||
session.commit()
|
||
logger.info("删除评级等级 | grade_name=%s", grade_name)
|
||
|
||
|
||
# ==================== 客户评级（每客户一行，落库） ====================
|
||
|
||
|
||
def resolve_grade(
|
||
grades: list[CustomerRatingGrade], total_score: Decimal
|
||
) -> CustomerRatingGrade | None:
|
||
"""等级判定纯函数：按 sort_order 升序取首个 min_score ≤ 总分的等级。
|
||
|
||
与前端打分卡的实时判定同规则同序（前端按数值比较，此处 Decimal 比较
|
||
——同一规则的两端实现，测试锁定边界行为一致）。无命中（总分低于全部
|
||
下限或等级表为空）返回 None，由调用方决定展示与落库（快照记 NULL）。
|
||
"""
|
||
# id 兜底让同序等级次序稳定（纯函数也接受未持久化对象：id 可能为 None）
|
||
for grade in sorted(grades, key=lambda g: (g.sort_order, g.id or 0)):
|
||
if grade.min_score <= total_score:
|
||
return grade
|
||
return None
|
||
|
||
|
||
def _to_rating_out(rating: CustomerRating) -> CustomerRatingOut:
|
||
"""客户评级行转响应模型。分值 Decimal → str（分值红线出参口径）。"""
|
||
return CustomerRatingOut(
|
||
customer_id=rating.customer_id,
|
||
score_ship_amount=str(rating.score_ship_amount),
|
||
score_receipt_rate=str(rating.score_receipt_rate),
|
||
score_profit_rate=str(rating.score_profit_rate),
|
||
score_contract_cooperation=str(rating.score_contract_cooperation),
|
||
score_site_construction=str(rating.score_site_construction),
|
||
total_score=str(rating.total_score),
|
||
grade_name=rating.grade_name,
|
||
remark=rating.remark,
|
||
scorer_name=rating.scorer_name,
|
||
create_time=rating.create_time,
|
||
update_time=rating.update_time,
|
||
)
|
||
|
||
|
||
def _ensure_item_scores(scores: list[Decimal]) -> None:
|
||
"""单项评分值域：0 ≤ 分 ≤ 20（总分由五项求和自然落在 0-100）。
|
||
|
||
须在审计上下文内调用，失败也要留痕（同 _ensure_score_range 口径）。
|
||
"""
|
||
for score in scores:
|
||
if not (_SCORE_FLOOR <= score <= _ITEM_SCORE_CEIL):
|
||
raise BadRequestError("单项评分须在 0~20 之间")
|
||
|
||
|
||
def get_customer_rating(session: Session, customer_id: int) -> CustomerRatingOut | None:
|
||
"""查某客户当前评级；未评级返回 None（接口信封 data 为 null）。
|
||
|
||
「未评级」是正常态而非错误，不用 404 表达；客户不存在才是 NotFound。
|
||
|
||
:raises NotFoundError: 客户不存在
|
||
"""
|
||
customer = customer_repository.get_by_id(session, customer_id)
|
||
if customer is None:
|
||
raise NotFoundError("客户不存在")
|
||
rating = customer_rating_repository.get_by_customer_id(session, customer_id)
|
||
return _to_rating_out(rating) if rating is not None else None
|
||
|
||
|
||
def save_customer_rating(
|
||
session: Session,
|
||
customer_id: int,
|
||
payload: CustomerRatingSaveIn,
|
||
*,
|
||
operator: User,
|
||
meta: RequestMeta | None = None,
|
||
) -> CustomerRatingOut:
|
||
"""保存客户评级（upsert：首次创建，再次保存整体覆盖）。
|
||
|
||
总分与等级**服务端重算**：总分按五项求和（不采信前端回传），等级按
|
||
保存时点的等级映射判定后**快照固化**进评级行 —— 等级表后续增删改
|
||
不影响该行（见 ``models/customer_rating.py``）。
|
||
|
||
:raises NotFoundError: 客户不存在
|
||
:raises BadRequestError: 单项评分须在 0~20 之间（审计上下文内校验，留痕）
|
||
"""
|
||
# 只读预取在审计上下文之前：客户不存在（NotFound 失败）也能留下可读记录，
|
||
# 摘要需要客户名
|
||
customer = customer_repository.get_by_id(session, customer_id)
|
||
existing = customer_rating_repository.get_by_customer_id(session, customer_id)
|
||
with audit_service.audit(
|
||
session,
|
||
operator=operator,
|
||
record=AuditRecord(
|
||
module=MODULE_RATING,
|
||
action=ACTION_CREATE if existing is None else ACTION_UPDATE,
|
||
summary=f"评定客户「{customer.company_name if customer else f'#{customer_id}'}」评级",
|
||
request_data=payload.model_dump(exclude_none=True),
|
||
),
|
||
meta=meta,
|
||
):
|
||
# 值域校验在审计上下文内：分值非法的 400 也要留痕（同等级配置口径）
|
||
item_scores = [
|
||
_to_decimal(payload.score_ship_amount),
|
||
_to_decimal(payload.score_receipt_rate),
|
||
_to_decimal(payload.score_profit_rate),
|
||
_to_decimal(payload.score_contract_cooperation),
|
||
_to_decimal(payload.score_site_construction),
|
||
]
|
||
_ensure_item_scores(item_scores)
|
||
if customer is None:
|
||
raise NotFoundError("客户不存在")
|
||
|
||
total_score = sum(item_scores, _SCORE_FLOOR)
|
||
grade = resolve_grade(rating_repository.list_all(session), total_score)
|
||
|
||
def _apply(row: CustomerRating) -> None:
|
||
"""把五项分值/总分/等级快照写入评级行（创建与并发兜底共用）。"""
|
||
(
|
||
row.score_ship_amount,
|
||
row.score_receipt_rate,
|
||
row.score_profit_rate,
|
||
row.score_contract_cooperation,
|
||
row.score_site_construction,
|
||
) = item_scores
|
||
row.total_score = total_score
|
||
row.grade_name = grade.grade_name if grade is not None else None
|
||
row.remark = payload.remark
|
||
row.scorer_id = operator.id
|
||
row.scorer_name = operator.username
|
||
|
||
try:
|
||
if existing is None:
|
||
rating = CustomerRating(
|
||
customer_id=customer_id,
|
||
scorer_id=operator.id,
|
||
scorer_name=operator.username,
|
||
)
|
||
_apply(rating)
|
||
customer_rating_repository.create(session, rating)
|
||
else:
|
||
_apply(existing)
|
||
rating = existing
|
||
out = _to_rating_out(rating)
|
||
session.commit()
|
||
except IntegrityError as exc:
|
||
# 先查后写不是原子操作：并发为同一客户首次评级会撞
|
||
# uk_customer_ratings_customer_id，回滚后转为更新兜底
|
||
session.rollback()
|
||
if not is_duplicate_entry(exc):
|
||
raise
|
||
raced = customer_rating_repository.get_by_customer_id(session, customer_id)
|
||
if raced is None:
|
||
# 1062 但行不在：数据态异常，交全局 IntegrityError 兜底
|
||
raise
|
||
_apply(raced)
|
||
rating = raced
|
||
out = _to_rating_out(rating)
|
||
session.commit()
|
||
|
||
logger.info(
|
||
"保存客户评级 | customer_id=%s | total=%s | grade=%s",
|
||
customer_id,
|
||
out.total_score,
|
||
out.grade_name,
|
||
)
|
||
return out
|