feat(pl): add customer stop pref service and unit tests

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/services/pl_customer_stop_pref_service.py

@@ -0,0 +1,168 @@
+"""客户装/送货物流默认记忆"""
+from __future__ import annotations
+
+import logging
+from decimal import Decimal
+from typing import Any, Mapping, Optional, Sequence
+
+from sqlalchemy.orm import Session
+
+from app.core.utils import generate_uuid
+from app.models.pl_models import PlCustomerStopPref
+
+logger = logging.getLogger(__name__)
+
+STOP_LOAD = "load"
+STOP_DELIVERY = "delivery"
+
+
+def _as_str(value: Any) -> str:
+    return str(value or "").strip()
+
+
+def _row_to_item(stop_type: str, row: Mapping[str, Any]) -> Optional[dict]:
+    customer_id = _as_str(row.get("customer_id") or row.get("customerId"))
+    if not customer_id:
+        return None
+    lng = row.get("lng")
+    lat = row.get("lat")
+    return {
+        "customer_id": customer_id,
+        "customer_name": _as_str(row.get("customer_name") or row.get("customerName")) or None,
+        "stop_type": stop_type,
+        "address": (row.get("address") if row.get("address") is not None else "") or "",
+        "lng": lng,
+        "lat": lat,
+        "contact": (row.get("contact") if row.get("contact") is not None else "") or "",
+        "notes": (row.get("notes") if row.get("notes") is not None else "") or "",
+    }
+
+
+def build_stop_pref_upsert_items(
+    *,
+    load: Optional[Mapping[str, Any]],
+    deliveries: Optional[Sequence[Mapping[str, Any]]],
+) -> list[dict]:
+    """抽出待 upsert 条目；同客户同 stop_type 后者覆盖（保持列表中每种键一条）。"""
+    merged: dict[tuple[str, str], dict] = {}
+    if load:
+        item = _row_to_item(STOP_LOAD, load)
+        if item:
+            merged[(item["customer_id"], item["stop_type"])] = item
+    for row in deliveries or []:
+        item = _row_to_item(STOP_DELIVERY, row)
+        if item:
+            merged[(item["customer_id"], item["stop_type"])] = item
+    return list(merged.values())
+
+
+def get_customer_stop_pref(
+    db: Session,
+    tenant_no: str,
+    customer_id: str,
+    stop_type: str,
+) -> Optional[PlCustomerStopPref]:
+    cid = _as_str(customer_id)
+    st = _as_str(stop_type)
+    if not cid or st not in (STOP_LOAD, STOP_DELIVERY):
+        return None
+    return (
+        db.query(PlCustomerStopPref)
+        .filter(
+            PlCustomerStopPref.tenant_no == tenant_no,
+            PlCustomerStopPref.customer_id == cid,
+            PlCustomerStopPref.stop_type == st,
+            PlCustomerStopPref.del_flag == 0,
+        )
+        .first()
+    )
+
+
+def upsert_customer_stop_pref(
+    db: Session,
+    *,
+    tenant_no: str,
+    item: Mapping[str, Any],
+    user_id: Optional[str],
+    dept_id: Optional[str],
+) -> None:
+    customer_id = _as_str(item.get("customer_id"))
+    stop_type = _as_str(item.get("stop_type"))
+    if not customer_id or stop_type not in (STOP_LOAD, STOP_DELIVERY):
+        return
+    row = get_customer_stop_pref(db, tenant_no, customer_id, stop_type)
+    lng = item.get("lng")
+    lat = item.get("lat")
+    if row is None:
+        row = PlCustomerStopPref(
+            id=generate_uuid(),
+            tenant_no=tenant_no,
+            customer_id=customer_id,
+            create_by=user_id,
+            create_dept_id=dept_id or "",
+        )
+        db.add(row)
+    row.customer_name = item.get("customer_name")
+    row.stop_type = stop_type
+    row.address = item.get("address") if item.get("address") is not None else ""
+    row.lng = Decimal(str(lng)) if lng is not None and str(lng) != "" else None
+    row.lat = Decimal(str(lat)) if lat is not None and str(lat) != "" else None
+    row.contact = item.get("contact") if item.get("contact") is not None else ""
+    row.notes = item.get("notes") if item.get("notes") is not None else ""
+    row.update_by = user_id
+    row.del_flag = 0
+
+
+def upsert_prefs_from_plan_stops(
+    db: Session,
+    *,
+    tenant_no: str,
+    load: Optional[Mapping[str, Any]],
+    deliveries: Optional[Sequence[Mapping[str, Any]]],
+    user_id: Optional[str],
+    dept_id: Optional[str],
+) -> None:
+    """供 plan 保存调用；异常由调用方用 savepoint 吞掉。"""
+    for item in build_stop_pref_upsert_items(load=load, deliveries=deliveries):
+        upsert_customer_stop_pref(
+            db,
+            tenant_no=tenant_no,
+            item=item,
+            user_id=user_id,
+            dept_id=dept_id,
+        )
+
+
+def safe_upsert_prefs_from_plan_stops(
+    db: Session,
+    *,
+    tenant_no: str,
+    load: Optional[Any],
+    deliveries: Optional[Sequence[Any]],
+    user_id: Optional[str],
+    dept_id: Optional[str],
+) -> None:
+    """savepoint 包裹；失败只记日志，不抛给计划主流程。"""
+    load_map = None
+    if load is not None:
+        load_map = load.model_dump(by_alias=False) if hasattr(load, "model_dump") else dict(load)
+    delivery_maps = []
+    for d in deliveries or []:
+        delivery_maps.append(
+            d.model_dump(by_alias=False) if hasattr(d, "model_dump") else dict(d)
+        )
+    try:
+        with db.begin_nested():
+            upsert_prefs_from_plan_stops(
+                db,
+                tenant_no=tenant_no,
+                load=load_map,
+                deliveries=delivery_maps,
+                user_id=user_id,
+                dept_id=dept_id,
+            )
+    except Exception:
+        logger.exception(
+            "upsert customer stop pref failed tenant=%s (plan save continues)",
+            tenant_no,
+        )

backend/tests/test_pl_customer_stop_pref_service.py

@@ -0,0 +1,54 @@
+"""客户装/送货默认记忆：条目合并规则"""
+from app.services.pl_customer_stop_pref_service import build_stop_pref_upsert_items
+
+
+def test_build_items_skips_empty_customer():
+    items = build_stop_pref_upsert_items(
+        load={"customer_id": "", "address": "A"},
+        deliveries=[{"customer_id": "c1", "address": "D1", "contact": "1", "notes": "n"}],
+    )
+    assert len(items) == 1
+    assert items[0]["stop_type"] == "delivery"
+    assert items[0]["customer_id"] == "c1"
+
+
+def test_build_items_load_and_delivery_independent():
+    items = build_stop_pref_upsert_items(
+        load={
+            "customer_id": "c1",
+            "customer_name": "甲",
+            "address": "装址",
+            "lng": 1.1,
+            "lat": 2.2,
+            "contact": "装联",
+            "notes": "装注",
+        },
+        deliveries=[
+            {
+                "customer_id": "c1",
+                "customer_name": "甲",
+                "address": "送址",
+                "contact": "送联",
+                "notes": "送注",
+            }
+        ],
+    )
+    assert len(items) == 2
+    by_type = {i["stop_type"]: i for i in items}
+    assert by_type["load"]["address"] == "装址"
+    assert by_type["delivery"]["address"] == "送址"
+
+
+def test_build_items_same_delivery_customer_last_wins():
+    items = build_stop_pref_upsert_items(
+        load=None,
+        deliveries=[
+            {"customer_id": "c1", "address": "第一段", "contact": "a", "notes": "1"},
+            {"customer_id": "c2", "address": "别家", "contact": "b", "notes": "2"},
+            {"customer_id": "c1", "address": "最后段", "contact": "c", "notes": "3"},
+        ],
+    )
+    delivery_c1 = [i for i in items if i["stop_type"] == "delivery" and i["customer_id"] == "c1"]
+    assert len(delivery_c1) == 1
+    assert delivery_c1[0]["address"] == "最后段"
+    assert delivery_c1[0]["contact"] == "c"