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

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

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

from app.core.config import settings
from app.services.pl_mobile_stop_meta import (
    validate_mobile_meta_request,
    fold_mobile_selection_into_web_fields,
    attach_web_selection_as_mobile_meta,
)
from app.services.pl_mobile_product_identity import validate_mobile_product_links
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.services.pl_plan_execution_lock import (
    apply_force_void_downstream,
    collect_force_void_impact,
    execution_lock_message,
    load_plan_execution_locks,
    plan_execution_lock_reason,
)
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,
    PlPlanContract,
    PlPlanDelivery,
    PlPlanLoad,
    PlPlanProduct,
    PlPlanTransport,
    PlTransport,
    PlTransportDbLink,
    PlVehicle,
)
from app.schemas.pl import PlanProductCreate
from app.schemas.pl.cancel_reason import CancelReasonBody, normalize_optional_reason
from app.schemas.pl.plan import (
    CreateFromContractRequest,
    DeliverySlipPrintRead,
    PlanArchiveBody,
    PlanArchiveBrandQtyItem,
    PlanContractGapListResponse,
    PlanContractGapRead,
    PlanContractLink,
    PlanCreate,
    PlanCreateFull,
    PlanDriverFeedbackUpdateBody,
    HazardCertReadyBody,
    ErpEnteredBody,
    PlanForceFollowStatusBody,
    PlanForceVoidPreview,
    PlanInboundSlipTemplateSave,
    PlanDeliverySlipTemplateSave,
    PlanSubpackSlipTemplateSave,
    PlanDeliveryBody,
    PlanLoadBody,
    PlanListQuery,
    PlanListResponse,
    PlanProductCertsUpdate,
    PlanProductRead,
    PlanRead,
    PlanSideContractsUpdateBody,
    PlanSiteFeedbackUpdateBody,
    PlanStopContractsUpdateBody,
    PlanUpdate,
    PlanWithProducts,
)
from app.services.pl.plan_column_facet_context import strip_plan_facet_self_params
from app.services.pl_plan_contract_gap import (
    apply_contract_gap_keyword_filter,
    apply_contract_gap_type_filter,
    list_plan_contract_gaps,
    plan_has_contract_gap_exists,
    stop_contract_gap,
    update_plan_side_contracts,
    update_plan_stop_contracts,
)
from app.services.pl_brand_archive import (
    expand_export_rows_for_plan_brand,
    load_brand_qtys_by_plan_product_ids,
    load_brand_qtys_by_stop_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,
    lock_archive_inventory,
)
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 (
    ARCHIVE_STATUS_LABELS,
    FOLLOW_STATUS_LABELS,
    PLAN_TYPE_LABELS,
    _format_plate_display,
    build_plan_archive_export_rows,
    export_plan_archive_xlsx_response,
    export_plan_list_xlsx_response,
    format_stop_contract_export_label,
)
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_calendar_customer_names import attach_calendar_routes
from app.services.pl_calendar_vehicle_types import attach_calendar_vehicle_types
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_rules import assert_load_delivery_same_products
from app.services.pl_plan_stop_service import (
    attach_plan_stops_to_item,
    collect_stop_qty_warnings,
    load_plan_stops_grouped,
    replace_plan_stops,
    sync_plan_legacy_from_stops,
    uses_multi_stop_payload,
    assert_external_stop_products_have_contract_line,
    validate_plan_stops_write,
    validate_mobile_draft_stops,
    has_pending_stop_products,
    validate_mobile_stop_sources,
)
from app.services.pl_customer_stop_pref_service import safe_upsert_prefs_from_plan_stops
from app.services.pl.plan_list_export_segments import (
    format_stop_segment_label as _format_stop_segment_label,
    iter_segment_product_slots as _iter_segment_product_slots,
)
from app.services.pl.plan_list_filters import (
    apply_stock_role_filter,
    build_plan_column_filters,
    parse_single_stop_kind,
    parse_stop_customer_keywords,
    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,
    is_belongs_company_contract_error,
    apply_plan_product_fields,
    build_plan_product,
    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_display(
    product: PlPlanProduct,
    *,
    standard_by_id: dict[str, str],
    alias_options_by_id: dict[str, list[str]],
) -> tuple[str, Optional[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 ""), alias


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_no_in=_param_get(params, "planNoIn", "plan_no_in"),
        belongs_company=_param_get(params, "belongsCompany", "belongs_company"),
        belongs_company_in=_param_get(
            params, "belongsCompanyIn", "belongs_company_in"
        ),
        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"),
        archive_list=_truthy_param(_param_get(params, "archiveList", "archive_list")),
        customer_name=_param_get(params, "customerName", "customer_name"),
        product_name=_param_get(params, "productName", "product_name"),
        product_name_in=_param_get(params, "productNameIn", "product_name_in"),
        package_spec=_param_get(params, "packageSpec", "package_spec"),
        package_spec_in=_param_get(params, "packageSpecIn", "package_spec_in"),
        license_plate=_param_get(params, "licensePlate", "license_plate"),
        license_plate_in=_param_get(params, "licensePlateIn", "license_plate_in"),
        stop_kind=_param_get(params, "stopKind", "stop_kind"),
        stop_kind_in=_param_get(params, "stopKindIn", "stop_kind_in"),
        stop_customers=_param_get(params, "stopCustomers", "stop_customers"),
        load_customers_in=_param_get(params, "loadCustomersIn", "load_customers_in"),
        delivery_customers_in=_param_get(params, "deliveryCustomersIn", "delivery_customers_in"),
        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"),
        erp_entered=_param_get(params, "erpEntered", "erp_entered"),
        erp_entered_in=_param_get(params, "erpEnteredIn", "erp_entered_in"),
        carrier_name=_param_get(params, "carrierName", "carrier_name"),
        carrier_name_in=_param_get(params, "carrierNameIn", "carrier_name_in"),
        personnel_summary=_param_get(params, "personnelSummary", "personnel_summary"),
        follow_user_name=_param_get(params, "followUserName", "follow_user_name"),
        stop_address=_param_get(params, "stopAddress", "stop_address"),
        contract_no=_param_get(params, "contractNo", "contract_no"),
        contract_no_in=_param_get(params, "contractNoIn", "contract_no_in"),
        product_spec=_param_get(params, "productSpec", "product_spec"),
        product_spec_in=_param_get(params, "productSpecIn", "product_spec_in"),
        product_unit=_param_get(params, "productUnit", "product_unit"),
        product_unit_in=_param_get(params, "productUnitIn", "product_unit_in"),
        product_plan_qty_op=_param_get(params, "productPlanQtyOp", "product_plan_qty_op"),
        product_plan_qty=_param_get(params, "productPlanQty", "product_plan_qty"),
        product_plan_qty_min=_param_get(
            params, "productPlanQtyMin", "product_plan_qty_min"
        ),
        product_plan_qty_max=_param_get(
            params, "productPlanQtyMax", "product_plan_qty_max"
        ),
        product_actual_qty_op=_param_get(
            params, "productActualQtyOp", "product_actual_qty_op"
        ),
        product_actual_qty=_param_get(params, "productActualQty", "product_actual_qty"),
        product_actual_qty_min=_param_get(
            params, "productActualQtyMin", "product_actual_qty_min"
        ),
        product_actual_qty_max=_param_get(
            params, "productActualQtyMax", "product_actual_qty_max"
        ),
        product_inbound_qty_op=_param_get(
            params, "productInboundQtyOp", "product_inbound_qty_op"
        ),
        product_inbound_qty=_param_get(
            params, "productInboundQty", "product_inbound_qty"
        ),
        product_inbound_qty_min=_param_get(
            params, "productInboundQtyMin", "product_inbound_qty_min"
        ),
        product_inbound_qty_max=_param_get(
            params, "productInboundQtyMax", "product_inbound_qty_max"
        ),
        product_outbound_qty_op=_param_get(
            params, "productOutboundQtyOp", "product_outbound_qty_op"
        ),
        product_outbound_qty=_param_get(
            params, "productOutboundQty", "product_outbound_qty"
        ),
        product_outbound_qty_min=_param_get(
            params, "productOutboundQtyMin", "product_outbound_qty_min"
        ),
        product_outbound_qty_max=_param_get(
            params, "productOutboundQtyMax", "product_outbound_qty_max"
        ),
        product_unit_price_op=_param_get(
            params, "productUnitPriceOp", "product_unit_price_op"
        ),
        product_unit_price=_param_get(params, "productUnitPrice", "product_unit_price"),
        product_unit_price_min=_param_get(
            params, "productUnitPriceMin", "product_unit_price_min"
        ),
        product_unit_price_max=_param_get(
            params, "productUnitPriceMax", "product_unit_price_max"
        ),
        product_amount_op=_param_get(params, "productAmountOp", "product_amount_op"),
        product_amount=_param_get(params, "productAmount", "product_amount"),
        product_amount_min=_param_get(params, "productAmountMin", "product_amount_min"),
        product_amount_max=_param_get(params, "productAmountMax", "product_amount_max"),
        stop_dibang=_param_get(params, "stopDibang", "stop_dibang"),
        stop_dibang_in=_param_get(params, "stopDibangIn", "stop_dibang_in"),
    )


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


class PlanColumnFacetBody(BaseModel):
    """运输计划列头可选值（去重）。额外字段与列表过滤同名，用于 Excel 级联上下文。"""

    model_config = ConfigDict(extra="allow")

    field: str = Field(
        ...,
        description="planNo | licensePlate | carrierName | loadCustomers | deliveryCustomers | contractNo | productName | productSpec | packageSpec | productUnit",
    )
    keyword: Optional[str] = Field(default=None, description="可选值关键字")
    limit: Optional[int] = Field(default=500, ge=1, le=500, description="返回条数上限")


def _truthy_param(value: Any) -> bool:
    if value is True:
        return True
    if value in (1, "1", "true", "True", "yes", "YES"):
        return True
    return False


def _build_plan_facet_base_query(
    db: Session,
    *,
    tenant_no: str,
    user: DbUserContext,
    params: dict,
):
    """与列表同源的计划上下文查询（已排除当前 facet 列自身条件）。"""
    query = db.query(PlPlan).filter(
        PlPlan.tenant_no == tenant_no,
        PlPlan.del_flag == 0,
    )
    contract_gap_only = _truthy_param(
        _param_get(params, "contractGapOnly", "contract_gap_only")
    )

    q_kw = (_param_get(params, "q") or "")
    if isinstance(q_kw, str):
        q_kw = q_kw.strip()
    else:
        q_kw = str(q_kw).strip() if q_kw else ""
    if contract_gap_only:
        query = apply_contract_gap_keyword_filter(query, tenant_no, q_kw)
    elif 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),
            )
        )

    follow_status = _param_get(params, "followStatus", "follow_status")
    query = _apply_follow_status_filter(query, follow_status)
    exclude_follow_status = _param_get(
        params, "excludeFollowStatus", "exclude_follow_status"
    )
    query = _apply_exclude_follow_status_filter(query, exclude_follow_status)

    has_hazard = _param_get(params, "hasHazardProduct", "has_hazard_product")
    if _truthy_param(has_hazard):
        query = query.filter(plan_has_hazard_product_exists(tenant_no))

    if contract_gap_only:
        query = query.filter(
            PlPlan.follow_status != "cancelled",
            plan_has_contract_gap_exists(tenant_no),
        )

    gap_type = _param_get(params, "gapType", "gap_type")
    if gap_type:
        query = apply_contract_gap_type_filter(query, tenant_no, str(gap_type))

    customer_keyword = _param_get(params, "customerKeyword", "customer_keyword")
    if contract_gap_only:
        query = apply_contract_gap_keyword_filter(
            query, tenant_no, str(customer_keyword or ""), include_plan_no=False
        )
    elif customer_keyword and str(customer_keyword).strip():
        kw = str(customer_keyword).strip()
        kw_lower = kw.lower()
        query = query.filter(
            or_(
                func.lower(PlPlan.supplier_name).contains(kw_lower),
                func.lower(PlPlan.customer_name).contains(kw_lower),
            )
        )

    column_filters = _filters_from_params(params, db=db, tenant_no=tenant_no)
    if column_filters:
        query = query.filter(*column_filters)

    stock_role = _param_get(params, "stockRole", "stock_role")
    try:
        query = apply_stock_role_filter(
            query, stock_role=stock_role, tenant_no=tenant_no
        )
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc

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


@router.post("/column-facet-values")
def plan_column_facet_values(
    body: PlanColumnFacetBody,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
):
    """运输计划列头可选值：在列表上下文（排除本列）下去重，对齐 Excel 级联。"""
    from sqlalchemy import func as sa_func

    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    lim = min(max(1, int(body.limit or 500)), 500)
    kw = (body.keyword or "").strip()
    field = str(body.field or "").strip()

    raw_params = _body_to_params(body)
    params = strip_plan_facet_self_params(raw_params, field)
    base = _build_plan_facet_base_query(
        db, tenant_no=tenant_no, user=user, params=params
    )
    plan_id_sq = base.with_entities(PlPlan.id).subquery()

    def _finish(rows):
        out = []
        for row in rows:
            value = str(row.value).strip() if row.value is not None else ""
            if not value:
                continue
            out.append({"value": value, "label": value, "count": int(row.cnt or 0)})
        return out

    def _plan_col_facet(col):
        q = (
            db.query(col.label("value"), sa_func.count(PlPlan.id).label("cnt"))
            .select_from(PlPlan)
            .filter(
                PlPlan.id.in_(plan_id_sq),
                col.isnot(None),
                col != "",
            )
        )
        if kw:
            q = q.filter(col.like(f"%{kw}%"))
        return _finish(q.group_by(col).order_by(col.asc()).limit(lim).all())

    def _sub_facet(model, col):
        q = (
            db.query(col.label("value"), sa_func.count(PlPlan.id).label("cnt"))
            .select_from(PlPlan)
            .join(model, model.plan_id == PlPlan.id)
            .filter(
                PlPlan.id.in_(plan_id_sq),
                model.tenant_no == tenant_no,
                model.del_flag == 0,
                col.isnot(None),
                col != "",
            )
        )
        if kw:
            q = q.filter(col.like(f"%{kw}%"))
        return _finish(q.group_by(col).order_by(col.asc()).limit(lim).all())

    if field == "planNo":
        return {"list": _plan_col_facet(PlPlan.plan_no)}
    if field == "licensePlate":
        return {"list": _plan_col_facet(PlPlan.license_plate)}
    if field in ("loadCustomers", "loadCustomer"):
        return {"list": _sub_facet(PlPlanLoad, PlPlanLoad.customer_name)}
    if field in ("deliveryCustomers", "deliveryCustomer"):
        return {"list": _sub_facet(PlPlanDelivery, PlPlanDelivery.customer_name)}
    if field == "productName":
        return {"list": _sub_facet(PlPlanProduct, PlPlanProduct.product_name)}
    if field == "productSpec":
        return {"list": _sub_facet(PlPlanProduct, PlPlanProduct.model)}
    if field in ("carrierName",):
        col = PlTransport.carrier_name
        q = (
            db.query(col.label("value"), sa_func.count(PlPlan.id).label("cnt"))
            .select_from(PlPlan)
            .join(PlTransport, PlTransport.id == PlPlan.transport_id)
            .filter(
                PlPlan.id.in_(plan_id_sq),
                PlTransport.tenant_no == tenant_no,
                PlTransport.del_flag == 0,
                col.isnot(None),
                col != "",
            )
        )
        if kw:
            q = q.filter(col.like(f"%{kw}%"))
        return {"list": _finish(q.group_by(col).order_by(col.asc()).limit(lim).all())}

    if field in ("packageSpec", "productPackageSpec"):
        return {"list": _sub_facet(PlPlanProduct, PlPlanProduct.package_spec)}
    if field == "productUnit":
        return {"list": _sub_facet(PlPlanProduct, PlPlanProduct.unit)}

    if field in ("contractNo", "productContract"):
        from app.services.pl.plan_list_filters import (
            CONTRACT_PLACEHOLDER_INBOUND,
            CONTRACT_PLACEHOLDER_OUTBOUND,
            plan_internal_contract_placeholder_exists,
        )

        link_q = (
            db.query(
                PlPlanContract.contract_no.label("value"),
                sa_func.count(PlPlan.id).label("cnt"),
            )
            .select_from(PlPlan)
            .join(PlPlanContract, PlPlanContract.plan_id == PlPlan.id)
            .filter(
                PlPlan.id.in_(plan_id_sq),
                PlPlanContract.tenant_no == tenant_no,
                PlPlanContract.del_flag == 0,
                PlPlanContract.contract_no.isnot(None),
                PlPlanContract.contract_no != "",
            )
        )
        header_q = (
            db.query(
                PlPlan.contract_no.label("value"),
                sa_func.count(PlPlan.id).label("cnt"),
            )
            .filter(
                PlPlan.id.in_(plan_id_sq),
                PlPlan.contract_no.isnot(None),
                PlPlan.contract_no != "",
            )
        )
        if kw:
            link_q = link_q.filter(PlPlanContract.contract_no.like(f"%{kw}%"))
            header_q = header_q.filter(PlPlan.contract_no.like(f"%{kw}%"))
        merged: dict = {}
        for row in link_q.group_by(PlPlanContract.contract_no).all():
            value = str(row.value).strip()
            if value:
                merged[value] = merged.get(value, 0) + int(row.cnt or 0)
        for row in header_q.group_by(PlPlan.contract_no).all():
            value = str(row.value).strip()
            if value:
                merged[value] = merged.get(value, 0) + int(row.cnt or 0)
        items = sorted(merged.items(), key=lambda kv: kv[0])[:lim]
        list_items = [{"value": v, "label": v, "count": c} for v, c in items]
        # 山峡/云栋缺合同占位：固定置于可选值前列，便于按入库/出库过滤
        placeholder_prefix = []
        for placeholder in (CONTRACT_PLACEHOLDER_OUTBOUND, CONTRACT_PLACEHOLDER_INBOUND):
            if kw and kw not in placeholder:
                continue
            base = db.query(PlPlan.id).filter(PlPlan.id.in_(plan_id_sq))
            cnt = (
                base.filter(
                    plan_internal_contract_placeholder_exists(tenant_no, placeholder)
                ).count()
            )
            placeholder_prefix.append(
                {"value": placeholder, "label": placeholder, "count": int(cnt or 0)}
            )
        return {"list": placeholder_prefix + list_items}

    raise HTTPException(status_code=400, detail="不支持的列")


@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）",
    ),
    include_customer_short_names: int = Query(
        default=0, alias="includeCustomerShortNames", description="1=返回日历客户简称路线和车辆归属"
    ),
    include_document_attachments: Optional[int] = Query(
        default=0,
        alias="includeDocumentAttachments",
        description="1=聚合单据附件（归档列表需要；默认关闭以降低列表耗时）",
    ),
    include_arrange_copy_flags: Optional[int] = Query(
        default=0,
        alias="includeArrangeCopyFlags",
        description="1=计算复制给司机/现场/磅单标志（运输计划列表需要；默认关闭）",
    ),
    stock_role: Optional[str] = Query(
        default=None,
        alias="stockRole",
        description="inbound=送货段山峡/云栋；outbound=装货段山峡/云栋",
    ),
    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)

    try:
        query = apply_stock_role_filter(
            query, stock_role=stock_role, tenant_no=tenant_no
        )
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc

    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)
    need_copy_flags = int(include_arrange_copy_flags or 0) == 1
    need_document_attachments = int(include_document_attachments or 0) == 1
    execution_locks = load_plan_execution_locks(db, tenant_no, plan_ids)
    arrange_copy_flags = (
        load_plan_arrange_copy_flags(db, tenant_no, plans) if need_copy_flags else {}
    )
    document_attachments_by_plan = (
        load_plan_document_attachments_by_plan_ids(
            db,
            tenant_no=tenant_no,
            plan_ids=plan_ids,
        )
        if need_document_attachments
        else {}
    )
    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
            )
            list_product_name, list_alias_name = _resolve_list_product_display(
                product,
                standard_by_id=standard_by_id,
                alias_options_by_id=alias_options_by_id,
            )
            item = item.model_copy(
                update={
                    "is_toxic": is_toxic,
                    "is_explosive": is_explosive,
                    "brand_qtys": brand_qtys,
                    "product_name": list_product_name,
                    "alias_name": list_alias_name,
                }
            )
            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
        plan_item.execution_lock_reason = execution_locks.get(plan.id)
        plan_item.execution_started = bool(plan_item.execution_lock_reason)
        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)

    if include_customer_short_names == 1:
        attach_calendar_routes(db, tenant_no=tenant_no, plans=result_list)
        attach_calendar_vehicle_types(db, tenant_no=tenant_no, plans=result_list)
    return PlanListResponse(total=total, list=result_list)



PLAN_EXPORT_COLUMNS = [
    ExportColumn(title="计划编号", field="planNo"),
    ExportColumn(title="段次", field="stopSegment"),
    ExportColumn(title="归属公司", field="belongsCompany"),
    ExportColumn(title="跟进状态", field="followStatus"),
    ExportColumn(title="计划日期", field="planDate"),
    ExportColumn(title="送货时间", field="deliveryTimes"),
    ExportColumn(title="单据类型", field="planType"),
    ExportColumn(title="车牌号", field="licensePlate"),
    ExportColumn(title="人员", field="personnelSummary"),
    ExportColumn(title="跟进人", field="followUserName"),
    ExportColumn(title="装货客户", field="loadCustomer"),
    ExportColumn(title="送货客户", field="deliveryCustomer"),
    ExportColumn(title="装货地址/时间", field="loadAddressTime"),
    ExportColumn(title="送货地址/时间", field="deliveryAddressTime"),
    ExportColumn(title="合同", field="productContract"),
    ExportColumn(title="产品名称", field="productName"),
    ExportColumn(title="品牌", field="brandName"),
    ExportColumn(title="别名", field="aliasName"),
    ExportColumn(title="型号/规格", field="model"),
    ExportColumn(title="包装规格", field="packageSpec"),
    ExportColumn(title="单位", field="unit", dict_type="crm_product_unit"),
    ExportColumn(title="计划量", field="planQty"),
    ExportColumn(title="实际量", field="actualQty"),
    ExportColumn(title="装货地磅", field="loadDibang"),
    ExportColumn(title="送货地磅", field="deliveryDibang"),
    ExportColumn(title="供方", field="supplierName"),
    ExportColumn(title="需方", field="customerName"),
    ExportColumn(title="司机", field="driverName"),
    ExportColumn(title="运输状态", field="transportStatus"),
    ExportColumn(title="归档状态", field="archiveStatus"),
    ExportColumn(title="ERP", field="erpEntered"),
    ExportColumn(title="备注", field="remark"),
]

# 列表可见列 prop → 导出 field（与归档导出字段对齐）
_PLAN_EXPORT_COLUMN_ALIASES = {
    "productBrand": "brandName",
    "productAlias": "aliasName",
    "productSpec": "model",
    "productPackageSpec": "packageSpec",
    "productUnit": "unit",
}

_PLAN_COMPANY_LABELS = {
    "shanxia": "安庆山峡",
    "yundong": "安徽云栋",
}

_TRANSPORT_STATUS_LABELS = {
    "draft": "草稿",
    "pending": "待执行",
    "executing": "执行中",
    "completed": "已完成",
    "cancelled": "已作废",
    "arranged": "已安排",
}


def _normalize_plan_export_columns(
    requested: Optional[Sequence[str]],
) -> Optional[list[str]]:
    if not requested:
        return None
    return [
        _PLAN_EXPORT_COLUMN_ALIASES.get(str(key), str(key)) for key in requested
    ]


def _plan_list_export_filename_stem(
    stock_role: Optional[str], date_str: str
) -> str:
    role = (stock_role or "").strip().lower()
    if role == "inbound":
        return f"入库单_{date_str}"
    if role == "outbound":
        return f"出库单_{date_str}"
    return f"计划列表_{date_str}"


def _plan_list_export_sheet_title(stock_role: Optional[str]) -> str:
    role = (stock_role or "").strip().lower()
    if role == "inbound":
        return "入库单"
    if role == "outbound":
        return "出库单"
    return "计划列表"


def _format_export_date_text(value: Any) -> str:
    if value is None:
        return ""
    text = str(value).strip()
    return text[:10] if text else ""


def _format_stop_address_time(address: Any, time_value: Any) -> str:
    addr = (str(address or "")).strip()
    time_text = _format_export_date_text(time_value)
    if addr and time_text:
        return f"{addr} / {time_text}"
    return addr or time_text


def _format_stop_dibang_label(stop: Any) -> str:
    if stop is None:
        return ""
    generate = int(getattr(stop, "generate_dibang", 0) or 0)
    if not generate:
        return ""
    direction = (getattr(stop, "dibang_direction", None) or "").strip()
    if direction == "outbound":
        return "出库"
    if direction == "inbound":
        return "入库"
    if direction == "none":
        return "无"
    return "是"


def _apply_plan_list_export_labels(data: dict) -> None:
    company = str(data.get("belongsCompany") or "")
    data["belongsCompany"] = _PLAN_COMPANY_LABELS.get(company, company)
    follow = str(data.get("followStatus") or "")
    data["followStatus"] = FOLLOW_STATUS_LABELS.get(follow, follow)
    plan_type = str(data.get("planType") or "")
    data["planType"] = PLAN_TYPE_LABELS.get(plan_type, plan_type)
    transport = str(data.get("transportStatus") or "")
    data["transportStatus"] = _TRANSPORT_STATUS_LABELS.get(transport, transport)
    archive = str(data.get("archiveStatus") or "")
    data["archiveStatus"] = ARCHIVE_STATUS_LABELS.get(archive, archive)


def _apply_plan_list_export_segment_stop_fields(
    data: dict,
    *,
    kind: str,
    load_read: Any = None,
    delivery_read: Any = None,
    delivery_seq: Optional[int] = None,
) -> None:
    """按当前导出行所属段填写装/送侧字段；对侧留空。"""
    data["_stopKind"] = kind
    data["_deliverySeq"] = delivery_seq
    data["stopSegment"] = _format_stop_segment_label(kind, delivery_seq)
    data["loadCustomer"] = ""
    data["deliveryCustomer"] = ""
    data["loadAddressTime"] = ""
    data["deliveryAddressTime"] = ""
    data["deliveryTimes"] = ""
    data["productContract"] = ""
    data["loadDibang"] = ""
    data["deliveryDibang"] = ""

    if kind == "load" and load_read is not None:
        data["loadCustomer"] = (getattr(load_read, "customer_name", None) or "").strip()
        data["loadAddressTime"] = _format_stop_address_time(
            getattr(load_read, "address", None),
            getattr(load_read, "load_time", None),
        )
        data["loadDibang"] = _format_stop_dibang_label(load_read)
        # 与列表/归档一致：有编号导出编号；山峡/云栋缺合同 → 出库
        data["productContract"] = format_stop_contract_export_label(
            "load",
            getattr(load_read, "contracts", None),
            data["loadCustomer"],
        )
        return

    if kind == "delivery" and delivery_read is not None:
        data["deliveryCustomer"] = (
            getattr(delivery_read, "customer_name", None) or ""
        ).strip()
        data["deliveryAddressTime"] = _format_stop_address_time(
            getattr(delivery_read, "address", None),
            getattr(delivery_read, "delivery_time", None),
        )
        data["deliveryTimes"] = _format_export_date_text(
            getattr(delivery_read, "delivery_time", None)
        )
        data["deliveryDibang"] = _format_stop_dibang_label(delivery_read)
        # 与列表/归档一致：有编号导出编号；山峡/云栋缺合同 → 入库
        data["productContract"] = format_stop_contract_export_label(
            "delivery",
            getattr(delivery_read, "contracts", None),
            data["deliveryCustomer"],
        )


def _blank_plan_list_export_stop_fields(data: dict) -> None:
    data["stopSegment"] = ""
    data["loadCustomer"] = ""
    data["deliveryCustomer"] = ""
    data["loadAddressTime"] = ""
    data["deliveryAddressTime"] = ""
    data["deliveryTimes"] = ""
    data["productContract"] = ""
    data["loadDibang"] = ""
    data["deliveryDibang"] = ""


def _blank_plan_list_export_product_fields(row: dict) -> None:
    row["productName"] = ""
    row["brandName"] = ""
    row["aliasName"] = ""
    row["model"] = ""
    row["packageSpec"] = ""
    row["unit"] = ""
    row["planQty"] = None
    row["actualQty"] = None


def _index_plan_products_for_list_export(
    plan_products: Sequence[Any],
) -> tuple[dict[str, Any], dict[str, Any]]:
    by_id: dict[str, Any] = {}
    by_product_id: dict[str, Any] = {}
    for product in plan_products:
        plan_product_id = str(getattr(product, "id", None) or "").strip()
        if plan_product_id:
            by_id[plan_product_id] = product
        product_id = str(getattr(product, "product_id", None) or "").strip()
        if product_id and product_id not in by_product_id:
            by_product_id[product_id] = product
    return by_id, by_product_id


def _lookup_plan_list_export_plan_product(
    stop_product: Any,
    plan_products_by_id: dict[str, Any],
    plan_products_by_product_id: dict[str, Any],
) -> Optional[Any]:
    plan_product_id = str(getattr(stop_product, "plan_product_id", None) or "").strip()
    if plan_product_id and plan_product_id in plan_products_by_id:
        return plan_products_by_id[plan_product_id]
    product_id = str(getattr(stop_product, "product_id", None) or "").strip()
    if product_id and product_id in plan_products_by_product_id:
        return plan_products_by_product_id[product_id]
    return None


def _fill_plan_list_export_legacy_product(
    row: dict,
    product: Any,
    *,
    standard_by_id: dict[str, str],
    alias_options_by_id: dict[str, list[str]],
) -> None:
    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,
    )
    row["productName"] = display_name or (product.product_name or "")
    row["brandName"] = ""
    row["aliasName"] = (alias or "").strip()
    row["model"] = (product.model or "").strip()
    row["packageSpec"] = (product.package_spec or "").strip()
    row["unit"] = (product.unit or "").strip()
    row["planQty"] = product.quantity
    row["actualQty"] = product.actual_qty


def _fill_plan_list_export_segment_product(
    row: dict,
    product: Any,
    kind: str,
    *,
    plan_products_by_id: dict[str, Any],
    plan_products_by_product_id: dict[str, Any],
    standard_by_id: dict[str, str],
    alias_options_by_id: dict[str, list[str]],
) -> Optional[Any]:
    """填充段产品字段；返回关联的计划产品（供品牌明细回退）。"""
    if product is None:
        _blank_plan_list_export_product_fields(row)
        return None
    plan_product = _lookup_plan_list_export_plan_product(
        product, plan_products_by_id, plan_products_by_product_id
    )
    saved_name = getattr(product, "product_name", None) or ""
    display_name, alias = resolve_product_display_names(
        saved_product_name=saved_name,
        product_id=getattr(product, "product_id", None),
        standard_by_id=standard_by_id,
        alias_options_by_id=alias_options_by_id,
    )
    row["productName"] = display_name or saved_name
    row["brandName"] = ""
    row["aliasName"] = (alias or "").strip()
    row["model"] = (getattr(product, "model", None) or "").strip()
    row["packageSpec"] = (getattr(product, "package_spec", None) or "").strip()
    row["unit"] = (getattr(product, "unit", None) or "").strip()
    row["planQty"] = getattr(product, "plan_qty", None)
    if kind == "load":
        row["actualQty"] = getattr(product, "inbound_qty", None)
    else:
        row["actualQty"] = getattr(product, "outbound_qty", None)
    return plan_product


def _build_plan_list_export_rows(
    db: Session,
    *,
    tenant_no: str,
    plans: list[PlPlan],
    stops_by_plan: dict,
) -> list[dict]:
    """计划列表导出：有段时按段次×产品展开；有归档品牌明细时再按品牌拆行。"""
    plan_ids = [plan.id for plan in plans]
    products_by_plan: dict[str, list[PlPlanProduct]] = defaultdict(list)
    product_ids: set[str] = set()
    plan_product_ids: list[str] = []
    stop_product_ids: list[str] = []
    if plan_ids:
        product_rows = (
            db.query(PlPlanProduct)
            .filter(
                PlPlanProduct.tenant_no == tenant_no,
                PlPlanProduct.plan_id.in_(plan_ids),
                PlPlanProduct.del_flag == 0,
            )
            .order_by(PlPlanProduct.order_num.asc(), PlPlanProduct.id.asc())
            .all()
        )
        for product in product_rows:
            products_by_plan[product.plan_id].append(product)
            if product.product_id:
                product_ids.add(product.product_id)
            if product.id:
                plan_product_ids.append(str(product.id))
    for plan in plans:
        load_read, delivery_reads = stops_by_plan.get(plan.id, (None, []))
        if load_read is None and not delivery_reads:
            continue
        for _kind, _load, _delivery, _seq, stop_product in _iter_segment_product_slots(
            load_read, delivery_reads
        ):
            if stop_product is None:
                continue
            stop_crm_id = str(getattr(stop_product, "product_id", None) or "").strip()
            if stop_crm_id:
                product_ids.add(stop_crm_id)
            stop_row_id = str(getattr(stop_product, "id", None) or "").strip()
            if stop_row_id:
                stop_product_ids.append(stop_row_id)
    standard_by_id, alias_options_by_id = load_crm_product_alias_maps(
        db, tenant_no=tenant_no, product_ids=product_ids
    )
    brand_by_plan_product = load_brand_qtys_by_plan_product_ids(
        db, tenant_no=tenant_no, plan_product_ids=plan_product_ids
    )
    brand_by_stop = load_brand_qtys_by_stop_product_ids(
        db, tenant_no=tenant_no, stop_product_ids=stop_product_ids
    )

    items: list[dict] = []
    for plan in plans:
        data = PlanWithProducts.model_validate(plan).model_dump(by_alias=True)
        data.pop("products", None)
        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 "否"
        data["licensePlate"] = _format_plate_display(
            data.get("licensePlate"), data.get("trailerNo")
        )
        _apply_plan_list_export_labels(data)
        load_read, delivery_reads = stops_by_plan.get(plan.id, (None, []))
        plan_products = products_by_plan.get(plan.id) or []

        if load_read is None and not delivery_reads:
            _blank_plan_list_export_stop_fields(data)
            if not plan_products:
                _blank_plan_list_export_product_fields(data)
                data["_stopKind"] = ""
                data["_deliverySeq"] = None
                items.append(data)
                continue
            for product in plan_products:
                row = dict(data)
                row["_stopKind"] = ""
                row["_deliverySeq"] = None
                _fill_plan_list_export_legacy_product(
                    row,
                    product,
                    standard_by_id=standard_by_id,
                    alias_options_by_id=alias_options_by_id,
                )
                items.extend(
                    expand_export_rows_for_plan_brand(
                        row,
                        archive_status=getattr(plan, "archive_status", None),
                        plan_product=product,
                        stop_product_id=None,
                        plan_product_id=str(getattr(product, "id", None) or ""),
                        by_stop_id=brand_by_stop,
                        by_plan_product_id=brand_by_plan_product,
                    )
                )
            continue

        by_id, by_product_id = _index_plan_products_for_list_export(plan_products)
        for kind, slot_load, slot_delivery, seq, stop_product in _iter_segment_product_slots(
            load_read, delivery_reads
        ):
            row = dict(data)
            _apply_plan_list_export_segment_stop_fields(
                row,
                kind=kind,
                load_read=slot_load,
                delivery_read=slot_delivery,
                delivery_seq=seq,
            )
            plan_product = _fill_plan_list_export_segment_product(
                row,
                stop_product,
                kind,
                plan_products_by_id=by_id,
                plan_products_by_product_id=by_product_id,
                standard_by_id=standard_by_id,
                alias_options_by_id=alias_options_by_id,
            )
            items.extend(
                expand_export_rows_for_plan_brand(
                    row,
                    archive_status=getattr(plan, "archive_status", None),
                    plan_product=plan_product,
                    stop_product_id=str(getattr(stop_product, "id", None) or "")
                    if stop_product is not None
                    else None,
                    plan_product_id=str(getattr(plan_product, "id", None) or "")
                    if plan_product is not None
                    else str(getattr(stop_product, "plan_product_id", None) or "")
                    if stop_product is not None
                    else None,
                    by_stop_id=brand_by_stop,
                    by_plan_product_id=brand_by_plan_product,
                )
            )
    return items


PLAN_ARCHIVE_EXPORT_COLUMNS = [
    ExportColumn(title="序号", field="rowNo"),
    ExportColumn(title="计划编号", field="planNo"),
    ExportColumn(title="段次", field="stopSegment"),
    ExportColumn(title="合同编号", field="contractNo"),
    ExportColumn(title="计划日期", field="planDate"),
    ExportColumn(title="归档状态", field="archiveStatus"),
    ExportColumn(title="跟进状态", field="followStatus"),
    ExportColumn(title="单据类型", field="planType"),
    ExportColumn(title="装货客户", field="loadCustomer"),
    ExportColumn(title="装货地址", field="loadAddress"),
    ExportColumn(title="装货时间", field="loadTime"),
    ExportColumn(title="送货客户", field="deliveryCustomer"),
    ExportColumn(title="送货地址", field="deliveryAddress"),
    ExportColumn(title="送货时间", field="deliveryTime"),
    ExportColumn(title="产品名称", field="productName"),
    ExportColumn(title="品牌", field="brandName"),
    ExportColumn(title="别名", field="aliasName"),
    ExportColumn(title="型号/规格", field="model"),
    ExportColumn(title="包装", field="packageSpec"),
    ExportColumn(title="单位", field="unit", dict_type="crm_product_unit"),
    ExportColumn(title="计划量", field="planQty"),
    ExportColumn(title="实际量", field="actualQty"),
    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")
    stock_role: Optional[str] = Field(default=None, alias="stockRole")
    ids: Optional[List[str]] = None
    columns: Optional[List[str]] = None
    export_format: Optional[str] = Field(
        default=None,
        alias="exportFormat",
        description="archive=运输计划归档 Excel；archive_zip=归档数据及附件 ZIP",
    )

    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)
        try:
            query = apply_stock_role_filter(
                query, stock_role=body.stock_role, tenant_no=tenant_no
            )
        except ValueError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        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")
        export_format = (body.export_format or "").strip().lower()
        is_archive_export = export_format == "archive"
        is_archive_zip = export_format == "archive_zip"
        if len(plans) > EXPORT_MAX_ROWS:
            raise ExportValidationError(f"导出超过 {EXPORT_MAX_ROWS} 条，请缩小筛选范围")
        if is_archive_export or is_archive_zip:
            if not plans:
                raise ExportValidationError("没有可导出的数据")
            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,
                stop_kind=parse_single_stop_kind(
                    _param_get(export_params, "stopKind", "stop_kind"),
                    _param_get(export_params, "stopKindIn", "stop_kind_in"),
                ),
                stop_customer_keywords=parse_stop_customer_keywords(
                    _param_get(export_params, "stopCustomers", "stop_customers")
                ),
            )
            cols = filter_export_columns(
                PLAN_ARCHIVE_EXPORT_COLUMNS, _normalize_plan_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]
            if is_archive_zip:
                from app.services.pl_plan_archive_zip_export import (
                    build_archive_zip_streaming_response,
                    collect_hazard_cert_items,
                    load_plan_products_by_plan_ids,
                    merge_plan_attachment_items,
                )

                plan_ids = [p.id for p in plans]
                docs = load_plan_document_attachments_by_plan_ids(
                    db, tenant_no=tenant_no, plan_ids=plan_ids
                )
                products_by_plan = load_plan_products_by_plan_ids(
                    db, tenant_no=tenant_no, plan_ids=plan_ids
                )
                attachments_by_plan = {
                    pid: merge_plan_attachment_items(
                        docs.get(pid, []),
                        collect_hazard_cert_items(products_by_plan.get(pid, [])),
                    )
                    for pid in plan_ids
                }
                stamp = datetime.now().strftime("%Y%m%d%H%M")
                return build_archive_zip_streaming_response(
                    db=db,
                    tenant_no=tenant_no,
                    columns=cols,
                    rows=items,
                    attachments_by_plan=attachments_by_plan,
                    stamp=stamp,
                )
            filename_stem = f"计划归档_{date_str}"
            return export_plan_archive_xlsx_response(
                db,
                tenant_no=tenant_no,
                columns=cols,
                rows=items,
                filename_stem=filename_stem,
            )
        else:
            plan_ids = [plan.id for plan in plans]
            stops_by_plan = load_plan_stops_grouped(db, tenant_no, plan_ids)
            items = _build_plan_list_export_rows(
                db,
                tenant_no=tenant_no,
                plans=plans,
                stops_by_plan=stops_by_plan,
            )
            cols = filter_export_columns(
                PLAN_EXPORT_COLUMNS, _normalize_plan_export_columns(body.columns)
            )
            sheet_title = _plan_list_export_sheet_title(body.stock_role)
            filename_stem = _plan_list_export_filename_stem(body.stock_role, date_str)
            if not (body.stock_role or "").strip():
                return export_plan_list_xlsx_response(
                    db,
                    tenant_no=tenant_no,
                    sheet_title=sheet_title,
                    columns=cols,
                    rows=items,
                    filename_stem=filename_stem,
                )
        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(default="", alias="productId"),
    package_spec: Optional[str] = Query(default=None, alias="packageSpec"),
    exclude_plan_id: Optional[str] = Query(default=None, alias="excludePlanId"),
    contract_product_id: Optional[str] = Query(default=None, alias="contractProductId"),
    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,
        contract_product_id=contract_product_id,
    )


@router.post("/contract-product-arrange-qty/batch")
def batch_contract_product_arrange_qty(
    body: dict,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> dict:
    """
    批量可安排数量摘要（不含已安排计划明细）。
    仅供运输计划新增/编辑/调整运输表单打开时预取，问号弹窗仍走单条详情接口。
    """
    from app.services.pl_contract_plan_qty import batch_contract_product_arrange_qty_summaries

    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    items = body.get("items") if isinstance(body, dict) else None
    if not isinstance(items, list):
        items = []
    rows = batch_contract_product_arrange_qty_summaries(
        db, tenant_no=tenant_no, items=items
    )
    return {"items": rows}


@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,
    )
    column_filters = _filters_from_params(params, db=db, tenant_no=tenant_no)
    total, items = list_plan_contract_gaps(
        db,
        tenant_no=tenant_no,
        user=user,
        page=page,
        page_size=page_size,
        gap_type=_param_get(params, "gapType", "gapTypeIn", "gap_type"),
        customer_keyword=_param_get(params, "customerKeyword", "customer_keyword"),
        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"),
        column_filters=column_filters,
    )
    return PlanContractGapListResponse(total=total, list=items)



CONTRACT_GAP_EXPORT_COLUMNS = [
    ExportColumn(title="计划编号", field="planNo"),
    ExportColumn(title="段次", field="stopSegment"),
    ExportColumn(title="计划日期", field="planDate"),
    ExportColumn(title="送货时间", field="deliveryTimes"),
    ExportColumn(title="单据类型", field="planType"),
    ExportColumn(title="跟进状态", field="followStatus"),
    ExportColumn(title="装货客户", field="loadCustomer"),
    ExportColumn(title="送货客户", field="deliveryCustomer"),
    ExportColumn(title="产品名称", field="productName"),
    ExportColumn(title="供方", field="supplierName"),
    ExportColumn(title="需方", field="customerName"),
    ExportColumn(title="缺口类型", field="gapType"),
]


def _contract_gap_export_base(item: Any) -> dict:
    """计划级字段；单据类型、跟进状态沿用列表导出的中文标签。"""
    if hasattr(item, "model_dump"):
        data = item.model_dump(by_alias=True)
    else:
        data = {
            "planNo": getattr(item, "plan_no", None) or getattr(item, "planNo", "") or "",
            "planDate": getattr(item, "plan_date", None)
            if hasattr(item, "plan_date")
            else getattr(item, "planDate", None),
            "planType": getattr(item, "plan_type", None)
            if hasattr(item, "plan_type")
            else getattr(item, "planType", None),
            "followStatus": getattr(item, "follow_status", None)
            if hasattr(item, "follow_status")
            else getattr(item, "followStatus", None),
            "supplierName": getattr(item, "supplier_name", None)
            if hasattr(item, "supplier_name")
            else getattr(item, "supplierName", None),
            "customerName": getattr(item, "customer_name", None)
            if hasattr(item, "customer_name")
            else getattr(item, "customerName", None),
        }
    plan_type = str(data.get("planType") or "")
    follow = str(data.get("followStatus") or "")
    return {
        "planNo": data.get("planNo") or "",
        "planDate": _format_export_date_text(data.get("planDate")),
        "planType": PLAN_TYPE_LABELS.get(plan_type, plan_type),
        "followStatus": FOLLOW_STATUS_LABELS.get(follow, follow),
        "supplierName": data.get("supplierName") or "",
        "customerName": data.get("customerName") or "",
    }


def _contract_gap_row_gap_type(kind: str, stop: Any) -> str:
    """本行段缺口：装货缺供方合同，送货缺需方合同；无缺口留空。"""
    if stop is None:
        return ""
    customer_name = getattr(stop, "customer_name", None)
    contracts = getattr(stop, "contracts", None) or []
    if not stop_contract_gap(customer_name, len(contracts)):
        return ""
    if kind == "load":
        return "缺供方合同"
    return "缺需方合同"


def build_contract_gap_export_rows(
    gap_items: Sequence[Any],
    stops_by_plan: dict,
) -> list[dict]:
    """把有缺口的计划按段次×产品展开；无缺口侧仍导出，gapType 为空。"""
    rows: list[dict] = []
    for item in gap_items:
        plan_id = str(getattr(item, "id", "") or "")
        load_read, delivery_reads = (stops_by_plan or {}).get(plan_id, (None, []))
        base = _contract_gap_export_base(item)
        for kind, slot_load, slot_delivery, seq, product in _iter_segment_product_slots(
            load_read, delivery_reads
        ):
            row = {
                "planNo": base["planNo"],
                "stopSegment": _format_stop_segment_label(kind, seq),
                "planDate": base["planDate"],
                "deliveryTimes": "",
                "planType": base["planType"],
                "followStatus": base["followStatus"],
                "loadCustomer": "",
                "deliveryCustomer": "",
                "productName": (
                    (getattr(product, "product_name", None) or "").strip()
                    if product is not None
                    else ""
                ),
                "supplierName": base["supplierName"],
                "customerName": base["customerName"],
                "gapType": "",
            }
            if kind == "load":
                row["loadCustomer"] = (
                    getattr(slot_load, "customer_name", None) or ""
                ).strip()
                row["gapType"] = _contract_gap_row_gap_type("load", slot_load)
            else:
                row["deliveryCustomer"] = (
                    getattr(slot_delivery, "customer_name", None) or ""
                ).strip()
                row["deliveryTimes"] = _format_export_date_text(
                    getattr(slot_delivery, "delivery_time", None)
                    if slot_delivery is not None
                    else None
                )
                row["gapType"] = _contract_gap_row_gap_type("delivery", slot_delivery)
            rows.append(row)
    return rows


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)
        column_filters = _filters_from_params(params, db=db, tenant_no=tenant_no)
        _, items = list_plan_contract_gaps(
            db,
            tenant_no=tenant_no,
            user=user,
            page=1,
            page_size=EXPORT_MAX_ROWS,
            gap_type=_param_get(params, "gapType", "gapTypeIn", "gap_type"),
            customer_keyword=_param_get(params, "customerKeyword", "customer_keyword"),
            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"),
            column_filters=column_filters,
        )
        plan_ids = [item.id for item in items]
        stops_by_plan = load_plan_stops_grouped(db, tenant_no, plan_ids)
        rows = build_contract_gap_export_rows(items, stops_by_plan)
        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())
    from app.services.pl_plan_stop_service import assert_plan_create_child_ids
    assert_plan_create_child_ids(payload)

    # 生成计划编号
    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())
    from app.services.pl_plan_stop_service import assert_plan_create_child_ids
    assert_plan_create_child_ids(payload)
    fold_mobile_selection_into_web_fields(payload)
    validate_mobile_meta_request(payload, settings.PL_MOBILE_PLAN_PRODUCT_META_ENABLED)
    validate_mobile_product_links(
        payload.products,
        [p for stop in ([payload.load] if payload.load else []) + list(payload.deliveries or []) for p in stop.products],
        creating=True,
        allow_unlinked_draft=bool(payload.mobile_product_meta and getattr(payload, 'save_mode', '') == 'draft'),
    )
    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 multi_stop and payload.load is not None:
        assert_load_delivery_same_products(payload.load, payload.deliveries)

    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 and payload.mobile_product_meta:
            load_body, delivery_bodies = validate_mobile_draft_stops(
                db, tenant_no, payload.load, payload.deliveries, payload.products,
                payload.belongs_company)
            qty_warnings = collect_stop_qty_warnings(payload.products, load_body, delivery_bodies)
        elif 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,
                    payload.belongs_company,
                )
                qty_warnings = collect_stop_qty_warnings(
                    payload.products, load_body, delivery_bodies
                )
            except HTTPException as exc:
                if is_belongs_company_contract_error(exc):
                    raise
                load_body, delivery_bodies = None, []
                multi_stop = False
    elif multi_stop:
        if payload.mobile_product_meta:
            validate_mobile_stop_sources(
                db, tenant_no, payload.load, payload.deliveries or [], payload.products,
                payload.belongs_company,
            )
        load_body, delivery_bodies = validate_plan_stops_write(
            db,
            tenant_no,
            payload.load,
            payload.deliveries,
            payload.products,
            payload.belongs_company,
        )
        assert_external_stop_products_have_contract_line(load_body, delivery_bodies)
        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,
            allow_unlinked_draft=bool(payload.mobile_product_meta and is_draft),
            preserve_pending_draft=is_draft,
        )
        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("/mobile-product-capability")
def get_mobile_product_capability(
    user: DbUserContext = Depends(get_db_current_user),
):
    return {"mobileProductMetaVersion": 1}


@router.get("/{id}", response_model=PlanWithProducts)
def get_plan(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
    include_mobile_product_meta: Annotated[bool, Query(alias="includeMobileProductMeta")] = False,
) -> 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
    result.execution_lock_reason = plan_execution_lock_reason(db, tenant_no, id)
    result.execution_started = bool(result.execution_lock_reason)
    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)
    if include_mobile_product_meta:
        result.mobile_product_meta_version = 1
        standard_by_id, alias_options_by_id = load_crm_product_alias_maps(
            db,
            tenant_no=tenant_no,
            product_ids=[product.product_id for product in result.products or [] if product.product_id],
        )
        attach_web_selection_as_mobile_meta(result, standard_by_id, alias_options_by_id)
        if result.follow_status == "drafting":
            for stop in ([result.load] if result.load else []) + list(result.deliveries or []):
                for product in stop.products:
                    product.unlinked_draft = not bool(product.plan_product_id)
    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="已作废的计划不能修改")

    fold_mobile_selection_into_web_fields(payload)
    validate_mobile_meta_request(payload, settings.PL_MOBILE_PLAN_PRODUCT_META_ENABLED)
    validate_mobile_product_links(
        payload.products,
        [p for stop in ([payload.load] if payload.load else []) + list(payload.deliveries or []) for p in stop.products],
        allow_unlinked_draft=bool(payload.mobile_product_meta and getattr(payload, 'save_mode', '') == 'draft'),
    )
    from app.services.pl_plan_stop_service import (
        preflight_plan_child_update, complete_plan_stops_for_validation,
    )
    prepared_product_ids, prepared_stop_ids = preflight_plan_child_update(db, tenant_no, id, payload)
    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)
    validation_load, validation_deliveries = (payload.load, payload.deliveries)
    if multi_stop:
        validation_load, validation_deliveries = complete_plan_stops_for_validation(
            db, tenant_no, id, payload.load, payload.deliveries)
        if validation_load is not None:
            assert_load_delivery_same_products(validation_load, validation_deliveries)

    qty_warnings: list[str] = []
    load_body: Optional[PlanLoadBody] = None
    delivery_bodies: list[PlanDeliveryBody] = []

    if not is_draft and has_pending_stop_products(db, tenant_no, id):
        if payload.load is None or any(
            not p.plan_product_id and not getattr(p, 'plan_product_client_key', None)
            for stop in [payload.load] + list(payload.deliveries or []) for p in stop.products
        ):
            raise HTTPException(422, '草稿含待关联产品，请补齐关联后正式保存')

    # 更新主表字段（跟进状态仅由业务动作 / saveMode 决议维护，忽略客户端 follow_status）
    update_data = payload.model_dump(
        exclude_unset=True,
        exclude={
            "mobile_product_meta",
            "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 validation_load is not None and payload.mobile_product_meta:
            load_body, delivery_bodies = validate_mobile_draft_stops(
                db, tenant_no, validation_load, validation_deliveries, payload.products,
                payload.belongs_company if payload.belongs_company is not None else plan.belongs_company)
            qty_warnings = collect_stop_qty_warnings(payload.products, load_body, delivery_bodies)
            sync_plan_legacy_from_stops(plan, load_body, delivery_bodies)
        elif multi_stop and validation_load is not None:
            try:
                load_body, delivery_bodies = validate_plan_stops_write(
                    db,
                    tenant_no,
                    validation_load,
                    validation_deliveries,
                    payload.products,
                    payload.belongs_company if payload.belongs_company is not None else plan.belongs_company,
                )
                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 as exc:
                if is_belongs_company_contract_error(exc):
                    raise
                # 跳过段更新，保留库内原段
                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:
        if payload.mobile_product_meta:
            validate_mobile_stop_sources(
                db, tenant_no, validation_load, validation_deliveries or [], payload.products,
                payload.belongs_company if payload.belongs_company is not None else plan.belongs_company,
            )
        load_body, delivery_bodies = validate_plan_stops_write(
            db,
            tenant_no,
            validation_load,
            validation_deliveries,
            payload.products,
            payload.belongs_company if payload.belongs_company is not None else plan.belongs_company,
        )
        assert_external_stop_products_have_contract_line(load_body, delivery_bodies)
        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 if row.del_flag == 0}
        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 = prepared_product_ids[idx]
            existing = existing_all_by_id.get(row_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(plan_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:
        # Validation uses a complete read-only view; persistence receives only submitted collections.
        if payload.load is not None and payload.load.contracts is not None:
            payload.load.contracts = load_body.contracts
        for partial, validated in zip(payload.deliveries or [], delivery_bodies):
            if partial.contracts is not None:
                partial.contracts = validated.contracts
        replace_plan_stops(
            db,
            tenant_no=tenant_no,
            plan_id=id,
            load=payload.load,
            deliveries=payload.deliveries,
            user_id=user.id,
            dept_id=user.dept_id,
            plan_products=plan_products,
            allow_unlinked_draft=bool(payload.mobile_product_meta and is_draft),
            preserve_pending_draft=is_draft,
            resolved_stop_ids=prepared_stop_ids,
        )
        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"
FORCE_VOID_PERM = "pl:plan:force-void"
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,
    body: Optional[CancelReasonBody] = Body(default=None),
    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="已归档的计划不能作废")

    lock_reason = plan_execution_lock_reason(db, tenant_no, plan.id)
    if lock_reason:
        raise HTTPException(
            status_code=400, detail=execution_lock_message(lock_reason, "void")
        )

    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
    )

    now = datetime.now()
    reason = normalize_optional_reason(body.cancel_reason if body else None)
    plan.follow_status = "cancelled"
    plan.archive_status = "cancelled"
    plan.transport_status = "pending"
    plan.cancel_reason = reason
    plan.cancel_time = now
    plan.cancel_by = user.id
    plan.update_by = user.id
    db.commit()
    db.refresh(plan)
    return plan


def _assert_can_force_void(db: Session, user: DbUserContext) -> None:
    if user.role == SUPER_ADMIN_ROLE:
        return
    if FORCE_VOID_PERM not in get_user_menu_codes(db, user.id):
        raise HTTPException(status_code=403, detail="无强制作废权限")


def _load_force_void_plan(
    db: Session, user: DbUserContext, plan_id: str
):
    _assert_can_force_void(db, user)
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    plan = _get_plan(db, tenant_no, plan_id, user)
    if plan.follow_status == "cancelled":
        raise HTTPException(status_code=400, detail="计划已作废")
    if plan.archive_status == "archived":
        raise HTTPException(status_code=400, detail="已归档的计划不能强制作废")
    lock_reason = plan_execution_lock_reason(db, tenant_no, plan.id)
    if not lock_reason:
        raise HTTPException(status_code=400, detail="该计划尚未开始，请使用作废")
    return tenant_no, plan, lock_reason


@router.get("/{id}/force-void-preview", response_model=PlanForceVoidPreview)
def preview_force_void_plan(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> PlanForceVoidPreview:
    """强制作废前的冲销清单。仅超级管理员，或已被授予 pl:plan:force-void 的用户。"""
    tenant_no, _plan, lock_reason = _load_force_void_plan(db, user, id)
    impact = collect_force_void_impact(db, tenant_no=tenant_no, plan_id=id)
    return PlanForceVoidPreview(
        execution_lock_reason=lock_reason,
        weighing_slips=impact.weighing_slips,
        inventories=impact.inventories,
        io_orders=impact.io_orders,
        notice=impact.notice(),
    )


@router.post("/{id}/force-void", response_model=PlanRead)
def force_void_plan(
    id: str,
    body: Optional[CancelReasonBody] = Body(default=None),
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> PlanRead:
    """强制作废已开始的运输计划：作废地磅单、退回数量、作废出入库单。"""
    tenant_no, plan, _lock_reason = _load_force_void_plan(db, user, id)
    reason = normalize_optional_reason(body.cancel_reason if body else None)
    if not reason:
        raise HTTPException(status_code=400, detail="请填写作废原因")

    apply_force_void_downstream(
        db,
        tenant_no=tenant_no,
        plan_id=plan.id,
        transport_id=plan.transport_id,
        user_id=user.id,
        reason=reason,
    )
    now = datetime.now()
    plan.follow_status = "cancelled"
    plan.archive_status = "cancelled"
    plan.transport_status = "pending"
    plan.cancel_reason = reason
    plan.cancel_time = now
    plan.cancel_by = user.id
    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)
    lock_archive_inventory(db, tenant_no=tenant_no, plan=plan, refresh_plan=True)

    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="请填写全部送货段归档")
        from app.services.pl_contract_plan_qty import (
            ARCHIVE_LINE_QTY_OVERAGE_CODE,
            collect_archive_line_overages,
        )

        overages = collect_archive_line_overages(
            db,
            tenant_no=tenant_no,
            plan_id=plan.id,
            load_block=payload.load,
            delivery_blocks=payload.deliveries,
        )
        if overages and not payload.confirm_overage:
            raise HTTPException(
                status_code=422,
                detail={
                    "code": ARCHIVE_LINE_QTY_OVERAGE_CODE,
                    "message": "归档实际数量超过合同行剩余，请确认超交",
                    "items": overages,
                },
            )
        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