refactor(business): 编辑出参取校验返回值省重查，submit_change 批量预载

- update_receipt 取 _ensure_shipment_of_business 返回值组装出参
  （shipment_id 已赋为 payload 新值，与 get_with_refs 重查同源），
  省一次三表 join；update_shipment 同理取锁定行的 business_no
- 删 business_repository.count_items（无任何调用的死代码，删除保护
  实际走 has_docs）
- submit_change 校验段定义与岗位各一条 IN 批量预载（逐条查询在批次
  扩大时是写侧 N+1）；岗位字典键 casefold 归一保持与单条版 ci 校对
  等价的命中语义；旧值留痕保持逐条（窗口函数优化留待 M3 一并设计）

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

backend/app/repositories/business_repository.py

@@ -197,18 +197,6 @@ def replace_items(session: Session, business_id: int, rows: list[BusinessItem])
     session.flush()
 
 
-def count_items(session: Session, business_ids: list[int]) -> dict[int, int]:
-    """业务明细行数（删除保护备用）。"""
-    if not business_ids:
-        return {}
-    rows = session.execute(
-        select(BusinessItem.business_id, func.count())
-        .where(BusinessItem.business_id.in_(business_ids))
-        .group_by(BusinessItem.business_id)
-    ).all()
-    return dict(rows)
-
-
 def add_status_log(session: Session, log: BusinessStatusLog) -> None:
     """追加状态留痕（不 commit）。"""
     session.add(log)

backend/app/repositories/business_role_repository.py

@@ -25,6 +25,20 @@ def get_by_code(session: Session, role_code: str) -> BusinessRole | None:
     return session.execute(stmt).scalar_one_or_none()
 
 
+def list_by_codes(session: Session, role_codes: list[str]) -> dict[str, BusinessRole]:
+    """按编码批量取岗位（参数批次校验段用，消除逐条 N+1）。
+
+    返回 {编码: 岗位}。MySQL ci 校对下编码大小写变形也能命中，
+    键取库里的规范 role_code，调用方据此把维度值归一为规范值。
+    """
+    if not role_codes:
+        return {}
+    rows = session.execute(
+        select(BusinessRole).where(BusinessRole.role_code.in_(role_codes))
+    ).scalars()
+    return {role.role_code: role for role in rows}
+
+
 def list_roles(
     session: Session,
     *,

backend/app/repositories/param_repository.py

@@ -81,6 +81,21 @@ def get_definition_by_code(session: Session, param_code: str) -> ParamDefinition
     return session.execute(stmt).scalar_one_or_none()
 
 
+def list_definitions_by_codes(
+    session: Session, param_codes: list[str]
+) -> dict[str, ParamDefinition]:
+    """按编码批量取参数定义（提交批次的校验段用，消除逐条 N+1）。
+
+    返回 {编码: 定义}；未定义的编码不在结果中，由调用方判缺失。
+    """
+    if not param_codes:
+        return {}
+    rows = session.execute(
+        select(ParamDefinition).where(ParamDefinition.param_code.in_(param_codes))
+    ).scalars()
+    return {definition.param_code: definition for definition in rows}
+
+
 def list_history(
     session: Session,
     param_id: int,

backend/app/services/param_service.py

@@ -218,11 +218,28 @@ def submit_change(
         if payload.change_type == 2 and effective < date.today():
             raise BadRequestError("指定生效日期不得早于今日")
 
-        # 先全量校验、再写入：任何一项不合法整批拒绝（不留半批脏数据）
+        # 先全量校验、再写入：任何一项不合法整批拒绝（不留半批脏数据）。
+        # 定义与岗位各一条 IN 批量查询预载（逐条查询在批次扩大时是写侧 N+1）；
+        # 岗位预载收集全部非空维度值 —— 集合可能略大于实际所需，但省去按
+        # 维度类型先分类的第二轮遍历，无害多取由校验逻辑自行过滤
+        codes = list({item.param_code for item in payload.items})
+        definitions = param_repository.list_definitions_by_codes(session, codes)
+        dimension_values = list(
+            {item.dimension_value for item in payload.items if item.dimension_value}
+        )
+        # 预载字典键归一为 casefold：单条版 get_by_code 走 SQL ci 校对（大小写
+        # 变形可命中），批量版改字典查找后须自行等价归一，否则变形大小写会
+        # 被误判「不是已存在的岗位」；岗位编码均为 ascii 小写蛇形，casefold 足够
+        roles = {
+            code.casefold(): role
+            for code, role in business_role_repository.list_by_codes(
+                session, dimension_values
+            ).items()
+        }
         normalized: list[tuple[ParamDefinition, str, str]] = []
         seen: set[tuple[str, str]] = set()
         for item in payload.items:
-            definition = param_repository.get_definition_by_code(session, item.param_code)
+            definition = definitions.get(item.param_code)
             if definition is None:
                 # 用户输入先行查库给 400；未注册编码出现在库里属数据与注册表漂移
                 raise BadRequestError(f"参数 {item.param_code} 未在系统中定义")
@@ -234,7 +251,7 @@ def submit_change(
             if spec.dimension_type == "NONE" and dimension != "":
                 raise BadRequestError(f"参数 {item.param_code} 不带维度，维度值须为空")
             if spec.dimension_type == "BUSINESS_ROLE" and dimension:
-                role = business_role_repository.get_by_code(session, dimension)
+                role = roles.get(dimension.casefold())
                 if role is None:
                     raise BadRequestError(
                         f"参数 {item.param_code} 的维度值 {dimension!r} 不是已存在的岗位"

backend/app/services/receipt_service.py

@@ -284,8 +284,11 @@ def update_receipt(
         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)
+        # 校验返回的 (shipment_no, ship_date) 即新批次 DTO 所需 —— shipment_id
+        # 已被赋为 payload 新值，与重查 get_with_refs 同源，省一次三表 join
+        shipment_no, ship_date = _ensure_shipment_of_business(
+            session, payload.shipment_id, receipt.business_id
+        )
 
         receipt.receipt_date = payload.receipt_date
         receipt.shipment_id = payload.shipment_id
@@ -293,7 +296,7 @@ def update_receipt(
         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))
+        out = _to_out((receipt, business.business_no, shipment_no, ship_date))
         batch_no = receipt.receipt_batch_no
         session.commit()
         # commit 后不再触碰 ORM 属性（audit_service 约定）：批次号已固化

backend/app/services/shipment_service.py

@@ -317,7 +317,8 @@ def update_shipment(
         total = shipment_repository.sum_item_amount(session, shipment_id)
         shipment.total_amount = total if total is not None else Decimal("0")
         _recalc_totals(session, business)
-        out = _to_out(shipment_repository.get_with_business_no(session, shipment_id))
+        # business_id 创建时定死不可改挂，锁定的 business.business_no 与重查同源
+        out = _to_out((shipment, business.business_no))
         shipment_no = shipment.shipment_no
         session.commit()
         # commit 后不再触碰 ORM 属性（audit_service 约定）：单号已固化