shanxia2.0/backend/app/services/pl_plan_archive_export.py

"""运输计划归档导出：按段次×产品展开（装货行与各送货段各一行）。"""
from __future__ import annotations

from datetime import date, datetime
from decimal import Decimal
from typing import Optional, Union

from sqlalchemy.orm import Session

from app.models.pl_models import PlPlan, PlPlanContract, PlPlanProduct, PlTransport
from app.schemas.pl.plan import PlanDeliveryRead, PlanLoadRead, PlanStopProductRead
from app.services.pl.plan_list_export_segments import (
    format_stop_segment_label,
    iter_segment_product_slots,
)
from app.services.pl.plan_list_filters import customer_name_matches_keywords
from app.services.pl_brand_archive import (
    expand_export_rows_for_plan_brand,
    load_brand_qtys_by_plan_product_ids,
    load_brand_qtys_by_stop_product_ids,
)
from app.services.pl_internal_customer import is_internal_customer_name
from app.services.pl_plan_stop_service import load_plan_stops_grouped
from app.services.pl_product_display import load_crm_product_alias_maps, resolve_product_display_names

StopKind = str  # "load" | "delivery"


def _unique_contract_nos(*sources: Optional[str]) -> str:
    """按出现顺序去重拼接合同编号。"""
    seen: list[str] = []
    for raw in sources:
        for part in str(raw or "").replace("，", ",").split(","):
            no = part.strip()
            if no and no not in seen:
                seen.append(no)
    return "、".join(seen)


def _contract_nos_from_stop_links(contracts: Optional[list]) -> list[str]:
    """段次合同关联 → 编号列表（去重保序）。"""
    seen: list[str] = []
    for item in contracts or []:
        no = (
            getattr(item, "contract_no", None)
            or (item.get("contractNo") if isinstance(item, dict) else None)
            or getattr(item, "contractNo", None)
            or ""
        )
        no = str(no or "").strip()
        if not no:
            cid = (
                getattr(item, "contract_id", None)
                or (item.get("contractId") if isinstance(item, dict) else None)
                or getattr(item, "contractId", None)
                or ""
            )
            no = str(cid or "").strip()
        if no and no not in seen:
            seen.append(no)
    return seen


def format_stop_contract_export_label(
    kind: StopKind,
    contracts: Optional[list],
    customer_name: Optional[str],
) -> str:
    """与前端列表一致：有编号显示编号；山峡/云栋缺合同 → 装货出库 / 送货入库。"""
    nos = _contract_nos_from_stop_links(contracts)
    if nos:
        return "、".join(nos)
    if is_internal_customer_name(customer_name):
        return "出库" if kind == "load" else "入库"
    return ""


def _load_contract_nos_by_plan(
    db: Session, tenant_no: str, plan_ids: list[str]
) -> dict[str, list[str]]:
    if not plan_ids:
        return {}
    rows = (
        db.query(PlPlanContract)
        .filter(
            PlPlanContract.tenant_no == tenant_no,
            PlPlanContract.del_flag == 0,
            PlPlanContract.plan_id.in_(plan_ids),
        )
        .order_by(PlPlanContract.order_num.asc(), PlPlanContract.create_time.asc())
        .all()
    )
    out: dict[str, list[str]] = {pid: [] for pid in plan_ids}
    for row in rows:
        no = (row.contract_no or "").strip()
        if not no:
            continue
        bucket = out.setdefault(row.plan_id, [])
        if no not in bucket:
            bucket.append(no)
    return out


PLAN_TYPE_LABELS = {
    "direct": "直销单",
    "inbound": "入库单",
    "outbound": "出库单",
}

ARCHIVE_STATUS_LABELS = {
    "cancelled": "已取消",
    "archived": "已归档",
    "unarchived": "未归档",
    "draft": "已暂存",
}

FOLLOW_STATUS_LABELS = {
    "drafting": "草稿中",
    "wait_notice": "等通知",
    "pending_arrange": "待安排",
    "arranged": "已安排",
    "completed": "已完成",
    "cancelled": "已取消",
    "draft": "待安排",
    "processing": "进行中",
}

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


def resolve_archive_list_io_column(kind: StopKind, customer_name: Optional[str]) -> str:
    """装货/送货段入库出库列方向（与前端 resolveArchiveListIoColumn 一致）。"""
    internal = is_internal_customer_name(customer_name)
    if kind == "load":
        return "outbound" if internal else "inbound"
    return "inbound" if internal else "outbound"


def _format_date_value(value: Optional[Union[date, datetime, str]]) -> str:
    if value is None:
        return ""
    return str(value)[:10]


def _format_plate_display(license_plate: Optional[str], trailer_no: Optional[str]) -> str:
    plate = (license_plate or "").strip()
    trailer = (trailer_no or "").strip()
    if not plate and not trailer:
        return ""
    if plate and trailer:
        return f"{plate}{trailer}"
    return plate or trailer


def _get_stop_archive_actual_qty(product: PlanStopProductRead, kind: StopKind) -> Optional[Decimal]:
    return product.inbound_qty if kind == "load" else product.outbound_qty


def _get_stop_archive_unit_price(product: PlanStopProductRead, kind: StopKind) -> Optional[Decimal]:
    return product.inbound_price if kind == "load" else product.outbound_price


def _get_stop_archive_amount(product: PlanStopProductRead, kind: StopKind) -> Optional[Decimal]:
    stored = product.inbound_amount if kind == "load" else product.outbound_amount
    if stored is not None:
        return stored
    qty = _get_stop_archive_actual_qty(product, kind)
    price = _get_stop_archive_unit_price(product, kind)
    if qty is not None and price is not None:
        return qty * price
    return None


def _stop_time(kind: StopKind, stop: Union[PlanLoadRead, PlanDeliveryRead]) -> str:
    if kind == "load":
        return _format_date_value(getattr(stop, "load_time", None))
    return _format_date_value(getattr(stop, "delivery_time", None))


def _load_legacy_products_by_plan(
    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 _plan_common_export_fields(
    plan: PlPlan,
    transport: Optional[PlTransport],
    *,
    contract_nos: Optional[list[str]] = None,
) -> dict:
    freight_type_raw = transport.freight_type if transport else "by_weight"
    freight_type_label = 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_date = _format_date_value(plan.plan_date)
    link_nos = contract_nos or []
    return {
        "planId": plan.id or "",
        "planNo": plan.plan_no or "",
        "contractNo": _unique_contract_nos(plan.contract_no, *link_nos),
        "planDate": plan_date,
        "archiveStatus": ARCHIVE_STATUS_LABELS.get(plan.archive_status or "", plan.archive_status or ""),
        "followStatus": FOLLOW_STATUS_LABELS.get(plan.follow_status or "", plan.follow_status or ""),
        "planType": PLAN_TYPE_LABELS.get(plan.plan_type or "", plan.plan_type or ""),
        "licensePlate": _format_plate_display(plan.license_plate, plan.trailer_no),
        "erpEntered": "是" if int(plan.erp_entered or 0) == 1 else "否",
        "freightType": freight_type_label,
        "freightPrice": freight_price,
        "freightAmount": freight_amount,
        "remark": plan.remark or "",
    }


def _empty_pair_side_fields(*, side: str) -> dict:
    """side: load | delivery"""
    prefix = "load" if side == "load" else "delivery"
    return {
        f"{prefix}Customer": "",
        f"{prefix}Address": "",
        f"{prefix}Time": "",
        f"{prefix}PlanQty": None,
        f"{prefix}ActualQty": None,
        f"{prefix}InboundQty": None,
        f"{prefix}OutboundQty": None,
        f"{prefix}UnitPrice": None,
        f"{prefix}Amount": None,
    }


def _empty_product_identity_fields() -> dict:
    return {
        "productName": "",
        "brandName": "",
        "aliasName": "",
        "model": "",
        "packageSpec": "",
        "unit": "",
    }


def _resolve_export_product_display(
    plan_product: Optional[PlPlanProduct],
    stop_product: Optional[PlanStopProductRead],
    *,
    standard_by_id: dict[str, str],
    alias_options_by_id: dict[str, list[str]],
) -> tuple[str, str]:
    """返回 (展示用产品名称, 别名)。"""
    saved_name = ""
    product_id = ""
    if plan_product is not None:
        saved_name = plan_product.product_name or ""
        product_id = plan_product.product_id or ""
    elif stop_product is not None:
        saved_name = stop_product.product_name or ""
        product_id = stop_product.product_id or ""
    display_name, alias = resolve_product_display_names(
        saved_product_name=saved_name,
        product_id=product_id,
        standard_by_id=standard_by_id,
        alias_options_by_id=alias_options_by_id,
    )
    return (display_name or saved_name or "").strip(), (alias or "").strip()


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


def _product_identity_from_stop(
    product: PlanStopProductRead,
    *,
    plan_product: Optional[PlPlanProduct] = None,
    standard_by_id: Optional[dict[str, str]] = None,
    alias_options_by_id: Optional[dict[str, list[str]]] = None,
) -> dict:
    display_name, alias_name = _resolve_export_product_display(
        plan_product,
        product,
        standard_by_id=standard_by_id or {},
        alias_options_by_id=alias_options_by_id or {},
    )
    return {
        "productName": display_name or (product.product_name or ""),
        "brandName": "",
        "aliasName": alias_name,
        "model": product.model or "",
        "packageSpec": product.package_spec or "",
        "unit": product.unit or "",
    }


def _segment_slot_customer(
    kind: str,
    load: Optional[PlanLoadRead],
    delivery: Optional[PlanDeliveryRead],
) -> str:
    if kind == "load" and load is not None:
        return load.customer_name or ""
    if kind == "delivery" and delivery is not None:
        return delivery.customer_name or ""
    return ""


def _build_archive_segment_row(
    row_no: int,
    plan_common: dict,
    kind: str,
    load: Optional[PlanLoadRead],
    delivery: Optional[PlanDeliveryRead],
    delivery_seq: Optional[int],
    product: Optional[PlanStopProductRead],
    *,
    plan_products_by_id: dict[str, PlPlanProduct],
    plan_products_by_product_id: dict[str, PlPlanProduct],
    standard_by_id: dict[str, str],
    alias_options_by_id: dict[str, list[str]],
) -> dict:
    """一段次×一个产品一行：只填本侧客户/地址/时间/合同与本侧量价。"""
    fallback_contract = str(plan_common.get("contractNo") or "")
    row = {
        **plan_common,
        "rowNo": row_no,
        "_stopKind": kind,
        "_deliverySeq": delivery_seq,
        "stopSegment": format_stop_segment_label(kind, delivery_seq),
        "loadCustomer": "",
        "loadAddress": "",
        "loadTime": "",
        "deliveryCustomer": "",
        "deliveryAddress": "",
        "deliveryTime": "",
        "contractNo": fallback_contract,
        **_empty_product_identity_fields(),
        "planQty": None,
        "actualQty": None,
        "unitPrice": None,
        "amount": None,
    }
    if kind == "load" and load is not None:
        customer_name = load.customer_name or ""
        row["loadCustomer"] = customer_name
        row["loadAddress"] = load.address or ""
        row["loadTime"] = _stop_time("load", load)
        row["contractNo"] = format_stop_contract_export_label(
            "load", load.contracts, customer_name
        ) or fallback_contract
    elif kind == "delivery" and delivery is not None:
        customer_name = delivery.customer_name or ""
        row["deliveryCustomer"] = customer_name
        row["deliveryAddress"] = delivery.address or ""
        row["deliveryTime"] = _stop_time("delivery", delivery)
        row["contractNo"] = format_stop_contract_export_label(
            "delivery", delivery.contracts, customer_name
        ) or fallback_contract
    if product is not None:
        plan_product = _lookup_plan_product(
            product, plan_products_by_id, plan_products_by_product_id
        )
        row.update(
            _product_identity_from_stop(
                product,
                plan_product=plan_product,
                standard_by_id=standard_by_id,
                alias_options_by_id=alias_options_by_id,
            )
        )
        row["planQty"] = product.plan_qty
        row["actualQty"] = _get_stop_archive_actual_qty(product, kind)
        row["unitPrice"] = _get_stop_archive_unit_price(product, kind)
        row["amount"] = _get_stop_archive_amount(product, kind)
    return row


def _append_legacy_plan_product_rows(
    rows: list[dict],
    row_no: int,
    plan: PlPlan,
    plan_common: dict,
    products: list[PlPlanProduct],
    *,
    standard_by_id: Optional[dict[str, str]] = None,
    alias_options_by_id: Optional[dict[str, list[str]]] = None,
    brand_by_plan_product: Optional[dict[str, list]] = None,
) -> int:
    """无段表数据时回退计划产品级导出（兼容旧数据）；有品牌明细时按品牌拆行。"""
    product_rows = products or [None]
    fallback_no = str(plan_common.get("contractNo") or "")
    if fallback_no:
        legacy_contract_no = fallback_no
    else:
        legacy_contract_no = _unique_contract_nos(
            format_stop_contract_export_label("load", None, plan.supplier_name),
            format_stop_contract_export_label(
                "delivery", None, plan.customer_name or plan.destination_name
            ),
        )
    brand_index = brand_by_plan_product or {}
    for product in product_rows:
        row = {
            **plan_common,
            "rowNo": 0,
            "_stopKind": "",
            "_deliverySeq": None,
            "stopSegment": "",
            "deliveryStopKind": "",
            **_empty_pair_side_fields(side="load"),
            **_empty_pair_side_fields(side="delivery"),
            **_empty_product_identity_fields(),
            "loadCustomer": plan.supplier_name or plan.customer_name or "",
            "loadAddress": "",
            "deliveryCustomer": plan.customer_name or "",
            "deliveryAddress": plan.destination_name or "",
            "deliveryTime": _format_date_value(plan.delivery_time),
            "contractNo": legacy_contract_no,
            # 兼容旧字段
            "stopKind": "",
            "stopCustomer": plan.customer_name or plan.supplier_name or "",
            "stopAddress": plan.destination_name or "",
            "stopTime": _format_date_value(plan.delivery_time),
            "planQty": None,
            "actualQty": None,
            "inboundQty": None,
            "outboundQty": None,
            "unitPrice": None,
            "amount": None,
        }
        if product is not None:
            display_name, alias_name = _resolve_export_product_display(
                product,
                None,
                standard_by_id=standard_by_id or {},
                alias_options_by_id=alias_options_by_id or {},
            )
            row.update(
                {
                    "productName": display_name or (product.product_name or ""),
                    "brandName": "",
                    "aliasName": alias_name,
                    "model": product.model or "",
                    "packageSpec": product.package_spec or "",
                    "unit": product.unit or "",
                    "loadPlanQty": product.quantity,
                    "loadActualQty": product.actual_qty,
                    "loadInboundQty": product.inbound_qty,
                    "loadOutboundQty": product.outbound_qty,
                    "loadUnitPrice": product.inbound_price or product.outbound_price,
                    "loadAmount": product.inbound_amount or product.outbound_amount,
                    "planQty": product.quantity,
                    "actualQty": product.actual_qty,
                    "inboundQty": product.inbound_qty,
                    "outboundQty": product.outbound_qty,
                    "unitPrice": product.inbound_price or product.outbound_price,
                    "amount": product.inbound_amount or product.outbound_amount,
                }
            )
        for expanded in expand_export_rows_for_plan_brand(
            row,
            archive_status=getattr(plan, "archive_status", None),
            plan_product=product,
            stop_product_id=None,
            plan_product_id=str(getattr(product, "id", None) or "")
            if product is not None
            else None,
            by_stop_id={},
            by_plan_product_id=brand_index,
            apply_unit_price=True,
        ):
            row_no += 1
            expanded["rowNo"] = row_no
            rows.append(expanded)
    return row_no


def build_plan_archive_export_rows(
    db: Session,
    tenant_no: str,
    plans: list[PlPlan],
    transports_by_id: dict[str, PlTransport],
    *,
    stop_kind: Optional[str] = None,
    stop_customer_keywords: Optional[list] = None,
) -> list[dict]:
    """构建归档导出数据行。

    默认按段次×产品展开：先装货产品行，再各送货段产品行。
    stop_kind 为 load / delivery 时只保留对应侧。
    stop_customer_keywords 非空时按该行所属侧客户过滤。
    无段表数据时回退计划产品级导出，stopSegment 为空。
    """
    if not plans:
        return []

    include_load = stop_kind != "delivery"
    include_delivery = stop_kind != "load"
    keywords = [str(item).strip() for item in (stop_customer_keywords or []) if str(item).strip()]

    def _keep_customer(name: Optional[str]) -> bool:
        if not keywords:
            return True
        return customer_name_matches_keywords(name, keywords)

    plan_ids = [plan.id for plan in plans]
    stops_by_plan = load_plan_stops_grouped(db, tenant_no, plan_ids)
    legacy_products_by_plan = _load_legacy_products_by_plan(db, tenant_no, plan_ids)
    contract_nos_by_plan = _load_contract_nos_by_plan(db, tenant_no, plan_ids)
    crm_product_ids: list[str] = []
    for products in legacy_products_by_plan.values():
        for product in products:
            if product.product_id:
                crm_product_ids.append(product.product_id)
    for load_read, delivery_reads in stops_by_plan.values():
        for stop in ([load_read] if load_read is not None else []) + list(delivery_reads or []):
            for stop_product in stop.products or []:
                if stop_product.product_id:
                    crm_product_ids.append(stop_product.product_id)
    standard_by_id, alias_options_by_id = load_crm_product_alias_maps(
        db,
        tenant_no=tenant_no,
        product_ids=crm_product_ids,
    )
    plan_product_ids: list[str] = []
    stop_product_ids: list[str] = []
    for products in legacy_products_by_plan.values():
        for product in products:
            if product.id:
                plan_product_ids.append(str(product.id))
    for load_read, delivery_reads in stops_by_plan.values():
        for stop in ([load_read] if load_read is not None else []) + list(delivery_reads or []):
            for stop_product in stop.products or []:
                sid = str(getattr(stop_product, "id", None) or "").strip()
                if sid:
                    stop_product_ids.append(sid)
    brand_by_plan_product = load_brand_qtys_by_plan_product_ids(
        db, tenant_no=tenant_no, plan_product_ids=plan_product_ids
    )
    brand_by_stop = load_brand_qtys_by_stop_product_ids(
        db, tenant_no=tenant_no, stop_product_ids=stop_product_ids
    )

    rows: list[dict] = []
    row_no = 0
    for plan in plans:
        transport = transports_by_id.get(plan.transport_id) if plan.transport_id else None
        plan_common = _plan_common_export_fields(
            plan,
            transport,
            contract_nos=contract_nos_by_plan.get(plan.id, []),
        )
        load_read, delivery_reads = stops_by_plan.get(plan.id, (None, []))
        plan_products = legacy_products_by_plan.get(plan.id, [])
        plan_products_by_id = {str(p.id): p for p in plan_products if p.id}
        plan_products_by_product_id: dict[str, PlPlanProduct] = {}
        for product in plan_products:
            pid = str(product.product_id or "").strip()
            if pid and pid not in plan_products_by_product_id:
                plan_products_by_product_id[pid] = product

        has_stops = load_read is not None or bool(delivery_reads)
        if not has_stops:
            if not include_load:
                continue
            legacy_customer = plan.customer_name or plan.supplier_name or ""
            if not _keep_customer(legacy_customer):
                continue
            row_no = _append_legacy_plan_product_rows(
                rows,
                row_no,
                plan,
                plan_common,
                plan_products,
                standard_by_id=standard_by_id,
                alias_options_by_id=alias_options_by_id,
                brand_by_plan_product=brand_by_plan_product,
            )
            continue

        for kind, slot_load, slot_delivery, seq, stop_product in iter_segment_product_slots(
            load_read, delivery_reads
        ):
            if kind == "load" and not include_load:
                continue
            if kind == "delivery" and not include_delivery:
                continue
            if not _keep_customer(_segment_slot_customer(kind, slot_load, slot_delivery)):
                continue
            base_row = _build_archive_segment_row(
                0,
                plan_common,
                kind,
                slot_load,
                slot_delivery,
                seq,
                stop_product,
                plan_products_by_id=plan_products_by_id,
                plan_products_by_product_id=plan_products_by_product_id,
                standard_by_id=standard_by_id,
                alias_options_by_id=alias_options_by_id,
            )
            plan_product = None
            if stop_product is not None:
                plan_product = _lookup_plan_product(
                    stop_product, plan_products_by_id, plan_products_by_product_id
                )
            for expanded in expand_export_rows_for_plan_brand(
                base_row,
                archive_status=getattr(plan, "archive_status", None),
                plan_product=plan_product,
                stop_product_id=str(getattr(stop_product, "id", None) or "")
                if stop_product is not None
                else None,
                plan_product_id=str(getattr(plan_product, "id", None) or "")
                if plan_product is not None
                else str(getattr(stop_product, "plan_product_id", None) or "")
                if stop_product is not None
                else None,
                by_stop_id=brand_by_stop,
                by_plan_product_id=brand_by_plan_product,
                apply_unit_price=True,
            ):
                row_no += 1
                expanded["rowNo"] = row_no
                rows.append(expanded)

    return rows


ARCHIVE_MERGED_SHEET_TITLE = "计划归档（合并）"

# 与列表 PL_PLAN_SPAN_PROPS 对应：整张计划纵向合并（含装货侧）。
_PLAN_LEVEL_MERGE_FIELDS = frozenset(
    {
        "rowNo",
        "planNo",
        "planDate",
        "archiveStatus",
        "followStatus",
        "planType",
        "erpEntered",
        "licensePlate",
        "freightType",
        "freightPrice",
        "freightAmount",
        "remark",
        "loadCustomer",
        "loadAddress",
        "loadTime",
    }
)
# 与列表 PL_DELIVERY_SPAN_PROPS 对应：同一送货段合并；仅一个送货段时连装货行一起合并。
_DELIVERY_LEVEL_MERGE_FIELDS = frozenset(
    {
        "deliveryCustomer",
        "deliveryAddress",
        "deliveryTime",
    }
)


def _iter_equal_spans(rows: list[dict], key_fn) -> list[tuple[int, int]]:
    """返回需要合并的闭区间（数据行下标，不含表头）。单行不合并。"""
    spans: list[tuple[int, int]] = []
    index = 0
    total = len(rows)
    while index < total:
        key = key_fn(rows[index])
        end = index + 1
        while end < total and key_fn(rows[end]) == key:
            end += 1
        if end - index > 1:
            spans.append((index, end - 1))
        index = end
    return spans


_PLAN_LIST_PLAN_MERGE_FIELDS = frozenset(
    {
        "planNo",
        "belongsCompany",
        "followStatus",
        "planDate",
        "deliveryTimes",
        "planType",
        "licensePlate",
        "personnelSummary",
        "followUserName",
        "archiveStatus",
        "erpEntered",
        "hazardCertReady",
        "loadCustomer",
        "loadAddressTime",
        "loadDibang",
    }
)
_PLAN_LIST_DELIVERY_MERGE_FIELDS = frozenset(
    {
        "deliveryCustomer",
        "deliveryAddressTime",
        "deliveryDibang",
    }
)
_PLAN_LIST_LOAD_COPY_FIELDS = ("loadCustomer", "loadAddressTime", "loadDibang")


def _export_plan_group_key(row: dict) -> str:
    return str(row.get("planId") or row.get("id") or "")


def prepare_archive_list_display_rows(
    rows: list[dict],
    *,
    contract_field: str = "contractNo",
    load_fields: tuple[str, ...] = ("loadCustomer", "loadAddress", "loadTime"),
    delivery_fields: Optional[frozenset] = None,
    number_plans: bool = True,
    join_plan_fields: tuple[str, ...] = (),
) -> list[dict]:
    """按列表合并规则整理单元格取值。

    序号按计划计，一张计划一个号（归档导出）。
    只有一个送货段时，送货侧字段写到该计划首行，以便和装货行合并后仍能看见。
    多个送货段时，装货行的送货客户为「—」。
    """
    if delivery_fields is None:
        delivery_fields = _DELIVERY_LEVEL_MERGE_FIELDS
    prepared: list[dict] = []
    plan_index = 0
    cursor = 0
    while cursor < len(rows):
        plan_id = _export_plan_group_key(rows[cursor])
        end = cursor + 1
        while end < len(rows) and _export_plan_group_key(rows[end]) == plan_id:
            end += 1
        group = [dict(item) for item in rows[cursor:end]]
        plan_index += 1
        delivery_seqs: list[int] = []
        for item in group:
            if item.get("_stopKind") == "delivery" and item.get("_deliverySeq") is not None:
                seq = int(item["_deliverySeq"])
                if seq not in delivery_seqs:
                    delivery_seqs.append(seq)
        sole_seq = delivery_seqs[0] if len(delivery_seqs) == 1 else None
        for item in group:
            kind = item.get("_stopKind") or ""
            if kind == "load":
                merge_seq = sole_seq if sole_seq is not None else 0
            elif kind == "delivery":
                merge_seq = int(item.get("_deliverySeq") or 1)
            else:
                merge_seq = 1
            item["_mergeDeliverySeq"] = merge_seq
            item["_contractMergeKey"] = (
                f"{kind}|{merge_seq}|{item.get(contract_field) or ''}"
            )
            if number_plans:
                item["rowNo"] = plan_index
        if sole_seq is not None:
            delivery_row = next((item for item in group if item.get("_stopKind") == "delivery"), None)
            if delivery_row is not None:
                for field in delivery_fields:
                    group[0][field] = delivery_row.get(field) or ""
        else:
            for item in group:
                if item.get("_mergeDeliverySeq") != 0:
                    continue
                if not item.get("deliveryCustomer"):
                    item["deliveryCustomer"] = "—"
        load_row = next((item for item in group if item.get("_stopKind") == "load"), None)
        if load_row is not None:
            for field in load_fields:
                if not group[0].get(field):
                    group[0][field] = load_row.get(field) or ""
        for field in join_plan_fields:
            seen: list[str] = []
            for item in group:
                text = str(item.get(field) or "").strip()
                if text and text not in seen:
                    seen.append(text)
            if seen:
                group[0][field] = "\n".join(seen)
        prepared.extend(group)
        cursor = end
    return prepared


def append_archive_list_merge_sheet(
    wb,
    columns,
    rows: list[dict],
    resolver,
    *,
    sheet_title: Optional[str] = None,
    plan_fields: Optional[frozenset] = None,
    delivery_fields: Optional[frozenset] = None,
    contract_field: str = "contractNo",
    load_fields: tuple[str, ...] = ("loadCustomer", "loadAddress", "loadTime"),
    number_plans: bool = True,
    join_plan_fields: tuple[str, ...] = (),
) -> None:
    """追加按列表合并的 sheet。默认标题为「计划归档（合并）」。"""
    from openpyxl.styles import Alignment

    if delivery_fields is None:
        delivery_fields = _DELIVERY_LEVEL_MERGE_FIELDS
    if plan_fields is None:
        plan_fields = _PLAN_LEVEL_MERGE_FIELDS
    display_rows = prepare_archive_list_display_rows(
        list(rows),
        contract_field=contract_field,
        load_fields=load_fields,
        delivery_fields=delivery_fields,
        number_plans=number_plans,
        join_plan_fields=join_plan_fields,
    )
    sheet = wb.create_sheet((sheet_title or ARCHIVE_MERGED_SHEET_TITLE)[:31])
    sheet.append([column.title for column in columns])
    for row in display_rows:
        sheet.append(resolver.row_to_cells(row, columns))

    field_column = {column.field: index + 1 for index, column in enumerate(columns)}
    vertical = Alignment(vertical="center", wrap_text=True)

    def merge_fields(fields, spans: list[tuple[int, int]]) -> None:
        for field in fields:
            column_index = field_column.get(field)
            if not column_index:
                continue
            for start, end in spans:
                sheet.merge_cells(
                    start_row=start + 2,
                    start_column=column_index,
                    end_row=end + 2,
                    end_column=column_index,
                )
                sheet.cell(start + 2, column_index).alignment = vertical

    plan_key = _export_plan_group_key
    plan_spans = _iter_equal_spans(display_rows, plan_key)
    merged_plan_fields = set(plan_fields)
    merged_plan_fields.update(field for field in field_column if str(field).startswith("attach"))
    merge_fields(merged_plan_fields, plan_spans)
    merge_fields(
        delivery_fields,
        _iter_equal_spans(
            display_rows,
            lambda row: (_export_plan_group_key(row), row.get("_mergeDeliverySeq")),
        ),
    )
    if contract_field in field_column:
        merge_fields(
            [contract_field],
            _iter_equal_spans(
                display_rows,
                lambda row: (_export_plan_group_key(row), row.get("_contractMergeKey")),
            ),
        )


def export_plan_list_xlsx_response(
    db,
    *,
    tenant_no: str,
    sheet_title: str,
    columns,
    rows: list[dict],
    filename_stem: str,
):
    """运输计划 / 危化品证列表 Excel：明细页 + 按列表合并的第二页。"""
    from app.utils.excel_export import (
        DictLabelResolver,
        ExportValidationError,
        assert_export_row_count,
        workbook_to_streaming_response,
    )
    from openpyxl import Workbook

    assert_export_row_count(len(rows))
    if not columns:
        raise ExportValidationError("没有可导出的列")
    resolver = DictLabelResolver(db, columns, tenant_no=tenant_no)
    workbook = Workbook()
    detail = workbook.active
    detail.title = sheet_title[:31]
    detail.append([column.title for column in columns])
    for row in rows:
        detail.append(resolver.row_to_cells(row, columns))
    append_archive_list_merge_sheet(
        workbook,
        columns,
        rows,
        resolver,
        sheet_title=f"{sheet_title}（合并）",
        plan_fields=_PLAN_LIST_PLAN_MERGE_FIELDS,
        delivery_fields=_PLAN_LIST_DELIVERY_MERGE_FIELDS,
        contract_field="productContract",
        load_fields=_PLAN_LIST_LOAD_COPY_FIELDS,
        number_plans=False,
        join_plan_fields=("deliveryTimes",),
    )
    return workbook_to_streaming_response(workbook, filename_stem=filename_stem)


def export_plan_archive_xlsx_response(
    db,
    *,
    tenant_no: str,
    columns,
    rows: list[dict],
    filename_stem: str,
):
    """归档 Excel：第一页为段次明细，第二页为列表合并样式。"""
    from app.utils.excel_export import (
        DictLabelResolver,
        ExportValidationError,
        assert_export_row_count,
        workbook_to_streaming_response,
    )
    from openpyxl import Workbook

    assert_export_row_count(len(rows))
    if not columns:
        raise ExportValidationError("没有可导出的列")
    resolver = DictLabelResolver(db, columns, tenant_no=tenant_no)
    workbook = Workbook()
    detail = workbook.active
    detail.title = "计划归档"
    detail.append([column.title for column in columns])
    for row in rows:
        detail.append(resolver.row_to_cells(row, columns))
    append_archive_list_merge_sheet(workbook, columns, rows, resolver)
    return workbook_to_streaming_response(workbook, filename_stem=filename_stem)