feat(business): 回款管理接口（批次号/手动指定批次/已核算禁改/合计回写）

- 批次 HK 生成；shipment_id 同业务强校验（统一失败面防探测）
- 已核算回款禁改禁删（M3 防重锚点）；payment_days 派生（发货日首日=第 1 天）

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

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

@@ -0,0 +1,89 @@
+"""回款管理接口（需求 2.2.3）。
+
+回款按业务登记到账款项，批次号自动生成（HK 前缀）。校验规则（发货
+批次同业务强校验、已核算禁改禁删、汇总重算）见 `receipt_service` 模块
+docstring；基础数据属业务操作，登录即可访问。
+"""
+
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Path, Query
+
+from app.api.deps import DbSession, RequestMetaDep, get_current_user
+from app.core.response import ApiResponse, PageResult, ok
+from app.models import User
+from app.schemas.receipt import (
+    ReceiptCreateIn,
+    ReceiptOut,
+    ReceiptQuery,
+    ReceiptUpdateIn,
+)
+from app.services import receipt_service
+
+# 基础数据是业务操作：登录即可，不要求管理员（区别于系统管理的 AdminUser）
+CurrentUser = Annotated[User, Depends(get_current_user)]
+
+router = APIRouter(prefix="/receipts", tags=["回款管理"])
+
+
+@router.get("", response_model=ApiResponse[PageResult[ReceiptOut]], summary="回款列表")
+def list_receipts(
+    session: DbSession,
+    _: CurrentUser,
+    query: Annotated[ReceiptQuery, Query()],
+) -> ApiResponse[PageResult[ReceiptOut]]:
+    """分页查询回款。附业务号/发货单号与回款天数派生值，可按业务/日期/抵顶/核算状态筛选。"""
+    return ok(receipt_service.list_receipts(session, query))
+
+
+@router.post("", response_model=ApiResponse[ReceiptOut], summary="新增回款")
+def create_receipt(
+    session: DbSession,
+    current_user: CurrentUser,
+    payload: ReceiptCreateIn,
+    meta: RequestMetaDep,
+) -> ApiResponse[ReceiptOut]:
+    """新增回款。批次号自动生成；发货批次须与本回款同属一个业务；业务须执行中。"""
+    return ok(
+        receipt_service.create_receipt(session, payload, operator=current_user, meta=meta),
+        message="回款已登记",
+    )
+
+
+@router.get("/{receipt_id}", response_model=ApiResponse[ReceiptOut], summary="回款详情")
+def get_receipt(
+    session: DbSession,
+    _: CurrentUser,
+    receipt_id: Annotated[int, Path(ge=1)],
+) -> ApiResponse[ReceiptOut]:
+    """回款详情（含回款天数派生值，未指定发货批次时为空）。"""
+    return ok(receipt_service.get_detail(session, receipt_id))
+
+
+@router.put("/{receipt_id}", response_model=ApiResponse[ReceiptOut], summary="编辑回款")
+def update_receipt(
+    session: DbSession,
+    current_user: CurrentUser,
+    receipt_id: Annotated[int, Path(ge=1)],
+    payload: ReceiptUpdateIn,
+    meta: RequestMetaDep,
+) -> ApiResponse[ReceiptOut]:
+    """编辑回款（全量替换并重算合计）。已核算回款禁止修改；发货批次重新校验同业务。"""
+    return ok(
+        receipt_service.update_receipt(
+            session, receipt_id, payload, operator=current_user, meta=meta
+        ),
+        message="已保存",
+    )
+
+
+@router.delete("/{receipt_id}", response_model=ApiResponse[None], summary="删除回款")
+def delete_receipt(
+    session: DbSession,
+    current_user: CurrentUser,
+    receipt_id: Annotated[int, Path(ge=1)],
+    meta: RequestMetaDep,
+) -> ApiResponse[None]:
+    """删除回款。已核算回款禁止删除；业务已完成或已终止时禁删。"""
+    receipt_service.delete_receipt(session, receipt_id, operator=current_user, meta=meta)
+    return ok(message="回款已删除")

backend/app/repositories/receipt_repository.py

@@ -0,0 +1,127 @@
+"""回款数据访问：回款行与关联引用（业务号/发货单号/发货日）。
+
+只读写数据并 flush；commit/rollback 与业务异常一律在 service 层。
+
+- 列表与详情统一 join 出 ``business_no / shipment_no / ship_date``，
+  ship_date 供 service 按 ``formula.payment_days`` 派生回款天数 —— 关联
+  引用经本层查询一次供给，service 不再对发货单二次查询
+- shipment_id 是手动指定（可空），发货单关联一律 outerjoin
+"""
+
+from datetime import date
+
+from sqlalchemy import Select, func, select
+from sqlalchemy.orm import Session
+
+from app.models import Business, Receipt, Shipment
+
+
+def get_by_id(session: Session, receipt_id: int) -> Receipt | None:
+    """按主键取回款行（供摘要预取与状态判定；不锁行，写保护锚在业务行）。"""
+    return session.get(Receipt, receipt_id)
+
+
+def get_with_refs(
+    session: Session, receipt_id: int
+) -> tuple[Receipt, str, str | None, date | None] | None:
+    """回款行 + 业务号 + 发货单号 + 发货日（DTO 与 payment_days 派生共用）。
+
+    business 主表为 inner join（FK NOT NULL，业务与回款同存亡）；发货单
+    可未指定，outerjoin 后 shipment_no / ship_date 可能为 None。
+    """
+    row = session.execute(
+        select(Receipt, Business.business_no, Shipment.shipment_no, Shipment.ship_date)
+        .join(Business, Business.id == Receipt.business_id)
+        .outerjoin(Shipment, Shipment.id == Receipt.shipment_id)
+        .where(Receipt.id == receipt_id)
+    ).first()
+    return row
+
+
+def get_shipment(session: Session, shipment_id: int) -> Shipment | None:
+    """按主键取发货单（同业务强校验的只读预查；FK RESTRICT 为最终防线）。"""
+    return session.get(Shipment, shipment_id)
+
+
+def _apply_filters(
+    stmt: Select,
+    *,
+    business_id: int | None,
+    date_from: date | None,
+    date_to: date | None,
+    is_barter: int | None,
+    settlement_status: int | None,
+) -> Select:
+    """回款列表的筛选条件（全部落在 Receipt 自身列上）。"""
+    if business_id is not None:
+        stmt = stmt.where(Receipt.business_id == business_id)
+    if date_from is not None:
+        stmt = stmt.where(Receipt.receipt_date >= date_from)
+    if date_to is not None:
+        stmt = stmt.where(Receipt.receipt_date <= date_to)
+    if is_barter is not None:
+        stmt = stmt.where(Receipt.is_barter == is_barter)
+    if settlement_status is not None:
+        stmt = stmt.where(Receipt.settlement_status == settlement_status)
+    return stmt
+
+
+def list_receipts(
+    session: Session,
+    *,
+    offset: int,
+    limit: int,
+    business_id: int | None,
+    date_from: date | None,
+    date_to: date | None,
+    is_barter: int | None,
+    settlement_status: int | None,
+) -> list[tuple[Receipt, str, str | None, date | None]]:
+    """分页查询回款（附业务号/发货单号/发货日），按 id 倒序。"""
+    stmt = _apply_filters(
+        select(Receipt, Business.business_no, Shipment.shipment_no, Shipment.ship_date)
+        .join(Business, Business.id == Receipt.business_id)
+        .outerjoin(Shipment, Shipment.id == Receipt.shipment_id),
+        business_id=business_id,
+        date_from=date_from,
+        date_to=date_to,
+        is_barter=is_barter,
+        settlement_status=settlement_status,
+    )
+    return list(
+        session.execute(stmt.order_by(Receipt.id.desc()).offset(offset).limit(limit)).all()
+    )
+
+
+def count_receipts(
+    session: Session,
+    *,
+    business_id: int | None,
+    date_from: date | None,
+    date_to: date | None,
+    is_barter: int | None,
+    settlement_status: int | None,
+) -> int:
+    """回款总数（筛选条件与列表共用同一 apply_filters，防两侧口径漂移）。"""
+    stmt = _apply_filters(
+        select(func.count()).select_from(Receipt),
+        business_id=business_id,
+        date_from=date_from,
+        date_to=date_to,
+        is_barter=is_barter,
+        settlement_status=settlement_status,
+    )
+    return int(session.execute(stmt).scalar_one())
+
+
+def create(session: Session, receipt: Receipt) -> Receipt:
+    """新增回款行。只 flush 以取得自增 id，编号唯一键冲突由 service 兜底。"""
+    session.add(receipt)
+    session.flush()
+    return receipt
+
+
+def delete(session: Session, receipt: Receipt) -> None:
+    """物理删除回款行（已核算与否的准入判定在 service 层）。"""
+    session.delete(receipt)
+    session.flush()

backend/app/schemas/receipt.py

@@ -0,0 +1,104 @@
+"""回款管理的请求/响应模型（需求 2.2.3）。
+
+金额入参出参一律 str（服务层 Decimal，全程禁 float）；字段契约**纯蛇形
+直出**（2026-10-13 定稿，无驼峰 alias）。
+
+payment_days 是派生值（发货日 → 回款日的天数，对照表 #1），不在写接口
+入参内 —— 落库列只有 shipment_id 源数据，派生值实时计算，发货批次变更
+后无需回写历史。
+"""
+
+from datetime import date
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+from app.core.response import PageQuery
+
+# 金额校验是工资系统的红线口径（数字/非负/容量/两位小数拒绝），跨 schema
+# 单点复用 T7 定稿工具，禁止各处复制导致口径漂移；AMOUNT_MAX 即
+# DECIMAL(14,2) 的容量上限，与回款列容量同源
+from app.schemas.business import AMOUNT_MAX, _check_amount
+
+
+class ReceiptCreateIn(BaseModel):
+    """新增回款：业务必选，发货批次可暂不指定（手动指定，对照表 #18）。
+
+    shipment_id 指向的发货单必须与本回款同属一个业务（服务层强校验，
+    数据完整性红线）；shipment_id 可空 = 暂未指定，指定后回款天数方可派生。
+    """
+
+    model_config = ConfigDict(str_strip_whitespace=True)
+
+    business_id: int = Field(ge=1, description="业务ID")
+    shipment_id: int | None = Field(
+        default=None, ge=1, description="发货单ID，可空=暂未指定"
+    )
+    receipt_date: date = Field(description="回款到账日（M3 按此日匹配参数版本）")
+    amount: str = Field(description="回款金额（元），≥0，最多两位小数")
+    is_barter: int = Field(default=0, ge=0, le=1, description="以物抵顶标记：1是 0否")
+    remark: str | None = Field(default=None, max_length=255, description="备注")
+
+    @field_validator("amount")
+    @classmethod
+    def _check_amount_value(cls, value: str) -> str:
+        return _check_amount(value, "回款金额", AMOUNT_MAX) or value
+
+
+class ReceiptUpdateIn(BaseModel):
+    """编辑回款（全量替换语义）。
+
+    shipment_id 传 None = 清除指定（回到「暂未指定」）；amount 是必填项，
+    全量替换下不存在「不传即不变」的歧义。已核算（settlement_status=1）
+    的回款在服务层禁改，schema 层不感知该状态。
+    """
+
+    model_config = ConfigDict(str_strip_whitespace=True)
+
+    shipment_id: int | None = Field(
+        default=None, ge=1, description="发货单ID，可空=暂未指定（传空清除既有指定）"
+    )
+    receipt_date: date = Field(description="回款到账日")
+    amount: str = Field(description="回款金额（元），≥0，最多两位小数")
+    is_barter: int = Field(default=0, ge=0, le=1, description="以物抵顶标记：1是 0否")
+    remark: str | None = Field(default=None, max_length=255, description="备注")
+
+    @field_validator("amount")
+    @classmethod
+    def _check_amount_value(cls, value: str) -> str:
+        return _check_amount(value, "回款金额", AMOUNT_MAX) or value
+
+
+class ReceiptOut(BaseModel):
+    """回款响应（列表与写操作回包共用）。
+
+    business_no/shipment_no 为关联冗余展示；payment_days 为派生值：
+    未指定发货批次时为 None，指定后 = 发货日到回款日的天数（发货日首日
+    计第 1 天）。
+    """
+
+    model_config = ConfigDict(from_attributes=True)
+
+    id: int
+    receipt_batch_no: str
+    business_id: int
+    business_no: str
+    shipment_id: int | None = None
+    shipment_no: str | None = None
+    receipt_date: date
+    amount: str
+    is_barter: int
+    settlement_status: int
+    payment_days: int | None = None
+    remark: str | None = None
+
+
+class ReceiptQuery(PageQuery):
+    """回款列表查询条件（筛选参数纯蛇形直出，同 BusinessQuery 契约）。"""
+
+    business_id: int | None = Field(default=None, ge=1, description="业务筛选")
+    date_from: date | None = Field(default=None, description="回款到账日起（含）")
+    date_to: date | None = Field(default=None, description="回款到账日止（含）")
+    is_barter: int | None = Field(default=None, ge=0, le=1, description="以物抵顶筛选：1是 0否")
+    settlement_status: int | None = Field(
+        default=None, ge=0, le=1, description="核算状态筛选：0未核算 1已核算"
+    )

backend/app/services/receipt_service.py

@@ -0,0 +1,349 @@
+"""回款管理业务编排（需求 2.2.3）。
+
+事务边界在本层：repository 只 flush，commit/rollback 只写在这里。
+
+- **写操作前 FOR UPDATE 锁业务行**（delete_business 先例）：同业务的
+  发货/回款/编辑相互串行化，防止并发写后 ``_recalc_totals`` 重算
+  total_received_amount 互相覆盖（丢失更新）；批次号生成（MAX+FOR
+  UPDATE 锁当日序号区间）与唯一键（1062）冲突重试一次兜底，模式同
+  business_service
+- **shipment_id 同业务强校验**（数据完整性红线）：指定的发货单必须存在
+  且与本回款同属一个业务 —— 跨业务指定会让回款天数与台账归属漂移；
+  存在性与归属两类失败统一同一文案（防通过响应差异探测发货单存在性，
+  与登录判定的防枚举同哲学），FK RESTRICT 是最终防线
+- **settlement_status 是 M3 核算防重锚点**（需规 2.3.3「每笔回款仅核算
+  一次」）：已核算回款禁改禁删 —— 改金额/删行会让 M3 台账与源数据漂移；
+  M2 只写入 0（未核算），置 1 由 M3 核算服务负责
+- 回款合计回写统一收口 ``business_service._recalc_totals``（增删改后
+  都重算，禁各处手写 SUM）
+- payment_days 为**只读派生值**：shipment_id 非空且发货单在库时按
+  ``formula.payment_days(ship_date, receipt_date)`` 实时计算（发货日首日
+  计第 1 天），不落库；未指定批次为 None
+- 一期禁负数金额（≥0）：schema 拒绝；红字/退款冲销需求出现时先补设计
+"""
+
+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_RECEIPT,
+    NO_PREFIX_RECEIPT,
+)
+from app.core.exceptions import BadRequestError, NotFoundError
+from app.core.response import PageResult
+from app.models import Receipt
+from app.models.business import BUSINESS_STATUS_ACTIVE
+from app.models.receipt import SETTLEMENT_DONE, SETTLEMENT_PENDING
+from app.repositories import business_repository, receipt_repository
+from app.schemas.receipt import (
+    ReceiptCreateIn,
+    ReceiptOut,
+    ReceiptQuery,
+    ReceiptUpdateIn,
+)
+from app.services import audit_service
+from app.services.audit_service import AuditRecord, RequestMeta
+from app.services.business_service import _recalc_totals
+from app.services.formula import payment_days
+from app.services.no_generator import next_no
+
+logger = logging.getLogger(__name__)
+
+
+def _to_out(row: tuple) -> ReceiptOut:
+    """(receipt, business_no, shipment_no, ship_date) → 响应模型。
+
+    金额出参 str；payment_days 为派生值 —— ship_date 为 None（未指定批次）
+    时输出 None；回款日早于发货日时按公式照实输出负值，忠实反映源数据
+    （钳制会掩盖录入异常，M3 台账同口径）。
+    """
+    receipt, business_no, shipment_no, ship_date = row
+    return ReceiptOut(
+        id=receipt.id,
+        receipt_batch_no=receipt.receipt_batch_no,
+        business_id=receipt.business_id,
+        business_no=business_no,
+        shipment_id=receipt.shipment_id,
+        shipment_no=shipment_no,
+        receipt_date=receipt.receipt_date,
+        amount=str(receipt.amount),
+        is_barter=receipt.is_barter,
+        settlement_status=receipt.settlement_status,
+        payment_days=(
+            payment_days(ship_date, receipt.receipt_date) if ship_date is not None else None
+        ),
+        remark=receipt.remark,
+    )
+
+
+def _generate_no(session: Session) -> str:
+    """生成回款批次号（HK 前缀；MAX+FOR UPDATE 与 1062 兜底见 no_generator）。"""
+    return next_no(
+        session, model=Receipt, column_name="receipt_batch_no", prefix=NO_PREFIX_RECEIPT
+    )
+
+
+def _ensure_shipment_of_business(
+    session: Session, shipment_id: int | None, business_id: int
+) -> tuple[str | None, object | None]:
+    """发货批次同业务强校验，返回 (shipment_no, ship_date) 供 DTO 组装。
+
+    「必须存在且 business_id 相同」两类失败统一同一文案（防探测发货单
+    存在性）；shipment_id 为 None = 暂未指定，直接放行。
+
+    :raises BadRequestError: 发货批次不存在或不属于该业务
+    """
+    if shipment_id is None:
+        return None, None
+    shipment = receipt_repository.get_shipment(session, shipment_id)
+    if shipment is None or shipment.business_id != business_id:
+        raise BadRequestError("发货批次不属于该业务，无法指定")
+    return shipment.shipment_no, shipment.ship_date
+
+
+def _ensure_business_writable(session: Session, business_id: int, action: str):
+    """锁业务行并要求状态=1 执行中（新增/编辑/删除共用的写准入）。
+
+    锁在判定之前取：判定结果基于锁定后的行，配合 _recalc_totals 的重算
+    同事务提交，并发写不会丢失更新。
+
+    :raises NotFoundError: 业务不存在
+    :raises BadRequestError: 业务已完成或已终止
+    """
+    business = business_repository.get_by_id(session, business_id, for_update=True)
+    if business is None:
+        raise NotFoundError("业务不存在")
+    if business.status != BUSINESS_STATUS_ACTIVE:
+        raise BadRequestError(f"业务已完成或已终止，禁止{action}回款")
+    return business
+
+
+def list_receipts(session: Session, query: ReceiptQuery) -> PageResult[ReceiptOut]:
+    """分页查询回款（附业务号/发货单号与 payment_days 派生值）。"""
+    rows = receipt_repository.list_receipts(
+        session,
+        offset=query.offset,
+        limit=query.page_size,
+        business_id=query.business_id,
+        date_from=query.date_from,
+        date_to=query.date_to,
+        is_barter=query.is_barter,
+        settlement_status=query.settlement_status,
+    )
+    total = receipt_repository.count_receipts(
+        session,
+        business_id=query.business_id,
+        date_from=query.date_from,
+        date_to=query.date_to,
+        is_barter=query.is_barter,
+        settlement_status=query.settlement_status,
+    )
+    return PageResult[ReceiptOut](
+        records=[_to_out(row) for row in rows],
+        total=total,
+        page_num=query.page_num,
+        page_size=query.page_size,
+    )
+
+
+def get_detail(session: Session, receipt_id: int) -> ReceiptOut:
+    """回款详情（同列表 DTO，含 payment_days 派生值）。
+
+    :raises NotFoundError: 回款不存在
+    """
+    row = receipt_repository.get_with_refs(session, receipt_id)
+    if row is None:
+        raise NotFoundError("回款不存在")
+    return _to_out(row)
+
+
+def create_receipt(
+    session: Session,
+    payload: ReceiptCreateIn,
+    *,
+    operator,
+    meta: RequestMeta | None = None,
+) -> ReceiptOut:
+    """新增回款：锁业务行 → 校验 → 建行 + 重算合计，一个事务。
+
+    摘要三要素（批次号/业务号/金额）在进入审计前就绪：业务号取自锁行
+    结果、批次号先行生成（SELECT MAX FOR UPDATE，不产生未提交变更）、
+    金额来自入参；引用校验（业务状态/发货批次同业务）在审计上下文内，
+    失败也留痕 —— 失败路径回滚释放已取序号，当日序号出现空洞无害
+    （唯一性不依赖连续）。
+
+    :raises NotFoundError: 业务不存在
+    :raises BadRequestError: 业务已完成或已终止 / 发货批次不属于该业务
+    """
+    business = business_repository.get_by_id(session, payload.business_id, for_update=True)
+    batch_no = _generate_no(session)
+    with audit_service.audit(
+        session,
+        operator=operator,
+        record=AuditRecord(
+            module=MODULE_RECEIPT,
+            action=ACTION_CREATE,
+            summary=(
+                f"新增回款 {batch_no}"
+                f"（业务 {business.business_no if business else f'#{payload.business_id}'}，"
+                f"金额 {payload.amount}）"
+            ),
+            request_data=payload.model_dump(exclude_none=True),
+        ),
+        meta=meta,
+    ):
+        if business is None:
+            raise NotFoundError("业务不存在")
+        if business.status != BUSINESS_STATUS_ACTIVE:
+            raise BadRequestError("业务已完成或已终止，禁止新增回款")
+        shipment_no, ship_date = _ensure_shipment_of_business(
+            session, payload.shipment_id, payload.business_id
+        )
+
+        try:
+            receipt = receipt_repository.create(
+                session,
+                Receipt(
+                    receipt_batch_no=batch_no,
+                    business_id=payload.business_id,
+                    shipment_id=payload.shipment_id,
+                    receipt_date=payload.receipt_date,
+                    amount=Decimal(payload.amount),
+                    is_barter=payload.is_barter,
+                    settlement_status=SETTLEMENT_PENDING,
+                    remark=payload.remark,
+                ),
+            )
+            _recalc_totals(session, business)
+            out = _to_out((receipt, business.business_no, shipment_no, ship_date))
+            new_id = receipt.id
+            session.commit()
+        except IntegrityError as exc:
+            session.rollback()
+            if _is_duplicate_entry(exc):
+                # 批次号撞唯一键的残余竞态：兜底重试一次（防无限重试掩盖死锁）
+                return _retry_create(session, payload, operator=operator, meta=meta)
+            raise
+
+        logger.info(
+            "新增回款 | receipt_id=%s | no=%s | business_id=%s",
+            new_id,
+            batch_no,
+            payload.business_id,
+        )
+        return out
+
+
+def _retry_create(
+    session: Session,
+    payload: ReceiptCreateIn,
+    *,
+    operator,
+    meta: RequestMeta | None,
+) -> ReceiptOut:
+    """批次号撞唯一键后的单次重试（重新锁行与生成编号，防死锁类缺陷掩盖）。"""
+    return create_receipt(session, payload, operator=operator, meta=meta)
+
+
+def update_receipt(
+    session: Session,
+    receipt_id: int,
+    payload: ReceiptUpdateIn,
+    *,
+    operator,
+    meta: RequestMeta | None = None,
+) -> ReceiptOut:
+    """编辑回款（全量替换）+ 重算合计，一个事务。
+
+    **已核算回款禁止修改**（M3 防重锚点）：M3 台账已按该行固化核算，
+    改金额/到账日/批次指定会让台账与源数据漂移。shipment_id 变更后
+    重新做同业务强校验（全量替换语义下每次编辑都校验，无「未变即免检」
+    的旁路）。
+
+    :raises NotFoundError: 回款不存在 / 业务不存在
+    :raises BadRequestError: 已核算禁改 / 业务已完成或已终止 / 发货批次不属于该业务
+    """
+    receipt = receipt_repository.get_by_id(session, receipt_id)
+    with audit_service.audit(
+        session,
+        operator=operator,
+        record=AuditRecord(
+            module=MODULE_RECEIPT,
+            action=ACTION_UPDATE,
+            summary=f"修改回款 {receipt.receipt_batch_no if receipt else f'#{receipt_id}'}",
+            request_data=payload.model_dump(exclude_none=True),
+        ),
+        meta=meta,
+    ):
+        if receipt is None:
+            raise NotFoundError("回款不存在")
+        if receipt.settlement_status == SETTLEMENT_DONE:
+            raise BadRequestError("已核算回款禁止修改")
+        business = _ensure_business_writable(session, receipt.business_id, "修改")
+        # 返回值不取用：DTO 组装以 get_with_refs 重查为准（shipment 变更后同源）
+        _ensure_shipment_of_business(session, payload.shipment_id, receipt.business_id)
+
+        receipt.receipt_date = payload.receipt_date
+        receipt.shipment_id = payload.shipment_id
+        receipt.amount = Decimal(payload.amount)
+        receipt.is_barter = payload.is_barter
+        receipt.remark = payload.remark
+        _recalc_totals(session, business)
+        out = _to_out(receipt_repository.get_with_refs(session, receipt_id))
+        batch_no = receipt.receipt_batch_no
+        session.commit()
+        # commit 后不再触碰 ORM 属性（audit_service 约定）：批次号已固化
+        logger.info("修改回款 | receipt_id=%s | no=%s", receipt_id, batch_no)
+        return out
+
+
+def delete_receipt(
+    session: Session,
+    receipt_id: int,
+    *,
+    operator,
+    meta: RequestMeta | None = None,
+) -> None:
+    """删除回款（物理删除）+ 重算合计，一个事务。
+
+    **已核算回款禁止删除**（M3 防重锚点，理由同编辑）；业务须状态=1
+    执行中 —— 终态业务的回款是 M3 核算与报表的既成事实，不得删改。
+
+    :raises NotFoundError: 回款不存在 / 业务不存在
+    :raises BadRequestError: 已核算禁删 / 业务已完成或已终止
+    """
+    receipt = receipt_repository.get_by_id(session, receipt_id)
+    with audit_service.audit(
+        session,
+        operator=operator,
+        record=AuditRecord(
+            module=MODULE_RECEIPT,
+            action=ACTION_DELETE,
+            summary=f"删除回款 {receipt.receipt_batch_no if receipt else f'#{receipt_id}'}",
+        ),
+        meta=meta,
+    ):
+        if receipt is None:
+            raise NotFoundError("回款不存在")
+        if receipt.settlement_status == SETTLEMENT_DONE:
+            raise BadRequestError("已核算回款禁止删除")
+        business = _ensure_business_writable(session, receipt.business_id, "删除")
+
+        batch_no = receipt.receipt_batch_no
+        receipt_repository.delete(session, receipt)
+        _recalc_totals(session, business)
+        session.commit()
+        # commit 后不再触碰 ORM 属性（audit_service 约定）：批次号已固化
+        logger.info("删除回款 | receipt_id=%s | no=%s", receipt_id, batch_no)
+
+
+def _is_duplicate_entry(exc: IntegrityError) -> bool:
+    """识别 MySQL 唯一键冲突（1062），与 business_service 同口径。"""
+    args = getattr(exc.orig, "args", ())
+    return bool(args) and args[0] == 1062