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

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

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

from app.core.config import settings
from app.core.database import get_db
from app.core.db_auth import DbUserContext, get_db_current_user
from app.core.delete_guard import assert_row_deletable_by_creator, user_info_from_context
from app.core.menu_guard import SUPER_ADMIN_ROLE, get_user_menu_codes
from app.services.pl_data_scope import apply_pl_scope, assert_pl_row_in_scope
from app.services.pl_plan_arrange_copy_flags import load_plan_arrange_copy_flags
from app.core.tenant_context import get_tenant_no
from app.utils.excel_export import (
    EXPORT_MAX_ROWS,
    ExportColumn,
    ExportValidationError,
    export_rows_to_xlsx_response,
    filter_export_columns,
)
from app.core.utils import generate_uuid
from app.models.crm_models import CrmContract, CrmContractProduct, CrmProduct
from app.models.db_models import DbInboundOutbound
from app.models.pl_models import (
    PlCarrier,
    PlDriver,
    PlPlan,
    PlPlanProduct,
    PlPlanTransport,
    PlTransport,
    PlTransportDbLink,
    PlTransportSiteLink,
    PlVehicle,
)
from app.schemas.pl import PlanProductCreate
from app.schemas.pl.plan import (
    CreateFromContractRequest,
    PlanArchiveBody,
    PlanArchiveBrandQtyItem,
    PlanContractGapListResponse,
    PlanContractGapRead,
    PlanCreate,
    PlanCreateFull,
    PlanDriverFeedbackUpdateBody,
    HazardCertReadyBody,
    ErpEnteredBody,
    PlanForceFollowStatusBody,
    PlanInboundSlipTemplateSave,
    PlanDeliverySlipTemplateSave,
    PlanSubpackSlipTemplateSave,
    PlanListQuery,
    PlanListResponse,
    PlanProductCertsUpdate,
    PlanProductRead,
    PlanRead,
    PlanSideContractsUpdateBody,
    PlanSiteFeedbackUpdateBody,
    PlanUpdate,
    PlanWithProducts,
)
from app.services.pl_plan_contract_gap import list_plan_contract_gaps, update_plan_side_contracts
from app.services.pl_brand_archive import load_brand_qtys_by_plan_product_ids
from app.services.pl_product_display import (
    load_crm_product_alias_maps,
    resolve_product_display_names,
)
from app.schemas.pl.transport import (
    PlanArrangeInfoRead,
    TransportInventoryLinkRead,
    TransportIoOrderBrief,
    TransportSiteLinkRead,
)
from app.services.pl_dibang_bridge_service import revoke_regeneratable_dibang_for_transport
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_site_link_read import build_transport_site_link_read_from_row
from app.services.pl_transport_site_service import revoke_site_links_for_transport
from app.services.pl_inventory_service import apply_archive_inventory_for_plan
from app.services.pl_plan_list_attachments import load_plan_document_attachments_by_plan_ids
from app.services.pl_plan_service import (
    apply_archive_freight,
    apply_archive_product_quantities,
    apply_plan_driver_feedback,
    apply_plan_site_feedback,
    apply_vehicle_prefill_to_plan,
    assert_contract_selectable_for_plan,
    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,
    plan_has_incomplete_hazard_cert_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,
    validate_create_full,
    validate_hazard_cert_ready_for_transport,
    validate_plan_party_contracts,
)

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


_PLAN_NO_SEQ_WIDTH = 3


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


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


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

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

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

    seq = max_seq + 1

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


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


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


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

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

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

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

    return None


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


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

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

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

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

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


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


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


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

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


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


def _plan_product_name_exists(db: Session, tenant_no: str, keyword: str):
    """子查询：计划产品名称模糊匹配。"""
    return (
        db.query(PlPlanProduct.id)
        .filter(
            PlPlanProduct.plan_id == PlPlan.id,
            PlPlanProduct.tenant_no == tenant_no,
            PlPlanProduct.del_flag == 0,
            func.lower(PlPlanProduct.product_name).contains(keyword.lower()),
        )
        .exists()
    )


def _plan_plate_match(keyword: str):
    """车牌号 / 挂车号模糊匹配。"""
    kw = keyword.lower()
    return or_(
        func.lower(PlPlan.license_plate).contains(kw),
        func.lower(PlPlan.trailer_no).contains(kw),
    )


@router.get("/", response_model=PlanListResponse)
def list_plans(
    q: Optional[str] = Query(
        default=None,
        description="模糊搜索（计划编号/供方名称/需方名称/产品名称/车牌号）",
    ),
    plan_no: Optional[str] = Query(default=None, description="计划编号"),
    plan_type: Optional[str] = Query(default=None, alias="planType", description="单据类型"),
    follow_status: Optional[str] = Query(default=None, alias="followStatus", description="跟进状态"),
    transport_status: Optional[str] = Query(default=None, alias="transportStatus", description="运输状态"),
    archive_status: Optional[str] = Query(default=None, alias="archiveStatus", description="归档状态"),
    customer_name: Optional[str] = Query(default=None, alias="customerName", description="客户名称"),
    product_name: Optional[str] = Query(default=None, alias="productName", 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="计划日期结束"),
    has_hazard_product: Optional[int] = Query(
        default=None,
        alias="hasHazardProduct",
        description="1=仅含易制毒/易制爆产品的计划",
    ),
    cert_upload_status: Optional[str] = Query(
        default=None,
        alias="certUploadStatus",
        description="incomplete=至少一张必需危化证书未上传",
    ),
    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 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),
                )
            )

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

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

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

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

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

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

    if product_name:
        product_name_kw = product_name.strip()
        if product_name_kw:
            query = query.filter(_plan_product_name_exists(db, tenant_no, product_name_kw))

    if license_plate:
        plate_kw = license_plate.strip()
        if plate_kw:
            query = query.filter(_plan_plate_match(plate_kw))

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

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

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

    if (cert_upload_status or "").strip().lower() == "incomplete":
        query = query.filter(plan_has_incomplete_hazard_cert_exists(tenant_no))

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

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

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

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

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

    result_list: list[PlanWithProducts] = []
    for plan in plans:
        enriched_products: list[PlanProductRead] = []
        for product in products_by_plan.get(plan.id, []):
            is_toxic, is_explosive = resolve_hazard_flags_from_crm(
                db,
                tenant_no,
                product.product_id,
                hazard_map=hazard_map,
            )
            item = PlanProductRead.model_validate(product)
            raw_brand_qtys = brand_qtys_by_pp.get(product.id) or None
            brand_qtys = (
                [PlanArchiveBrandQtyItem.model_validate(x) for x in raw_brand_qtys]
                if raw_brand_qtys
                else None
            )
            item = item.model_copy(
                update={
                    "is_toxic": is_toxic,
                    "is_explosive": is_explosive,
                    "brand_qtys": brand_qtys,
                    "product_name": _resolve_list_product_name(
                        product,
                        standard_by_id=standard_by_id,
                        alias_options_by_id=alias_options_by_id,
                    ),
                }
            )
            enriched_products.append(item)

        plan_item = PlanWithProducts.model_validate(plan)
        plan_item.products = enriched_products
        supplier_contracts, demander_contracts = contracts_by_plan.get(plan.id, ([], []))
        plan_item.supplier_contracts = supplier_contracts
        plan_item.demander_contracts = demander_contracts
        flags = arrange_copy_flags.get(plan.id)
        if flags:
            plan_item.can_copy_to_driver = flags.can_copy_to_driver
            plan_item.can_copy_to_site = flags.can_copy_to_site
            plan_item.can_copy_pound_qrcode = flags.can_copy_pound_qrcode
        if not (plan_item.personnel_summary or "").strip():
            snapshot = plan_item.resource_snapshot
            if snapshot and snapshot.personnel_summary:
                plan_item.personnel_summary = snapshot.personnel_summary
            elif plan_item.driver_name:
                plan_item.personnel_summary = f"{plan_item.driver_name}(驾)"
        plan_item.document_attachments = document_attachments_by_plan.get(plan.id, [])
        result_list.append(plan_item)

    return PlanListResponse(total=total, list=result_list)



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

PLAN_ARCHIVE_EXPORT_COLUMNS = [
    ExportColumn(title="序号", field="rowNo"),
    ExportColumn(title="到货日期", field="deliveryDate"),
    ExportColumn(title="供方名称", field="supplierName"),
    ExportColumn(title="需方名称", field="customerName"),
    ExportColumn(title="产品名称", field="productName"),
    ExportColumn(title="型号/规格", field="model"),
    ExportColumn(title="包装", field="packageSpec"),
    ExportColumn(title="单位", field="unit"),
    ExportColumn(title="入库数", field="inboundQty"),
    ExportColumn(title="入库价", field="inboundPrice"),
    ExportColumn(title="入库金额", field="inboundAmount"),
    ExportColumn(title="出库数", field="outboundQty"),
    ExportColumn(title="出库价", field="outboundPrice"),
    ExportColumn(title="出库金额", field="outboundAmount"),
    ExportColumn(title="车牌号", field="licensePlate"),
    ExportColumn(title="运费单位", field="freightType"),
    ExportColumn(title="运费单价", field="freightPrice"),
    ExportColumn(title="运费金额", field="freightAmount"),
    ExportColumn(title="备注", field="remark"),
]

PLAN_ARCHIVE_FREIGHT_TYPE_LABELS = {
    "by_weight": "按重量",
    "by_trip": "按趟",
}


def _format_plan_delivery_date_for_export(plan: PlPlan) -> str:
    if plan.delivery_time is None:
        return ""
    return str(plan.delivery_time)[:10]


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}


def _load_products_by_plan_id(db: Session, tenant_no: str, plan_ids: list[str]) -> dict[str, list[PlPlanProduct]]:
    if not plan_ids:
        return {}
    products = (
        db.query(PlPlanProduct)
        .filter(
            PlPlanProduct.tenant_no == tenant_no,
            PlPlanProduct.del_flag == 0,
            PlPlanProduct.plan_id.in_(plan_ids),
        )
        .order_by(PlPlanProduct.order_num)
        .all()
    )
    grouped: dict[str, list[PlPlanProduct]] = {}
    for product in products:
        grouped.setdefault(product.plan_id, []).append(product)
    return grouped


def _build_plan_archive_export_rows(
    plans: list[PlPlan],
    products_by_plan: dict[str, list[PlPlanProduct]],
    transports_by_id: dict[str, PlTransport],
) -> list[dict]:
    rows: list[dict] = []
    row_no = 0
    for plan in plans:
        products = products_by_plan.get(plan.id, [])
        transport = transports_by_id.get(plan.transport_id) if plan.transport_id else None
        freight_type_raw = transport.freight_type if transport else "by_weight"
        freight_type_label = PLAN_ARCHIVE_FREIGHT_TYPE_LABELS.get(
            freight_type_raw, freight_type_raw or ""
        )
        freight_price = transport.freight_price if transport else None
        freight_amount = plan.freight_amount
        if freight_amount is None and transport is not None:
            freight_amount = transport.freight_amount
        plan_common = {
            "deliveryDate": _format_plan_delivery_date_for_export(plan),
            "supplierName": plan.supplier_name or "",
            "customerName": plan.customer_name or "",
            "licensePlate": plan.license_plate or "",
            "freightType": freight_type_label,
            "freightPrice": freight_price,
            "freightAmount": freight_amount,
            "remark": "",
        }
        product_rows = products or [None]
        for product in product_rows:
            row_no += 1
            row = {
                **plan_common,
                "rowNo": row_no,
                "productName": "",
                "model": "",
                "packageSpec": "",
                "unit": "",
                "inboundQty": None,
                "inboundPrice": None,
                "inboundAmount": None,
                "outboundQty": None,
                "outboundPrice": None,
                "outboundAmount": None,
            }
            if product is not None:
                row.update(
                    {
                        "productName": product.product_name or "",
                        "model": product.model or "",
                        "packageSpec": product.package_spec or "",
                        "unit": product.unit or "",
                        "inboundQty": product.inbound_qty,
                        "inboundPrice": product.inbound_price,
                        "inboundAmount": product.inbound_amount,
                        "outboundQty": product.outbound_qty,
                        "outboundPrice": product.outbound_price,
                        "outboundAmount": product.outbound_amount,
                    }
                )
            rows.append(row)
    return 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")
    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")
    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")
    has_hazard_product: Optional[int] = Field(default=None, alias="hasHazardProduct")
    cert_upload_status: Optional[str] = Field(default=None, alias="certUploadStatus")
    ids: Optional[List[str]] = None
    columns: Optional[List[str]] = None
    export_format: Optional[str] = Field(default=None, alias="exportFormat", description="archive=运输计划归档专用导出")

    model_config = {"populate_by_name": True}


@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),
                    )
                )
        if body.plan_no:
            query = query.filter(PlPlan.plan_no.contains(body.plan_no))
        if body.plan_type:
            query = query.filter(PlPlan.plan_type == body.plan_type)
        if body.follow_status:
            query = query.filter(PlPlan.follow_status == body.follow_status)
        if body.transport_status:
            query = query.filter(PlPlan.transport_status == body.transport_status)
        if body.archive_status:
            query = query.filter(PlPlan.archive_status == body.archive_status)
        if body.customer_name:
            query = query.filter(func.lower(PlPlan.customer_name).contains(body.customer_name.lower()))
        if body.product_name:
            product_name_kw = body.product_name.strip()
            if product_name_kw:
                query = query.filter(_plan_product_name_exists(db, tenant_no, product_name_kw))
        if body.license_plate:
            plate_kw = body.license_plate.strip()
            if plate_kw:
                query = query.filter(_plan_plate_match(plate_kw))
        if body.plan_date_from:
            query = query.filter(PlPlan.plan_date >= body.plan_date_from)
        if body.plan_date_to:
            query = query.filter(PlPlan.plan_date <= body.plan_date_to)
        if body.has_hazard_product == 1:
            query = query.filter(plan_has_hazard_product_exists(tenant_no))
        if (body.cert_upload_status or "").strip().lower() == "incomplete":
            query = query.filter(plan_has_incomplete_hazard_cert_exists(tenant_no))
        if body.ids:
            query = query.filter(PlPlan.id.in_(body.ids))
        query = apply_pl_scope(query, PlPlan, user, db, tenant_no)
        plans = query.order_by(PlPlan.create_time.desc()).limit(EXPORT_MAX_ROWS + 1).all()
        date_str = datetime.now().strftime("%Y%m%d")
        is_archive_export = (body.export_format or "").strip().lower() == "archive"
        if is_archive_export:
            plan_ids = [plan.id for plan in plans]
            products_by_plan = _load_products_by_plan_id(db, tenant_no, plan_ids)
            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(plans, products_by_plan, transports_by_id)
            cols = PLAN_ARCHIVE_EXPORT_COLUMNS
            sheet_title = "计划归档"
            filename_stem = f"计划归档_{date_str}"
        else:
            items = []
            for plan in plans:
                data = PlanWithProducts.model_validate(plan).model_dump(by_alias=True)
                if data.get("planDate") is not None:
                    data["planDate"] = str(data["planDate"])[:10]
                if data.get("erpEntered") is not None:
                    data["erpEntered"] = "是" if int(data["erpEntered"]) else "否"
                items.append(data)
            cols = filter_export_columns(PLAN_EXPORT_COLUMNS, body.columns)
            sheet_title = "计划列表"
            filename_stem = f"计划列表_{date_str}"
        return export_rows_to_xlsx_response(
            db,
            tenant_no=tenant_no,
            sheet_title=sheet_title,
            columns=cols,
            rows=items,
            filename_stem=filename_stem,
        )
    except ExportValidationError as e:
        raise HTTPException(status_code=400, detail=str(e)) from e


@router.get("/count-by-contract")
def count_plans_by_contract(
    contract_ids: str = Query(..., alias="contractIds", description="合同ID，逗号分隔"),
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> dict:
    from app.models.pl_models import PlPlanContract

    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    ids = [item.strip() for item in contract_ids.split(",") if item.strip()]
    if not ids:
        return {"counts": {}}

    # 旧单：主表 contract_id；新单：pl_plan_contract 分侧关联（按计划去重计数）
    legacy_rows = (
        db.query(PlPlan.contract_id, func.count(PlPlan.id))
        .filter(
            PlPlan.tenant_no == tenant_no,
            PlPlan.del_flag == 0,
            PlPlan.contract_id.in_(ids),
        )
    )
    legacy_rows = apply_pl_scope(legacy_rows, PlPlan, user, db, tenant_no).group_by(PlPlan.contract_id).all()
    counts: dict[str, int] = {contract_id: count for contract_id, count in legacy_rows}

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

    return {"counts": counts}


@router.get("/contract-gaps", response_model=PlanContractGapListResponse)
def list_contract_gaps(
    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"),
    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())
    total, items = list_plan_contract_gaps(
        db,
        tenant_no=tenant_no,
        user=user,
        page=page,
        page_size=page_size,
        plan_no=plan_no,
        gap_type=gap_type,
        customer_keyword=customer_keyword,
        plan_date_from=plan_date_from,
        plan_date_to=plan_date_to,
        q=q,
    )
    return PlanContractGapListResponse(total=total, list=items)



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


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


@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())
        _, items = list_plan_contract_gaps(
            db,
            tenant_no=tenant_no,
            user=user,
            page=1,
            page_size=EXPORT_MAX_ROWS,
            plan_no=body.plan_no,
            gap_type=body.gap_type,
            customer_keyword=body.customer_keyword,
            plan_date_from=body.plan_date_from,
            plan_date_to=body.plan_date_to,
            q=body.q,
            ids=body.ids,
            max_page_size=EXPORT_MAX_ROWS,
        )
        plan_type_labels = {"direct": "直销单", "inbound": "入库单", "outbound": "出库单"}
        gap_labels = {"supplier": "仅缺供方合同", "demander": "仅缺需方合同", "both": "双侧缺口"}
        rows = []
        for item in items:
            data = item.model_dump(by_alias=True)
            data["planType"] = plan_type_labels.get(str(data.get("planType") or ""), data.get("planType"))
            sides = set(data.get("gapSides") or [])
            if "supplier" in sides and "demander" in sides:
                data["gapType"] = "双侧缺口"
            elif "supplier" in sides:
                data["gapType"] = "仅缺供方合同"
            elif "demander" in sides:
                data["gapType"] = "仅缺需方合同"
            else:
                data["gapType"] = ""
            for k, v in list(data.items()):
                if hasattr(v, "isoformat"):
                    data[k] = str(v)[:19]
            rows.append(data)
        cols = filter_export_columns(CONTRACT_GAP_EXPORT_COLUMNS, body.columns)
        date_str = datetime.now().strftime("%Y%m%d")
        return export_rows_to_xlsx_response(
            db, tenant_no=tenant_no, sheet_title="合同完善", columns=cols, rows=rows,
            filename_stem=f"计划合同完善_{date_str}",
        )
    except ExportValidationError as e:
        raise HTTPException(status_code=400, detail=str(e)) from e


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


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


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

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

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

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

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

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

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


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

    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)

    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
    return result


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


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

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

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

    assert_contract_selectable_for_plan(db, contract.id, tenant_no)

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

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

    plan = PlPlan(
        id=generate_uuid(),
        tenant_no=tenant_no,
        plan_no=plan_no,
        create_date=_resolve_create_date(payload.create_date),
        plan_date=payload.plan_date,
        belongs_company=contract.belongs_company or "shanxia",
        plan_type=plan_type,
        follow_status="pending_arrange",
        transport_status="pending",
        archive_status="unarchived",
        complete_status="no",
        contract_id=contract.id,
        contract_no=contract.contract_no,
        supplier_id=contract.supplier_id,
        supplier_name=(contract.party_a or contract.supplier_name or "").strip() or None,
        destination_id=contract.customer_id,
        # 客户名称严格跟合同「客户名称」；目的地名称与客户保持一致（不再取需方可章名 party_b）
        destination_name=(contract.customer_name or "").strip() or None,
        customer_id=contract.customer_id,
        customer_name=(contract.customer_name or "").strip() or None,
        contact_name=contract.contact_name,
        contact_phone=contract.contact_phone,
        party_a=contract.party_a,
        party_b=contract.party_b,
        contract_start_date=contract.start_date,
        contract_end_date=contract.end_date,
        delivery_method=contract.delivery_method,
        settlement_method=contract.settlement_method,
        has_freight=1 if contract.settlement_method and "含运" in contract.settlement_method else 0,
        follow_user_id=payload.follow_user_id or user.id,
        follow_user_name=payload.follow_user_name or user.nick_name,
        create_by=user.id,
        create_dept_id=user.dept_id,
    )
    db.add(plan)

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

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

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


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


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

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

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

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

    result = PlanWithProducts.model_validate(plan)
    result.products = enriched_products
    supplier_contracts, demander_contracts = load_plan_contracts_grouped(
        db, tenant_no, [id]
    ).get(id, ([], []))
    result.supplier_contracts = supplier_contracts
    result.demander_contracts = demander_contracts
    return result


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


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

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

    # 更新主表字段（跟进状态仅由业务动作接口维护，忽略客户端写入）
    update_data = payload.model_dump(
        exclude_unset=True,
        exclude={"products", "transport", "follow_status", "supplier_contracts", "demander_contracts"},
    )
    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
    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

    # 更新产品明细
    if payload.products is not None:
        existing_products = (
            db.query(PlPlanProduct)
            .filter(
                PlPlanProduct.plan_id == id,
                PlPlanProduct.del_flag == 0,
            )
            .all()
        )
        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
        }

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

        plan_products: list[PlPlanProduct] = []
        for idx, prod in enumerate(payload.products):
            is_toxic, is_explosive = resolve_hazard_flags_from_crm(
                db,
                tenant_no,
                prod.product_id,
            )
            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,
                }
            )
            pp = build_plan_product(
                tenant_no=tenant_no,
                plan_id=id,
                plan_type=plan_type,
                prod=prod_payload,
                order_num=idx,
                user_id=user.id,
                dept_id=user.dept_id,
            )
            db.add(pp)
            plan_products.append(pp)

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

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

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

    plan.update_by = user.id
    db.commit()
    db.refresh(plan)
    result = PlanRead.model_validate(plan)
    result.supplier_contracts = supplier_contracts
    result.demander_contracts = demander_contracts
    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=[])
    if plan.follow_status == "cancelled" or plan.transport_status != "arranged" or not plan.transport_id:
        return empty

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

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

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

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

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

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

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

    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,
        )
        if not site_records and token and header.status != "voided":
            site_records = [
                TransportSiteLinkRead(
                    plan_product_id="",
                    product_name=header.order_no,
                    site_link_id=header.id,
                    plan_no=plan.plan_no or "",
                    business_type="入库" if kind == "inbound" else "出库",
                    site_access_token=token,
                    site_url=build_io_order_site_url(kind, token),
                )
            ]

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


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


@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 != "pending_arrange":
        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"
ALLOWED_FORCE_FOLLOW_STATUSES = frozenset(
    {"wait_notice", "pending_arrange", "arranged", "completed", "cancelled"}
)


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

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

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

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


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


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

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

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

    if plan.transport_id:
        transport = (
            db.query(PlTransport)
            .filter(
                PlTransport.id == plan.transport_id,
                PlTransport.tenant_no == tenant_no,
                PlTransport.del_flag == 0,
            )
            .first()
        )
        if transport:
            try:
                revoke_regeneratable_dibang_for_transport(
                    db,
                    tenant_no=tenant_no,
                    transport_id=transport.id,
                    user_id=user.id,
                )
            except ValueError as exc:
                raise HTTPException(status_code=400, detail=str(exc)) from exc
            revoke_site_links_for_transport(
                db,
                tenant_no=tenant_no,
                transport_id=transport.id,
                user_id=user.id,
            )
            transport.transport_status = "cancelled"
            transport.update_by = user.id

    void_io_orders_for_plan(
        db, tenant_no=tenant_no, plan_id=plan.id, user_id=user.id
    )

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


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


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

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

    already_archived = plan.archive_status == "archived"

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

    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 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}/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"):
        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