shanxia2.0/backend/app/services/pl_customer_stop_pref_service.py

"""客户装/送货物流默认记忆"""
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,
        )