From 68564c61f0a65cea581e6015783dbc76d1f5d8c3 Mon Sep 17 00:00:00 2001 From: zlwu Date: Sat, 5 Sep 2026 18:40:29 +0800 Subject: [PATCH] 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 --- backend/app/api/pl/plan.py | 75 ++- backend/app/api/pl/transport.py | 29 +- .../app/services/pl_dibang_bridge_service.py | 446 ++++++++++++++++-- .../tests/test_pl_dibang_upsert_on_arrange.py | 280 +++++++++++ 4 files changed, 749 insertions(+), 81 deletions(-) create mode 100644 backend/tests/test_pl_dibang_upsert_on_arrange.py diff --git a/backend/app/api/pl/plan.py b/backend/app/api/pl/plan.py index 3765b5e62..7dc654f1c 100644 --- a/backend/app/api/pl/plan.py +++ b/backend/app/api/pl/plan.py @@ -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( diff --git a/backend/app/api/pl/transport.py b/backend/app/api/pl/transport.py index e630be99a..3ad62f15e 100644 --- a/backend/app/api/pl/transport.py +++ b/backend/app/api/pl/transport.py @@ -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 diff --git a/backend/app/services/pl_dibang_bridge_service.py b/backend/app/services/pl_dibang_bridge_service.py index 0a90a2275..a9bb5dd9f 100644 --- a/backend/app/services/pl_dibang_bridge_service.py +++ b/backend/app/services/pl_dibang_bridge_service.py @@ -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") - if not business_type: - raise ValueError("装货段未配置地磅方向，无法生成地磅单") - stop_products = _load_stop_products(db, tenant_no, load.id) - if not stop_products: - raise ValueError("装货段无产品明细，无法生成地磅单") - partner = resolve_stop_dibang_partner("load", load, deliveries) - if not partner: - raise ValueError("无法解析装货段地磅往来单位，请先配置送货客户") - results.extend( - _sync_stop_inventory_links( - db, - tenant_no=tenant_no, - plan=plan, - transport=transport, - stop_type="load", - stop_id=load.id, - partner=partner, - business_type=business_type, - 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") + 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(f"送货段【{delivery.customer_name or delivery.seq}】未配置地磅方向，无法生成地磅单") - stop_products = _delivery_stop_products(db, tenant_no, delivery.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(f"送货段【{delivery.customer_name or delivery.seq}】无产品明细，无法生成地磅单") - partner = resolve_stop_dibang_partner("delivery", 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="delivery", - stop_id=delivery.id, + stop_type=stop_type, + stop_id=stop.id, partner=partner, business_type=business_type, - dibang_date=_resolve_stop_dibang_date( - plan, - stop_type="delivery", - delivery_time=delivery.delivery_time, - ), + 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, + ), + ) + + for delivery in deliveries: + _handle_stop( + stop_type="delivery", + 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, + ), + ) + + 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 diff --git a/backend/tests/test_pl_dibang_upsert_on_arrange.py b/backend/tests/test_pl_dibang_upsert_on_arrange.py new file mode 100644 index 000000000..c26a7db53 --- /dev/null +++ b/backend/tests/test_pl_dibang_upsert_on_arrange.py @@ -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()