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

"""运输计划 · 装货/送货段读取与列表摘要。"""
from __future__ import annotations

from collections import defaultdict
from datetime import date, datetime
from decimal import Decimal
from typing import Optional

from fastapi import HTTPException
from sqlalchemy.orm import Session

from app.core.utils import generate_uuid
from app.models.crm_models import CrmContractProduct
from app.models.pl_models import (
    PlPlan,
    PlPlanContract,
    PlPlanDelivery,
    PlPlanDeliveryProduct,
    PlPlanLoad,
    PlPlanLoadProduct,
    PlTransportAddress,
)
from app.schemas.pl.plan import (
    PlanArchiveBrandQtyItem,
    PlanContractLink,
    PlanDeliveryBody,
    PlanDeliveryRead,
    PlanLoadBody,
    PlanLoadRead,
    PlanStopProductBody,
    PlanStopProductRead,
)
from app.services.pl_brand_archive import load_brand_qtys_by_stop_product_ids
from app.services.pl_internal_customer import is_internal_customer_name
from app.services.pl_plan_stop_rules import normalize_stop_dibang_fields, warn_qty_imbalance
from app.services.pl_plan_service import (
    PLAN_CONTRACT_SIDE_DEMANDER,
    PLAN_CONTRACT_SIDE_SUPPLIER,
    assert_contract_selectable_for_plan,
)
from app.services.pl_product_display import load_crm_product_alias_maps, resolve_product_display_names

PLAN_CONTRACT_SIDE_LOAD = "load"
PLAN_CONTRACT_SIDE_DELIVERY = "delivery"


def _contract_link(row: PlPlanContract) -> PlanContractLink:
    return PlanContractLink(
        contractId=row.contract_id,
        contractNo=row.contract_no,
        stopId=row.stop_id,
    )


def _attach_stop_brand_qtys(
    product: PlanStopProductRead,
    brand_qtys_by_stop: dict[str, list[dict]],
) -> PlanStopProductRead:
    raw = brand_qtys_by_stop.get(product.id) or None
    if raw:
        product.brand_qtys = [PlanArchiveBrandQtyItem.model_validate(x) for x in raw]
    else:
        product.brand_qtys = None
    return product


def _resolve_stop_list_product_name(
    *,
    saved_product_name: Optional[str],
    product_id: Optional[str],
    standard_by_id: dict[str, str],
    alias_options_by_id: dict[str, list[str]],
) -> str:
    """列表展示用：优先 CRM 标准名，不展示别名。"""
    display_name, _alias = resolve_product_display_names(
        saved_product_name=saved_product_name,
        product_id=product_id,
        standard_by_id=standard_by_id,
        alias_options_by_id=alias_options_by_id,
    )
    return display_name or (saved_product_name or "")


def _apply_stop_list_product_name(
    product: PlanStopProductRead,
    *,
    standard_by_id: dict[str, str],
    alias_options_by_id: dict[str, list[str]],
) -> PlanStopProductRead:
    product.product_name = _resolve_stop_list_product_name(
        saved_product_name=product.product_name,
        product_id=product.product_id,
        standard_by_id=standard_by_id,
        alias_options_by_id=alias_options_by_id,
    )
    return product


def _load_stop_product(
    row: PlPlanLoadProduct,
    brand_qtys_by_stop: dict[str, list[dict]] | None = None,
    *,
    standard_by_id: dict[str, str] | None = None,
    alias_options_by_id: dict[str, list[str]] | None = None,
) -> PlanStopProductRead:
    product = PlanStopProductRead.model_validate(row)
    if standard_by_id is not None and alias_options_by_id is not None:
        product = _apply_stop_list_product_name(
            product,
            standard_by_id=standard_by_id,
            alias_options_by_id=alias_options_by_id,
        )
    if brand_qtys_by_stop is not None:
        return _attach_stop_brand_qtys(product, brand_qtys_by_stop)
    return product


def _delivery_stop_product(
    row: PlPlanDeliveryProduct,
    brand_qtys_by_stop: dict[str, list[dict]] | None = None,
    *,
    standard_by_id: dict[str, str] | None = None,
    alias_options_by_id: dict[str, list[str]] | None = None,
) -> PlanStopProductRead:
    product = PlanStopProductRead.model_validate(row)
    if standard_by_id is not None and alias_options_by_id is not None:
        product = _apply_stop_list_product_name(
            product,
            standard_by_id=standard_by_id,
            alias_options_by_id=alias_options_by_id,
        )
    if brand_qtys_by_stop is not None:
        return _attach_stop_brand_qtys(product, brand_qtys_by_stop)
    return product


def load_stop_contract_rows_grouped(
    db: Session,
    tenant_no: str,
    plan_ids: list[str],
) -> dict[str, list[PlPlanContract]]:
    if not plan_ids:
        return {}
    rows = (
        db.query(PlPlanContract)
        .filter(
            PlPlanContract.tenant_no == tenant_no,
            PlPlanContract.plan_id.in_(plan_ids),
            PlPlanContract.del_flag == 0,
        )
        .order_by(PlPlanContract.order_num.asc(), PlPlanContract.create_time.asc())
        .all()
    )
    grouped: dict[str, list[PlPlanContract]] = defaultdict(list)
    for row in rows:
        grouped[row.plan_id].append(row)
    return grouped


def _contracts_for_stop(
    contract_rows: list[PlPlanContract],
    stop_id: str,
    legacy_sides: tuple[str, ...],
) -> list[PlanContractLink]:
    by_stop = [_contract_link(r) for r in contract_rows if (r.stop_id or "") == stop_id]
    if by_stop:
        return by_stop
    return [
        _contract_link(r)
        for r in contract_rows
        if not (r.stop_id or "").strip() and r.side in legacy_sides
    ]


def _to_decimal(value) -> Optional[Decimal]:
    if value is None or value == "":
        return None
    try:
        return Decimal(str(value))
    except Exception:
        return None


def resolve_stop_address_from_library(
    db: Session,
    tenant_no: str,
    address: Optional[str],
    lng,
    lat,
) -> Optional[str]:
    """段 address 为空时，按经纬度从运输地址库补全 full_address。"""
    text = (address or "").strip()
    if text:
        return text
    lat_d = _to_decimal(lat)
    lng_d = _to_decimal(lng)
    if lat_d is None and lng_d is None:
        return text or None

    query = db.query(PlTransportAddress).filter(
        PlTransportAddress.tenant_no == tenant_no,
        PlTransportAddress.del_flag == 0,
        PlTransportAddress.status == "1",
    )
    if lat_d is not None:
        query = query.filter(PlTransportAddress.lat == lat_d)
    if lng_d is not None:
        query = query.filter(PlTransportAddress.lng == lng_d)
    rows = query.order_by(PlTransportAddress.create_time.desc()).all()
    if not rows:
        return text or None
    if len(rows) == 1:
        full = (rows[0].full_address or "").strip()
        return full or None
    if lng_d is not None:
        for row in rows:
            if row.lng == lng_d:
                full = (row.full_address or "").strip()
                if full:
                    return full
    full = (rows[0].full_address or "").strip()
    return full or None


def enrich_stop_addresses_from_library(
    db: Session,
    tenant_no: str,
    load: PlanLoadBody,
    deliveries: list[PlanDeliveryBody],
) -> None:
    resolved_load = resolve_stop_address_from_library(
        db, tenant_no, load.address, load.lng, load.lat
    )
    if resolved_load:
        load.address = resolved_load
    for delivery in deliveries:
        resolved = resolve_stop_address_from_library(
            db, tenant_no, delivery.address, delivery.lng, delivery.lat
        )
        if resolved:
            delivery.address = resolved


def load_plan_stops_grouped(
    db: Session,
    tenant_no: str,
    plan_ids: list[str],
) -> dict[str, tuple[Optional[PlanLoadRead], list[PlanDeliveryRead]]]:
    """批量读取装货段与送货段（列表/详情共用）。"""
    if not plan_ids:
        return {}

    loads = (
        db.query(PlPlanLoad)
        .filter(
            PlPlanLoad.tenant_no == tenant_no,
            PlPlanLoad.plan_id.in_(plan_ids),
            PlPlanLoad.del_flag == 0,
        )
        .all()
    )
    load_by_plan = {row.plan_id: row for row in loads}
    load_ids = [row.id for row in loads]

    load_products: dict[str, list[PlPlanLoadProduct]] = defaultdict(list)
    if load_ids:
        for row in (
            db.query(PlPlanLoadProduct)
            .filter(
                PlPlanLoadProduct.tenant_no == tenant_no,
                PlPlanLoadProduct.load_id.in_(load_ids),
                PlPlanLoadProduct.del_flag == 0,
            )
            .order_by(PlPlanLoadProduct.order_num.asc(), PlPlanLoadProduct.create_time.asc())
            .all()
        ):
            load_products[row.load_id].append(row)

    deliveries = (
        db.query(PlPlanDelivery)
        .filter(
            PlPlanDelivery.tenant_no == tenant_no,
            PlPlanDelivery.plan_id.in_(plan_ids),
            PlPlanDelivery.del_flag == 0,
        )
        .order_by(PlPlanDelivery.seq.asc(), PlPlanDelivery.create_time.asc())
        .all()
    )
    deliveries_by_plan: dict[str, list[PlPlanDelivery]] = defaultdict(list)
    delivery_ids: list[str] = []
    for row in deliveries:
        deliveries_by_plan[row.plan_id].append(row)
        delivery_ids.append(row.id)

    delivery_products: dict[str, list[PlPlanDeliveryProduct]] = defaultdict(list)
    if delivery_ids:
        for row in (
            db.query(PlPlanDeliveryProduct)
            .filter(
                PlPlanDeliveryProduct.tenant_no == tenant_no,
                PlPlanDeliveryProduct.delivery_id.in_(delivery_ids),
                PlPlanDeliveryProduct.del_flag == 0,
            )
            .order_by(PlPlanDeliveryProduct.order_num.asc(), PlPlanDeliveryProduct.create_time.asc())
            .all()
        ):
            delivery_products[row.delivery_id].append(row)

    stop_product_ids: list[str] = []
    crm_product_ids: set[str] = set()
    for rows in load_products.values():
        for p in rows:
            if p.id:
                stop_product_ids.append(p.id)
            if (p.product_id or "").strip():
                crm_product_ids.add(p.product_id.strip())
    for rows in delivery_products.values():
        for p in rows:
            if p.id:
                stop_product_ids.append(p.id)
            if (p.product_id or "").strip():
                crm_product_ids.add(p.product_id.strip())
    brand_qtys_by_stop = load_brand_qtys_by_stop_product_ids(
        db, tenant_no=tenant_no, stop_product_ids=stop_product_ids
    )
    standard_by_id, alias_options_by_id = load_crm_product_alias_maps(
        db, tenant_no=tenant_no, product_ids=crm_product_ids
    )

    contracts_by_plan = load_stop_contract_rows_grouped(db, tenant_no, plan_ids)

    result: dict[str, tuple[Optional[PlanLoadRead], list[PlanDeliveryRead]]] = {}
    for plan_id in plan_ids:
        contract_rows = contracts_by_plan.get(plan_id, [])
        load_row = load_by_plan.get(plan_id)
        load_read: Optional[PlanLoadRead] = None
        if load_row:
            load_read = PlanLoadRead.model_validate(load_row)
            resolved_load_address = resolve_stop_address_from_library(
                db,
                tenant_no,
                load_read.address,
                load_read.lng,
                load_read.lat,
            )
            if resolved_load_address:
                load_read.address = resolved_load_address
            load_read.products = [
                _load_stop_product(
                    p,
                    brand_qtys_by_stop,
                    standard_by_id=standard_by_id,
                    alias_options_by_id=alias_options_by_id,
                )
                for p in load_products.get(load_row.id, [])
            ]
            load_read.contracts = _contracts_for_stop(
                contract_rows,
                load_row.id,
                (PLAN_CONTRACT_SIDE_LOAD, PLAN_CONTRACT_SIDE_SUPPLIER),
            )

        delivery_reads: list[PlanDeliveryRead] = []
        plan_deliveries = deliveries_by_plan.get(plan_id, [])
        for idx, delivery_row in enumerate(plan_deliveries):
            delivery_read = PlanDeliveryRead.model_validate(delivery_row)
            resolved_delivery_address = resolve_stop_address_from_library(
                db,
                tenant_no,
                delivery_read.address,
                delivery_read.lng,
                delivery_read.lat,
            )
            if resolved_delivery_address:
                delivery_read.address = resolved_delivery_address
            delivery_read.products = [
                _delivery_stop_product(
                    p,
                    brand_qtys_by_stop,
                    standard_by_id=standard_by_id,
                    alias_options_by_id=alias_options_by_id,
                )
                for p in delivery_products.get(delivery_row.id, [])
            ]
            legacy_sides = (PLAN_CONTRACT_SIDE_DELIVERY, PLAN_CONTRACT_SIDE_DEMANDER)
            # 未迁移合同仅挂到第一个送货段
            if idx == 0:
                delivery_read.contracts = _contracts_for_stop(
                    contract_rows,
                    delivery_row.id,
                    legacy_sides,
                )
            else:
                delivery_read.contracts = _contracts_for_stop(
                    contract_rows,
                    delivery_row.id,
                    (PLAN_CONTRACT_SIDE_DELIVERY,),
                )
            delivery_reads.append(delivery_read)

        result[plan_id] = (load_read, delivery_reads)

    return result


def resolve_load_customer_name(
    plan: PlPlan,
    load: Optional[PlanLoadRead],
) -> Optional[str]:
    if load and (load.customer_name or "").strip():
        return load.customer_name.strip()
    legacy = (plan.supplier_name or "").strip()
    return legacy or None


def resolve_delivery_customer_summary(
    plan: PlPlan,
    deliveries: list[PlanDeliveryRead],
) -> Optional[str]:
    names: list[str] = []
    seen: set[str] = set()
    for delivery in deliveries:
        name = (delivery.customer_name or "").strip()
        if name and name not in seen:
            seen.add(name)
            names.append(name)

    if names:
        if len(names) <= 2:
            return "、".join(names)
        return f"{names[0]}、{names[1]}等{len(names)}家"

    legacy = (plan.customer_name or plan.destination_name or "").strip()
    return legacy or None


def collect_unique_delivery_date_strings(deliveries) -> list[str]:
    """送货段 delivery_time → 去重升序 YYYY-MM-DD。"""
    days: set[date] = set()
    for row in deliveries or []:
        raw = getattr(row, "delivery_time", None)
        if raw is None or raw == "":
            continue
        if isinstance(raw, datetime):
            days.add(raw.date())
        elif isinstance(raw, date):
            days.add(raw)
        else:
            text = str(raw).strip()
            if not text:
                continue
            try:
                days.add(datetime.strptime(text[:10], "%Y-%m-%d").date())
            except ValueError:
                continue
    return [d.isoformat() for d in sorted(days)]


def attach_plan_stops_to_item(
    plan_item,
    plan: PlPlan,
    stops: tuple[Optional[PlanLoadRead], list[PlanDeliveryRead]],
) -> None:
    load_read, delivery_reads = stops
    plan_item.load = load_read
    plan_item.deliveries = delivery_reads
    plan_item.load_customer_name = resolve_load_customer_name(plan, load_read)
    plan_item.delivery_customer_summary = resolve_delivery_customer_summary(plan, delivery_reads)
    if load_read and (load_read.address or "").strip():
        plan_item.pickup_address = load_read.address
    if delivery_reads and (delivery_reads[0].address or "").strip():
        plan_item.destination_address = delivery_reads[0].address
    plan_item.delivery_times = collect_unique_delivery_date_strings(delivery_reads)


def uses_multi_stop_payload(load: Optional[PlanLoadBody], deliveries: Optional[list[PlanDeliveryBody]]) -> bool:
    return load is not None or deliveries is not None


def _normalize_contract_links(links: Optional[list[PlanContractLink]]) -> list[PlanContractLink]:
    if not links:
        return []
    seen: set[str] = set()
    out: list[PlanContractLink] = []
    for link in links:
        cid = (link.contract_id or "").strip()
        if not cid or cid in seen:
            continue
        seen.add(cid)
        out.append(link)
    return out


def assert_at_most_one_contract(
    links: list[PlanContractLink],
    side_label: str,
) -> list[PlanContractLink]:
    if len(links) > 1:
        raise HTTPException(
            status_code=422,
            detail=f"{side_label}只能选择 1 份",
        )
    return links


def _assert_product_rows_in_contract_allowed(
    products: Optional[list],
    allowed: set[str],
    *,
    row_label: str = "装货产品",
) -> None:
    for prod in products or []:
        pid = str(getattr(prod, "product_id", None) or "").strip()
        name = (getattr(prod, "product_name", None) or pid or "未命名").strip()
        if not pid or pid not in allowed:
            raise HTTPException(
                status_code=422,
                detail=f"{row_label}「{name}」不在所选合同明细中",
            )


def _contract_product_rows(db: Session, tenant_no: str, contract_id: str) -> list:
    return (
        db.query(CrmContractProduct)
        .filter(
            CrmContractProduct.contract_id == contract_id,
            CrmContractProduct.tenant_no == tenant_no,
            CrmContractProduct.del_flag == 0,
        )
        .all()
    )


def _contract_allowed_product_ids(db: Session, tenant_no: str, contract_id: str) -> set[str]:
    rows = _contract_product_rows(db, tenant_no, contract_id)
    return {
        str(r.product_id).strip()
        for r in rows
        if getattr(r, "product_id", None) and str(r.product_id).strip()
    }


def _assert_stop_products_package_for_contract(
    db: Session,
    tenant_no: str,
    contract_id: str,
    stop_products: Optional[list],
) -> None:
    from app.services.pl_contract_plan_qty import (
        validate_stop_products_package_against_contract,
    )

    rows = _contract_product_rows(db, tenant_no, contract_id)
    ok, err = validate_stop_products_package_against_contract(
        contract_products=rows,
        stop_products=stop_products or [],
    )
    if not ok:
        raise HTTPException(status_code=422, detail=err or "包装规格不符合合同明细")


def assert_load_products_in_selected_contract(
    db: Session,
    tenant_no: str,
    load: PlanLoadBody,
    plan_products: Optional[list] = None,
) -> None:
    if is_internal_customer_name(load.customer_name):
        return
    links = _normalize_contract_links(load.contracts)
    if not links:
        return
    contract_id = (links[0].contract_id or "").strip()
    if not contract_id:
        return
    allowed = _contract_allowed_product_ids(db, tenant_no, contract_id)
    _assert_product_rows_in_contract_allowed(load.products, allowed, row_label="装货产品")
    _assert_product_rows_in_contract_allowed(plan_products, allowed, row_label="装货产品")
    merged = list(load.products or []) + list(plan_products or [])
    _assert_stop_products_package_for_contract(db, tenant_no, contract_id, merged)


def assert_delivery_products_in_selected_contract(
    db: Session,
    tenant_no: str,
    delivery: PlanDeliveryBody,
    seq: int,
) -> None:
    if is_internal_customer_name(delivery.customer_name):
        return
    links = _normalize_contract_links(delivery.contracts)
    if not links:
        return
    contract_id = (links[0].contract_id or "").strip()
    if not contract_id:
        return
    allowed = _contract_allowed_product_ids(db, tenant_no, contract_id)
    _assert_product_rows_in_contract_allowed(
        delivery.products,
        allowed,
        row_label=f"送货段#{seq}产品",
    )
    _assert_stop_products_package_for_contract(
        db, tenant_no, contract_id, delivery.products
    )


def assert_stop_contract_required(
    customer_name: Optional[str],
    contracts: Optional[list[PlanContractLink]],
    side_label: str,
) -> None:
    """非内部客户装/送货段保存时必须至少选一份合同。"""
    if is_internal_customer_name(customer_name):
        return
    if not _normalize_contract_links(contracts):
        raise HTTPException(status_code=422, detail=f"请为{side_label}选择合同")


def _resolve_stop_contracts(
    db: Session,
    tenant_no: str,
    customer_id: Optional[str],
    customer_name: Optional[str],
    contracts: Optional[list[PlanContractLink]],
    side_label: str,
) -> list[PlanContractLink]:
    if is_internal_customer_name(customer_name):
        return []
    links = _normalize_contract_links(contracts)
    links = assert_at_most_one_contract(links, side_label)
    if not links:
        return []
    resolved: list[PlanContractLink] = []
    for link in links:
        contract = assert_contract_selectable_for_plan(
            db,
            link.contract_id,
            tenant_no,
            expected_customer_id=customer_id,
            side_label=side_label,
        )
        resolved.append(
            PlanContractLink(
                contractId=contract.id,
                contractNo=contract.contract_no or link.contract_no,
            )
        )
    return resolved


def validate_plan_stops_write(
    db: Session,
    tenant_no: str,
    load: Optional[PlanLoadBody],
    deliveries: Optional[list[PlanDeliveryBody]],
    products: Optional[list],
) -> tuple[PlanLoadBody, list[PlanDeliveryBody]]:
    if load is None:
        raise HTTPException(status_code=422, detail="请填写装货信息")
    if not (load.customer_id or "").strip() or not (load.customer_name or "").strip():
        raise HTTPException(status_code=422, detail="请选择装货客户")
    if not deliveries:
        raise HTTPException(status_code=422, detail="请至少添加一段送货信息")
    if products is not None and not products:
        raise HTTPException(status_code=422, detail="请至少保留一条产品明细")

    assert_stop_contract_required(load.customer_name, load.contracts, "装货客户")

    load.contracts = _resolve_stop_contracts(
        db,
        tenant_no,
        load.customer_id,
        load.customer_name,
        load.contracts,
        "装货合同",
    )

    resolved_deliveries: list[PlanDeliveryBody] = []
    for idx, delivery in enumerate(deliveries, start=1):
        if not (delivery.customer_id or "").strip() or not (delivery.customer_name or "").strip():
            raise HTTPException(status_code=422, detail=f"送货段#{idx}请选择客户")
        assert_stop_contract_required(delivery.customer_name, delivery.contracts, f"送货段#{idx}")
        delivery.contracts = _resolve_stop_contracts(
            db,
            tenant_no,
            delivery.customer_id,
            delivery.customer_name,
            delivery.contracts,
            f"送货段#{idx}合同",
        )
        resolved_deliveries.append(delivery)

    enrich_stop_addresses_from_library(db, tenant_no, load, resolved_deliveries)
    assert_load_products_in_selected_contract(db, tenant_no, load, plan_products=products)
    for idx, delivery in enumerate(resolved_deliveries, start=1):
        assert_delivery_products_in_selected_contract(db, tenant_no, delivery, idx)
    return load, resolved_deliveries


def collect_stop_qty_warnings(
    products: Optional[list],
    load: PlanLoadBody,
    deliveries: list[PlanDeliveryBody],
) -> list[str]:
    plan_rows = products or []
    load_rows = load.products or []
    delivery_rows: list[PlanStopProductBody] = []
    for delivery in deliveries:
        delivery_rows.extend(delivery.products or [])
    return warn_qty_imbalance(plan_rows, load_rows, delivery_rows)


def sync_plan_legacy_from_stops(
    plan: PlPlan,
    load: PlanLoadBody,
    deliveries: list[PlanDeliveryBody],
) -> None:
    plan.supplier_id = load.customer_id
    plan.supplier_name = load.customer_name
    plan.pickup_address = load.address
    plan.pickup_lng = load.lng
    plan.pickup_lat = load.lat
    plan.pickup_time = load.load_time
    plan.pickup_contact = load.contact
    plan.pickup_notes = load.notes

    first = deliveries[0]
    plan.customer_id = first.customer_id
    plan.customer_name = first.customer_name
    plan.destination_id = first.customer_id
    plan.destination_name = first.customer_name
    plan.destination_address = first.address
    plan.destination_lng = first.lng
    plan.destination_lat = first.lat
    plan.delivery_time = first.delivery_time
    plan.delivery_contact = first.contact
    plan.destination_notes = first.notes


def _soft_delete_load_products_and_stop_contracts(
    db: Session,
    tenant_no: str,
    plan_id: str,
    user_id: Optional[str],
) -> None:
    """软删装货产品与装/送货段合同；送货段本体由 upsert 保留主键，不在此全量软删。"""
    now = datetime.now()
    load_ids = [
        row.id
        for row in db.query(PlPlanLoad)
        .filter(
            PlPlanLoad.plan_id == plan_id,
            PlPlanLoad.tenant_no == tenant_no,
            PlPlanLoad.del_flag == 0,
        )
        .all()
    ]
    if load_ids:
        db.query(PlPlanLoadProduct).filter(
            PlPlanLoadProduct.load_id.in_(load_ids),
            PlPlanLoadProduct.del_flag == 0,
        ).update({"del_flag": 1, "update_by": user_id, "update_time": now}, synchronize_session=False)

    db.query(PlPlanContract).filter(
        PlPlanContract.plan_id == plan_id,
        PlPlanContract.tenant_no == tenant_no,
        PlPlanContract.del_flag == 0,
        PlPlanContract.side.in_(
            [PLAN_CONTRACT_SIDE_LOAD, PLAN_CONTRACT_SIDE_DELIVERY]
        ),
    ).update({"del_flag": 1, "update_by": user_id, "update_time": now}, synchronize_session=False)


def _soft_delete_delivery_products(
    db: Session,
    delivery_ids: list[str],
    user_id: Optional[str],
    now: datetime,
) -> None:
    if not delivery_ids:
        return
    db.query(PlPlanDeliveryProduct).filter(
        PlPlanDeliveryProduct.delivery_id.in_(delivery_ids),
        PlPlanDeliveryProduct.del_flag == 0,
    ).update({"del_flag": 1, "update_by": user_id, "update_time": now}, synchronize_session=False)


def resolve_delivery_id_for_upsert(
    *,
    plan_id: str,
    payload_id: Optional[str],
    existing_by_id: dict,
) -> tuple[str, Optional[object]]:
    pid = (payload_id or "").strip()
    if not pid:
        return generate_uuid(), None
    row = existing_by_id.get(pid)
    if row is None or getattr(row, "plan_id", None) != plan_id:
        raise HTTPException(status_code=422, detail=f"送货段 id 无效或不属于本计划：{pid}")
    return pid, row


def _write_stop_contracts(
    db: Session,
    tenant_no: str,
    plan_id: str,
    stop_id: str,
    side: str,
    contracts: list[PlanContractLink],
    user_id: Optional[str],
    dept_id: Optional[str],
    order_num_start: int,
) -> int:
    now = datetime.now()
    order_num = order_num_start
    for link in contracts:
        order_num += 1
        db.add(
            PlPlanContract(
                id=generate_uuid(),
                plan_id=plan_id,
                side=side,
                stop_id=stop_id,
                contract_id=link.contract_id,
                contract_no=link.contract_no,
                tenant_no=tenant_no,
                order_num=order_num,
                create_by=user_id,
                create_dept_id=dept_id or "",
                create_time=now,
                update_by=user_id,
                update_time=now,
                del_flag=0,
            )
        )
    return order_num


def _resolve_plan_product_id(
    prod: PlanStopProductBody,
    plan_products: list,
) -> str:
    explicit = (prod.plan_product_id or "").strip()
    if explicit:
        for row in plan_products:
            if row.id == explicit:
                return explicit
    product_id = (prod.product_id or "").strip()
    if product_id:
        for row in plan_products:
            if (row.product_id or "").strip() == product_id:
                return row.id
    product_no = (prod.product_no or "").strip()
    if product_no:
        for row in plan_products:
            if (row.product_no or "").strip() == product_no:
                return row.id
    raise HTTPException(
        status_code=422,
        detail=f"无法匹配计划产品行：{prod.product_name or explicit or product_id}",
    )


def _resolve_or_create_plan_load(
    db: Session,
    tenant_no: str,
    plan_id: str,
    load: PlanLoadBody,
    user_id: Optional[str],
    dept_id: Optional[str],
    now: datetime,
) -> str:
    """装货段与 plan 1:1；uk_pl_plan_load_plan 仅按 plan_id 唯一，软删后不可再 INSERT，须原地更新。"""
    generate_dibang, dibang_direction = normalize_stop_dibang_fields(
        load.customer_name,
        "load",
        load.generate_dibang,
        load.dibang_direction,
    )

    existing = (
        db.query(PlPlanLoad)
        .filter(
            PlPlanLoad.plan_id == plan_id,
            PlPlanLoad.tenant_no == tenant_no,
        )
        .order_by(PlPlanLoad.del_flag.asc(), PlPlanLoad.update_time.desc())
        .first()
    )
    if existing:
        existing.customer_id = load.customer_id
        existing.customer_name = load.customer_name
        existing.address = load.address
        existing.lng = load.lng
        existing.lat = load.lat
        existing.load_time = load.load_time
        existing.contact = load.contact
        existing.notes = load.notes
        existing.generate_dibang = generate_dibang
        existing.dibang_direction = dibang_direction
        existing.remark = load.remark
        existing.del_flag = 0
        existing.update_by = user_id
        existing.update_time = now
        return existing.id

    load_id = generate_uuid()
    db.add(
        PlPlanLoad(
            id=load_id,
            plan_id=plan_id,
            customer_id=load.customer_id,
            customer_name=load.customer_name,
            address=load.address,
            lng=load.lng,
            lat=load.lat,
            load_time=load.load_time,
            contact=load.contact,
            notes=load.notes,
            generate_dibang=generate_dibang,
            dibang_direction=dibang_direction,
            tenant_no=tenant_no,
            remark=load.remark,
            create_by=user_id,
            create_dept_id=dept_id or "",
            create_time=now,
            update_by=user_id,
            update_time=now,
            del_flag=0,
        )
    )
    return load_id


def replace_plan_stops(
    db: Session,
    tenant_no: str,
    plan_id: str,
    load: PlanLoadBody,
    deliveries: list[PlanDeliveryBody],
    user_id: Optional[str],
    dept_id: Optional[str],
    plan_products: Optional[list] = None,
) -> None:
    _soft_delete_load_products_and_stop_contracts(db, tenant_no, plan_id, user_id)
    now = datetime.now()

    load_id = _resolve_or_create_plan_load(
        db,
        tenant_no,
        plan_id,
        load,
        user_id,
        dept_id,
        now,
    )

    plan_product_rows = plan_products or []

    for idx, prod in enumerate(load.products or []):
        plan_product_id = _resolve_plan_product_id(prod, plan_product_rows)
        db.add(
            PlPlanLoadProduct(
                id=generate_uuid(),
                load_id=load_id,
                plan_id=plan_id,
                plan_product_id=plan_product_id,
                product_id=prod.product_id,
                product_no=prod.product_no,
                product_name=prod.product_name,
                model=prod.model,
                unit=prod.unit or "吨",
                package_spec=prod.package_spec,
                delivery_method=prod.delivery_method,
                plan_qty=prod.plan_qty,
                tenant_no=tenant_no,
                order_num=idx,
                remark=prod.remark,
                create_by=user_id,
                create_dept_id=dept_id or "",
                create_time=now,
                update_by=user_id,
                update_time=now,
                del_flag=0,
            )
        )

    contract_order = 0
    contract_order = _write_stop_contracts(
        db,
        tenant_no,
        plan_id,
        load_id,
        PLAN_CONTRACT_SIDE_LOAD,
        load.contracts or [],
        user_id,
        dept_id,
        contract_order,
    )

    existing_deliveries = (
        db.query(PlPlanDelivery)
        .filter(
            PlPlanDelivery.plan_id == plan_id,
            PlPlanDelivery.tenant_no == tenant_no,
        )
        .all()
    )
    existing_by_id = {row.id: row for row in existing_deliveries}
    kept_ids: set[str] = set()

    for delivery in deliveries:
        generate_dibang, dibang_direction = normalize_stop_dibang_fields(
            delivery.customer_name,
            "delivery",
            delivery.generate_dibang,
            delivery.dibang_direction,
        )

        delivery_id, existing = resolve_delivery_id_for_upsert(
            plan_id=plan_id,
            payload_id=delivery.id,
            existing_by_id=existing_by_id,
        )
        kept_ids.add(delivery_id)

        if existing is not None:
            existing.seq = delivery.seq or 1
            existing.customer_id = delivery.customer_id
            existing.customer_name = delivery.customer_name
            existing.address = delivery.address
            existing.lng = delivery.lng
            existing.lat = delivery.lat
            existing.delivery_time = delivery.delivery_time
            existing.contact = delivery.contact
            existing.notes = delivery.notes
            existing.generate_dibang = generate_dibang
            existing.dibang_direction = dibang_direction
            existing.remark = delivery.remark
            existing.del_flag = 0
            existing.update_by = user_id
            existing.update_time = now
        else:
            db.add(
                PlPlanDelivery(
                    id=delivery_id,
                    plan_id=plan_id,
                    seq=delivery.seq or 1,
                    customer_id=delivery.customer_id,
                    customer_name=delivery.customer_name,
                    address=delivery.address,
                    lng=delivery.lng,
                    lat=delivery.lat,
                    delivery_time=delivery.delivery_time,
                    contact=delivery.contact,
                    notes=delivery.notes,
                    generate_dibang=generate_dibang,
                    dibang_direction=dibang_direction,
                    tenant_no=tenant_no,
                    remark=delivery.remark,
                    create_by=user_id,
                    create_dept_id=dept_id or "",
                    create_time=now,
                    update_by=user_id,
                    update_time=now,
                    del_flag=0,
                )
            )

        _soft_delete_delivery_products(db, [delivery_id], user_id, now)

        for idx, prod in enumerate(delivery.products or []):
            plan_product_id = _resolve_plan_product_id(prod, plan_product_rows)
            db.add(
                PlPlanDeliveryProduct(
                    id=generate_uuid(),
                    delivery_id=delivery_id,
                    plan_id=plan_id,
                    plan_product_id=plan_product_id,
                    product_id=prod.product_id,
                    product_no=prod.product_no,
                    product_name=prod.product_name,
                    model=prod.model,
                    unit=prod.unit or "吨",
                    package_spec=prod.package_spec,
                    delivery_method=prod.delivery_method,
                    plan_qty=prod.plan_qty,
                    tenant_no=tenant_no,
                    order_num=idx,
                    remark=prod.remark,
                    create_by=user_id,
                    create_dept_id=dept_id or "",
                    create_time=now,
                    update_by=user_id,
                    update_time=now,
                    del_flag=0,
                )
            )

        contract_order = _write_stop_contracts(
            db,
            tenant_no,
            plan_id,
            delivery_id,
            PLAN_CONTRACT_SIDE_DELIVERY,
            delivery.contracts or [],
            user_id,
            dept_id,
            contract_order,
        )

    removed_ids = [row.id for row in existing_deliveries if row.id not in kept_ids and row.del_flag == 0]
    if removed_ids:
        _soft_delete_delivery_products(db, removed_ids, user_id, now)
        db.query(PlPlanDelivery).filter(
            PlPlanDelivery.id.in_(removed_ids),
            PlPlanDelivery.tenant_no == tenant_no,
            PlPlanDelivery.del_flag == 0,
        ).update({"del_flag": 1, "update_by": user_id, "update_time": now}, synchronize_session=False)