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

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

from fastapi import APIRouter, Body, Depends, HTTPException, Query, Request, status
from pydantic import BaseModel, Field
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.core.menu_guard import SUPER_ADMIN_ROLE, get_user_menu_codes
from app.services.pl_data_scope import apply_pl_scope, assert_pl_row_in_scope
from app.services.pl.list_sort import order_plan, parse_sort_params
from app.services.pl_plan_arrange_copy_flags import load_plan_arrange_copy_flags
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.models.crm_models import CrmContract, CrmContractProduct, CrmProduct
from app.models.db_models import DbInboundOutbound
from app.models.pl_models import (
    PlCarrier,
    PlDriver,
    PlPlan,
    PlPlanDelivery,
    PlPlanLoad,
    PlPlanProduct,
    PlPlanTransport,
    PlTransport,
    PlTransportDbLink,
    PlVehicle,
)
from app.schemas.pl import PlanProductCreate
from app.schemas.pl.plan import (
    CreateFromContractRequest,
    DeliverySlipPrintRead,
    PlanArchiveBody,
    PlanArchiveBrandQtyItem,
    PlanContractGapListResponse,
    PlanContractGapRead,
    PlanContractLink,
    PlanCreate,
    PlanCreateFull,
    PlanDriverFeedbackUpdateBody,
    HazardCertReadyBody,
    ErpEnteredBody,
    PlanForceFollowStatusBody,
    PlanInboundSlipTemplateSave,
    PlanDeliverySlipTemplateSave,
    PlanSubpackSlipTemplateSave,
    PlanDeliveryBody,
    PlanLoadBody,
    PlanListQuery,
    PlanListResponse,
    PlanProductCertsUpdate,
    PlanProductRead,
    PlanRead,
    PlanSideContractsUpdateBody,
    PlanSiteFeedbackUpdateBody,
    PlanStopContractsUpdateBody,
    PlanUpdate,
    PlanWithProducts,
)
from app.services.pl_plan_contract_gap import (
    list_plan_contract_gaps,
    update_plan_side_contracts,
    update_plan_stop_contracts,
)
from app.services.pl_brand_archive import load_brand_qtys_by_plan_product_ids
from app.services.pl_product_display import (
    load_crm_product_alias_maps,
    resolve_product_display_names,
)
from app.schemas.pl.transport import (
    PlanArrangeInfoRead,
    TransportInventoryLinkRead,
    TransportIoOrderBrief,
    TransportSiteLinkRead,
)
from app.services.pl_dibang_bridge_service import (
    StopDibangOption,
    revoke_regeneratable_dibang_for_transport,
    sync_linked_dibang_dates_for_plan,
    upsert_stop_dibang_links,
)
from app.services.pl_inventory_link_read import build_transport_inventory_link_read
from app.services.pl_io_order_service import get_active_io_order_for_plan, void_io_orders_for_plan
from app.services.pl_internal_customer import plan_requires_site_link
from app.services.pl_plan_site_link_service import build_site_records_for_arranged_plan
from app.services.pl_transport_site_service import revoke_site_links_for_transport
from app.services.pl_inventory_service import (
    apply_archive_inventory_for_plan,
    apply_archive_inventory_for_plan_stops,
)
from app.services.pl_plan_archive_stop_service import (
    apply_archive_stop_quantities,
    sum_delivery_outbound_qty,
    sync_plan_products_from_stop_archive,
)
from app.services.pl_plan_archive_export import build_plan_archive_export_rows
from app.services.pl_plan_price_mask import (
    mask_plan_with_products_prices,
    user_can_archive_plan,
    user_can_view_archive_prices,
)
from app.services.pl_plan_list_attachments import load_plan_document_attachments_by_plan_ids
from app.services.pl_plan_print_service import get_delivery_slip_print_context
from app.services.pl_plan_stop_service import (
    attach_plan_stops_to_item,
    collect_stop_qty_warnings,
    collect_unique_delivery_date_strings,
    load_plan_stops_grouped,
    replace_plan_stops,
    sync_plan_legacy_from_stops,
    uses_multi_stop_payload,
    validate_plan_stops_write,
)
from app.services.pl_customer_stop_pref_service import safe_upsert_prefs_from_plan_stops
from app.services.pl.plan_list_filters import (
    build_plan_column_filters,
    plan_plate_match,
    plan_product_name_exists,
)
from app.services.pl_plan_service import (
    apply_archive_brand_qtys_only,
    apply_archive_freight,
    apply_archive_product_quantities,
    apply_archive_stop_feedback_attachments,
    apply_plan_driver_feedback,
    apply_plan_site_feedback,
    apply_vehicle_prefill_to_plan,
    assert_contract_selectable_for_plan,
    apply_plan_product_fields,
    build_plan_product,
    resolve_plan_product_id_for_upsert,
    calc_plan_total_amount,
    clear_draft_transport_for_plan,
    clear_hazard_cert_ready,
    collect_hazard_product_ids,
    determine_plan_type,
    fetch_hazard_flags_map,
    load_plan_contracts_grouped,
    plan_has_hazard_product_exists,
    record_plan_change,
    replace_plan_contracts,
    resolve_hazard_flags_from_crm,
    set_hazard_cert_ready,
    set_plan_erp_entered,
    sync_transport_prefill_for_plan,
    update_plan_product_certs,
    assert_can_save_as_draft,
    resolve_follow_status_on_save,
    validate_create_full,
    validate_draft_min_fields,
    validate_hazard_cert_ready_for_transport,
    validate_plan_party_contracts,
)

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


_PLAN_NO_SEQ_WIDTH = 3


def _resolve_list_product_name(
    product: PlPlanProduct,
    *,
    standard_by_id: dict[str, str],
    alias_options_by_id: dict[str, list[str]],
) -> str:
    """列表展示用：优先 CRM 标准名（详情仍返回库内 product_name，以便编辑反推别名）。"""
    display_name, _alias = resolve_product_display_names(
        saved_product_name=product.product_name,
        product_id=product.product_id,
        standard_by_id=standard_by_id,
        alias_options_by_id=alias_options_by_id,
    )
    return display_name or (product.product_name or "")


def _resolve_create_date(value: Optional[date]) -> date:
    return value or date.today()


def _generate_plan_no(db: Session) -> str:
    """生成计划编号 POYYYYMMDD###（当日 3 位序号，每日重置）"""
    today = date.today()
    prefix = f"PO{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}"


_SUBPACK_SLIP_NO_SEQ_WIDTH = 3
_SUBPACK_SLIP_NO_CANONICAL_PATTERN = re.compile(r"^NO:SXFZ(\d{8})-(\d+)$")
_SUBPACK_SLIP_NO_BARE_PATTERN = re.compile(r"^SXFZ(\d{8})-(\d+)$")
_SUBPACK_SLIP_NO_LEGACY_PATTERNS = (
    re.compile(r"^NO:SXFZ-(\d{4})-(\d{2})-(\d{2})-(\d+)$"),
    re.compile(r"^SXFZ-(\d{4})-(\d{2})-(\d{2})-(\d+)$"),
)


def _format_subpack_slip_no(day: str, seq: int) -> str:
    return f"NO:SXFZ{day}-{int(seq):0{_SUBPACK_SLIP_NO_SEQ_WIDTH}d}"


def _normalize_subpack_slip_no(slip_no: str) -> Optional[str]:
    """将历史分装单号统一为 NO:SXFZYYYYMMDD-###"""
    raw = (slip_no or "").strip()
    if not raw:
        return None

    matched = _SUBPACK_SLIP_NO_CANONICAL_PATTERN.match(raw)
    if matched:
        day, seq = matched.groups()
        return _format_subpack_slip_no(day, int(seq))

    matched = _SUBPACK_SLIP_NO_BARE_PATTERN.match(raw)
    if matched:
        day, seq = matched.groups()
        return _format_subpack_slip_no(day, int(seq))

    for pattern in _SUBPACK_SLIP_NO_LEGACY_PATTERNS:
        matched = pattern.match(raw)
        if matched:
            year, month, day, seq = matched.groups()
            return _format_subpack_slip_no(f"{year}{month}{day}", int(seq))

    return None


def _subpack_slip_no_seq(slip_no: str, day_prefix: str) -> int:
    normalized = _normalize_subpack_slip_no(slip_no)
    if not normalized or not normalized.startswith(day_prefix):
        return 0
    suffix = normalized[len(day_prefix) :]
    return int(suffix) if suffix.isdigit() else 0


def _ensure_subpack_slip_no(db: Session, tenant_no: str, plan: PlPlan) -> str:
    """分配分装单号 NO:SXFZYYYYMMDD-###（按租户每日递增）"""
    existing = (plan.subpack_slip_no or "").strip()
    if existing:
        normalized = _normalize_subpack_slip_no(existing)
        if normalized:
            if normalized != existing:
                plan.subpack_slip_no = normalized
            return normalized

    today = date.today()
    prefix = f"NO:SXFZ{today.strftime('%Y%m%d')}-"

    rows = (
        db.query(PlPlan.subpack_slip_no)
        .filter(
            PlPlan.tenant_no == tenant_no,
            PlPlan.subpack_slip_no.isnot(None),
            PlPlan.subpack_slip_no != "",
        )
        .all()
    )

    max_seq = 0
    for (slip_no,) in rows:
        max_seq = max(max_seq, _subpack_slip_no_seq(slip_no or "", prefix))

    next_no = f"{prefix}{max_seq + 1:0{_SUBPACK_SLIP_NO_SEQ_WIDTH}d}"
    plan.subpack_slip_no = next_no
    return next_no


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)}吨
备注："""


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


def _parse_status_csv(value: Optional[str]) -> List[str]:
    if not value:
        return []
    return [item.strip() for item in value.split(",") if item.strip()]


def _apply_follow_status_filter(query, follow_status: Optional[str]):
    statuses = _parse_status_csv(follow_status)
    if len(statuses) == 1:
        return query.filter(PlPlan.follow_status == statuses[0])
    if len(statuses) > 1:
        return query.filter(PlPlan.follow_status.in_(statuses))
    return query


def _apply_exclude_follow_status_filter(query, exclude_follow_status: Optional[str]):
    statuses = _parse_status_csv(exclude_follow_status)
    if len(statuses) == 1:
        return query.filter(PlPlan.follow_status != statuses[0])
    if len(statuses) > 1:
        return query.filter(~PlPlan.follow_status.in_(statuses))
    return query


def _merge_query(request: Request, **known) -> dict:
    out = {k: v for k, v in known.items() if v is not None and v != ""}
    for k, v in request.query_params.multi_items():
        if k in ("page", "pageSize"):
            continue
        if k not in out and v not in (None, ""):
            out[k] = v
    return out


def _param_get(params: dict, *keys: str) -> Any:
    for key in keys:
        if key in params and params[key] not in (None, ""):
            return params[key]
    return None


def _filters_from_params(params: dict, *, db: Session, tenant_no: str) -> list:
    return build_plan_column_filters(
        db=db,
        tenant_no=tenant_no,
        plan_no=_param_get(params, "planNo", "plan_no"),
        plan_type=_param_get(params, "planType", "plan_type"),
        plan_type_in=_param_get(params, "planTypeIn", "plan_type_in"),
        transport_status=_param_get(params, "transportStatus", "transport_status"),
        transport_status_in=_param_get(params, "transportStatusIn", "transport_status_in"),
        archive_status=_param_get(params, "archiveStatus", "archive_status"),
        archive_status_in=_param_get(params, "archiveStatusIn", "archive_status_in"),
        customer_name=_param_get(params, "customerName", "customer_name"),
        product_name=_param_get(params, "productName", "product_name"),
        package_spec=_param_get(params, "packageSpec", "package_spec"),
        license_plate=_param_get(params, "licensePlate", "license_plate"),
        plan_date_from=_param_get(params, "planDateFrom", "plan_date_from"),
        plan_date_to=_param_get(params, "planDateTo", "plan_date_to"),
        delivery_time_from=_param_get(params, "deliveryTimeFrom", "delivery_time_from"),
        delivery_time_to=_param_get(params, "deliveryTimeTo", "delivery_time_to"),
        cert_upload_status=_param_get(params, "certUploadStatus", "cert_upload_status"),
        hazard_cert_ready=_param_get(params, "hazardCertReady", "hazard_cert_ready"),
        hazard_cert_ready_in=_param_get(
            params, "hazardCertReadyIn", "hazard_cert_ready_in"
        ),
        supplier_name=_param_get(params, "supplierName", "supplier_name"),
        destination_name=_param_get(params, "destinationName", "destination_name"),
    )


def _body_to_params(body: BaseModel) -> dict:
    data = body.model_dump(by_alias=True, exclude_none=True)
    for key in ("ids", "columns", "exportFormat", "export_format"):
        data.pop(key, None)
    extra = getattr(body, "model_extra", None) or {}
    for key, value in extra.items():
        if key not in data and value not in (None, ""):
            data[key] = value
    return data


@router.get("/", response_model=PlanListResponse)
def list_plans(
    request: Request,
    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="跟进状态，多个以英文逗号分隔",
    ),
    exclude_follow_status: Optional[str] = Query(
        default=None,
        alias="excludeFollowStatus",
        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="客户名称"),
    product_name: Optional[str] = Query(default=None, alias="productName", description="产品名称"),
    package_spec: Optional[str] = Query(default=None, alias="packageSpec", description="包装规格"),
    license_plate: Optional[str] = Query(default=None, alias="licensePlate", description="车牌号"),
    plan_date_from: Optional[date] = Query(default=None, alias="planDateFrom", description="计划日期开始"),
    plan_date_to: Optional[date] = Query(default=None, alias="planDateTo", description="计划日期结束"),
    delivery_time_from: Optional[date] = Query(
        default=None, alias="deliveryTimeFrom", description="送货时间开始"
    ),
    delivery_time_to: Optional[date] = Query(
        default=None, alias="deliveryTimeTo", description="送货时间结束"
    ),
    has_hazard_product: Optional[int] = Query(
        default=None,
        alias="hasHazardProduct",
        description="1=仅含易制毒/易制爆产品的计划",
    ),
    cert_upload_status: Optional[str] = Query(
        default=None,
        alias="certUploadStatus",
        description="incomplete=至少一张必需危化证书未上传",
    ),
    hazard_cert_ready: Optional[int] = Query(
        default=None,
        alias="hazardCertReady",
        description="0=未确认齐备 1=已确认齐备",
    ),
    contract_id: Optional[str] = Query(
        default=None,
        alias="contractId",
        description="CRM合同ID：返回关联该合同的运输计划（含 pl_plan_contract 与遗留主表 contract_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),
) -> 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 contract_id and str(contract_id).strip():
        from app.services.pl_contract_plan_qty import plan_ids_linked_to_contract

        linked_ids = plan_ids_linked_to_contract(
            db, tenant_no=tenant_no, contract_id=str(contract_id).strip()
        )
        if not linked_ids:
            return PlanListResponse(total=0, list=[])
        query = query.filter(PlPlan.id.in_(linked_ids))

    if q:
        q_kw = q.strip()
        if q_kw:
            q_lower = q_kw.lower()
            query = query.filter(
                or_(
                    PlPlan.plan_no.contains(q_kw),
                    func.lower(PlPlan.supplier_name).contains(q_lower),
                    func.lower(PlPlan.customer_name).contains(q_lower),
                    func.lower(PlPlan.destination_name).contains(q_lower),
                    plan_product_name_exists(db, tenant_no, q_kw),
                    plan_plate_match(q_kw),
                )
            )

    query = _apply_follow_status_filter(query, follow_status)
    query = _apply_exclude_follow_status_filter(query, exclude_follow_status)

    if has_hazard_product == 1:
        query = query.filter(plan_has_hazard_product_exists(tenant_no))

    params = _merge_query(
        request,
        planNo=plan_no,
        planType=plan_type,
        transportStatus=transport_status,
        archiveStatus=archive_status,
        customerName=customer_name,
        productName=product_name,
        packageSpec=package_spec,
        licensePlate=license_plate,
        planDateFrom=plan_date_from,
        planDateTo=plan_date_to,
        deliveryTimeFrom=delivery_time_from,
        deliveryTimeTo=delivery_time_to,
        certUploadStatus=cert_upload_status,
        hazardCertReady=hazard_cert_ready,
    )
    column_filters = _filters_from_params(params, db=db, tenant_no=tenant_no)
    if column_filters:
        query = query.filter(*column_filters)

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

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

    # 分页
    offset = (page - 1) * page_size
    sf, so = parse_sort_params(params)
    plans = (
        query.order_by(order_plan(sf, so), PlPlan.id.desc())
        .offset(offset)
        .limit(page_size)
        .all()
    )

    plan_ids = [plan.id for plan in plans]
    products_by_plan: dict[str, list[PlPlanProduct]] = {}
    all_products: 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)

    product_ids = [product.product_id for product in all_products if product.product_id]
    hazard_map = fetch_hazard_flags_map(
        db,
        tenant_no,
        product_ids,
    )
    standard_by_id, alias_options_by_id = load_crm_product_alias_maps(
        db,
        tenant_no=tenant_no,
        product_ids=product_ids,
    )
    brand_qtys_by_pp = load_brand_qtys_by_plan_product_ids(
        db,
        tenant_no=tenant_no,
        plan_product_ids=[p.id for p in all_products],
    )
    contracts_by_plan = load_plan_contracts_grouped(db, tenant_no, plan_ids)
    arrange_copy_flags = load_plan_arrange_copy_flags(db, tenant_no, plans)
    document_attachments_by_plan = load_plan_document_attachments_by_plan_ids(
        db,
        tenant_no=tenant_no,
        plan_ids=plan_ids,
    )
    stops_by_plan = load_plan_stops_grouped(db, tenant_no, plan_ids)
    can_price = user_can_view_archive_prices(db, user)

    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,
                hazard_map=hazard_map,
            )
            item = PlanProductRead.model_validate(product)
            raw_brand_qtys = brand_qtys_by_pp.get(product.id) or None
            brand_qtys = (
                [PlanArchiveBrandQtyItem.model_validate(x) for x in raw_brand_qtys]
                if raw_brand_qtys
                else None
            )
            item = item.model_copy(
                update={
                    "is_toxic": is_toxic,
                    "is_explosive": is_explosive,
                    "brand_qtys": brand_qtys,
                    "product_name": _resolve_list_product_name(
                        product,
                        standard_by_id=standard_by_id,
                        alias_options_by_id=alias_options_by_id,
                    ),
                }
            )
            enriched_products.append(item)

        plan_item = PlanWithProducts.model_validate(plan)
        plan_item.products = enriched_products
        supplier_contracts, demander_contracts = contracts_by_plan.get(plan.id, ([], []))
        plan_item.supplier_contracts = supplier_contracts
        plan_item.demander_contracts = demander_contracts
        flags = arrange_copy_flags.get(plan.id)
        if flags:
            plan_item.can_copy_to_driver = flags.can_copy_to_driver
            plan_item.can_copy_to_site = flags.can_copy_to_site
            plan_item.can_copy_pound_qrcode = flags.can_copy_pound_qrcode
        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}(驾)"
        plan_item.document_attachments = document_attachments_by_plan.get(plan.id, [])
        attach_plan_stops_to_item(
            plan_item,
            plan,
            stops_by_plan.get(plan.id, (None, [])),
        )
        plan_item = mask_plan_with_products_prices(plan_item, enabled=not can_price)
        result_list.append(plan_item)

    return PlanListResponse(total=total, list=result_list)



PLAN_EXPORT_COLUMNS = [
    ExportColumn(title="计划编号", field="planNo"),
    ExportColumn(title="跟进状态", field="followStatus"),
    ExportColumn(title="计划日期", field="planDate"),
    ExportColumn(title="送货时间", field="deliveryTimes"),
    ExportColumn(title="单据类型", field="planType"),
    ExportColumn(title="供方", field="supplierName"),
    ExportColumn(title="需方", field="customerName"),
    ExportColumn(title="车牌号", field="licensePlate"),
    ExportColumn(title="司机", field="driverName"),
    ExportColumn(title="运输状态", field="transportStatus"),
    ExportColumn(title="归档状态", field="archiveStatus"),
    ExportColumn(title="ERP", field="erpEntered"),
    ExportColumn(title="备注", field="remark"),
]

PLAN_ARCHIVE_EXPORT_COLUMNS = [
    ExportColumn(title="序号", field="rowNo"),
    ExportColumn(title="计划编号", field="planNo"),
    ExportColumn(title="计划日期", field="planDate"),
    ExportColumn(title="归档状态", field="archiveStatus"),
    ExportColumn(title="跟进状态", field="followStatus"),
    ExportColumn(title="单据类型", field="planType"),
    ExportColumn(title="段次", field="stopKind"),
    ExportColumn(title="客户", field="stopCustomer"),
    ExportColumn(title="地址", field="stopAddress"),
    ExportColumn(title="段次时间", field="stopTime"),
    ExportColumn(title="产品名称", field="productName"),
    ExportColumn(title="型号/规格", field="model"),
    ExportColumn(title="包装", field="packageSpec"),
    ExportColumn(title="单位", field="unit"),
    ExportColumn(title="计划数量", field="planQty"),
    ExportColumn(title="实际数量", field="actualQty"),
    ExportColumn(title="入库数量", field="inboundQty"),
    ExportColumn(title="出库数量", field="outboundQty"),
    ExportColumn(title="单价（元）", field="unitPrice"),
    ExportColumn(title="金额（元）", field="amount"),
    ExportColumn(title="车牌号", field="licensePlate"),
    ExportColumn(title="运费单位", field="freightType"),
    ExportColumn(title="运费单价", field="freightPrice"),
    ExportColumn(title="运费金额", field="freightAmount"),
    ExportColumn(title="ERP", field="erpEntered"),
    ExportColumn(title="备注", field="remark"),
]


def _load_transports_by_id(db: Session, tenant_no: str, transport_ids: list[str]) -> dict[str, PlTransport]:
    if not transport_ids:
        return {}
    rows = (
        db.query(PlTransport)
        .filter(
            PlTransport.tenant_no == tenant_no,
            PlTransport.del_flag == 0,
            PlTransport.id.in_(transport_ids),
        )
        .all()
    )
    return {row.id: row for row in rows}


class PlanExportBody(BaseModel):
    q: Optional[str] = None
    plan_no: Optional[str] = Field(default=None, alias="planNo")
    plan_type: Optional[str] = Field(default=None, alias="planType")
    follow_status: Optional[str] = Field(default=None, alias="followStatus")
    exclude_follow_status: Optional[str] = Field(default=None, alias="excludeFollowStatus")
    transport_status: Optional[str] = Field(default=None, alias="transportStatus")
    archive_status: Optional[str] = Field(default=None, alias="archiveStatus")
    customer_name: Optional[str] = Field(default=None, alias="customerName")
    product_name: Optional[str] = Field(default=None, alias="productName")
    package_spec: Optional[str] = Field(default=None, alias="packageSpec")
    license_plate: Optional[str] = Field(default=None, alias="licensePlate")
    plan_date_from: Optional[date] = Field(default=None, alias="planDateFrom")
    plan_date_to: Optional[date] = Field(default=None, alias="planDateTo")
    delivery_time_from: Optional[date] = Field(default=None, alias="deliveryTimeFrom")
    delivery_time_to: Optional[date] = Field(default=None, alias="deliveryTimeTo")
    has_hazard_product: Optional[int] = Field(default=None, alias="hasHazardProduct")
    cert_upload_status: Optional[str] = Field(default=None, alias="certUploadStatus")
    hazard_cert_ready: Optional[int] = Field(default=None, alias="hazardCertReady")
    ids: Optional[List[str]] = None
    columns: Optional[List[str]] = None
    export_format: Optional[str] = Field(default=None, alias="exportFormat", description="archive=运输计划归档专用导出")

    model_config = {"populate_by_name": True, "extra": "allow"}


@router.post("/export")
def export_plans(
    body: PlanExportBody,
    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(PlPlan).filter(
            PlPlan.tenant_no == tenant_no,
            PlPlan.del_flag == 0,
        )
        if body.q:
            q_kw = body.q.strip()
            if q_kw:
                q_lower = q_kw.lower()
                query = query.filter(
                    or_(
                        PlPlan.plan_no.contains(q_kw),
                        func.lower(PlPlan.supplier_name).contains(q_lower),
                        func.lower(PlPlan.customer_name).contains(q_lower),
                        func.lower(PlPlan.destination_name).contains(q_lower),
                        plan_product_name_exists(db, tenant_no, q_kw),
                        plan_plate_match(q_kw),
                    )
                )
        query = _apply_follow_status_filter(query, body.follow_status)
        query = _apply_exclude_follow_status_filter(query, body.exclude_follow_status)
        if body.has_hazard_product == 1:
            query = query.filter(plan_has_hazard_product_exists(tenant_no))
        column_filters = _filters_from_params(
            _body_to_params(body), db=db, tenant_no=tenant_no
        )
        if column_filters:
            query = query.filter(*column_filters)
        if body.ids:
            query = query.filter(PlPlan.id.in_(body.ids))
        query = apply_pl_scope(query, PlPlan, user, db, tenant_no)
        export_params = _body_to_params(body)
        sf, so = parse_sort_params(export_params)
        plans = (
            query.order_by(order_plan(sf, so), PlPlan.id.desc())
            .limit(EXPORT_MAX_ROWS + 1)
            .all()
        )
        date_str = datetime.now().strftime("%Y%m%d")
        is_archive_export = (body.export_format or "").strip().lower() == "archive"
        if is_archive_export:
            transport_ids = [plan.transport_id for plan in plans if plan.transport_id]
            transports_by_id = _load_transports_by_id(db, tenant_no, transport_ids)
            items = build_plan_archive_export_rows(db, tenant_no, plans, transports_by_id)
            cols = filter_export_columns(PLAN_ARCHIVE_EXPORT_COLUMNS, body.columns)
            if not user_can_view_archive_prices(db, user):
                cols = [c for c in cols if c.field not in ARCHIVE_PRICE_EXPORT_FIELDS]
            sheet_title = "计划归档"
            filename_stem = f"计划归档_{date_str}"
        else:
            plan_ids = [plan.id for plan in plans]
            stops_by_plan = load_plan_stops_grouped(db, tenant_no, plan_ids)
            items = []
            for plan in plans:
                data = PlanWithProducts.model_validate(plan).model_dump(by_alias=True)
                if data.get("planDate") is not None:
                    data["planDate"] = str(data["planDate"])[:10]
                if data.get("erpEntered") is not None:
                    data["erpEntered"] = "是" if int(data["erpEntered"]) else "否"
                _load_read, delivery_reads = stops_by_plan.get(plan.id, (None, []))
                times = collect_unique_delivery_date_strings(delivery_reads)
                data["deliveryTimes"] = " / ".join(times) if times else ""
                items.append(data)
            cols = filter_export_columns(PLAN_EXPORT_COLUMNS, body.columns)
            sheet_title = "计划列表"
            filename_stem = f"计划列表_{date_str}"
        return export_rows_to_xlsx_response(
            db,
            tenant_no=tenant_no,
            sheet_title=sheet_title,
            columns=cols,
            rows=items,
            filename_stem=filename_stem,
        )
    except ExportValidationError as e:
        raise HTTPException(status_code=400, detail=str(e)) from e


@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:
    from app.models.pl_models import PlPlanContract

    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": {}}

    # 旧单：主表 contract_id；新单：pl_plan_contract 分侧关联（按计划去重计数）
    legacy_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),
        )
    )
    legacy_rows = apply_pl_scope(legacy_rows, PlPlan, user, db, tenant_no).group_by(PlPlan.contract_id).all()
    counts: dict[str, int] = {contract_id: count for contract_id, count in legacy_rows}

    link_q = (
        db.query(PlPlanContract.contract_id, PlPlan.id)
        .join(PlPlan, PlPlan.id == PlPlanContract.plan_id)
        .filter(
            PlPlanContract.tenant_no == tenant_no,
            PlPlanContract.del_flag == 0,
            PlPlanContract.contract_id.in_(ids),
            PlPlan.tenant_no == tenant_no,
            PlPlan.del_flag == 0,
        )
    )
    link_q = apply_pl_scope(link_q, PlPlan, user, db, tenant_no)
    seen: set[tuple[str, str]] = set()
    for contract_id, plan_id in link_q.all():
        key = (contract_id, plan_id)
        if key in seen:
            continue
        seen.add(key)
        # 若旧主表已计过同一计划+合同，不再重复加（新单主表 contract_id 应为空）
        counts[contract_id] = counts.get(contract_id, 0) + 1

    return {"counts": counts}


@router.get("/contract-product-arrange-qty")
def get_contract_product_arrange_qty(
    contract_id: str = Query(..., alias="contractId"),
    product_id: str = Query(..., alias="productId"),
    package_spec: Optional[str] = Query(default=None, alias="packageSpec"),
    exclude_plan_id: Optional[str] = Query(default=None, alias="excludePlanId"),
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> dict:
    """表单可安排数量 + 问号弹窗：合同用量与已安排计划列表。"""
    from app.services.pl_contract_plan_qty import get_contract_product_arrange_qty_detail

    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    return get_contract_product_arrange_qty_detail(
        db,
        tenant_no=tenant_no,
        contract_id=contract_id,
        product_id=product_id,
        package_spec=package_spec,
        exclude_plan_id=exclude_plan_id,
    )


@router.get("/contract-gaps", response_model=PlanContractGapListResponse)
def list_contract_gaps(
    request: Request,
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=20, ge=1, le=500, alias="pageSize"),
    plan_no: Optional[str] = Query(default=None, alias="planNo", description="计划编号"),
    gap_type: Optional[str] = Query(
        default=None,
        alias="gapType",
        description="缺口类型：supplier | demander | both，多个以英文逗号分隔",
    ),
    customer_keyword: Optional[str] = Query(
        default=None,
        alias="customerKeyword",
        description="供方或需方名称关键词",
    ),
    plan_date_from: Optional[date] = Query(default=None, alias="planDateFrom"),
    plan_date_to: Optional[date] = Query(default=None, alias="planDateTo"),
    delivery_time_from: Optional[date] = Query(default=None, alias="deliveryTimeFrom"),
    delivery_time_to: Optional[date] = Query(default=None, alias="deliveryTimeTo"),
    plan_type: Optional[str] = Query(default=None, alias="planType"),
    follow_status: Optional[str] = Query(default=None, alias="followStatus"),
    q: Optional[str] = Query(
        default=None,
        description="模糊搜索（计划编号/供方名称/需方名称）",
    ),
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> PlanContractGapListResponse:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    params = _merge_query(
        request,
        planNo=plan_no,
        gapType=gap_type,
        customerKeyword=customer_keyword,
        planDateFrom=plan_date_from,
        planDateTo=plan_date_to,
        deliveryTimeFrom=delivery_time_from,
        deliveryTimeTo=delivery_time_to,
        planType=plan_type,
        followStatus=follow_status,
    )
    total, items = list_plan_contract_gaps(
        db,
        tenant_no=tenant_no,
        user=user,
        page=page,
        page_size=page_size,
        plan_no=_param_get(params, "planNo", "plan_no"),
        gap_type=_param_get(params, "gapType", "gapTypeIn", "gap_type"),
        customer_keyword=_param_get(params, "customerKeyword", "customer_keyword"),
        plan_date_from=_param_get(params, "planDateFrom", "plan_date_from"),
        plan_date_to=_param_get(params, "planDateTo", "plan_date_to"),
        delivery_time_from=_param_get(params, "deliveryTimeFrom", "delivery_time_from"),
        delivery_time_to=_param_get(params, "deliveryTimeTo", "delivery_time_to"),
        plan_type=_param_get(params, "planType", "plan_type"),
        plan_type_in=_param_get(params, "planTypeIn", "plan_type_in"),
        follow_status=_param_get(params, "followStatus", "followStatusIn", "follow_status"),
        q=q,
        sort_field=_param_get(params, "sortField", "sort_field"),
        sort_order=_param_get(params, "sortOrder", "sort_order"),
    )
    return PlanContractGapListResponse(total=total, list=items)



CONTRACT_GAP_EXPORT_COLUMNS = [
    ExportColumn(title="计划编号", field="planNo"),
    ExportColumn(title="计划日期", field="planDate"),
    ExportColumn(title="送货时间", field="deliveryTimes"),
    ExportColumn(title="单据类型", field="planType"),
    ExportColumn(title="跟进状态", field="followStatus"),
    ExportColumn(title="供方", field="supplierName"),
    ExportColumn(title="需方", field="customerName"),
    ExportColumn(title="缺口类型", field="gapType"),
]


class ContractGapExportBody(BaseModel):
    plan_no: Optional[str] = Field(default=None, alias="planNo")
    gap_type: Optional[str] = Field(default=None, alias="gapType")
    customer_keyword: Optional[str] = Field(default=None, alias="customerKeyword")
    plan_date_from: Optional[date] = Field(default=None, alias="planDateFrom")
    plan_date_to: Optional[date] = Field(default=None, alias="planDateTo")
    delivery_time_from: Optional[date] = Field(default=None, alias="deliveryTimeFrom")
    delivery_time_to: Optional[date] = Field(default=None, alias="deliveryTimeTo")
    q: Optional[str] = None
    ids: Optional[List[str]] = None
    columns: Optional[List[str]] = None
    model_config = {"populate_by_name": True, "extra": "allow"}


@router.post("/contract-gaps/export")
def export_contract_gaps(
    body: ContractGapExportBody,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
):
    try:
        tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
        params = _body_to_params(body)
        _, items = list_plan_contract_gaps(
            db,
            tenant_no=tenant_no,
            user=user,
            page=1,
            page_size=EXPORT_MAX_ROWS,
            plan_no=_param_get(params, "planNo", "plan_no"),
            gap_type=_param_get(params, "gapType", "gapTypeIn", "gap_type"),
            customer_keyword=_param_get(params, "customerKeyword", "customer_keyword"),
            plan_date_from=_param_get(params, "planDateFrom", "plan_date_from"),
            plan_date_to=_param_get(params, "planDateTo", "plan_date_to"),
            delivery_time_from=_param_get(
                params, "deliveryTimeFrom", "delivery_time_from"
            ),
            delivery_time_to=_param_get(
                params, "deliveryTimeTo", "delivery_time_to"
            ),
            plan_type=_param_get(params, "planType", "plan_type"),
            plan_type_in=_param_get(params, "planTypeIn", "plan_type_in"),
            follow_status=_param_get(
                params, "followStatus", "followStatusIn", "follow_status"
            ),
            q=_param_get(params, "q"),
            ids=body.ids,
            max_page_size=EXPORT_MAX_ROWS,
            sort_field=_param_get(params, "sortField", "sort_field"),
            sort_order=_param_get(params, "sortOrder", "sort_order"),
        )
        plan_type_labels = {"direct": "直销单", "inbound": "入库单", "outbound": "出库单"}
        gap_labels = {"supplier": "仅缺供方合同", "demander": "仅缺需方合同", "both": "双侧缺口"}
        rows = []
        for item in items:
            data = item.model_dump(by_alias=True)
            data["planType"] = plan_type_labels.get(str(data.get("planType") or ""), data.get("planType"))
            sides = set(data.get("gapSides") or [])
            if "supplier" in sides and "demander" in sides:
                data["gapType"] = "双侧缺口"
            elif "supplier" in sides:
                data["gapType"] = "仅缺供方合同"
            elif "demander" in sides:
                data["gapType"] = "仅缺需方合同"
            else:
                data["gapType"] = ""
            times = data.get("deliveryTimes") or []
            if isinstance(times, list):
                data["deliveryTimes"] = " / ".join(str(t) for t in times) if times else ""
            for k, v in list(data.items()):
                if hasattr(v, "isoformat"):
                    data[k] = str(v)[:19]
            rows.append(data)
        cols = filter_export_columns(CONTRACT_GAP_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=rows,
            filename_stem=f"计划合同完善_{date_str}",
        )
    except ExportValidationError as e:
        raise HTTPException(status_code=400, detail=str(e)) from e


@router.put("/{id}/stop-contracts", response_model=PlanContractGapRead)
def put_plan_stop_contracts(
    id: str,
    payload: PlanStopContractsUpdateBody,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> PlanContractGapRead:
    """保存某装货/送货段合同关联（合同补全）。"""
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    plan = _get_plan(db, tenant_no, id, user)
    result = update_plan_stop_contracts(
        db,
        tenant_no=tenant_no,
        plan=plan,
        stop_type=payload.stop_type,
        stop_id=payload.stop_id,
        contracts=payload.contracts,
        user_id=user.id,
        dept_id=user.dept_id,
    )
    plan.update_by = user.id
    db.commit()
    return result


@router.put("/{id}/contracts/{side}", response_model=PlanContractGapRead)
def put_plan_side_contracts(
    id: str,
    side: str,
    payload: PlanSideContractsUpdateBody,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> PlanContractGapRead:
    """单侧保存计划合同关联（合同补全）。"""
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    plan = _get_plan(db, tenant_no, id, user)
    result = update_plan_side_contracts(
        db,
        tenant_no=tenant_no,
        plan=plan,
        side=side,
        contracts=payload.contracts,
        user_id=user.id,
        dept_id=user.dept_id,
    )
    plan.update_by = user.id
    db.commit()
    return result


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


@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 or "").strip()

    plan = PlPlan(
        id=generate_uuid(),
        tenant_no=tenant_no,
        plan_no=plan_no,
        create_date=_resolve_create_date(payload.create_date),
        plan_date=payload.plan_date,
        belongs_company=payload.belongs_company,
        plan_type=plan_type,
        follow_status="pending_arrange",
        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_contact=payload.pickup_contact,
        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,
        delivery_contact=payload.delivery_contact,
        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:
    """创建运输计划（完整表单：供方/需方 + 产品库产品 + 可选预选车辆，不自动安排运输）"""
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    save_mode = (getattr(payload, "save_mode", None) or "formal").strip() or "formal"
    is_draft = save_mode == "draft"
    multi_stop = uses_multi_stop_payload(payload.load, payload.deliveries)
    qty_warnings: list[str] = []
    load_body = None
    delivery_bodies: list = []
    supplier_contracts: list[PlanContractLink] = []
    demander_contracts: list[PlanContractLink] = []

    if is_draft:
        validate_draft_min_fields(
            belongs_company=payload.belongs_company,
            create_date=payload.create_date,
        )
        if multi_stop and payload.load is not None:
            try:
                load_body, delivery_bodies = validate_plan_stops_write(
                    db,
                    tenant_no,
                    payload.load,
                    payload.deliveries,
                    payload.products,
                )
                qty_warnings = collect_stop_qty_warnings(
                    payload.products, load_body, delivery_bodies
                )
            except HTTPException:
                load_body, delivery_bodies = None, []
                multi_stop = False
    elif multi_stop:
        load_body, delivery_bodies = validate_plan_stops_write(
            db,
            tenant_no,
            payload.load,
            payload.deliveries,
            payload.products,
        )
        qty_warnings = collect_stop_qty_warnings(payload.products, load_body, delivery_bodies)
    else:
        validate_create_full(payload)
        supplier_contracts, demander_contracts = validate_plan_party_contracts(
            db,
            tenant_no,
            supplier_id=payload.supplier_id,
            supplier_name=payload.supplier_name,
            customer_id=payload.customer_id,
            customer_name=payload.customer_name,
            supplier_contracts=payload.supplier_contracts,
            demander_contracts=payload.demander_contracts,
            require_products=payload.products,
        )
        load_body = None
        delivery_bodies = []

    plan_no = _generate_plan_no(db)

    # 单据类型由前端显式选择；未选则存空，不再自动推断为直销单
    plan_type = (payload.plan_type or "").strip()
    has_vehicle = bool(payload.transport and payload.transport.vehicle_id)
    demander_name = (payload.customer_name or "").strip() or None

    plan = PlPlan(
        id=generate_uuid(),
        tenant_no=tenant_no,
        plan_no=plan_no,
        create_date=_resolve_create_date(payload.create_date),
        plan_date=payload.plan_date,
        belongs_company=payload.belongs_company,
        plan_type=plan_type,
        follow_status=resolve_follow_status_on_save(
            save_mode=save_mode, current=None, is_create=True
        ),
        transport_status="pending",
        archive_status="unarchived",
        complete_status="no",
        contract_id=None,
        contract_no=None,
        supplier_id=payload.supplier_id,
        supplier_name=payload.supplier_name,
        destination_id=payload.customer_id or payload.destination_id,
        destination_name=demander_name or 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,
        freight_remark=payload.freight_remark,
        pickup_address=payload.pickup_address,
        pickup_lng=payload.pickup_lng,
        pickup_lat=payload.pickup_lat,
        pickup_time=payload.pickup_time,
        pickup_contact=payload.pickup_contact,
        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,
        delivery_contact=payload.delivery_contact,
        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()

    if multi_stop and load_body is not None:
        sync_plan_legacy_from_stops(plan, load_body, delivery_bodies)

    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_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 multi_stop and load_body is not None:
        replace_plan_stops(
            db,
            tenant_no=tenant_no,
            plan_id=plan.id,
            load=load_body,
            deliveries=delivery_bodies,
            user_id=user.id,
            dept_id=user.dept_id,
            plan_products=plan_products,
        )
        safe_upsert_prefs_from_plan_stops(
            db,
            tenant_no=tenant_no,
            load=load_body,
            deliveries=delivery_bodies,
            user_id=user.id,
            dept_id=user.dept_id,
        )
    else:
        replace_plan_contracts(
            db,
            tenant_no=tenant_no,
            plan_id=plan.id,
            supplier_contracts=supplier_contracts,
            demander_contracts=demander_contracts,
            user_id=user.id,
            dept_id=user.dept_id,
        )

    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
    result.supplier_contracts = supplier_contracts
    result.demander_contracts = demander_contracts
    result.warnings = qty_warnings
    stops = load_plan_stops_grouped(db, tenant_no, [plan.id]).get(plan.id, (None, []))
    attach_plan_stops_to_item(result, plan, stops)
    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="合同不存在")

    assert_contract_selectable_for_plan(db, contract.id, tenant_no)

    # 生成计划编号
    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,
        create_date=_resolve_create_date(payload.create_date),
        plan_date=payload.plan_date,
        belongs_company=contract.belongs_company or "shanxia",
        plan_type=plan_type,
        follow_status="pending_arrange",
        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,
        # 客户名称严格跟合同「客户名称」；目的地名称与客户保持一致（不再取需方可章名 party_b）
        destination_name=(contract.customer_name or "").strip() or None,
        customer_id=contract.customer_id,
        customer_name=(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,
        )
        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,
            brand_mode=cp.brand_mode,
            brand_id=cp.brand_id,
            brand_name=cp.brand_name,
            model=cp.model,
            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()
    )

    hazard_map = fetch_hazard_flags_map(
        db,
        tenant_no,
        [product.product_id for product in products if product.product_id],
    )
    brand_qtys_by_pp = load_brand_qtys_by_plan_product_ids(
        db,
        tenant_no=tenant_no,
        plan_product_ids=[p.id for p in products],
    )

    enriched_products = []
    for product in products:
        is_toxic, is_explosive = resolve_hazard_flags_from_crm(
            db,
            tenant_no,
            product.product_id,
            hazard_map=hazard_map,
        )
        item = PlanProductRead.model_validate(product)
        raw_brand_qtys = brand_qtys_by_pp.get(product.id) or None
        brand_qtys = (
            [PlanArchiveBrandQtyItem.model_validate(x) for x in raw_brand_qtys]
            if raw_brand_qtys
            else None
        )
        item = item.model_copy(
            update={
                "is_toxic": is_toxic,
                "is_explosive": is_explosive,
                "brand_qtys": brand_qtys,
            }
        )
        enriched_products.append(item)

    result = PlanWithProducts.model_validate(plan)
    result.products = enriched_products
    supplier_contracts, demander_contracts = load_plan_contracts_grouped(
        db, tenant_no, [id]
    ).get(id, ([], []))
    result.supplier_contracts = supplier_contracts
    result.demander_contracts = demander_contracts
    stops = load_plan_stops_grouped(db, tenant_no, [id]).get(id, (None, []))
    attach_plan_stops_to_item(result, plan, stops)
    can_price = user_can_view_archive_prices(db, user)
    return mask_plan_with_products_prices(result, enabled=not can_price)


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


@router.put("/{id}", response_model=PlanWithProducts)
def update_plan(
    id: str,
    payload: PlanUpdate,
    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="已作废的计划不能修改")

    save_mode = (getattr(payload, "save_mode", None) or "formal").strip() or "formal"
    is_draft = save_mode == "draft"
    if is_draft:
        assert_can_save_as_draft(plan.follow_status)
        validate_draft_min_fields(
            belongs_company=(
                payload.belongs_company
                if payload.belongs_company is not None
                else plan.belongs_company
            ),
            create_date=(
                payload.create_date if payload.create_date is not None else plan.create_date
            ),
        )

    multi_stop = uses_multi_stop_payload(payload.load, payload.deliveries)
    qty_warnings: list[str] = []
    load_body: Optional[PlanLoadBody] = None
    delivery_bodies: list[PlanDeliveryBody] = []

    # 更新主表字段（跟进状态仅由业务动作 / saveMode 决议维护，忽略客户端 follow_status）
    update_data = payload.model_dump(
        exclude_unset=True,
        exclude={
            "products",
            "transport",
            "follow_status",
            "save_mode",
            "supplier_contracts",
            "demander_contracts",
            "load",
            "deliveries",
        },
    )
    for key, value in update_data.items():
        setattr(plan, key, value)

    # 新模型：单一合同字段不再作为写入源
    if "supplier_contracts" in payload.model_fields_set or "demander_contracts" in payload.model_fields_set:
        plan.contract_id = None
        plan.contract_no = None

    if payload.customer_id is not None or payload.customer_name is not None:
        plan.destination_id = plan.customer_id
        plan.destination_name = plan.customer_name

    supplier_id = plan.supplier_id
    supplier_name = plan.supplier_name
    customer_id = plan.customer_id
    customer_name = plan.customer_name

    if is_draft:
        supplier_contracts: list[PlanContractLink] = []
        demander_contracts: list[PlanContractLink] = []
        if multi_stop and payload.load is not None:
            try:
                load_body, delivery_bodies = validate_plan_stops_write(
                    db,
                    tenant_no,
                    payload.load,
                    payload.deliveries,
                    payload.products,
                )
                products_for_warn = payload.products if payload.products is not None else None
                qty_warnings = collect_stop_qty_warnings(
                    products_for_warn, load_body, delivery_bodies
                )
                sync_plan_legacy_from_stops(plan, load_body, delivery_bodies)
            except HTTPException:
                # 跳过段更新，保留库内原段
                load_body, delivery_bodies = None, []
                multi_stop = False
        if not multi_stop:
            supplier_contracts, demander_contracts = load_plan_contracts_grouped(
                db, tenant_no, [id]
            ).get(id, ([], []))
    elif multi_stop:
        load_body, delivery_bodies = validate_plan_stops_write(
            db,
            tenant_no,
            payload.load,
            payload.deliveries,
            payload.products,
        )
        products_for_warn = payload.products if payload.products is not None else None
        qty_warnings = collect_stop_qty_warnings(products_for_warn, load_body, delivery_bodies)
        sync_plan_legacy_from_stops(plan, load_body, delivery_bodies)
        supplier_contracts = []
        demander_contracts = []
    else:
        supplier_contracts_in = (
            payload.supplier_contracts
            if "supplier_contracts" in payload.model_fields_set
            else load_plan_contracts_grouped(db, tenant_no, [id]).get(id, ([], []))[0]
        )
        demander_contracts_in = (
            payload.demander_contracts
            if "demander_contracts" in payload.model_fields_set
            else load_plan_contracts_grouped(db, tenant_no, [id]).get(id, ([], []))[1]
        )
        # 编辑保存若带了产品或任一侧合同字段，按新规则完整校验供方/需方
        if (
            payload.products is not None
            or "supplier_contracts" in payload.model_fields_set
            or "demander_contracts" in payload.model_fields_set
            or payload.supplier_id is not None
            or payload.customer_id is not None
        ):
            supplier_contracts, demander_contracts = validate_plan_party_contracts(
                db,
                tenant_no,
                supplier_id=supplier_id,
                supplier_name=supplier_name,
                customer_id=customer_id,
                customer_name=customer_name,
                supplier_contracts=supplier_contracts_in,
                demander_contracts=demander_contracts_in,
                require_products=payload.products if payload.products is not None else None,
            )
            replace_plan_contracts(
                db,
                tenant_no=tenant_no,
                plan_id=id,
                supplier_contracts=supplier_contracts,
                demander_contracts=demander_contracts,
                user_id=user.id,
                dept_id=user.dept_id,
            )
        else:
            supplier_contracts, demander_contracts = load_plan_contracts_grouped(db, tenant_no, [id]).get(
                id, ([], [])
            )

    plan_type = payload.plan_type or plan.plan_type

    # 更新产品明细（按 id upsert，保持 pl_plan_product.id 稳定）
    if payload.products is not None:
        existing_all = (
            db.query(PlPlanProduct)
            .filter(
                PlPlanProduct.plan_id == id,
                PlPlanProduct.tenant_no == tenant_no,
            )
            .all()
        )
        existing_all_by_id = {row.id: row for row in existing_all}
        existing_products = [row for row in existing_all if row.del_flag == 0]
        old_hazard_product_ids = collect_hazard_product_ids(db, tenant_no, existing_products)
        cert_by_product_id = {
            row.product_id: (
                row.purchase_cert_url,
                row.transport_cert_url,
                row.legal_use_cert_url,
            )
            for row in existing_products
        }

        kept_ids: set[str] = set()
        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,
            )
            purchase_cert = prod.purchase_cert_url
            transport_cert = prod.transport_cert_url
            legal_use_cert = prod.legal_use_cert_url
            old_certs = cert_by_product_id.get(prod.product_id, (None, None, None))
            old_purchase, old_transport, old_legal_use = old_certs
            purchase_cert = purchase_cert or old_purchase
            transport_cert = transport_cert or old_transport
            legal_use_cert = legal_use_cert or old_legal_use
            prod_payload = prod.model_copy(
                update={
                    "is_toxic": is_toxic,
                    "is_explosive": is_explosive,
                    "purchase_cert_url": purchase_cert,
                    "transport_cert_url": transport_cert,
                    "legal_use_cert_url": legal_use_cert,
                }
            )
            row_id, existing = resolve_plan_product_id_for_upsert(
                plan_id=id,
                payload_id=prod.id,
                existing_by_id=existing_all_by_id,
            )
            if existing is not None:
                apply_plan_product_fields(
                    existing,
                    plan_type=plan_type,
                    prod=prod_payload,
                    order_num=idx,
                    user_id=user.id,
                )
                kept_ids.add(existing.id)
                plan_products.append(existing)
            else:
                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,
                    product_row_id=row_id,
                )
                db.add(pp)
                kept_ids.add(pp.id)
                plan_products.append(pp)

        for row in existing_products:
            if row.id not in kept_ids:
                row.del_flag = 1
                row.update_by = user.id

        new_hazard_product_ids = collect_hazard_product_ids(db, tenant_no, payload.products)
        if old_hazard_product_ids != new_hazard_product_ids:
            clear_hazard_cert_ready(plan)

        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,
            )

    if multi_stop and load_body is not None:
        replace_plan_stops(
            db,
            tenant_no=tenant_no,
            plan_id=id,
            load=load_body,
            deliveries=delivery_bodies,
            user_id=user.id,
            dept_id=user.dept_id,
            plan_products=plan_products,
        )
        safe_upsert_prefs_from_plan_stops(
            db,
            tenant_no=tenant_no,
            load=load_body,
            deliveries=delivery_bodies,
            user_id=user.id,
            dept_id=user.dept_id,
        )

    plan.update_by = user.id
    plan.follow_status = resolve_follow_status_on_save(
        save_mode=save_mode,
        current=plan.follow_status,
        is_create=False,
    )
    dibang_date_warnings = sync_linked_dibang_dates_for_plan(
        db,
        tenant_no=tenant_no,
        plan=plan,
        user_id=user.id,
        dept_id=user.dept_id or "",
    )
    dibang_upsert_warnings: list[str] = []
    if (
        plan.transport_status == "arranged"
        and plan.transport_id
        and multi_stop
        and load_body is not None
    ):
        transport_row = (
            db.query(PlTransport)
            .filter(
                PlTransport.id == plan.transport_id,
                PlTransport.tenant_no == tenant_no,
                PlTransport.del_flag == 0,
            )
            .first()
        )
        if transport_row:
            stop_options: list[StopDibangOption] = []
            load_row = (
                db.query(PlPlanLoad)
                .filter(
                    PlPlanLoad.plan_id == id,
                    PlPlanLoad.tenant_no == tenant_no,
                    PlPlanLoad.del_flag == 0,
                )
                .first()
            )
            if load_row:
                stop_options.append(
                    StopDibangOption(
                        stop_type="load",
                        stop_id=load_row.id,
                        generate_dibang=bool(load_row.generate_dibang),
                    )
                )
            delivery_rows = (
                db.query(PlPlanDelivery)
                .filter(
                    PlPlanDelivery.plan_id == id,
                    PlPlanDelivery.tenant_no == tenant_no,
                    PlPlanDelivery.del_flag == 0,
                )
                .order_by(PlPlanDelivery.seq.asc(), PlPlanDelivery.create_time.asc())
                .all()
            )
            for delivery_row in delivery_rows:
                stop_options.append(
                    StopDibangOption(
                        stop_type="delivery",
                        stop_id=delivery_row.id,
                        generate_dibang=bool(delivery_row.generate_dibang),
                    )
                )
            if stop_options:
                try:
                    _link_results, dibang_upsert_warnings = upsert_stop_dibang_links(
                        db,
                        tenant_no=tenant_no,
                        plan=plan,
                        transport=transport_row,
                        stop_options=stop_options,
                        user_id=user.id,
                        dept_id=user.dept_id or "",
                    )
                except ValueError as exc:
                    db.rollback()
                    raise HTTPException(status_code=400, detail=str(exc)) from exc
    db.commit()
    db.refresh(plan)
    result = PlanWithProducts.model_validate(plan)
    result.products = plan_products
    result.supplier_contracts = supplier_contracts
    result.demander_contracts = demander_contracts
    result.warnings = [*qty_warnings, *dibang_date_warnings, *dibang_upsert_warnings]
    stops = load_plan_stops_grouped(db, tenant_no, [id]).get(id, (None, []))
    attach_plan_stops_to_item(result, plan, stops)
    return result


@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


@router.patch("/{id}/hazard-cert-ready", response_model=PlanWithProducts)
def patch_plan_hazard_cert_ready(
    id: str,
    payload: HazardCertReadyBody,
    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)
    set_hazard_cert_ready(
        db,
        tenant_no=tenant_no,
        plan_id=id,
        ready=payload.ready,
        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


@router.patch("/{id}/erp-entered", response_model=PlanWithProducts)
def patch_plan_erp_entered(
    id: str,
    payload: ErpEnteredBody,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> PlanWithProducts:
    """标记或取消运输计划已录入 ERP。"""
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    set_plan_erp_entered(
        db,
        tenant_no=tenant_no,
        plan_id=id,
        entered=payload.entered,
        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
    result.document_attachments = load_plan_document_attachments_by_plan_ids(
        db,
        tenant_no=tenant_no,
        plan_ids=[id],
    ).get(id, [])
    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=[],
        requires_site_link=False,
    )
    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
    requires_site_link = plan_requires_site_link(db, tenant_no=tenant_no, plan=plan)
    site_records = build_site_records_for_arranged_plan(
        db,
        tenant_no=tenant_no,
        plan=plan,
        transport=transport,
    )

    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 "",
                stop_type=link.stop_type,
                stop_id=link.stop_id,
                business_type=(inventory.business_type if inventory else None),
                partner_company=(inventory.partner_company if inventory else None),
            )
        )

    io_order_brief = None
    active_io = get_active_io_order_for_plan(db, tenant_no, plan)
    if active_io:
        kind, header = active_io
        from app.services.pl_io_order_service import build_io_order_site_url

        token = (header.site_access_token or "").strip()
        io_order_brief = TransportIoOrderBrief(
            kind=kind,
            order_id=header.id,
            order_no=header.order_no,
            site_access_token=token or None,
            site_url=build_io_order_site_url(kind, token) if token else None,
            status=header.status,
        )

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


# =============================================================================
# 跟进状态：等通知 / 取消等通知
# =============================================================================


@router.post("/{id}/wait-notice", response_model=PlanRead)
def mark_plan_wait_notice(
    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.follow_status not in ("pending_arrange", "drafting"):
        raise HTTPException(status_code=400, detail="仅待安排或草稿中状态可标记为等通知")

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


@router.post("/{id}/cancel-wait-notice", response_model=PlanRead)
def cancel_plan_wait_notice(
    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 != "wait_notice":
        raise HTTPException(status_code=400, detail="仅等通知状态可取消等通知")
    if plan.archive_status == "archived":
        raise HTTPException(status_code=400, detail="已归档的计划不能取消等通知")

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


FORCE_FOLLOW_STATUS_PERM = "pl:plan:force-follow-status"
ARCHIVE_PRICE_EXPORT_FIELDS = frozenset(
    {"unitPrice", "amount", "freightPrice", "freightAmount"}
)
ALLOWED_FORCE_FOLLOW_STATUSES = frozenset(
    {"drafting", "wait_notice", "pending_arrange", "arranged", "completed", "cancelled"}
)


@router.post("/{id}/force-follow-status", response_model=PlanRead)
def force_plan_follow_status(
    id: str,
    payload: PlanForceFollowStatusBody,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> PlanRead:
    """强制修改跟进状态（仅改字段，不作废联动、不补运输安排）。需权限 pl:plan:force-follow-status。"""
    if user.role != SUPER_ADMIN_ROLE:
        codes = get_user_menu_codes(db, user.id)
        if FORCE_FOLLOW_STATUS_PERM not in codes:
            raise HTTPException(status_code=403, detail="无强制修改跟进状态权限")

    next_status = (payload.follow_status or "").strip()
    if next_status not in ALLOWED_FORCE_FOLLOW_STATUSES:
        raise HTTPException(
            status_code=400,
            detail="跟进状态仅支持：草稿中/等通知/待安排/已安排/已完成/已取消",
        )

    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    plan = _get_plan(db, tenant_no, id, user)

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


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


@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

    void_io_orders_for_plan(
        db, tenant_no=tenant_no, plan_id=plan.id, user_id=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-draft", response_model=PlanWithProducts)
def archive_plan_draft(
    id: str,
    payload: PlanArchiveBody = Body(...),
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> PlanWithProducts:
    """暂存归档填写内容：不改库存、不标已归档。"""
    if not user_can_archive_plan(db, user):
        raise HTTPException(status_code=403, detail="无运输计划归档操作权限")
    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="已归档的计划请使用调整保存，不能暂存")

    use_stop_archive = payload.load is not None and payload.deliveries is not None
    if use_stop_archive:
        apply_archive_stop_quantities(
            db,
            tenant_no=tenant_no,
            plan=plan,
            load_block=payload.load,
            delivery_blocks=payload.deliveries or [],
            user_id=user.id,
            strict=False,
        )
        sync_plan_products_from_stop_archive(
            db, tenant_no=tenant_no, plan_id=plan.id, user_id=user.id
        )
        if payload.products:
            apply_archive_brand_qtys_only(
                db,
                tenant_no=tenant_no,
                plan=plan,
                items=payload.products,
                user_id=user.id,
            )
    elif payload.products:
        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,
            strict=False,
        )

    if payload.feedbacks is not None and plan.transport_id:
        apply_archive_stop_feedback_attachments(
            db,
            tenant_no=tenant_no,
            plan=plan,
            feedbacks=payload.feedbacks,
            user_id=user.id,
            dept_id=user.dept_id or "",
        )

    plan.archive_status = "draft"
    plan.update_by = user.id
    record_plan_change(
        db,
        tenant_no=tenant_no,
        plan_id=plan.id,
        change_type="archive_draft",
        change_content="暂存归档内容",
        user_id=user.id,
        user_name=user.name or None,
        dept_id=user.dept_id,
    )
    db.commit()
    return get_plan(id, db, user)


@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:
    if not user_can_archive_plan(db, user):
        raise HTTPException(status_code=403, detail="无运输计划归档操作权限")
    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 not already_archived:
        products = (
            db.query(PlPlanProduct)
            .filter(
                PlPlanProduct.plan_id == plan.id,
                PlPlanProduct.tenant_no == tenant_no,
                PlPlanProduct.del_flag == 0,
            )
            .order_by(PlPlanProduct.order_num)
            .all()
        )
        validate_hazard_cert_ready_for_transport(db, tenant_no, plan, products)

    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)

    use_stop_archive = payload.load is not None and payload.deliveries is not None
    if use_stop_archive:
        if not payload.deliveries:
            raise HTTPException(status_code=400, detail="请填写全部送货段归档")
        apply_archive_stop_quantities(
            db,
            tenant_no=tenant_no,
            plan=plan,
            load_block=payload.load,
            delivery_blocks=payload.deliveries,
            user_id=user.id,
        )
        sync_plan_products_from_stop_archive(
            db, tenant_no=tenant_no, plan_id=plan.id, user_id=user.id
        )
        if payload.products:
            apply_archive_brand_qtys_only(
                db,
                tenant_no=tenant_no,
                plan=plan,
                items=payload.products,
                user_id=user.id,
            )
        apply_archive_inventory_for_plan_stops(
            db,
            tenant_no=tenant_no,
            plan=plan,
            user_id=user.id,
            dept_id=user.dept_id or "",
        )
    else:
        if not payload.products:
            raise HTTPException(status_code=400, detail="请提交产品归档明细")
        apply_archive_product_quantities(
            db, tenant_no=tenant_no, plan=plan, items=payload.products, user_id=user.id
        )
        apply_archive_inventory_for_plan(
            db,
            tenant_no=tenant_no,
            plan=plan,
            user_id=user.id,
            dept_id=user.dept_id or "",
        )
    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 payload.feedbacks is not None and plan.transport_id:
        apply_archive_stop_feedback_attachments(
            db,
            tenant_no=tenant_no,
            plan=plan,
            feedbacks=payload.feedbacks,
            user_id=user.id,
            dept_id=user.dept_id or "",
        )

    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.patch("/{id}/driver-feedback", response_model=PlanWithProducts)
def update_plan_driver_feedback(
    id: str,
    payload: PlanDriverFeedbackUpdateBody,
    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)
    apply_plan_driver_feedback(
        db,
        tenant_no=tenant_no,
        plan=plan,
        payload=payload,
        user_id=user.id,
    )
    record_plan_change(
        db,
        tenant_no=tenant_no,
        plan_id=plan.id,
        change_type="feedback_adjust",
        change_content="归档人员更新司机回传信息",
        user_id=user.id,
        user_name=user.name or None,
        dept_id=user.dept_id,
    )
    plan.update_by = user.id
    db.commit()
    return get_plan(id, db, user)


@router.patch("/{id}/site-feedback/{link_id}", response_model=PlanWithProducts)
def update_plan_site_feedback(
    id: str,
    link_id: str,
    payload: PlanSiteFeedbackUpdateBody,
    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)
    link = apply_plan_site_feedback(
        db,
        tenant_no=tenant_no,
        plan=plan,
        link_id=link_id,
        payload=payload,
        user_id=user.id,
    )
    product_label = (link.product or "现场").strip()
    record_plan_change(
        db,
        tenant_no=tenant_no,
        plan_id=plan.id,
        change_type="feedback_adjust",
        change_content=f"归档人员更新现场回传信息（{product_label}）",
        user_id=user.id,
        user_name=user.name or None,
        dept_id=user.dept_id,
    )
    plan.update_by = user.id
    db.commit()
    return get_plan(id, db, user)


# =============================================================================
# 入库单打印模板
# =============================================================================


@router.get("/{id}/inbound-slip-template")
def get_inbound_slip_template(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> str:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    plan = _get_plan(db, tenant_no, id, user)
    return plan.slip_template_content or ""


@router.put("/{id}/inbound-slip-template", status_code=status.HTTP_204_NO_CONTENT)
def save_inbound_slip_template(
    id: str,
    payload: PlanInboundSlipTemplateSave,
    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)
    if plan.plan_type != "inbound":
        raise HTTPException(status_code=400, detail="仅入库单计划支持保存入库单打印模板")
    if plan.transport_status != "arranged":
        raise HTTPException(status_code=400, detail="请先完成运输安排后再保存入库单打印模板")
    plan.slip_template_content = payload.slip_template_content or ""
    plan.update_by = user.id
    db.commit()


# =============================================================================
# 送(提)货单打印模板
# =============================================================================


@router.get("/{id}/print-delivery-slip", response_model=DeliverySlipPrintRead)
def get_print_delivery_slip(
    id: str,
    customerId: Optional[str] = Query(default=None, description="送货客户ID（多客户时必填）"),
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
):
    """按送货客户聚合送(提)货单打印数据；不传 customerId 时仅返回客户选项。"""
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    plan = _get_plan(db, tenant_no, id, user)
    plan_date = plan.plan_date.isoformat() if plan.plan_date else None
    try:
        return get_delivery_slip_print_context(
            db,
            tenant_no=tenant_no,
            plan_id=plan.id,
            plan_no=plan.plan_no or "",
            plan_date=plan_date,
            customer_id=customerId,
        )
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc


@router.get("/{id}/delivery-slip-template")
def get_delivery_slip_template(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> str:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    plan = _get_plan(db, tenant_no, id, user)
    return plan.delivery_slip_template_content or ""


@router.put("/{id}/delivery-slip-template", status_code=status.HTTP_204_NO_CONTENT)
def save_delivery_slip_template(
    id: str,
    payload: PlanDeliverySlipTemplateSave,
    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)
    if plan.plan_type == "inbound":
        raise HTTPException(status_code=400, detail="入库单计划请使用入库单打印模板")
    if plan.transport_status != "arranged":
        raise HTTPException(status_code=400, detail="请先完成运输安排后再保存送(提)货单打印模板")
    plan.delivery_slip_template_content = payload.delivery_slip_template_content or ""
    plan.update_by = user.id
    db.commit()


# =============================================================================
# 分装单打印模板
# =============================================================================


@router.get("/{id}/subpack-slip-template")
def get_subpack_slip_template(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> str:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    plan = _get_plan(db, tenant_no, id, user)
    return plan.subpack_slip_template_content or ""


@router.put("/{id}/subpack-slip-template", status_code=status.HTTP_204_NO_CONTENT)
def save_subpack_slip_template(
    id: str,
    payload: PlanSubpackSlipTemplateSave,
    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)
    if plan.plan_type == "inbound":
        raise HTTPException(status_code=400, detail="入库单计划请使用入库单打印模板")
    plan.subpack_slip_template_content = payload.subpack_slip_template_content or ""
    plan.update_by = user.id
    db.commit()


@router.post("/{id}/subpack-slip-no")
def ensure_subpack_slip_no(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> str:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    plan = _get_plan(db, tenant_no, id, user)
    if plan.plan_type == "inbound":
        raise HTTPException(status_code=400, detail="入库单计划不支持分装单号")
    slip_no = _ensure_subpack_slip_no(db, tenant_no, plan)
    plan.update_by = user.id
    db.commit()
    return slip_no


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


@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)

    if plan.follow_status not in ("pending_arrange", "wait_notice", "drafting"):
        raise HTTPException(status_code=400, detail="只能删除待安排、草稿中或等通知且未安排运输的计划")
    if plan.transport_status != "pending":
        raise HTTPException(status_code=400, detail="已安排运输的计划不能删除")
    if plan.archive_status == "archived":
        raise HTTPException(status_code=400, detail="已归档的计划不能删除")

    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