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

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

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

from app.core.config import settings
from app.core.database import get_db
from app.core.db_auth import DbUserContext, get_db_current_user
from app.core.delete_guard import assert_row_deletable_by_creator, user_info_from_context
from app.services.pl_data_scope import apply_pl_scope, assert_pl_row_in_scope
from app.core.tenant_context import get_tenant_no
from app.core.utils import generate_uuid
from app.models.crm_models import CrmContract, CrmContractProduct, CrmProduct
from app.models.db_models import DbInboundOutbound
from app.models.pl_models import (
    PlCarrier,
    PlDriver,
    PlPlan,
    PlPlanProduct,
    PlPlanTransport,
    PlTransport,
    PlTransportDbLink,
    PlTransportSiteLink,
    PlVehicle,
)
from app.schemas.pl import PlanProductCreate
from app.schemas.pl.plan import (
    CreateFromContractRequest,
    PlanArchiveBody,
    PlanCreate,
    PlanCreateFull,
    PlanDriverFeedbackUpdateBody,
    HazardCertReadyBody,
    PlanInboundSlipTemplateSave,
    PlanDeliverySlipTemplateSave,
    PlanSubpackSlipTemplateSave,
    PlanListQuery,
    PlanListResponse,
    PlanProductCertsUpdate,
    PlanProductRead,
    PlanRead,
    PlanSiteFeedbackUpdateBody,
    PlanUpdate,
    PlanWithProducts,
)
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_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,
    record_plan_change,
    resolve_hazard_flags_from_crm,
    set_hazard_cert_ready,
    sync_transport_prefill_for_plan,
    update_plan_product_certs,
    validate_create_full,
    validate_hazard_cert_ready_for_transport,
)

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


_PLAN_NO_SEQ_WIDTH = 3


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


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


@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="产品名称"),
    plan_date_from: Optional[date] = Query(default=None, alias="planDateFrom", description="计划日期开始"),
    plan_date_to: Optional[date] = Query(default=None, alias="planDateTo", description="计划日期结束"),
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=20, ge=1, le=100, alias="pageSize"),
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> PlanListResponse:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    query = db.query(PlPlan).filter(
        PlPlan.tenant_no == tenant_no,
        PlPlan.del_flag == 0,
    )

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

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

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

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

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

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

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

    if product_name:
        product_name_kw = product_name.strip()
        if product_name_kw:
            query = query.filter(
                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(product_name_kw.lower()),
                )
                .exists()
            )

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

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

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

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

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

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

    hazard_map = fetch_hazard_flags_map(
        db,
        tenant_no,
        [product.product_id for product in all_products if product.product_id],
    )

    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)
            item = item.model_copy(update={"is_toxic": is_toxic, "is_explosive": is_explosive})
            enriched_products.append(item)

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

    return PlanListResponse(total=total, list=result_list)


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

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


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


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

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

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

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

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

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

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


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

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

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

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

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

    db.commit()
    db.refresh(plan)

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


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


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

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

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

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

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

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

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


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


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

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

    hazard_map = fetch_hazard_flags_map(
        db,
        tenant_no,
        [product.product_id for product in products if product.product_id],
    )

    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)
        item = item.model_copy(update={"is_toxic": is_toxic, "is_explosive": is_explosive})
        enriched_products.append(item)

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


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


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

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

    if plan.transport_status == "arranged" and plan.plan_type in ("inbound", "outbound"):
        raise HTTPException(
            status_code=400,
            detail="已安排的入库/出库计划不可修改，请作废后新建计划",
        )

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

    plan_type = payload.plan_type or plan.plan_type

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


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


@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


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