feat(pl): upsert dibang links by stop instead of revoke-and-rebuild.

Keep inventory ids across rearrange/plan edits; block cancel when weighing exists. Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-05 18:40:29 +08:00
1 parent c90ff7c419
commit 68564c61f0
@@ -30,6 +30,8 @@ from app.models.pl_models import (
    PlCarrier,
    PlDriver,
    PlPlan,
+    PlPlanDelivery,
+    PlPlanLoad,
    PlPlanProduct,
    PlPlanTransport,
    PlTransport,
@@ -84,8 +86,10 @@ from app.schemas.pl.transport import (
    TransportSiteLinkRead,
 )
 from app.services.pl_dibang_bridge_service import (
+    StopDibangOption,
    revoke_regeneratable_dibang_for_transport,
    sync_linked_dibang_dates_for_plan,
+    upsert_stop_dibang_links,
 )
 from app.services.pl_inventory_link_read import build_transport_inventory_link_read
 from app.services.pl_io_order_service import get_active_io_order_for_plan, void_io_orders_for_plan
@@ -1792,13 +1796,80 @@ def update_plan(
        user_id=user.id,
        dept_id=user.dept_id or "",
    )
+    dibang_upsert_warnings: list[str] = []
+    if (
+        plan.transport_status == "arranged"
+        and plan.transport_id
+        and multi_stop
+        and load_body is not None
+    ):
+        transport_row = (
+            db.query(PlTransport)
+            .filter(
+                PlTransport.id == plan.transport_id,
+                PlTransport.tenant_no == tenant_no,
+                PlTransport.del_flag == 0,
+            )
+            .first()
+        )
+        if transport_row:
+            stop_options: list[StopDibangOption] = []
+            load_row = (
+                db.query(PlPlanLoad)
+                .filter(
+                    PlPlanLoad.plan_id == id,
+                    PlPlanLoad.tenant_no == tenant_no,
+                    PlPlanLoad.del_flag == 0,
+                )
+                .first()
+            )
+            if load_row:
+                stop_options.append(
+                    StopDibangOption(
+                        stop_type="load",
+                        stop_id=load_row.id,
+                        generate_dibang=bool(load_row.generate_dibang),
+                    )
+                )
+            delivery_rows = (
+                db.query(PlPlanDelivery)
+                .filter(
+                    PlPlanDelivery.plan_id == id,
+                    PlPlanDelivery.tenant_no == tenant_no,
+                    PlPlanDelivery.del_flag == 0,
+                )
+                .order_by(PlPlanDelivery.seq.asc(), PlPlanDelivery.create_time.asc())
+                .all()
+            )
+            for delivery_row in delivery_rows:
+                stop_options.append(
+                    StopDibangOption(
+                        stop_type="delivery",
+                        stop_id=delivery_row.id,
+                        generate_dibang=bool(delivery_row.generate_dibang),
+                    )
+                )
+            if stop_options:
+                try:
+                    _link_results, dibang_upsert_warnings = upsert_stop_dibang_links(
+                        db,
+                        tenant_no=tenant_no,
+                        plan=plan,
+                        transport=transport_row,
+                        stop_options=stop_options,
+                        user_id=user.id,
+                        dept_id=user.dept_id or "",
+                    )
+                except ValueError as exc:
+                    db.rollback()
+                    raise HTTPException(status_code=400, detail=str(exc)) from exc
    db.commit()
    db.refresh(plan)
    result = PlanWithProducts.model_validate(plan)
    result.products = plan_products
    result.supplier_contracts = supplier_contracts
    result.demander_contracts = demander_contracts
-    result.warnings = [*qty_warnings, *dibang_date_warnings]
+    result.warnings = [*qty_warnings, *dibang_date_warnings, *dibang_upsert_warnings]
    stops = load_plan_stops_grouped(db, tenant_no, [id]).get(id, (None, []))
    attach_plan_stops_to_item(result, plan, stops)
    return result
@@ -1844,6 +1915,8 @@ def patch_plan_hazard_cert_ready(
    user: DbUserContext = Depends(get_db_current_user),
 ) -> PlanWithProducts:
    """标记或取消危化品证书齐备（人工确认）"""
+    if not user_can_manage_plan_cert_ready(db, user):
+        raise HTTPException(status_code=403, detail="无证书齐备操作权限")
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    _get_plan(db, tenant_no, id, user)
    set_hazard_cert_ready(
@@ -56,7 +56,6 @@ from app.schemas.pl.transport import (
 from app.services.pl_dibang_bridge_service import (
    StopDibangOption,
    plan_has_load_stop,
-    revoke_regeneratable_dibang_for_transport,
    sync_dibang_for_plan_stops,
    sync_dibang_for_transport,
 )
@@ -391,19 +390,6 @@ def _is_rearrange(plan: PlPlan, existing_transport: Optional[PlTransport]) -> bo
    return plan.transport_status == "arranged"


-def _transport_has_db_links(db: Session, tenant_no: str, transport_id: str) -> bool:
-    return (
-        db.query(PlTransportDbLink.id)
-        .filter(
-            PlTransportDbLink.tenant_no == tenant_no,
-            PlTransportDbLink.transport_id == transport_id,
-            PlTransportDbLink.del_flag == 0,
-        )
-        .first()
-        is not None
-    )
-
-
 def _apply_plan_logistics_from_payload(plan: PlPlan, payload: TransportArrange) -> None:
    if payload.pickup_address is not None:
        plan.pickup_address = payload.pickup_address
@@ -817,20 +803,7 @@ def arrange_transport(
    _apply_plan_logistics_from_payload(plan, payload)
    start_point, end_point = _resolve_transport_endpoints(plan, payload)

-    if (
-        is_rearrange
-        and existing_transport
-        and _transport_has_db_links(db, tenant_no, existing_transport.id)
-    ):
-        try:
-            revoke_regeneratable_dibang_for_transport(
-                db,
-                tenant_no=tenant_no,
-                transport_id=existing_transport.id,
-                user_id=user.id,
-            )
-        except ValueError as exc:
-            raise HTTPException(status_code=400, detail=str(exc)) from exc
+    # 地磅关联改为按段 upsert，不再在调整前整批 revoke 再建

    if existing_transport and existing_transport.transport_status != "cancelled":
        transport = existing_transport
@@ -515,6 +515,260 @@ def _inventory_date_editable(db: Session, inventory: DbInboundOutbound) -> bool:
    return True


+def _inventory_mutable_editable(db: Session, inventory: DbInboundOutbound) -> bool:
+    """未过磅且状态可作废时，可就地更新日期/往来/车牌/数量。"""
+    return _inventory_date_editable(db, inventory)
+
+
+def _find_active_stop_link(
+    db: Session,
+    *,
+    tenant_no: str,
+    transport_id: str,
+    plan_id: str,
+    stop_type: str,
+    stop_id: str,
+    plan_product_id: str,
+) -> Optional[PlTransportDbLink]:
+    return (
+        db.query(PlTransportDbLink)
+        .filter(
+            PlTransportDbLink.tenant_no == tenant_no,
+            PlTransportDbLink.transport_id == transport_id,
+            PlTransportDbLink.plan_id == plan_id,
+            PlTransportDbLink.stop_type == stop_type,
+            PlTransportDbLink.stop_id == stop_id,
+            PlTransportDbLink.plan_product_id == plan_product_id,
+            PlTransportDbLink.del_flag == 0,
+        )
+        .first()
+    )
+
+
+def _soft_delete_link_and_inventory(
+    db: Session,
+    *,
+    tenant_no: str,
+    link: PlTransportDbLink,
+    user_id: Optional[str],
+) -> None:
+    inventory = (
+        db.query(DbInboundOutbound)
+        .filter(
+            DbInboundOutbound.id == link.db_inventory_id,
+            DbInboundOutbound.tenant_no == tenant_no,
+            DbInboundOutbound.del_flag == 0,
+        )
+        .first()
+    )
+    if inventory:
+        _subtract_plan_contribution(
+            db,
+            tenant_no=tenant_no,
+            link=link,
+            inventory=inventory,
+            user_id=user_id,
+        )
+        inventory.del_flag = 1
+        inventory.update_by = user_id
+    link.del_flag = 1
+    link.update_by = user_id
+
+
+def _cancel_stop_dibang_links(
+    db: Session,
+    *,
+    tenant_no: str,
+    transport_id: str,
+    plan_id: str,
+    stop_type: str,
+    stop_id: str,
+    user_id: Optional[str],
+) -> None:
+    """取消段 generate_dibang：已过磅阻断；否则软删 link+出入库并回减月计划。"""
+    links = (
+        db.query(PlTransportDbLink)
+        .filter(
+            PlTransportDbLink.tenant_no == tenant_no,
+            PlTransportDbLink.transport_id == transport_id,
+            PlTransportDbLink.plan_id == plan_id,
+            PlTransportDbLink.stop_type == stop_type,
+            PlTransportDbLink.stop_id == stop_id,
+            PlTransportDbLink.del_flag == 0,
+        )
+        .all()
+    )
+    if not links:
+        return
+
+    blocking: list[str] = []
+    for link in links:
+        inventory = (
+            db.query(DbInboundOutbound)
+            .filter(
+                DbInboundOutbound.id == link.db_inventory_id,
+                DbInboundOutbound.tenant_no == tenant_no,
+                DbInboundOutbound.del_flag == 0,
+            )
+            .first()
+        )
+        if not inventory:
+            continue
+        label = inventory.batch_no or inventory.id
+        status = (inventory.status or "").strip()
+        if status and status not in _REVOKABLE_INVENTORY_STATUSES:
+            blocking.append(f"{label}（状态：{status}）")
+            continue
+        if _inventory_has_weighing_slip(db, inventory.id):
+            blocking.append(label)
+
+    if blocking:
+        raise ValueError(
+            "以下出入库单已执行或已关联地磅，无法取消生成地磅："
+            + "；".join(blocking)
+            + "。请在地磅管理中处理后再操作。"
+        )
+
+    for link in links:
+        _soft_delete_link_and_inventory(
+            db,
+            tenant_no=tenant_no,
+            link=link,
+            user_id=user_id,
+        )
+    db.flush()
+
+
+def _update_existing_stop_inventory(
+    db: Session,
+    *,
+    tenant_no: str,
+    plan: PlPlan,
+    link: PlTransportDbLink,
+    inventory: DbInboundOutbound,
+    product: _StopDibangProduct,
+    partner: str,
+    license_plate: str,
+    dibang_date: date,
+    user_id: Optional[str],
+    dept_id: str,
+    warnings: list[str],
+) -> DibangInventoryLinkResult:
+    """就地更新已有出入库可变字段；不可编辑时写 warnings，不抛错。"""
+    inv_product_id = (inventory.product_id or "").strip()
+    new_product_id = (product.product_id or "").strip()
+    if inv_product_id and new_product_id and inv_product_id != new_product_id:
+        label = inventory.batch_no or inventory.id
+        raise ValueError(
+            f"出入库单 {label} 已关联产品，禁止更换为其他产品。请取消地磅后重新勾选，或保留原产品。"
+        )
+
+    label = inventory.batch_no or inventory.id
+    editable = _inventory_mutable_editable(db, inventory)
+
+    if not editable:
+        status = (inventory.status or "").strip() or "未知状态"
+        if _inventory_has_weighing_slip(db, inventory.id):
+            reason = "已关联地磅单"
+        else:
+            reason = f"状态：{status}"
+        warnings.append(
+            f"出入库单 {label}（{reason}）未同步往来单位/车牌/数量等可变字段，请在地磅管理中手工处理"
+        )
+        # 备注中计划编号文案仍可校正
+        desired_remark = f"运输计划 {plan.plan_no}"
+        if (inventory.remark or "").strip() != desired_remark:
+            inventory.remark = desired_remark
+            inventory.update_by = user_id
+        if (link.remark or "").strip() != desired_remark:
+            link.remark = desired_remark
+            link.update_by = user_id
+        return DibangInventoryLinkResult(
+            plan_product_id=product.plan_product_id,
+            product_name=product.product_name or "",
+            db_purchase_plan_id=link.db_purchase_plan_id or "",
+            db_inventory_id=inventory.id,
+            batch_no=inventory.batch_no or "",
+            plan_no=(inventory.plan_no or plan.plan_no) or "",
+        )
+
+    # —— 可编辑：更新 partner / vehicle / quantity / date（跨月改挂）——
+    if partner and (inventory.partner_company or "").strip() != partner:
+        inventory.partner_company = partner
+
+    plate = (license_plate or "").strip()
+    if plate and (inventory.vehicle_no or "").strip() != plate:
+        inventory.vehicle_no = plate
+
+    new_qty = float(product.quantity)
+    old_qty = _resolve_link_contributed_qty(link, inventory)
+    if abs(new_qty - old_qty) > 1e-9:
+        delta = new_qty - old_qty
+        if link.db_purchase_plan_id:
+            db_plan = (
+                db.query(DbPurchasePlan)
+                .filter(
+                    DbPurchasePlan.id == link.db_purchase_plan_id,
+                    DbPurchasePlan.tenant_no == tenant_no,
+                    DbPurchasePlan.del_flag == 0,
+                )
+                .first()
+            )
+            if db_plan:
+                _apply_plan_quantity_delta(db_plan, delta, user_id=user_id)
+        inventory.quantity = Decimal(str(new_qty))
+        link.contributed_qty = Decimal(str(new_qty))
+
+    old_date = _normalize_to_date(inventory.date)
+    if old_date != dibang_date:
+        same_month = (
+            old_date is not None
+            and old_date.year == dibang_date.year
+            and old_date.month == dibang_date.month
+        )
+        inventory.date = dibang_date
+        if not same_month:
+            qty = _resolve_link_contributed_qty(link, inventory)
+            _subtract_plan_contribution(
+                db,
+                tenant_no=tenant_no,
+                link=link,
+                inventory=inventory,
+                user_id=user_id,
+            )
+            try:
+                new_db_plan = _ensure_monthly_db_plan_for_inventory(
+                    db,
+                    tenant_no=tenant_no,
+                    plan=plan,
+                    inventory=inventory,
+                    dibang_date=dibang_date,
+                    contributed_qty=qty,
+                    user_id=user_id,
+                    dept_id=dept_id or "",
+                )
+                link.db_purchase_plan_id = new_db_plan.id
+                inventory.purchase_plan_id = new_db_plan.id
+                inventory.plan_no = (new_db_plan.plan_no or "").strip() or inventory.plan_no
+            except ValueError as exc:
+                warnings.append(f"出入库单 {label} 跨月改挂失败：{exc}")
+
+    desired_remark = f"运输计划 {plan.plan_no}"
+    inventory.remark = desired_remark
+    link.remark = desired_remark
+    inventory.update_by = user_id
+    link.update_by = user_id
+
+    return DibangInventoryLinkResult(
+        plan_product_id=product.plan_product_id,
+        product_name=product.product_name or "",
+        db_purchase_plan_id=link.db_purchase_plan_id or "",
+        db_inventory_id=inventory.id,
+        batch_no=inventory.batch_no or "",
+        plan_no=(inventory.plan_no or plan.plan_no) or "",
+    )
+
+
 def _resolve_link_dibang_date(
    plan: PlPlan,
    link: PlTransportDbLink,
@@ -728,13 +982,16 @@ def _sync_stop_inventory_links(
    stop_products: list[_StopDibangProduct],
    user_id: Optional[str],
    dept_id: str,
+    warnings: Optional[list[str]] = None,
 ) -> list[DibangInventoryLinkResult]:
+    """按段产品 upsert 出入库+link：已有则就地更新，无则新建。"""
    if not partner:
        raise ValueError("无法解析往来单位，请检查段客户信息")

    month_start, month_end = _month_range(dibang_date)
    license_plate = (transport.license_plate or plan.license_plate or "").strip()
    results: list[DibangInventoryLinkResult] = []
+    warn_buf = warnings if warnings is not None else []

    for product in stop_products:
        if not product.product_id:
@@ -744,6 +1001,47 @@ def _sync_stop_inventory_links(
        if contributed_qty <= 0:
            continue

+        existing_link = _find_active_stop_link(
+            db,
+            tenant_no=tenant_no,
+            transport_id=transport.id,
+            plan_id=plan.id,
+            stop_type=stop_type,
+            stop_id=stop_id,
+            plan_product_id=product.plan_product_id,
+        )
+        if existing_link:
+            inventory = (
+                db.query(DbInboundOutbound)
+                .filter(
+                    DbInboundOutbound.id == existing_link.db_inventory_id,
+                    DbInboundOutbound.tenant_no == tenant_no,
+                    DbInboundOutbound.del_flag == 0,
+                )
+                .first()
+            )
+            if inventory:
+                results.append(
+                    _update_existing_stop_inventory(
+                        db,
+                        tenant_no=tenant_no,
+                        plan=plan,
+                        link=existing_link,
+                        inventory=inventory,
+                        product=product,
+                        partner=partner,
+                        license_plate=license_plate,
+                        dibang_date=dibang_date,
+                        user_id=user_id,
+                        dept_id=dept_id,
+                        warnings=warn_buf,
+                    )
+                )
+                continue
+            # link 悬空：软删后走新建
+            existing_link.del_flag = 1
+            existing_link.update_by = user_id
+
        plan_product = PlPlanProduct(
            id=product.plan_product_id,
            product_id=product.product_id,
@@ -899,7 +1197,7 @@ def _delivery_stop_products(
    ]


-def sync_dibang_for_plan_stops(
+def upsert_stop_dibang_links(
    db: Session,
    *,
    tenant_no: str,
@@ -908,8 +1206,11 @@ def sync_dibang_for_plan_stops(
    stop_options: list[StopDibangOption],
    user_id: Optional[str],
    dept_id: str,
-) -> list[DibangInventoryLinkResult]:
-    """按装货/送货段勾选生成地磅计划与出入库单。"""
+) -> tuple[list[DibangInventoryLinkResult], list[str]]:
+    """按段 upsert 地磅出入库与 link：新建 / 就地更新 / 取消勾选软删。
+
+    返回 (link_results, warnings)。已过磅取消勾选或产品变更时抛 ValueError。
+    """
    apply_stop_dibang_options(
        db,
        tenant_no=tenant_no,
@@ -920,6 +1221,7 @@ def sync_dibang_for_plan_stops(

    option_map = {(item.stop_type, item.stop_id): item for item in stop_options}
    results: list[DibangInventoryLinkResult] = []
+    warnings: list[str] = []

    deliveries = (
        db.query(PlPlanDelivery)
@@ -941,73 +1243,113 @@ def sync_dibang_for_plan_stops(
        )
        .first()
    )
-    if load:
-        option = option_map.get(("load", load.id))
-        if option and option.generate_dibang:
-            business_type = _resolve_stop_business_type(load, "load")
+
+    def _handle_stop(
+        *,
+        stop_type: Literal["load", "delivery"],
+        stop: PlPlanLoad | PlPlanDelivery,
+        partner: str,
+        dibang_date: date,
+    ) -> None:
+        option = option_map.get((stop_type, stop.id))
+        if option is None:
+            return
+        if not option.generate_dibang:
+            _cancel_stop_dibang_links(
+                db,
+                tenant_no=tenant_no,
+                transport_id=transport.id,
+                plan_id=plan.id,
+                stop_type=stop_type,
+                stop_id=stop.id,
+                user_id=user_id,
+            )
+            return
+
+        if stop_type == "load":
+            stop_products = _load_stop_products(db, tenant_no, stop.id)
+        else:
+            stop_products = _delivery_stop_products(db, tenant_no, stop.id)
+
+        business_type = _resolve_stop_business_type(stop, stop_type)
        if not business_type:
-                raise ValueError("装货段未配置地磅方向，无法生成地磅单")
-            stop_products = _load_stop_products(db, tenant_no, load.id)
+            stop_label = getattr(stop, "customer_name", None) or stop.id
+            raise ValueError(f"{'装货' if stop_type == 'load' else '送货'}段【{stop_label}】未配置地磅方向，无法生成地磅单")
        if not stop_products:
-                raise ValueError("装货段无产品明细，无法生成地磅单")
-            partner = resolve_stop_dibang_partner("load", load, deliveries)
+            stop_label = getattr(stop, "customer_name", None) or stop.id
+            raise ValueError(f"{'装货' if stop_type == 'load' else '送货'}段【{stop_label}】无产品明细，无法生成地磅单")
        if not partner:
-                raise ValueError("无法解析装货段地磅往来单位，请先配置送货客户")
+            raise ValueError(
+                "无法解析装货段地磅往来单位，请先配置送货客户"
+                if stop_type == "load"
+                else "无法解析送货段地磅往来单位，请先配置装货客户"
+            )
        results.extend(
            _sync_stop_inventory_links(
                db,
                tenant_no=tenant_no,
                plan=plan,
                transport=transport,
-                    stop_type="load",
-                    stop_id=load.id,
+                stop_type=stop_type,
+                stop_id=stop.id,
                partner=partner,
                business_type=business_type,
+                dibang_date=dibang_date,
+                stop_products=stop_products,
+                user_id=user_id,
+                dept_id=dept_id,
+                warnings=warnings,
+            )
+        )
+
+    if load:
+        _handle_stop(
+            stop_type="load",
+            stop=load,
+            partner=resolve_stop_dibang_partner("load", load, deliveries),
            dibang_date=_resolve_stop_dibang_date(
                plan,
                stop_type="load",
                load_time=load.load_time,
            ),
-                    stop_products=stop_products,
-                    user_id=user_id,
-                    dept_id=dept_id,
-                )
        )

    for delivery in deliveries:
-        option = option_map.get(("delivery", delivery.id))
-        if not option or not option.generate_dibang:
-            continue
-        business_type = _resolve_stop_business_type(delivery, "delivery")
-        if not business_type:
-            raise ValueError(f"送货段【{delivery.customer_name or delivery.seq}】未配置地磅方向，无法生成地磅单")
-        stop_products = _delivery_stop_products(db, tenant_no, delivery.id)
-        if not stop_products:
-            raise ValueError(f"送货段【{delivery.customer_name or delivery.seq}】无产品明细，无法生成地磅单")
-        partner = resolve_stop_dibang_partner("delivery", load, deliveries)
-        if not partner:
-            raise ValueError("无法解析送货段地磅往来单位，请先配置装货客户")
-        results.extend(
-            _sync_stop_inventory_links(
-                db,
-                tenant_no=tenant_no,
-                plan=plan,
-                transport=transport,
+        _handle_stop(
            stop_type="delivery",
-                stop_id=delivery.id,
-                partner=partner,
-                business_type=business_type,
+            stop=delivery,
+            partner=resolve_stop_dibang_partner("delivery", load, deliveries),
            dibang_date=_resolve_stop_dibang_date(
                plan,
                stop_type="delivery",
                delivery_time=delivery.delivery_time,
            ),
-                stop_products=stop_products,
+        )
+
+    db.flush()
+    return results, warnings
+
+
+def sync_dibang_for_plan_stops(
+    db: Session,
+    *,
+    tenant_no: str,
+    plan: PlPlan,
+    transport: PlTransport,
+    stop_options: list[StopDibangOption],
+    user_id: Optional[str],
+    dept_id: str,
+) -> list[DibangInventoryLinkResult]:
+    """按装货/送货段勾选 upsert 地磅计划与出入库单（兼容旧调用方，忽略 warnings）。"""
+    results, _warnings = upsert_stop_dibang_links(
+        db,
+        tenant_no=tenant_no,
+        plan=plan,
+        transport=transport,
+        stop_options=stop_options,
        user_id=user_id,
        dept_id=dept_id,
    )
-        )
-
    return results


@@ -0,0 +1,280 @@
+"""地磅按段 upsert — 取消勾选阻断 / 复用出入库 id"""
+from datetime import date, datetime
+from decimal import Decimal
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from app.models.db_models import DbInboundOutbound, DbPurchasePlan
+from app.models.pl_models import (
+    PlPlan,
+    PlPlanDelivery,
+    PlPlanDeliveryProduct,
+    PlPlanLoad,
+    PlPlanLoadProduct,
+    PlTransport,
+    PlTransportDbLink,
+)
+from app.services.pl_dibang_bridge_service import (
+    StopDibangOption,
+    upsert_stop_dibang_links,
+)
+
+
+def _make_query_router(mapping):
+    def query_router(model):
+        mock = MagicMock()
+        handler = mapping.get(model)
+        if handler is None:
+            return MagicMock()
+        return handler(mock)
+
+    return query_router
+
+
+@patch("app.services.pl_dibang_bridge_service._inventory_has_weighing_slip", return_value=True)
+def test_cancel_generate_dibang_blocked_when_weighing_exists(_mock_weighing):
+    plan = PlPlan(
+        id="plan-1",
+        plan_no="PO1",
+        plan_type="outbound",
+        plan_date=date(2026, 9, 1),
+        tenant_no="t1",
+    )
+    transport = PlTransport(id="tr-1", tenant_no="t1", license_plate="皖A12345", del_flag=0)
+    load = PlPlanLoad(
+        id="load-1",
+        plan_id="plan-1",
+        customer_name="客户A",
+        generate_dibang=1,
+        dibang_direction="outbound",
+        load_time=datetime(2026, 9, 1, 0, 0),
+        tenant_no="t1",
+        del_flag=0,
+    )
+    inventory = DbInboundOutbound(
+        id="inv-1",
+        tenant_no="t1",
+        date=date(2026, 9, 1),
+        partner_company="客户A",
+        business_type="出库",
+        product_id="prod-1",
+        product="硫酸",
+        quantity=30.0,
+        batch_no="CRKDH0002400",
+        status="待出库",
+        del_flag=0,
+    )
+    link = PlTransportDbLink(
+        id="link-1",
+        tenant_no="t1",
+        plan_id="plan-1",
+        transport_id="tr-1",
+        plan_product_id="pp-1",
+        stop_type="load",
+        stop_id="load-1",
+        db_purchase_plan_id="dbp-1",
+        db_inventory_id="inv-1",
+        contributed_qty=Decimal("30"),
+        del_flag=0,
+    )
+
+    db = MagicMock()
+
+    def link_q(m):
+        filtered = MagicMock()
+        filtered.all.return_value = [link]
+        filtered.first.return_value = link
+        m.filter.return_value = filtered
+        return m
+
+    def load_q(m):
+        m.filter.return_value.first.return_value = load
+        return m
+
+    def delivery_q(m):
+        m.filter.return_value.order_by.return_value.all.return_value = []
+        return m
+
+    def inv_q(m):
+        m.filter.return_value.first.return_value = inventory
+        return m
+
+    db.query.side_effect = _make_query_router(
+        {
+            PlTransportDbLink: link_q,
+            PlPlanLoad: load_q,
+            PlPlanDelivery: delivery_q,
+            DbInboundOutbound: inv_q,
+        }
+    )
+
+    with pytest.raises(ValueError, match="CRKDH0002400"):
+        upsert_stop_dibang_links(
+            db,
+            tenant_no="t1",
+            plan=plan,
+            transport=transport,
+            stop_options=[
+                StopDibangOption(stop_type="load", stop_id="load-1", generate_dibang=False),
+            ],
+            user_id="u1",
+            dept_id="d1",
+        )
+
+
+@patch("app.services.pl_dibang_bridge_service._inventory_has_weighing_slip", return_value=False)
+@patch("app.services.pl_dibang_bridge_service.find_monthly_db_plan")
+@patch("app.services.pl_dibang_bridge_service._create_db_inventory")
+@patch("app.services.pl_dibang_bridge_service.resolve_crm_product")
+def test_upsert_reuses_existing_inventory_id(
+    _mock_crm,
+    mock_create_inv,
+    mock_find_plan,
+    _mock_weighing,
+):
+    plan = PlPlan(
+        id="plan-1",
+        plan_no="PO1",
+        plan_type="outbound",
+        plan_date=date(2026, 9, 1),
+        pickup_time=datetime(2026, 9, 1, 0, 0),
+        license_plate="皖A12345",
+        tenant_no="t1",
+    )
+    transport = PlTransport(id="tr-1", tenant_no="t1", license_plate="皖A12345", del_flag=0)
+    load = PlPlanLoad(
+        id="load-1",
+        plan_id="plan-1",
+        customer_name="客户A",
+        generate_dibang=1,
+        dibang_direction="outbound",
+        load_time=datetime(2026, 9, 1, 0, 0),
+        tenant_no="t1",
+        del_flag=0,
+    )
+    delivery = PlPlanDelivery(
+        id="del-1",
+        plan_id="plan-1",
+        customer_name="客户B",
+        seq=1,
+        generate_dibang=0,
+        tenant_no="t1",
+        del_flag=0,
+    )
+    stop_product = PlPlanLoadProduct(
+        id="lp-1",
+        load_id="load-1",
+        plan_product_id="pp-1",
+        product_id="prod-1",
+        product_name="硫酸",
+        model="98%",
+        unit="吨",
+        plan_qty=Decimal("30"),
+        tenant_no="t1",
+        del_flag=0,
+        order_num=1,
+    )
+    db_plan = DbPurchasePlan(
+        id="dbp-1",
+        tenant_no="t1",
+        plan_no="SX-1",
+        planned_quantity=30.0,
+        del_flag=0,
+    )
+    inventory = DbInboundOutbound(
+        id="inv-1",
+        tenant_no="t1",
+        date=date(2026, 9, 1),
+        partner_company="客户B",
+        business_type="出库",
+        product_id="prod-1",
+        product="硫酸",
+        specification="98%",
+        quantity=30.0,
+        batch_no="CRKDH0000001",
+        vehicle_no="皖A12345",
+        status="待出库",
+        purchase_plan_id="dbp-1",
+        plan_no="SX-1",
+        del_flag=0,
+    )
+    link = PlTransportDbLink(
+        id="link-1",
+        tenant_no="t1",
+        plan_id="plan-1",
+        transport_id="tr-1",
+        plan_product_id="pp-1",
+        stop_type="load",
+        stop_id="load-1",
+        db_purchase_plan_id="dbp-1",
+        db_inventory_id="inv-1",
+        contributed_qty=Decimal("30"),
+        del_flag=0,
+    )
+
+    mock_create_inv.side_effect = AssertionError("must reuse existing inventory")
+    mock_find_plan.return_value = db_plan
+
+    db = MagicMock()
+
+    def link_q(m):
+        filtered = MagicMock()
+        filtered.all.return_value = [link]
+        filtered.first.return_value = link
+        m.filter.return_value = filtered
+        return m
+
+    def load_q(m):
+        m.filter.return_value.first.return_value = load
+        return m
+
+    def delivery_q(m):
+        m.filter.return_value.order_by.return_value.all.return_value = [delivery]
+        return m
+
+    def load_prod_q(m):
+        m.filter.return_value.order_by.return_value.all.return_value = [stop_product]
+        return m
+
+    def delivery_prod_q(m):
+        m.filter.return_value.order_by.return_value.all.return_value = []
+        return m
+
+    def inv_q(m):
+        m.filter.return_value.first.return_value = inventory
+        return m
+
+    def plan_q(m):
+        m.filter.return_value.first.return_value = db_plan
+        return m
+
+    db.query.side_effect = _make_query_router(
+        {
+            PlTransportDbLink: link_q,
+            PlPlanLoad: load_q,
+            PlPlanDelivery: delivery_q,
+            PlPlanLoadProduct: load_prod_q,
+            PlPlanDeliveryProduct: delivery_prod_q,
+            DbInboundOutbound: inv_q,
+            DbPurchasePlan: plan_q,
+        }
+    )
+
+    results, warnings = upsert_stop_dibang_links(
+        db,
+        tenant_no="t1",
+        plan=plan,
+        transport=transport,
+        stop_options=[
+            StopDibangOption(stop_type="load", stop_id="load-1", generate_dibang=True),
+        ],
+        user_id="u1",
+        dept_id="d1",
+    )
+
+    assert warnings == []
+    assert len(results) == 1
+    assert results[0].db_inventory_id == "inv-1"
+    assert results[0].batch_no == "CRKDH0000001"
+    mock_create_inv.assert_not_called()