shanxia2.0/backend/app/api/pl/transport.py

"""运输计划管理 - 运输 API"""
from datetime import date, datetime
from typing import List, Optional
import json
import secrets

from fastapi import APIRouter, Depends, HTTPException, Query, status
from pydantic import BaseModel, Field
from sqlalchemy import func, or_
from sqlalchemy.orm import Session

from app.core.database import get_db
from app.core.db_auth import DbUserContext, get_db_current_user
from app.services.pl_data_scope import apply_pl_scope, assert_pl_row_in_scope
from app.core.tenant_context import get_tenant_no
from app.utils.excel_export import (
    EXPORT_MAX_ROWS,
    ExportColumn,
    ExportValidationError,
    export_rows_to_xlsx_response,
    filter_export_columns,
)
from app.core.utils import generate_uuid
from app.core.config import settings
from app.models.pl_models import (
    PlCarrier,
    PlDriver,
    PlPersonnel,
    PlPlan,
    PlPlanProduct,
    PlPlanTransport,
    PlTransport,
    PlTransportApproval,
    PlTransportDbLink,
    PlVehicle,
)
from app.schemas.pl.io_order import BrandQtyItem
from app.schemas.pl.transport import (
    ApprovalRecordRead,
    ApprovalRequest,
    ApprovalSubmit,
    TransportArrange,
    TransportArrangeResult,
    TransportCreate,
    TransportFeedback,
    TransportInventoryLinkRead,
    TransportIoOrderBrief,
    TransportSiteLinkRead,
    TransportListQuery,
    TransportListResponse,
    TransportDriverProductFeedbackRead,
    TransportDriverStopProductFeedbackRead,
    TransportRead,
    TransportStopFeedbackRead,
)
from app.services.pl_dibang_bridge_service import (
    StopDibangOption,
    plan_has_load_stop,
    sync_dibang_for_plan_stops,
    sync_dibang_for_transport,
)
from app.services.pl_transport_stop_feedback_service import (
    build_transport_site_url,
    ensure_transport_site_token,
    load_driver_stop_segments,
    load_site_stop_segments,
)
from app.services.pl_inventory_link_read import build_transport_inventory_link_read
from app.services.pl_io_order_service import (
    apply_dibang_prefill_to_order,
    create_io_order_for_transport,
    void_io_orders_for_plan,
)
from app.services.pl_internal_customer import plan_requires_site_link
from app.services.pl_plan_service import (
    validate_hazard_cert_ready_for_transport,
)
from app.services.pl_transport_resource_snapshot_service import (
    apply_resource_snapshot_json,
    build_transport_resource_snapshot,
)
from app.services.pl_carrier_service import assert_carrier_eligible_for_transport
from app.services.pl_feedback_brand_qty import load_transport_driver_product_feedbacks

router = APIRouter(prefix="/order", tags=["运输计划-运输管理"])


def _transport_to_read(db: Session, transport: PlTransport) -> TransportRead:
    base = TransportRead.model_validate(transport)
    feedbacks = load_transport_driver_product_feedbacks(
        db,
        tenant_no=transport.tenant_no,
        transport_id=transport.id,
    )
    updates: dict = {}
    if feedbacks:
        updates["driver_product_feedbacks"] = [
            TransportDriverProductFeedbackRead.model_validate(item)
            for item in feedbacks
        ]

    link = (
        db.query(PlPlanTransport)
        .filter(
            PlPlanTransport.transport_id == transport.id,
            PlPlanTransport.del_flag == 0,
        )
        .order_by(PlPlanTransport.create_time.desc())
        .first()
    )
    if link and link.plan_id and plan_has_load_stop(db, transport.tenant_no, link.plan_id):
        updates["driver_stop_feedbacks"] = [
            TransportStopFeedbackRead(
                stop_type=segment.stop_type,
                stop_id=segment.stop_id,
                label=segment.label,
                customer_name=segment.customer_name,
                attachment_urls=segment.attachment_urls,
                products=[
                    TransportDriverStopProductFeedbackRead(
                        stop_product_id=product.stop_product_id,
                        plan_product_id=product.plan_product_id,
                        product_name=product.product_name,
                        model=product.model,
                        unit=product.unit,
                        package_spec=product.package_spec,
                        quantity=product.plan_qty,
                        actual_qty=product.actual_qty,
                        brand_qtys=[
                            BrandQtyItem(
                                brand_id=str(item.get("brandId") or item.get("brand_id") or ""),
                                brand_name=item.get("brandName") or item.get("brand_name"),
                                qty=item.get("qty") or 0,
                            )
                            for item in (product.brand_qtys or [])
                            if isinstance(item, dict)
                        ]
                        or None,
                    )
                    for product in segment.products
                    if product.actual_qty is not None
                    or (product.brand_qtys and len(product.brand_qtys) > 0)
                ],
                remark=segment.remark,
                feedback_time=segment.feedback_time,
            )
            for segment in load_driver_stop_segments(
                db,
                tenant_no=transport.tenant_no,
                plan_id=link.plan_id,
                transport_id=transport.id,
            )
            if segment.attachment_urls
            or segment.remark
            or segment.feedback_time
            or any(
                product.actual_qty is not None
                or (product.brand_qtys and len(product.brand_qtys) > 0)
                for product in segment.products
            )
        ]
        updates["site_stop_feedbacks"] = [
            TransportStopFeedbackRead(
                stop_type=segment.stop_type,
                stop_id=segment.stop_id,
                label=segment.label,
                customer_name=segment.customer_name,
                business_type=segment.business_type,
                quantity=segment.quantity,
                attachment_urls=segment.attachment_urls,
                products=[
                    TransportDriverStopProductFeedbackRead(
                        stop_product_id=product.stop_product_id,
                        plan_product_id=product.plan_product_id,
                        product_name=product.product_name,
                        model=product.model,
                        unit=product.unit,
                        package_spec=product.package_spec,
                        quantity=product.plan_qty,
                        actual_qty=product.actual_qty,
                        brand_qtys=[
                            BrandQtyItem(
                                brand_id=str(item.get("brandId") or item.get("brand_id") or ""),
                                brand_name=item.get("brandName") or item.get("brand_name"),
                                qty=item.get("qty") or 0,
                            )
                            for item in (product.brand_qtys or [])
                            if isinstance(item, dict)
                        ]
                        or None,
                    )
                    for product in segment.products
                    if product.actual_qty is not None
                    or (product.brand_qtys and len(product.brand_qtys) > 0)
                ],
                remark=segment.remark,
                feedback_time=segment.feedback_time,
            )
            for segment in load_site_stop_segments(
                db,
                tenant_no=transport.tenant_no,
                plan_id=link.plan_id,
                transport_id=transport.id,
            )
            if segment.quantity is not None
            or segment.attachment_urls
            or segment.remark
            or any(
                product.actual_qty is not None
                or (product.brand_qtys and len(product.brand_qtys) > 0)
                for product in segment.products
            )
        ]

    if not updates:
        return base
    return base.model_copy(update=updates)


def _generate_transport_no(db: Session) -> str:
    """生成运输编号 SXYS######"""
    now = date.today()
    prefix = f"SXYS{now.strftime('%Y%m')}"

    last = (
        db.query(PlTransport)
        .filter(
            PlTransport.transport_no.like(f"{prefix}%"),
            PlTransport.del_flag == 0,
        )
        .order_by(PlTransport.transport_no.desc())
        .first()
    )

    if last:
        last_seq = int(last.transport_no[-6:])
        seq = last_seq + 1
    else:
        seq = 1

    return f"{prefix}{seq:06d}"


def _resolve_driver_info(db: Session, tenant_no: str, driver_id: Optional[str]) -> Optional[dict]:
    """解析司机信息：优先 pl_personnel，兼容旧 pl_driver。"""
    if not driver_id:
        return None
    personnel = (
        db.query(PlPersonnel)
        .filter(
            PlPersonnel.id == driver_id,
            PlPersonnel.tenant_no == tenant_no,
            PlPersonnel.del_flag == 0,
        )
        .first()
    )
    if personnel:
        return {
            "id": personnel.id,
            "name": personnel.personnel_name,
            "phone": personnel.phone_no,
        }

    legacy = (
        db.query(PlDriver)
        .filter(
            PlDriver.id == driver_id,
            PlDriver.tenant_no == tenant_no,
            PlDriver.del_flag == 0,
        )
        .first()
    )
    if legacy:
        return {
            "id": legacy.id,
            "name": legacy.driver_name,
            "phone": legacy.phone_no,
        }
    return None


def _get_vehicle_info(db: Session, vehicle_id: str, tenant_no: Optional[str] = None) -> dict:
    vehicle = db.query(PlVehicle).filter(PlVehicle.id == vehicle_id).first()
    if not vehicle:
        return {}

    info = {
        "license_plate": vehicle.license_plate,
        "carrier_id": vehicle.carrier_id,
        "carrier_name": vehicle.carrier_name,
    }

    if vehicle.default_driver_id:
        driver_info = (
            _resolve_driver_info(db, tenant_no, vehicle.default_driver_id)
            if tenant_no
            else None
        )
        if not driver_info:
            personnel = db.query(PlPersonnel).filter(
                PlPersonnel.id == vehicle.default_driver_id,
                PlPersonnel.del_flag == 0,
            ).first()
            if personnel:
                driver_info = {
                    "id": personnel.id,
                    "name": personnel.personnel_name,
                    "phone": personnel.phone_no,
                }
            else:
                legacy = db.query(PlDriver).filter(
                    PlDriver.id == vehicle.default_driver_id,
                    PlDriver.del_flag == 0,
                ).first()
                if legacy:
                    driver_info = {
                        "id": legacy.id,
                        "name": legacy.driver_name,
                        "phone": legacy.phone_no,
                    }
        if driver_info:
            info["driver_id"] = driver_info["id"]
            info["driver_name"] = driver_info["name"]
            info["driver_phone"] = driver_info["phone"]
        elif vehicle.default_driver_name:
            info["driver_id"] = vehicle.default_driver_id
            info["driver_name"] = vehicle.default_driver_name

    return info


def _assert_vehicle_carrier_eligible(
    db: Session,
    tenant_no: str,
    vehicle_id: str,
    *,
    rearrange: bool = False,
) -> None:
    vehicle = (
        db.query(PlVehicle)
        .filter(
            PlVehicle.id == vehicle_id,
            PlVehicle.tenant_no == tenant_no,
            PlVehicle.del_flag == 0,
        )
        .first()
    )
    if not vehicle:
        raise HTTPException(status_code=400, detail="车辆不存在")
    if not vehicle.carrier_id:
        return

    carrier = (
        db.query(PlCarrier)
        .filter(
            PlCarrier.id == vehicle.carrier_id,
            PlCarrier.tenant_no == tenant_no,
            PlCarrier.del_flag == 0,
        )
        .first()
    )
    if not carrier:
        raise HTTPException(status_code=400, detail="承运商不存在")
    assert_carrier_eligible_for_transport(carrier, rearrange=rearrange)


def _ensure_driver_access_token(transport: PlTransport) -> str:
    token = (transport.driver_access_token or "").strip()
    if not token:
        token = secrets.token_urlsafe(32)[:64]
        transport.driver_access_token = token
    return token


def _rotate_driver_access_token(transport: PlTransport) -> str:
    token = secrets.token_urlsafe(32)[:64]
    transport.driver_access_token = token
    return token


def _is_rearrange(plan: PlPlan, existing_transport: Optional[PlTransport]) -> bool:
    """仅计划已正式安排才算「调整运输」。

    创建/编辑表单预选车辆会留下 draft 运输单，计划仍为 pending；
    若把非 cancelled 运输单都当成 rearrange，会把首次正式安排误判为调整。
    existing_transport 保留入参以兼容调用方；草稿复用由后续分支处理。
    """
    _ = existing_transport
    return plan.transport_status == "arranged"


def _apply_plan_logistics_from_payload(plan: PlPlan, payload: TransportArrange) -> None:
    if payload.pickup_address is not None:
        plan.pickup_address = payload.pickup_address
    elif payload.start_point is not None:
        plan.pickup_address = payload.start_point
    if "pickup_lng" in payload.model_fields_set or "pickup_lat" in payload.model_fields_set:
        plan.pickup_lng = payload.pickup_lng
        plan.pickup_lat = payload.pickup_lat
    if payload.pickup_time is not None:
        plan.pickup_time = payload.pickup_time
    if payload.pickup_contact is not None:
        plan.pickup_contact = payload.pickup_contact
    if payload.pickup_notes is not None:
        plan.pickup_notes = payload.pickup_notes
    if payload.destination_address is not None:
        plan.destination_address = payload.destination_address
    elif payload.end_point is not None:
        plan.destination_address = payload.end_point
    if "destination_lng" in payload.model_fields_set or "destination_lat" in payload.model_fields_set:
        plan.destination_lng = payload.destination_lng
        plan.destination_lat = payload.destination_lat
    if payload.delivery_time is not None:
        plan.delivery_time = payload.delivery_time
    if payload.delivery_contact is not None:
        plan.delivery_contact = payload.delivery_contact
    if payload.destination_notes is not None:
        plan.destination_notes = payload.destination_notes


def _resolve_transport_endpoints(plan: PlPlan, payload: TransportArrange) -> tuple[Optional[str], Optional[str]]:
    start_point = (
        payload.pickup_address
        or payload.start_point
        or plan.pickup_address
    )
    end_point = (
        payload.destination_address
        or payload.end_point
        or plan.destination_address
        or plan.destination_name
    )
    return start_point, end_point


def _apply_transport_arrange_fields(
    transport: PlTransport,
    *,
    payload: TransportArrange,
    vehicle_info: dict,
    driver_info: Optional[dict],
    carrier_name: Optional[str],
    start_point: Optional[str],
    end_point: Optional[str],
    freight_amount: Optional[float],
    user_id: str,
) -> None:
    transport.vehicle_id = payload.vehicle_id
    transport.license_plate = vehicle_info.get("license_plate")
    transport.driver_id = driver_info["id"] if driver_info else None
    transport.driver_name = driver_info["name"] if driver_info else None
    transport.driver_phone = driver_info["phone"] if driver_info else None
    transport.carrier_id = payload.carrier_id or vehicle_info.get("carrier_id")
    transport.carrier_name = carrier_name
    transport.transport_medium = payload.transport_medium
    transport.start_point = start_point
    transport.end_point = end_point
    transport.distance_km = payload.distance_km
    transport.unit = payload.unit or ("吨" if payload.freight_type == "by_weight" else "趟")
    transport.freight_price = payload.freight_price
    transport.freight_amount = freight_amount or 0
    transport.freight_type = payload.freight_type
    if payload.has_invoice is not None:
        transport.has_invoice = payload.has_invoice
    if payload.remark is not None:
        transport.remark = payload.remark
    transport.update_by = user_id


def _build_driver_url(token: str) -> str:
    origin = (settings.PUBLIC_WEB_ORIGIN or "").rstrip("/")
    return f"{origin}/#/public/pl/transport/{token}"


def _build_arrange_result(
    transport: PlTransport,
    site_records: list[TransportSiteLinkRead],
    inventory_records: list[TransportInventoryLinkRead],
    io_order: Optional[TransportIoOrderBrief] = None,
    contract_qty_reminder: Optional[dict] = None,
) -> TransportArrangeResult:
    token = transport.driver_access_token or ""
    result = TransportArrangeResult.model_validate(transport)
    result.driver_access_token = token
    result.driver_url = _build_driver_url(token) if token else None
    result.site_records = site_records
    result.inventory_records = inventory_records
    result.io_order = io_order
    if contract_qty_reminder:
        from app.schemas.pl.transport import ContractQtyReminder

        result.contract_qty_reminder = ContractQtyReminder.model_validate(contract_qty_reminder)
    return result


# =============================================================================
# 运输列表
# =============================================================================


@router.get("/", response_model=TransportListResponse)
def list_transports(
    q: Optional[str] = Query(default=None, description="模糊搜索"),
    transport_no: Optional[str] = Query(default=None, alias="transportNo", description="运输编号"),
    transport_status: Optional[str] = Query(default=None, alias="transportStatus", description="运输状态"),
    approval_status: Optional[str] = Query(default=None, alias="approvalStatus", description="审批状态"),
    vehicle_id: Optional[str] = Query(default=None, alias="vehicleId", description="车辆ID"),
    driver_id: Optional[str] = Query(default=None, alias="driverId", description="司机ID"),
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=20, ge=1, le=500, alias="pageSize"),
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> TransportListResponse:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    query = db.query(PlTransport).filter(
        PlTransport.tenant_no == tenant_no,
        PlTransport.del_flag == 0,
    )

    if q:
        query = query.filter(
            or_(
                PlTransport.transport_no.contains(q),
                PlTransport.license_plate.contains(q),
                PlTransport.driver_name.contains(q),
            )
        )

    if transport_no:
        query = query.filter(PlTransport.transport_no.contains(transport_no))

    if transport_status:
        query = query.filter(PlTransport.transport_status == transport_status)

    if approval_status:
        query = query.filter(PlTransport.approval_status == approval_status)

    if vehicle_id:
        query = query.filter(PlTransport.vehicle_id == vehicle_id)

    if driver_id:
        query = query.filter(PlTransport.driver_id == driver_id)

    query = apply_pl_scope(query, PlTransport, user, db, tenant_no)

    total = query.count()
    offset = (page - 1) * page_size
    transports = query.order_by(PlTransport.create_time.desc()).offset(offset).limit(page_size).all()

    return TransportListResponse(total=total, list=transports)


# =============================================================================
# 创建运输
# =============================================================================



TRANSPORT_EXPORT_COLUMNS = [
    ExportColumn(title="运输编号", field="transportNo"),
    ExportColumn(title="车牌号", field="licensePlate"),
    ExportColumn(title="司机", field="driverName"),
    ExportColumn(title="承运商", field="carrierName"),
    ExportColumn(title="运输介质", field="transportMedium"),
    ExportColumn(title="起点", field="startPoint"),
    ExportColumn(title="终点", field="endPoint"),
    ExportColumn(title="运输状态", field="transportStatus"),
    ExportColumn(title="审批状态", field="approvalStatus"),
    ExportColumn(title="备注", field="remark"),
]


class TransportExportBody(BaseModel):
    q: Optional[str] = None
    transport_no: Optional[str] = Field(default=None, alias="transportNo")
    transport_status: Optional[str] = Field(default=None, alias="transportStatus")
    approval_status: Optional[str] = Field(default=None, alias="approvalStatus")
    vehicle_id: Optional[str] = Field(default=None, alias="vehicleId")
    driver_id: Optional[str] = Field(default=None, alias="driverId")
    ids: Optional[List[str]] = None
    columns: Optional[List[str]] = None

    model_config = {"populate_by_name": True}


@router.post("/export")
def export_transports(
    body: TransportExportBody,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
):
    """导出运输列表（当前筛选全量，或 ids 已选）。"""
    try:
        tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
        query = db.query(PlTransport).filter(
            PlTransport.tenant_no == tenant_no,
            PlTransport.del_flag == 0,
        )
        if body.q:
            query = query.filter(
                or_(
                    PlTransport.transport_no.contains(body.q),
                    PlTransport.license_plate.contains(body.q),
                    PlTransport.driver_name.contains(body.q),
                )
            )
        if body.transport_no:
            query = query.filter(PlTransport.transport_no.contains(body.transport_no))
        if body.transport_status:
            query = query.filter(PlTransport.transport_status == body.transport_status)
        if body.approval_status:
            query = query.filter(PlTransport.approval_status == body.approval_status)
        if body.vehicle_id:
            query = query.filter(PlTransport.vehicle_id == body.vehicle_id)
        if body.driver_id:
            query = query.filter(PlTransport.driver_id == body.driver_id)
        if body.ids:
            query = query.filter(PlTransport.id.in_(body.ids))
        query = apply_pl_scope(query, PlTransport, user, db, tenant_no)
        rows = query.order_by(PlTransport.create_time.desc()).limit(EXPORT_MAX_ROWS + 1).all()
        items = [TransportRead.model_validate(r).model_dump(by_alias=True) for r in rows]
        cols = filter_export_columns(TRANSPORT_EXPORT_COLUMNS, body.columns)
        date_str = datetime.now().strftime("%Y%m%d")
        return export_rows_to_xlsx_response(
            db,
            tenant_no=tenant_no,
            sheet_title="运输列表",
            columns=cols,
            rows=items,
            filename_stem=f"运输列表_{date_str}",
        )
    except ExportValidationError as e:
        raise HTTPException(status_code=400, detail=str(e)) from e



@router.post("/", response_model=TransportRead, status_code=status.HTTP_201_CREATED)
def create_transport(
    payload: TransportCreate,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> TransportRead:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())

    transport_no = _generate_transport_no(db)

    transport = PlTransport(
        id=generate_uuid(),
        tenant_no=tenant_no,
        transport_no=transport_no,
        vehicle_id=payload.vehicle_id,
        license_plate=payload.license_plate,
        driver_id=payload.driver_id,
        driver_name=payload.driver_name,
        driver_phone=payload.driver_phone,
        carrier_id=payload.carrier_id,
        carrier_name=payload.carrier_name,
        transport_medium=payload.transport_medium,
        start_point=payload.start_point,
        end_point=payload.end_point,
        distance_km=payload.distance_km,
        unit=payload.unit,
        freight_price=payload.freight_price,
        freight_amount=payload.freight_amount,
        freight_type=payload.freight_type,
        has_invoice=payload.has_invoice,
        qr_expire_date=payload.qr_expire_date,
        transport_status=payload.transport_status or "draft",
        approval_status=payload.approval_status or "none",
        remark=payload.remark,
        create_by=user.id,
        create_dept_id=user.dept_id,
    )
    db.add(transport)

    # 关联计划
    if payload.plan_ids:
        for pid in payload.plan_ids:
            pt = PlPlanTransport(
                id=generate_uuid(),
                tenant_no=tenant_no,
                plan_id=pid,
                transport_id=transport.id,
                create_by=user.id,
                create_dept_id=user.dept_id,
            )
            db.add(pt)

            # 更新计划关联的运输信息
            plan = db.query(PlPlan).filter(PlPlan.id == pid, PlPlan.del_flag == 0).first()
            if plan:
                assert_pl_row_in_scope(db, plan, PlPlan, user, tenant_no)
                plan.transport_id = transport.id
                plan.transport_no = transport_no
                plan.vehicle_id = transport.vehicle_id
                plan.license_plate = transport.license_plate
                plan.driver_id = transport.driver_id
                plan.driver_name = transport.driver_name
                if transport.vehicle_id:
                    snapshot = build_transport_resource_snapshot(
                        db,
                        tenant_no,
                        transport.vehicle_id,
                        selected_driver_id=transport.driver_id,
                    )
                    apply_resource_snapshot_json(
                        plan=plan, transport=transport, snapshot=snapshot
                    )

    db.commit()
    db.refresh(transport)
    return transport


# =============================================================================
# 运输安排
# =============================================================================


@router.post("/arrange/{plan_id}", response_model=TransportArrangeResult, status_code=status.HTTP_201_CREATED)
def arrange_transport(
    plan_id: str,
    payload: TransportArrange,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> TransportArrangeResult:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())

    # 获取计划
    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="计划不存在")
    assert_pl_row_in_scope(db, plan, PlPlan, user, tenant_no)

    existing_transport = None
    if plan.transport_id:
        existing_transport = db.query(PlTransport).filter(
            PlTransport.id == plan.transport_id,
            PlTransport.tenant_no == tenant_no,
            PlTransport.del_flag == 0,
        ).first()

    is_rearrange = _is_rearrange(plan, existing_transport)

    if is_rearrange and existing_transport and existing_transport.vehicle_id:
        _assert_vehicle_carrier_eligible(
            db,
            tenant_no,
            existing_transport.vehicle_id,
            rearrange=True,
        )

    _assert_vehicle_carrier_eligible(db, tenant_no, payload.vehicle_id, rearrange=False)

    # 获取车辆信息
    vehicle_info = _get_vehicle_info(db, payload.vehicle_id, tenant_no)

    # 获取司机信息（可选；driverId 为人员 pl_personnel.id）
    driver_info = None
    if payload.driver_id:
        driver_info = _resolve_driver_info(db, tenant_no, payload.driver_id)
        if not driver_info:
            raise HTTPException(status_code=400, detail="司机不存在")

    # 检查产品属性（易制毒/易制爆需要证件，以 CRM 产品主数据为准）
    products = db.query(PlPlanProduct).filter(
        PlPlanProduct.plan_id == plan_id,
        PlPlanProduct.del_flag == 0,
    ).all()

    validate_hazard_cert_ready_for_transport(db, tenant_no, plan, products)

    from app.services.pl_contract_plan_qty import (
        assert_contract_qty_for_arrange,
        build_contract_qty_reminder_payload,
    )

    contract_qty_reminders = assert_contract_qty_for_arrange(
        db, tenant_no=tenant_no, plan_id=plan_id
    )
    contract_qty_reminder_payload = build_contract_qty_reminder_payload(contract_qty_reminders)

    # 计算运费
    freight_amount = payload.freight_amount
    if payload.freight_type == "by_weight" and payload.freight_price:
        total_qty = sum(
            float(p.quantity or p.inbound_estimate_qty or p.outbound_estimate_qty or 0)
            for p in products
        )
        freight_amount = total_qty * float(payload.freight_price)

    carrier_name = vehicle_info.get("carrier_name")
    if payload.carrier_id:
        carrier = db.query(PlCarrier).filter(PlCarrier.id == payload.carrier_id).first()
        if carrier:
            carrier_name = carrier.carrier_name

    _apply_plan_logistics_from_payload(plan, payload)
    start_point, end_point = _resolve_transport_endpoints(plan, payload)

    # 地磅关联改为按段 upsert，不再在调整前整批 revoke 再建

    if existing_transport and existing_transport.transport_status != "cancelled":
        transport = existing_transport
        _apply_transport_arrange_fields(
            transport,
            payload=payload,
            vehicle_info=vehicle_info,
            driver_info=driver_info,
            carrier_name=carrier_name,
            start_point=start_point,
            end_point=end_point,
            freight_amount=freight_amount,
            user_id=user.id,
        )
        transport_no = transport.transport_no
    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",
            create_by=user.id,
            create_dept_id=user.dept_id,
        )
        db.add(transport)
        _apply_transport_arrange_fields(
            transport,
            payload=payload,
            vehicle_info=vehicle_info,
            driver_info=driver_info,
            carrier_name=carrier_name,
            start_point=start_point,
            end_point=end_point,
            freight_amount=freight_amount,
            user_id=user.id,
        )

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

    plan.transport_id = transport.id
    plan.transport_no = transport_no
    plan.vehicle_id = transport.vehicle_id
    plan.license_plate = transport.license_plate
    plan.driver_id = transport.driver_id
    plan.driver_name = transport.driver_name
    if transport.vehicle_id:
        snapshot = build_transport_resource_snapshot(
            db,
            tenant_no,
            transport.vehicle_id,
            selected_driver_id=transport.driver_id,
        )
        apply_resource_snapshot_json(plan=plan, transport=transport, snapshot=snapshot)
    plan.transport_status = "arranged"
    plan.follow_status = "arranged"
    plan.freight_amount = freight_amount or 0
    plan.update_by = user.id

    # 安排运输时：关联且仍为「已审批」的合同自动开始执行
    from app.services.pl_contract_plan_qty import auto_start_execute_approved_contracts

    auto_start_execute_approved_contracts(
        db, tenant_no=tenant_no, plan=plan, user_id=user.id
    )

    needs_site_link = plan_requires_site_link(db, tenant_no=tenant_no, plan=plan)

    if is_rearrange:
        _ensure_driver_access_token(transport)
        if needs_site_link and plan_has_load_stop(db, tenant_no, plan.id):
            ensure_transport_site_token(transport)
        elif not needs_site_link:
            transport.site_access_token = None
            if plan.plan_type in ("inbound", "outbound"):
                void_io_orders_for_plan(
                    db,
                    tenant_no=tenant_no,
                    plan_id=plan.id,
                    user_id=user.id,
                )
    else:
        _ensure_driver_access_token(transport)

    site_records: list[TransportSiteLinkRead] = []
    inventory_records: list[TransportInventoryLinkRead] = []
    io_order_brief: Optional[TransportIoOrderBrief] = None

    if needs_site_link:
        use_multi_stop_site = plan_has_load_stop(db, tenant_no, plan.id)
        if use_multi_stop_site:
            site_token = ensure_transport_site_token(transport)
            site_url = build_transport_site_url(site_token)
            site_records = [
                TransportSiteLinkRead(
                    plan_product_id="",
                    product_name=plan.plan_no or transport.transport_no,
                    site_link_id=transport.id,
                    plan_no=plan.plan_no or "",
                    business_type="按段现场",
                    site_access_token=site_token,
                    site_url=site_url,
                )
            ]
        elif plan.plan_type in ("inbound", "outbound"):
            try:
                if is_rearrange:
                    # 调整运输：先作废旧出入库单（现场链接随之失效），再重建
                    void_io_orders_for_plan(
                        db,
                        tenant_no=tenant_no,
                        plan_id=plan.id,
                        user_id=user.id,
                    )
                io_result = create_io_order_for_transport(
                    db,
                    tenant_no=tenant_no,
                    plan=plan,
                    products=products,
                    transport=transport,
                    user_id=user.id,
                    dept_id=user.dept_id or "",
                    after_dibang=False,
                )
                if io_result:
                    io_order_brief = TransportIoOrderBrief(
                        kind=io_result.kind,
                        order_id=io_result.order_id,
                        order_no=io_result.order_no,
                        site_access_token=io_result.site_access_token,
                        site_url=io_result.site_url,
                        status="in_progress",
                    )
                    # 兼容旧前端：合成单条 siteRecords 供复制现场链接
                    site_records = [
                        TransportSiteLinkRead(
                            plan_product_id="",
                            product_name=io_result.order_no,
                            site_link_id=io_result.order_id,
                            plan_no=plan.plan_no or "",
                            business_type="入库" if io_result.kind == "inbound" else "出库",
                            site_access_token=io_result.site_access_token,
                            site_url=io_result.site_url,
                        )
                    ]
            except ValueError as exc:
                db.rollback()
                raise HTTPException(status_code=400, detail=str(exc)) from exc

    use_multi_stop_dibang = plan_has_load_stop(db, tenant_no, plan.id)
    stop_dibang_options: list[StopDibangOption] = []
    if use_multi_stop_dibang and payload.stop_dibang_options:
        stop_dibang_options = [
            StopDibangOption(
                stop_type=item.stop_type,  # type: ignore[arg-type]
                stop_id=item.stop_id,
                generate_dibang=item.generate_dibang,
            )
            for item in payload.stop_dibang_options
        ]
    should_sync_legacy_dibang = (
        not use_multi_stop_dibang
        and plan.plan_type in ("inbound", "outbound")
        and payload.generate_dibang_plan
    )
    # 有 stop 选项即同步：全 false 也要进 upsert，才能取消勾选 / 已过磅阻断
    should_sync_stop_dibang = use_multi_stop_dibang and bool(stop_dibang_options)

    if should_sync_stop_dibang or should_sync_legacy_dibang:
        try:
            if should_sync_stop_dibang:
                link_results = sync_dibang_for_plan_stops(
                    db,
                    tenant_no=tenant_no,
                    plan=plan,
                    transport=transport,
                    stop_options=stop_dibang_options,
                    user_id=user.id,
                    dept_id=user.dept_id or "",
                )
            else:
                link_results = sync_dibang_for_transport(
                    db,
                    tenant_no=tenant_no,
                    plan=plan,
                    products=products,
                    transport=transport,
                    user_id=user.id,
                    dept_id=user.dept_id or "",
                )
            inventory_records = [
                build_transport_inventory_link_read(
                    db,
                    plan_product_id=item.plan_product_id,
                    product_name=item.product_name,
                    db_purchase_plan_id=item.db_purchase_plan_id,
                    db_inventory_id=item.db_inventory_id,
                    batch_no=item.batch_no,
                    plan_no=item.plan_no,
                )
                for item in link_results
            ]
        except ValueError as exc:
            db.rollback()
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        apply_dibang_prefill_to_order(
            db, tenant_no=tenant_no, plan=plan, transport=transport
        )

    db.commit()
    db.refresh(transport)
    return _build_arrange_result(
        transport,
        site_records,
        inventory_records,
        io_order_brief,
        contract_qty_reminder=contract_qty_reminder_payload,
    )


# =============================================================================
# 获取运输详情
# =============================================================================


@router.get("/{id}", response_model=TransportRead)
def get_transport(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> TransportRead:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    transport = _get_transport(db, tenant_no, id, user)
    return _transport_to_read(db, transport)


# =============================================================================
# 生成二维码
# =============================================================================


class QrCodeResponse(BaseModel):
    transport_id: str = Field(alias="transportId")
    transport_no: str = Field(alias="transportNo")
    qr_code: str = Field(alias="qrCode")
    expire_date: Optional[date] = Field(default=None, alias="expireDate")

    model_config = {"from_attributes": True, "populate_by_name": True}


@router.get("/{id}/qrcode", response_model=QrCodeResponse)
def generate_qrcode(
    id: str,
    expire_days: int = Query(default=30, ge=1, le=365, description="有效期天数"),
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> QrCodeResponse:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    transport = _get_transport(db, tenant_no, id, user)

    from datetime import timedelta
    expire_date = date.today() + timedelta(days=expire_days)

    # 生成二维码内容（URL scheme）
    qr_content = f"/transport/scan/{transport.id}?code={transport.transport_no}"

    transport.qr_code = qr_content
    transport.qr_expire_date = expire_date
    db.commit()

    return QrCodeResponse(
        transport_id=transport.id,
        transport_no=transport.transport_no,
        qr_code=qr_content,
        expire_date=expire_date,
    )


# =============================================================================
# 司机扫码回传（无需登录）
# =============================================================================


@router.post("/scan/feedback")
def transport_scan_feedback(
    payload: TransportFeedback,
    db: Session = Depends(get_db),
) -> dict:
    """司机扫码回传，无需登录"""
    transport = (
        db.query(PlTransport)
        .filter(
            PlTransport.transport_no == payload.transport_no,
            PlTransport.del_flag == 0,
        )
        .first()
    )
    if not transport:
        raise HTTPException(status_code=404, detail="运输单不存在")

    if payload.verify_code and payload.verify_code != transport.transport_no:
        raise HTTPException(status_code=400, detail="验证码错误")

    # 更新回传信息
    if payload.feedback_qty:
        transport.feedback_qty = payload.feedback_qty

    if payload.feedback_images:
        transport.feedback_images = json.dumps(payload.feedback_images)

    if payload.feedback_remark:
        transport.feedback_remark = payload.feedback_remark

    transport.feedback_time = datetime.now()
    db.commit()

    return {"success": True, "message": "回传成功"}


# =============================================================================
# 提交审批
# =============================================================================


@router.post("/{id}/submit-approval", response_model=TransportRead)
def submit_approval(
    id: str,
    payload: ApprovalSubmit,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> TransportRead:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    transport = _get_transport(db, tenant_no, id, user)

    if transport.approval_status not in ("none", "rejected"):
        raise HTTPException(status_code=400, detail="当前状态不允许提交审批")

    # 创建审批记录
    for idx, approver_id in enumerate(payload.approver_ids):
        from app.models.models import SysUser

        approver = db.query(SysUser).filter(SysUser.id == approver_id).first()
        if not approver:
            raise HTTPException(status_code=400, detail=f"审批人不存在: {approver_id}")

        approval = PlTransportApproval(
            id=generate_uuid(),
            tenant_no=tenant_no,
            transport_id=id,
            approver_id=approver_id,
            approver_name=approver.nick_name,
            approval_order=idx + 1,
            status="pending",
            create_by=user.id,
            create_dept_id=user.dept_id,
        )
        db.add(approval)

    transport.approval_status = "pending"
    db.commit()
    db.refresh(transport)
    return transport


# =============================================================================
# 审批
# =============================================================================


@router.post("/approval/{id}", response_model=ApprovalRecordRead)
def process_approval(
    id: str,
    payload: ApprovalRequest,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> ApprovalRecordRead:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())

    approval = (
        db.query(PlTransportApproval)
        .filter(
            PlTransportApproval.id == id,
            PlTransportApproval.tenant_no == tenant_no,
            PlTransportApproval.del_flag == 0,
        )
        .first()
    )
    if not approval:
        raise HTTPException(status_code=404, detail="审批记录不存在")

    if approval.status != "pending":
        raise HTTPException(status_code=400, detail="该审批已处理")

    # 更新审批状态
    from datetime import datetime

    approval.status = "approved" if payload.action == "approve" else "rejected"
    approval.comment = payload.comment
    approval.approval_time = datetime.now()

    # 更新运输审批状态
    transport = db.query(PlTransport).filter(PlTransport.id == approval.transport_id).first()
    if transport:
        if payload.action == "approve":
            # 检查是否所有审批人都通过了
            pending = (
                db.query(PlTransportApproval)
                .filter(
                    PlTransportApproval.transport_id == approval.transport_id,
                    PlTransportApproval.status == "pending",
                    PlTransportApproval.del_flag == 0,
                )
                .count()
            )
            if pending == 0:
                transport.approval_status = "approved"
                transport.transport_status = "pending"
        else:
            transport.approval_status = "rejected"
            transport.transport_status = "draft"

    db.commit()
    db.refresh(approval)
    return approval


# =============================================================================
# 获取运输的审批记录
# =============================================================================


@router.get("/{id}/approvals", response_model=list[ApprovalRecordRead])
def get_transport_approvals(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> list[ApprovalRecordRead]:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())

    approvals = (
        db.query(PlTransportApproval)
        .filter(
            PlTransportApproval.transport_id == id,
            PlTransportApproval.tenant_no == tenant_no,
            PlTransportApproval.del_flag == 0,
        )
        .order_by(PlTransportApproval.approval_order)
        .all()
    )
    return approvals


# =============================================================================
# 辅助函数
# =============================================================================


def _get_transport(db: Session, tenant_no: str, transport_id: str, user: DbUserContext) -> PlTransport:
    transport = (
        db.query(PlTransport)
        .filter(
            PlTransport.id == transport_id,
            PlTransport.tenant_no == tenant_no,
            PlTransport.del_flag == 0,
        )
        .first()
    )
    if not transport:
        raise HTTPException(status_code=404, detail="运输不存在")
    assert_pl_row_in_scope(db, transport, PlTransport, user, tenant_no)
    return transport