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

"""运输计划管理 - 计划 API"""
from datetime import date
from typing import Optional

from fastapi import APIRouter, Body, Depends, HTTPException, Query, status
from sqlalchemy import func, or_
from sqlalchemy.orm import Session

from app.core.config import settings
from app.core.database import get_db
from app.core.db_auth import DbUserContext, get_db_current_user
from app.core.delete_guard import assert_row_deletable_by_creator, user_info_from_context
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.core.utils import generate_uuid
from app.models.crm_models import CrmContract, CrmContractProduct, CrmProduct
from app.models.db_models import DbInboundOutbound
from app.models.pl_models import (
    PlCarrier,
    PlDriver,
    PlPlan,
    PlPlanProduct,
    PlPlanTransport,
    PlTransport,
    PlTransportDbLink,
    PlTransportSiteLink,
    PlVehicle,
)
from app.schemas.pl import PlanProductCreate
from app.schemas.pl.plan import (
    CreateFromContractRequest,
    PlanArchiveBody,
    PlanCreate,
    PlanCreateFull,
    PlanListQuery,
    PlanListResponse,
    PlanProductCertsUpdate,
    PlanProductRead,
    PlanRead,
    PlanUpdate,
    PlanWithProducts,
)
from app.schemas.pl.transport import PlanArrangeInfoRead, TransportInventoryLinkRead, TransportSiteLinkRead
from app.services.pl_dibang_bridge_service import revoke_regeneratable_dibang_for_transport
from app.services.pl_inventory_link_read import build_transport_inventory_link_read
from app.services.pl_site_link_read import build_transport_site_link_read_from_row
from app.services.pl_transport_site_service import revoke_site_links_for_transport
from app.services.pl_plan_service import (
    apply_archive_freight,
    apply_archive_product_quantities,
    apply_vehicle_prefill_to_plan,
    build_plan_product,
    calc_plan_total_amount,
    clear_draft_transport_for_plan,
    determine_plan_type,
    record_plan_change,
    resolve_hazard_flags_from_crm,
    sync_transport_prefill_for_plan,
    update_plan_product_certs,
    validate_create_full,
)

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


_PLAN_NO_SEQ_WIDTH = 3


def _generate_plan_no(db: Session) -> str:
    """生成计划编号 SXDDHYYYYMMDD###（当日 3 位序号，每日重置）"""
    today = date.today()
    prefix = f"SXDDH{today.strftime('%Y%m%d')}"

    # uk_plan_no 对全表生效（含软删），取号时不能过滤 del_flag
    rows = (
        db.query(PlPlan.plan_no)
        .filter(PlPlan.plan_no.like(f"{prefix}%"))
        .all()
    )

    max_seq = 0
    for (plan_no,) in rows:
        suffix = (plan_no or "")[len(prefix) :]
        if suffix.isdigit():
            max_seq = max(max_seq, int(suffix))

    seq = max_seq + 1

    return f"{prefix}{seq:0{_PLAN_NO_SEQ_WIDTH}d}"


def _determine_plan_type(supplier_name: Optional[str], destination_name: Optional[str]) -> str:
    """根据供应商和目的地自动判断单据类型"""
    return determine_plan_type(supplier_name, destination_name)


def _get_plan_type_label(plan_type: str) -> str:
    labels = {
        "direct": "直销单",
        "inbound": "入库单",
        "outbound": "出库单",
    }
    return labels.get(plan_type, plan_type)


def _generate_notify_message(plan: PlPlan, products: list) -> str:
    """生成通知信息"""
    product_names = "、".join([p.product_name for p in products[:3]])
    if len(products) > 3:
        product_names += "等"

    return f"""计划号：{plan.plan_no}
企业名称：{plan.customer_name or ''}
提货单位：{plan.destination_name or ''}
到货时间：{plan.contract_end_date or ''}
药品名称：{product_names}
到货吨位：{sum(float(p.inbound_estimate_qty or p.outbound_estimate_qty or 0) for p in products)}吨
备注："""


# =============================================================================
# 计划列表
# =============================================================================


@router.get("/", response_model=PlanListResponse)
def list_plans(
    q: Optional[str] = Query(default=None, description="模糊搜索"),
    plan_no: Optional[str] = Query(default=None, description="计划编号"),
    plan_type: Optional[str] = Query(default=None, alias="planType", description="单据类型"),
    follow_status: Optional[str] = Query(default=None, alias="followStatus", description="跟进状态"),
    transport_status: Optional[str] = Query(default=None, alias="transportStatus", description="运输状态"),
    archive_status: Optional[str] = Query(default=None, alias="archiveStatus", description="归档状态"),
    customer_name: Optional[str] = Query(default=None, alias="customerName", description="客户名称"),
    plan_date_from: Optional[date] = Query(default=None, alias="planDateFrom", description="计划日期开始"),
    plan_date_to: Optional[date] = Query(default=None, alias="planDateTo", description="计划日期结束"),
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=20, ge=1, le=100, alias="pageSize"),
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> PlanListResponse:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    query = db.query(PlPlan).filter(
        PlPlan.tenant_no == tenant_no,
        PlPlan.del_flag == 0,
    )

    if q:
        query = query.filter(
            or_(
                PlPlan.plan_no.contains(q),
                func.lower(PlPlan.customer_name).contains(q.lower()),
            )
        )

    if plan_no:
        query = query.filter(PlPlan.plan_no.contains(plan_no))

    if plan_type:
        query = query.filter(PlPlan.plan_type == plan_type)

    if follow_status:
        query = query.filter(PlPlan.follow_status == follow_status)

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

    if archive_status:
        query = query.filter(PlPlan.archive_status == archive_status)

    if customer_name:
        query = query.filter(func.lower(PlPlan.customer_name).contains(customer_name.lower()))

    if plan_date_from:
        query = query.filter(PlPlan.plan_date >= plan_date_from)

    if plan_date_to:
        query = query.filter(PlPlan.plan_date <= plan_date_to)

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

    # 统计总数
    total = query.count()

    # 分页
    offset = (page - 1) * page_size
    plans = query.order_by(PlPlan.create_time.desc()).offset(offset).limit(page_size).all()

    plan_ids = [plan.id for plan in plans]
    products_by_plan: dict[str, list[PlPlanProduct]] = {}
    if plan_ids:
        all_products = (
            db.query(PlPlanProduct)
            .filter(
                PlPlanProduct.plan_id.in_(plan_ids),
                PlPlanProduct.del_flag == 0,
            )
            .order_by(PlPlanProduct.order_num)
            .all()
        )
        for product in all_products:
            products_by_plan.setdefault(product.plan_id, []).append(product)

    result_list: list[PlanWithProducts] = []
    for plan in plans:
        enriched_products: list[PlanProductRead] = []
        for product in products_by_plan.get(plan.id, []):
            is_toxic, is_explosive = resolve_hazard_flags_from_crm(
                db,
                tenant_no,
                product.product_id,
                product.is_toxic,
                product.is_explosive,
            )
            item = PlanProductRead.model_validate(product)
            if is_toxic != int(product.is_toxic or 0) or is_explosive != int(product.is_explosive or 0):
                item = item.model_copy(update={"is_toxic": is_toxic, "is_explosive": is_explosive})
            enriched_products.append(item)

        plan_item = PlanWithProducts.model_validate(plan)
        plan_item.products = enriched_products
        if not (plan_item.personnel_summary or "").strip():
            snapshot = plan_item.resource_snapshot
            if snapshot and snapshot.personnel_summary:
                plan_item.personnel_summary = snapshot.personnel_summary
            elif plan_item.driver_name:
                plan_item.personnel_summary = f"{plan_item.driver_name}(司机)"
        result_list.append(plan_item)

    return PlanListResponse(total=total, list=result_list)


@router.get("/count-by-contract")
def count_plans_by_contract(
    contract_ids: str = Query(..., alias="contractIds", description="合同ID，逗号分隔"),
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> dict:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    ids = [item.strip() for item in contract_ids.split(",") if item.strip()]
    if not ids:
        return {"counts": {}}

    rows = (
        db.query(PlPlan.contract_id, func.count(PlPlan.id))
        .filter(
            PlPlan.tenant_no == tenant_no,
            PlPlan.del_flag == 0,
            PlPlan.contract_id.in_(ids),
        )
    )
    rows = apply_pl_scope(rows, PlPlan, user, db, tenant_no).group_by(PlPlan.contract_id).all()
    return {"counts": {contract_id: count for contract_id, count in rows}}


# =============================================================================
# 创建计划
# =============================================================================


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

    # 生成计划编号
    plan_no = _generate_plan_no(db)

    # 自动判断单据类型
    plan_type = payload.plan_type
    if not plan_type:
        plan_type = _determine_plan_type(payload.supplier_name, payload.destination_name)

    plan = PlPlan(
        id=generate_uuid(),
        tenant_no=tenant_no,
        plan_no=plan_no,
        plan_date=payload.plan_date,
        belongs_company=payload.belongs_company,
        plan_type=plan_type,
        follow_status=payload.follow_status or "draft",
        transport_status=payload.transport_status or "pending",
        archive_status=payload.archive_status or "unarchived",
        complete_status=payload.complete_status or "no",
        contract_id=payload.contract_id,
        contract_no=payload.contract_no,
        supplier_id=payload.supplier_id,
        supplier_name=payload.supplier_name,
        destination_id=payload.destination_id,
        destination_name=payload.destination_name,
        customer_id=payload.customer_id,
        customer_name=payload.customer_name,
        contact_name=payload.contact_name,
        contact_phone=payload.contact_phone,
        party_a=payload.party_a,
        party_b=payload.party_b,
        contract_start_date=payload.contract_start_date,
        contract_end_date=payload.contract_end_date,
        delivery_method=payload.delivery_method,
        settlement_method=payload.settlement_method,
        has_freight=payload.has_freight,
        pickup_address=payload.pickup_address,
        pickup_lng=payload.pickup_lng,
        pickup_lat=payload.pickup_lat,
        pickup_time=payload.pickup_time,
        pickup_notes=payload.pickup_notes,
        destination_address=payload.destination_address,
        destination_lng=payload.destination_lng,
        destination_lat=payload.destination_lat,
        delivery_time=payload.delivery_time,
        destination_notes=payload.destination_notes,
        follow_user_id=payload.follow_user_id or user.id,
        follow_user_name=payload.follow_user_name or user.nick_name,
        remark=payload.remark,
        create_by=user.id,
        create_dept_id=user.dept_id,
    )
    db.add(plan)

    # 添加产品明细
    if payload.products:
        for idx, prod in enumerate(payload.products):
            pp = build_plan_product(
                tenant_no=tenant_no,
                plan_id=plan.id,
                plan_type=plan_type,
                prod=prod,
                order_num=idx,
                user_id=user.id,
                dept_id=user.dept_id,
            )
            db.add(pp)

    plan.total_amount = calc_plan_total_amount(payload.products or [])

    db.commit()
    db.refresh(plan)
    return plan


@router.post("/create-full", response_model=PlanWithProducts, status_code=status.HTTP_201_CREATED)
def create_plan_full(
    payload: PlanCreateFull,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> PlanWithProducts:
    """创建运输计划（完整表单：合同产品 + 可选预选车辆，不自动安排运输）"""
    validate_create_full(payload)
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    plan_no = _generate_plan_no(db)

    plan_type = payload.plan_type or _determine_plan_type(payload.supplier_name, payload.destination_name)
    has_vehicle = bool(payload.transport and payload.transport.vehicle_id)

    plan = PlPlan(
        id=generate_uuid(),
        tenant_no=tenant_no,
        plan_no=plan_no,
        plan_date=payload.plan_date,
        belongs_company=payload.belongs_company,
        plan_type=plan_type,
        follow_status="draft",
        transport_status="pending",
        archive_status="unarchived",
        complete_status="no",
        contract_id=payload.contract_id,
        contract_no=payload.contract_no,
        supplier_id=payload.supplier_id,
        supplier_name=payload.supplier_name,
        destination_id=payload.destination_id,
        destination_name=payload.destination_name,
        customer_id=payload.customer_id,
        customer_name=payload.customer_name,
        contact_name=payload.contact_name,
        contact_phone=payload.contact_phone,
        party_a=payload.party_a,
        party_b=payload.party_b,
        contract_start_date=payload.contract_start_date,
        contract_end_date=payload.contract_end_date,
        delivery_method=payload.delivery_method,
        settlement_method=payload.settlement_method,
        has_freight=payload.has_freight,
        pickup_address=payload.pickup_address,
        pickup_lng=payload.pickup_lng,
        pickup_lat=payload.pickup_lat,
        pickup_time=payload.pickup_time,
        pickup_notes=payload.pickup_notes,
        destination_address=payload.destination_address,
        destination_lng=payload.destination_lng,
        destination_lat=payload.destination_lat,
        delivery_time=payload.delivery_time,
        destination_notes=payload.destination_notes,
        total_amount=calc_plan_total_amount(payload.products or []),
        follow_user_id=payload.follow_user_id or user.id,
        follow_user_name=payload.follow_user_name or user.nick_name,
        remark=payload.remark,
        create_by=user.id,
        create_dept_id=user.dept_id,
    )
    db.add(plan)
    db.flush()

    plan_products: list[PlPlanProduct] = []
    for idx, prod in enumerate(payload.products or []):
        is_toxic, is_explosive = resolve_hazard_flags_from_crm(
            db,
            tenant_no,
            prod.product_id,
            prod.is_toxic,
            prod.is_explosive,
        )
        prod_payload = prod.model_copy(update={"is_toxic": is_toxic, "is_explosive": is_explosive})
        pp = build_plan_product(
            tenant_no=tenant_no,
            plan_id=plan.id,
            plan_type=plan_type,
            prod=prod_payload,
            order_num=idx,
            user_id=user.id,
            dept_id=user.dept_id,
        )
        db.add(pp)
        plan_products.append(pp)

    if has_vehicle and payload.transport:
        apply_vehicle_prefill_to_plan(
            db,
            tenant_no=tenant_no,
            plan=plan,
            transport_payload=payload.transport,
            products=plan_products,
            user_id=user.id,
            dept_id=user.dept_id,
        )

    db.commit()
    db.refresh(plan)

    result = PlanWithProducts.model_validate(plan)
    result.products = plan_products
    return result


# =============================================================================
# 从合同创建计划
# =============================================================================


@router.post("/create-from-contract", response_model=PlanRead, status_code=status.HTTP_201_CREATED)
def create_from_contract(
    payload: CreateFromContractRequest,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> PlanRead:
    """从 CRM 合同发起计划"""
    from app.models.crm_models import CrmContract, CrmContractProduct

    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())

    # 查询合同信息
    contract = (
        db.query(CrmContract)
        .filter(CrmContract.id == payload.contract_id, CrmContract.del_flag == 0)
        .first()
    )
    if not contract:
        raise HTTPException(status_code=404, detail="合同不存在")

    # 生成计划编号
    plan_no = _generate_plan_no(db)

    # 自动判断单据类型
    plan_type = _determine_plan_type(contract.supplier_name, contract.customer_name)

    plan = PlPlan(
        id=generate_uuid(),
        tenant_no=tenant_no,
        plan_no=plan_no,
        plan_date=payload.plan_date,
        belongs_company=contract.belongs_company or "shanxia",
        plan_type=plan_type,
        follow_status="draft",
        transport_status="pending",
        archive_status="unarchived",
        complete_status="no",
        contract_id=contract.id,
        contract_no=contract.contract_no,
        supplier_id=contract.supplier_id,
        supplier_name=(contract.party_a or contract.supplier_name or "").strip() or None,
        destination_id=contract.customer_id,
        destination_name=(contract.party_b or contract.customer_name or "").strip() or None,
        customer_id=contract.customer_id,
        customer_name=(contract.party_b or contract.customer_name or "").strip() or None,
        contact_name=contract.contact_name,
        contact_phone=contract.contact_phone,
        party_a=contract.party_a,
        party_b=contract.party_b,
        contract_start_date=contract.start_date,
        contract_end_date=contract.end_date,
        delivery_method=contract.delivery_method,
        settlement_method=contract.settlement_method,
        has_freight=1 if contract.settlement_method and "含运" in contract.settlement_method else 0,
        follow_user_id=payload.follow_user_id or user.id,
        follow_user_name=payload.follow_user_name or user.nick_name,
        create_by=user.id,
        create_dept_id=user.dept_id,
    )
    db.add(plan)

    # 查询合同产品明细
    contract_products = (
        db.query(CrmContractProduct)
        .filter(
            CrmContractProduct.contract_id == contract.id,
            CrmContractProduct.del_flag == 0,
        )
        .all()
    )

    # 添加产品明细
    for idx, cp in enumerate(contract_products):
        is_toxic, is_explosive = resolve_hazard_flags_from_crm(
            db,
            tenant_no,
            cp.product_id,
            0,
            0,
        )
        pp = PlPlanProduct(
            id=generate_uuid(),
            tenant_no=tenant_no,
            plan_id=plan.id,
            product_id=cp.product_id,
            product_no=cp.product_no,
            product_name=cp.product_name,
            model=cp.specification,
            unit=cp.unit or "吨",
            is_explosive=is_explosive,
            is_toxic=is_toxic,
            inbound_estimate_qty=cp.quantity,
            outbound_estimate_qty=cp.quantity,
            order_num=idx,
            create_by=user.id,
            create_dept_id=user.dept_id,
        )
        db.add(pp)

    db.commit()
    db.refresh(plan)
    return plan


# =============================================================================
# 获取计划详情
# =============================================================================


@router.get("/{id}", response_model=PlanWithProducts)
def get_plan(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> PlanWithProducts:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    plan = _get_plan(db, tenant_no, id, user)

    products = (
        db.query(PlPlanProduct)
        .filter(
            PlPlanProduct.plan_id == id,
            PlPlanProduct.del_flag == 0,
        )
        .order_by(PlPlanProduct.order_num)
        .all()
    )

    enriched_products = []
    for product in products:
        is_toxic, is_explosive = resolve_hazard_flags_from_crm(
            db,
            tenant_no,
            product.product_id,
            product.is_toxic,
            product.is_explosive,
        )
        item = PlanProductRead.model_validate(product)
        if is_toxic != int(product.is_toxic or 0) or is_explosive != int(product.is_explosive or 0):
            item = item.model_copy(update={"is_toxic": is_toxic, "is_explosive": is_explosive})
        enriched_products.append(item)

    result = PlanWithProducts.model_validate(plan)
    result.products = enriched_products
    return result


# =============================================================================
# 更新计划
# =============================================================================


@router.put("/{id}", response_model=PlanRead)
def update_plan(
    id: str,
    payload: PlanUpdate,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> PlanRead:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    plan = _get_plan(db, tenant_no, id, user)

    # 更新主表字段
    update_data = payload.model_dump(exclude_unset=True, exclude={"products", "transport"})
    for key, value in update_data.items():
        setattr(plan, key, value)

    plan_type = payload.plan_type or plan.plan_type

    # 更新产品明细
    if payload.products is not None:
        existing_products = (
            db.query(PlPlanProduct)
            .filter(
                PlPlanProduct.plan_id == id,
                PlPlanProduct.del_flag == 0,
            )
            .all()
        )
        cert_by_product_id = {
            row.product_id: (row.purchase_cert_url, row.transport_cert_url)
            for row in existing_products
        }

        db.query(PlPlanProduct).filter(
            PlPlanProduct.plan_id == id,
            PlPlanProduct.del_flag == 0,
        ).update({"del_flag": 1})

        plan_products: list[PlPlanProduct] = []
        for idx, prod in enumerate(payload.products):
            is_toxic, is_explosive = resolve_hazard_flags_from_crm(
                db,
                tenant_no,
                prod.product_id,
                prod.is_toxic,
                prod.is_explosive,
            )
            purchase_cert = prod.purchase_cert_url
            transport_cert = prod.transport_cert_url
            if not purchase_cert or not transport_cert:
                old_purchase, old_transport = cert_by_product_id.get(prod.product_id, (None, None))
                purchase_cert = purchase_cert or old_purchase
                transport_cert = transport_cert or old_transport
            prod_payload = prod.model_copy(
                update={
                    "is_toxic": is_toxic,
                    "is_explosive": is_explosive,
                    "purchase_cert_url": purchase_cert,
                    "transport_cert_url": transport_cert,
                }
            )
            pp = build_plan_product(
                tenant_no=tenant_no,
                plan_id=id,
                plan_type=plan_type,
                prod=prod_payload,
                order_num=idx,
                user_id=user.id,
                dept_id=user.dept_id,
            )
            db.add(pp)
            plan_products.append(pp)

        plan.total_amount = calc_plan_total_amount(payload.products)
    else:
        plan_products = (
            db.query(PlPlanProduct)
            .filter(
                PlPlanProduct.plan_id == id,
                PlPlanProduct.del_flag == 0,
            )
            .all()
        )

    # 同步运输预选 / 关联运输单运费
    if payload.transport is not None:
        if payload.transport.vehicle_id:
            sync_transport_prefill_for_plan(
                db,
                tenant_no=tenant_no,
                plan=plan,
                transport_payload=payload.transport,
                products=plan_products,
                user_id=user.id,
                dept_id=user.dept_id,
            )
        else:
            clear_draft_transport_for_plan(
                db,
                tenant_no=tenant_no,
                plan=plan,
                user_id=user.id,
            )

    plan.update_by = user.id
    db.commit()
    db.refresh(plan)
    return plan


@router.patch("/{id}/product-certs", response_model=PlanWithProducts)
def patch_plan_product_certs(
    id: str,
    payload: PlanProductCertsUpdate,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> PlanWithProducts:
    """更新计划危化品产品购买证/合法使用证明（不触发安排运输）"""
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    _get_plan(db, tenant_no, id, user)
    update_plan_product_certs(
        db,
        tenant_no=tenant_no,
        plan_id=id,
        items=payload.products,
        user_id=user.id,
    )
    plan = _get_plan(db, tenant_no, id, user)
    products = (
        db.query(PlPlanProduct)
        .filter(
            PlPlanProduct.plan_id == id,
            PlPlanProduct.del_flag == 0,
        )
        .order_by(PlPlanProduct.order_num)
        .all()
    )
    result = PlanWithProducts.model_validate(plan)
    result.products = products
    return result


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


@router.get("/{id}/arrange-info", response_model=PlanArrangeInfoRead)
def get_plan_arrange_info(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> PlanArrangeInfoRead:
    """获取计划已安排运输后的司机链接与地磅出入库关联（供查看弹窗复制简述/二维码）。"""
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    plan = _get_plan(db, tenant_no, id, user)

    empty = PlanArrangeInfoRead(driver_url=None, license_plate=None, site_records=[], inventory_records=[])
    if plan.follow_status == "cancelled" or plan.transport_status != "arranged" or not plan.transport_id:
        return empty

    transport = (
        db.query(PlTransport)
        .filter(
            PlTransport.id == plan.transport_id,
            PlTransport.tenant_no == tenant_no,
            PlTransport.del_flag == 0,
        )
        .first()
    )
    if not transport or transport.transport_status == "cancelled":
        return empty

    token = (transport.driver_access_token or "").strip()
    driver_url = _build_driver_url(token) if token else None

    site_rows = (
        db.query(PlTransportSiteLink)
        .filter(
            PlTransportSiteLink.tenant_no == tenant_no,
            PlTransportSiteLink.plan_id == id,
            PlTransportSiteLink.transport_id == transport.id,
            PlTransportSiteLink.del_flag == 0,
        )
        .order_by(PlTransportSiteLink.create_time.asc())
        .all()
    )

    product_name_by_id = {
        item.id: item.product_name or ""
        for item in db.query(PlPlanProduct)
        .filter(
            PlPlanProduct.plan_id == id,
            PlPlanProduct.del_flag == 0,
        )
        .all()
    }

    site_records: list[TransportSiteLinkRead] = [
        build_transport_site_link_read_from_row(
            row,
            product_name=product_name_by_id.get(row.plan_product_id, row.product or ""),
        )
        for row in site_rows
    ]

    links = (
        db.query(PlTransportDbLink)
        .filter(
            PlTransportDbLink.tenant_no == tenant_no,
            PlTransportDbLink.plan_id == id,
            PlTransportDbLink.transport_id == transport.id,
            PlTransportDbLink.del_flag == 0,
        )
        .all()
    )

    inventory_records: list[TransportInventoryLinkRead] = []
    for link in links:
        product = (
            db.query(PlPlanProduct)
            .filter(
                PlPlanProduct.id == link.plan_product_id,
                PlPlanProduct.del_flag == 0,
            )
            .first()
        )
        inventory = (
            db.query(DbInboundOutbound)
            .filter(
                DbInboundOutbound.id == link.db_inventory_id,
                DbInboundOutbound.tenant_no == tenant_no,
                DbInboundOutbound.del_flag == 0,
            )
            .first()
        )
        inventory_records.append(
            build_transport_inventory_link_read(
                db,
                plan_product_id=link.plan_product_id,
                product_name=(product.product_name if product else "") or "",
                db_purchase_plan_id=link.db_purchase_plan_id,
                db_inventory_id=link.db_inventory_id,
                batch_no=(inventory.batch_no if inventory else "") or "",
                plan_no=plan.plan_no or "",
            )
        )

    return PlanArrangeInfoRead(
        driver_url=driver_url,
        license_plate=transport.license_plate or plan.license_plate,
        site_records=site_records,
        inventory_records=inventory_records,
    )


# =============================================================================
# 取消计划
# =============================================================================


@router.post("/{id}/cancel", response_model=PlanRead)
def cancel_plan(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> PlanRead:
    """作废运输计划：取消跟进状态、作废关联地磅出入库单、取消运输单。"""
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    plan = _get_plan(db, tenant_no, id, user)

    if plan.follow_status == "cancelled":
        raise HTTPException(status_code=400, detail="计划已作废")

    if plan.archive_status == "archived":
        raise HTTPException(status_code=400, detail="已归档的计划不能作废")

    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:
            try:
                revoke_regeneratable_dibang_for_transport(
                    db,
                    tenant_no=tenant_no,
                    transport_id=transport.id,
                    user_id=user.id,
                )
            except ValueError as exc:
                raise HTTPException(status_code=400, detail=str(exc)) from exc
            revoke_site_links_for_transport(
                db,
                tenant_no=tenant_no,
                transport_id=transport.id,
                user_id=user.id,
            )
            transport.transport_status = "cancelled"
            transport.update_by = user.id

    plan.follow_status = "cancelled"
    plan.transport_status = "pending"
    plan.update_by = user.id
    db.commit()
    db.refresh(plan)
    return plan


# =============================================================================
# 归档计划
# =============================================================================


@router.post("/{id}/archive", response_model=PlanWithProducts)
def archive_plan(
    id: str,
    payload: Optional[PlanArchiveBody] = Body(default=None),
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> PlanWithProducts:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    plan = _get_plan(db, tenant_no, id, user)

    if plan.follow_status == "cancelled":
        raise HTTPException(status_code=400, detail="已作废的计划不能归档")

    already_archived = plan.archive_status == "archived"

    if payload is None:
        if already_archived:
            raise HTTPException(status_code=400, detail="计划已归档，请提交归档表单以调整内容")
        plan.archive_status = "archived"
        plan.complete_status = "yes"
        plan.follow_status = "completed"
        plan.update_by = user.id
        db.commit()
        return get_plan(id, db, user)

    apply_archive_product_quantities(
        db, tenant_no=tenant_no, plan=plan, items=payload.products, user_id=user.id
    )
    if payload.freight is not None:
        apply_archive_freight(
            db, tenant_no=tenant_no, plan=plan, freight=payload.freight, user_id=user.id
        )
    elif int(plan.has_freight or 0) == 1 and not already_archived:
        if not plan.transport_id:
            raise HTTPException(status_code=400, detail="计划有运输费用但未安排运输，请先安排运输")

    if not already_archived:
        plan.archive_status = "archived"
        plan.complete_status = "yes"
        plan.follow_status = "completed"

    plan.update_by = user.id
    record_plan_change(
        db,
        tenant_no=tenant_no,
        plan_id=plan.id,
        change_type="archive_adjust" if already_archived else "archive",
        change_content="调整归档内容" if already_archived else "运输计划归档",
        user_id=user.id,
        user_name=user.name or None,
        dept_id=user.dept_id,
    )
    db.commit()
    return get_plan(id, db, user)


# =============================================================================
# 删除计划
# =============================================================================


@router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_plan(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> None:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    plan = _get_plan(db, tenant_no, id, user)

    assert_row_deletable_by_creator(plan, user_info_from_context(user))
    plan.del_flag = 1
    plan.update_by = user.id

    db.query(PlPlanProduct).filter(
        PlPlanProduct.plan_id == id,
        PlPlanProduct.tenant_no == tenant_no,
        PlPlanProduct.del_flag == 0,
    ).update({"del_flag": 1, "update_by": user.id}, synchronize_session=False)

    db.query(PlPlanTransport).filter(
        PlPlanTransport.plan_id == id,
        PlPlanTransport.tenant_no == tenant_no,
        PlPlanTransport.del_flag == 0,
    ).update({"del_flag": 1, "update_by": user.id}, synchronize_session=False)

    db.commit()


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


def _get_plan(db: Session, tenant_no: str, plan_id: str, user: DbUserContext) -> 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="计划不存在")
    assert_pl_row_in_scope(db, plan, PlPlan, user, tenant_no)
    return plan