feat(business): 客户评级等级映射配置接口（区间重叠判定/名称唯一）

- 判定=按 sort_order 升序首个 min_score≤总分，边界相邻不算重叠
- 新增同样执行重叠判定（判定表语义下插出重叠即脏配置）
- 等级配置无下游引用（评级为纯计算器，对照表 #25），允许清空

Co-Authored-By: Claude Code <noreply@anthropic.com>

backend/app/api/v1/business/__init__.py

@@ -0,0 +1,8 @@
+"""业务数据管理接口（需求 2.2）。
+
+挂载于 `/api/v1/business/*`，含业务单据维护、客户评级等级配置等业务数据入口。
+
+与 `api/v1/system` 的权限分工：业务数据是**业务操作**，登录即可访问
+（不要求管理员）；写操作同样经 `RequestMetaDep` 带入请求元信息，
+由 service 层写入操作日志。
+"""

backend/app/api/v1/business/ratings.py

@@ -0,0 +1,74 @@
+"""客户评级等级配置接口（需求 2.2.5）。
+
+评级计算器的等级映射判定表：判定按 sort_order 升序取首个 min_score ≤ 总分
+的等级，max_score 仅展示。区间合法（0≤min≤max≤100）与互不重叠由后端校验
+—— 前端表单只是展示约束，后端校验是唯一防线。
+
+列表不分页全量返回：等级映射是判定表，数量个位数，分页无意义。
+业务数据属业务操作，登录用户即可访问（区别于系统管理模块的 AdminUser）；
+写操作经 RequestMetaDep 带入请求元信息，由 service 层写入操作日志。
+"""
+
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Path
+
+from app.api.deps import DbSession, RequestMetaDep, get_current_user
+from app.core.response import ApiResponse, ok
+from app.models import User
+from app.schemas.rating import RatingGradeCreateIn, RatingGradeOut, RatingGradeUpdateIn
+from app.services import rating_service
+
+CurrentUser = Annotated[User, Depends(get_current_user)]
+
+router = APIRouter(prefix="/rating-grades", tags=["客户评级"])
+
+
+@router.get("", response_model=ApiResponse[list[RatingGradeOut]], summary="评级等级列表")
+def list_grades(
+    session: DbSession,
+    _: CurrentUser,
+) -> ApiResponse[list[RatingGradeOut]]:
+    """全量等级列表，按判定顺序（sort_order 升序）返回，不分页。"""
+    return ok(rating_service.list_grades(session))
+
+
+@router.post("", response_model=ApiResponse[RatingGradeOut], summary="新增评级等级")
+def create_grade(
+    session: DbSession,
+    current_user: CurrentUser,
+    payload: RatingGradeCreateIn,
+    meta: RequestMetaDep,
+) -> ApiResponse[RatingGradeOut]:
+    """新增评级等级。区间须 0≤min≤max≤100 且与既有等级不重叠。"""
+    return ok(
+        rating_service.create_grade(session, payload, operator=current_user, meta=meta),
+        message="评级等级已创建",
+    )
+
+
+@router.put("/{grade_id}", response_model=ApiResponse[RatingGradeOut], summary="编辑评级等级")
+def update_grade(
+    session: DbSession,
+    current_user: CurrentUser,
+    grade_id: Annotated[int, Path(ge=1)],
+    payload: RatingGradeUpdateIn,
+    meta: RequestMetaDep,
+) -> ApiResponse[RatingGradeOut]:
+    """编辑评级等级（全量替换）。区间须合法且与其他等级不重叠。"""
+    return ok(
+        rating_service.update_grade(session, grade_id, payload, operator=current_user, meta=meta),
+        message="已保存",
+    )
+
+
+@router.delete("/{grade_id}", response_model=ApiResponse[None], summary="删除评级等级")
+def delete_grade(
+    session: DbSession,
+    current_user: CurrentUser,
+    grade_id: Annotated[int, Path(ge=1)],
+    meta: RequestMetaDep,
+) -> ApiResponse[None]:
+    """物理删除评级等级（等级表无下游引用，允许清空，理由见 service 说明）。"""
+    rating_service.delete_grade(session, grade_id, operator=current_user, meta=meta)
+    return ok(message="评级等级已删除")

backend/app/repositories/rating_repository.py

@@ -0,0 +1,58 @@
+"""客户评级等级配置数据访问。
+
+本层只 flush 不 commit —— 事务边界由 services 层控制。
+"""
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.models import CustomerRatingGrade
+
+# ==================== 读取 ====================
+
+
+def list_all(session: Session) -> list[CustomerRatingGrade]:
+    """全量等级，按 sort_order 升序、id 兜底（同序次序稳定）。
+
+    不分页：等级映射是判定表，数量个位数，分页无意义。
+    """
+    stmt = select(CustomerRatingGrade).order_by(
+        CustomerRatingGrade.sort_order, CustomerRatingGrade.id
+    )
+    return list(session.execute(stmt).scalars())
+
+
+def get_by_id(session: Session, grade_id: int) -> CustomerRatingGrade | None:
+    """按主键查询。
+
+    等级表无下游引用（评级是纯计算器不落评分记录），无需 ``for_update``
+    锁行 —— product_repository 锁行防的是并发引用悬空，此处不存在该风险。
+    """
+    return session.get(CustomerRatingGrade, grade_id)
+
+
+def get_by_name(session: Session, grade_name: str) -> CustomerRatingGrade | None:
+    """按等级名称查询。
+
+    grade_name 是 uk_customer_rating_grades_grade_name 唯一键的键值，先查
+    用于给出即时业务提示；并发窗口仍存在，最终由 IntegrityError 兜底
+    （见 rating_service）。
+    """
+    stmt = select(CustomerRatingGrade).where(CustomerRatingGrade.grade_name == grade_name)
+    return session.execute(stmt).scalar_one_or_none()
+
+
+# ==================== 写入 ====================
+
+
+def create(session: Session, grade: CustomerRatingGrade) -> CustomerRatingGrade:
+    """新增等级。只 flush 以取得自增 id。"""
+    session.add(grade)
+    session.flush()
+    return grade
+
+
+def delete(session: Session, grade: CustomerRatingGrade) -> None:
+    """物理删除等级。等级表无下游引用，调用方无需引用保护校验。"""
+    session.delete(grade)
+    session.flush()

backend/app/schemas/rating.py

@@ -0,0 +1,79 @@
+"""客户评级等级配置的请求/响应模型。
+
+分值（min_score/max_score）入参出参一律用**字符串**承载：前端 JS 的 number
+无法精确表示全部十进制小数，字符串是分值红线的入口闸门，Decimal 转换统一
+收敛在 service 层（与 schemas/product.py 同一口径）。
+
+注意：schema 只拦**格式**（可解析、有限数）；0≤min≤max≤100 与区间不重叠
+是横跨两字段/多行的业务值域规则，在 service 层审计上下文内统一判定 ——
+值域校验失败的 400 也要留痕（2026-10-13 定稿口径）。
+"""
+
+from datetime import datetime
+from decimal import Decimal, InvalidOperation
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+
+def _parse_score(value: str, *, label: str) -> str:
+    """分值字符串格式校验：必须可被 Decimal 解析且为有限数。
+
+    ``NaN`` 参与比较抛 InvalidOperation、``Infinity`` 落库即溢出，均须在
+    入口用 ``is_finite()`` 拦下（与 schemas/product.py 的 _parse_non_negative
+    同口径）；非负与 ≤100 属业务值域，留给 service 层在审计上下文内判定。
+    """
+    try:
+        score = Decimal(value.strip())
+    except InvalidOperation as exc:
+        raise ValueError(f"{label}必须是有效数字：{value}") from exc
+    if not score.is_finite():
+        raise ValueError(f"{label}必须为有限数字：{value}")
+    return value
+
+
+class RatingGradeOut(BaseModel):
+    """评级等级响应（列表直出）。分值按红线以字符串出参。"""
+
+    id: int
+    grade_name: str
+    min_score: str
+    max_score: str
+    sort_order: int
+    remark: str | None = None
+    create_time: datetime
+    update_time: datetime
+
+
+class RatingGradeCreateIn(BaseModel):
+    """新增评级等级。
+
+    min_score/max_score 必填：区间是判定表的判定依据，缺省无从判定。
+    sort_order 缺省 0（列默认值同源），判定按其升序取首个命中等级。
+    """
+
+    model_config = ConfigDict(str_strip_whitespace=True)
+
+    grade_name: str = Field(min_length=1, max_length=32, description="等级名称（如 A/优秀）")
+    min_score: str = Field(description="总分下限（含），0-100 数字字符串")
+    max_score: str = Field(description="总分上限（展示用），0-100 数字字符串")
+    sort_order: int = Field(default=0, ge=0, description="判定顺序，升序")
+    remark: str | None = Field(default=None, max_length=255, description="说明")
+
+    @field_validator("min_score")
+    @classmethod
+    def _check_min_score(cls, value: str) -> str:
+        """总分下限须为有限数字（值域 0-100 由 service 层判定）。"""
+        return _parse_score(value, label="总分下限")
+
+    @field_validator("max_score")
+    @classmethod
+    def _check_max_score(cls, value: str) -> str:
+        """总分上限须为有限数字（值域 0-100 由 service 层判定）。"""
+        return _parse_score(value, label="总分上限")
+
+
+class RatingGradeUpdateIn(RatingGradeCreateIn):
+    """编辑评级等级。
+
+    前端表单全量提交，故为全量替换语义；等级表无启停状态列，字段与新增一致。
+    """

backend/app/services/rating_service.py

@@ -0,0 +1,279 @@
+"""客户评级等级配置业务编排。
+
+评级计算器定位（对照表 #25）：仅打分工具、不影响绩效金额，评分记录不落库
+——本表承载的等级映射是**判定表**，判定规则为按 sort_order 升序取首个
+min_score ≤ 总分的等级，max_score 仅展示。两条硬约束（违反会算错等级）：
+
+- 分值全程 ``Decimal``：入参出参 str、schema 拦格式，service 层负责字符串
+  到 Decimal 的转换与落库；0≤min≤max≤100 的区间校验在**审计上下文内**
+  完成，失败的 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
+  靠前的等级，相邻无歧义
+
+删除允许清空：等级表无下游引用（评分记录不落库，判定只读本表现值、不
+回溯历史快照），物理删除无悬空风险；全部删空后计算器只显示总分、不出
+等级（前端已对应提示），故不强制保留任何一条。
+"""
+
+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.exceptions import BadRequestError, NotFoundError
+from app.models import CustomerRatingGrade, User
+from app.repositories import rating_repository
+from app.schemas.rating import 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")
+
+# 唯一键冲突的 MySQL 错误码（pymysql 原始异常 args 首位）
+_ERR_DUP_ENTRY = 1062
+
+
+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 是「已存在」语义，可转业务提示；其余完整性错误（外键、
+    数据溢出等）语义未知，必须原样上抛交由全局处理器记 500，不能吞成 400。
+    """
+    orig_args = getattr(exc.orig, "args", ())
+    return bool(orig_args) and orig_args[0] == _ERR_DUP_ENTRY
+
+
+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), 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), 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:
+    """物理删除评级等级。
+
+    等级表无下游引用 —— 评级是纯计算器不落评分记录（对照表 #25），判定
+    只读本表现值、不回溯历史快照，故可物理删除且无需锁行防并发引用。
+
+    **允许清空**：全部删空后计算器只显示总分、不出等级（前端已对应提示），
+    不强制保留任何一条。
+
+    :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)