From 4f69a9b9042c3baaec80cad86946b25ce69b9995 Mon Sep 17 00:00:00 2001
From: zlwu <704541015@qq.com>
Date: Sun, 19 Jul 2026 16:39:20 +0800
Subject: [PATCH] =?UTF-8?q?feat(pl):=20=E5=87=BA=E5=85=A5=E5=BA=93?=
 =?UTF-8?q?=E5=8D=95=E8=AF=BB=E6=97=B6=20enrichment=20=E5=8F=96=E9=80=81?=
 =?UTF-8?q?=E8=B4=A7=E4=B8=8E=E4=BA=A7=E5=93=81=E6=89=A9=E5=B1=95=E5=AD=97?=
 =?UTF-8?q?=E6=AE=B5?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 backend/app/services/pl_io_order_service.py  | 519 +++++++++++++++++++
 backend/tests/test_pl_io_order_enrichment.py |  65 +++
 2 files changed, 584 insertions(+)
 create mode 100644 backend/app/services/pl_io_order_service.py
 create mode 100644 backend/tests/test_pl_io_order_enrichment.py

diff --git a/backend/app/services/pl_io_order_service.py b/backend/app/services/pl_io_order_service.py
new file mode 100644
index 000000000..601aed388
--- /dev/null
+++ b/backend/app/services/pl_io_order_service.py
@@ -0,0 +1,519 @@
+"""运输计划 · 入库单 / 出库单（安排运输落单、现场回传、作废）"""
+from __future__ import annotations
+
+import json
+import secrets
+from dataclasses import dataclass
+from datetime import date, datetime
+from decimal import Decimal
+from typing import Any, Optional, Union
+
+from sqlalchemy.orm import Session
+
+from app.core.config import settings
+from app.core.utils import generate_uuid
+from app.models.db_models import DbInboundOutbound
+from app.models.pl_models import (
+    PlInboundOrder,
+    PlInboundOrderItem,
+    PlOutboundOrder,
+    PlOutboundOrderItem,
+    PlPlan,
+    PlPlanProduct,
+    PlTransport,
+    PlTransportDbLink,
+)
+from app.services.db.mappers import parse_date_str
+from app.services.pl_dibang_bridge_service import _product_quantity, resolve_partner_company
+
+
+OrderHeader = Union[PlInboundOrder, PlOutboundOrder]
+OrderItem = Union[PlInboundOrderItem, PlOutboundOrderItem]
+
+
+@dataclass
+class IoOrderCreateResult:
+    kind: str  # inbound | outbound
+    order_id: str
+    order_no: str
+    site_access_token: str
+    site_url: str
+
+
+def build_io_order_site_url(kind: str, token: str) -> str:
+    origin = (settings.PUBLIC_WEB_ORIGIN or "").rstrip("/")
+    path = "inbound" if kind == "inbound" else "outbound"
+    return f"{origin}/#/public/pl/{path}/{token}"
+
+
+def _normalize_plan_date(plan: PlPlan) -> date:
+    plan_date = plan.plan_date
+    if isinstance(plan_date, datetime):
+        return plan_date.date()
+    if isinstance(plan_date, str):
+        return parse_date_str(plan_date) or date.today()
+    if isinstance(plan_date, date):
+        return plan_date
+    return date.today()
+
+
+def _next_order_no(db: Session, tenant_no: str, prefix: str, model: type) -> str:
+    day = datetime.now().strftime("%Y%m%d")
+    like = f"{prefix}{day}%"
+    last = (
+        db.query(model.order_no)
+        .filter(model.tenant_no == tenant_no, model.order_no.like(like), model.del_flag == 0)
+        .order_by(model.order_no.desc())
+        .first()
+    )
+    seq = 1
+    if last and last[0] and len(last[0]) >= len(prefix) + 8 + 4:
+        try:
+            seq = int(last[0][-4:]) + 1
+        except ValueError:
+            seq = 1
+    return f"{prefix}{day}{seq:04d}"
+
+
+def _dibang_qty_by_plan_product(
+    db: Session, tenant_no: str, transport_id: str
+) -> dict[str, Decimal]:
+    links = (
+        db.query(PlTransportDbLink)
+        .filter(
+            PlTransportDbLink.tenant_no == tenant_no,
+            PlTransportDbLink.transport_id == transport_id,
+            PlTransportDbLink.del_flag == 0,
+        )
+        .all()
+    )
+    if not links:
+        return {}
+    inv_ids = [x.db_inventory_id for x in links if x.db_inventory_id]
+    if not inv_ids:
+        return {}
+    inv_rows = (
+        db.query(DbInboundOutbound)
+        .filter(DbInboundOutbound.id.in_(inv_ids), DbInboundOutbound.del_flag == 0)
+        .all()
+    )
+    inv_map = {r.id: r for r in inv_rows}
+    out: dict[str, Decimal] = {}
+    for link in links:
+        inv = inv_map.get(link.db_inventory_id)
+        if not inv or inv.quantity is None:
+            continue
+        out[link.plan_product_id] = Decimal(str(inv.quantity))
+    return out
+
+
+def create_io_order_for_transport(
+    db: Session,
+    *,
+    tenant_no: str,
+    plan: PlPlan,
+    products: list[PlPlanProduct],
+    transport: PlTransport,
+    user_id: Optional[str],
+    dept_id: str,
+    after_dibang: bool = False,
+) -> Optional[IoOrderCreateResult]:
+    """安排运输后创建入库/出库单。直销返回 None。"""
+    if plan.plan_type not in ("inbound", "outbound"):
+        return None
+
+    kind = plan.plan_type
+    header_model = PlInboundOrder if kind == "inbound" else PlOutboundOrder
+    item_model = PlInboundOrderItem if kind == "inbound" else PlOutboundOrderItem
+    prefix = "RK" if kind == "inbound" else "CK"
+
+    partner = resolve_partner_company(plan, plan.plan_type)
+    if not partner:
+        raise ValueError("无法解析往来单位，请检查合同供方/需方信息")
+
+    token = secrets.token_urlsafe(32)[:64]
+    order_no = _next_order_no(db, tenant_no, prefix, header_model)
+    plan_date = _normalize_plan_date(plan)
+    license_plate = (transport.license_plate or plan.license_plate or "").strip()
+    dibang_map = _dibang_qty_by_plan_product(db, tenant_no, transport.id) if after_dibang else {}
+
+    header = header_model(
+        id=generate_uuid(),
+        order_no=order_no,
+        plan_id=plan.id,
+        transport_id=transport.id,
+        status="in_progress",
+        partner_company=partner,
+        vehicle_no=license_plate or None,
+        plan_no=(plan.plan_no or "").strip() or None,
+        plan_date=plan_date,
+        site_access_token=token,
+        create_by=user_id,
+        create_dept_id=dept_id or "",
+        tenant_no=tenant_no,
+    )
+    db.add(header)
+
+    for idx, product in enumerate(products):
+        plan_qty = Decimal(str(_product_quantity(product)))
+        actual = dibang_map.get(product.id)
+        qty_source = "dibang" if actual is not None else "plan"
+        if actual is None:
+            actual = plan_qty
+        db.add(
+            item_model(
+                id=generate_uuid(),
+                order_id=header.id,
+                plan_product_id=product.id,
+                product_id=product.product_id,
+                product_name=product.product_name,
+                specification=product.model,
+                unit=product.unit,
+                plan_qty=plan_qty,
+                actual_qty=actual,
+                qty_source=qty_source,
+                order_num=idx + 1,
+                create_by=user_id,
+                create_dept_id=dept_id or "",
+                tenant_no=tenant_no,
+            )
+        )
+
+    db.flush()
+    return IoOrderCreateResult(
+        kind=kind,
+        order_id=header.id,
+        order_no=order_no,
+        site_access_token=token,
+        site_url=build_io_order_site_url(kind, token),
+    )
+
+
+def apply_dibang_prefill_to_order(
+    db: Session,
+    *,
+    tenant_no: str,
+    plan: PlPlan,
+    transport: PlTransport,
+) -> None:
+    """地磅单生成后，用数量预填尚未被现场改过的明细 actual_qty。"""
+    if plan.plan_type not in ("inbound", "outbound"):
+        return
+    header_model = PlInboundOrder if plan.plan_type == "inbound" else PlOutboundOrder
+    item_model = PlInboundOrderItem if plan.plan_type == "inbound" else PlOutboundOrderItem
+    header = (
+        db.query(header_model)
+        .filter(
+            header_model.tenant_no == tenant_no,
+            header_model.plan_id == plan.id,
+            header_model.transport_id == transport.id,
+            header_model.status != "voided",
+            header_model.del_flag == 0,
+        )
+        .order_by(header_model.create_time.desc())
+        .first()
+    )
+    if not header:
+        return
+    dibang_map = _dibang_qty_by_plan_product(db, tenant_no, transport.id)
+    if not dibang_map:
+        return
+    items = (
+        db.query(item_model)
+        .filter(item_model.order_id == header.id, item_model.del_flag == 0)
+        .all()
+    )
+    for item in items:
+        qty = dibang_map.get(item.plan_product_id)
+        if qty is None:
+            continue
+        if item.qty_source == "manual":
+            continue
+        item.actual_qty = qty
+        item.qty_source = "dibang"
+    db.flush()
+
+
+def void_io_orders_for_plan(
+    db: Session,
+    *,
+    tenant_no: str,
+    plan_id: str,
+    user_id: Optional[str],
+) -> None:
+    for model in (PlInboundOrder, PlOutboundOrder):
+        rows = (
+            db.query(model)
+            .filter(
+                model.tenant_no == tenant_no,
+                model.plan_id == plan_id,
+                model.status != "voided",
+                model.del_flag == 0,
+            )
+            .all()
+        )
+        for row in rows:
+            row.status = "voided"
+            row.update_by = user_id
+    db.flush()
+
+
+def get_active_io_order_for_plan(
+    db: Session, tenant_no: str, plan: PlPlan
+) -> Optional[tuple[str, OrderHeader]]:
+    if plan.plan_type == "inbound":
+        row = (
+            db.query(PlInboundOrder)
+            .filter(
+                PlInboundOrder.tenant_no == tenant_no,
+                PlInboundOrder.plan_id == plan.id,
+                PlInboundOrder.del_flag == 0,
+            )
+            .order_by(PlInboundOrder.create_time.desc())
+            .first()
+        )
+        return ("inbound", row) if row else None
+    if plan.plan_type == "outbound":
+        row = (
+            db.query(PlOutboundOrder)
+            .filter(
+                PlOutboundOrder.tenant_no == tenant_no,
+                PlOutboundOrder.plan_id == plan.id,
+                PlOutboundOrder.del_flag == 0,
+            )
+            .order_by(PlOutboundOrder.create_time.desc())
+            .first()
+        )
+        return ("outbound", row) if row else None
+    return None
+
+
+def parse_feedback_images(raw: Optional[str]) -> list[str]:
+    if not raw:
+        return []
+    try:
+        parsed = json.loads(raw)
+        if isinstance(parsed, list):
+            return [str(item) for item in parsed if item]
+    except (TypeError, json.JSONDecodeError):
+        pass
+    return [part.strip() for part in raw.split(",") if part.strip()]
+
+
+def dump_feedback_images(images: list[str]) -> str:
+    return json.dumps(images, ensure_ascii=False)
+
+
+def apply_site_feedback(
+    db: Session,
+    *,
+    header: OrderHeader,
+    items: list[OrderItem],
+    item_qty: dict[str, Decimal],
+    remark: Optional[str],
+    images: list[str],
+    user_id: Optional[str] = None,
+) -> None:
+    if header.status == "voided":
+        raise ValueError("单据已作废，无法提交")
+    now = datetime.now()
+    for item in items:
+        if item.id not in item_qty:
+            continue
+        item.actual_qty = item_qty[item.id]
+        item.qty_source = "manual"
+        item.update_by = user_id
+    header.site_feedback_remark = (remark or "").strip() or None
+    header.site_feedback_images = dump_feedback_images(images) if images else header.site_feedback_images
+    header.site_feedback_time = now
+    if not header.site_first_submitted_at:
+        header.site_first_submitted_at = now
+        if header.status == "in_progress":
+            header.status = "completed"
+    header.update_by = user_id
+    db.flush()
+
+
+CRM_PRODUCT_UNIT = "crm_product_unit"
+
+PLAN_TYPE_BUSINESS_LABEL = {
+    "inbound": "入库",
+    "outbound": "出库",
+    "direct": "直销",
+}
+
+
+def resolve_unit_label(unit_map: dict[str, str], unit_code: Optional[str]) -> str:
+    code = (unit_code or "").strip()
+    if not code:
+        return "吨"
+    return unit_map.get(code, code)
+
+
+def merge_plan_logistics_into_order(data: dict, plan: Optional[PlPlan]) -> None:
+    if not plan:
+        data.setdefault("businessType", None)
+        data.setdefault("planRemark", None)
+        data.setdefault("pickupAddress", None)
+        data.setdefault("pickupLng", None)
+        data.setdefault("pickupLat", None)
+        data.setdefault("pickupTime", None)
+        data.setdefault("pickupNotes", None)
+        data.setdefault("destinationName", None)
+        data.setdefault("destinationAddress", None)
+        data.setdefault("destinationLng", None)
+        data.setdefault("destinationLat", None)
+        data.setdefault("deliveryTime", None)
+        data.setdefault("destinationNotes", None)
+        return
+    data["businessType"] = PLAN_TYPE_BUSINESS_LABEL.get(plan.plan_type, plan.plan_type)
+    data["planRemark"] = (plan.remark or "").strip() or None
+    data["pickupAddress"] = plan.pickup_address
+    data["pickupLng"] = plan.pickup_lng
+    data["pickupLat"] = plan.pickup_lat
+    data["pickupTime"] = plan.pickup_time
+    data["pickupNotes"] = plan.pickup_notes
+    data["destinationName"] = plan.destination_name
+    data["destinationAddress"] = plan.destination_address or plan.destination_name
+    data["destinationLng"] = plan.destination_lng
+    data["destinationLat"] = plan.destination_lat
+    data["deliveryTime"] = plan.delivery_time
+    data["destinationNotes"] = plan.destination_notes
+
+
+def merge_plan_product_into_item(
+    item: dict,
+    product: Optional[PlPlanProduct],
+    *,
+    unit_map: dict[str, str],
+) -> None:
+    if not product:
+        item.setdefault("productNo", None)
+        item.setdefault("packageSpec", None)
+        item.setdefault("remark", None)
+        item["unitLabel"] = resolve_unit_label(unit_map, item.get("unit"))
+        return
+    item["productNo"] = product.product_no
+    item["packageSpec"] = product.package_spec
+    item["remark"] = (product.remark or "").strip() or None
+    if not item.get("specification") and product.model:
+        item["specification"] = product.model
+    item["unitLabel"] = resolve_unit_label(unit_map, item.get("unit") or product.unit)
+
+
+def enrich_io_order_dict(
+    db: Session,
+    data: dict,
+    *,
+    tenant_no: str,
+    plan_id: Optional[str],
+    items: list[Any] | None,
+    public_unit_as_label: bool = False,
+) -> dict:
+    """读时附加计划取送货与产品扩展字段。"""
+    from app.utils.excel_export import load_dict_label_maps
+
+    plan = None
+    if plan_id:
+        plan = (
+            db.query(PlPlan)
+            .filter(PlPlan.id == plan_id, PlPlan.del_flag == 0)
+            .first()
+        )
+    merge_plan_logistics_into_order(data, plan)
+
+    unit_map = load_dict_label_maps(
+        db, [CRM_PRODUCT_UNIT], tenant_no=tenant_no
+    ).get(CRM_PRODUCT_UNIT, {})
+
+    raw_items = data.get("items")
+    if raw_items is None:
+        return data
+
+    product_ids = [
+        it.get("planProductId")
+        for it in raw_items
+        if it.get("planProductId")
+    ]
+    products_by_id: dict[str, PlPlanProduct] = {}
+    if product_ids:
+        rows = (
+            db.query(PlPlanProduct)
+            .filter(
+                PlPlanProduct.id.in_(product_ids),
+                PlPlanProduct.del_flag == 0,
+            )
+            .all()
+        )
+        products_by_id = {r.id: r for r in rows}
+
+    for it in raw_items:
+        pp = products_by_id.get(it.get("planProductId") or "")
+        merge_plan_product_into_item(it, pp, unit_map=unit_map)
+        if public_unit_as_label:
+            it["unit"] = it.get("unitLabel") or resolve_unit_label(unit_map, it.get("unit"))
+
+    return data
+
+
+def order_header_to_dict(
+    kind: str,
+    row: OrderHeader,
+    items: list[Any] | None = None,
+    *,
+    db: Session | None = None,
+    public_unit_as_label: bool = False,
+) -> dict:
+    token = row.site_access_token or ""
+    data = {
+        "id": row.id,
+        "orderNo": row.order_no,
+        "planId": row.plan_id,
+        "transportId": row.transport_id,
+        "status": row.status,
+        "partnerCompany": row.partner_company,
+        "vehicleNo": row.vehicle_no,
+        "planNo": row.plan_no,
+        "planDate": row.plan_date.isoformat() if row.plan_date else None,
+        "siteAccessToken": token,
+        "siteUrl": build_io_order_site_url(kind, token) if token else None,
+        "siteFeedbackRemark": row.site_feedback_remark,
+        "siteFeedbackImages": parse_feedback_images(row.site_feedback_images),
+        "siteFeedbackTime": row.site_feedback_time,
+        "siteFirstSubmittedAt": row.site_first_submitted_at,
+        "remark": row.remark,
+        "tenantNo": row.tenant_no,
+        "createTime": row.create_time,
+        "updateTime": row.update_time,
+        "kind": kind,
+    }
+    if items is not None:
+        data["items"] = [
+            {
+                "id": it.id,
+                "orderId": it.order_id,
+                "planProductId": it.plan_product_id,
+                "productId": it.product_id,
+                "productName": it.product_name,
+                "specification": it.specification,
+                "unit": it.unit,
+                "planQty": it.plan_qty,
+                "actualQty": it.actual_qty,
+                "qtySource": it.qty_source,
+            }
+            for it in items
+        ]
+        data["productSummary"] = "、".join(
+            (it.product_name or "").strip() for it in items if (it.product_name or "").strip()
+        )[:200]
+        data["actualQtyTotal"] = float(sum(Decimal(str(it.actual_qty or 0)) for it in items))
+    if db is not None:
+        enrich_io_order_dict(
+            db,
+            data,
+            tenant_no=row.tenant_no,
+            plan_id=row.plan_id,
+            items=items,
+            public_unit_as_label=public_unit_as_label,
+        )
+    return data
diff --git a/backend/tests/test_pl_io_order_enrichment.py b/backend/tests/test_pl_io_order_enrichment.py
new file mode 100644
index 000000000..18f64f00a
--- /dev/null
+++ b/backend/tests/test_pl_io_order_enrichment.py
@@ -0,0 +1,65 @@
+from app.services.pl_io_order_service import (
+    resolve_unit_label,
+    PLAN_TYPE_BUSINESS_LABEL,
+    merge_plan_logistics_into_order,
+    merge_plan_product_into_item,
+)
+
+
+def test_resolve_unit_label_maps_dun():
+    assert resolve_unit_label({"dun": "吨"}, "dun") == "吨"
+    assert resolve_unit_label({}, "dun") == "dun"
+    assert resolve_unit_label({}, None) == "吨"
+    assert resolve_unit_label({}, "") == "吨"
+
+
+def test_merge_plan_logistics_into_order():
+    data = {"remark": "order-remark"}
+    plan = type(
+        "P",
+        (),
+        {
+            "plan_type": "outbound",
+            "remark": "plan-remark",
+            "pickup_address": "取货路1号",
+            "pickup_lng": 117.0,
+            "pickup_lat": 30.5,
+            "pickup_time": None,
+            "pickup_notes": "戴帽",
+            "destination_name": "仓库A",
+            "destination_address": "送货路2号",
+            "destination_lng": None,
+            "destination_lat": None,
+            "delivery_time": None,
+            "destination_notes": "北门",
+        },
+    )()
+    merge_plan_logistics_into_order(data, plan)
+    assert data["businessType"] == "出库"
+    assert data["planRemark"] == "plan-remark"
+    assert data["remark"] == "order-remark"  # 单据备注不被覆盖
+    assert data["pickupAddress"] == "取货路1号"
+    assert data["destinationAddress"] == "送货路2号"
+    assert data["destinationName"] == "仓库A"
+
+
+def test_merge_plan_product_into_item():
+    item = {"unit": "dun", "productName": "尿素"}
+    product = type(
+        "PP",
+        (),
+        {
+            "product_no": "P001",
+            "package_spec": "50kg/袋",
+            "remark": "易潮",
+            "unit": "dun",
+            "model": "46.4%",
+            "product_name": "工业级尿素",
+        },
+    )()
+    merge_plan_product_into_item(item, product, unit_map={"dun": "吨"})
+    assert item["productNo"] == "P001"
+    assert item["packageSpec"] == "50kg/袋"
+    assert item["remark"] == "易潮"
+    assert item["unitLabel"] == "吨"
+    assert item["unit"] == "dun"