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

"""运输计划业务逻辑"""
import json
from datetime import date, datetime
from decimal import Decimal
from typing import Optional, Union

from fastapi import HTTPException
from sqlalchemy import and_, exists, func, or_, select
from sqlalchemy.orm import Session

from app.core.utils import generate_uuid
from app.services.pl_mobile_product_identity import bind_mobile_product_key
from app.models.crm_models import CrmProduct
from app.models.pl_models import (
    PlCarrier,
    PlFreightSettlement,
    PlFreightSettlementTransport,
    PlPersonnel,
    PlPlan,
    PlPlanChange,
    PlPlanContract,
    PlPlanProduct,
    PlPlanTransport,
    PlTransport,
    PlTransportSiteLink,
    PlVehicle,
    PlVehiclePersonnel,
)
from app.schemas.pl.plan import (
    PlanArchiveFeedbacksBlock,
    PlanArchiveFreightBlock,
    PlanArchiveProductItem,
    PlanContractLink,
    PlanCreateFull,
    PlanDriverFeedbackUpdateBody,
    PlanProductCreate,
    PlanSiteFeedbackUpdateBody,
    PlanTransportCreateBlock,
)
from app.services.pl_brand_archive import persist_plan_product_brand_qtys, validate_brand_qty_rows
from app.services.pl_feedback_brand_qty import persist_transport_product_feedback
from app.services.pl_freight_settlement_service import sync_draft_settlement_freight_for_transport
from app.services.pl_transport_resource_snapshot_service import (
    apply_resource_snapshot_json,
    build_transport_resource_snapshot,
    clear_resource_snapshot,
)


def determine_plan_type(supplier_name: Optional[str], destination_name: Optional[str]) -> str:
    if destination_name and ("安庆山峡" in destination_name or "安徽云栋" in destination_name):
        return "inbound"
    if supplier_name and ("安庆山峡" in supplier_name or "安徽云栋" in supplier_name):
        return "outbound"
    return "direct"


def is_ton_product_unit(unit: Optional[str]) -> bool:
    code = (unit or "").strip()
    if not code:
        return True
    return code in ("dun", "吨")


def _product_quantity(prod: PlanProductCreate) -> Decimal:
    if prod.quantity is not None:
        return Decimal(str(prod.quantity))
    if prod.inbound_estimate_qty is not None:
        return Decimal(str(prod.inbound_estimate_qty))
    if prod.outbound_estimate_qty is not None:
        return Decimal(str(prod.outbound_estimate_qty))
    return Decimal("0")


def _product_create_actual_qty(prod: PlanProductCreate) -> Decimal:
    if prod.actual_qty is not None:
        return Decimal(str(prod.actual_qty))
    return Decimal("0")


def _plan_product_actual_qty(prod: PlPlanProduct) -> Decimal:
    if prod.actual_qty is not None:
        return Decimal(str(prod.actual_qty))
    return Decimal("0")


def sum_product_actual_quantities(products: list[PlanProductCreate]) -> Decimal:
    return sum(_product_create_actual_qty(p) for p in products)


def sum_plan_product_actual_quantities(products: list[PlPlanProduct]) -> Decimal:
    return sum(_plan_product_actual_qty(p) for p in products)


def _estimate_qty_fields(plan_type: str, qty: Optional[Decimal]) -> dict:
    if qty is None:
        return {}
    q = float(qty)
    if plan_type == "inbound":
        return {"inbound_estimate_qty": q, "outbound_estimate_qty": q}
    if plan_type == "outbound":
        return {"outbound_estimate_qty": q, "inbound_estimate_qty": q}
    return {"inbound_estimate_qty": q, "outbound_estimate_qty": q}


def build_plan_product(
    *,
    tenant_no: str,
    plan_id: str,
    plan_type: str,
    prod: PlanProductCreate,
    order_num: int,
    user_id: str,
    dept_id: str,
    product_row_id: Optional[str] = None,
) -> PlPlanProduct:
    qty = _product_quantity(prod)
    estimates = _estimate_qty_fields(plan_type, qty if qty else None)
    tax_price = prod.tax_price
    row = PlPlanProduct(
        id=(product_row_id or "").strip() or generate_uuid(),
        tenant_no=tenant_no,
        plan_id=plan_id,
        product_id=prod.product_id,
        product_no=prod.product_no,
        product_name=prod.product_name,
        brand_mode=prod.brand_mode,
        brand_id=prod.brand_id,
        brand_name=prod.brand_name,
        model=prod.model,
        unit=prod.unit or "吨",
        category_name=prod.category_name,
        package_spec=prod.package_spec,
        delivery_method=prod.delivery_method,
        quantity=qty if qty else None,
        actual_qty=Decimal(str(prod.actual_qty)) if prod.actual_qty is not None else None,
        tax_price=tax_price,
        tax_amount=prod.tax_amount,
        tax_subtotal=prod.tax_subtotal,
        is_explosive=prod.is_explosive,
        is_toxic=prod.is_toxic,
        purchase_cert_url=prod.purchase_cert_url,
        transport_cert_url=prod.transport_cert_url,
        legal_use_cert_url=prod.legal_use_cert_url,
        inbound_price=tax_price,
        outbound_price=tax_price,
        inbound_amount=prod.tax_subtotal,
        outbound_amount=prod.tax_subtotal,
        remark=prod.remark,
        order_num=order_num,
        create_by=user_id,
        create_dept_id=dept_id,
        **estimates,
    )

    bind_mobile_product_key(row, prod)
    return row


def apply_plan_product_fields(
    row: PlPlanProduct,
    *,
    plan_type: str,
    prod: PlanProductCreate,
    order_num: int,
    user_id: str,
) -> None:
    """只更新提交的字段；计划编辑不覆盖归档数量、单价与金额。"""
    values = prod.model_dump(exclude_unset=True, exclude={'id', 'client_key'})
    for field, value in values.items():
        setattr(row, field, value)
    if any(field in values for field in ('quantity', 'inbound_estimate_qty', 'outbound_estimate_qty')):
        qty = _product_quantity(prod)
        row.quantity = qty
        for field, value in _estimate_qty_fields(plan_type, qty).items():
            setattr(row, field, value)
    row.order_num = order_num
    row.update_by = user_id
    row.del_flag = 0
    bind_mobile_product_key(row, prod)


def prepare_plan_product_identities(products, existing):
    """Resolve one request before writes; omitted optional fields cannot change identity."""
    from app.services.pl_child_sync import validate_child_ids

    by_id, explicit_ids = validate_child_ids(products, existing, '计划产品')
    claimed, resolved = set(explicit_ids), []
    key_fields = ('model', 'package_spec', 'unit', 'brand_id', 'delivery_method')
    for product in products or []:
        row = by_id.get((product.id or '').strip())
        if row is None:
            candidates = [candidate for candidate in existing
                          if candidate.id not in explicit_ids
                          and candidate.product_id == product.product_id
                          and all((getattr(candidate, field, None) or '') == (getattr(product, field) or '')
                                  for field in key_fields if field in product.model_fields_set)]
            if len(candidates) > 1 or (candidates and candidates[0].id in claimed):
                raise HTTPException(422, '计划产品缺少 id 且业务键不唯一，请刷新后回传明细 id')
            if candidates:
                row = candidates[0]
                claimed.add(row.id)
        resolved.append((product, row, row.id if row is not None else generate_uuid()))
    # Restore only verified existing identities. New UUIDs stay in the prepared result,
    # never masquerading as client-supplied IDs in later validation.
    for product, row, _ in resolved:
        if row is not None:
            product.id = row.id
    return resolved, [row for row in existing if row.id not in claimed]


def resolve_plan_product_id_for_upsert(
    *,
    plan_id: str,
    payload_id: Optional[str],
    existing_by_id: dict,
) -> tuple[str, Optional[object]]:
    """解析计划产品行 upsert 主键：有合法 id 则保留；无 id 则新建；非法 id 则 422。"""
    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 _apply_vehicle_snapshot_to_plan_transport(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
    transport: Optional[PlTransport],
    vehicle: PlVehicle,
    driver_id: Optional[str],
    escort_id: Optional[str] = None,
) -> None:
    snapshot = build_transport_resource_snapshot(
        db,
        tenant_no,
        vehicle.id,
        selected_driver_id=driver_id,
        selected_escort_id=escort_id,
    )
    apply_resource_snapshot_json(plan=plan, transport=transport, snapshot=snapshot)


def _resolve_personnel_by_vehicle_role(
    db: Session,
    tenant_no: str,
    vehicle_id: str,
    personnel_id: Optional[str],
    *,
    role: str,
) -> tuple[Optional[str], Optional[str], Optional[str]]:
    """解析驾押人员：显式选择优先，否则取车辆关联该角色；均无则允许为空。"""
    if personnel_id:
        personnel = (
            db.query(PlPersonnel)
            .filter(
                PlPersonnel.id == personnel_id,
                PlPersonnel.tenant_no == tenant_no,
                PlPersonnel.del_flag == 0,
            )
            .first()
        )
        if personnel:
            return personnel.id, personnel.personnel_name, personnel.phone_no

    default_link = (
        db.query(PlVehiclePersonnel)
        .filter(
            PlVehiclePersonnel.vehicle_id == vehicle_id,
            PlVehiclePersonnel.tenant_no == tenant_no,
            PlVehiclePersonnel.del_flag == 0,
            PlVehiclePersonnel.role == role,
        )
        .order_by(PlVehiclePersonnel.is_default.desc(), PlVehiclePersonnel.order_num.asc())
        .first()
    )
    if default_link:
        personnel = (
            db.query(PlPersonnel)
            .filter(
                PlPersonnel.id == default_link.personnel_id,
                PlPersonnel.del_flag == 0,
            )
            .first()
        )
        if personnel:
            return personnel.id, personnel.personnel_name, personnel.phone_no

    return None, None, None


def _resolve_driver(
    db: Session,
    tenant_no: str,
    vehicle_id: str,
    driver_id: Optional[str],
) -> tuple[Optional[str], Optional[str], Optional[str]]:
    """解析司机：显式选择优先，否则取车辆关联司机；均无则允许为空。"""
    return _resolve_personnel_by_vehicle_role(
        db, tenant_no, vehicle_id, driver_id, role="driver"
    )


def _resolve_escort(
    db: Session,
    tenant_no: str,
    vehicle_id: str,
    escort_id: Optional[str],
) -> tuple[Optional[str], Optional[str], Optional[str]]:
    """解析押运员：显式选择优先，否则取车辆关联押运；均无则允许为空。"""
    return _resolve_personnel_by_vehicle_role(
        db, tenant_no, vehicle_id, escort_id, role="escort"
    )


def _calc_freight_amount(
    freight_type: str,
    products: list[PlPlanProduct],
    transport_qty: Optional[Decimal],
    freight_price: Optional[Decimal],
    freight_amount: Optional[Decimal],
) -> Decimal:
    if freight_type == "by_trip":
        return Decimal(str(freight_amount or 0))

    qty = transport_qty
    if qty is None:
        qty = sum_plan_product_actual_quantities(products)
    if freight_price is not None:
        computed = qty * Decimal(str(freight_price))
        if freight_amount is not None:
            return Decimal(str(freight_amount))
        return computed
    return Decimal(str(freight_amount or 0))


def fetch_hazard_flags_map(
    db: Session,
    tenant_no: str,
    product_ids: Optional[list[str]] = None,
) -> dict[str, tuple[int, int]]:
    """批量从 CRM 产品主数据读取易制毒/易制爆标识。"""
    from app.models.crm_models import CrmProduct

    ids = [str(pid) for pid in (product_ids or []) if pid]
    if not ids:
        return {}
    rows = (
        db.query(CrmProduct.id, CrmProduct.is_toxic, CrmProduct.is_explosive)
        .filter(
            CrmProduct.id.in_(ids),
            CrmProduct.tenant_no == tenant_no,
            CrmProduct.del_flag == 0,
        )
        .all()
    )
    return {
        str(row.id): (int(row.is_toxic or 0), int(row.is_explosive or 0))
        for row in rows
    }


def resolve_hazard_flags_from_crm(
    db: Session,
    tenant_no: str,
    product_id: Optional[str],
    *,
    hazard_map: Optional[dict[str, tuple[int, int]]] = None,
) -> tuple[int, int]:
    """从 CRM 产品主数据读取易制毒/易制爆标识（不读计划产品快照）。"""
    if not product_id:
        return 0, 0
    if hazard_map is not None:
        return hazard_map.get(str(product_id), (0, 0))
    from app.models.crm_models import CrmProduct

    row = (
        db.query(CrmProduct)
        .filter(
            CrmProduct.id == product_id,
            CrmProduct.tenant_no == tenant_no,
            CrmProduct.del_flag == 0,
        )
        .first()
    )
    if not row:
        return 0, 0
    return int(row.is_toxic or 0), int(row.is_explosive or 0)


def plan_has_hazard_products(
    db: Session,
    tenant_no: str,
    products: list[PlPlanProduct],
) -> bool:
    return bool(collect_hazard_product_ids(db, tenant_no, products))


def _cert_url_blank(column):
    """证书 URL 为空（NULL / 空串 / 纯空白）。"""
    return func.nullif(func.trim(func.coalesce(column, "")), "").is_(None)


def plan_has_hazard_product_exists(tenant_no: str):
    """列表筛选：计划下存在 CRM 标识为易制毒/易制爆的产品。"""
    return exists(
        select(PlPlanProduct.id)
        .select_from(PlPlanProduct)
        .join(
            CrmProduct,
            and_(
                CrmProduct.id == PlPlanProduct.product_id,
                CrmProduct.tenant_no == tenant_no,
                CrmProduct.del_flag == 0,
            ),
        )
        .where(
            PlPlanProduct.plan_id == PlPlan.id,
            PlPlanProduct.tenant_no == tenant_no,
            PlPlanProduct.del_flag == 0,
            or_(CrmProduct.is_toxic == 1, CrmProduct.is_explosive == 1),
        )
    )


def plan_has_incomplete_hazard_cert_exists(tenant_no: str):
    """列表筛选：存在危化产品且至少一张必需证书未上传。"""
    toxic_incomplete = and_(
        CrmProduct.is_toxic == 1,
        or_(
            _cert_url_blank(PlPlanProduct.purchase_cert_url),
            _cert_url_blank(PlPlanProduct.transport_cert_url),
        ),
    )
    explosive_incomplete = and_(
        CrmProduct.is_explosive == 1,
        _cert_url_blank(PlPlanProduct.legal_use_cert_url),
    )
    return exists(
        select(PlPlanProduct.id)
        .select_from(PlPlanProduct)
        .join(
            CrmProduct,
            and_(
                CrmProduct.id == PlPlanProduct.product_id,
                CrmProduct.tenant_no == tenant_no,
                CrmProduct.del_flag == 0,
            ),
        )
        .where(
            PlPlanProduct.plan_id == PlPlan.id,
            PlPlanProduct.tenant_no == tenant_no,
            PlPlanProduct.del_flag == 0,
            or_(toxic_incomplete, explosive_incomplete),
        )
    )


def collect_hazard_product_ids(
    db: Session,
    tenant_no: str,
    products: list[Union[PlPlanProduct, PlanProductCreate]],
) -> frozenset[str]:
    product_ids = [str(p.product_id) for p in products if getattr(p, "product_id", None)]
    if not product_ids:
        return frozenset()
    hazard_map = fetch_hazard_flags_map(db, tenant_no, product_ids)
    hazard_ids: set[str] = set()
    for prod in products:
        product_id = getattr(prod, "product_id", None)
        if not product_id:
            continue
        is_toxic, is_explosive = resolve_hazard_flags_from_crm(
            db,
            tenant_no,
            product_id,
            hazard_map=hazard_map,
        )
        if is_toxic or is_explosive:
            hazard_ids.add(str(product_id))
    return frozenset(hazard_ids)


def clear_hazard_cert_ready(plan: PlPlan) -> None:
    plan.hazard_cert_ready = 0
    plan.hazard_cert_ready_by = None
    plan.hazard_cert_ready_time = None


def _required_hazard_cert_fields(is_toxic: int, is_explosive: int) -> list[str]:
    fields: list[str] = []
    if is_toxic:
        fields.extend(["purchase_cert_url", "transport_cert_url"])
    if is_explosive:
        fields.append("legal_use_cert_url")
    return fields


def _is_hazard_product_cert_complete(
    product: PlPlanProduct,
    *,
    is_toxic: int,
    is_explosive: int,
) -> bool:
    for field in _required_hazard_cert_fields(is_toxic, is_explosive):
        if not str(getattr(product, field, "") or "").strip():
            return False
    return True


def all_hazard_products_cert_complete(
    db: Session,
    tenant_no: str,
    products: list[PlPlanProduct],
) -> bool:
    """计划内所有危化品产品的所需证书槽位均已上传。"""
    if not products:
        return True
    hazard_map = fetch_hazard_flags_map(
        db,
        tenant_no,
        [product.product_id for product in products if product.product_id],
    )
    has_hazard = False
    for product in products:
        is_toxic, is_explosive = resolve_hazard_flags_from_crm(
            db,
            tenant_no,
            product.product_id,
            hazard_map=hazard_map,
        )
        if not (is_toxic or is_explosive):
            continue
        has_hazard = True
        if not _is_hazard_product_cert_complete(
            product,
            is_toxic=is_toxic,
            is_explosive=is_explosive,
        ):
            return False
    return has_hazard


def maybe_auto_set_hazard_cert_ready(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
    products: list[PlPlanProduct],
    user_id: str,
) -> None:
    """全部危化品证书已上传时，自动标记证书齐备（与人工确认等效）。"""
    if int(plan.hazard_cert_ready or 0):
        return
    if not all_hazard_products_cert_complete(db, tenant_no, products):
        return
    plan.hazard_cert_ready = 1
    plan.hazard_cert_ready_by = user_id
    plan.hazard_cert_ready_time = datetime.now()


def validate_hazard_cert_ready_for_transport(
    db: Session,
    tenant_no: str,
    plan: PlPlan,
    products: list[PlPlanProduct],
) -> None:
    if not plan_has_hazard_products(db, tenant_no, products):
        return
    if int(plan.hazard_cert_ready or 0):
        return
    if all_hazard_products_cert_complete(db, tenant_no, products):
        return
    raise HTTPException(
        status_code=400,
        detail="该计划危化品证书尚未确认齐备，请至危化品证上传页操作",
    )


def set_hazard_cert_ready(
    db: Session,
    *,
    tenant_no: str,
    plan_id: str,
    ready: bool,
    user_id: str,
) -> PlPlan:
    plan = (
        db.query(PlPlan)
        .filter(
            PlPlan.id == plan_id,
            PlPlan.tenant_no == tenant_no,
            PlPlan.del_flag == 0,
        )
        .first()
    )
    if not plan:
        raise HTTPException(status_code=404, detail="计划不存在")
    if plan.archive_status == "archived":
        raise HTTPException(status_code=400, detail="已归档的计划不能修改证书齐备状态")
    if plan.follow_status == "cancelled":
        raise HTTPException(status_code=400, detail="已作废的计划不能修改证书齐备状态")

    products = (
        db.query(PlPlanProduct)
        .filter(
            PlPlanProduct.plan_id == plan_id,
            PlPlanProduct.tenant_no == tenant_no,
            PlPlanProduct.del_flag == 0,
        )
        .all()
    )
    has_hazard = plan_has_hazard_products(db, tenant_no, products)

    if ready:
        if not has_hazard:
            raise HTTPException(status_code=400, detail="该计划不含易制毒/易制爆产品，无需确认证书齐备")
        plan.hazard_cert_ready = 1
        plan.hazard_cert_ready_by = user_id
        plan.hazard_cert_ready_time = datetime.now()
    else:
        clear_hazard_cert_ready(plan)

    plan.update_by = user_id
    db.commit()
    db.refresh(plan)
    return plan


def set_plan_erp_entered(
    db: Session,
    *,
    tenant_no: str,
    plan_id: str,
    entered: bool,
    user_id: str,
) -> PlPlan:
    plan = (
        db.query(PlPlan)
        .filter(
            PlPlan.id == plan_id,
            PlPlan.tenant_no == tenant_no,
            PlPlan.del_flag == 0,
        )
        .first()
    )
    if not plan:
        raise HTTPException(status_code=404, detail="计划不存在")
    if plan.follow_status == "cancelled":
        raise HTTPException(status_code=400, detail="已作废的计划不能修改ERP录入状态")

    plan.erp_entered = 1 if entered else 0
    plan.update_by = user_id
    db.commit()
    db.refresh(plan)
    return plan


def update_plan_product_certs(
    db: Session,
    *,
    tenant_no: str,
    plan_id: str,
    items: list,
    user_id: str,
) -> list[PlPlanProduct]:
    plan = (
        db.query(PlPlan)
        .filter(
            PlPlan.id == plan_id,
            PlPlan.tenant_no == tenant_no,
            PlPlan.del_flag == 0,
        )
        .first()
    )
    if not plan:
        raise HTTPException(status_code=404, detail="计划不存在")
    if plan.archive_status == "archived":
        raise HTTPException(status_code=400, detail="已归档的计划不能修改")
    if plan.follow_status == "cancelled":
        raise HTTPException(status_code=400, detail="已作废的计划不能修改危化品证书")

    existing_rows = (
        db.query(PlPlanProduct)
        .filter(
            PlPlanProduct.plan_id == plan_id,
            PlPlanProduct.tenant_no == tenant_no,
            PlPlanProduct.del_flag == 0,
        )
        .all()
    )
    hazard_map = fetch_hazard_flags_map(
        db,
        tenant_no,
        [row.product_id for row in existing_rows if row.product_id],
    )

    updated: list[PlPlanProduct] = []
    for item in items:
        row = (
            db.query(PlPlanProduct)
            .filter(
                PlPlanProduct.id == item.id,
                PlPlanProduct.plan_id == plan_id,
                PlPlanProduct.tenant_no == tenant_no,
                PlPlanProduct.del_flag == 0,
            )
            .first()
        )
        if not row:
            raise HTTPException(status_code=404, detail=f"计划产品不存在: {item.id}")
        is_toxic, is_explosive = resolve_hazard_flags_from_crm(
            db, tenant_no, row.product_id, hazard_map=hazard_map
        )
        if not (is_toxic or is_explosive):
            raise HTTPException(status_code=400, detail=f"产品【{row.product_name}】无需上传危化品证书")
        row.purchase_cert_url = (item.purchase_cert_url or "").strip() or None
        row.transport_cert_url = (item.transport_cert_url or "").strip() or None
        row.legal_use_cert_url = (item.legal_use_cert_url or "").strip() or None
        row.update_by = user_id
        updated.append(row)

    maybe_auto_set_hazard_cert_ready(
        db,
        tenant_no=tenant_no,
        plan=plan,
        products=existing_rows,
        user_id=user_id,
    )
    plan.update_by = user_id
    db.commit()
    for row in updated:
        db.refresh(row)
    return updated


def _generate_transport_no(db: Session) -> str:
    now = date.today()
    prefix = f"SXYS{now.strftime('%Y%m')}"
    last = (
        db.query(PlTransport)
        .filter(PlTransport.transport_no.like(f"{prefix}%"))
        .order_by(PlTransport.transport_no.desc())
        .first()
    )
    seq = 1
    if last and last.transport_no:
        try:
            seq = int(last.transport_no[-6:]) + 1
        except ValueError:
            seq = 1
    return f"{prefix}{seq:06d}"


def _resolve_transport_context(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
    transport_payload: PlanTransportCreateBlock,
    products: list[PlPlanProduct],
) -> tuple[
    PlVehicle,
    Optional[str],
    Optional[str],
    Optional[str],
    Optional[str],
    Optional[str],
    Optional[str],
    Optional[str],
    Optional[str],
    Decimal,
    Optional[Decimal],
]:
    vehicle = (
        db.query(PlVehicle)
        .filter(
            PlVehicle.id == transport_payload.vehicle_id,
            PlVehicle.tenant_no == tenant_no,
            PlVehicle.del_flag == 0,
        )
        .first()
    )
    if not vehicle:
        raise HTTPException(status_code=400, detail="车辆不存在")

    driver_id, driver_name, driver_phone = _resolve_driver(
        db, tenant_no, transport_payload.vehicle_id, transport_payload.driver_id
    )
    escort_id, escort_name, escort_phone = _resolve_escort(
        db, tenant_no, transport_payload.vehicle_id, transport_payload.escort_id
    )
    carrier_id = transport_payload.carrier_id or vehicle.carrier_id
    carrier_name = vehicle.carrier_name
    if carrier_id:
        carrier = (
            db.query(PlCarrier)
            .filter(PlCarrier.id == carrier_id, PlCarrier.del_flag == 0)
            .first()
        )
        if carrier:
            carrier_name = carrier.carrier_name

    freight_amount = _calc_freight_amount(
        transport_payload.freight_type,
        products,
        transport_payload.transport_qty,
        transport_payload.freight_price,
        transport_payload.freight_amount,
    )
    transport_qty = transport_payload.transport_qty
    if transport_qty is None and transport_payload.freight_type == "by_weight":
        transport_qty = sum_plan_product_actual_quantities(products)

    return (
        vehicle,
        driver_id,
        driver_name,
        driver_phone,
        escort_id,
        escort_name,
        escort_phone,
        carrier_id,
        carrier_name,
        freight_amount,
        transport_qty,
    )


def _apply_payload_to_transport_row(
    transport: PlTransport,
    *,
    vehicle: PlVehicle,
    driver_id: Optional[str],
    driver_name: Optional[str],
    driver_phone: Optional[str],
    escort_id: Optional[str],
    escort_name: Optional[str],
    escort_phone: Optional[str],
    carrier_id: Optional[str],
    carrier_name: Optional[str],
    plan: PlPlan,
    transport_payload: PlanTransportCreateBlock,
    freight_amount: Decimal,
    transport_qty: Optional[Decimal],
    user_id: str,
) -> None:
    transport.vehicle_id = vehicle.id
    transport.license_plate = vehicle.license_plate
    transport.driver_id = driver_id
    transport.driver_name = driver_name
    transport.driver_phone = driver_phone
    transport.escort_id = escort_id
    transport.escort_name = escort_name
    transport.escort_phone = escort_phone
    transport.carrier_id = carrier_id
    transport.carrier_name = carrier_name
    transport.start_point = plan.pickup_address
    transport.end_point = plan.destination_address or plan.destination_name
    transport.distance_km = transport_payload.distance_km
    transport.transport_qty = transport_qty
    transport.freight_price = transport_payload.freight_price
    transport.freight_amount = freight_amount or 0
    transport.freight_type = transport_payload.freight_type
    if transport_payload.has_invoice is not None:
        transport.has_invoice = transport_payload.has_invoice
    transport.unit = "吨" if transport_payload.freight_type == "by_weight" else "趟"
    if transport_payload.remark is not None:
        transport.remark = transport_payload.remark
    transport.update_by = user_id


def upsert_draft_transport_for_plan(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
    transport_payload: PlanTransportCreateBlock,
    products: list[PlPlanProduct],
    user_id: str,
    dept_id: str,
) -> None:
    """写入/更新草稿运输单（pl_transport + pl_plan_transport），计划保持待安排。"""
    (
        vehicle,
        driver_id,
        driver_name,
        driver_phone,
        escort_id,
        escort_name,
        escort_phone,
        carrier_id,
        carrier_name,
        freight_amount,
        transport_qty,
    ) = _resolve_transport_context(
        db,
        tenant_no=tenant_no,
        plan=plan,
        transport_payload=transport_payload,
        products=products,
    )

    transport: Optional[PlTransport] = None
    if plan.transport_id:
        transport = (
            db.query(PlTransport)
            .filter(
                PlTransport.id == plan.transport_id,
                PlTransport.tenant_no == tenant_no,
                PlTransport.del_flag == 0,
            )
            .first()
        )

    if transport:
        _apply_payload_to_transport_row(
            transport,
            vehicle=vehicle,
            driver_id=driver_id,
            driver_name=driver_name,
            driver_phone=driver_phone,
            escort_id=escort_id,
            escort_name=escort_name,
            escort_phone=escort_phone,
            carrier_id=carrier_id,
            carrier_name=carrier_name,
            plan=plan,
            transport_payload=transport_payload,
            freight_amount=freight_amount,
            transport_qty=transport_qty,
            user_id=user_id,
        )
        if transport.transport_status == "draft":
            pass
    else:
        transport_no = _generate_transport_no(db)
        transport = PlTransport(
            id=generate_uuid(),
            tenant_no=tenant_no,
            transport_no=transport_no,
            transport_status="draft",
            approval_status="none",
            remark=transport_payload.remark,
            create_by=user_id,
            create_dept_id=dept_id,
        )
        db.add(transport)
        db.flush()
        db.add(
            PlPlanTransport(
                id=generate_uuid(),
                tenant_no=tenant_no,
                plan_id=plan.id,
                transport_id=transport.id,
                create_by=user_id,
                create_dept_id=dept_id,
            )
        )
        plan.transport_id = transport.id
        plan.transport_no = transport_no
        _apply_payload_to_transport_row(
            transport,
            vehicle=vehicle,
            driver_id=driver_id,
            driver_name=driver_name,
            driver_phone=driver_phone,
            escort_id=escort_id,
            escort_name=escort_name,
            escort_phone=escort_phone,
            carrier_id=carrier_id,
            carrier_name=carrier_name,
            plan=plan,
            transport_payload=transport_payload,
            freight_amount=freight_amount,
            transport_qty=transport_qty,
            user_id=user_id,
        )

    plan.vehicle_id = vehicle.id
    plan.license_plate = vehicle.license_plate
    plan.driver_id = driver_id
    plan.driver_name = driver_name
    plan.escort_id = escort_id
    plan.escort_name = escort_name
    plan.escort_phone = escort_phone
    _apply_vehicle_snapshot_to_plan_transport(
        db,
        tenant_no=tenant_no,
        plan=plan,
        transport=transport,
        vehicle=vehicle,
        driver_id=driver_id,
        escort_id=escort_id,
    )
    plan.freight_amount = freight_amount or 0
    if plan.transport_id:
        sync_draft_settlement_freight_for_transport(
            db,
            tenant_no=tenant_no,
            transport_id=plan.transport_id,
            freight_amount=freight_amount or Decimal("0"),
            user_id=user_id,
        )
    if plan.transport_status != "arranged":
        plan.transport_status = "pending"


def clear_draft_transport_for_plan(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
    user_id: str,
) -> None:
    """清除计划上的草稿运输单及车辆预选。"""
    transport: Optional[PlTransport] = None
    if plan.transport_id:
        transport = (
            db.query(PlTransport)
            .filter(
                PlTransport.id == plan.transport_id,
                PlTransport.tenant_no == tenant_no,
                PlTransport.del_flag == 0,
            )
            .first()
        )
        if transport and transport.transport_status == "draft":
            transport.del_flag = 1
            transport.update_by = user_id
            db.query(PlPlanTransport).filter(
                PlPlanTransport.plan_id == plan.id,
                PlPlanTransport.transport_id == transport.id,
                PlPlanTransport.tenant_no == tenant_no,
                PlPlanTransport.del_flag == 0,
            ).update({"del_flag": 1, "update_by": user_id})
        plan.transport_id = None
        plan.transport_no = None

    plan.vehicle_id = None
    plan.license_plate = None
    plan.driver_id = None
    plan.driver_name = None
    plan.escort_id = None
    plan.escort_name = None
    plan.escort_phone = None
    clear_resource_snapshot(plan=plan, transport=transport)
    plan.freight_amount = 0
    if plan.transport_status != "arranged":
        plan.transport_status = "pending"


def sync_transport_prefill_for_plan(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
    transport_payload: PlanTransportCreateBlock,
    products: list[PlPlanProduct],
    user_id: str,
    dept_id: str = "",
) -> None:
    """同步运输预选到草稿运输子表（pl_transport）。"""
    upsert_draft_transport_for_plan(
        db,
        tenant_no=tenant_no,
        plan=plan,
        transport_payload=transport_payload,
        products=products,
        user_id=user_id,
        dept_id=dept_id or user_id,
    )


def apply_vehicle_prefill_to_plan(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
    transport_payload: PlanTransportCreateBlock,
    products: list[PlPlanProduct],
    user_id: str = "",
    dept_id: str = "",
) -> None:
    """兼容旧调用：写入草稿运输子表。"""
    upsert_draft_transport_for_plan(
        db,
        tenant_no=tenant_no,
        plan=plan,
        transport_payload=transport_payload,
        products=products,
        user_id=user_id,
        dept_id=dept_id or user_id,
    )


def create_transport_for_plan(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
    products: list[PlPlanProduct],
    transport_payload: PlanTransportCreateBlock,
    user_id: str,
    dept_id: str,
    generate_transport_no,
) -> PlTransport:
    vehicle = (
        db.query(PlVehicle)
        .filter(
            PlVehicle.id == transport_payload.vehicle_id,
            PlVehicle.tenant_no == tenant_no,
            PlVehicle.del_flag == 0,
        )
        .first()
    )
    if not vehicle:
        raise HTTPException(status_code=400, detail="车辆不存在")

    validate_hazard_cert_ready_for_transport(db, tenant_no, plan, products)

    driver_id, driver_name, driver_phone = _resolve_driver(
        db, tenant_no, transport_payload.vehicle_id, transport_payload.driver_id
    )
    escort_id, escort_name, escort_phone = _resolve_escort(
        db, tenant_no, transport_payload.vehicle_id, transport_payload.escort_id
    )

    carrier_name = vehicle.carrier_name
    carrier_id = transport_payload.carrier_id or vehicle.carrier_id
    if carrier_id:
        carrier = (
            db.query(PlCarrier)
            .filter(PlCarrier.id == carrier_id, PlCarrier.del_flag == 0)
            .first()
        )
        if carrier:
            carrier_name = carrier.carrier_name

    freight_amount = _calc_freight_amount(
        transport_payload.freight_type,
        products,
        transport_payload.transport_qty,
        transport_payload.freight_price,
        transport_payload.freight_amount,
    )

    transport_qty = transport_payload.transport_qty
    if transport_qty is None and transport_payload.freight_type == "by_weight":
        transport_qty = sum_plan_product_actual_quantities(products)

    transport_no = generate_transport_no(db)
    transport = PlTransport(
        id=generate_uuid(),
        tenant_no=tenant_no,
        transport_no=transport_no,
        vehicle_id=vehicle.id,
        license_plate=vehicle.license_plate,
        driver_id=driver_id,
        driver_name=driver_name,
        driver_phone=driver_phone,
        escort_id=escort_id,
        escort_name=escort_name,
        escort_phone=escort_phone,
        carrier_id=carrier_id,
        carrier_name=carrier_name,
        start_point=plan.pickup_address,
        end_point=plan.destination_address or plan.destination_name,
        distance_km=transport_payload.distance_km,
        unit="吨" if transport_payload.freight_type == "by_weight" else "趟",
        transport_qty=transport_qty,
        freight_price=transport_payload.freight_price,
        freight_amount=freight_amount or 0,
        freight_type=transport_payload.freight_type,
        has_invoice=transport_payload.has_invoice if transport_payload.has_invoice is not None else 0,
        transport_status="draft",
        approval_status="none",
        remark=transport_payload.remark,
        create_by=user_id,
        create_dept_id=dept_id,
    )
    db.add(transport)

    db.add(
        PlPlanTransport(
            id=generate_uuid(),
            tenant_no=tenant_no,
            plan_id=plan.id,
            transport_id=transport.id,
            create_by=user_id,
            create_dept_id=dept_id,
        )
    )

    plan.transport_id = transport.id
    plan.transport_no = transport_no
    plan.vehicle_id = vehicle.id
    plan.license_plate = vehicle.license_plate
    plan.driver_id = driver_id
    plan.driver_name = driver_name
    plan.escort_id = escort_id
    plan.escort_name = escort_name
    plan.escort_phone = escort_phone
    _apply_vehicle_snapshot_to_plan_transport(
        db,
        tenant_no=tenant_no,
        plan=plan,
        transport=transport,
        vehicle=vehicle,
        driver_id=driver_id,
        escort_id=escort_id,
    )
    plan.transport_status = "arranged"
    plan.follow_status = "arranged"
    plan.freight_amount = freight_amount or 0

    return transport


PLAN_SELECTABLE_CONTRACT_STATUSES = ("approved", "executing")
PLAN_CONTRACT_SIDE_SUPPLIER = "supplier"
PLAN_CONTRACT_SIDE_DEMANDER = "demander"


PLAN_BELONGS_COMPANY_TRADING_ENTITY = {
    "shanxia": "安庆市山峡化工有限公司",
    "yundong": "安徽云栋化工有限公司",
}


def belongs_company_trading_entity(belongs_company: Optional[str]) -> str:
    return PLAN_BELONGS_COMPANY_TRADING_ENTITY.get((belongs_company or "").strip(), "")


def is_belongs_company_contract_error(exc: HTTPException) -> bool:
    detail = exc.detail if isinstance(exc.detail, str) else ""
    return detail == "请先选择归属公司" or detail.endswith("的交易主体与归属公司不一致")


def assert_contract_selectable_for_plan(
    db: Session,
    contract_id: str,
    tenant_no: str,
    *,
    expected_customer_id: Optional[str] = None,
    expected_supplier_name: Optional[str] = None,
    side_label: str = "合同",
):
    from app.crud.crm.supplier_info import get_supplier_info
    from app.models.crm_models import CrmContract

    contract = (
        db.query(CrmContract)
        .filter(
            CrmContract.id == contract_id,
            CrmContract.tenant_no == tenant_no,
            CrmContract.del_flag == 0,
        )
        .first()
    )
    if not contract:
        raise HTTPException(status_code=404, detail=f"{side_label}不存在")
    if contract.status not in PLAN_SELECTABLE_CONTRACT_STATUSES:
        raise HTTPException(status_code=422, detail=f"仅已审批或执行中的合同可关联运输计划（{side_label}）")
    if expected_customer_id and contract.customer_id != expected_customer_id:
        raise HTTPException(status_code=422, detail=f"{side_label}不属于所选客户")
    expected_name = (expected_supplier_name or "").strip()
    if expected_name:
        supplier = (
            get_supplier_info(db, contract.supplier_id, tenant_no=tenant_no)
            if contract.supplier_id
            else None
        )
        actual = (supplier.supplier_name if supplier else "") or ""
        if actual.strip() != expected_name:
            raise HTTPException(status_code=422, detail=f"{side_label}的交易主体与归属公司不一致")
    return contract


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 validate_plan_party_contracts(
    db: Session,
    tenant_no: str,
    *,
    supplier_id: Optional[str],
    supplier_name: Optional[str],
    customer_id: Optional[str],
    customer_name: Optional[str],
    supplier_contracts: Optional[list[PlanContractLink]],
    demander_contracts: Optional[list[PlanContractLink]],
    require_products: Optional[list] = None,
) -> tuple[list[PlanContractLink], list[PlanContractLink]]:
    if not (supplier_id or "").strip() or not (supplier_name or "").strip():
        raise HTTPException(status_code=422, detail="请选择供方客户")
    if not (customer_id or "").strip() or not (customer_name or "").strip():
        raise HTTPException(status_code=422, detail="请选择需方客户")
    if require_products is not None and not require_products:
        raise HTTPException(status_code=422, detail="请至少保留一条产品明细")

    supplier_links = _normalize_contract_links(supplier_contracts)
    demander_links = _normalize_contract_links(demander_contracts)
    # Local import: pl_plan_stop_service imports this module at top level.
    from app.services.pl_plan_stop_service import assert_at_most_one_contract

    supplier_links = assert_at_most_one_contract(supplier_links, "供方合同")
    demander_links = assert_at_most_one_contract(demander_links, "需方合同")

    resolved_supplier: list[PlanContractLink] = []
    for link in supplier_links:
        contract = assert_contract_selectable_for_plan(
            db,
            link.contract_id,
            tenant_no,
            expected_customer_id=supplier_id,
            side_label="供方合同",
        )
        resolved_supplier.append(
            PlanContractLink(contractId=contract.id, contractNo=contract.contract_no or link.contract_no)
        )

    resolved_demander: list[PlanContractLink] = []
    for link in demander_links:
        contract = assert_contract_selectable_for_plan(
            db,
            link.contract_id,
            tenant_no,
            expected_customer_id=customer_id,
            side_label="需方合同",
        )
        resolved_demander.append(
            PlanContractLink(contractId=contract.id, contractNo=contract.contract_no or link.contract_no)
        )

    return resolved_supplier, resolved_demander


def prepare_plan_contract_scope(
    db, *, tenant_no, plan_id, side, stop_id=None, contracts=None,
):
    from app.services.pl_child_sync import prepare_keyed_rows
    if contracts is None:
        return None
    scope = db.query(PlPlanContract).filter(
        PlPlanContract.tenant_no == tenant_no, PlPlanContract.plan_id == plan_id,
        PlPlanContract.side == side, PlPlanContract.del_flag == 0)
    if stop_id:
        scope = scope.filter(PlPlanContract.stop_id == stop_id)
    else:
        scope = scope.filter(or_(PlPlanContract.stop_id.is_(None), PlPlanContract.stop_id == ''))
    values = [dict(tenant_no=tenant_no, plan_id=plan_id, side=side, stop_id=stop_id,
                   contract_id=link.contract_id, contract_no=link.contract_no) for link in contracts]
    return prepare_keyed_rows(scope.all(), values, ('contract_id',), '计划合同关联')


def sync_plan_contract_scope(
    db, *, tenant_no, plan_id, side, stop_id=None, contracts=None,
    user_id=None, dept_id=None, order_start=0,
):
    from app.services.pl_child_sync import apply_keyed_rows
    prepared = prepare_plan_contract_scope(db, tenant_no=tenant_no, plan_id=plan_id,
        side=side, stop_id=stop_id, contracts=contracts)
    if prepared is not None:
        apply_keyed_rows(db, PlPlanContract, prepared, user_id=user_id,
                         dept_id=dept_id, order_start=order_start)


def replace_plan_contracts(
    db: Session, *, tenant_no: str, plan_id: str,
    supplier_contracts: Optional[list[PlanContractLink]],
    demander_contracts: Optional[list[PlanContractLink]],
    user_id: Optional[str], dept_id: Optional[str],
) -> None:
    from app.services.pl_child_sync import apply_keyed_rows
    prepared = [prepare_plan_contract_scope(db, tenant_no=tenant_no, plan_id=plan_id,
                side=side, contracts=links) for side, links in
                ((PLAN_CONTRACT_SIDE_SUPPLIER, supplier_contracts),
                 (PLAN_CONTRACT_SIDE_DEMANDER, demander_contracts))]
    for scope in prepared:
        if scope is not None:
            apply_keyed_rows(db, PlPlanContract, scope, user_id=user_id, dept_id=dept_id)


def load_plan_contracts_grouped(
    db: Session,
    tenant_no: str,
    plan_ids: list[str],
) -> dict[str, tuple[list[PlanContractLink], list[PlanContractLink]]]:
    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()
    )
    buckets: dict[str, list[list[PlanContractLink]]] = {
        pid: [[], []] for pid in plan_ids
    }
    for row in rows:
        bucket = buckets.setdefault(row.plan_id, [[], []])
        link = PlanContractLink(
            contractId=row.contract_id,
            contractNo=row.contract_no,
            stopId=row.stop_id,
        )
        if row.side == PLAN_CONTRACT_SIDE_SUPPLIER:
            bucket[0].append(link)
        elif row.side == PLAN_CONTRACT_SIDE_DEMANDER:
            bucket[1].append(link)
    return {pid: (lists[0], lists[1]) for pid, lists in buckets.items()}


DRAFTABLE_FOLLOW_STATUSES = frozenset({"drafting", "pending_arrange"})


def validate_draft_min_fields(*, belongs_company: Optional[str], create_date) -> None:
    if not (belongs_company or "").strip():
        raise HTTPException(status_code=422, detail="请选择归属公司")
    # create_date 允许为空，由 create/update 用 _resolve_create_date 兜底当天


def resolve_follow_status_on_save(
    *,
    save_mode: str,
    current: Optional[str],
    is_create: bool,
) -> str:
    mode = (save_mode or "formal").strip() or "formal"
    if mode == "draft":
        return "drafting"
    if is_create:
        return "pending_arrange"
    if current == "drafting":
        return "pending_arrange"
    return (current or "pending_arrange").strip() or "pending_arrange"


def assert_can_save_as_draft(current_status: Optional[str]) -> None:
    status = (current_status or "").strip()
    if status not in DRAFTABLE_FOLLOW_STATUSES:
        raise HTTPException(status_code=400, detail="当前状态不可暂存")


def validate_create_full(payload: PlanCreateFull) -> None:
    if not (payload.supplier_id or "").strip() or not (payload.supplier_name or "").strip():
        raise HTTPException(status_code=422, detail="请选择供方客户")
    if not (payload.customer_id or "").strip() or not (payload.customer_name or "").strip():
        raise HTTPException(status_code=422, detail="请选择需方客户")
    if not payload.products:
        raise HTTPException(status_code=422, detail="请至少保留一条产品明细")

    transport = payload.transport
    if not transport or not transport.vehicle_id:
        return

    if payload.has_freight:
        if transport.freight_type == "by_weight":
            if not transport.freight_price or Decimal(str(transport.freight_price)) <= 0:
                raise HTTPException(status_code=422, detail="按重量计费时请填写运输单价")
            qty = transport.transport_qty or sum_product_actual_quantities(payload.products)
            if qty <= 0:
                raise HTTPException(status_code=422, detail="运输量必须大于0，请填写产品实际数量或运输量")
        elif transport.freight_type == "by_trip":
            if not transport.freight_amount or Decimal(str(transport.freight_amount)) <= 0:
                raise HTTPException(status_code=422, detail="按趟计费时请填写运费金额")


def calc_plan_total_amount(products: list[PlanProductCreate]) -> Decimal:
    total = Decimal("0")
    for prod in products:
        if prod.tax_subtotal is not None:
            total += Decimal(str(prod.tax_subtotal))
    return total


_LOCKED_SETTLEMENT_STATUSES = ("approving", "approved", "confirmed")


def transport_has_locked_freight_settlement(
    db: Session, tenant_no: str, transport_id: Optional[str]
) -> bool:
    if not transport_id:
        return False
    return (
        db.query(PlFreightSettlementTransport.id)
        .join(
            PlFreightSettlement,
            PlFreightSettlement.id == PlFreightSettlementTransport.settlement_id,
        )
        .filter(
            PlFreightSettlementTransport.transport_id == transport_id,
            PlFreightSettlementTransport.tenant_no == tenant_no,
            PlFreightSettlementTransport.del_flag == 0,
            PlFreightSettlement.tenant_no == tenant_no,
            PlFreightSettlement.del_flag == 0,
            PlFreightSettlement.settlement_status.in_(_LOCKED_SETTLEMENT_STATUSES),
        )
        .first()
        is not None
    )


def sync_archive_qty_from_io_order(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
    items: list[PlanArchiveProductItem],
) -> list[PlanArchiveProductItem]:
    """若存在入库/出库单，用明细 actual_qty 覆盖归档提交中的实际数量（权威来源）。"""
    from app.models.pl_models import PlInboundOrder, PlInboundOrderItem, PlOutboundOrder, PlOutboundOrderItem

    if plan.plan_type not in ("inbound", "outbound"):
        return items
    header_m = PlInboundOrder if plan.plan_type == "inbound" else PlOutboundOrder
    item_m = PlInboundOrderItem if plan.plan_type == "inbound" else PlOutboundOrderItem
    header = (
        db.query(header_m)
        .filter(
            header_m.tenant_no == tenant_no,
            header_m.plan_id == plan.id,
            header_m.status != "voided",
            header_m.del_flag == 0,
        )
        .order_by(header_m.create_time.desc())
        .first()
    )
    if not header:
        return items
    io_items = (
        db.query(item_m)
        .filter(item_m.order_id == header.id, item_m.del_flag == 0)
        .all()
    )
    qty_by_plan_product = {
        row.plan_product_id: Decimal(str(row.actual_qty))
        for row in io_items
        if row.actual_qty is not None
    }
    if not qty_by_plan_product:
        return items
    out: list[PlanArchiveProductItem] = []
    for item in items:
        qty = qty_by_plan_product.get(item.id)
        if qty is not None:
            out.append(
                PlanArchiveProductItem(
                    id=item.id,
                    actual_qty=qty,
                    inbound_qty=item.inbound_qty,
                    inbound_price=item.inbound_price,
                    outbound_qty=item.outbound_qty,
                    outbound_price=item.outbound_price,
                    brand_qtys=item.brand_qtys,
                )
            )
        else:
            out.append(item)
    return out


def apply_archive_product_quantities(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
    items: list[PlanArchiveProductItem],
    user_id: str,
    sync_from_io_order: bool = True,
) -> None:
    if sync_from_io_order:
        items = sync_archive_qty_from_io_order(db, tenant_no=tenant_no, plan=plan, items=items)
    product_map = {
        row.id: row
        for row in db.query(PlPlanProduct).filter(
            PlPlanProduct.plan_id == plan.id,
            PlPlanProduct.tenant_no == tenant_no,
            PlPlanProduct.del_flag == 0,
        ).all()
    }
    for item in items:
        prod = product_map.get(item.id)
        if not prod:
            raise HTTPException(status_code=400, detail=f"产品明细不存在: {item.id}")
        actual = (
            Decimal(str(item.actual_qty)) if item.actual_qty is not None else None
        )
        if actual is not None:
            prod.actual_qty = actual

        if item.inbound_qty is not None:
            prod.inbound_qty = Decimal(str(item.inbound_qty))
        elif actual is not None and plan.plan_type == "inbound":
            prod.inbound_qty = actual
        if item.inbound_price is not None:
            prod.inbound_price = Decimal(str(item.inbound_price))
        inbound_qty = prod.inbound_qty
        inbound_price = prod.inbound_price or prod.tax_price
        if inbound_qty is not None and inbound_price is not None:
            prod.inbound_amount = Decimal(str(inbound_qty)) * Decimal(str(inbound_price))

        if item.outbound_qty is not None:
            prod.outbound_qty = Decimal(str(item.outbound_qty))
        elif actual is not None and plan.plan_type == "outbound":
            prod.outbound_qty = actual
        if item.outbound_price is not None:
            prod.outbound_price = Decimal(str(item.outbound_price))
        outbound_qty = prod.outbound_qty
        outbound_price = prod.outbound_price or prod.tax_price
        if outbound_qty is not None and outbound_price is not None:
            prod.outbound_amount = Decimal(str(outbound_qty)) * Decimal(str(outbound_price))

        prod.update_by = user_id

        brand_rows = None if item.brand_qtys is None else []
        if item.brand_qtys:
            brand_rows = [
                {
                    "brandId": bq.brand_id or "",
                    "brandName": bq.brand_name,
                    "qty": bq.qty,
                    **({"unitPrice": bq.unit_price} if bq.unit_price is not None else {}),
                }
                for bq in item.brand_qtys
            ]
        if actual is not None and brand_rows is not None:
            try:
                validate_brand_qty_rows(
                    brand_mode=prod.brand_mode,
                    required_brand_id=prod.brand_id,
                    actual_qty=actual,
                    rows=brand_rows,
                )
            except ValueError as exc:
                raise HTTPException(status_code=400, detail=str(exc)) from exc
        persist_plan_product_brand_qtys(
            db,
            tenant_no=tenant_no,
            plan_id=plan.id,
            plan_product_id=prod.id,
            brand_mode=prod.brand_mode,
            rows=brand_rows,
            user_id=user_id,
            dept_id=getattr(prod, "create_dept_id", "") or "",
        )


def apply_archive_brand_qtys_only(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
    items: list[PlanArchiveProductItem],
    user_id: str,
) -> None:
    """一车多卸：段级数量已写入，此处仅处理计划产品品牌分量。"""
    product_map = {
        row.id: row
        for row in db.query(PlPlanProduct).filter(
            PlPlanProduct.plan_id == plan.id,
            PlPlanProduct.tenant_no == tenant_no,
            PlPlanProduct.del_flag == 0,
        ).all()
    }
    for item in items:
        if item.brand_qtys is None:
            continue
        prod = product_map.get(item.id)
        if not prod:
            raise HTTPException(status_code=400, detail=f"产品明细不存在: {item.id}")
        actual = Decimal(str(prod.actual_qty or 0))
        brand_rows = [
            {
                "brandId": bq.brand_id or "",
                "brandName": bq.brand_name,
                "qty": bq.qty,
                **({"unitPrice": bq.unit_price} if bq.unit_price is not None else {}),
            }
            for bq in item.brand_qtys
        ]
        try:
            validate_brand_qty_rows(
                brand_mode=prod.brand_mode,
                required_brand_id=prod.brand_id,
                actual_qty=actual,
                rows=brand_rows,
            )
        except ValueError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        persist_plan_product_brand_qtys(
            db,
            tenant_no=tenant_no,
            plan_id=plan.id,
            plan_product_id=prod.id,
            brand_mode=prod.brand_mode,
            rows=brand_rows,
            user_id=user_id,
            dept_id=getattr(prod, "create_dept_id", "") or "",
        )


def apply_archive_freight(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
    freight: PlanArchiveFreightBlock,
    user_id: str,
    strict: bool = True,
) -> None:
    has_freight = int(freight.has_freight if freight.has_freight is not None else plan.has_freight or 0)
    plan.has_freight = has_freight
    if freight.freight_remark is not None:
        plan.freight_remark = freight.freight_remark
    plan.update_by = user_id

    if has_freight != 1:
        plan.freight_amount = Decimal("0")
        if not plan.transport_id:
            return
        if transport_has_locked_freight_settlement(db, tenant_no, plan.transport_id):
            raise HTTPException(status_code=400, detail="运输单已纳入运费结算，无法修改运费")
        transport = (
            db.query(PlTransport)
            .filter(
                PlTransport.id == plan.transport_id,
                PlTransport.tenant_no == tenant_no,
                PlTransport.del_flag == 0,
            )
            .first()
        )
        if not transport:
            raise HTTPException(status_code=400, detail="关联运输单不存在")
        transport.freight_amount = Decimal("0")
        transport.freight_price = None
        transport.transport_qty = None
        transport.update_by = user_id
        sync_draft_settlement_freight_for_transport(
            db,
            tenant_no=tenant_no,
            transport_id=plan.transport_id,
            freight_amount=Decimal("0"),
            user_id=user_id,
        )
        return

    if not plan.transport_id:
        if strict:
            raise HTTPException(status_code=400, detail="计划有运输费用但未安排运输，请先安排运输")
        return
    if transport_has_locked_freight_settlement(db, tenant_no, plan.transport_id):
        raise HTTPException(status_code=400, detail="运输单已纳入运费结算，无法修改运费")

    transport = (
        db.query(PlTransport)
        .filter(
            PlTransport.id == plan.transport_id,
            PlTransport.tenant_no == tenant_no,
            PlTransport.del_flag == 0,
        )
        .first()
    )
    if not transport:
        raise HTTPException(status_code=400, detail="关联运输单不存在")

    transport.distance_km = freight.distance_km
    transport.freight_type = freight.freight_type or "by_weight"
    transport.freight_price = freight.freight_price
    amount = freight.freight_amount or Decimal("0")
    transport.freight_amount = amount
    freight_type = freight.freight_type or transport.freight_type or "by_weight"
    plan_products = (
        db.query(PlPlanProduct)
        .filter(
            PlPlanProduct.plan_id == plan.id,
            PlPlanProduct.tenant_no == tenant_no,
            PlPlanProduct.del_flag == 0,
        )
        .all()
    )
    if freight.transport_qty is not None:
        transport.transport_qty = freight.transport_qty
    elif freight_type == "by_weight":
        from app.services.pl_plan_archive_stop_service import sum_delivery_outbound_qty
        from app.models.pl_models import PlPlanLoad

        has_load_stop = (
            db.query(PlPlanLoad.id)
            .filter(
                PlPlanLoad.plan_id == plan.id,
                PlPlanLoad.tenant_no == tenant_no,
                PlPlanLoad.del_flag == 0,
            )
            .first()
        )
        if has_load_stop:
            transport.transport_qty = sum_delivery_outbound_qty(
                db, tenant_no=tenant_no, plan_id=plan.id
            )
        else:
            transport.transport_qty = sum_plan_product_actual_quantities(plan_products)
    if freight_type == "by_weight" and (transport.transport_qty or 0) <= 0:
        if strict:
            raise HTTPException(status_code=400, detail="运输量必须大于0，请先填写产品实际数量")
        # 草稿：允许运输量暂缺
    if freight.has_invoice is not None:
        transport.has_invoice = freight.has_invoice
    transport.update_by = user_id

    plan.freight_amount = amount
    sync_draft_settlement_freight_for_transport(
        db,
        tenant_no=tenant_no,
        transport_id=plan.transport_id,
        freight_amount=amount,
        user_id=user_id,
    )


def record_plan_change(
    db: Session,
    *,
    tenant_no: str,
    plan_id: str,
    change_type: str,
    change_content: str,
    user_id: str,
    user_name: Optional[str],
    dept_id: str,
) -> None:
    db.add(
        PlPlanChange(
            id=generate_uuid(),
            tenant_no=tenant_no,
            plan_id=plan_id,
            change_type=change_type,
            change_content=change_content,
            change_user_id=user_id,
            change_user_name=user_name,
            create_by=user_id,
            create_dept_id=dept_id,
        )
    )


def apply_plan_driver_feedback(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
    payload: PlanDriverFeedbackUpdateBody,
    user_id: str,
) -> PlTransport:
    if plan.follow_status == "cancelled":
        raise HTTPException(status_code=400, detail="已作废的计划不能编辑司机回传")

    if not plan.transport_id:
        raise HTTPException(status_code=400, detail="计划未安排运输，无法编辑司机回传")

    transport = (
        db.query(PlTransport)
        .filter(
            PlTransport.id == plan.transport_id,
            PlTransport.tenant_no == tenant_no,
            PlTransport.del_flag == 0,
        )
        .first()
    )
    if not transport:
        raise HTTPException(status_code=404, detail="运输单不存在")

    if payload.products is not None:
        products_payload = [
            {
                "planProductId": item.plan_product_id,
                "actualQty": item.actual_qty,
                "brandQtys": ([b.model_dump(by_alias=True) for b in item.brand_qtys] if item.brand_qtys is not None else None),
            }
            for item in payload.products
        ]
        try:
            persist_transport_product_feedback(
                db,
                tenant_no=tenant_no,
                transport=transport,
                plan_id=plan.id,
                products=products_payload,
                user_id=user_id,
                enforce_brand=True,
            )
        except ValueError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc
    elif payload.feedback_qty is not None:
        transport.feedback_qty = payload.feedback_qty
    if payload.feedback_remark is not None:
        transport.feedback_remark = payload.feedback_remark.strip() if payload.feedback_remark else None
    if payload.feedback_images is not None:
        transport.feedback_images = json.dumps([item for item in payload.feedback_images if item])
    transport.feedback_time = payload.feedback_time or datetime.now()
    transport.update_by = user_id
    return transport


def apply_plan_site_feedback(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
    link_id: str,
    payload: PlanSiteFeedbackUpdateBody,
    user_id: str,
) -> PlTransportSiteLink:
    if plan.follow_status == "cancelled":
        raise HTTPException(status_code=400, detail="已作废的计划不能编辑现场回传")

    link = (
        db.query(PlTransportSiteLink)
        .filter(
            PlTransportSiteLink.id == link_id,
            PlTransportSiteLink.plan_id == plan.id,
            PlTransportSiteLink.tenant_no == tenant_no,
            PlTransportSiteLink.del_flag == 0,
        )
        .first()
    )
    if not link:
        raise HTTPException(status_code=404, detail="现场链接不存在")

    if payload.site_feedback_remark is not None:
        link.site_feedback_remark = (
            payload.site_feedback_remark.strip() if payload.site_feedback_remark else None
        )
    if payload.site_feedback_images is not None:
        link.site_feedback_images = json.dumps([item for item in payload.site_feedback_images if item])
    link.site_feedback_time = payload.site_feedback_time or datetime.now()
    link.update_by = user_id
    return link


def _resolve_stop_business_type_for_archive(
    db: Session,
    *,
    tenant_no: str,
    plan_id: str,
    stop_type: str,
    stop_id: str,
) -> str:
    from app.models.pl_models import PlPlanDelivery, PlPlanLoad

    direction: Optional[str] = None
    if stop_type == "load":
        row = (
            db.query(PlPlanLoad)
            .filter(
                PlPlanLoad.id == stop_id,
                PlPlanLoad.plan_id == plan_id,
                PlPlanLoad.tenant_no == tenant_no,
                PlPlanLoad.del_flag == 0,
            )
            .first()
        )
        direction = (row.dibang_direction or "").strip() if row else None
    else:
        row = (
            db.query(PlPlanDelivery)
            .filter(
                PlPlanDelivery.id == stop_id,
                PlPlanDelivery.plan_id == plan_id,
                PlPlanDelivery.tenant_no == tenant_no,
                PlPlanDelivery.del_flag == 0,
            )
            .first()
        )
        direction = (row.dibang_direction or "").strip() if row else None

    if direction == "outbound":
        return "出库"
    if direction == "inbound":
        return "入库"
    return "出库"


def apply_archive_stop_feedback_attachments(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
    feedbacks: PlanArchiveFeedbacksBlock,
    user_id: str,
    dept_id: str = "",
) -> None:
    if not plan.transport_id:
        return

    from app.models.pl_models import PlTransportDriverStopFeedback, PlTransportSiteStopFeedback
    from app.services.pl_transport_stop_feedback_service import (
        save_driver_stop_feedback,
        save_site_stop_feedback,
    )

    transport_id = plan.transport_id
    plan_id = plan.id

    for item in feedbacks.driver_stops or []:
        urls = [url for url in (item.attachment_urls or []) if url]
        existing = (
            db.query(PlTransportDriverStopFeedback)
            .filter(
                PlTransportDriverStopFeedback.transport_id == transport_id,
                PlTransportDriverStopFeedback.tenant_no == tenant_no,
                PlTransportDriverStopFeedback.stop_type == item.stop_type,
                PlTransportDriverStopFeedback.stop_id == item.stop_id,
                PlTransportDriverStopFeedback.del_flag == 0,
            )
            .first()
        )
        if not urls and not existing:
            continue
        save_driver_stop_feedback(
            db,
            tenant_no=tenant_no,
            transport_id=transport_id,
            plan_id=plan_id,
            stop_type=item.stop_type,
            stop_id=item.stop_id,
            attachment_urls=urls,
            user_id=user_id,
            dept_id=dept_id,
        )

    for item in feedbacks.site_stops or []:
        urls = [url for url in (item.attachment_urls or []) if url]
        existing = (
            db.query(PlTransportSiteStopFeedback)
            .filter(
                PlTransportSiteStopFeedback.transport_id == transport_id,
                PlTransportSiteStopFeedback.tenant_no == tenant_no,
                PlTransportSiteStopFeedback.stop_type == item.stop_type,
                PlTransportSiteStopFeedback.stop_id == item.stop_id,
                PlTransportSiteStopFeedback.del_flag == 0,
            )
            .first()
        )
        if not urls and not existing:
            continue
        business_type = _resolve_stop_business_type_for_archive(
            db,
            tenant_no=tenant_no,
            plan_id=plan_id,
            stop_type=item.stop_type,
            stop_id=item.stop_id,
        )
        save_site_stop_feedback(
            db,
            tenant_no=tenant_no,
            transport_id=transport_id,
            plan_id=plan_id,
            stop_type=item.stop_type,
            stop_id=item.stop_id,
            business_type=business_type,
            attachment_urls=urls,
            user_id=user_id,
            dept_id=dept_id,
            require_qty_before_attachments=False,
        )